"""Black-box UDP integration harness for FreeDMR. This module is test-only. It starts bridge_master.py with a generated loopback configuration and interacts with it over UDP as emulated HBP repeaters. """ from __future__ import annotations from dataclasses import dataclass, field from hashlib import blake2b, sha1 from hmac import new as hmac_new from pathlib import Path import os import random import socket import subprocess import sys import tempfile import threading import time import unittest import venv from tests.harness.deterministic import ( HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, HBPF_VOICE, PacketSpec, bytes_3, bytes_4, parse_dmr_fields, ) RPTL = b"RPTL" RPTACK = b"RPTACK" RPTK = b"RPTK" RPTC = b"RPTC" RPTPING = b"RPTPING" MSTPONG = b"MSTPONG" DMRD = b"DMRD" DMRE = b"DMRE" BCKA = b"BCKA" BCSQ = b"BCSQ" BCST = b"BCST" BCVE = b"BCVE" FBP_VERSION = 5 FBP_PASSPHRASE = b"test-passphrase".ljust(20, b"\x00")[:20] REQUIRED_RUNTIME_MODULES = ("bitarray", "twisted", "setproctitle") PROXY_RUNTIME_MODULES = ("Pyro5",) @dataclass(frozen=True) class LinkImpairment: """Deterministic test-only impairment for packets sent by fake UDP peers.""" drop_rate: float = 0.0 duplicate_rate: float = 0.0 delay_range: tuple[float, float] = (0.0, 0.0) jitter_range: tuple[float, float] = (0.0, 0.0) drop_indices: frozenset[int] = frozenset() duplicate_indices: frozenset[int] = frozenset() delay_by_index: dict[int, float] = field(default_factory=dict) seed: int = 1 def schedule( self, packets: list[bytes], *, cadence_seconds: float | None = None, ) -> list[tuple[float, bytes]]: rng = random.Random(self.seed) scheduled = [] for index, packet in enumerate(packets): if index in self.drop_indices or rng.random() < self.drop_rate: continue base_time = index * (cadence_seconds or 0.0) jitter = rng.uniform(*self.jitter_range) delay = self.delay_by_index.get(index, rng.uniform(*self.delay_range)) send_at = max(0.0, base_time + jitter + delay) scheduled.append((send_at, packet)) if index in self.duplicate_indices or rng.random() < self.duplicate_rate: scheduled.append((send_at + 0.001, packet)) return sorted(scheduled, key=lambda item: item[0]) class ImpairmentProfiles: @staticmethod def clean() -> LinkImpairment: return LinkImpairment() @staticmethod def provider_vxlan_reorder(delayed_index: int = 1, delay: float = 0.08) -> LinkImpairment: return LinkImpairment(delay_by_index={delayed_index: delay}) @staticmethod def mobile_flutter(*drop_indices: int) -> LinkImpairment: return LinkImpairment(drop_indices=frozenset(drop_indices)) @staticmethod def duplicate_udp(*duplicate_indices: int) -> LinkImpairment: return LinkImpairment(duplicate_indices=frozenset(duplicate_indices)) @staticmethod def burst_loss(start: int, count: int) -> LinkImpairment: return LinkImpairment(drop_indices=frozenset(range(start, start + count))) @dataclass(frozen=True) class StreamProfile: packets: list[PacketSpec] cadence_seconds: float = 0.03 @classmethod def voice_over( cls, *, peer_id: int | bytes, rf_src: int | bytes = 3120001, dst_id: int | bytes = 91, slot: int = 2, stream_id: int | bytes = 0x01020304, start_seq: int = 0, voice_bursts: int = 1, include_header: bool = False, include_terminator: bool = False, ) -> "StreamProfile": packets = [] seq = start_seq if include_header: packets.append( PacketSpec( peer_id=peer_id, rf_src=rf_src, dst_id=dst_id, slot=slot, stream_id=stream_id, seq=seq, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VHEAD, ) ) seq = (seq + 1) % 256 for index in range(voice_bursts): packets.append( PacketSpec( peer_id=peer_id, rf_src=rf_src, dst_id=dst_id, slot=slot, stream_id=stream_id, seq=seq, frame_type=HBPF_VOICE, dtype_vseq=index % 6, ) ) seq = (seq + 1) % 256 if include_terminator: packets.append( PacketSpec( peer_id=peer_id, rf_src=rf_src, dst_id=dst_id, slot=slot, stream_id=stream_id, seq=seq, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VTERM, ) ) return cls(packets=packets) def send_scheduled_packets( send_packet, packets: list[bytes], *, cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, ) -> None: if impairment is None: for index, packet in enumerate(packets): if index and cadence_seconds is not None: time.sleep(cadence_seconds) send_packet(packet) return started_at = time.monotonic() for send_at, packet in impairment.schedule(packets, cadence_seconds=cadence_seconds): wait = started_at + send_at - time.monotonic() if wait > 0: time.sleep(wait) send_packet(packet) @dataclass(frozen=True) class RecordedPacketFixture: """Small line-oriented recorded packet fixture. Fixture files contain one hex-encoded UDP payload per non-empty line. Lines starting with ``#`` are ignored. The format is intentionally transport-only: replay preserves bytes and leaves protocol mutation to FreeDMR. """ packets: list[bytes] @classmethod def from_file(cls, path: Path) -> "RecordedPacketFixture": packets = [] for line in path.read_text(encoding="ascii").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue packets.append(bytes.fromhex(stripped)) return cls(packets) def replay( self, send_packet, *, cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, ) -> None: send_scheduled_packets( send_packet, self.packets, cadence_seconds=cadence_seconds, impairment=impairment, ) def require_udp_integration_enabled() -> None: if os.environ.get("FREEDMR_RUN_UDP_TESTS") != "1": raise unittest.SkipTest("set FREEDMR_RUN_UDP_TESTS=1 to run black-box UDP tests") def _venv_python(venv_dir: Path) -> Path: if os.name == "nt": return venv_dir / "Scripts" / "python.exe" return venv_dir / "bin" / "python" def python_has_runtime_deps( python_executable: str | Path, modules: tuple[str, ...] = REQUIRED_RUNTIME_MODULES, ) -> bool: return _runtime_import_check(python_executable, modules).returncode == 0 def _runtime_import_check( python_executable: str | Path, modules: tuple[str, ...] = REQUIRED_RUNTIME_MODULES, ) -> subprocess.CompletedProcess: imports = "; ".join(f"import {module}" for module in modules) return subprocess.run( [str(python_executable), "-c", imports], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) class DependencySandbox: """Resolve or create a Python runtime that can start bridge_master.py.""" def __init__( self, repo_root: Path, extra_runtime_modules: tuple[str, ...] = (), ) -> None: self.repo_root = repo_root self.runtime_modules = REQUIRED_RUNTIME_MODULES + extra_runtime_modules self._tempdir: tempfile.TemporaryDirectory | None = None def cleanup(self) -> None: if self._tempdir is not None: self._tempdir.cleanup() self._tempdir = None def resolve_python(self) -> str: explicit_python = os.environ.get("FREEDMR_UDP_PYTHON") if explicit_python: if not python_has_runtime_deps(explicit_python, self.runtime_modules): raise unittest.SkipTest( "FREEDMR_UDP_PYTHON does not have FreeDMR runtime dependencies" ) return explicit_python if python_has_runtime_deps(sys.executable, self.runtime_modules): return sys.executable if os.environ.get("FREEDMR_UDP_BOOTSTRAP_VENV") != "1": raise unittest.SkipTest( "missing FreeDMR runtime dependencies; set " "FREEDMR_UDP_BOOTSTRAP_VENV=1 to install them in a test venv" ) return str(self._bootstrap_venv_python()) def _bootstrap_venv_python(self) -> Path: venv_dir_env = os.environ.get("FREEDMR_UDP_VENV_DIR") if venv_dir_env: venv_dir = Path(venv_dir_env).expanduser().resolve() else: self._tempdir = tempfile.TemporaryDirectory(prefix="freedmr-udp-venv-") venv_dir = Path(self._tempdir.name) python_executable = _venv_python(venv_dir) if not python_executable.exists(): venv.EnvBuilder(with_pip=True).create(venv_dir) if not python_has_runtime_deps(python_executable, self.runtime_modules): requirements = self.repo_root / "requirements.txt" subprocess.check_call( [str(python_executable), "-m", "pip", "install", "-r", str(requirements)] ) import_check = _runtime_import_check(python_executable, self.runtime_modules) if import_check.returncode != 0: raise RuntimeError( "test venv was created but FreeDMR dependencies still fail to import:\n" + import_check.stderr ) return python_executable def free_udp_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] def free_udp_port_range(count: int) -> int: if count < 1: raise ValueError("count must be at least 1") for _attempt in range(100): start = random.randint(20000, 60000 - count) sockets = [] try: for port in range(start, start + count): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("127.0.0.1", port)) sockets.append(sock) return start except OSError: pass finally: for sock in sockets: sock.close() raise RuntimeError(f"could not reserve a contiguous UDP port range of {count}") def parse_udp_dmr_fields(packet: bytes) -> dict[str, object]: if packet[:4] != DMRE: return parse_dmr_fields(packet) fields = parse_dmr_fields(DMRD + packet[4:53]) fields.update( { "opcode": packet[:4], "ber": packet[53:54], "rssi": packet[54:55], "fbp_version": packet[55] if len(packet) > 55 else None, } ) if len(packet) >= 89 and packet[55] > 4: fields.update( { "timestamp": packet[56:64], "source_server": packet[64:68], "source_rptr": packet[68:72], "hops": packet[72], "hash": packet[73:89], } ) elif len(packet) >= 85: fields.update( { "timestamp": packet[56:64], "source_server": packet[64:68], "source_rptr": b"\x00\x00\x00\x00", "hops": packet[68], "hash": packet[69:85], } ) return fields def fbp_packet_from_spec( packet: PacketSpec, *, network_id: int | bytes, fbp_version: int = FBP_VERSION, source_server: int | bytes = 9991, source_rptr: int | bytes = 1001, hops: int = 0, timestamp_ns: int | None = None, corrupt_hash: bool = False, passphrase: bytes = FBP_PASSPHRASE, ) -> bytes: dmr = packet.data() body = b"".join( [ DMRE, dmr[4:11], bytes_4(network_id), dmr[15:53], packet.ber, packet.rssi, bytes([fbp_version]), (time.time_ns() if timestamp_ns is None else timestamp_ns).to_bytes(8, "big"), bytes_4(source_server), ] ) if fbp_version > 4: body += bytes_4(source_rptr) + bytes([hops]) else: body += bytes([hops]) digest = blake2b(body, key=passphrase, digest_size=16).digest() if corrupt_hash: digest = bytes([digest[0] ^ 0xFF]) + digest[1:] return body + digest def obp_v1_packet_from_spec( packet: PacketSpec, *, network_id: int | bytes, corrupt_hash: bool = False, passphrase: bytes = FBP_PASSPHRASE, ) -> bytes: dmr = packet.data() body = b"".join([DMRD, dmr[4:11], bytes_4(network_id), dmr[15:53]]) digest = hmac_new(passphrase, body, sha1).digest() if corrupt_hash: digest = bytes([digest[0] ^ 0xFF]) + digest[1:] return body + digest def fbp_control_packet( opcode: bytes, payload: bytes = b"", *, corrupt_hash: bool = False, passphrase: bytes = FBP_PASSPHRASE, ) -> bytes: body = opcode + payload if opcode == BCVE: digest_payload = payload else: digest_payload = body digest = hmac_new(passphrase, digest_payload, sha1).digest() if corrupt_hash: digest = bytes([digest[0] ^ 0xFF]) + digest[1:] return body + digest def repeater_config_packet(radio_id: bytes, callsign: str = "TST") -> bytes: return b"".join( [ radio_id, callsign.encode("ascii", "ignore").ljust(8)[:8], b"439000000", b"430000000", b"10", b"01", b"000.0000", b"000.00000", b"000", b"FreeDMR Test".ljust(20), b"UDP Harness".ljust(19), b"2", b"http://localhost".ljust(124), b"freedmr-test".ljust(40), b"freedmr-test".ljust(40), ] ) def write_bridge_master_config( path: Path, system_ports: dict[str, int], *, fbp_system_ports: dict[str, int] | None = None, fbp_target_ports: dict[str, int] | None = None, fbp_network_ids: dict[str, int] | None = None, fbp_proto_versions: dict[str, int] | None = None, global_use_acl: bool = False, global_sub_acl: str = "PERMIT:ALL", global_tg1_acl: str = "PERMIT:ALL", global_tg2_acl: str = "PERMIT:ALL", ts1_static: str = "", ts2_static: str = "91", dial_a_tg: bool = True, dynamic_tg_routing: bool = True, network_direct_dial_slot: int = 1, master_extra_config: str = "", ) -> None: global_use_acl_text = "True" if global_use_acl else "False" dial_a_tg_text = "True" if dial_a_tg else "False" dynamic_tg_routing_text = "True" if dynamic_tg_routing else "False" systems = [] for name, port in system_ports.items(): systems.append( f""" [{name}] MODE: MASTER ENABLED: True REPEAT: False MAX_PEERS: 4 IP: 127.0.0.1 PORT: {port} PASSPHRASE: GROUP_HANGTIME: 0 USE_ACL: False REG_ACL: PERMIT:ALL SUB_ACL: PERMIT:ALL TGID_TS1_ACL: PERMIT:ALL TGID_TS2_ACL: PERMIT:ALL DEFAULT_UA_TIMER: 1 SINGLE_MODE: True VOICE_IDENT: False DIAL_A_TG: {dial_a_tg_text} DYNAMIC_TG_ROUTING: {dynamic_tg_routing_text} NETWORK_DIRECT_DIAL_SLOT: {network_direct_dial_slot} TS1_STATIC: {ts1_static} TS2_STATIC: {ts2_static} DEFAULT_DIAL_TS1: 0 DEFAULT_DIAL_TS2: 0 DEFAULT_REFLECTOR: 0 ANNOUNCEMENT_LANGUAGE: en_GB GENERATOR: 0 ALLOW_UNREG_ID: True PROXY_CONTROL: False OVERRIDE_IDENT_TG: {master_extra_config} """ ) for name, port in (fbp_system_ports or {}).items(): target_port = (fbp_target_ports or {})[name] network_id = (fbp_network_ids or {})[name] proto_version = (fbp_proto_versions or {}).get(name, FBP_VERSION) systems.append( f""" [{name}] MODE: OPENBRIDGE ENABLED: True NETWORK_ID: {network_id} IP: 127.0.0.1 PORT: {port} PASSPHRASE: test-passphrase TARGET_IP: 127.0.0.1 TARGET_PORT: {target_port} USE_ACL: False SUB_ACL: PERMIT:ALL TGID_ACL: PERMIT:ALL RELAX_CHECKS: True ENHANCED_OBP: True PROTO_VER: {proto_version} """ ) path.write_text( f"""[GLOBAL] PATH: ./ PING_TIME: 1 MAX_MISSED: 3 USE_ACL: {global_use_acl_text} REG_ACL: PERMIT:ALL SUB_ACL: {global_sub_acl} TGID_TS1_ACL: {global_tg1_acl} TGID_TS2_ACL: {global_tg2_acl} GEN_STAT_BRIDGES: False ALLOW_NULL_PASSPHRASE: True ANNOUNCEMENT_LANGUAGES: en_GB SERVER_ID: 9990 DATA_GATEWAY: False VALIDATE_SERVER_IDS: False DEBUG_BRIDGES: False ENABLE_API: False [REPORTS] REPORT: False REPORT_INTERVAL: 60 REPORT_PORT: 0 REPORT_CLIENTS: 127.0.0.1 [LOGGER] LOG_FILE: /dev/null LOG_HANDLERS: console LOG_LEVEL: INFO LOG_NAME: FreeDMR-Test [ALIASES] TRY_DOWNLOAD: False PATH: ./ PEER_FILE: SUBSCRIBER_FILE: TGID_FILE: LOCAL_SUBSCRIBER_FILE: STALE_DAYS: 1 SUB_MAP_FILE: SERVER_ID_FILE: CHECKSUM_FILE: KEYS_FILE: [ALLSTAR] ENABLED: False USER: test PASS: test SERVER: 127.0.0.1 PORT: 5038 NODE: 0 {''.join(systems)} """, encoding="utf-8", ) def write_hotspot_proxy_config( path: Path, *, listen_port: int, dest_port_start: int, dest_port_end: int, master: str = "127.0.0.1", ) -> None: path.write_text( f"""[PROXY] Master: {master} ListenPort: {listen_port} ListenIP: 127.0.0.1 DestportStart: {dest_port_start} DestPortEnd: {dest_port_end} Timeout: 30 Stats: False Debug: False ClientInfo: False BlackList: [] IPBlackList: {{}} """, encoding="utf-8", ) @dataclass class UdpCapture: packet: bytes sockaddr: tuple[str, int] received_at: float fields: dict[str, object] = field(init=False) def __post_init__(self) -> None: self.fields = parse_udp_dmr_fields(self.packet) class HbpRepeater: def __init__( self, master_port: int, radio_id: int, timeout: float = 2.0, bind_host: str = "127.0.0.1", master_host: str = "127.0.0.1", ) -> None: self.master = (master_host, master_port) self.radio_id = bytes_4(radio_id) self.timeout = timeout self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind((bind_host, 0)) self.sock.settimeout(timeout) self.captures: list[UdpCapture] = [] @property def sockaddr(self) -> tuple[str, int]: return self.sock.getsockname() def close(self) -> None: self.sock.close() def send(self, packet: bytes) -> None: self.sock.sendto(packet, self.master) def recv(self, timeout: float | None = None) -> UdpCapture: old_timeout = self.sock.gettimeout() if timeout is not None: self.sock.settimeout(timeout) try: packet, sockaddr = self.sock.recvfrom(4096) finally: if timeout is not None: self.sock.settimeout(old_timeout) capture = UdpCapture(packet, sockaddr, time.monotonic()) self.captures.append(capture) return capture def drain(self, seconds: float = 0.2) -> list[UdpCapture]: deadline = time.monotonic() + seconds captures = [] while True: remaining = deadline - time.monotonic() if remaining <= 0: return captures try: captures.append(self.recv(timeout=remaining)) except TimeoutError: return captures except socket.timeout: return captures def login(self, startup_timeout: float = 8.0) -> None: deadline = time.monotonic() + startup_timeout while True: self.send(RPTL + self.radio_id) try: challenge = self.recv(timeout=min(self.timeout, max(0.1, deadline - time.monotonic()))) break except (TimeoutError, socket.timeout): if time.monotonic() >= deadline: raise if not challenge.packet.startswith(RPTACK): raise AssertionError(f"expected RPTACK challenge, got {challenge.packet!r}") self.send(RPTK + self.radio_id) auth = self.recv() if not auth.packet.startswith(RPTACK): raise AssertionError(f"expected RPTACK auth, got {auth.packet!r}") self.send(RPTC + repeater_config_packet(self.radio_id)) configured = self.recv() if not configured.packet.startswith(RPTACK): raise AssertionError(f"expected RPTACK config, got {configured.packet!r}") def ping(self) -> None: self.send(RPTPING + self.radio_id) pong = self.recv() if not pong.packet.startswith(MSTPONG): raise AssertionError(f"expected MSTPONG, got {pong.packet!r}") def send_dmr(self, packet: PacketSpec) -> None: self.send(packet.data()) def replay_fixture( self, fixture: RecordedPacketFixture, *, cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, ) -> None: fixture.replay( self.send, cadence_seconds=cadence_seconds, impairment=impairment, ) def send_stream( self, packets: list[PacketSpec], cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, ) -> None: send_scheduled_packets( self.send, [packet.data() for packet in packets], cadence_seconds=cadence_seconds, impairment=impairment, ) class FbpPeer: def __init__(self, obp_port: int, network_id: int, timeout: float = 2.0) -> None: self.master = ("127.0.0.1", obp_port) self.network_id = network_id self.timeout = timeout self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(("127.0.0.1", 0)) self.sock.settimeout(timeout) self.captures: list[UdpCapture] = [] @property def sockaddr(self) -> tuple[str, int]: return self.sock.getsockname() def close(self) -> None: self.sock.close() def send(self, packet: bytes) -> None: self.sock.sendto(packet, self.master) def recv(self, timeout: float | None = None) -> UdpCapture: old_timeout = self.sock.gettimeout() if timeout is not None: self.sock.settimeout(timeout) try: packet, sockaddr = self.sock.recvfrom(4096) finally: if timeout is not None: self.sock.settimeout(old_timeout) capture = UdpCapture(packet, sockaddr, time.monotonic()) self.captures.append(capture) return capture def drain(self, seconds: float = 0.2) -> list[UdpCapture]: deadline = time.monotonic() + seconds captures = [] while True: remaining = deadline - time.monotonic() if remaining <= 0: return captures try: captures.append(self.recv(timeout=remaining)) except TimeoutError: return captures except socket.timeout: return captures def recv_dmre(self, timeout: float = 2.0) -> UdpCapture: deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: raise socket.timeout("timed out waiting for DMRE") capture = self.recv(timeout=remaining) if capture.packet[:4] == DMRE: return capture def recv_opcode(self, opcode: bytes, timeout: float = 2.0) -> UdpCapture: deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: raise socket.timeout(f"timed out waiting for {opcode!r}") capture = self.recv(timeout=remaining) if capture.packet[: len(opcode)] == opcode: return capture def send_bcka(self) -> None: self.send(fbp_control_packet(BCKA)) def send_bcve(self, version: int = FBP_VERSION) -> None: self.send(fbp_control_packet(BCVE, bytes([version]))) def send_invalid_bcve(self, version: int = FBP_VERSION) -> None: self.send(fbp_control_packet(BCVE, bytes([version]), corrupt_hash=True)) def send_bcst(self) -> None: self.send(fbp_control_packet(BCST)) def send_bcsq(self, tgid: int | bytes, stream_id: int | bytes) -> None: self.send(fbp_control_packet(BCSQ, bytes_3(tgid) + bytes_4(stream_id))) def send_invalid_bcsq(self, tgid: int | bytes, stream_id: int | bytes) -> None: self.send( fbp_control_packet( BCSQ, bytes_3(tgid) + bytes_4(stream_id), corrupt_hash=True, ) ) def send_fbp( self, packet: PacketSpec, *, network_id: int | bytes | None = None, fbp_version: int = FBP_VERSION, source_server: int | bytes = 9991, source_rptr: int | bytes = 1001, hops: int = 0, timestamp_ns: int | None = None, corrupt_hash: bool = False, ) -> None: self.send( fbp_packet_from_spec( packet, network_id=self.network_id if network_id is None else network_id, fbp_version=fbp_version, source_server=source_server, source_rptr=source_rptr, hops=hops, timestamp_ns=timestamp_ns, corrupt_hash=corrupt_hash, ) ) def send_obp_v1( self, packet: PacketSpec, *, network_id: int | bytes | None = None, corrupt_hash: bool = False, ) -> None: self.send( obp_v1_packet_from_spec( packet, network_id=self.network_id if network_id is None else network_id, corrupt_hash=corrupt_hash, ) ) def replay_fixture( self, fixture: RecordedPacketFixture, *, cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, ) -> None: fixture.replay( self.send, cadence_seconds=cadence_seconds, impairment=impairment, ) def send_fbp_stream( self, packets: list[PacketSpec], *, cadence_seconds: float | None = None, impairment: LinkImpairment | None = None, network_id: int | bytes | None = None, fbp_version: int = FBP_VERSION, source_server: int | bytes = 9991, source_rptr: int | bytes = 1001, hops: int = 0, timestamp_ns: int | None = None, corrupt_hash: bool = False, ) -> None: raw_packets = [ fbp_packet_from_spec( packet, network_id=self.network_id if network_id is None else network_id, fbp_version=fbp_version, source_server=source_server, source_rptr=source_rptr, hops=hops, timestamp_ns=timestamp_ns, corrupt_hash=corrupt_hash, ) for packet in packets ] send_scheduled_packets( self.send, raw_packets, cadence_seconds=cadence_seconds, impairment=impairment, ) class FreeDmrProcess: def __init__(self, repo_root: Path, config_path: Path, python_executable: str) -> None: self.repo_root = repo_root self.config_path = config_path self.python_executable = python_executable self.proc: subprocess.Popen | None = None self._output_lines: list[str] = [] self._output_lock = threading.Lock() self._reader: threading.Thread | None = None def __enter__(self): self.proc = subprocess.Popen( [self.python_executable, "bridge_master.py", "-c", str(self.config_path), "-l", "INFO"], cwd=str(self.repo_root), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, ) self._reader = threading.Thread(target=self._read_output, daemon=True) self._reader.start() return self def __exit__(self, exc_type, exc, tb) -> None: if self.proc is None: return if self.proc.poll() is None: self.proc.terminate() try: self.proc.wait(timeout=5) except subprocess.TimeoutExpired: self.proc.kill() self.proc.wait(timeout=5) if self.proc.stdout is not None: self.proc.stdout.close() if self._reader is not None: self._reader.join(timeout=1) def _read_output(self) -> None: if self.proc is None or self.proc.stdout is None: return for line in self.proc.stdout: with self._output_lock: self._output_lines.append(line) def output(self) -> str: with self._output_lock: return "".join(self._output_lines) def wait_for_log(self, text: str, timeout: float = 2.0) -> str: deadline = time.monotonic() + timeout while time.monotonic() < deadline: output = self.output() if text in output: return output time.sleep(0.05) raise AssertionError(f"log text {text!r} not found in output:\n{self.output()}") def wait_for_start(self, timeout: float = 8.0) -> None: assert self.proc is not None deadline = time.monotonic() + timeout ready_at = time.monotonic() + 0.5 while time.monotonic() < deadline: if self.proc.poll() is not None: raise RuntimeError("FreeDMR exited before startup:\n" + self.output()) # bridge_master does not provide a dedicated readiness signal. Give # Twisted enough time to bind loopback UDP sockets, then let the # first client login be the real readiness check. if time.monotonic() >= ready_at: return time.sleep(0.05) raise TimeoutError("FreeDMR did not reach startup wait window") class HotspotProxyProcess(FreeDmrProcess): def __enter__(self): self.proc = subprocess.Popen( [self.python_executable, "hotspot_proxy_v2.py", "-c", str(self.config_path)], cwd=str(self.repo_root), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, ) self._reader = threading.Thread(target=self._read_output, daemon=True) self._reader.start() return self def wait_for_start(self, timeout: float = 8.0) -> None: assert self.proc is not None deadline = time.monotonic() + timeout ready_at = time.monotonic() + 0.5 while time.monotonic() < deadline: if self.proc.poll() is not None: raise RuntimeError("hotspot_proxy_v2 exited before startup:\n" + self.output()) if time.monotonic() >= ready_at: return time.sleep(0.05) raise TimeoutError("hotspot_proxy_v2 did not reach startup wait window") class UdpBlackBoxScenario: def __init__( self, repo_root: Path | None = None, *, global_use_acl: bool = False, global_sub_acl: str = "PERMIT:ALL", global_tg1_acl: str = "PERMIT:ALL", global_tg2_acl: str = "PERMIT:ALL", ts1_static: str = "", ts2_static: str = "91", dial_a_tg: bool = True, dynamic_tg_routing: bool = True, network_direct_dial_slot: int = 1, master_extra_config: str = "", fbp_systems: dict[str, int] | None = None, fbp_proto_versions: dict[str, int] | None = None, enable_proxy: bool = False, proxy_dest_count: int = 1, ) -> None: self.repo_root = repo_root or Path(__file__).resolve().parents[2] self.tempdir = tempfile.TemporaryDirectory(prefix="freedmr-udp-test-") self.config_path = Path(self.tempdir.name) / "freedmr-test.cfg" self.proxy_config_path = Path(self.tempdir.name) / "hotspot-proxy-test.cfg" self.enable_proxy = enable_proxy self.proxy_dest_ports = [] self.proxy_port = free_udp_port() if enable_proxy else None if enable_proxy: proxy_dest_start = free_udp_port_range(proxy_dest_count) self.proxy_dest_ports = list( range(proxy_dest_start, proxy_dest_start + proxy_dest_count) ) master_b_port = free_udp_port() while master_b_port in self.proxy_dest_ports: master_b_port = free_udp_port() self.system_ports = { "MASTER-A": self.proxy_dest_ports[0], "MASTER-B": master_b_port, } for index, port in enumerate(self.proxy_dest_ports[1:], start=1): self.system_ports[f"PROXY-{index}"] = port else: self.system_ports = {"MASTER-A": free_udp_port(), "MASTER-B": free_udp_port()} self.fbp_network_ids = fbp_systems or {} self.fbp_proto_versions = fbp_proto_versions or {} self.fbp_system_ports = {name: free_udp_port() for name in self.fbp_network_ids} self.fbp_peers: dict[str, FbpPeer] = {} self.process: FreeDmrProcess | None = None self.proxy_process: HotspotProxyProcess | None = None self.deps = DependencySandbox( self.repo_root, extra_runtime_modules=PROXY_RUNTIME_MODULES if enable_proxy else (), ) self.config_options = { "global_use_acl": global_use_acl, "global_sub_acl": global_sub_acl, "global_tg1_acl": global_tg1_acl, "global_tg2_acl": global_tg2_acl, "ts1_static": ts1_static, "ts2_static": ts2_static, "dial_a_tg": dial_a_tg, "dynamic_tg_routing": dynamic_tg_routing, "network_direct_dial_slot": network_direct_dial_slot, "master_extra_config": master_extra_config, } def __enter__(self): python_executable = self.deps.resolve_python() self.fbp_peers = { name: FbpPeer(self.fbp_system_ports[name], network_id) for name, network_id in self.fbp_network_ids.items() } write_bridge_master_config( self.config_path, self.system_ports, fbp_system_ports=self.fbp_system_ports, fbp_target_ports={ name: peer.sockaddr[1] for name, peer in self.fbp_peers.items() }, fbp_network_ids=self.fbp_network_ids, fbp_proto_versions=self.fbp_proto_versions, **self.config_options, ) self.process = FreeDmrProcess(self.repo_root, self.config_path, python_executable) self.process.__enter__() self.process.wait_for_start() if self.enable_proxy: assert self.proxy_port is not None write_hotspot_proxy_config( self.proxy_config_path, listen_port=self.proxy_port, dest_port_start=self.proxy_dest_ports[0], dest_port_end=self.proxy_dest_ports[-1], ) self.proxy_process = HotspotProxyProcess( self.repo_root, self.proxy_config_path, python_executable, ) self.proxy_process.__enter__() self.proxy_process.wait_for_start() return self def __exit__(self, exc_type, exc, tb) -> None: if self.proxy_process is not None: self.proxy_process.__exit__(exc_type, exc, tb) if self.process is not None: self.process.__exit__(exc_type, exc, tb) for peer in self.fbp_peers.values(): peer.close() self.deps.cleanup() self.tempdir.cleanup() def repeater(self, system_name: str, radio_id: int) -> HbpRepeater: return HbpRepeater(self.system_ports[system_name], radio_id) def proxied_repeater(self, radio_id: int) -> HbpRepeater: if self.proxy_port is None: raise RuntimeError("UdpBlackBoxScenario was not created with enable_proxy=True") return HbpRepeater(self.proxy_port, radio_id, bind_host="127.0.0.2") def fbp_peer(self, system_name: str) -> FbpPeer: return self.fbp_peers[system_name]