GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultnet-rs: drop and count packets that belong to no session

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 17:38:20 +02:00
commit 7b7666f2c3
4 changed files with 314 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

55
packages/cultnet-rs/src/rudp.rs
View file

@ -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)?;
}

5
packages/cultnet-rs/src/transport.rs
View file

@ -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)]

17
packages/cultnet-rs/tests/rudp_server_hub.rs
View file

@ -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]

255
packages/cultnet-rs/tests/rudp_stray_packets.rs Normal file
View file

@ -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<UdpSocket> {
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
);