From 7b7666f2c301bbad5daff6828ade7c802b3a2fc0 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 17:38:20 +0200 Subject: [PATCH] cultnet-rs: drop and count packets that belong to no session A foreign connection id, a malformed frame, a sender no Connect admitted, a Connect with the peer table full, and any packet the session refuses ended the transport loop with an Err. Odin died twice this way after nginx reloads moved its UDP flows (require_connection: connection id does not match). The hub and the socket transport now drop such packets and count them in CultNetTransportStats::packets_dropped. Only socket I/O errors leave the loop. The socket transport checks the connection id before it touches the peer endpoint or session, so a stray Connect can no longer move the endpoint or reset the session. Co-Authored-By: Claude Sonnet 5 --- packages/cultnet-rs/src/rudp.rs | 55 +++- packages/cultnet-rs/src/transport.rs | 5 + packages/cultnet-rs/tests/rudp_server_hub.rs | 17 +- .../cultnet-rs/tests/rudp_stray_packets.rs | 255 ++++++++++++++++++ 4 files changed, 314 insertions(+), 18 deletions(-) create mode 100644 packages/cultnet-rs/tests/rudp_stray_packets.rs diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 174739be..226f5c7b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1572,8 +1572,15 @@ impl CultNetRudpServerHub { }; wire.truncate(received); self.stats.bytes_received += received as u64; - let packet = decode_rudp_packet(&wire)?; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's id and a sender no + // Connect admitted are dropped and counted. Only the socket ends a loop. + let Ok(packet) = decode_rudp_packet(&wire) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; if packet.connection_id != self.connection_id { + self.stats.packets_dropped += 1; return Ok(true); } @@ -1581,7 +1588,10 @@ impl CultNetRudpServerHub { if let Some(peer) = self.peers.get_mut(&remote_addr) && peer.context.connect_payload == packet.payload { - let _ = peer.session.receive(&packet, now_ms())?; + if peer.session.receive(&packet, now_ms()).is_err() { + self.stats.packets_dropped += 1; + return Ok(true); + } let reply = peer .session .pending_accept_for_resend(now_ms()) @@ -1590,7 +1600,8 @@ impl CultNetRudpServerHub { return Ok(true); } if !self.peers.contains_key(&remote_addr) && self.peers.len() >= self.max_peers { - return Err(anyhow!("RUDP server hub peer limit reached")); + self.stats.packets_dropped += 1; + return Ok(true); } let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { connection_id: self.connection_id, @@ -1610,7 +1621,10 @@ impl CultNetRudpServerHub { session_generation: generation, connect_payload: packet.payload.clone(), }; - let accept = session.accept_connect(&packet, now_ms(), Vec::new())?; + let Ok(accept) = session.accept_connect(&packet, now_ms(), Vec::new()) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; self.send_packet(remote_addr, &accept)?; let replaced = self.peers.insert( remote_addr, @@ -1632,9 +1646,13 @@ impl CultNetRudpServerHub { } let Some(peer) = self.peers.get_mut(&remote_addr) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; + let Ok(result) = peer.session.receive(&packet, now_ms()) else { + self.stats.packets_dropped += 1; return Ok(true); }; - let result = peer.session.receive(&packet, now_ms())?; let context = peer.context.clone(); let ack = if packet.reliable { Some(peer.session.create_ack_for_received(packet.sequence)) @@ -1970,7 +1988,20 @@ impl CultNetRudpSocketTransportConnection { wire.truncate(received); self.stats.bytes_received += received as u64; - let packet = decode_rudp_packet(&wire)?; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's id, a sender that is not + // this transport's peer and a packet the session refuses are dropped and + // counted. Only the socket ends a loop. The id is checked before the + // peer endpoint or session state is touched, so a stray Connect can + // neither move the endpoint nor reset the session. + let Ok(packet) = decode_rudp_packet(&wire) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; + if packet.connection_id != self.session.connection_id() { + self.stats.packets_dropped += 1; + return Ok(true); + } if let Some(expected) = self.remote_addr { if expected != remote_addr { if self.mode == CultNetRudpSocketMode::Server @@ -1978,6 +2009,7 @@ impl CultNetRudpSocketTransportConnection { { self.remote_addr = Some(remote_addr); } else { + self.stats.packets_dropped += 1; return Ok(true); } } @@ -1985,6 +2017,7 @@ impl CultNetRudpSocketTransportConnection { if self.mode == CultNetRudpSocketMode::Server && packet.packet_type != CultNetRudpPacketType::Connect { + self.stats.packets_dropped += 1; return Ok(true); } self.remote_addr = Some(remote_addr); @@ -1993,12 +2026,18 @@ impl CultNetRudpSocketTransportConnection { && packet.packet_type == CultNetRudpPacketType::Connect { self.session.reset_peer_state(); - let accept = self.session.accept_connect(&packet, now_ms(), Vec::new())?; + let Ok(accept) = self.session.accept_connect(&packet, now_ms(), Vec::new()) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; self.send_packet(&accept)?; return Ok(true); } - let result = self.session.receive(&packet, now_ms())?; + let Ok(result) = self.session.receive(&packet, now_ms()) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; if let Some(reply) = result.reply { self.send_packet(&reply)?; } diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index d097aa3c..71e714c4 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -26,6 +26,11 @@ pub struct CultNetTransportStats { /// payload that lost a fragment on the wire. Zero for non-fragmenting /// transports. pub fragment_sets_evicted: u64, + /// Datagrams read and discarded because they belong to no session on this + /// transport: malformed frames, another session's connection id, an + /// unadmitted sender, or a packet the session refused. A steady rate means + /// a flow is being moved or scanned, not that this transport is failing. + pub packets_dropped: u64, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/packages/cultnet-rs/tests/rudp_server_hub.rs b/packages/cultnet-rs/tests/rudp_server_hub.rs index 201cf3b4..80cfde3b 100644 --- a/packages/cultnet-rs/tests/rudp_server_hub.rs +++ b/packages/cultnet-rs/tests/rudp_server_hub.rs @@ -415,16 +415,13 @@ fn session_and_hub_reject_configured_memory_bounds() -> Result<()> { let mut second = client(server_addr)?; connect(&mut hub, &mut first, b"first")?; second.connect(b"second".to_vec())?; - for _ in 0..20 { - match hub.receive_event_once() { - Err(error) => { - assert!(error.to_string().contains("peer limit")); - return Ok(()); - } - Ok(_) => continue, - } - } - Err(anyhow!("hub did not enforce peer limit")) + // The peer limit refuses the second Connect without ending the hub loop: + // a full table is not a fault of the process that holds it. + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert_eq!(hub.sessions().len(), 1); + assert_eq!(hub.stats().packets_dropped, 1); + Ok(()) } #[test] diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs new file mode 100644 index 00000000..58f491fa --- /dev/null +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -0,0 +1,255 @@ +//! A packet that does not belong to a session is dropped; only socket failures +//! end a transport loop. Each test puts one class of stray in front of a real +//! session and requires the real session to keep being served. + +use std::net::{SocketAddr, UdpSocket}; +use std::time::Duration; + +use anyhow::Result; +use cultnet_rs::*; + +const CONNECTION_ID: u32 = 0x4355_4c54; +const FOREIGN_ID: u32 = 0x0BAD_F10E; + +#[derive(Clone, Copy)] +enum Stray { + /// Bytes that are not a RUDP frame. + Malformed, + /// A well-formed session that carries someone else's connection id: what a + /// moved UDP flow delivers. + ForeignConnectionId, + /// The right connection id from a sender no Connect ever admitted, or one + /// that already closed. + UnadmittedSender, +} + +fn socket() -> Result { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(20)))?; + Ok(socket) +} + +fn raw_session(connection_id: u32) -> CultNetRudpSession { + CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id, + initial_sequence: 1, + resend_delay_ms: 250, + max_pending_reliable_packets: None, + }) +} + +fn reliable() -> CultNetRudpSendOptions { + CultNetRudpSendOptions { + reliable: true, + ..Default::default() + } +} + +fn send_to(socket: &UdpSocket, to: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { + socket.send_to(&encode_rudp_packet(packet)?, to)?; + Ok(()) +} + +fn send_stray(kind: Stray, from: &UdpSocket, to: SocketAddr) -> Result<()> { + match kind { + Stray::Malformed => { + from.send_to(b"not a rudp packet", to)?; + from.send_to(&[], to)?; + } + Stray::ForeignConnectionId => { + let mut foreign = raw_session(FOREIGN_ID); + send_to(from, to, &foreign.create_connect(0, b"foreign".to_vec())?)?; + foreign.assume_connected(0); + send_to(from, to, &foreign.send("schema", b"foreign".to_vec(), reliable())?)?; + send_to(from, to, &foreign.create_ack())?; + send_to(from, to, &foreign.create_disconnect(b"foreign".to_vec()))?; + } + Stray::UnadmittedSender => { + let mut orphan = raw_session(CONNECTION_ID); + orphan.assume_connected(0); + send_to(from, to, &orphan.send("schema", b"orphan".to_vec(), reliable())?)?; + send_to(from, to, &orphan.create_ack())?; + send_to(from, to, &orphan.create_disconnect(b"orphan".to_vec()))?; + } + } + Ok(()) +} + +fn hub_keeps_serving_its_session(kind: Stray) -> Result<()> { + let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + options.max_peers = 1; + let mut hub = CultNetRudpServerHub::new(options)?; + let hub_addr = hub.local_addr()?; + let mut real = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("real", socket()?, hub_addr, CONNECTION_ID), + )?; + real.connect(b"real".to_vec())?; + let mut admitted = false; + for _ in 0..20 { + if let Some(CultNetRudpServerEvent::Connected { .. }) = hub.receive_event_once()? { + admitted = true; + break; + } + } + assert!(admitted); + let _ = real.receive_once()?; + + let stray = socket()?; + send_stray(kind, &stray, hub_addr)?; + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert!(hub.stats().packets_dropped > 0, "strays are counted"); + + real.send("schema", b"still here".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + match hub.receive_event_once()? { + Some(CultNetRudpServerEvent::Frame { frame, .. }) => { + assert_eq!(frame.payload, b"still here"); + } + other => panic!("the real session must still be served, got {other:?}"), + } + Ok(()) +} + +#[test] +fn hub_drops_a_connect_that_arrives_with_the_peer_table_full() -> Result<()> { + let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + options.max_peers = 1; + let mut hub = CultNetRudpServerHub::new(options)?; + let hub_addr = hub.local_addr()?; + let mut real = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("real", socket()?, hub_addr, CONNECTION_ID), + )?; + real.connect(b"real".to_vec())?; + while !matches!( + hub.receive_event_once()?, + Some(CultNetRudpServerEvent::Connected { .. }) + ) {} + let _ = real.receive_once()?; + + send_to( + &socket()?, + hub_addr, + &raw_session(CONNECTION_ID).create_connect(0, b"late".to_vec())?, + )?; + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert_eq!(hub.sessions().len(), 1); + assert_eq!(hub.stats().packets_dropped, 1); + + real.send("schema", b"still here".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + assert!(matches!( + hub.receive_event_once()?, + Some(CultNetRudpServerEvent::Frame { .. }) + )); + Ok(()) +} + +/// A moved flow answers a client from its server address with someone else's +/// session. +fn client_keeps_its_session(kind: Stray) -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let connect = decode_rudp_packet(&buffer[..received])?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = server_session.accept_connect(&connect, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + assert!(client.connected()); + + send_stray(kind, &server, client_addr)?; + send_to( + &server, + client_addr, + &server_session.send("schema", b"real".to_vec(), reliable())?, + )?; + let frame = client + .receive_once()? + .expect("the real frame behind the strays"); + assert_eq!(frame.payload, b"real"); + assert!(client.stats().packets_dropped > 0, "strays are counted"); + Ok(()) +} + +/// A server-mode transport owns one peer. A stray, including a foreign Connect +/// from another address, neither resets that peer's session nor moves the +/// endpoint the transport answers. +fn server_mode_keeps_its_peer(kind: Stray) -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + let mut buffer = vec![0_u8; 65_535]; + let (received, _) = peer.recv_from(&mut buffer)?; + peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?; + + send_stray(kind, &socket()?, server_addr)?; + send_stray(kind, &peer, server_addr)?; + send_to( + &peer, + server_addr, + &peer_session.send("schema", b"real".to_vec(), reliable())?, + )?; + let frame = server + .receive_once()? + .expect("the real frame behind the strays"); + assert_eq!(frame.payload, b"real"); + assert!(server.stats().packets_dropped > 0, "strays are counted"); + Ok(()) +} + +macro_rules! per_stray { + ($body:ident, $malformed:ident, $foreign:ident, $unadmitted:ident) => { + #[test] + fn $malformed() -> Result<()> { + $body(Stray::Malformed) + } + #[test] + fn $foreign() -> Result<()> { + $body(Stray::ForeignConnectionId) + } + #[test] + fn $unadmitted() -> Result<()> { + $body(Stray::UnadmittedSender) + } + }; + ($body:ident, $malformed:ident, $foreign:ident) => { + #[test] + fn $malformed() -> Result<()> { + $body(Stray::Malformed) + } + #[test] + fn $foreign() -> Result<()> { + $body(Stray::ForeignConnectionId) + } + }; +} + +per_stray!( + hub_keeps_serving_its_session, + hub_survives_malformed_frames, + hub_survives_foreign_connection_ids, + hub_survives_unadmitted_senders +); +per_stray!( + client_keeps_its_session, + client_survives_malformed_frames, + client_survives_foreign_connection_ids +); +per_stray!( + server_mode_keeps_its_peer, + server_mode_survives_malformed_frames, + server_mode_survives_foreign_connection_ids +);