line_codec.rs (14651B)
1 use anyhow::{Context, Result}; 2 use tokio::{ 3 io::{AsyncBufReadExt, AsyncRead, BufReader}, 4 net::tcp::OwnedReadHalf, 5 }; 6 7 use crate::app::{GossipEnvelope, TRANSACTION_BATCH_LIMIT}; 8 9 use super::{ 10 GossipNetwork, MAX_BLOCK_BATCH, MAX_GOSSIP_LINE_BYTES, MAX_INVENTORY_ITEMS, 11 MAX_OBJECT_REQUESTS, MAX_PEER_LIST, MAX_SNAPSHOT_BLOCKS, metrics::P2pMetricsCounters, 12 }; 13 14 pub(super) struct LimitedLineReader<R> { 15 reader: BufReader<R>, 16 pending: Vec<u8>, 17 } 18 19 impl<R: AsyncRead + Unpin> LimitedLineReader<R> { 20 pub(super) fn new(reader: R) -> Self { 21 Self { 22 reader: BufReader::new(reader), 23 pending: Vec::new(), 24 } 25 } 26 27 pub(super) async fn read_line(&mut self) -> Result<Option<String>> { 28 loop { 29 let available = self.reader.fill_buf().await?; 30 if available.is_empty() { 31 if self.pending.is_empty() { 32 return Ok(None); 33 } 34 anyhow::bail!("peer closed before completing a gossip message"); 35 } 36 37 if let Some(newline) = available.iter().position(|byte| *byte == b'\n') { 38 if self.pending.len() + newline > MAX_GOSSIP_LINE_BYTES { 39 anyhow::bail!("p2p message exceeds {} byte limit", MAX_GOSSIP_LINE_BYTES); 40 } 41 self.pending.extend_from_slice(&available[..newline]); 42 self.reader.consume(newline + 1); 43 if self.pending.ends_with(b"\r") { 44 self.pending.pop(); 45 } 46 let bytes = std::mem::take(&mut self.pending); 47 return String::from_utf8(bytes) 48 .context("p2p message is not valid UTF-8") 49 .map(Some); 50 } 51 52 if self.pending.len() + available.len() > MAX_GOSSIP_LINE_BYTES { 53 anyhow::bail!("p2p message exceeds {} byte limit", MAX_GOSSIP_LINE_BYTES); 54 } 55 let consumed = available.len(); 56 self.pending.extend_from_slice(available); 57 self.reader.consume(consumed); 58 } 59 } 60 } 61 62 pub(super) async fn read_session_envelope( 63 network: &GossipNetwork, 64 connection_label: &str, 65 reader: &mut LimitedLineReader<OwnedReadHalf>, 66 ) -> Result<Option<GossipEnvelope>> { 67 let Some(line) = reader.read_line().await? else { 68 return Ok(None); 69 }; 70 P2pMetricsCounters::add(&network.inner.metrics.bytes_received, line.len() as u64 + 1); 71 if line.trim().is_empty() { 72 P2pMetricsCounters::inc(&network.inner.metrics.empty_frames); 73 P2pMetricsCounters::set_last( 74 &network.inner.metrics.last_empty_frame_remote, 75 connection_label.to_string(), 76 ); 77 anyhow::bail!("empty p2p envelope"); 78 } 79 80 match parse_envelope(&line) { 81 Ok(envelope) => { 82 P2pMetricsCounters::inc(&network.inner.metrics.envelopes_received); 83 record_received_envelope_kind(&network.inner.metrics, &envelope); 84 Ok(Some(envelope)) 85 } 86 Err(error) => { 87 P2pMetricsCounters::inc(&network.inner.metrics.parse_errors); 88 P2pMetricsCounters::set_last( 89 &network.inner.metrics.last_parse_error, 90 format!("{connection_label}: {error:#}"), 91 ); 92 Err(error) 93 } 94 } 95 } 96 97 pub(super) fn record_received_envelope_kind( 98 metrics: &P2pMetricsCounters, 99 envelope: &GossipEnvelope, 100 ) { 101 match envelope { 102 GossipEnvelope::Hello(_) => { 103 P2pMetricsCounters::inc(&metrics.hello_envelopes_received); 104 } 105 GossipEnvelope::PeerStatus { .. } => { 106 P2pMetricsCounters::inc(&metrics.peer_status_envelopes_received); 107 } 108 GossipEnvelope::Inventory { .. } => { 109 P2pMetricsCounters::inc(&metrics.inventory_envelopes_received); 110 } 111 GossipEnvelope::BlindedTransaction(_) => { 112 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 113 P2pMetricsCounters::inc(&metrics.blinded_transaction_envelopes_received); 114 P2pMetricsCounters::inc(&metrics.blinded_transactions_received); 115 } 116 GossipEnvelope::BlindedTransactions { transactions } => { 117 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 118 P2pMetricsCounters::inc(&metrics.blinded_transaction_envelopes_received); 119 P2pMetricsCounters::add( 120 &metrics.blinded_transactions_received, 121 transactions.len() as u64, 122 ); 123 } 124 GossipEnvelope::MineAction(_) => { 125 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 126 } 127 GossipEnvelope::MineActions { .. } => { 128 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 129 } 130 GossipEnvelope::BlindedReveal(_) => { 131 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 132 P2pMetricsCounters::inc(&metrics.blinded_reveal_envelopes_received); 133 P2pMetricsCounters::inc(&metrics.blinded_reveals_received); 134 } 135 GossipEnvelope::BlindedReveals { reveals } => { 136 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 137 P2pMetricsCounters::inc(&metrics.blinded_reveal_envelopes_received); 138 P2pMetricsCounters::add(&metrics.blinded_reveals_received, reveals.len() as u64); 139 } 140 GossipEnvelope::RevealBundle(bundle) => { 141 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 142 P2pMetricsCounters::inc(&metrics.blinded_reveal_envelopes_received); 143 P2pMetricsCounters::add( 144 &metrics.blinded_reveals_received, 145 bundle.reveals.len() as u64, 146 ); 147 } 148 GossipEnvelope::RevealBundles { bundles } => { 149 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 150 P2pMetricsCounters::inc(&metrics.blinded_reveal_envelopes_received); 151 P2pMetricsCounters::add( 152 &metrics.blinded_reveals_received, 153 bundles 154 .iter() 155 .map(|bundle| bundle.reveals.len() as u64) 156 .sum::<u64>(), 157 ); 158 } 159 GossipEnvelope::Block(_) 160 | GossipEnvelope::Blocks { .. } 161 | GossipEnvelope::ChainSnapshot(_) => { 162 P2pMetricsCounters::inc(&metrics.data_envelopes_received); 163 } 164 GossipEnvelope::ChainSnapshotRequest 165 | GossipEnvelope::BlockRangeRequest { .. } 166 | GossipEnvelope::BlockRequest { .. } 167 | GossipEnvelope::PeerAnnouncement { .. } 168 | GossipEnvelope::PeerVerificationChallenge { .. } 169 | GossipEnvelope::PeerVerificationResponse { .. } 170 | GossipEnvelope::PeerList { .. } => { 171 P2pMetricsCounters::inc(&metrics.control_envelopes_received); 172 } 173 } 174 } 175 176 pub(super) fn parse_envelope(line: &str) -> Result<GossipEnvelope> { 177 if line.trim().is_empty() { 178 anyhow::bail!("empty p2p envelope"); 179 } 180 let envelope = serde_json::from_str(line).context("invalid p2p envelope JSON")?; 181 validate_envelope_limits(&envelope)?; 182 Ok(envelope) 183 } 184 185 pub(super) fn validate_envelope_limits(envelope: &GossipEnvelope) -> Result<()> { 186 match envelope { 187 GossipEnvelope::BlockRangeRequest { limit, .. } => { 188 ensure_len("block range request", *limit, MAX_BLOCK_BATCH)?; 189 } 190 GossipEnvelope::BlockRequest { hashes } => { 191 ensure_len("block request", hashes.len(), MAX_OBJECT_REQUESTS)?; 192 } 193 GossipEnvelope::Inventory { blocks } => { 194 ensure_len("block inventory", blocks.len(), MAX_INVENTORY_ITEMS)?; 195 } 196 GossipEnvelope::BlindedTransactions { transactions } => { 197 ensure_len( 198 "blinded transaction batch", 199 transactions.len(), 200 TRANSACTION_BATCH_LIMIT, 201 )?; 202 } 203 GossipEnvelope::MineActions { transactions } => { 204 ensure_len( 205 "mine action batch", 206 transactions.len(), 207 TRANSACTION_BATCH_LIMIT, 208 )?; 209 } 210 GossipEnvelope::BlindedReveals { reveals } => { 211 ensure_len( 212 "blinded reveal batch", 213 reveals.len(), 214 TRANSACTION_BATCH_LIMIT, 215 )?; 216 } 217 GossipEnvelope::RevealBundles { bundles } => { 218 ensure_len( 219 "reveal bundle batch", 220 bundles.len(), 221 TRANSACTION_BATCH_LIMIT, 222 )?; 223 } 224 GossipEnvelope::Blocks { blocks } => { 225 ensure_len("block batch", blocks.len(), MAX_BLOCK_BATCH)?; 226 } 227 GossipEnvelope::ChainSnapshot(snapshot) => { 228 ensure_len("chain snapshot", snapshot.blocks.len(), MAX_SNAPSHOT_BLOCKS)?; 229 } 230 GossipEnvelope::PeerList { peers } => { 231 ensure_len("peer list", peers.len(), MAX_PEER_LIST)?; 232 } 233 GossipEnvelope::Hello(_) 234 | GossipEnvelope::ChainSnapshotRequest 235 | GossipEnvelope::PeerStatus { .. } 236 | GossipEnvelope::BlindedTransaction(_) 237 | GossipEnvelope::MineAction(_) 238 | GossipEnvelope::BlindedReveal(_) 239 | GossipEnvelope::RevealBundle(_) 240 | GossipEnvelope::Block(_) 241 | GossipEnvelope::PeerAnnouncement { .. } 242 | GossipEnvelope::PeerVerificationChallenge { .. } 243 | GossipEnvelope::PeerVerificationResponse { .. } => {} 244 } 245 Ok(()) 246 } 247 248 fn ensure_len(label: &str, len: usize, max: usize) -> Result<()> { 249 if len > max { 250 anyhow::bail!("{label} has {len} items, exceeding limit {max}"); 251 } 252 Ok(()) 253 } 254 255 #[cfg(test)] 256 mod tests { 257 use tokio::io::AsyncWriteExt; 258 259 use crate::{ 260 adapters::p2p::{MAX_INVENTORY_ITEMS, MAX_OBJECT_REQUESTS, metrics::P2pMetricsCounters}, 261 app::{BlockInventory, GossipEnvelope}, 262 domain::{BlindedReveal, BlindedTransaction}, 263 }; 264 265 use super::{ 266 LimitedLineReader, parse_envelope, record_received_envelope_kind, validate_envelope_limits, 267 }; 268 269 #[test] 270 fn oversized_inventory_is_rejected_before_processing() { 271 let envelope = GossipEnvelope::Inventory { 272 blocks: vec![ 273 BlockInventory { 274 height: 1, 275 hash: "hash".to_string() 276 }; 277 MAX_INVENTORY_ITEMS + 1 278 ], 279 }; 280 281 let error = validate_envelope_limits(&envelope).unwrap_err(); 282 283 assert!(error.to_string().contains("block inventory")); 284 } 285 286 #[test] 287 fn parser_applies_envelope_limits() { 288 let line = serde_json::to_string(&GossipEnvelope::BlockRequest { 289 hashes: vec!["hash".to_string(); MAX_OBJECT_REQUESTS + 1], 290 }) 291 .unwrap(); 292 293 let error = parse_envelope(&line).unwrap_err(); 294 295 assert!(error.to_string().contains("block request")); 296 } 297 298 #[test] 299 fn parser_rejects_empty_envelope_without_json_eof() { 300 let error = parse_envelope("").unwrap_err(); 301 302 assert!(error.to_string().contains("empty p2p envelope")); 303 assert!(!format!("{error:#}").contains("EOF while parsing")); 304 } 305 306 #[test] 307 fn parser_accepts_legacy_peer_status_without_mempool_fields() { 308 let envelope = 309 parse_envelope(r#"{"type":"peer_status","height":7,"tip_hash":"tip"}"#).unwrap(); 310 311 assert_eq!( 312 envelope, 313 GossipEnvelope::PeerStatus { 314 height: 7, 315 tip_hash: "tip".to_string(), 316 time_ms: 0, 317 } 318 ); 319 } 320 321 #[test] 322 fn received_envelope_metrics_are_categorized() { 323 let metrics = P2pMetricsCounters::default(); 324 let blinded_tx = BlindedTransaction { 325 commitment: "commitment".to_string(), 326 inputs: Vec::new(), 327 fee: 3, 328 encrypted_size: 128, 329 expires_at_height: 20, 330 nonce: "nonce".to_string(), 331 ciphertext: "ciphertext".to_string(), 332 payload_hash: "payload-hash".to_string(), 333 }; 334 let blinded_reveal = BlindedReveal { 335 commitment: "commitment".to_string(), 336 key: "key".to_string(), 337 }; 338 339 record_received_envelope_kind( 340 &metrics, 341 &GossipEnvelope::PeerStatus { 342 height: 7, 343 tip_hash: "tip".to_string(), 344 time_ms: 1_000, 345 }, 346 ); 347 record_received_envelope_kind(&metrics, &GossipEnvelope::Inventory { blocks: Vec::new() }); 348 record_received_envelope_kind(&metrics, &GossipEnvelope::Blocks { blocks: Vec::new() }); 349 record_received_envelope_kind( 350 &metrics, 351 &GossipEnvelope::BlindedTransactions { 352 transactions: vec![blinded_tx.clone(), blinded_tx], 353 }, 354 ); 355 record_received_envelope_kind(&metrics, &GossipEnvelope::BlindedReveal(blinded_reveal)); 356 record_received_envelope_kind(&metrics, &GossipEnvelope::ChainSnapshotRequest); 357 358 let snapshot = metrics.snapshot(); 359 assert_eq!(snapshot.peer_status_envelopes_received, 1); 360 assert_eq!(snapshot.inventory_envelopes_received, 1); 361 assert_eq!(snapshot.data_envelopes_received, 3); 362 assert_eq!(snapshot.blinded_transaction_envelopes_received, 1); 363 assert_eq!(snapshot.blinded_transactions_received, 2); 364 assert_eq!(snapshot.blinded_reveal_envelopes_received, 1); 365 assert_eq!(snapshot.blinded_reveals_received, 1); 366 assert_eq!(snapshot.control_envelopes_received, 1); 367 } 368 369 #[tokio::test] 370 async fn limited_line_reader_keeps_partial_line_after_cancelled_read() { 371 let (mut writer, reader) = tokio::io::duplex(1024); 372 let mut reader = LimitedLineReader::new(reader); 373 let line = serde_json::to_string(&GossipEnvelope::PeerStatus { 374 height: 7, 375 tip_hash: "tip".to_string(), 376 time_ms: 1_000, 377 }) 378 .unwrap(); 379 let split_at = line.len() / 2; 380 381 writer 382 .write_all(&line.as_bytes()[..split_at]) 383 .await 384 .unwrap(); 385 let cancelled = 386 tokio::time::timeout(std::time::Duration::from_millis(25), reader.read_line()).await; 387 388 assert!(cancelled.is_err()); 389 390 writer 391 .write_all(&line.as_bytes()[split_at..]) 392 .await 393 .unwrap(); 394 writer.write_all(b"\n").await.unwrap(); 395 396 assert_eq!( 397 reader.read_line().await.unwrap().as_deref(), 398 Some(line.as_str()) 399 ); 400 } 401 }