Talon/RemSound
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
05b218825d01f60e9b4cbf9c0aabd95c947d309d
RemSound/server/remsound-relay.py
T
EdnunpandClaude Opus 4.8 84c4a47411 v3.9.1: plain/WASAPI streams now play on an ASIO-mode receiver (silent-mic fix) 
Receiver: a Mixed (plain) session is rendered on an active lane in BothIndependent mode instead of being skipped, so a WASAPI-only sender is no longer silent to a receiver that has an ASIO driver selected. Also port the per-session buffer depth-drain (stops the receive jitter buffer bloating).

Sender: add sndAudFr meter (audio frames actually sent) to localise capture vs send.

App: startup sound cue (machine-wide, Preferences); stop sending audio when no peer is reachable (issue #8). Version 3.9.1.

Server (relay): fix updater version-compare for multi-dot tags, guard the main loop against crashes, reject spoofed BYE from a mismatched endpoint.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-12 19:00:33 +01:00

574 lines
22 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
RemSound UDP relay, dual-protocol.
Listens on a single UDP port and handles two protocol versions concurrently:
- 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.
- 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
import argparse
import logging
import logging.handlers
import os
import select
import signal
import socket
import struct
import sys
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional
LISTEN_HOST = "0.0.0.0"
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.
MAGIC = b"RMND"
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
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 # 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 # v1 slot joins/leaves/replacements
lobby_changes: int = 0 # v2 joins/leaves/expiries
def setup_logger(log_path: str) -> logging.Logger:
logger = logging.getLogger("remsound-relay")
logger.setLevel(logging.INFO)
fmt = logging.Formatter(
fmt="%(asctime)s level=%(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
try:
fh = logging.handlers.WatchedFileHandler(log_path, encoding="utf-8")
fh.setFormatter(fmt)
logger.addHandler(fh)
except OSError as e:
sys.stderr.write(f"remsound-relay: could not open {log_path}: {e}\n")
sh = logging.StreamHandler(sys.stderr)
sh.setFormatter(fmt)
logger.addHandler(sh)
return logger
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("<H", data, 6)[0]
sequence = struct.unpack_from("<I", data, 8)[0]
return pkt_type, stream_id, sequence
def parse_header_v2(data: bytes) -> 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("<H", data, 6)[0]
sequence = struct.unpack_from("<I", data, 8)[0]
client_id_bytes = bytes(data[V2_CLIENT_ID_OFFSET:V2_CLIENT_ID_OFFSET + V2_CLIENT_ID_LEN])
return pkt_type, stream_id, sequence, client_id_bytes
def _fmt_addr(addr: tuple[str, int]) -> 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:
"""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.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()
# ------- v1 (pairwise) -------------------------------------------------
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 _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.v1_peers:
if (now - p.last_seen) <= IDLE_TIMEOUT_SECONDS:
kept.append(p)
else:
dropped.append(p.addr)
if dropped:
self.v1_peers = kept
for addr in dropped:
self.log.info(
"event=peer_dropped reason=idle addr=%s remaining=%d",
_fmt_addr(addr), len(self.v1_peers),
)
self.stats.pair_changes += 1
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",
_fmt_addr(addr), len(self.v1_peers),
)
self.stats.pair_changes += 1
if len(self.v1_peers) == 2:
self.log.info(
"event=peer_paired a=%s b=%s",
_fmt_addr(self.v1_peers[0].addr),
_fmt_addr(self.v1_peers[1].addr),
)
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",
_fmt_addr(old_addr), _fmt_addr(addr),
)
self.stats.pair_changes += 1
return oldest
return -1
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
now = time.monotonic()
idx = self._v1_find_slot(addr)
if idx is None:
self._v1_expire_idle(now)
idx = self._v1_admit_or_replace(addr, now)
if idx < 0:
self.stats.dropped_unpaired += 1
return
peer = self.v1_peers[idx]
peer.last_seen = now
peer.rx_packets += 1
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 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("<H", header, 6, 0) # stream_id (unused)
struct.pack_into("<I", header, 8, 0) # sequence (unused)
header[V2_CLIENT_ID_OFFSET:V2_CLIENT_ID_OFFSET + V2_CLIENT_ID_LEN] = SERVER_CLIENT_ID_BYTES
payload = bytearray()
members = list(self.v2_clients.items())[:255] # 1-byte count
payload.append(len(members))
for cid, entry in members:
payload.extend(cid.bytes)
payload.extend(_encode_lobby_name(entry.display_name))
return bytes(header) + bytes(payload)
def _v2_broadcast_roster(self) -> 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("<H", header, 6, 0)
struct.pack_into("<I", header, 8, 0)
header[V2_CLIENT_ID_OFFSET:V2_CLIENT_ID_OFFSET + V2_CLIENT_ID_LEN] = SERVER_CLIENT_ID_BYTES
# Payload: 1 byte current count, 1 byte max count.
payload = bytes([len(self.v2_clients) & 0xFF, self.max_clients & 0xFF])
try:
self.sock.sendto(bytes(header) + payload, addr)
except OSError as e:
self.log.warning(
"event=send_failed proto=v2 reason=lobby_full to=%s err=%s",
_fmt_addr(addr), e,
)
self.log.info(
"event=lobby_full attempted_client_id=%s addr=%s count=%d max=%d",
attempted_client_id, _fmt_addr(addr),
len(self.v2_clients), self.max_clients,
)
self.stats.dropped_lobby_full += 1
def _v2_expire_idle(self, now: float) -> 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)
# Capture whether this packet came from the endpoint this client_id is CURRENTLY registered
# at, BEFORE the NAT-rebind update below overwrites entry.addr. Used to reject spoofed
# control packets: the roster broadcast ships every member's client_id to all members, so on
# an internet-facing relay anyone who joins learns the others' ids and could otherwise forge
# a BYE to evict them. A genuine BYE always comes from the client's own registered endpoint.
from_registered_endpoint = entry is not None and entry.addr == addr
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:
# Only the endpoint a client is registered at may say goodbye for it — otherwise a
# forged BYE bearing a known client_id (learned from the roster) could evict any peer.
if not from_registered_endpoint:
self.log.warning(
"event=bye_rejected reason=endpoint_mismatch client_id=%s from=%s",
client_id, _fmt_addr(addr),
)
return
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
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 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,
)
self.stats = RelayStats()
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 (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(
"--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_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)
try:
sock.bind((args.host, args.port))
except OSError as e:
log.error("event=bind_failed err=%s", e)
return 1
relay = Relay(sock, log, args.max_clients)
stop_flag = {"stop": False}
def _stop_signal(_signum, _frame):
stop_flag["stop"] = True
signal.signal(signal.SIGTERM, _stop_signal)
signal.signal(signal.SIGINT, _stop_signal)
try:
while not stop_flag["stop"]:
try:
ready, _, _ = select.select([sock], [], [], SOCKET_POLL_TIMEOUT_SECONDS)
except InterruptedError:
continue
except OSError as e:
# select() itself failed (e.g. a transient resource-pressure error on a long-
# running, low-RAM host). Log and pause briefly rather than spin or exit.
log.warning("event=select_failed err=%s", e)
time.sleep(0.1)
continue
now = time.monotonic()
# Per-iteration work, fully guarded. A relay that must stay up for DAYS — and that is
# reachable from the open internet — can never let a single packet or a housekeeping
# tick crash the whole process: that would drop EVERY connected client and force a ~5s
# systemd restart. Anything unexpected is logged (with a traceback) and we carry on.
try:
if ready:
data, addr = sock.recvfrom(RECV_BUFFER_BYTES)
relay.handle_packet(data, addr)
relay.tick(now)
relay.maybe_log_stats(now)
except OSError as e:
# recvfrom, or a sendto that escaped its own guard — transient; keep serving.
log.warning("event=io_error err=%s", e)
except Exception:
log.exception("event=loop_error — recovered, continuing")
finally:
log.info("event=shutdown")
sock.close()
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink