diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 9d351ac2..75d980d8 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -89,14 +89,12 @@ server body, is deliberately not made. outside any conformance harness, and its stream, negotiation, and shared-memory frame-ring surfaces have never been checked against the browser, Kotlin, and TypeScript CultMesh bodies they must interoperate with. Runtime-local tests pass; that is not a parity claim. -- Finish porting RUDP hardening from the retiring GameCult/cultnet-rs lineage. Its exact - acknowledgement receipts are in. Two commits are not: bounded session memory (4a54f46) and - hostile sequence rollover rejection (91609ff). They rebuild the receive path -- stale-sequence - rejection, per-channel tracking caps, reassembly and ordered-buffer byte accounting -- in the - same code this lineage reworked with sequence domains and ACK horizons, and they change when - packets are dropped and reassembly refused. That is wire-observable, so the interop lanes are - the gate, not a local build. Until they land, this runtime accepts sequence and reassembly - input the other lineage rejects. Do not delete that repository before this is resolved. +- Carry the RUDP receive-path bounds to the other runtimes. Rust now caps tracked ordered + and sequenced channels, caps the out-of-order hold buffer by frames and bytes, refuses a + packet far ahead of the highest received so a forged sequence cannot starve a session, and + reports reliable sequence exhaustion instead of aborting. C#, TypeScript, Python and Kotlin + still accept all of it. These are refusals rather than wire-format changes, so the interop + lanes stay green either way -- which is exactly why the gap will not announce itself. - Keep performance claims evidence-bound: C# owns SoA; Python has benchmark comparison gates; other runtimes should advertise measured native hot paths only when those paths exist. diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index ab16bf0d..0ff9f322 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -33,6 +33,15 @@ const RUDP_FIXED_HEADER_BYTES: usize = 36; pub const CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024; pub const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS: usize = 32; const RUDP_RECEIVED_SEQUENCE_WINDOW: usize = 4_096; +/// A peer chooses its own channel ids and its own send order, so the per-channel +/// maps and the out-of-order hold buffer are both attacker-sized unless capped. +const RUDP_MAX_TRACKED_CHANNELS: usize = 64; +/// A peer cannot be trusted to pick the next sequence honestly. One packet far +/// ahead of the current highest would otherwise advance the receive state past +/// every legitimate sequence still in flight and starve the session. +const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024; +const RUDP_MAX_ORDERED_BUFFERED_FRAMES: usize = 1_024; +const RUDP_MAX_ORDERED_BUFFERED_BYTES: usize = 4 * 1024 * 1024; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum CultNetRudpPacketType { @@ -529,6 +538,25 @@ impl CultNetRudpSession { }); } + let far_ahead = packet.reliable + && self.highest_received_sequence.is_some_and(|highest| { + packet.sequence > highest + && packet.sequence - highest > RUDP_RECEIVE_AHEAD_WINDOW + }); + if far_ahead { + // Refuse without remembering. Recording it would move the highest + // received sequence out of reach of everything still in flight. + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + ready_to_send, + reply: None, + pong: false, + pong_payload: Vec::new(), + disconnected: false, + disconnect_reason: Vec::new(), + }); + } + let duplicate = packet.reliable && (self.received_sequences.contains(&packet.sequence) || self.highest_received_sequence.is_some_and(|highest| { @@ -562,8 +590,15 @@ impl CultNetRudpSession { }); }; let delivered = if ordered { - self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized) + self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)? } else if packet.sequenced { + if !self + .latest_sequenced_by_channel + .contains_key(&frame.channel_id) + && self.latest_sequenced_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS + { + return Err(anyhow!("RUDP session has too many sequenced channels")); + } let latest = self .latest_sequenced_by_channel .get(&frame.channel_id) @@ -724,8 +759,7 @@ impl CultNetRudpSession { let sequence = self.next_sequence; self.next_sequence = self .next_sequence - .checked_add(1) - .expect("reliable sequence overflow"); + .saturating_add(1); sequence } else if sequenced { let next = self @@ -733,7 +767,7 @@ impl CultNetRudpSession { .entry(channel_id.to_string()) .or_insert(1); let sequence = *next; - *next = next.checked_add(1).expect("sequenced channel overflow"); + *next = next.saturating_add(1); sequence } else { 0 @@ -808,6 +842,14 @@ impl CultNetRudpSession { return Err(anyhow!("RUDP reliable send queue is full")); } } + // Refuse before allocation rather than at it. A session that runs out of + // reliable sequence space must report it, not abort the process. + let required = u32::try_from(packet_count) + .ok() + .filter(|required| self.next_sequence.checked_add(*required).is_some()); + if required.is_none() { + return Err(anyhow!("RUDP reliable sequence space is exhausted")); + } Ok(()) } @@ -958,13 +1000,30 @@ impl CultNetRudpSession { Ok(()) } + fn ordered_buffered_frames(&self) -> usize { + self.ordered_buffers.values().map(BTreeMap::len).sum() + } + + fn ordered_buffered_bytes(&self) -> usize { + self.ordered_buffers + .values() + .flat_map(BTreeMap::values) + .map(|pending| pending.frame.payload.len()) + .sum() + } + fn deliver_ordered( &mut self, frame: CultNetRudpDeliveredFrame, next_sequence_after_frame: u32, expected_sequence_if_uninitialized: u32, - ) -> Vec { + ) -> Result> { let channel_id = frame.channel_id.clone(); + if !self.ordered_next_sequence_by_channel.contains_key(&channel_id) + && self.ordered_next_sequence_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS + { + return Err(anyhow!("RUDP session has too many ordered channels")); + } let mut next = if let Some(next) = self .ordered_next_sequence_by_channel .get(&channel_id) @@ -992,10 +1051,16 @@ impl CultNetRudpSession { } if frame.sequence < next { - return Vec::new(); + return Ok(Vec::new()); } if frame.sequence > next { + if self.ordered_buffered_frames() >= RUDP_MAX_ORDERED_BUFFERED_FRAMES + || self.ordered_buffered_bytes() + frame.payload.len() + > RUDP_MAX_ORDERED_BUFFERED_BYTES + { + return Err(anyhow!("RUDP ordered hold buffer is full")); + } self.ordered_buffers.entry(channel_id).or_default().insert( frame.sequence, PendingOrderedFrame { @@ -1003,14 +1068,14 @@ impl CultNetRudpSession { next_sequence: next_sequence_after_frame, }, ); - return Vec::new(); + return Ok(Vec::new()); } self.ordered_next_sequence_by_channel .insert(channel_id.clone(), next_sequence_after_frame); let mut delivered = vec![frame]; delivered.extend(self.drain_ordered(&channel_id)); - delivered + Ok(delivered) } fn drain_ordered(&mut self, channel_id: &str) -> Vec { diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 293b63a6..b56c14ba 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -2417,3 +2417,222 @@ fn schema_discovery_round_trips_over_legacy_gamecult_contract_when_inline_schema ); Ok(()) } + +#[test] +fn rudp_sequenced_channel_tracking_is_bounded() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 401, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 401, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + + // A peer picks its own channel ids, so the sequenced map is attacker-sized. + let mut error = None; + for channel in 0..256u32 { + let packet = sender.send( + &format!("channel-{channel}"), + b"state".to_vec(), + CultNetRudpSendOptions { + sequenced: true, + ..CultNetRudpSendOptions::default() + }, + )?; + if let Err(refused) = receiver.receive(&packet, 3) { + error = Some(refused); + break; + } + } + let error = error.expect("unbounded sequenced channel tracking"); + assert!( + error.to_string().contains("too many sequenced channels"), + "unexpected refusal: {error}" + ); + Ok(()) +} + +#[test] +fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 402, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 402, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + + let mut error = None; + for channel in 0..256u32 { + let packet = sender.send( + &format!("ordered-{channel}"), + b"state".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..CultNetRudpSendOptions::default() + }, + )?; + match receiver.receive(&packet, 3) { + Ok(_) => { + sender.receive(&receiver.create_ack_for(packet.sequence), 3)?; + } + Err(refused) => { + error = Some(refused); + break; + } + } + } + let error = error.expect("unbounded ordered channel tracking"); + assert!( + error.to_string().contains("too many ordered channels"), + "unexpected refusal: {error}" + ); + Ok(()) +} + +#[test] +fn rudp_ordered_hold_buffer_is_bounded() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 403, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 403, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + + let options = CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..CultNetRudpSendOptions::default() + }; + // Withhold the first frame so every later one has to be held. + let withheld = sender.send("schema", b"first".to_vec(), options.clone())?; + sender.receive(&receiver.create_ack_for(withheld.sequence), 3)?; + + let mut error = None; + for step in 0..4_096u32 { + let packet = sender.send("schema", vec![(step % 251) as u8], options.clone())?; + match receiver.receive(&packet, 4) { + Ok(result) => { + assert!(result.delivered.is_empty(), "gap should hold every frame"); + sender.receive(&receiver.create_ack_for(packet.sequence), 4)?; + } + Err(refused) => { + error = Some(refused); + break; + } + } + } + let error = error.expect("unbounded ordered hold buffer"); + assert!( + error.to_string().contains("ordered hold buffer is full"), + "unexpected refusal: {error}" + ); + Ok(()) +} + +#[test] +fn rudp_far_ahead_sequence_cannot_starve_the_session() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 404, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 404, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + + let options = CultNetRudpSendOptions { + reliable: true, + ..CultNetRudpSendOptions::default() + }; + let first = sender.send("schema", b"before".to_vec(), options.clone())?; + assert_eq!(receiver.receive(&first, 3)?.delivered.len(), 1); + sender.receive(&receiver.create_ack_for(first.sequence), 3)?; + + // A forged packet far past the current highest. Recording it would push the + // receive state beyond every sequence still in flight. + let forged = CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Data, + connection_id: 404, + sequence: first.sequence.saturating_add(500_000), + ack: 0, + ack_mask: 0, + channel_id: "schema".to_string(), + reliable: true, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: b"hostile".to_vec(), + }; + assert!(receiver.receive(&forged, 4)?.delivered.is_empty()); + + // The session must still accept the next legitimate packet. + let second = sender.send("schema", b"after".to_vec(), options)?; + assert_eq!( + receiver.receive(&second, 5)?.delivered[0].payload, + b"after".to_vec() + ); + Ok(()) +} + +#[test] +fn rudp_reliable_sequence_exhaustion_is_reported_not_fatal() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 405, + // One sequence left: the handshake takes it, so the next send is the + // first allocation with nowhere to go. + initial_sequence: u32::MAX - 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 405, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + + let error = sender + .send( + "schema", + b"one past the end".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ..CultNetRudpSendOptions::default() + }, + ) + .unwrap_err(); + assert!( + error.to_string().contains("sequence space is exhausted"), + "unexpected error: {error}" + ); + Ok(()) +}