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 +);