cultnet-py: RUDP socket transport drops and counts strays

A malformed frame, a foreign connection id or a refused packet raised out of
receive_once. They are now counted in stats.packets_dropped and receive_once
returns None. The id is checked before the peer endpoint is claimed, so a
stray arriving first can no longer become the server-mode peer.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

packages/cultnet-py/src/cultnet_py/transport.py

@@ -23,6 +23,10 @@ class CultNetTransportStats:
     bytes_sent: int = 0
     frames_received: int = 0
     frames_sent: int = 0
+    # Datagrams read and discarded because they belong to no session on this
+    # transport: malformed frames, another session's connection id, a sender
+    # that is not the peer, or a packet the session refuses.
+    packets_dropped: int = 0
 
 
 @dataclass(frozen=True)
@@ -921,6 +925,7 @@ class CultNetRudpSocketTransportConnection:
         self._bytes_sent = 0
         self._frames_received = 0
         self._frames_sent = 0
+        self._packets_dropped = 0
         self._delivered_frames: deque[CultNetTransportFrame] = deque()
         self._closed = False
         self.disconnect_reason: bytes | None = None
@@ -943,6 +948,7 @@ class CultNetRudpSocketTransportConnection:
             bytes_sent=self._bytes_sent,
             frames_received=self._frames_received,
             frames_sent=self._frames_sent,
+            packets_dropped=self._packets_dropped,
         )
 
     @property
@@ -977,17 +983,39 @@ class CultNetRudpSocketTransportConnection:
             return None
         self._bytes_received += len(wire)
 
+        # What a datagram carries is the sender's business, not a fault of this
+        # process: a malformed frame, another session's connection id, a sender
+        # that is not the 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 is claimed, so a stray cannot become the peer.
+        try:
+            packet = decode_rudp_packet(wire)
+        except ValueError:
+            self._packets_dropped += 1
+            return None
+        if packet.connection_id != self.session.connection_id:
+            self._packets_dropped += 1
+            return None
         if self.remote_addr is None:
             self.remote_addr = remote_addr
         elif remote_addr != self.remote_addr:
+            self._packets_dropped += 1
             return None
 
-        packet = decode_rudp_packet(wire)
         if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type == CultNetRudpPacketType.CONNECT:
-            self._send_packet(self.session.accept_connect(packet, _now_ms()))
+            try:
+                accept = self.session.accept_connect(packet, _now_ms())
+            except ValueError:
+                self._packets_dropped += 1
+                return None
+            self._send_packet(accept)
             return None
 
-        result = self.session.receive(packet, _now_ms())
+        try:
+            result = self.session.receive(packet, _now_ms())
+        except ValueError:
+            self._packets_dropped += 1
+            return None
         if result.reply is not None:
             self._send_packet(result.reply)
         for ready in result.ready_to_send:

packages/cultnet-py/tests/test_cultnet.py

@@ -918,6 +918,65 @@ class CultNetTests(unittest.TestCase):
             client.close()
             server.close()
 
+    def test_cultnet_rudp_socket_transport_drops_strays_and_keeps_serving_its_peer(self) -> None:
+        server_socket = bind_udp_socket()
+        client_socket = bind_udp_socket()
+        stray_socket = bind_udp_socket()
+        connection_id = 0x10203050
+        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,
+            )
+        )
+        foreign = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=0x0BADF10E))
+        strays = [
+            b"not a rudp packet",
+            b"",
+            encode_rudp_packet(foreign.create_connect(0, b"foreign")),
+        ]
+
+        try:
+            # A stray that arrives before the peer must not become the peer.
+            for stray in strays:
+                stray_socket.sendto(stray, server_socket.getsockname())
+            for _ in strays:
+                self.assertIsNone(server.receive_once())
+            self.assertEqual(server.stats.packets_dropped, len(strays))
+
+            client.connect(b"join")
+            pump_rudp_handshake(client, server)
+
+            # A moved flow answers from the peer's own address with someone else's session.
+            for stray in strays:
+                server_socket.sendto(stray, client_socket.getsockname())
+                client_socket.sendto(stray, server_socket.getsockname())
+            for _ in strays:
+                self.assertIsNone(client.receive_once())
+                self.assertIsNone(server.receive_once())
+            self.assertEqual(client.stats.packets_dropped, len(strays))
+            self.assertEqual(server.stats.packets_dropped, 2 * len(strays))
+
+            client.send("schema", b"still here")
+            self.assertEqual(receive_rudp_frame(server).payload, b"still here")
+            server.send("schema", b"server still here")
+            self.assertEqual(receive_rudp_frame(client).payload, b"server still here")
+        finally:
+            stray_socket.close()
+            client.close()
+            server.close()
+
     def test_cultnet_rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames(self) -> None:
         server_socket = bind_udp_socket()
         client_socket = bind_udp_socket()