diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 73bd1e62..14c33fcb 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/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: diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 475305d3..549d0803 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/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()