line_codec.rs (15075B)
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::{ 261 MAX_GOSSIP_LINE_BYTES, MAX_INVENTORY_ITEMS, MAX_OBJECT_REQUESTS, 262 metrics::P2pMetricsCounters, 263 }, 264 app::{BlockInventory, GossipEnvelope}, 265 domain::{BlindedReveal, BlindedTransaction}, 266 }; 267 268 use super::{ 269 LimitedLineReader, parse_envelope, record_received_envelope_kind, validate_envelope_limits, 270 }; 271 272 #[test] 273 fn oversized_inventory_is_rejected_before_processing() { 274 let envelope = GossipEnvelope::Inventory { 275 blocks: vec![ 276 BlockInventory { 277 height: 1, 278 hash: "hash".to_string() 279 }; 280 MAX_INVENTORY_ITEMS + 1 281 ], 282 }; 283 284 let error = validate_envelope_limits(&envelope).unwrap_err(); 285 286 assert!(error.to_string().contains("block inventory")); 287 } 288 289 #[test] 290 fn parser_applies_envelope_limits() { 291 let line = serde_json::to_string(&GossipEnvelope::BlockRequest { 292 hashes: vec!["hash".to_string(); MAX_OBJECT_REQUESTS + 1], 293 }) 294 .unwrap(); 295 296 let error = parse_envelope(&line).unwrap_err(); 297 298 assert!(error.to_string().contains("block request")); 299 } 300 301 #[test] 302 fn parser_rejects_empty_envelope_without_json_eof() { 303 let error = parse_envelope("").unwrap_err(); 304 305 assert!(error.to_string().contains("empty p2p envelope")); 306 assert!(!format!("{error:#}").contains("EOF while parsing")); 307 } 308 309 #[test] 310 fn parser_accepts_legacy_peer_status_without_mempool_fields() { 311 let envelope = 312 parse_envelope(r#"{"type":"peer_status","height":7,"tip_hash":"tip"}"#).unwrap(); 313 314 assert_eq!( 315 envelope, 316 GossipEnvelope::PeerStatus { 317 height: 7, 318 tip_hash: "tip".to_string(), 319 time_ms: 0, 320 } 321 ); 322 } 323 324 #[test] 325 fn received_envelope_metrics_are_categorized() { 326 let metrics = P2pMetricsCounters::default(); 327 let blinded_tx = BlindedTransaction { 328 commitment: "commitment".to_string(), 329 inputs: Vec::new(), 330 fee: 3, 331 encrypted_size: 128, 332 expires_at_height: 20, 333 nonce: "nonce".to_string(), 334 ciphertext: "ciphertext".to_string(), 335 payload_hash: "payload-hash".to_string(), 336 }; 337 let blinded_reveal = BlindedReveal { 338 commitment: "commitment".to_string(), 339 key: "key".to_string(), 340 }; 341 342 record_received_envelope_kind( 343 &metrics, 344 &GossipEnvelope::PeerStatus { 345 height: 7, 346 tip_hash: "tip".to_string(), 347 time_ms: 1_000, 348 }, 349 ); 350 record_received_envelope_kind(&metrics, &GossipEnvelope::Inventory { blocks: Vec::new() }); 351 record_received_envelope_kind(&metrics, &GossipEnvelope::Blocks { blocks: Vec::new() }); 352 record_received_envelope_kind( 353 &metrics, 354 &GossipEnvelope::BlindedTransactions { 355 transactions: vec![blinded_tx.clone(), blinded_tx], 356 }, 357 ); 358 record_received_envelope_kind(&metrics, &GossipEnvelope::BlindedReveal(blinded_reveal)); 359 record_received_envelope_kind(&metrics, &GossipEnvelope::ChainSnapshotRequest); 360 361 let snapshot = metrics.snapshot(); 362 assert_eq!(snapshot.peer_status_envelopes_received, 1); 363 assert_eq!(snapshot.inventory_envelopes_received, 1); 364 assert_eq!(snapshot.data_envelopes_received, 3); 365 assert_eq!(snapshot.blinded_transaction_envelopes_received, 1); 366 assert_eq!(snapshot.blinded_transactions_received, 2); 367 assert_eq!(snapshot.blinded_reveal_envelopes_received, 1); 368 assert_eq!(snapshot.blinded_reveals_received, 1); 369 assert_eq!(snapshot.control_envelopes_received, 1); 370 } 371 372 #[tokio::test] 373 async fn limited_line_reader_keeps_partial_line_after_cancelled_read() { 374 let (mut writer, reader) = tokio::io::duplex(1024); 375 let mut reader = LimitedLineReader::new(reader); 376 let line = serde_json::to_string(&GossipEnvelope::PeerStatus { 377 height: 7, 378 tip_hash: "tip".to_string(), 379 time_ms: 1_000, 380 }) 381 .unwrap(); 382 let split_at = line.len() / 2; 383 384 writer 385 .write_all(&line.as_bytes()[..split_at]) 386 .await 387 .unwrap(); 388 let cancelled = 389 tokio::time::timeout(std::time::Duration::from_millis(25), reader.read_line()).await; 390 391 assert!(cancelled.is_err()); 392 393 writer 394 .write_all(&line.as_bytes()[split_at..]) 395 .await 396 .unwrap(); 397 writer.write_all(b"\n").await.unwrap(); 398 399 assert_eq!( 400 reader.read_line().await.unwrap().as_deref(), 401 Some(line.as_str()) 402 ); 403 } 404 405 #[tokio::test] 406 async fn limited_line_reader_rejects_oversized_partial_frame_without_newline() { 407 let bytes = vec![b'a'; MAX_GOSSIP_LINE_BYTES + 1]; 408 let mut reader = LimitedLineReader::new(bytes.as_slice()); 409 410 let error = reader.read_line().await.unwrap_err(); 411 412 assert!(error.to_string().contains("p2p message exceeds")); 413 } 414 }