From 042dc75c82de77207cefdbb8ad48f86dc59119d9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 19:39:35 +0200 Subject: [PATCH] cultnet-py: a server-mode transport is claimed only by a Connect; a refusing session ends Server mode claimed the endpoint on the first datagram of any type with the right connection id, so an orphan Ack made a later real client a stranger. Only a Connect claims it now. A session that refuses a packet has already recorded its reliable sequence, so the transport ends it: the session forgets the peer (so the goodbye does not acknowledge the refused frame), a Disconnect goes to the peer, disconnect_reason is set, and a server-mode transport forgets its endpoint. CultNetRudpSession gains reset_peer_state and rejects an initial_sequence of 0xFFFFFFFF. Co-Authored-By: Claude Opus 5.5 --- .../cultnet-py/src/cultnet_py/transport.py | 40 +++++++ packages/cultnet-py/tests/test_cultnet.py | 113 +++++++++++++++++- 2 files changed, 152 insertions(+), 1 deletion(-) diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 14c33fcb..3462b047 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -410,6 +410,8 @@ class CultNetRudpSession: raise ValueError("RUDP max_pending_reliable_packets must be greater than zero") self.max_pending_reliable_packets = options.max_pending_reliable_packets self._next_sequence = _uint32(options.initial_sequence, "initial_sequence") + if self._next_sequence == 0xFFFFFFFF: + raise ValueError("RUDP initial_sequence must leave room for a reliable packet") self._next_sequenced_by_channel: dict[str, int] = {} self._next_fragment_id = 1 self._connected = False @@ -453,6 +455,21 @@ class CultNetRudpSession: def outstanding_reliable_packet_count(self) -> int: return len(self._pending_reliable) + len(self._queued_reliable) + def reset_peer_state(self) -> None: + """Forgets everything learned from the peer; sequence numbers already issued stay issued.""" + self._connected = False + self._last_received_at_ms = None + self._highest_received_sequence = None + self._received_sequences.clear() + self._next_sequenced_by_channel.clear() + self._latest_sequenced_by_channel.clear() + self._pending_reliable.clear() + self._queued_reliable.clear() + self._ordered_next_sequence_by_channel.clear() + self._ordered_buffers.clear() + self._fragment_buffers.clear() + self._fragment_sets_evicted = 0 + def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket: self._ensure_reliable_capacity(1) packet = self._create_packet( @@ -997,6 +1014,10 @@ class CultNetRudpSocketTransportConnection: self._packets_dropped += 1 return None if self.remote_addr is None: + # Only a Connect claims the endpoint of a server-mode transport. + if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type != CultNetRudpPacketType.CONNECT: + self._packets_dropped += 1 + return None self.remote_addr = remote_addr elif remote_addr != self.remote_addr: self._packets_dropped += 1 @@ -1014,7 +1035,11 @@ class CultNetRudpSocketTransportConnection: try: result = self.session.receive(packet, _now_ms()) except ValueError: + # receive() has already recorded the packet's reliable sequence, so the + # session cannot be kept: a retransmit would be acknowledged and the + # frame silently lost. End it and tell the peer. self._packets_dropped += 1 + self._end_refused_session() return None if result.reply is not None: self._send_packet(result.reply) @@ -1089,6 +1114,21 @@ class CultNetRudpSocketTransportConnection: self._closed = True self.socket.close() + def _end_refused_session(self) -> None: + reason = b"session refused a packet" + # The goodbye is built after the reset, or its ack field would acknowledge + # the very frame the session refused. + self.session.reset_peer_state() + goodbye = self.session.create_disconnect(reason) + self.disconnect_reason = reason + try: + self._send_packet(goodbye) + except OSError: + pass # Best effort: the session ends whether or not the peer hears it. + if self.mode == CultNetRudpSocketMode.SERVER: + # Only a Connect can claim the endpoint again. + self.remote_addr = None + def _send_packet(self, packet: CultNetRudpPacket) -> None: if self.remote_addr is None: raise ValueError("RUDP socket transport does not have a remote endpoint") diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 549d0803..5c2680ef 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -8,7 +8,7 @@ import unittest import hashlib import hmac import io -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import Callable @@ -977,6 +977,117 @@ class CultNetTests(unittest.TestCase): client.close() server.close() + def test_cultnet_rudp_server_mode_is_claimed_only_by_a_connect(self) -> None: + server_socket = bind_udp_socket() + orphan_socket = bind_udp_socket() + client_socket = bind_udp_socket() + connection_id = 0x10203052 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + client = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-client", + socket=client_socket, + mode=CultNetRudpSocketMode.CLIENT, + remote_addr=server_socket.getsockname(), + connection_id=connection_id, + ) + ) + orphan = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + try: + # A right-id packet that is not a Connect must claim nothing. + orphan_socket.sendto(encode_rudp_packet(orphan.create_ack()), server_socket.getsockname()) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 1) + self.assertIsNone(server.remote_addr) + + client.connect(b"join") + pump_rudp_handshake(client, server) + self.assertTrue(client.connected) + self.assertEqual(server.remote_addr, client_socket.getsockname()) + finally: + orphan_socket.close() + client.close() + server.close() + + def test_cultnet_rudp_socket_transport_ends_the_session_that_refuses_a_reliable_frame(self) -> None: + server_socket = bind_udp_socket() + peer_socket = bind_udp_socket() + connection_id = 0x10203053 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + peer = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + + def to_server(packet) -> None: + peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname()) + + def drain_peer() -> list: + packets = [] + while True: + try: + wire, _ = peer_socket.recvfrom(65535) + except TimeoutError: + return packets + packets.append(decode_rudp_packet(wire)) + + try: + to_server(peer.create_connect(0, b"join")) + server.receive_once() + accept = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0] + peer.receive(accept, 0) + + (frame,) = peer.send_many("schema", b"poison", CultNetRudpSendOptions(reliable=True, ordered=True)) + poison = replace(frame, fragment_count=2, fragment_id=0) + to_server(poison) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 1) + self.assertFalse(server.connected) + self.assertEqual(server.disconnect_reason, b"session refused a packet") + + # The peer is told, and the goodbye does not acknowledge the refused frame. + (goodbye,) = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.DISCONNECT] + self.assertTrue(peer.receive(goodbye, 1).disconnected) + self.assertIn(poison.sequence, peer.pending_reliable_sequences) + + # A retransmit is dropped, not acknowledged. + to_server(poison) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 2) + self.assertEqual(drain_peer(), []) + self.assertIn(poison.sequence, peer.pending_reliable_sequences) + + # A Connect claims the endpoint again. + again = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + to_server(again.create_connect(0, b"again")) + server.receive_once() + self.assertEqual( + [p.packet_type for p in drain_peer()], [CultNetRudpPacketType.ACCEPT] + ) + finally: + peer_socket.close() + server.close() + + def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None: + with self.assertRaises(ValueError): + CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF)) + with self.assertRaises(ValueError): + CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, max_pending_reliable_packets=0)) + CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFE, max_pending_reliable_packets=1) + ) + def test_cultnet_rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames(self) -> None: server_socket = bind_udp_socket() client_socket = bind_udp_socket()