From 4cd0fec0497dd00b4d5e158367369f8f7143e1c6 Mon Sep 17 00:00:00 2001 From: Ednunp Date: Fri, 15 May 2026 17:24:20 +0100 Subject: [PATCH] =?UTF-8?q?server:=20update=20to=20v2.3=20=E2=80=94=20dual?= =?UTF-8?q?-protocol=20relay=20(v1=20pair=20+=20v2=20lobby)=20+=20auto-upd?= =?UTF-8?q?ater=20with=20EXIT=20trap=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/remsound-relay.py | 464 ++++++++++++++++++++++++++++++--------- 1 file changed, 355 insertions(+), 109 deletions(-) diff --git a/server/remsound-relay.py b/server/remsound-relay.py index 8081ad1..796d0f9 100644 --- a/server/remsound-relay.py +++ b/server/remsound-relay.py @@ -1,19 +1,28 @@ #!/usr/bin/env python3 """ -RemSound UDP relay. +RemSound UDP relay, dual-protocol. -Listens on a single UDP port and reflects RemSound packets between up to two -peer endpoints. Validates the 12-byte RemSound header (magic 'RMND', version 1) -and silently drops anything else. Never decodes audio. +Listens on a single UDP port and handles two protocol versions concurrently: -Pairing model: pragmatic v1. The relay does not key on stream id (two RemSound -instances will use different per-sender stream ids, so keying on it would -prevent pairing). Instead the first two distinct UDP endpoints to send a valid -RemSound packet claim the two peer slots; subsequent valid packets from a -slot's endpoint are reflected to the other slot. Slots that go silent for more -than IDLE_TIMEOUT_SECONDS are replaced when a fresh endpoint arrives. +- v1 ("pairwise"): 12-byte header, two-slot reflector. First two distinct + UDP endpoints to send a valid RemSound v1 packet claim the slots; subsequent + v1 packets from one slot's endpoint are reflected to the other. Slots idle + for IDLE_TIMEOUT_SECONDS are eligible for replacement. This is the original + remsound-relay.py behaviour, preserved here unchanged so legacy clients keep + working against the new server. -Spec and operational notes: see README.md alongside this script. +- v2 ("lobby"): 28-byte header with embedded CLIENT_ID (UUID). Up to + REMSOUND_MAX_CLIENTS instances (default 10) form a single lobby. Each + incoming packet is forwarded unmodified to every OTHER registered client. + Identity is the CLIENT_ID, not the network endpoint — NAT rebinds and + same-NAT-multiple-clients are no longer special cases. Periodic LobbyRoster + packets keep clients informed of the current membership. + +The two protocols share state only via the listening socket and the stats +counters. They never interact otherwise: a v1 client and a v2 client cannot +hear each other in this release (deliberate — see the design doc). + +Owner: Pi thread. Spec: D:\\Dropbox\\proj\\pi\\remsound server update.md. """ from __future__ import annotations @@ -21,13 +30,15 @@ from __future__ import annotations import argparse import logging import logging.handlers +import os import select import signal import socket import struct import sys import time -from dataclasses import dataclass +import uuid +from dataclasses import dataclass, field from typing import Optional LISTEN_HOST = "0.0.0.0" @@ -35,39 +46,68 @@ DEFAULT_PORT = 47830 RECV_BUFFER_BYTES = 2048 IDLE_TIMEOUT_SECONDS = 60 STATS_INTERVAL_SECONDS = 60 +ROSTER_HEARTBEAT_SECONDS = 1.0 # v2 only — periodic roster broadcast SOCKET_POLL_TIMEOUT_SECONDS = 1.0 DEFAULT_LOG_PATH = "/var/log/remsound-relay.log" +DEFAULT_MAX_CLIENTS = 10 +LOBBY_NAME_BYTES = 32 # bytes reserved for a display name on the wire -# Wire format constants — little-endian, see RemSound.Core.RemPacket. -HEADER_LEN = 12 +# Wire format constants. MAGIC = b"RMND" -VERSION = 1 +V1_VERSION = 1 +V2_VERSION = 2 +V1_HEADER_LEN = 12 +V2_HEADER_LEN = 28 +V2_CLIENT_ID_OFFSET = 12 +V2_CLIENT_ID_LEN = 16 + +# Packet types (v1 + v2 shared range; v2-only types are 6+). TYPE_FORMAT = 1 TYPE_AUDIO = 2 TYPE_KEEPALIVE = 3 TYPE_HEARTBEAT = 4 -PACKET_TYPE_NAMES = { - TYPE_FORMAT: "Format", - TYPE_AUDIO: "Audio", - TYPE_KEEPALIVE: "KeepAlive", - TYPE_HEARTBEAT: "Heartbeat", +TYPE_CONTROL = 5 +TYPE_LOBBY_HELLO = 6 +TYPE_LOBBY_ROSTER = 7 +TYPE_LOBBY_FULL = 8 +TYPE_LOBBY_BYE = 9 +V2_FORWARDABLE_TYPES = { + TYPE_FORMAT, TYPE_AUDIO, TYPE_KEEPALIVE, TYPE_HEARTBEAT, TYPE_CONTROL, } +# A zero UUID identifies the server in outbound v2 packets that we originate +# (LobbyRoster, LobbyFull, LobbyBye-from-server). Clients can recognise this +# as "from server" rather than from another peer. +SERVER_CLIENT_ID_BYTES = b"\x00" * V2_CLIENT_ID_LEN + @dataclass class PeerSlot: + """v1 protocol — one of (up to) two peer endpoints in a pair.""" addr: tuple[str, int] last_seen: float rx_packets: int = 0 tx_packets: int = 0 +@dataclass +class ClientEntry: + """v2 protocol — one client in the lobby, keyed by CLIENT_ID.""" + addr: tuple[str, int] + display_name: str + last_seen: float + rx_packets: int = 0 + tx_packets: int = 0 + + @dataclass class RelayStats: forwarded: int = 0 - dropped_unpaired: int = 0 + dropped_unpaired: int = 0 # v1: third endpoint while pair active + dropped_lobby_full: int = 0 # v2: 11th client when at cap rejected_bad_header: int = 0 - pair_changes: int = 0 + pair_changes: int = 0 # v1 slot joins/leaves/replacements + lobby_changes: int = 0 # v2 joins/leaves/expiries def setup_logger(log_path: str) -> logging.Logger: @@ -89,172 +129,379 @@ def setup_logger(log_path: str) -> logging.Logger: return logger -def parse_header(data: bytes) -> Optional[tuple[int, int, int, int]]: - """Validate the RemSound header. Returns (version, type, stream_id, sequence) or None.""" - if len(data) < HEADER_LEN: - return None - if data[0:4] != MAGIC: - return None - version = data[4] - if version != VERSION: +def parse_header_v1(data: bytes) -> Optional[tuple[int, int, int]]: + """Validate a v1 header. Returns (type, stream_id, sequence) or None.""" + if len(data) < V1_HEADER_LEN: return None pkt_type = data[5] stream_id = struct.unpack_from(" Optional[tuple[int, int, int, bytes]]: + """Validate a v2 header. Returns (type, stream_id, sequence, client_id_bytes) or None.""" + if len(data) < V2_HEADER_LEN: + return None + pkt_type = data[5] + stream_id = struct.unpack_from(" str: + return f"{addr[0]}:{addr[1]}" + + +def _decode_lobby_name(raw: bytes) -> str: + """Decode the 32-byte null-padded UTF-8 display-name field. Tolerant of garbage.""" + end = raw.find(b"\x00") + if end >= 0: + raw = raw[:end] + try: + return raw.decode("utf-8", errors="replace").strip() + except Exception: + return "" + + +def _encode_lobby_name(name: str) -> bytes: + """Encode a display name into LOBBY_NAME_BYTES, null-padded.""" + encoded = (name or "").encode("utf-8", errors="replace")[:LOBBY_NAME_BYTES] + return encoded + b"\x00" * (LOBBY_NAME_BYTES - len(encoded)) class Relay: - def __init__(self, sock: socket.socket, log: logging.Logger): + """Dispatcher that owns both the v1 pair state and the v2 lobby state.""" + + def __init__(self, sock: socket.socket, log: logging.Logger, max_clients: int): self.sock = sock self.log = log - self.peers: list[PeerSlot] = [] + self.max_clients = max_clients + # v1 state + self.v1_peers: list[PeerSlot] = [] + # v2 state + self.v2_clients: dict[uuid.UUID, ClientEntry] = {} + self.v2_roster_dirty = False # set when membership changes + self.v2_last_roster_broadcast = 0.0 + # shared self.stats = RelayStats() self.last_stats_log = time.monotonic() - @staticmethod - def _fmt_addr(addr: tuple[str, int]) -> str: - return f"{addr[0]}:{addr[1]}" + # ------- v1 (pairwise) ------------------------------------------------- - def find_slot(self, addr: tuple[str, int]) -> Optional[int]: - for i, p in enumerate(self.peers): + def _v1_find_slot(self, addr: tuple[str, int]) -> Optional[int]: + for i, p in enumerate(self.v1_peers): if p.addr == addr: return i return None - def expire_idle(self, now: float) -> None: - if not self.peers: + def _v1_expire_idle(self, now: float) -> None: + if not self.v1_peers: return kept: list[PeerSlot] = [] dropped: list[tuple[str, int]] = [] - for p in self.peers: + for p in self.v1_peers: if (now - p.last_seen) <= IDLE_TIMEOUT_SECONDS: kept.append(p) else: dropped.append(p.addr) if dropped: - self.peers = kept + self.v1_peers = kept for addr in dropped: self.log.info( "event=peer_dropped reason=idle addr=%s remaining=%d", - self._fmt_addr(addr), - len(self.peers), + _fmt_addr(addr), len(self.v1_peers), ) self.stats.pair_changes += 1 - def admit_or_replace(self, addr: tuple[str, int], now: float) -> int: - if len(self.peers) < 2: - self.peers.append(PeerSlot(addr=addr, last_seen=now)) + def _v1_admit_or_replace(self, addr: tuple[str, int], now: float) -> int: + if len(self.v1_peers) < 2: + self.v1_peers.append(PeerSlot(addr=addr, last_seen=now)) self.log.info( "event=peer_joined addr=%s slots_filled=%d", - self._fmt_addr(addr), - len(self.peers), + _fmt_addr(addr), len(self.v1_peers), ) self.stats.pair_changes += 1 - if len(self.peers) == 2: + if len(self.v1_peers) == 2: self.log.info( "event=peer_paired a=%s b=%s", - self._fmt_addr(self.peers[0].addr), - self._fmt_addr(self.peers[1].addr), + _fmt_addr(self.v1_peers[0].addr), + _fmt_addr(self.v1_peers[1].addr), ) - return len(self.peers) - 1 - # Both slots occupied. Replace the stalest one if it has been idle. - oldest = 0 if self.peers[0].last_seen <= self.peers[1].last_seen else 1 - if (now - self.peers[oldest].last_seen) > IDLE_TIMEOUT_SECONDS: - old_addr = self.peers[oldest].addr - self.peers[oldest] = PeerSlot(addr=addr, last_seen=now) + return len(self.v1_peers) - 1 + oldest = 0 if self.v1_peers[0].last_seen <= self.v1_peers[1].last_seen else 1 + if (now - self.v1_peers[oldest].last_seen) > IDLE_TIMEOUT_SECONDS: + old_addr = self.v1_peers[oldest].addr + self.v1_peers[oldest] = PeerSlot(addr=addr, last_seen=now) self.log.info( "event=peer_replaced old=%s new=%s", - self._fmt_addr(old_addr), - self._fmt_addr(addr), + _fmt_addr(old_addr), _fmt_addr(addr), ) self.stats.pair_changes += 1 return oldest - return -1 # Both slots active — packet from a third endpoint is ignored. + return -1 - def handle_packet(self, data: bytes, addr: tuple[str, int]) -> None: - parsed = parse_header(data) - if parsed is None: + def _handle_v1(self, data: bytes, addr: tuple[str, int]) -> None: + if parse_header_v1(data) is None: self.stats.rejected_bad_header += 1 return - # We do not log per-packet detail (would flood the log). Stats covers it. - _version, _pkt_type, _stream_id, _sequence = parsed - now = time.monotonic() - idx = self.find_slot(addr) + idx = self._v1_find_slot(addr) if idx is None: - # Take the chance to age out idle slots first. - self.expire_idle(now) - idx = self.admit_or_replace(addr, now) + self._v1_expire_idle(now) + idx = self._v1_admit_or_replace(addr, now) if idx < 0: self.stats.dropped_unpaired += 1 return - - peer = self.peers[idx] + peer = self.v1_peers[idx] peer.last_seen = now peer.rx_packets += 1 - - if len(self.peers) == 2: - other = self.peers[1 - idx] + if len(self.v1_peers) == 2: + other = self.v1_peers[1 - idx] try: self.sock.sendto(data, other.addr) other.tx_packets += 1 self.stats.forwarded += 1 except OSError as e: self.log.warning( - "event=send_failed to=%s err=%s", - self._fmt_addr(other.addr), - e, + "event=send_failed proto=v1 to=%s err=%s", + _fmt_addr(other.addr), e, ) else: self.stats.dropped_unpaired += 1 + # ------- v2 (lobby) ---------------------------------------------------- + + def _v2_build_roster_packet(self) -> bytes: + """Build a LobbyRoster packet with the current membership.""" + # Use a separate per-build sequence — clients can ignore it; we use 0. + header = bytearray(V2_HEADER_LEN) + header[0:4] = MAGIC + header[4] = V2_VERSION + header[5] = TYPE_LOBBY_ROSTER + struct.pack_into(" None: + if not self.v2_clients: + self.v2_roster_dirty = False + self.v2_last_roster_broadcast = time.monotonic() + return + packet = self._v2_build_roster_packet() + for entry in self.v2_clients.values(): + try: + self.sock.sendto(packet, entry.addr) + except OSError as e: + self.log.warning( + "event=send_failed proto=v2 reason=roster to=%s err=%s", + _fmt_addr(entry.addr), e, + ) + self.v2_roster_dirty = False + self.v2_last_roster_broadcast = time.monotonic() + + def _v2_send_lobby_full(self, attempted_client_id: uuid.UUID, addr: tuple[str, int]) -> None: + """Send a LobbyFull packet back to an over-cap client and log it.""" + header = bytearray(V2_HEADER_LEN) + header[0:4] = MAGIC + header[4] = V2_VERSION + header[5] = TYPE_LOBBY_FULL + struct.pack_into(" None: + if not self.v2_clients: + return + expired: list[uuid.UUID] = [] + for cid, entry in self.v2_clients.items(): + if (now - entry.last_seen) > IDLE_TIMEOUT_SECONDS: + expired.append(cid) + for cid in expired: + entry = self.v2_clients.pop(cid) + self.log.info( + "event=client_idle_expired client_id=%s addr=%s", + cid, _fmt_addr(entry.addr), + ) + self.stats.lobby_changes += 1 + self.v2_roster_dirty = True + + def _handle_v2(self, data: bytes, addr: tuple[str, int]) -> None: + parsed = parse_header_v2(data) + if parsed is None: + self.stats.rejected_bad_header += 1 + return + pkt_type, _stream_id, _sequence, cid_bytes = parsed + try: + client_id = uuid.UUID(bytes=cid_bytes) + except ValueError: + self.stats.rejected_bad_header += 1 + return + now = time.monotonic() + entry = self.v2_clients.get(client_id) + if entry is None: + # Admit attempt. + if len(self.v2_clients) >= self.max_clients: + self._v2_send_lobby_full(client_id, addr) + return + entry = ClientEntry(addr=addr, display_name="", last_seen=now) + self.v2_clients[client_id] = entry + self.log.info( + "event=client_joined client_id=%s addr=%s count=%d", + client_id, _fmt_addr(addr), len(self.v2_clients), + ) + self.stats.lobby_changes += 1 + self.v2_roster_dirty = True + else: + # Refresh endpoint (handles NAT rebind) and last-seen. + if entry.addr != addr: + self.log.info( + "event=client_endpoint_update client_id=%s old=%s new=%s", + client_id, _fmt_addr(entry.addr), _fmt_addr(addr), + ) + entry.addr = addr + entry.last_seen = now + entry.rx_packets += 1 + + # Type-specific handling. + if pkt_type == TYPE_LOBBY_HELLO: + payload = data[V2_HEADER_LEN:V2_HEADER_LEN + LOBBY_NAME_BYTES] + new_name = _decode_lobby_name(payload) + if new_name != entry.display_name: + entry.display_name = new_name + self.log.info( + "event=client_named client_id=%s name=%r", client_id, new_name, + ) + self.v2_roster_dirty = True + return + if pkt_type == TYPE_LOBBY_BYE: + self.v2_clients.pop(client_id, None) + self.log.info( + "event=client_left client_id=%s addr=%s reason=bye", + client_id, _fmt_addr(addr), + ) + self.stats.lobby_changes += 1 + self.v2_roster_dirty = True + return + if pkt_type not in V2_FORWARDABLE_TYPES: + # Unknown / server-originated type from a client. Ignore quietly. + return + + # Fan-out forwarding to every OTHER client. + for other_id, other in self.v2_clients.items(): + if other_id == client_id: + continue + try: + self.sock.sendto(data, other.addr) + other.tx_packets += 1 + self.stats.forwarded += 1 + except OSError as e: + self.log.warning( + "event=send_failed proto=v2 to=%s err=%s", + _fmt_addr(other.addr), e, + ) + + # ------- shared -------------------------------------------------------- + + def handle_packet(self, data: bytes, addr: tuple[str, int]) -> None: + if len(data) < 6 or data[0:4] != MAGIC: + self.stats.rejected_bad_header += 1 + return + version = data[4] + if version == V1_VERSION: + self._handle_v1(data, addr) + elif version == V2_VERSION: + self._handle_v2(data, addr) + else: + self.stats.rejected_bad_header += 1 + + def tick(self, now: float) -> None: + """Periodic housekeeping: idle expiry + roster broadcast.""" + self._v1_expire_idle(now) + self._v2_expire_idle(now) + if self.v2_clients and ( + self.v2_roster_dirty + or (now - self.v2_last_roster_broadcast) >= ROSTER_HEARTBEAT_SECONDS + ): + self._v2_broadcast_roster() + def maybe_log_stats(self, now: float) -> None: if (now - self.last_stats_log) < STATS_INTERVAL_SECONDS: return self.last_stats_log = now s = self.stats - peers_summary = ", ".join( - f"{self._fmt_addr(p.addr)}(rx={p.rx_packets},tx={p.tx_packets})" - for p in self.peers + v1_summary = ", ".join( + f"{_fmt_addr(p.addr)}(rx={p.rx_packets},tx={p.tx_packets})" + for p in self.v1_peers + ) or "none" + v2_summary = ", ".join( + f"{cid}@{_fmt_addr(e.addr)}(rx={e.rx_packets},tx={e.tx_packets})" + for cid, e in self.v2_clients.items() ) or "none" self.log.info( - "event=stats forwarded=%d dropped_unpaired=%d rejected_bad_header=%d pair_changes=%d peers=[%s]", - s.forwarded, - s.dropped_unpaired, - s.rejected_bad_header, - s.pair_changes, - peers_summary, + "event=stats forwarded=%d dropped_unpaired=%d dropped_lobby_full=%d " + "rejected_bad_header=%d pair_changes=%d lobby_changes=%d " + "client_count=%d v1_peers=[%s] v2_clients=[%s]", + s.forwarded, s.dropped_unpaired, s.dropped_lobby_full, + s.rejected_bad_header, s.pair_changes, s.lobby_changes, + len(self.v2_clients), v1_summary, v2_summary, ) - # Reset counters so the next stats line shows a per-minute rate. self.stats = RelayStats() - for p in self.peers: + for p in self.v1_peers: p.rx_packets = 0 p.tx_packets = 0 + for e in self.v2_clients.values(): + e.rx_packets = 0 + e.tx_packets = 0 def main() -> int: - parser = argparse.ArgumentParser(description="RemSound UDP relay") + parser = argparse.ArgumentParser(description="RemSound UDP relay (dual-protocol v1+v2)") + parser.add_argument("--port", type=int, default=DEFAULT_PORT, + help=f"UDP port to listen on (default {DEFAULT_PORT})") + parser.add_argument("--host", default=LISTEN_HOST, + help=f"Bind address (default {LISTEN_HOST})") + parser.add_argument("--log-path", default=DEFAULT_LOG_PATH, + help=f"Log file path (default {DEFAULT_LOG_PATH})") parser.add_argument( - "--port", - type=int, - default=DEFAULT_PORT, - help=f"UDP port to listen on (default {DEFAULT_PORT})", - ) - parser.add_argument( - "--host", - default=LISTEN_HOST, - help=f"Bind address (default {LISTEN_HOST})", - ) - parser.add_argument( - "--log-path", - default=DEFAULT_LOG_PATH, - help=f"Log file path (default {DEFAULT_LOG_PATH})", + "--max-clients", type=int, + default=int(os.environ.get("REMSOUND_MAX_CLIENTS", str(DEFAULT_MAX_CLIENTS))), + help=f"v2 lobby capacity (default {DEFAULT_MAX_CLIENTS}, " + "overridable via REMSOUND_MAX_CLIENTS env var)", ) args = parser.parse_args() + if args.max_clients < 2: + sys.stderr.write("remsound-relay: --max-clients must be >= 2\n") + return 2 log = setup_logger(args.log_path) - log.info("event=startup version=1 listen=%s:%d", args.host, args.port) + log.info( + "event=startup version_supported=v1,v2 listen=%s:%d max_clients=%d", + args.host, args.port, args.max_clients, + ) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -264,8 +511,7 @@ def main() -> int: log.error("event=bind_failed err=%s", e) return 1 - relay = Relay(sock, log) - + relay = Relay(sock, log, args.max_clients) stop_flag = {"stop": False} def _stop_signal(_signum, _frame): @@ -288,7 +534,7 @@ def main() -> int: log.warning("event=recv_failed err=%s", e) continue relay.handle_packet(data, addr) - relay.expire_idle(now) + relay.tick(now) relay.maybe_log_stats(now) finally: log.info("event=shutdown")