diff options
Diffstat (limited to 'stacks/daemon')
35 files changed, 8132 insertions, 0 deletions
diff --git a/stacks/daemon/README.md b/stacks/daemon/README.md new file mode 100644 index 0000000..bfba87e --- /dev/null +++ b/stacks/daemon/README.md @@ -0,0 +1,72 @@ +# max25d — Linux daemon + +**max25d** supervises TNC/modem hardware, runs per-device RX/TX backends, and exposes M25/1 to `max25-terminal`. + +Linux only. Config: `share/max25/max25d.ini.example`. + +## Components + +| File | Role | +|------|------| +| `max25d` | Python 3 daemon — M25/1 server, stack lifecycle | +| `device_backends.py` | Backend abstraction (TNC, BayCom, CRDOP) | +| `kiss_bridge.py` | Serial KISS for command-mode TNCs (TNC2C/PK-TNC2) | +| `banlist.py` | AX.25 source ban list (silent RX drop) | +| `test_multi_device.py` | Multi-device routing smoke | + +## Backends + +| Device id | Backend | Notes | +|-----------|---------|-------| +| `tnc2c`, `pktnc2` | `KissSerialBackend` | CI-tested | +| `max25e0` (`max25-bcpr:bc0`) | bcpr / KISS PTY | BayCom/based SER12 userspace — [docs/BAYCOM.md](../../docs/BAYCOM.md) | +| `baycom-kiss` | `KissRawSerialBackend` | USB/async KISS (not SER12 product face) | +| `soft-crdop` | `CrdopTcpBackend` | MAX25-SoftModem (CRDOP) — standard; KISS default; sound IN/OUT + radio like hardware modem | + +Untested backends log a startup warning; SEND/RX return `ERR link not ready` when the stack path is unavailable. + +## Quick start + +```bash +# Protocol only (no hardware): +./max25d --no-stack -c ../../share/max25/max25d.ini.example + +# With stack auto-start: +sudo ./max25d -c /etc/max25/max25d.ini +``` + +## One RF device per Linux host + +**Target:** one `[devices]` id per `max25d`. See [ARCHITECTURE.md](../../docs/ARCHITECTURE.md#linux-host-policy--one-rf-device). + +```ini +[devices] +default = tnc2c +tnc2c = /dev/ttyS4 +``` + +**Legacy multi-id** (deprecated for new sites): `share/max25/max25d.full-station.ini.example`, `max25d.dual-baycom.ini.example`. + +M25/1: `devices=` in `STATUS`, `SET DEVICE <id>`, `GET DEVICES`, `RX device=<id> …`. + +## Source ban list + +Block unwanted AX.25 callers (source address on incoming UI frames). Banned traffic is dropped silently — no terminal display, no daemon log line. + +```ini +[modem] +bans_file = /etc/max25/bans.txt +``` + +File format: one callsign per line (`#` comments allowed). Ban without SSID blocks all SSIDs of that call (`DG1ABC` blocks `DG1ABC-7`). Ban with SSID (`DK0WC-7`) matches only that SSID. + +M25/1: `BAN <callsign>`, `UNBAN <callsign>`, `BANS`. Changes persist to `bans_file` immediately. + +## Install + +```bash +./scripts/install-max25.sh --deps +# Installs: /usr/local/bin/max25d, kiss_bridge.py, device_backends.py, max25-terminal +``` + +Host setup guide: [docs/LINUX-HOST-SETUP.md](../../docs/LINUX-HOST-SETUP.md). Protocol: [include/max25/protocol.md](../../include/max25/protocol.md). Client: [docs/MAX25-CLIENT.md](../../docs/MAX25-CLIENT.md). diff --git a/stacks/daemon/ax25_codec.py b/stacks/daemon/ax25_codec.py new file mode 100644 index 0000000..03ee671 --- /dev/null +++ b/stacks/daemon/ax25_codec.py @@ -0,0 +1,176 @@ +""" +AX.25 UI frame codec for max25d — address fields, CRC-CCITT FCS, UI parse/build. + +Derived from libax25 axutils.c (address encoding) and ax25ipd/crc.c (RFC 1171 FCS). +KISS DATA carries the AX.25 body without FCS; over-the-air and some paths include FCS. +""" +from __future__ import annotations + +import re +from typing import Optional + +# HDLC CRC-CCITT (polynomial 0x1021), RFC 1171 / ax25ipd table +_FCS_TABLE: tuple[int, ...] = ( + 0x0000, 0x1189, 0x2312, 0x329B, 0x4624, 0x57AD, 0x6536, 0x74BF, + 0x8C48, 0x9DC1, 0xAF5A, 0xBED3, 0xCA6C, 0xDBE5, 0xE97E, 0xF8F7, + 0x1081, 0x0108, 0x3393, 0x221A, 0x56A5, 0x472C, 0x75B7, 0x643E, + 0x9CC9, 0x8D40, 0xBFDB, 0xAE52, 0xDAED, 0xCB64, 0xF9FF, 0xE876, + 0x2102, 0x308B, 0x0210, 0x1399, 0x6726, 0x76AF, 0x4434, 0x55BD, + 0xAD4A, 0xBCC3, 0x8E58, 0x9FD1, 0xEB6E, 0xFAE7, 0xC87C, 0xD9F5, + 0x3183, 0x200A, 0x1291, 0x0318, 0x77A7, 0x662E, 0x54B5, 0x453C, + 0xBDCB, 0xAC42, 0x9ED9, 0x8F50, 0xFBEF, 0xEA66, 0xD8FD, 0xC974, + 0x4204, 0x538D, 0x6116, 0x709F, 0x0420, 0x15A9, 0x2732, 0x36BB, + 0xCE4C, 0xDFC5, 0xED5E, 0xFCD7, 0x8868, 0x99E1, 0xAB7A, 0xBAF3, + 0x5285, 0x430C, 0x7197, 0x601E, 0x14A1, 0x0528, 0x37B3, 0x263A, + 0xDECD, 0xCF44, 0xFDDF, 0xEC56, 0x98E9, 0x8960, 0xBBFB, 0xAA72, + 0x6306, 0x728F, 0x4014, 0x519D, 0x2522, 0x34AB, 0x0630, 0x17B9, + 0xEF4E, 0xFEC7, 0xCC5C, 0xDDD5, 0xA96A, 0xB8E3, 0x8A78, 0x9BF1, + 0x7387, 0x620E, 0x5095, 0x411C, 0x35A3, 0x242A, 0x16B1, 0x0738, + 0xFFCF, 0xEE46, 0xDCDD, 0xCD54, 0xB9EB, 0xA862, 0x9AF9, 0x8B70, + 0x8408, 0x9581, 0xA71A, 0xB693, 0xC22C, 0xD3A5, 0xE13E, 0xF0B7, + 0x0840, 0x19C9, 0x2B52, 0x3ADB, 0x4E64, 0x5FED, 0x6D76, 0x7CFF, + 0x9489, 0x8500, 0xB79B, 0xA612, 0xD2AD, 0xC324, 0xF1BF, 0xE036, + 0x18C1, 0x0948, 0x3BD3, 0x2A5A, 0x5EE5, 0x4F6C, 0x7DF7, 0x6C7E, + 0xA50A, 0xB483, 0x8618, 0x9791, 0xE32E, 0xF2A7, 0xC03C, 0xD1B5, + 0x2942, 0x38CB, 0x0A50, 0x1BD9, 0x6F66, 0x7EEF, 0x4C74, 0x5DFD, + 0xB58B, 0xA402, 0x9699, 0x8710, 0xF3AF, 0xE226, 0xD0BD, 0xC134, + 0x39C3, 0x284A, 0x1AD1, 0x0B58, 0x7FE7, 0x6E6E, 0x5CF5, 0x4D7C, + 0xC60C, 0xD785, 0xE51E, 0xF497, 0x8028, 0x91A1, 0xA33A, 0xB2B3, + 0x4A44, 0x5BCD, 0x6956, 0x78DF, 0x0C60, 0x1DE9, 0x2F72, 0x3EFB, + 0xD68D, 0xC704, 0xF59F, 0xE416, 0x90A9, 0x8120, 0xB3BB, 0xA232, + 0x5AC5, 0x4B4C, 0x79D7, 0x685E, 0x1CE1, 0x0D68, 0x3FF3, 0x2E7A, + 0xE70E, 0xF687, 0xC41C, 0xD595, 0xA12A, 0xB0A3, 0x8238, 0x93B1, + 0x6B46, 0x7ACF, 0x4854, 0x59DD, 0x2D62, 0x3CEB, 0x0E70, 0x1FF9, + 0xF78F, 0xE606, 0xD49D, 0xC514, 0xB1AB, 0xA022, 0x92B9, 0x8330, + 0x7BC7, 0x6A4E, 0x58D5, 0x495C, 0x3DE3, 0x2C6A, 0x1EF1, 0x0F78, +) + +_FCS_GOOD = 0xF0B8 # ax25ipd PPPGOODFCS — residual after valid frame + FCS + +AX25_UI_CONTROL = 0x03 +AX25_UI_PID = 0xF0 +MIN_UI_FRAME = 16 # dest(7) + src(7) + ctrl + pid + +_CALLSIGN_RE = re.compile(r"^([A-Z0-9]{1,6})(?:-([0-9]{1,2}))?$") + + +def parse_callsign(text: str) -> tuple[str, int]: + """Parse operator callsign text; invalid SSID clamps to 0 (legacy helper).""" + text = text.strip().upper() + if "-" in text: + call, ssid_s = text.split("-", 1) + try: + ssid = int(ssid_s) + except ValueError: + ssid = 0 + if ssid < 0 or ssid > 15: + ssid = 0 + return call[:6], ssid + return text[:6], 0 + + +def validate_callsign(text: str) -> tuple[str, int]: + """Strict callsign parse (libax25 ax25_aton_entry rules).""" + text = text.strip().upper() + match = _CALLSIGN_RE.match(text) + if not match: + raise ValueError(f"invalid AX.25 callsign: {text!r}") + call = match.group(1) + ssid = int(match.group(2)) if match.group(2) is not None else 0 + if ssid < 0 or ssid > 15: + raise ValueError(f"invalid AX.25 SSID in {text!r}") + return call, ssid + + +def format_callsign(call: str, ssid: int) -> str: + """Textual callsign; omit -0 suffix (libax25 ax25_ntoa convention).""" + call = call.strip().upper() + if ssid <= 0: + return call + if ssid >= 10: + return f"{call}-{ssid}" + return f"{call}-{ssid}" + + +def ax25_crc(data: bytes) -> int: + """Compute AX.25 FCS over body (addresses + control + PID + info).""" + fcs = 0xFFFF + for b in data: + fcs = (fcs >> 8) ^ _FCS_TABLE[(fcs ^ b) & 0xFF] + return fcs ^ 0xFFFF + + +def ax25_crc_valid(frame: bytes) -> bool: + """True when frame includes a valid little-endian FCS trailer.""" + fcs = 0xFFFF + for b in frame: + fcs = (fcs >> 8) ^ _FCS_TABLE[(fcs ^ b) & 0xFF] + return fcs == _FCS_GOOD + + +def ax25_encode_address(call: str, ssid: int, last: bool) -> bytes: + """Seven-byte AX.25 address field (libax25 ax25_aton_entry layout).""" + padded = call.upper().ljust(6)[:6] + raw = bytes((ord(c) << 1) for c in padded) + ssid_b = ((ssid & 0x0F) << 1) | (0x01 if last else 0x00) + return raw + bytes([ssid_b]) + + +def ax25_decode_address(raw: bytes) -> tuple[str, int, bool]: + call = "".join(chr((b >> 1) & 0x7F) for b in raw[:6]).strip() + ssid = (raw[6] >> 1) & 0x0F + last = bool(raw[6] & 0x01) + return call, ssid, last + + +def ax25_build_ui(src: str, dst: str, info: bytes) -> bytes: + src_call, src_ssid = validate_callsign(src) + dst_call, dst_ssid = validate_callsign(dst) + body = ( + ax25_encode_address(dst_call, dst_ssid, last=False) + + ax25_encode_address(src_call, src_ssid, last=True) + + bytes([AX25_UI_CONTROL, AX25_UI_PID]) + + info + ) + crc = ax25_crc(body) + return body + bytes((crc & 0xFF, crc >> 8)) + + +def ax25_parse_ui(frame: bytes) -> Optional[tuple[str, str, bytes]]: + """ + Parse a UI frame. Strips FCS only when the trailer validates (ax25ipd ok_crc). + KISS payloads are usually FCS-free; over-the-air captures may include FCS. + """ + if len(frame) < MIN_UI_FRAME: + return None + + body = frame + if len(frame) >= MIN_UI_FRAME + 2 and ax25_crc_valid(frame): + body = frame[:-2] + + pos = 0 + addresses: list[tuple[str, int, bool]] = [] + while pos + 7 <= len(body): + call, ssid, last = ax25_decode_address(body[pos : pos + 7]) + addresses.append((call, ssid, last)) + pos += 7 + if last: + break + else: + return None + + if len(addresses) < 2: + return None + if pos + 2 > len(body): + return None + if body[pos] != AX25_UI_CONTROL or body[pos + 1] != AX25_UI_PID: + return None + + payload = body[pos + 2 :] + dst_call, dst_ssid, _ = addresses[0] + src_call, src_ssid, _ = addresses[-1] + return ( + format_callsign(src_call, src_ssid), + format_callsign(dst_call, dst_ssid), + payload, + ) diff --git a/stacks/daemon/banlist.py b/stacks/daemon/banlist.py new file mode 100644 index 0000000..061ed1f --- /dev/null +++ b/stacks/daemon/banlist.py @@ -0,0 +1,89 @@ +"""Simple AX.25 source ban list for max25d — silent RX drop.""" +from __future__ import annotations + +import re +import sys +import threading +from pathlib import Path +from typing import Optional + +DEFAULT_BANS_FILE = Path("/etc/max25/bans.txt") + +_AX25_UI_SRC_RE = re.compile(r"\[AX25 UI ([^>]+)>") + + +def extract_ax25_source(line: str) -> Optional[str]: + """Return source callsign from a formatted AX.25 UI RX line, or None.""" + match = _AX25_UI_SRC_RE.search(line) + if not match: + return None + return match.group(1).strip().upper() + + +def callsign_banned(ban_entry: str, source: str) -> bool: + """Match ban entry against an incoming source callsign.""" + ban = ban_entry.strip().upper() + src = source.strip().upper() + if not ban or not src: + return False + if "-" in ban: + return ban == src + return ban == src.split("-", 1)[0] + + +class BanList: + """Persistent set of banned AX.25 source addresses.""" + + def __init__(self, path: Path | str = DEFAULT_BANS_FILE) -> None: + self.path = Path(path) + self._lock = threading.Lock() + self._entries: set[str] = set() + self.load() + + def load(self) -> None: + entries: set[str] = set() + try: + if self.path.is_file(): + text = self.path.read_text(encoding="utf-8") + for raw in text.splitlines(): + line = raw.split("#", 1)[0].strip().upper() + if line: + entries.add(line) + except OSError as exc: + print(f"max25d: banlist load failed ({self.path}): {exc}", file=sys.stderr) + with self._lock: + self._entries = entries + + def save(self) -> None: + lines = sorted(self._entries) + content = "\n".join(lines) + if content: + content += "\n" + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(content, encoding="utf-8") + + def list(self) -> list[str]: + with self._lock: + return sorted(self._entries) + + def add(self, callsign: str) -> None: + call = callsign.strip().upper() + with self._lock: + self._entries.add(call) + self.save() + + def remove(self, callsign: str) -> bool: + call = callsign.strip().upper() + with self._lock: + if call not in self._entries: + return False + self._entries.remove(call) + self.save() + return True + + def is_banned(self, source: str) -> bool: + src = source.strip().upper() + if not src: + return False + with self._lock: + return any(callsign_banned(entry, src) for entry in self._entries) diff --git a/stacks/daemon/daemon_log.py b/stacks/daemon/daemon_log.py new file mode 100644 index 0000000..f1696df --- /dev/null +++ b/stacks/daemon/daemon_log.py @@ -0,0 +1,237 @@ +""" +Structured stderr logging for max25d — human- and machine-readable lines. + +Format: + max25d [LEVEL] [area] message + max25d [LEVEL] [area] [device] message + +Levels: INFO, OK, WARN, ERROR, EVENT, RECOVERY +""" +from __future__ import annotations + +import sys +from dataclasses import dataclass +from typing import Callable, Optional + +PREFIX = "max25d" +DEVICE_TOKENS = ("tnc2c", "pktnc2", "tmodem", "max25e0", "max25e0:bc0", "max25e0:bc1", "baycom-kiss", "soft-crdop") + + +@dataclass(frozen=True) +class DeviceSummary: + device_id: str + backend: str + hardware: str + serial: str + enabled: bool + tested: bool + + +class DaemonLogger: + """Thread-safe enough for max25d (GIL); all lines go to stderr.""" + + def __init__(self, emit: Optional[Callable[[str], None]] = None) -> None: + self._emit = emit or self._default_emit + + @staticmethod + def _default_emit(line: str) -> None: + print(line, file=sys.stderr, flush=True) + + def _line( + self, + level: str, + msg: str, + *, + area: str = "", + device: str = "", + ) -> None: + parts = [PREFIX, f"[{level}]"] + if area: + parts.append(f"[{area}]") + if device: + parts.append(f"[{device}]") + parts.append(msg) + self._emit(" ".join(parts)) + + def info(self, msg: str, *, area: str = "", device: str = "") -> None: + self._line("INFO", msg, area=area, device=device) + + def ok(self, msg: str, *, area: str = "", device: str = "") -> None: + self._line("OK", msg, area=area, device=device) + + def warn(self, msg: str, *, area: str = "", device: str = "") -> None: + self._line("WARN", msg, area=area, device=device) + + def error(self, msg: str, *, area: str = "", device: str = "") -> None: + self._line("ERROR", msg, area=area, device=device) + + def event(self, msg: str, *, area: str = "", device: str = "") -> None: + self._line("EVENT", msg, area=area, device=device) + + def recovery(self, msg: str, *, device: str = "") -> None: + self._line("RECOVERY", msg, area="serial", device=device) + + def _parse_device_prefix(self, text: str) -> tuple[str, str]: + head, sep, tail = text.partition(":") + if sep and head in DEVICE_TOKENS: + return head, tail.strip() + device = "" + for token in DEVICE_TOKENS: + if f"({token})" in text or f" {token})" in text: + device = token + break + return device, text + + def emit_unstructured(self, msg: str) -> None: + """Map legacy free-form strings to structured levels.""" + device, text = self._parse_device_prefix(msg.strip()) + lower = text.lower() + if lower.startswith("warning:"): + self.warn(text[8:].strip(), device=device) + elif lower.startswith("recovery:"): + self.recovery(text[9:].strip(), device=device) + elif " prep ok" in lower or lower.startswith("ok:"): + self.ok(text, device=device) + elif any(x in lower for x in ("failed", "error", "fail:")): + self.error(text, device=device) + elif lower.startswith("serial watch:"): + self.info(text, area="watch", device=device) + elif lower.startswith("serial "): + self.info(text, area="serial", device=device) + elif lower.startswith("stack "): + self.info(text, area="stack", device=device) + elif lower.startswith("rx ") or lower.startswith("tx "): + self.event(text, device=device) + else: + self.info(text, device=device) + + def banner(self, title: str) -> None: + self._emit(f"{PREFIX} === {title} ===") + + def section(self, name: str) -> None: + self._emit(f"{PREFIX} [{name}]") + + def kv(self, key: str, value: str, *, indent: int = 0) -> None: + pad = " " * indent + self._emit(f"{PREFIX} {pad}{key}={value}") + + +LOGGER = DaemonLogger() + + +def device_serial_label(dev) -> str: + """Human-readable serial/KISS path for startup summary.""" + spec = (dev.device_spec or dev.serial_device or "").strip() + if dev.backend_type == "kiss-serial": + baud = dev.serial_baud or "?" + line = (dev.serial_line or "?").upper() + dtr = dev.serial_dtr_rts or "default" + kiss = dev.serial_kiss_entry or "default" + path = spec or dev.serial_device or "?" + return f"{path} {baud} {line} dtr={dtr} kiss_entry={kiss}" + if dev.backend_type == "baycom-kiss": + return f"baycom:{dev.baycom_modem or 'a'} kiss={dev.kiss_link or '?'}" + if dev.backend_type == "crdop-tcp": + host = dev.crdop_host or "127.0.0.1" + port = dev.crdop_port or "?" + return f"crdop-tcp {host}:{port}" + if spec: + return spec + return dev.backend_type or "auto" + + +def emit_startup_banner( + *, + config_path: Optional[str], + cfg, + devices: list, + tested_fn: Callable[[str], bool], +) -> None: + LOGGER.banner("MAX25d starting") + LOGGER.section("config") + if config_path: + LOGGER.kv("ini", config_path, indent=1) + else: + LOGGER.kv("ini", "(built-in defaults)", indent=1) + LOGGER.kv("mode", cfg.mode, indent=1) + LOGGER.kv("default_device", cfg.default_device or cfg.device, indent=1) + LOGGER.kv("devices", str(len(devices)), indent=1) + + LOGGER.section("network") + LOGGER.kv("tcp", f"{cfg.tcp_host}:{cfg.tcp_port}", indent=1) + LOGGER.kv("unix", cfg.unix_socket or "(disabled)", indent=1) + if cfg.tcp_password: + LOGGER.kv("tcp_auth", "enabled", indent=1) + else: + LOGGER.warn("TCP has no password — set tcp_password before exposing LAN", area="security") + + LOGGER.section("modem") + LOGGER.kv("callerid", cfg.callerid, indent=1) + LOGGER.kv("callid", cfg.callid, indent=1) + LOGGER.kv("ax25_ui", "yes" if cfg.ax25_ui else "no", indent=1) + if cfg.bans_file: + LOGGER.kv("bans_file", cfg.bans_file, indent=1) + + LOGGER.section("stack") + LOGGER.kv("auto_start", "yes" if cfg.auto_start else "no", indent=1) + LOGGER.kv("serial_enabled", "yes" if cfg.serial_enabled else "no", indent=1) + LOGGER.kv("stack_recover_only", "yes" if cfg.stack_recover_only else "no", indent=1) + LOGGER.kv("serial_watch", "yes" if cfg.serial_watch else "no", indent=1) + if cfg.serial_watch: + LOGGER.kv("serial_watch_interval", f"{cfg.serial_watch_interval}s", indent=1) + LOGGER.kv("serial_watch_startup_grace", f"{cfg.serial_watch_startup_grace}s", indent=1) + LOGGER.kv("serial_bootwait_escalate", "yes" if cfg.serial_bootwait_escalate else "no", indent=1) + + LOGGER.section("session") + LOGGER.kv( + "detach", + "max25d --session tmux | max25d-session start/attach/stop", + indent=1, + ) + + LOGGER.section("devices") + for dev in devices: + if not dev.enabled: + LOGGER.kv( + dev.device_id, + f"disabled backend={dev.backend_type or 'auto'}", + indent=1, + ) + continue + serial = device_serial_label(dev) + hw = dev.hardware or cfg.hardware + tested = tested_fn(dev.device_id) + note = "" if tested else " — not hardware-validated in CI" + LOGGER.kv( + dev.device_id, + f"{hw} {serial}{note}", + indent=1, + ) + if not tested and dev.backend_type == "kiss-serial": + LOGGER.warn( + "RF path not CI-validated — verify on hardware before production", + area="devices", + device=dev.device_id, + ) + + +def emit_startup_complete( + *, + device_lines: list[tuple[str, str, str]], + tcp_host: str, + tcp_port: int, + unix_socket: str, +) -> None: + LOGGER.banner("MAX25d ready") + LOGGER.section("listen") + LOGGER.kv("tcp", f"{tcp_host}:{tcp_port}", indent=1) + LOGGER.kv("unix", unix_socket or "(disabled)", indent=1) + if device_lines: + LOGGER.section("device_status") + for dev_id, stack_st, link_st in device_lines: + LOGGER.kv(dev_id, f"stack={stack_st} link={link_st}", indent=1) + LOGGER.info( + "attach: max25-terminal -U /run/max25/modem.sock | " + "detach session: max25d-session attach", + area="operator", + ) diff --git a/stacks/daemon/device_backends.py b/stacks/daemon/device_backends.py new file mode 100644 index 0000000..b4c6c29 --- /dev/null +++ b/stacks/daemon/device_backends.py @@ -0,0 +1,1076 @@ +""" +Device backends for max25d — heterogeneous RF paths (TNC, BayCom, CRDOP). + +Each enabled [devices] id gets one backend instance. Backends without hardware +validation log a startup warning but still wire real stack paths (not silent no-ops). +""" +from __future__ import annotations + +import os +import select +import socket +import struct +import termios +import threading +import time +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Callable, Optional + +from kiss_bridge import ( + MAX_PAYLOAD, + KissDecoder, + SerialProfile, + ax25_build_ui, + ax25_parse_ui, + format_rx_line, + kiss_data_frame, + serial_profile_for_device, +) +from kiss_bridge import KissBridge # noqa: E402 — re-exported wrapper target +from paths import normalize_max25_bcpr_path + +LogFn = Callable[[str], None] +RxFn = Callable[[str], None] +InvalidFn = Callable[[], None] + + +def _spec_int(raw: str, default: int) -> int: + try: + return int(str(raw).strip()) + except (TypeError, ValueError): + return default + +# max25e0 host address defaults (overridable in max25d.ini [device.max25e0]) +MAX25E0_DEFAULT_IPV4 = "127.0.0.25/8" +MAX25E0_DEFAULT_IPV6 = "::25/128" + +# manifest.yaml device ids → default hardware + backend kind +DEVICE_REGISTRY: dict[str, dict[str, str | bool]] = { + "tnc2c": {"hardware": "tncs", "backend": "kiss-serial", "tested": True}, + "pktnc2": {"hardware": "tncs", "backend": "kiss-serial", "tested": False}, + "tmodem": {"hardware": "tncs", "backend": "kiss-raw-serial", "tested": False}, + # Kernel baycom-ser12/par96 removed 2026-07-18 — use max25-bcpr → device max25e0 + "baycom-kiss": {"hardware": "modems", "backend": "kiss-raw-serial", "tested": False}, + "pccom-kiss": {"hardware": "modems", "backend": "kiss-raw-serial", "tested": False}, + "max25e0": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True}, + "max25e0:bc0": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True}, + "max25e0:bc1": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True}, + "soft-crdop": {"hardware": "soft-modems", "backend": "crdop-tcp", "tested": True}, + "audio-dummy": {"hardware": "acoustic-bench", "backend": "audio-dummy", "tested": True}, +} + + +def registry_hardware(device_id: str, fallback: str = "tncs") -> str: + entry = DEVICE_REGISTRY.get(device_id, {}) + return str(entry.get("hardware", fallback)) + + +def registry_backend(device_id: str) -> str: + entry = DEVICE_REGISTRY.get(device_id, {}) + return str(entry.get("backend", "kiss-serial")) + + +def registry_tested(device_id: str) -> bool: + entry = DEVICE_REGISTRY.get(device_id, {}) + return bool(entry.get("tested", False)) + + +def baycom_ctl_device_id(dev_cfg: DeviceBackendConfig) -> str: + """Legacy helper: kernel BayCom ctl device id (stack removed — prefer max25-bcpr).""" + entry = DEVICE_REGISTRY.get(dev_cfg.device_id, {}) + if entry.get("backend") == "baycom-kiss": + return dev_cfg.device_id + return "max25e0" + + +@dataclass +class DeviceBackendConfig: + device_id: str + hardware: str = "" + backend_type: str = "" + device_spec: str = "" + enabled: bool = True + # Serial (TNC / baycom-kiss USB) + serial_device: str = "" + serial_baud: int = 0 + serial_line: str = "" + serial_dtr_rts: str = "" + serial_kiss_entry: str = "" + # BayCom kernel KISS PTY + kiss_link: str = "" + baycom_modem: str = "a" + baycom_ini: str = "" + # max25-bcpr userspace SER12 — max25e0 (+ forks max25e0:bcN) + max25_bcpr_ini: str = "" + max25_bcpr_device: str = "" # bc0 | bc1 + # Host addresses (max25e0 family only; forks inherit from max25e0) + ipv4: str = "" + ipv6: str = "" + # Legacy field aliases (read-only mirrors filled by parser) + bcpr_ini: str = "" + bcpr_device: str = "" + # CRDOP TCP + crdop_host: str = "127.0.0.1" + crdop_port: int = 8515 + crdop_profile: str = "default" + crdop_listen: bool = True + # Acoustic bench / audio-dummy + audio_mode: str = "loopback" # loopback | alsa | host + audio_capture: str = "" + audio_playback: str = "" + audio_sample_rate: int = 48000 + audio_host_port: int = 8520 + + +class DeviceBackend(ABC): + """Common RX/TX/PTT surface for max25d.""" + + device_id: str + status: str = "closed" + backend_type: str = "" + + @abstractmethod + def open(self) -> bool: + ... + + @abstractmethod + def close(self) -> None: + ... + + @abstractmethod + def attach_session(self, mycall: str) -> bool: + ... + + @abstractmethod + def detach_session(self) -> None: + ... + + @abstractmethod + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + ... + + +class KissSerialBackend(DeviceBackend): + """TNC2C / PK-TNC2 — command-mode serial entry into KISS.""" + + backend_type = "kiss-serial" + + def __init__( + self, + cfg: DeviceBackendConfig, + root: str, + on_rx: RxFn, + log: Optional[LogFn] = None, + prefix: Optional[str] = None, + on_invalid: Optional[InvalidFn] = None, + ) -> None: + self.device_id = cfg.device_id + self._cfg = cfg + self._root = root + self._prefix = prefix + self._on_rx = on_rx + self._on_invalid = on_invalid + self._log = log or (lambda _m: None) + self._bridge: Optional[KissBridge] = None + self.status = "closed" + + def _bridge_log(self, msg: str) -> None: + self._log(f"{self.device_id}: {msg}") + + def _ini_overrides(self) -> dict[str, str]: + out: dict[str, str] = {} + if self._cfg.serial_device: + out["device"] = self._cfg.serial_device + if self._cfg.serial_baud: + out["baud"] = str(self._cfg.serial_baud) + if self._cfg.serial_line: + out["line"] = self._cfg.serial_line + if self._cfg.serial_dtr_rts: + out["dtr_rts"] = self._cfg.serial_dtr_rts + if self._cfg.serial_kiss_entry: + out["kiss_entry"] = self._cfg.serial_kiss_entry + return out + + def open(self) -> bool: + profile = serial_profile_for_device( + self.device_id, + self._root, + self._ini_overrides(), + prefix=self._prefix, + ) + bridge = KissBridge( + profile, + self._on_rx, + self._bridge_log, + tree_root=self._root, + install_prefix=self._prefix, + on_invalid=self._on_invalid, + ) + if not bridge.open(): + self._bridge = bridge + self.status = bridge.status + return False + self._bridge = bridge + self.status = bridge.status + return True + + def close(self) -> None: + if self._bridge is not None: + self._bridge.close() + self.status = self._bridge.status + self._bridge = None + else: + self.status = "closed" + + def stabilize_session(self, mycall: str, *, force: bool = False) -> bool: + if self._bridge is None: + return False + ok = self._bridge.stabilize_session(mycall, force=force) + self.status = self._bridge.status + return ok + + def attach_session(self, mycall: str) -> bool: + if self._bridge is None: + return False + ok = self._bridge.attach_session(mycall) + self.status = self._bridge.status + return ok + + def detach_session(self) -> None: + if self._bridge is None: + return + self._bridge.detach_session() + self.status = self._bridge.status + + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + if self._bridge is None: + return False, "serial not ready" + ok, display = self._bridge.transmit(src, dst, text, ax25_ui) + self.status = self._bridge.status + return ok, display + + +class KissRawBackend(DeviceBackend): + """Raw KISS on serial or BayCom KISS PTY (no command-mode entry).""" + + backend_type = "kiss-raw" + + def __init__( + self, + cfg: DeviceBackendConfig, + path: str, + profile: SerialProfile, + on_rx: RxFn, + log: Optional[LogFn] = None, + *, + is_pty: bool = False, + ) -> None: + self.device_id = cfg.device_id + self._path = path + self._profile = profile + self._on_rx = on_rx + self._log = log or (lambda _m: None) + self._is_pty = is_pty + self._fd: Optional[int] = None + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.Lock() + self._mycall = "" + self._kiss_active = False + self._decoder = KissDecoder() + self.status = "closed" + + def open(self) -> bool: + path = self._path + if not path: + self.status = "error-no-path" + self._log(f"{self.device_id}: no KISS path configured") + return False + if not os.path.exists(path): + self.status = "error-no-device" + self._log(f"{self.device_id}: path missing: {path}") + return False + try: + fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) + if not self._is_pty: + self._configure_serial(fd) + termios.tcflush(fd, termios.TCIOFLUSH) + except OSError as exc: + self.status = "error-open" + self._log(f"{self.device_id}: open failed: {exc}") + return False + self._fd = fd + self.status = "open" + self._stop.clear() + self._thread = threading.Thread( + target=self._rx_loop, + name=f"kiss-raw-{self.device_id}", + daemon=True, + ) + self._thread.start() + self._log(f"{self.device_id}: raw KISS open {path}") + return True + + def _configure_serial(self, fd: int) -> None: + from kiss_bridge import _parse_baud, _parse_line + + speed = _parse_baud(self._profile.baud) + databits, parity = _parse_line(self._profile.line) + t = termios.tcgetattr(fd) + t[0] = t[1] = 0 + t[2] = termios.CLOCAL | termios.CREAD | databits | parity + t[3] = t[4] = t[5] = speed + t[6][termios.VMIN] = 0 + t[6][termios.VTIME] = 5 + termios.tcsetattr(fd, termios.TCSANOW, t) + flags = struct.unpack("I", __import__("fcntl").ioctl(fd, 0x5415, struct.pack("I", 0)))[0] + if self._profile.dtr_rts: + flags |= 0x004 | 0x002 + __import__("fcntl").ioctl(fd, 0x5416, struct.pack("I", flags)) + + def close(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + with self._lock: + if self._fd is not None: + try: + os.close(self._fd) + except OSError: + pass + self._fd = None + self._kiss_active = False + self.status = "closed" + self._decoder = KissDecoder() + + def attach_session(self, mycall: str) -> bool: + if self._fd is None: + return False + self._mycall = mycall.upper() + self._kiss_active = True + self.status = "ready" + return True + + def detach_session(self) -> None: + self._kiss_active = False + if self._fd is not None: + self.status = "open" + + def stabilize_session(self, mycall: str, *, force: bool = False) -> bool: + """Reopen KISS path after dead PTY / EIO (e.g. bcprd recycled outside max25d).""" + path = self._path + if ( + not force + and self._kiss_active + and self.status == "ready" + and self._fd is not None + and path + and os.path.exists(path) + ): + if not self._is_pty: + return True + # PTY symlink may have been retargeted while our fd still points at a + # deleted slave — force reopen when the live path inode differs. + try: + cur = os.stat(path) + fd_st = os.fstat(self._fd) + if cur.st_ino == fd_st.st_ino and cur.st_dev == fd_st.st_dev: + return True + except OSError: + pass + was_active = self._kiss_active or force + call = (mycall or self._mycall or "").upper() + self.close() + if not self.open(): + return False + if was_active and call: + return self.attach_session(call) + return self._fd is not None + + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + if self._fd is None or not self._kiss_active: + return False, "KISS not ready" + if len(text.encode("utf-8")) > MAX_PAYLOAD: + return False, "payload too long" + info = text.encode("utf-8") + frame = ax25_build_ui(src, dst, info) + pkt = kiss_data_frame(0, frame) + with self._lock: + try: + os.write(self._fd, pkt) + # PTY: never tcdrain — if the master side dies, drain can hang forever. + if not self._is_pty: + termios.tcdrain(self._fd) + except OSError as exc: + self.status = "error-tx" + return False, f"tx failed: {exc}" + display = format_rx_line(src, dst, info, ax25_ui) + return True, display + + def _rx_loop(self) -> None: + while not self._stop.is_set(): + fd = self._fd + if fd is None: + break + try: + chunk = os.read(fd, 4096) + except BlockingIOError: + time.sleep(0.05) + continue + except OSError: + break + if not chunk: + time.sleep(0.05) + continue + for _port, payload in self._decoder.feed(chunk): + if not payload: + continue + parsed = ax25_parse_ui(payload) + if parsed is None: + continue + src, dst, info = parsed + line = format_rx_line(src, dst, info, ax25_ui=True) + self._on_rx(line) + + +class BayComKissBackend(KissRawBackend): + """BayCom kernel modem (SER12 / PAR96) via baycom-pr KISS PTY.""" + + backend_type = "baycom-kiss" + + def __init__( + self, + cfg: DeviceBackendConfig, + on_rx: RxFn, + log: Optional[LogFn] = None, + ) -> None: + modem = cfg.baycom_modem or "a" + kiss = cfg.kiss_link or f"/var/run/baycom-pr/kiss-{modem}" + if modem == "a" and not cfg.kiss_link: + default = "/var/run/baycom-pr/kiss" + if os.path.exists(default) or not os.path.exists(kiss): + kiss = default + profile = SerialProfile(baud=9600, line="8n1", dtr_rts=False) + super().__init__(cfg, kiss, profile, on_rx, log, is_pty=True) + + +class Max25BcprKissBackend(KissRawBackend): + """max25-bcpr userspace SER12 via KISS PTY (max25e0 / max25e0:bcN). + + Hardware is a TCM3105-class AFSK modem chip (bits↔AFSK + PTT) only — not a TNC. + """ + + backend_type = "max25-bcpr-kiss" + + def __init__( + self, + cfg: DeviceBackendConfig, + on_rx: RxFn, + log: Optional[LogFn] = None, + ) -> None: + tag = (cfg.max25_bcpr_device or cfg.bcpr_device or "bc0").strip() or "bc0" + kiss = cfg.kiss_link or f"/tmp/max25-bcpr/kiss-{tag}" + kiss = normalize_max25_bcpr_path(kiss) + profile = SerialProfile(baud=9600, line="8n1", dtr_rts=False) + super().__init__(cfg, kiss, profile, on_rx, log, is_pty=True) + + +class KissRawSerialBackend(KissRawBackend): + """USB/async BayCom KISS serial (kiss-serial backend).""" + + backend_type = "kiss-raw-serial" + + def __init__( + self, + cfg: DeviceBackendConfig, + root: str, + on_rx: RxFn, + log: Optional[LogFn] = None, + prefix: Optional[str] = None, + ) -> None: + overrides: dict[str, str] = {} + if cfg.serial_device: + overrides["device"] = cfg.serial_device + if cfg.serial_baud: + overrides["baud"] = str(cfg.serial_baud) + if cfg.serial_line: + overrides["line"] = cfg.serial_line + if cfg.serial_dtr_rts: + overrides["dtr_rts"] = cfg.serial_dtr_rts + prof = serial_profile_for_device(cfg.device_id, root, overrides, prefix=prefix) + path = cfg.serial_device or prof.device + super().__init__(cfg, path, prof, on_rx, log, is_pty=False) + + +class CrdopTcpBackend(DeviceBackend): + """MAX25-SoftModem (CRDOP) via TCP host interface (:8515 / :8516). + + Native M25/KISS host protocol (MAX25-SoftModem) only. + """ + + backend_type = "crdop-tcp" + + def __init__( + self, + cfg: DeviceBackendConfig, + on_rx: RxFn, + log: Optional[LogFn] = None, + ) -> None: + self.device_id = cfg.device_id + self._cfg = cfg + self._on_rx = on_rx + self._log = log or (lambda _m: None) + self._ctrl: Optional[socket.socket] = None + self._data: Optional[socket.socket] = None + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.RLock() + self._mycall = "" + self._connected = False + self.status = "closed" + + def _line_term(self) -> str: + return "\n" + + def open(self) -> bool: + host = self._cfg.crdop_host + port = self._cfg.crdop_port + ctrl = None + data = None + try: + ctrl = socket.create_connection((host, port), timeout=5.0) + ctrl.settimeout(0.5) + data = socket.create_connection((host, port + 1), timeout=5.0) + data.settimeout(0.5) + except OSError as exc: + if ctrl is not None: + try: + ctrl.close() + except OSError: + pass + self.status = "error-connect" + self._log(f"{self.device_id}: CRDOP TCP connect failed ({host}:{port}): {exc}") + return False + self._ctrl = ctrl + self._data = data + self.status = "open" + self._stop.clear() + self._thread = threading.Thread( + target=self._rx_loop, + name=f"crdop-rx-{self.device_id}", + daemon=True, + ) + self._thread.start() + self._log(f"{self.device_id}: CRDOP TCP open {host}:{port}") + return True + + def close(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + for sock in (self._ctrl, self._data): + if sock is not None: + try: + sock.close() + except OSError: + pass + self._ctrl = None + self._data = None + self._connected = False + self.status = "closed" + + def _cmd(self, text: str) -> str: + if self._ctrl is None: + return "" + term = self._line_term() + payload = (text.rstrip(term) + term).encode("ascii", errors="replace") + with self._lock: + self._ctrl.sendall(payload) + return self._read_line_unlocked() + + def _read_line_unlocked(self) -> str: + if self._ctrl is None: + return "" + term = self._line_term() + term_b = term.encode("ascii") + buf = b"" + deadline = time.time() + 3.0 + while time.time() < deadline: + try: + chunk = self._ctrl.recv(4096) + except socket.timeout: + continue + except OSError: + break + if not chunk: + break + buf += chunk + while term_b in buf: + raw, buf = buf.split(term_b, 1) + line = raw.decode("ascii", errors="replace").strip() + if line: + return line + return "" + + def attach_session(self, mycall: str) -> bool: + if self._ctrl is None: + return False + self._mycall = mycall.upper() + cmds = [ + "INITIALIZE", + "PROTOCOLMODE KISS", + f"MYCALL {self._mycall}", + ] + if self._cfg.crdop_listen: + cmds.append("LISTEN TRUE") + for cmd in cmds: + reply = self._cmd(cmd) + self._log(f"{self.device_id}: {cmd} → {reply or '(no reply)'}") + self._connected = True + self.status = "ready" + return True + + def detach_session(self) -> None: + if self._ctrl is not None and self._connected: + self._cmd("ABORT") + self._connected = False + if self._ctrl is not None: + self.status = "open" + + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + if self._ctrl is None or self._data is None or not self._connected: + return False, "CRDOP not ready" + payload = text.encode("utf-8") + if len(payload) > MAX_PAYLOAD: + return False, "payload too long" + with self._lock: + try: + body = ax25_build_ui(src, dst, payload) + if len(body) >= 2: + body = body[:-2] + self._data.sendall(body) + except OSError as exc: + self.status = "error-tx" + return False, f"tx failed: {exc}" + if ax25_ui: + display = f"[CRDOP AX25 UI {src}>{dst}] {text}" + else: + display = text + return True, display + + def _rx_loop(self) -> None: + while not self._stop.is_set(): + ctrl = self._ctrl + if ctrl is None: + break + try: + ready, _, _ = select.select([ctrl], [], [], 0.5) + if not ready: + continue + chunk = ctrl.recv(4096) + except (OSError, socket.timeout): + continue + if not chunk: + time.sleep(0.05) + continue + term = self._line_term() + for line in chunk.decode("ascii", errors="replace").split(term): + line = line.strip() + if not line: + continue + if line.startswith("STATUS"): + self._on_rx(f"[CRDOP RX {self.device_id}] {line}") + + +class AudioDummyBackend(DeviceBackend): + """Acoustic bench dummy — loopback, ALSA sniff, or M25 host TCP.""" + + backend_type = "audio-dummy" + + def __init__( + self, + cfg: DeviceBackendConfig, + on_rx: RxFn, + log: Optional[LogFn] = None, + ) -> None: + self.device_id = cfg.device_id + self._cfg = cfg + self._on_rx = on_rx + self._log = log or (lambda _m: None) + self._ctrl: Optional[socket.socket] = None + self._data: Optional[socket.socket] = None + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.RLock() + self._mycall = "" + self._connected = False + self._engine = None + self.status = "closed" + + def _import_engine(self): + import sys + from pathlib import Path + + lib = Path(__file__).resolve().parents[1] / "crdop" / "lib" + if str(lib) not in sys.path: + sys.path.insert(0, str(lib)) + from acoustic_engine import AcousticEngine # noqa: WPS433 + from sound_proxy import SoundConfig # noqa: WPS433 + + sound = SoundConfig( + capture=self._cfg.audio_capture or "default", + playback=self._cfg.audio_playback or self._cfg.audio_capture or "default", + sample_rate=self._cfg.audio_sample_rate, + ) + return AcousticEngine(sample_rate=self._cfg.audio_sample_rate, sound=sound) + + def open(self) -> bool: + mode = (self._cfg.audio_mode or "loopback").lower() + if mode == "host": + host = "127.0.0.1" + port = self._cfg.audio_host_port + ctrl = None + data = None + try: + ctrl = socket.create_connection((host, port), timeout=3.0) + ctrl.settimeout(0.5) + data = socket.create_connection((host, port + 1), timeout=3.0) + data.settimeout(0.5) + except OSError as exc: + if ctrl is not None: + try: + ctrl.close() + except OSError: + pass + self.status = "error-connect" + self._log(f"{self.device_id}: audio-dummy host connect failed: {exc}") + return False + self._ctrl = ctrl + self._data = data + self._stop.clear() + self._thread = threading.Thread( + target=self._host_rx_loop, + name=f"audio-dummy-{self.device_id}", + daemon=True, + ) + self._thread.start() + self.status = "open" + self._log(f"{self.device_id}: audio-dummy host {host}:{port}") + return True + + try: + self._engine = self._import_engine() + except Exception as exc: + self.status = "error-engine" + self._log(f"{self.device_id}: audio engine load failed: {exc}") + return False + + if mode == "alsa" and self._cfg.audio_capture: + self._stop.clear() + self._thread = threading.Thread( + target=self._alsa_sniff_loop, + name=f"audio-sniff-{self.device_id}", + daemon=True, + ) + self._thread.start() + self.status = "open" + self._log(f"{self.device_id}: audio-dummy mode={mode}") + return True + + def close(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + for sock in (self._ctrl, self._data): + if sock is not None: + try: + sock.close() + except OSError: + pass + self._ctrl = None + self._data = None + self._connected = False + self.status = "closed" + + def _host_cmd(self, text: str) -> str: + if self._ctrl is None: + return "" + payload = (text.rstrip("\n") + "\n").encode("ascii", errors="replace") + with self._lock: + self._ctrl.sendall(payload) + buf = b"" + deadline = time.time() + 2.0 + while time.time() < deadline: + try: + chunk = self._ctrl.recv(4096) + except socket.timeout: + continue + if not chunk: + break + buf += chunk + if b"\n" in buf: + line, _ = buf.split(b"\n", 1) + return line.decode("ascii", errors="replace").strip() + return "" + + def attach_session(self, mycall: str) -> bool: + self._mycall = mycall.upper() + if self._ctrl is not None: + for cmd in ( + "INITIALIZE", + "PROTOCOLMODE KISS", + f"MYCALL {self._mycall}", + "LISTEN TRUE", + ): + reply = self._host_cmd(cmd) + self._log(f"{self.device_id}: {cmd} → {reply or '(no reply)'}") + self._connected = True + self.status = "ready" + return True + + def detach_session(self) -> None: + self._connected = False + if self._ctrl is not None: + self.status = "open" + else: + self.status = "closed" + + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + if not self._connected: + return False, "audio-dummy not ready" + payload = text.encode("utf-8") + if len(payload) > MAX_PAYLOAD: + return False, "payload too long" + + if self._engine is not None: + pcm = self._engine.encode_ax25_ui(src, dst, text) + rep = self._engine.analyze_pcm(pcm) + for line in rep.decode_lines: + self._on_rx(f"[AUDIO RX {self.device_id}] {line}") + display = f"[AX25 UI {src}>{dst}] {text}" if ax25_ui else text + return True, display + + if self._data is None: + return False, "no data channel" + import sys + from pathlib import Path + + lib = Path(__file__).resolve().parents[1] / "crdop" / "lib" + if str(lib) not in sys.path: + sys.path.insert(0, str(lib)) + from ax25_codec import ax25_build_ui # noqa: WPS433 + + body = ax25_build_ui(src, dst, payload) + if len(body) >= 2: + body = body[:-2] + try: + with self._lock: + self._data.sendall(body) + except OSError as exc: + return False, f"tx failed: {exc}" + display = f"[AX25 UI {src}>{dst}] {text}" if ax25_ui else text + return True, display + + def _alsa_sniff_loop(self) -> None: + import sys + from pathlib import Path + + lib = Path(__file__).resolve().parents[1] / "crdop" / "lib" + if str(lib) not in sys.path: + sys.path.insert(0, str(lib)) + from sound_proxy import SoundProxy # noqa: WPS433 + + if self._engine is None: + return + proxy = SoundProxy(self._engine.sound) + + def on_pcm(chunk: bytes) -> None: + rep = self._engine.analyze_pcm(chunk) + for line in rep.decode_lines: + self._on_rx(f"[SNIFF {self.device_id}] {line}") + + try: + proxy.sniff_loop(chunk_symbols=40, on_pcm=on_pcm, stop=self._stop) + except Exception as exc: + self._log(f"{self.device_id}: sniff error: {exc}") + + def _host_rx_loop(self) -> None: + while not self._stop.is_set(): + if self._ctrl is None: + break + try: + ready, _, _ = select.select([self._ctrl], [], [], 0.5) + if not ready: + continue + chunk = self._ctrl.recv(4096) + except (OSError, socket.timeout): + continue + if not chunk: + time.sleep(0.05) + continue + for line in chunk.decode("ascii", errors="replace").split("\n"): + line = line.strip() + if line.startswith("STATUS"): + self._on_rx(f"[AUDIO RX {self.device_id}] {line}") + + +def parse_device_spec(device_id: str, spec: str, cp, cfg_defaults: dict) -> DeviceBackendConfig: + """Build backend config from [devices] value and optional [device.<id>].""" + dev = DeviceBackendConfig(device_id=device_id) + dev.hardware = registry_hardware(device_id, cfg_defaults.get("hardware", "tncs")) + dev.backend_type = registry_backend(device_id) + + section = f"device.{device_id}" + sec_opts: dict[str, str] = {} + if cp.has_section(section): + sec_opts = {k: cp.get(section, k) for k in cp.options(section)} + + if sec_opts.get("hardware"): + dev.hardware = sec_opts["hardware"] + if sec_opts.get("backend"): + dev.backend_type = sec_opts["backend"] + + spec = (spec or "").strip() + dev.device_spec = spec + + if spec.startswith("baycom:"): + dev.backend_type = "baycom-kiss" + dev.baycom_modem = spec.split(":", 1)[1].strip() or "a" + dev.hardware = "modems" + elif spec.startswith("max25-bcpr:") or spec.startswith("bcpr:"): + # Userspace SER12 — product face max25-bcpr; device id remains max25e0 + dev.backend_type = "max25-bcpr-kiss" + tag = spec.split(":", 1)[1].strip() or "bc0" + dev.max25_bcpr_device = tag + dev.bcpr_device = tag # legacy alias + dev.hardware = "modems" + if not sec_opts.get("kiss_link"): + dev.kiss_link = f"/tmp/max25-bcpr/kiss-{tag}" + elif spec.startswith("crdop:"): + dev.backend_type = "crdop-tcp" + dev.crdop_profile = spec.split(":", 1)[1].strip() or "default" + dev.hardware = "soft-modems" + elif spec.startswith("audio:"): + dev.backend_type = "audio-dummy" + dev.audio_mode = spec.split(":", 1)[1].strip() or "loopback" + dev.hardware = "acoustic-bench" + elif spec.startswith("/") or spec.startswith("dev:"): + if dev.backend_type in ("baycom-kiss",): + dev.kiss_link = spec + else: + dev.serial_device = spec.removeprefix("dev:") + if dev.backend_type == "crdop-tcp": + dev.backend_type = "kiss-serial" + elif spec: + dev.serial_device = spec + + if sec_opts.get("kiss_link"): + dev.kiss_link = sec_opts["kiss_link"] + if dev.backend_type in ("max25-bcpr-kiss", "bcpr-kiss") and dev.kiss_link: + dev.kiss_link = normalize_max25_bcpr_path(dev.kiss_link) + if sec_opts.get("modem"): + dev.baycom_modem = sec_opts["modem"] + if sec_opts.get("baycom_ini"): + dev.baycom_ini = sec_opts["baycom_ini"] + if sec_opts.get("max25_bcpr_ini") or sec_opts.get("bcpr_ini"): + ini_path = sec_opts.get("max25_bcpr_ini") or sec_opts.get("bcpr_ini") or "" + dev.max25_bcpr_ini = ini_path + dev.bcpr_ini = ini_path + if sec_opts.get("max25_bcpr_device") or sec_opts.get("bcpr_device"): + tag = sec_opts.get("max25_bcpr_device") or sec_opts.get("bcpr_device") or "" + dev.max25_bcpr_device = tag + dev.bcpr_device = tag + if sec_opts.get("ipv4"): + dev.ipv4 = sec_opts["ipv4"].strip() + if sec_opts.get("ipv6"): + dev.ipv6 = sec_opts["ipv6"].strip() + if sec_opts.get("host"): + dev.crdop_host = sec_opts["host"] + if sec_opts.get("port"): + dev.crdop_port = _spec_int(sec_opts["port"], dev.crdop_port) + if sec_opts.get("listen"): + dev.crdop_listen = sec_opts["listen"].lower() in ("1", "yes", "true", "on") + if sec_opts.get("mode"): + dev.audio_mode = sec_opts["mode"] + if sec_opts.get("capture"): + dev.audio_capture = sec_opts["capture"] + if sec_opts.get("playback"): + dev.audio_playback = sec_opts["playback"] + if sec_opts.get("sample_rate"): + dev.audio_sample_rate = _spec_int(sec_opts["sample_rate"], dev.audio_sample_rate) + if sec_opts.get("host_port"): + dev.audio_host_port = _spec_int(sec_opts["host_port"], dev.audio_host_port) + + serial_sec = f"serial.{device_id}" + if cp.has_section(serial_sec): + if cp.has_option(serial_sec, "device"): + dev.serial_device = cp.get(serial_sec, "device") + if cp.has_option(serial_sec, "baud"): + dev.serial_baud = cp.getint(serial_sec, "baud") + if cp.has_option(serial_sec, "line"): + dev.serial_line = cp.get(serial_sec, "line") + if cp.has_option(serial_sec, "dtr_rts"): + dev.serial_dtr_rts = cp.get(serial_sec, "dtr_rts") + if cp.has_option(serial_sec, "kiss_entry"): + dev.serial_kiss_entry = cp.get(serial_sec, "kiss_entry") + + # max25e0 family: hardcoded host addresses (forks inherit from root max25e0) + if device_id == "max25e0" or device_id.startswith("max25e0:"): + root_opts: dict[str, str] = {} + if device_id != "max25e0" and cp.has_section("device.max25e0"): + root_opts = {k: cp.get("device.max25e0", k) for k in cp.options("device.max25e0")} + if not dev.ipv4: + dev.ipv4 = (root_opts.get("ipv4") or "").strip() or MAX25E0_DEFAULT_IPV4 + if not dev.ipv6: + dev.ipv6 = (root_opts.get("ipv6") or "").strip() or MAX25E0_DEFAULT_IPV6 + + return dev + + +def create_backend( + dev_cfg: DeviceBackendConfig, + root: str, + on_rx: RxFn, + log: Optional[LogFn] = None, + prefix: Optional[str] = None, + on_invalid: Optional[InvalidFn] = None, +) -> DeviceBackend: + kind = dev_cfg.backend_type or registry_backend(dev_cfg.device_id) + if kind == "kiss-serial": + return KissSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix, on_invalid=on_invalid) + if kind == "baycom-kiss": + return BayComKissBackend(dev_cfg, on_rx, log) + if kind in ("max25-bcpr-kiss", "bcpr-kiss"): + return Max25BcprKissBackend(dev_cfg, on_rx, log) + if kind == "kiss-raw-serial": + return KissRawSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix) + if kind == "crdop-tcp": + return CrdopTcpBackend(dev_cfg, on_rx, log) + if kind == "audio-dummy": + return AudioDummyBackend(dev_cfg, on_rx, log) + return KissSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix) + + +def backend_needs_stack(kind: str) -> bool: + return kind in ( + "kiss-serial", + "baycom-kiss", + "max25-bcpr-kiss", "bcpr-kiss", + "kiss-raw-serial", + "crdop-tcp", + "audio-dummy", + ) + + +def backend_serial_label(backend: Optional[DeviceBackend]) -> str: + if backend is None: + return "n/a" + return backend.status + + +# Legacy alias (tests / transitional) +BcprKissBackend = Max25BcprKissBackend diff --git a/stacks/daemon/kiss_bridge.py b/stacks/daemon/kiss_bridge.py new file mode 100644 index 0000000..b433d0e --- /dev/null +++ b/stacks/daemon/kiss_bridge.py @@ -0,0 +1,574 @@ +""" +KISS serial bridge for max25d — AX.25 UI over TNC2C / PK-TNC2. + +PTT: TNC firmware keys on KISS DATA (requires MYCALL); kernel BayCom (baycom_ser_fdx) +keys RTS in the driver when the KISS bridge accepts a DATA frame — no max25d PTT command. +""" +from __future__ import annotations + +import fcntl +import os +import struct +import termios +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, Optional + +from ax25_codec import ( # noqa: E402 + ax25_build_ui, + ax25_crc, + ax25_crc_valid, + ax25_parse_ui, + format_callsign, + parse_callsign, + validate_callsign, +) +from tx_pace import tx_pace_before_send # noqa: E402 + +FEND = 0xC0 +FESC = 0xDB +TFEND = 0xDC +TFESC = 0xDD + +# Native KISS return (TheFirmware TF 2.7) — firmware reset to banner, not TAPR kiss off +KISS_RETURN_FRAME = b"\xc0\xff\xc0" + +KISS_CMD_DATA = 0x00 +MAX_FRAME = 1024 +MAX_PAYLOAD = 256 + + +@dataclass +class SerialProfile: + device: str = "/dev/ttyS4" + baud: int = 19200 + line: str = "8n1" + dtr_rts: bool = True + kiss_entry: str = "kiss_on" # kiss_on | auto + + +class KissDecoder: + def __init__(self) -> None: + self._buf = bytearray() + self._in_frame = False + self._escape = False + + def feed(self, data: bytes) -> list[tuple[int, bytes]]: + frames: list[tuple[int, bytes]] = [] + for byte in data: + if byte == FEND: + if self._in_frame and self._buf: + parsed = self._deliver() + if parsed is not None: + frames.append(parsed) + self._in_frame = True + self._escape = False + self._buf.clear() + continue + if not self._in_frame: + continue + if self._escape: + if byte == TFEND: + byte = FEND + elif byte == TFESC: + byte = FESC + self._escape = False + elif byte == FESC: + self._escape = True + continue + if len(self._buf) >= MAX_FRAME: + self._buf.clear() + self._in_frame = False + continue + self._buf.append(byte) + return frames + + def _deliver(self) -> Optional[tuple[int, bytes]]: + if len(self._buf) < 1: + return None + cmd_byte = self._buf[0] + if (cmd_byte & 0x0F) != KISS_CMD_DATA: + return None + port = (cmd_byte >> 4) & 0x0F + payload = bytes(self._buf[1:]) + return port, payload + + +def kiss_escape(data: bytes) -> bytes: + out = bytearray() + for b in data: + if b == FEND: + out.extend((FESC, TFEND)) + elif b == FESC: + out.extend((FESC, TFESC)) + else: + out.append(b) + return bytes(out) + + +def kiss_encode(port: int, cmd: int, payload: bytes) -> bytes: + cmd_byte = ((port & 0x0F) << 4) | (cmd & 0x0F) + return b"\xC0" + kiss_escape(bytes([cmd_byte]) + payload) + b"\xC0" + + +def kiss_data_frame(port: int, ax25_frame: bytes) -> bytes: + """Build KISS DATA; strip FCS when CRC validates (ax25ipd / KISS convention).""" + if ax25_crc_valid(ax25_frame): + ax25_frame = ax25_frame[:-2] + return kiss_encode(port, KISS_CMD_DATA, ax25_frame) + + +def format_rx_line(src: str, dst: str, payload: bytes, ax25_ui: bool) -> str: + try: + text = payload.decode("utf-8") + except UnicodeDecodeError: + text = payload.decode("utf-8", errors="replace") + if ax25_ui: + return f"[AX25 UI {src}>{dst}] {text}" + return text + + +def load_env_file(path: str) -> dict[str, str]: + out: dict[str, str] = {} + if not path or not os.path.isfile(path): + return out + with open(path, encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + key, val = line.split("=", 1) + out[key.strip()] = val.strip() + return out + + +def _load_serial_env(device_id: str, root: str, prefix: Optional[str] = None) -> dict[str, str]: + from paths import serial_env_candidates + + tree = Path(root) + pref = Path(prefix) if prefix else None + for path in serial_env_candidates(device_id, tree, pref): + env = load_env_file(str(path)) + if env: + return env + return {} + + +def serial_profile_for_device( + device_id: str, + root: str, + ini: dict[str, str], + prefix: Optional[str] = None, +) -> SerialProfile: + prof = SerialProfile() + if ini.get("device"): + prof.device = ini["device"] + if ini.get("baud"): + prof.baud = int(ini["baud"]) + if ini.get("line"): + prof.line = ini["line"].lower() + if ini.get("dtr_rts"): + prof.dtr_rts = ini["dtr_rts"].lower() in ("1", "yes", "true", "on") + if ini.get("kiss_entry"): + prof.kiss_entry = ini["kiss_entry"].lower() + + env = _load_serial_env(device_id, root, prefix) + + if device_id == "tnc2c": + prof.device = ini.get("device") or env.get("TNC2C_DEV", prof.device) + prof.baud = int(ini.get("baud") or env.get("TNC2C_BAUD", prof.baud)) + prof.line = (ini.get("line") or env.get("TNC2C_LINE", prof.line)).lower() + if "dtr_rts" not in ini: + prof.dtr_rts = True + if "kiss_entry" not in ini: + prof.kiss_entry = "kiss_on" + elif device_id == "pktnc2": + prof.device = ini.get("device") or env.get("PKTNC2_DEV") or env.get("TNC_DEV", prof.device) + prof.baud = int(ini.get("baud") or env.get("PKTNC2_BAUD") or env.get("TNC_BAUD", "9600")) + prof.line = (ini.get("line") or env.get("PKTNC2_LINE") or env.get("TNC_LINE", "8n1")).lower() + if "dtr_rts" not in ini: + prof.dtr_rts = False + if "kiss_entry" not in ini: + prof.kiss_entry = "auto" + elif device_id in ("pccom-kiss", "baycom-kiss"): + prof.device = ini.get("device") or env.get("PCCOM_KISS_DEV", prof.device or "/dev/ttyUSB0") + prof.baud = int(ini.get("baud") or env.get("PCCOM_KISS_BAUD", "9600")) + prof.line = (ini.get("line") or env.get("PCCOM_KISS_LINE", "8n1")).lower() + dtr = ini.get("dtr_rts") or env.get("PCCOM_KISS_DTR_RTS", "no") + prof.dtr_rts = str(dtr).lower() in ("1", "yes", "true", "on") + if "kiss_entry" not in ini: + prof.kiss_entry = "none" + elif device_id == "tmodem": + default_dev = "/dev/cuaU0" if sys.platform.startswith("freebsd") else "/dev/ttyACM0" + prof.device = ini.get("device") or env.get("TMODEM_DEV", prof.device or default_dev) + prof.baud = int(ini.get("baud") or env.get("TMODEM_BAUD", "115200")) + prof.line = (ini.get("line") or env.get("TMODEM_LINE", "8n1")).lower() + dtr = ini.get("dtr_rts") or env.get("TMODEM_DTR_RTS", "no") + prof.dtr_rts = str(dtr).lower() in ("1", "yes", "true", "on") + if "kiss_entry" not in ini: + prof.kiss_entry = "none" + return prof + + +def _parse_line(line: str) -> tuple[int, int]: + line = line.lower() + if line == "7e1": + return termios.CS7, termios.PARENB + return termios.CS8, 0 + + +def _parse_baud(baud: int) -> int: + table = { + 1200: termios.B1200, + 2400: termios.B2400, + 4800: termios.B4800, + 9600: termios.B9600, + 19200: termios.B19200, + } + if baud not in table: + raise ValueError(f"unsupported baud: {baud}") + return table[baud] + + +class KissBridge: + """Thread-safe KISS bridge on a serial TNC port.""" + + def __init__( + self, + profile: SerialProfile, + on_rx: Callable[[str], None], + log: Optional[Callable[[str], None]] = None, + *, + tree_root: str = "", + install_prefix: Optional[str] = None, + on_invalid: Optional[Callable[[], None]] = None, + ) -> None: + self.profile = profile + self._on_rx = on_rx + self._on_invalid = on_invalid + self._log = log or (lambda _m: None) + self._tree_root = tree_root + self._install_prefix = install_prefix + self._fd: Optional[int] = None + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.Lock() + self._kiss_active = False + self._mycall = "" + self.status = "closed" + self._decoder = KissDecoder() + + def open(self) -> bool: + dev = self.profile.device + if not os.access(dev, os.R_OK | os.W_OK): + self.status = "error-no-device" + self._log(f"serial: no access to {dev}") + return False + try: + speed = _parse_baud(self.profile.baud) + databits, parity = _parse_line(self.profile.line) + except ValueError as exc: + self.status = "error-config" + self._log(f"serial: {exc}") + return False + try: + fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) + t = termios.tcgetattr(fd) + t[0] = t[1] = 0 + t[2] = termios.CLOCAL | termios.CREAD | databits | parity + t[3] = t[4] = t[5] = speed + t[6][termios.VMIN] = 0 + t[6][termios.VTIME] = 5 + termios.tcsetattr(fd, termios.TCSANOW, t) + termios.tcflush(fd, termios.TCIOFLUSH) + flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0] + if self.profile.dtr_rts: + flags |= 0x004 | 0x002 + fcntl.ioctl(fd, 0x5416, struct.pack("I", flags)) + except OSError as exc: + self.status = "error-open" + self._log(f"serial open failed: {exc}") + return False + self._fd = fd + self._log(f"serial open {dev} {self.profile.baud} {self.profile.line.upper()}") + if self.profile.dtr_rts: + time.sleep(2.0) + self._log("serial: DTR settle (2s)") + self.status = "open" + return True + + def close(self) -> None: + self._stop_rx_thread() + with self._lock: + if self._fd is not None: + if self._kiss_active: + self._write_unlocked(KISS_RETURN_FRAME) + try: + os.close(self._fd) + except OSError: + pass + self._fd = None + self._kiss_active = False + self.status = "closed" + self._decoder = KissDecoder() + + def _stop_rx_thread(self) -> None: + """Stop KISS RX thread so recovery owns the serial FD exclusively.""" + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + self._stop.clear() + + def _start_rx_thread(self) -> None: + """Start KISS RX after terminal recovery and KISS entry succeed.""" + if self._fd is None or self._thread is not None or not self._kiss_active: + return + self._stop.clear() + self._thread = threading.Thread(target=self._rx_loop, name="kiss-rx", daemon=True) + self._thread.start() + + def attach_session(self, mycall: str) -> bool: + if self._fd is None: + return False + self._mycall = mycall.upper() + self._stop_rx_thread() + with self._lock: + ok = self._stabilize_unlocked(self._mycall, force_ladder=False) + if ok: + self._start_rx_thread() + return ok + + def stabilize_session(self, mycall: str, *, force: bool = False) -> bool: + """Probe terminal/KISS health and repair without closing the port (keeps DTR).""" + if self._fd is None: + self.status = "error-open" + return False + if not force and self._kiss_active and self.status == "ready": + return True + self._mycall = mycall.upper() + self._stop_rx_thread() + with self._lock: + ok = self._stabilize_unlocked(self._mycall, force_ladder=force) + if ok: + self._start_rx_thread() + return ok + + def _load_recovery_mod(self): + import importlib.util + + from paths import tnc_serial_recovery_path + + path = None + if self._tree_root: + prefix = Path(self._install_prefix) if self._install_prefix else None + path = tnc_serial_recovery_path(Path(self._tree_root), prefix) + if path is None: + path = Path(__file__).resolve().parents[1] / "tncs" / "tnc_serial_recovery.py" + if not path.is_file(): + return None + spec = importlib.util.spec_from_file_location("tnc_serial_recovery", path) + if spec is None or spec.loader is None: + return None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def _recovery_io(self) -> tuple[Callable[[bytes], None], Callable[[float], bytes]]: + def wf(data: bytes) -> None: + self._write_unlocked(data) + + def rf(seconds: float) -> bytes: + return self._drain_unlocked(seconds) + + return wf, rf + + def _leave_kiss_unlocked(self) -> None: + if not self._kiss_active: + return + self._write_unlocked(KISS_RETURN_FRAME) + time.sleep(0.3) + self._drain_unlocked(0.2) + self._kiss_active = False + + def _enter_kiss_session_unlocked(self) -> bool: + if not self._set_mycall_unlocked(self._mycall): + self._log("serial: MYCALL may have failed") + if not self._enter_kiss_unlocked(): + self.status = "error-kiss" + self._kiss_active = False + return False + self._kiss_active = True + self.status = "ready" + return True + + def _stabilize_unlocked(self, mycall: str, *, force_ladder: bool) -> bool: + """Host probe, optional recovery ladder, MYCALL + KISS — port stays open.""" + self._mycall = mycall.upper() + try: + mod = self._load_recovery_mod() + wf, rf = self._recovery_io() + self._leave_kiss_unlocked() + + if mod is None: + self._log("serial: tnc_serial_recovery.py not found") + self.status = "error-config" + self._kiss_active = False + return False + + ok, probe_data, only_echo = mod.probe_info(wf, rf, pause=0.25) + if not ok or only_echo or force_ladder: + self._log( + f"serial: initial probe — {mod.format_rx_brief(probe_data)}, " + f"echo_only={only_echo}, banner={mod.has_banner(probe_data)}" + ) + if ok and not only_echo and not force_ladder: + return self._enter_kiss_session_unlocked() + if only_echo or not ok or force_ladder: + label = "auto-repair" if force_ladder else "recovery ladder" + self._log(f"serial: {label}") + ok, _ = mod.recover_terminal(wf, rf, log=self._log) + if not ok: + self.status = "error-host" + self._kiss_active = False + return False + return self._enter_kiss_session_unlocked() + self.status = "error-host" + self._kiss_active = False + return False + except OSError as exc: + self.status = "error-io" + self._kiss_active = False + self._log(f"serial: stabilize I/O error ({exc})") + return False + + def _recover_terminal_unlocked(self) -> bool: + """Legacy hook — full ladder when probe fails.""" + try: + mod = self._load_recovery_mod() + if mod is None: + return True + wf, rf = self._recovery_io() + ok, _, only_echo = mod.probe_info(wf, rf) + if ok and not only_echo: + self._log("serial: terminal mode OK") + return True + self._log("serial: software recovery ladder") + ok, _ = mod.recover_terminal(wf, rf, log=self._log) + if ok: + self._log("serial: recovery OK") + return ok + except Exception as exc: + self._log(f"serial: recovery skipped ({exc})") + return True + + def detach_session(self) -> None: + self._stop_rx_thread() + with self._lock: + if self._fd is not None and self._kiss_active: + self._write_unlocked(KISS_RETURN_FRAME) + time.sleep(0.2) + self._kiss_active = False + if self._fd is not None: + self.status = "open" + + def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]: + if self._fd is None or not self._kiss_active: + return False, "serial not ready" + if len(text.encode("utf-8")) > MAX_PAYLOAD: + return False, "payload too long" + try: + validate_callsign(src) + validate_callsign(dst) + except ValueError as exc: + return False, str(exc) + info = text.encode("utf-8") + try: + frame = ax25_build_ui(src, dst, info) + except ValueError as exc: + return False, str(exc) + pkt = kiss_data_frame(0, frame) + with self._lock: + try: + tx_pace_before_send() + self._write_unlocked(pkt) + termios.tcdrain(self._fd) + except OSError as exc: + self.status = "error-tx" + return False, f"tx failed: {exc}" + display = format_rx_line(src, dst, info, ax25_ui) + return True, display + + def _write_unlocked(self, data: bytes) -> None: + if self._fd is None: + return + os.write(self._fd, data) + + def _drain_unlocked(self, seconds: float) -> bytes: + if self._fd is None: + return b"" + end = time.time() + seconds + chunks: list[bytes] = [] + while time.time() < end: + try: + chunk = os.read(self._fd, 4096) + if chunk: + chunks.append(chunk) + except BlockingIOError: + time.sleep(0.02) + return b"".join(chunks) + + def _set_mycall_unlocked(self, call: str) -> bool: + mod = self._load_recovery_mod() + if mod is not None and hasattr(mod, "tf_mycall_frame"): + cmd = mod.tf_mycall_frame(call) + else: + cmd = f"\x1bI {call.upper()}\r".encode("ascii", errors="replace") + self._write_unlocked(cmd) + time.sleep(0.4) + reply = self._drain_unlocked(0.6) + return b"?" not in reply[:32] + + def _enter_kiss_unlocked(self) -> bool: + entry = self.profile.kiss_entry + if entry == "tapr": + self._write_unlocked(b"kiss on\r") + time.sleep(0.5) + self._drain_unlocked(0.3) + return True + self._write_unlocked(b"\x1b@K") + time.sleep(0.5) + self._drain_unlocked(0.3) + return True + + def _rx_loop(self) -> None: + while not self._stop.is_set(): + fd = self._fd + if fd is None: + break + try: + chunk = os.read(fd, 4096) + except BlockingIOError: + time.sleep(0.05) + continue + except OSError: + self._log("serial: rx I/O error — watch will repair") + self.status = "error-io" + break + if not chunk: + time.sleep(0.05) + continue + for _port, payload in self._decoder.feed(chunk): + if not payload: + continue + parsed = ax25_parse_ui(payload) + if parsed is None: + if self._on_invalid is not None and len(payload) >= 16: + self._on_invalid() + continue + src, dst, info = parsed + line = format_rx_line(src, dst, info, ax25_ui=True) + self._on_rx(line) diff --git a/stacks/daemon/max25_platform.py b/stacks/daemon/max25_platform.py new file mode 100644 index 0000000..2148fae --- /dev/null +++ b/stacks/daemon/max25_platform.py @@ -0,0 +1,83 @@ +"""MAX25 host platform — Linux full stack, FreeBSD server + CRDOP/OSS.""" +from __future__ import annotations + +import os +import sys + + +def system_name() -> str: + return sys.platform + + +def is_linux() -> bool: + return sys.platform == "linux" + + +def is_freebsd() -> bool: + return sys.platform.startswith("freebsd") + + +def is_bsd() -> bool: + return sys.platform.startswith(("freebsd", "openbsd", "netbsd", "darwin")) + + +def max25d_supported() -> bool: + """Daemon may run (full stack on Linux, server+CRDOP on FreeBSD).""" + return is_linux() or is_freebsd() + + +def default_unix_socket() -> str: + if is_linux(): + return "/run/max25/modem.sock" + if is_freebsd(): + return "/var/run/max25/modem.sock" + return "/tmp/max25/modem.sock" + + +def default_bans_file() -> str: + if is_linux(): + return "/var/lib/max25/bans.txt" + return "/var/db/max25/bans.txt" + + +def supported_device_ids() -> frozenset[str]: + if is_linux(): + return frozenset( + { + "tnc2c", + "pktnc2", + "tmodem", + "baycom-kiss", + "pccom-kiss", + # BayCom/based SER12 (max25-bcpr → device max25e0; not usable by default) + "max25e0", + "max25e0:bc0", + "max25e0:bc1", + "soft-crdop", + "audio-dummy", + } + ) + if is_freebsd(): + # USB KISS (tmodem/c1224) + CRDOP/OSS + max25-bcpr (SER12 via sysarch I386_SET_IOPERM) + return frozenset({"tmodem", "soft-crdop", "audio-dummy", "max25e0", "max25e0:bc0", "max25e0:bc1"}) + return frozenset({"soft-crdop"}) + + +def crdop_audio_backend() -> str: + """Host audio API for CRDOP sound-proxy.""" + env = os.environ.get("MAX25_AUDIO_BACKEND", "").strip().lower() + if env in ("alsa", "oss"): + return env + if is_linux(): + return "alsa" + if is_freebsd(): + return "oss" + return "alsa" + + +def platform_label() -> str: + if is_linux(): + return "Linux/KLinux" + if is_freebsd(): + return "FreeBSD" + return sys.platform diff --git a/stacks/daemon/max25d b/stacks/daemon/max25d new file mode 100755 index 0000000..d6c075b --- /dev/null +++ b/stacks/daemon/max25d @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# max25d launcher — ps(1) shows this path, not "python3 …/max25d". +set -euo pipefail +ROOT="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)" +case "${1:-}" in + session) + shift + for candidate in \ + "${ROOT}/../../scripts/max25d-session.sh" \ + "${ROOT}/../scripts/max25d-session.sh" \ + "$(command -v max25d-session 2>/dev/null || true)" + do + if [[ -n "${candidate}" && -x "${candidate}" ]]; then + exec "${candidate}" "$@" + fi + done + echo "max25d: max25d-session helper not found" >&2 + exit 1 + ;; +esac +exec -a "$0" python3 "${ROOT}/max25d.py" "$@" diff --git a/stacks/daemon/max25d.py b/stacks/daemon/max25d.py new file mode 100755 index 0000000..2488f8d --- /dev/null +++ b/stacks/daemon/max25d.py @@ -0,0 +1,2185 @@ +#!/usr/bin/env python3 +""" +max25d — MainAX25-Stack daemon. + +Linux/KLinux: full hardware stack. FreeBSD: server + CRDOP/OSS (modular TCP/IP service). +""" +from __future__ import annotations + +import argparse +import configparser +import os +import re +import select +import signal +import socket +import subprocess +import sys +import tempfile +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional, Set + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from device_backends import ( # noqa: E402 + DeviceBackend, + DeviceBackendConfig, + backend_serial_label, + baycom_ctl_device_id, + create_backend, + parse_device_spec, + registry_backend, + registry_tested, +) +from banlist import BanList, extract_ax25_source # noqa: E402 +from reporting_quality import DataQualityTracker, RxOutcome, classify_rx_line # noqa: E402 +from daemon_log import LOGGER, emit_startup_banner, emit_startup_complete # noqa: E402 +from kiss_bridge import KissBridge # noqa: E402 — tests patch this symbol +from paths import ( # noqa: E402 + ctl_path, + MAX25_BCPR_KISS_DEFAULT, + normalize_max25_bcpr_path, + resolve_baycom_ini, + resolve_layout, +) +from max25_platform import ( # noqa: E402 + default_bans_file, + default_unix_socket, + max25d_supported, + platform_label, + supported_device_ids, +) +from privilege_drop import RunAsConfig, drop_privileges_or_exit, parse_run_as # noqa: E402 +from modular_tcp_server import ModularTcpMainService, ModularTcpConfig, load_modular_tcp # noqa: E402 + +_EXE = Path(__file__).resolve() +TREE, PREFIX = resolve_layout(_EXE) +ROOT = TREE # dev checkout root or MAX25_ROOT / install prefix + +DEFAULT_TCP_PORT = 7325 +DEFAULT_UNIX = default_unix_socket() +M25_MAX_LINE_BUF = 65536 +CALLSIGN_RE = re.compile(r"^[A-Z0-9]{1,6}(-(1[0-5]|[0-9]))?$") +RESERVED_DEVICE_KEYS = frozenset({"default", "enabled"}) + + +def _device_is_baycom(dev: DeviceBackendConfig) -> bool: + if dev.device_id.startswith("baycom"): + return True + spec = (dev.device_spec or "").strip() + if spec.startswith("baycom:"): + return True + return dev.backend_type == "baycom-kiss" and dev.hardware == "modems" + + +def _device_is_pccom(dev: DeviceBackendConfig) -> bool: + if "pccom" in dev.device_id.lower(): + return True + ini = (dev.baycom_ini or "").lower() + return "pccom" in ini + + + +def _device_is_max25_bcpr(dev: DeviceBackendConfig) -> bool: + # Product device id = max25e0 (+ forks max25e0:bcN); backend = max25-bcpr + if dev.device_id == "max25e0" or dev.device_id.startswith("max25e0:"): + return True + spec = (dev.device_spec or "").strip() + if spec.startswith("max25-bcpr:") or spec.startswith("bcpr:"): + return True + return dev.backend_type in ("max25-bcpr-kiss", "bcpr-kiss") + + +def _device_is_tmodem(dev: DeviceBackendConfig) -> bool: + return dev.device_id == "tmodem" + + +def _device_allowed_by_features(dev: DeviceBackendConfig, cfg: DaemonConfig) -> bool: + if _device_is_baycom(dev) and not cfg.feature_baycom: + LOGGER.warn( + f"device {dev.device_id}: BayCom disabled — set [features] baycom=yes", + area="config", + ) + return False + if _device_is_pccom(dev) and not cfg.feature_pccom: + LOGGER.warn( + f"device {dev.device_id}: PC-COM disabled — set [features] pccom=yes", + area="config", + ) + return False + if _device_is_max25_bcpr(dev) and not cfg.feature_max25_bcpr: + LOGGER.warn( + f"device {dev.device_id}: max25-bcpr disabled — set [features] max25_bcpr=yes", + area="config", + ) + return False + if _device_is_tmodem(dev) and not cfg.feature_tmodem: + LOGGER.warn( + f"device {dev.device_id}: T-Modem disabled — set [features] tmodem=yes", + area="config", + ) + return False + return True + + +@dataclass +class TotConfig: + """Software TOT for BayCom/based (max25-bcpr) — host policy, not radio TOT.""" + + enabled: bool = True + max_key_sec: int = 25 + min_gap_sec: float = 1.5 + max_consecutive: int = 3 + max_bursts: int = 8 + recover_sec: int = 300 + + +@dataclass +class DaemonConfig: + mode: str = "standalone" + hardware: str = "tncs" + device: str = "tnc2c" + default_device: str = "" + devices: list[DeviceBackendConfig] = field(default_factory=list) + tcp_host: str = "0.0.0.0" + tcp_port: int = DEFAULT_TCP_PORT + unix_socket: str = DEFAULT_UNIX + tcp_password: str = "" + callerid: str = "CB-0" + callid: str = "QST" + ax25_ui: bool = True + auto_start: bool = True + serial_enabled: bool = True + serial_watch: bool = True + serial_watch_interval: int = 60 + serial_repair_cooldown: int = 20 + serial_watch_startup_grace: int = 45 + stack_recover_only: bool = True + stack_retry_interval: int = 120 + serial_bootwait_escalate: bool = True + serial_bootwait_escalate_after: int = 3 + serial_bootwait_escalate_cooldown: int = 300 + bans_file: str = field(default_factory=default_bans_file) + config_path: str = "" + feature_baycom: bool = True + feature_pccom: bool = True + feature_max25_bcpr: bool = True + feature_tmodem: bool = False + hybbx_release_attach: bool = False + run_user: str = "" + run_group: str = "" + run_uid: Optional[int] = None + run_gid: Optional[int] = None + report_error_transmissions: bool = True + report_voice_transmissions: bool = True + report_data_passes: int = 3 + report_data_quality_min: int = 50 + report_data_pass_seconds: int = 20 + # Legacy single-device [serial] overrides (used when [devices] absent). + serial_device: str = "" + serial_baud: int = 0 + serial_line: str = "" + serial_dtr_rts: str = "" + serial_kiss_entry: str = "" + modular_tcp: ModularTcpConfig = field(default_factory=ModularTcpConfig) + tot: TotConfig = field(default_factory=TotConfig) + + +@dataclass +class DeviceRuntime: + cfg: DeviceBackendConfig + backend: Optional[DeviceBackend] = None + stack_proc: Optional[subprocess.Popen] = None + stack_status: str = "stopped" + link_status: str = "n/a" + last_watch: float = 0.0 + last_repair: float = 0.0 + last_stack_retry: float = 0.0 + prep_done: bool = False + inline_repair_failures: int = 0 + last_bootwait_escalate: float = 0.0 + quality: DataQualityTracker = field(default_factory=DataQualityTracker) + tot_paused: bool = False + tot_paused_until: float = 0.0 + tot_trip_reason: str = "" + + +@dataclass +class DaemonState: + cfg: DaemonConfig + connected: bool = False + monitor_only: bool = False + selected_device: str = "" + devices: dict[str, DeviceRuntime] = field(default_factory=dict) + clients: Set[socket.socket] = field(default_factory=set) + bans: BanList = field(default_factory=BanList) + lock: threading.Lock = field(default_factory=threading.Lock) + started_at: float = 0.0 + + +def log(msg: str) -> None: + """Legacy log hook — structured stderr (human + machine readable).""" + LOGGER.emit_unstructured(msg) + + +def valid_callsign(value: str) -> bool: + return bool(value and CALLSIGN_RE.match(value.upper())) + + +def _truthy(value: str) -> bool: + return value.lower() in ("1", "yes", "true", "on") + + +def _ini_int( + cp: configparser.ConfigParser, + section: str, + key: str, + default: int, + *, + min_value: int = 1, + max_value: int = 86400, +) -> int: + if not cp.has_option(section, key): + return default + raw = cp.get(section, key) + try: + value = int(str(raw).strip()) + except (TypeError, ValueError): + LOGGER.warn( + f"[{section}] {key}={raw!r} invalid — using {default}", + area="config", + ) + return default + if value < min_value: + LOGGER.warn( + f"[{section}] {key}={value} below {min_value} — using {min_value}", + area="config", + ) + return min_value + if value > max_value: + LOGGER.warn( + f"[{section}] {key}={value} above {max_value} — using {max_value}", + area="config", + ) + return max_value + return value + + +def _safe_port(raw: str, default: int) -> int: + try: + port = int(str(raw).strip()) + except (TypeError, ValueError): + return default + if port < 1 or port > 65535: + return default + return port + + +def _serial_overrides_from_section(cp: configparser.ConfigParser, section: str) -> dict[str, str]: + if not cp.has_section(section): + return {} + out: dict[str, str] = {} + for key in ("device", "baud", "line", "dtr_rts", "kiss_entry"): + if cp.has_option(section, key): + out[key] = cp.get(section, key) + return out + + +def _apply_serial_overrides(dev: DeviceBackendConfig, overrides: dict[str, str]) -> None: + if overrides.get("device"): + dev.serial_device = overrides["device"] + if overrides.get("baud"): + dev.serial_baud = int(overrides["baud"]) + if overrides.get("line"): + dev.serial_line = overrides["line"] + if overrides.get("dtr_rts"): + dev.serial_dtr_rts = overrides["dtr_rts"] + if overrides.get("kiss_entry"): + dev.serial_kiss_entry = overrides["kiss_entry"] + + +def parse_devices(cp: configparser.ConfigParser, cfg: DaemonConfig) -> list[DeviceBackendConfig]: + """Build device list from [devices] or legacy [daemon] device= + [serial].""" + defaults = {"hardware": cfg.hardware} + if cp.has_section("devices"): + default_id = cp.get("devices", "default", fallback="").strip() + enabled_raw = cp.get("devices", "enabled", fallback="").strip() + enabled_set: Optional[set[str]] = None + if enabled_raw: + enabled_set = {x.strip() for x in enabled_raw.split(",") if x.strip()} + + entries: list[DeviceBackendConfig] = [] + for key in cp.options("devices"): + if key.lower() in RESERVED_DEVICE_KEYS: + continue + device_id = key.strip() + spec = cp.get("devices", key, fallback="").strip() + dev = parse_device_spec(device_id, spec, cp, defaults) + if enabled_set is not None: + dev.enabled = device_id in enabled_set + entries.append(dev) + + if not entries: + LOGGER.warn( + "[devices] section empty — falling back to legacy single device", + area="config", + ) + else: + if default_id: + cfg.default_device = default_id + if enabled_set is not None and default_id not in enabled_set: + enabled_entries = [d for d in entries if d.enabled] + if enabled_entries: + fallback = enabled_entries[0].device_id + LOGGER.warn( + f"[devices] default={default_id} not enabled — using {fallback}", + area="config", + ) + cfg.default_device = fallback + elif cfg.device: + cfg.default_device = cfg.device + else: + cfg.default_device = entries[0].device_id + return entries + + # Legacy single device + device_id = cfg.device or "tnc2c" + dev = parse_device_spec(device_id, "", cp, defaults) + legacy = _serial_overrides_from_section(cp, "serial") + if cfg.serial_device: + legacy.setdefault("device", cfg.serial_device) + if cfg.serial_baud: + legacy.setdefault("baud", str(cfg.serial_baud)) + if cfg.serial_line: + legacy.setdefault("line", cfg.serial_line) + if cfg.serial_dtr_rts: + legacy.setdefault("dtr_rts", cfg.serial_dtr_rts) + if cfg.serial_kiss_entry: + legacy.setdefault("kiss_entry", cfg.serial_kiss_entry) + _apply_serial_overrides(dev, legacy) + cfg.default_device = device_id + return [dev] + + +def _ini_float( + cp: configparser.ConfigParser, + section: str, + key: str, + default: float, + *, + min_value: float = 0.0, + max_value: float = 3600.0, +) -> float: + if not cp.has_option(section, key): + return default + raw = cp.get(section, key).strip() + try: + value = float(raw) + except ValueError: + LOGGER.warn(f"invalid float {section}.{key}={raw!r} — using {default}", area="config") + return default + if value < min_value: + return min_value + if value > max_value: + return max_value + return value + + +def load_config(path: Optional[Path]) -> DaemonConfig: + cfg = DaemonConfig() + if path is None: + for candidate in ( + Path(os.environ.get("MAX25D_INI", "")), + Path("/etc/max25/max25d.ini"), + ROOT / "share/max25/max25d.ini.example", + ): + if candidate and candidate.is_file(): + path = candidate + break + if path is None or not path.is_file(): + LOGGER.warn(f"using built-in defaults (no ini at {path})", area="config") + cfg.devices = [parse_device_spec(cfg.device, "", configparser.ConfigParser(), {"hardware": cfg.hardware})] + cfg.default_device = cfg.device + cfg.config_path = "" + return cfg + + cfg.config_path = str(path) + + cp = configparser.ConfigParser(strict=False) + cp.read(path) + if cp.has_section("daemon"): + cfg.mode = cp.get("daemon", "mode", fallback=cfg.mode) + cfg.hardware = cp.get("daemon", "hardware", fallback=cfg.hardware) + cfg.device = cp.get("daemon", "device", fallback=cfg.device) + run_as = parse_run_as(cp) + cfg.run_user = run_as.user + cfg.run_group = run_as.group + cfg.run_uid = run_as.uid + cfg.run_gid = run_as.gid + if cp.has_section("network"): + cfg.tcp_host = cp.get("network", "tcp_host", fallback=cfg.tcp_host) + cfg.tcp_port = _ini_int(cp, "network", "tcp_port", cfg.tcp_port, min_value=1, max_value=65535) + cfg.unix_socket = cp.get("network", "unix_socket", fallback=cfg.unix_socket) + cfg.tcp_password = cp.get("network", "tcp_password", fallback=cfg.tcp_password) + if cp.has_section("modem"): + cfg.callerid = cp.get("modem", "callerid", fallback=cfg.callerid).upper() + cfg.callid = cp.get("modem", "callid", fallback=cfg.callid).upper() + cfg.ax25_ui = _truthy(cp.get("modem", "ax25_ui", fallback="yes")) + cfg.bans_file = cp.get("modem", "bans_file", fallback=cfg.bans_file) + if cp.has_section("stack"): + cfg.auto_start = _truthy(cp.get("stack", "auto_start", fallback="yes")) + if cp.has_option("stack", "serial_watch"): + cfg.serial_watch = _truthy(cp.get("stack", "serial_watch")) + cfg.serial_watch_interval = _ini_int( + cp, "stack", "serial_watch_interval", cfg.serial_watch_interval + ) + cfg.serial_repair_cooldown = _ini_int( + cp, "stack", "serial_repair_cooldown", cfg.serial_repair_cooldown + ) + cfg.serial_watch_startup_grace = _ini_int( + cp, "stack", "serial_watch_startup_grace", cfg.serial_watch_startup_grace + ) + if cp.has_option("stack", "stack_recover_only"): + cfg.stack_recover_only = _truthy(cp.get("stack", "stack_recover_only")) + cfg.stack_retry_interval = _ini_int( + cp, "stack", "stack_retry_interval", cfg.stack_retry_interval + ) + if cp.has_option("stack", "serial_bootwait_escalate"): + cfg.serial_bootwait_escalate = _truthy(cp.get("stack", "serial_bootwait_escalate")) + cfg.serial_bootwait_escalate_after = _ini_int( + cp, "stack", "serial_bootwait_escalate_after", cfg.serial_bootwait_escalate_after + ) + cfg.serial_bootwait_escalate_cooldown = _ini_int( + cp, + "stack", + "serial_bootwait_escalate_cooldown", + cfg.serial_bootwait_escalate_cooldown, + ) + if cp.has_section("serial"): + cfg.serial_device = cp.get("serial", "device", fallback="") + cfg.serial_baud = cp.getint("serial", "baud", fallback=0) + cfg.serial_line = cp.get("serial", "line", fallback="") + cfg.serial_dtr_rts = cp.get("serial", "dtr_rts", fallback="") + cfg.serial_kiss_entry = cp.get("serial", "kiss_entry", fallback="") + + if cp.has_section("features"): + cfg.feature_baycom = _truthy(cp.get("features", "baycom", fallback="yes")) + cfg.feature_pccom = _truthy(cp.get("features", "pccom", fallback="yes")) + # Product key max25_bcpr; legacy bcpr= accepted + cfg.feature_max25_bcpr = _truthy( + cp.get("features", "max25_bcpr", fallback=cp.get("features", "bcpr", fallback="yes")) + ) + cfg.feature_tmodem = _truthy(cp.get("features", "tmodem", fallback="no")) + if cp.has_section("hybbx"): + cfg.hybbx_release_attach = _truthy( + cp.get("hybbx", "release_attach", fallback="no") + ) + if cp.has_option("hybbx", "attach"): + cfg.hybbx_release_attach = _truthy(cp.get("hybbx", "attach")) + cfg.report_error_transmissions = _truthy( + cp.get("reporting", "error_transmissions", fallback="yes") + ) + cfg.report_voice_transmissions = _truthy( + cp.get("reporting", "voice_transmissions", fallback="yes") + ) + cfg.report_data_passes = _ini_int( + cp, "reporting", "data_passes", 3, min_value=1, max_value=32 + ) + cfg.report_data_quality_min = _ini_int( + cp, "reporting", "data_quality_min", 50, min_value=1, max_value=100 + ) + cfg.report_data_pass_seconds = _ini_int( + cp, "reporting", "data_pass_seconds", 20, min_value=1, max_value=300 + ) + if cp.has_section("tot"): + cfg.tot.enabled = _truthy(cp.get("tot", "enabled", fallback="yes")) + cfg.tot.max_key_sec = _ini_int( + cp, "tot", "max_key_sec", cfg.tot.max_key_sec, min_value=1, max_value=600 + ) + if cp.has_option("tot", "max_key_ms"): + cfg.tot.max_key_sec = max( + 1, _ini_int(cp, "tot", "max_key_ms", cfg.tot.max_key_sec * 1000) // 1000 + ) + cfg.tot.min_gap_sec = _ini_float( + cp, "tot", "min_gap_sec", cfg.tot.min_gap_sec, min_value=0.0, max_value=60.0 + ) + if cp.has_option("tot", "min_gap_ms"): + cfg.tot.min_gap_sec = _ini_int(cp, "tot", "min_gap_ms", 1500) / 1000.0 + cfg.tot.max_consecutive = _ini_int( + cp, "tot", "max_consecutive", cfg.tot.max_consecutive, min_value=1, max_value=32 + ) + cfg.tot.max_bursts = _ini_int( + cp, "tot", "max_bursts", cfg.tot.max_bursts, min_value=1, max_value=64 + ) + cfg.tot.recover_sec = _ini_int( + cp, "tot", "recover_sec", cfg.tot.recover_sec, min_value=0, max_value=86400 + ) + + cfg.devices = parse_devices(cp, cfg) + cfg.devices = [d for d in cfg.devices if _device_allowed_by_features(d, cfg)] + allowed = supported_device_ids() + if allowed: + filtered: list[DeviceBackendConfig] = [] + for dev in cfg.devices: + if dev.device_id in allowed: + filtered.append(dev) + else: + LOGGER.warn( + f"device {dev.device_id} not supported on {platform_label()} — skipped", + area="config", + ) + cfg.devices = filtered + if not cfg.devices: + LOGGER.warn("no devices remain after config/platform filter", area="config") + cfg.modular_tcp = load_modular_tcp(cp) + if not cfg.default_device and cfg.devices: + cfg.default_device = cfg.devices[0].device_id + cfg.device = cfg.default_device + return cfg + + +def init_device_runtimes(state: DaemonState) -> None: + state.devices.clear() + for dev_cfg in state.cfg.devices: + if not dev_cfg.enabled: + continue + state.devices[dev_cfg.device_id] = DeviceRuntime( + cfg=dev_cfg, + quality=DataQualityTracker( + passes_required=state.cfg.report_data_passes, + min_good_percent=state.cfg.report_data_quality_min, + pass_window_sec=state.cfg.report_data_pass_seconds, + ), + ) + if not registry_tested(dev_cfg.device_id): + LOGGER.warn( + f"backend={dev_cfg.backend_type or 'auto'} — not hardware-validated in CI", + area="devices", + device=dev_cfg.device_id, + ) + if state.cfg.default_device in state.devices: + state.selected_device = state.cfg.default_device + elif state.devices: + state.selected_device = next(iter(state.devices)) + else: + state.selected_device = state.cfg.default_device + LOGGER.warn("no enabled device runtimes — SEND/CONNECT unavailable", area="devices") + + +def enabled_device_ids(state: DaemonState) -> list[str]: + return list(state.devices.keys()) + + +def device_hardware(state: DaemonState, dev_id: str) -> str: + rt = state.devices.get(dev_id) + if rt is None: + return state.cfg.hardware + return rt.cfg.hardware or state.cfg.hardware + + +def device_backend_kind(state: DaemonState, dev_id: str) -> str: + rt = state.devices.get(dev_id) + if rt is None: + return "kiss-serial" + return rt.cfg.backend_type or "kiss-serial" + + +def on_backend_rx(state: DaemonState, dev_id: str, line: str) -> None: + rt = state.devices.get(dev_id) + if rt is not None: + _sync_quality_tracker(state, rt) + rt.quality.record(classify_rx_line(line, state.cfg.callid)) + src = extract_ax25_source(line) + if src and state.bans.is_banned(src): + return + log(f"rx {dev_id}: {line}") + broadcast(state, f"RX device={dev_id} {line}") + + +def on_backend_invalid_frame(state: DaemonState, dev_id: str) -> None: + rt = state.devices.get(dev_id) + if rt is None: + return + _sync_quality_tracker(state, rt) + rt.quality.record_bad() + if state.cfg.report_error_transmissions: + broadcast(state, f"EVENT device={dev_id} error=invalid") + + +VOICE_BACKEND_KINDS = frozenset({"crdop-tcp", "audio-dummy"}) + + +def device_has_voice_path(rt: DeviceRuntime) -> bool: + kind = rt.cfg.backend_type or registry_backend(rt.cfg.device_id) + if kind in VOICE_BACKEND_KINDS: + return True + hw = (rt.cfg.hardware or "").lower() + return hw in ("acoustic-bench", "soft-modems") + + +def device_link_status(rt: DeviceRuntime) -> str: + if rt.backend is not None: + return backend_serial_label(rt.backend) + return rt.link_status + + +def link_status_is_healthy(status: str) -> bool: + if status in ("ready", "open", "n/a"): + return True + if status.startswith("error") or status in ("closed", "stopped"): + return False + return True + + +def _sync_quality_tracker(state: DaemonState, rt: DeviceRuntime) -> None: + required = state.cfg.report_data_passes + minimum = state.cfg.report_data_quality_min + window = state.cfg.report_data_pass_seconds + if ( + rt.quality.passes_required == required + and rt.quality.min_good_percent == minimum + and rt.quality.pass_window_sec == window + and rt.quality.passes.maxlen == required + ): + return + kept = list(rt.quality.passes)[-required:] + voice = rt.quality.voice_activity + current = rt.quality.current + rt.quality = DataQualityTracker( + passes_required=required, + min_good_percent=minimum, + pass_window_sec=window, + voice_activity=voice, + ) + for item in kept: + rt.quality.passes.append(item) + if current.started_at != 0.0: + rt.quality.current = current + + +def device_reporting_error(state: DaemonState, rt: DeviceRuntime) -> str: + if not state.cfg.report_error_transmissions: + return "invalid" + if not link_status_is_healthy(device_link_status(rt)): + return "invalid" + _sync_quality_tracker(state, rt) + if rt.quality.data_error_valid(reporting_enabled=True): + return "valid" + return "invalid" + + +def aggregate_reporting_error(state: DaemonState) -> str: + if not state.devices: + return "invalid" if not state.cfg.report_error_transmissions else "valid" + for rt in state.devices.values(): + if device_reporting_error(state, rt) == "invalid": + return "invalid" + return "valid" + + +def aggregate_reporting_voice(state: DaemonState) -> str: + if not state.cfg.report_voice_transmissions: + return "invalid" + voice_rts = [rt for rt in state.devices.values() if device_has_voice_path(rt)] + if not voice_rts: + return "valid" + for rt in voice_rts: + if not link_status_is_healthy(device_link_status(rt)): + return "invalid" + return "valid" + + +def device_reporting_voice(state: DaemonState, rt: DeviceRuntime) -> str: + if not device_has_voice_path(rt): + return "n/a" + _sync_quality_tracker(state, rt) + healthy = link_status_is_healthy(device_link_status(rt)) + if rt.quality.voice_signal_valid( + reporting_enabled=state.cfg.report_voice_transmissions, + link_healthy=healthy, + ): + return "valid" + return "invalid" + + +BACKEND_POLL_OPEN_STATUSES = frozenset( + { + "closed", + "error-open", + "error-connect", + "error-no-device", + "error-no-path", + } +) + +SERIAL_REPAIR_STATUSES = frozenset( + { + "error-host", + "error-kiss", + "error-tx", + "error-io", + "error-config", + } +) + +BACKEND_RETRY_STATUSES = BACKEND_POLL_OPEN_STATUSES | SERIAL_REPAIR_STATUSES + +HYBBX_ATTACH_MODES = frozenset({"hybbx-host", "hybbx-main", "hybbx-cohost"}) + + +def hybbx_release_attach(state: DaemonState) -> bool: + """HyBBX opens serial/KISS after max25d prep — max25d must not hold the fd.""" + mode = (state.cfg.mode or "").strip().lower() + if mode in HYBBX_ATTACH_MODES: + return True + return bool(getattr(state.cfg, "hybbx_release_attach", False)) + + +def uses_inline_tnc_prep(state: DaemonState, dev_id: str) -> bool: + """kiss-serial owned by max25d — no boot-wait subprocess (avoids port conflict).""" + if not state.cfg.stack_recover_only: + return False + return device_backend_kind(state, dev_id) == "kiss-serial" + + +def hybbx_host_hybbx_owns_serial(state: DaemonState, dev_id: str) -> bool: + """In HyBBX attach mode HyBBX opens serial/KISS after max25d prep.""" + if not hybbx_release_attach(state): + return False + kind = device_backend_kind(state, dev_id) + # kiss-raw + max25-bcpr: HyBBX owns the KISS attach; max25d must not hold the PTY. + if kind in ("kiss-raw-serial", "max25-bcpr-kiss", "bcpr-kiss"): + return True + if kind != "kiss-serial": + return False + rt = state.devices.get(dev_id) + return rt is not None and rt.prep_done and rt.backend is None + + +def prep_inline_serial_device(state: DaemonState, dev_id: str) -> None: + """Open serial and run initial recovery while holding DTR (no subprocess).""" + rt = state.devices[dev_id] + rt.stack_status = "ready" + LOGGER.info("inline prep — max25d owns serial recovery", area="stack", device=dev_id) + if not backend_enabled(state, dev_id): + return + if not open_backend(state, dev_id): + LOGGER.error(f"serial prep open failed status={rt.link_status}", area="serial", device=dev_id) + return + backend = rt.backend + stabilize = getattr(backend, "stabilize_session", None) + if stabilize is None: + return + ok = stabilize(state.cfg.callerid, force=False) + rt.prep_done = True + rt.link_status = backend.status + if ok: + LOGGER.ok("serial prep complete — terminal + KISS ready", area="serial", device=dev_id) + if hybbx_release_attach(state): + close_backend(state, dev_id) + rt.link_status = "ready" + LOGGER.info( + "HyBBX attach: serial released for HyBBX KISS attach", + area="serial", + device=dev_id, + ) + elif ( + backend.status == "error-host" + and state.cfg.serial_bootwait_escalate + and rt.stack_proc is None + ): + LOGGER.warn( + "inline ladder exhausted (error-host) — escalating to boot-wait + power-cycle hint", + area="serial", + device=dev_id, + ) + rt.last_bootwait_escalate = time.time() + escalate_to_bootwait_stack(state, dev_id) + else: + LOGGER.warn( + f"prep deferred status={backend.status} — serial watch will retry", + area="serial", + device=dev_id, + ) + + +def escalate_to_bootwait_stack(state: DaemonState, dev_id: str) -> None: + """Release inline serial and run boot-wait subprocess (DTR + power-cycle rescue).""" + rt = state.devices[dev_id] + close_backend(state, dev_id) + rt.prep_done = False + ctl = ctl_path(ROOT, PREFIX, _EXE) + if not ctl.is_file(): + rt.stack_status = "error-no-ctl" + log(f"serial watch: boot-wait escalate failed — no ctl ({dev_id})") + return + hw = device_hardware(state, dev_id) + args = [ + str(ctl), + "start", + "--mode", + state.cfg.mode, + "--hardware", + hw, + "--device", + dev_id, + ] + env = os.environ.copy() + env["MAX25_MODE"] = state.cfg.mode + env.pop("MAX25_TNC_PREP", None) + workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT)) + try: + proc = subprocess.Popen( + args, + cwd=workdir, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as exc: + log(f"serial watch: boot-wait escalate failed ({dev_id}): {exc}") + rt.stack_status = "error" + return + rt.stack_proc = proc + rt.stack_status = "running" + log( + f"serial watch: escalating to boot-wait ({dev_id}) pid={proc.pid} " + "— power OFF TNC 10s then ON while script runs (DTR held high)" + ) + broadcast(state, f"EVENT device={dev_id} serial=boot-wait-escalate") + + +def backend_needs_open(backend: Optional[DeviceBackend]) -> bool: + if backend is None: + return True + return backend.status in BACKEND_POLL_OPEN_STATUSES + + +def open_backend(state: DaemonState, dev_id: str) -> bool: + rt = state.devices.get(dev_id) + if rt is None or not backend_enabled(state, dev_id): + if rt is not None: + rt.link_status = "n/a" + return False + if hybbx_host_hybbx_owns_serial(state, dev_id): + rt.link_status = "ready" + rt.prep_done = True + rt.stack_status = "ready" + return True + if rt.backend is not None and rt.backend.status not in BACKEND_RETRY_STATUSES: + return rt.backend.status in ("open", "ready") + if ( + rt.backend is not None + and rt.backend.status in SERIAL_REPAIR_STATUSES + and rt.prep_done + ): + return rt.backend.status in ("open", "ready", "error-host", "error-kiss") + if rt.backend is not None and rt.backend.status != "closed": + rt.backend.close() + rt.backend = None + backend = create_backend( + rt.cfg, + str(ROOT), + lambda line, d=dev_id: on_backend_rx(state, d, line), + log, + prefix=str(PREFIX) if PREFIX else None, + on_invalid=lambda d=dev_id: on_backend_invalid_frame(state, d), + ) + if not backend.open(): + rt.backend = backend + rt.link_status = backend.status + return False + rt.backend = backend + rt.link_status = backend.status + return True + + +def close_backend(state: DaemonState, dev_id: str) -> None: + rt = state.devices.get(dev_id) + if rt is None or rt.backend is None: + return + rt.backend.close() + rt.link_status = rt.backend.status + rt.backend = None + + +def attach_backend_session(state: DaemonState, dev_id: str) -> bool: + if not backend_enabled(state, dev_id): + return True + if hybbx_host_hybbx_owns_serial(state, dev_id): + return True + rt = state.devices[dev_id] + if backend_needs_open(rt.backend): + if not open_backend(state, dev_id): + return False + assert rt.backend is not None + ok = rt.backend.attach_session(state.cfg.callerid) + rt.link_status = rt.backend.status + return ok + + +def detach_backend_session(state: DaemonState, dev_id: str) -> None: + rt = state.devices.get(dev_id) + if rt is None or rt.backend is None: + return + rt.backend.detach_session() + rt.link_status = rt.backend.status + + +def attach_all_sessions(state: DaemonState) -> bool: + ok = True + for dev_id in enabled_device_ids(state): + if backend_enabled(state, dev_id): + if not attach_backend_session(state, dev_id): + ok = False + return ok + + +def detach_all_sessions(state: DaemonState) -> None: + for dev_id in enabled_device_ids(state): + detach_backend_session(state, dev_id) + + +def backend_enabled(state: DaemonState, dev_id: str) -> bool: + if not state.cfg.serial_enabled: + return False + rt = state.devices.get(dev_id) + if rt is None: + return False + if rt.tot_paused: + return False + kind = rt.cfg.backend_type + return kind in ("kiss-serial", "baycom-kiss", "max25-bcpr-kiss", "bcpr-kiss", "kiss-raw-serial", "crdop-tcp") + + +def aggregate_stack_status(state: DaemonState) -> str: + if not state.devices: + return "stopped" + statuses = {rt.stack_status for rt in state.devices.values()} + if "running" in statuses: + return "running" + if any(s.startswith("error") for s in statuses): + return "error" + if statuses == {"ready"} or statuses == {"stopped"}: + return next(iter(statuses)) + if "ready" in statuses: + return "ready" + return "running" if "running" in statuses else "stopped" + + +def aggregate_link_status(state: DaemonState) -> str: + if not state.devices: + return "n/a" + if len(state.devices) == 1: + rt = next(iter(state.devices.values())) + return backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status + parts: list[str] = [] + for dev_id in sorted(state.devices): + rt = state.devices[dev_id] + st = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status + parts.append(f"{dev_id}={st}") + return ",".join(parts) + + +def status_line(state: DaemonState) -> str: + c = state.cfg + dev_list = ",".join(enabled_device_ids(state)) + selected = state.selected_device or c.default_device or c.device + return ( + f"STATUS hardware={c.hardware} device={selected} devices={dev_list} " + f"mode={c.mode} callerid={c.callerid} callid={c.callid} " + f"ax25_ui={'on' if c.ax25_ui else 'off'} " + f"connected={'yes' if state.connected else 'no'} " + f"stack={aggregate_stack_status(state)} serial={aggregate_link_status(state)} " + f"error={aggregate_reporting_error(state)} " + f"voice={aggregate_reporting_voice(state)}" + ) + + +def broadcast(state: DaemonState, line: str, skip: Optional[socket.socket] = None) -> None: + payload = (line + "\n").encode("utf-8") + dead: list[socket.socket] = [] + with state.lock: + for sock in state.clients: + if sock is skip: + continue + try: + sock.sendall(payload) + except OSError: + dead.append(sock) + for sock in dead: + state.clients.discard(sock) + + +def send_line(sock: socket.socket, line: str) -> None: + sock.sendall((line + "\n").encode("utf-8")) + + +def unix_path_is_live(path: str, *, timeout: float = 0.3) -> bool: + """True if path exists and accepts a Unix connect (live max25d listener).""" + if not path or not os.path.exists(path): + return False + probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + probe.settimeout(timeout) + probe.connect(path) + return True + except OSError: + return False + finally: + try: + probe.close() + except OSError: + pass + + +def unix_path_id(path: str) -> tuple[int, int] | None: + """Filesystem identity of a Unix socket path (dev, ino), or None.""" + try: + st = os.stat(path) + except OSError: + return None + return (st.st_dev, st.st_ino) + + +def unlink_unix_if_ours(path: str, bind_id: tuple[int, int] | None) -> None: + """Unlink path only when it still names the inode we bound. + + AF_UNIX: fstat(listen_fd) is sockfs — must not compare to path st_ino. + A second max25d that unlinks+rebinds must not lose its path when the + first instance exits (orphaned listen FD + ENOENT for clients). + """ + if not path or bind_id is None: + return + cur = unix_path_id(path) + if cur is None or cur != bind_id: + return + try: + os.unlink(path) + except FileNotFoundError: + pass + except OSError: + pass + + +def resolve_max25_bcpr_ini(explicit: str = "") -> Path | None: + """Resolve max25-bcpr.ini for userspace SER12 (max25e0).""" + from pathlib import Path as _P + if explicit: + p = _P(explicit) + return p if p.is_file() else None + for cand in ( + ROOT / "local" / "max25-bcpr.ini", + _P("/etc/max25/max25-bcpr.ini"), + _P("/etc/max25/bcpr.ini"), + ROOT / "local" / "bcpr.ini", + ROOT / "stacks" / "max25-bcpr" / "share" / "max25-bcpr.ini.example", + ROOT / "share" / "max25-bcpr" / "max25-bcpr.ini.example", + ): + if cand.is_file(): + return cand + return None + + +def resolve_max25_bcpr_ctl() -> Path | None: + from pathlib import Path as _P + for cand in ( + ROOT / "stacks" / "max25-bcpr" / "tools" / "max25-bcpr-ctl", + _P("/usr/local/sbin/max25-bcpr-ctl"), + _P("/usr/sbin/max25-bcpr-ctl"), + ): + if cand.is_file(): + return cand + return None + + +def bcpr_bc_index(cfg: DeviceBackendConfig) -> int: + tag = (cfg.max25_bcpr_device or cfg.bcpr_device or "").strip() + if not tag and cfg.device_id.startswith("max25e0:"): + tag = cfg.device_id.split(":", 1)[1] + if not tag: + tag = "bc0" + if tag.startswith("bc") and tag[2:].isdigit(): + return int(tag[2:]) + return 0 + + +def read_bcpr_state_dir(ini_path: Path) -> str: + cp = configparser.ConfigParser() + cp.read(ini_path) + for sect in ("max25-bcpr", "bcpr"): + if cp.has_section(sect) and cp.has_option(sect, "state_dir"): + return cp.get(sect, "state_dir").strip() or "/tmp/max25-bcpr" + return "/tmp/max25-bcpr" + + +def sync_tot_to_bcpr_ini(ini_path: Path, tot: TotConfig) -> None: + """Push max25d [tot] policy into max25-bcpr.ini before bcprd start.""" + if not ini_path.is_file(): + return + cp = configparser.ConfigParser() + cp.read(ini_path) + sect = "max25-bcpr" + if not cp.has_section(sect): + if cp.has_section("bcpr"): + sect = "bcpr" + else: + cp.add_section(sect) + cp.set(sect, "tot", "yes" if tot.enabled else "no") + cp.set(sect, "tot_max_key_sec", str(tot.max_key_sec)) + cp.set(sect, "tot_min_gap_ms", str(int(tot.min_gap_sec * 1000.0))) + cp.set(sect, "tot_max_consecutive", str(tot.max_consecutive)) + cp.set(sect, "tot_max_bursts", str(tot.max_bursts)) + with ini_path.open("w", encoding="utf-8") as fh: + cp.write(fh) + + +def tot_trip_path(state_dir: str, bc_index: int) -> Path: + return Path(state_dir) / f"tot-trip-bc{bc_index}" + + +def parse_tot_trip_file(path: Path) -> dict[str, str]: + out: dict[str, str] = {} + try: + for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): + if "=" not in line: + continue + key, val = line.split("=", 1) + out[key.strip()] = val.strip() + except OSError: + pass + return out + + +def device_tot_operational(state: DaemonState, dev_id: str) -> bool: + rt = state.devices.get(dev_id) + if rt is None: + return False + return not rt.tot_paused + + +def handle_tot_trip(state: DaemonState, dev_id: str, reason: str) -> None: + rt = state.devices.get(dev_id) + if rt is None or rt.tot_paused: + return + log(f"TOT trip ({dev_id}) reason={reason} — pause, stop stack, reset bcpr") + rt.tot_paused = True + rt.tot_trip_reason = reason or "unknown" + recover = state.cfg.tot.recover_sec + rt.tot_paused_until = time.time() + recover if recover > 0 else 0.0 + stop_device_stack(state, dev_id) + rt.stack_status = "tot-paused" + rt.link_status = "tot-paused" + rt.prep_done = False + broadcast( + state, + f"EVENT device={dev_id} tot=trip reason={rt.tot_trip_reason} stack=tot-paused", + ) + + +def poll_tot_trips(state: DaemonState) -> None: + if not state.cfg.tot.enabled: + return + for dev_id, rt in state.devices.items(): + kind = rt.cfg.backend_type or "" + if kind not in ("max25-bcpr-kiss", "bcpr-kiss"): + continue + if rt.tot_paused: + continue + explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or "" + resolved = resolve_max25_bcpr_ini(explicit) + if resolved is None: + continue + state_dir = read_bcpr_state_dir(resolved) + trip = tot_trip_path(state_dir, bcpr_bc_index(rt.cfg)) + if not trip.is_file(): + continue + meta = parse_tot_trip_file(trip) + if meta.get("tripped") != "1": + continue + handle_tot_trip(state, dev_id, meta.get("reason", "unknown")) + + +def poll_tot_recovery(state: DaemonState) -> None: + if not state.cfg.tot.enabled: + return + now = time.time() + for dev_id, rt in state.devices.items(): + if not rt.tot_paused: + continue + kind = rt.cfg.backend_type or "" + if kind not in ("max25-bcpr-kiss", "bcpr-kiss"): + continue + recover = state.cfg.tot.recover_sec + if recover <= 0: + continue + if rt.tot_paused_until > 0 and now < rt.tot_paused_until: + continue + explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or "" + resolved = resolve_max25_bcpr_ini(explicit) + if resolved is not None: + trip = tot_trip_path(read_bcpr_state_dir(resolved), bcpr_bc_index(rt.cfg)) + try: + trip.unlink(missing_ok=True) + except OSError: + pass + rt.tot_paused = False + rt.tot_trip_reason = "" + rt.tot_paused_until = 0.0 + log(f"TOT recover ({dev_id}) — restarting max25-bcpr stack") + if state.cfg.auto_start and backend_enabled(state, dev_id): + start_device_stack_for_tot(state, dev_id) + + +def start_device_stack_for_tot(state: DaemonState, dev_id: str) -> None: + """Restart bcpr stack after TOT cooldown (single device, no full start_stacks).""" + rt = state.devices[dev_id] + kind = rt.cfg.backend_type or "" + if kind in ("max25-bcpr-kiss", "bcpr-kiss"): + explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or "" + resolved = resolve_max25_bcpr_ini(explicit) + if resolved is None: + rt.stack_status = "error-no-ini" + return + sync_tot_to_bcpr_ini(resolved, state.cfg.tot) + start_max25_bcpr_stack(state, dev_id, resolved) + + +def _read_subprocess_capture(fh, limit: int = 4000) -> str: + """Read captured ctl stdout/stderr (UTF-8, truncated tail).""" + try: + fh.seek(0) + raw = fh.read() + except OSError: + return "" + text = raw.decode("utf-8", errors="replace").strip() + if len(text) > limit: + return f"...\n{text[-limit:]}" + return text + + +def start_max25_bcpr_stack(state: DaemonState, dev_id: str, ini: Path) -> None: + """Start max25-bcprd once via max25-bcpr-ctl for shared ini.""" + rt = state.devices[dev_id] + ctl = resolve_max25_bcpr_ctl() + if ctl is None: + rt.stack_status = "error-no-ctl" + log(f"max25-bcpr-ctl not found ({dev_id})") + return + args = [str(ctl), "-c", str(ini), "start"] + env = os.environ.copy() + # Temp file (not PIPE): bcprd backgrounded by ctl must not block communicate(). + # Capture ctl output; log tail only when rc!=0 or after ctl timeout. + proc: subprocess.Popen[bytes] | None = None + ctl_output = "" + try: + with tempfile.TemporaryFile(mode="w+b") as capfh: + try: + proc = subprocess.Popen( + args, + cwd=str(ROOT), + env=env, + stdin=subprocess.DEVNULL, + stdout=capfh, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + except OSError as exc: + log(f"max25-bcpr start failed ({dev_id}): {exc}") + rt.stack_status = "error" + return + try: + proc.communicate(timeout=30) + except subprocess.TimeoutExpired: + proc.kill() + try: + proc.communicate(timeout=2) + except (subprocess.TimeoutExpired, OSError): + pass + ctl_output = _read_subprocess_capture(capfh) + # May still have spawned max25-bcprd — fall through to kiss probe. + log(f"max25-bcpr-ctl start timed out ({dev_id}) — probing live kiss/pid") + if ctl_output: + log(f"max25-bcpr-ctl output ({dev_id}): {ctl_output}") + else: + ctl_output = _read_subprocess_capture(capfh) + except OSError as exc: + log(f"max25-bcpr start failed ({dev_id}): {exc}") + rt.stack_status = "error" + return + if proc is None: + rt.stack_status = "error" + return + if proc.returncode not in (0, None, -9, -15): + # -9/-15: we killed a hung ctl; still probe kiss below. + if proc.returncode > 0: + rt.stack_status = "error" + if ctl_output: + log(f"max25-bcpr-ctl start rc={proc.returncode} ({dev_id}): {ctl_output}") + else: + log(f"max25-bcpr-ctl start rc={proc.returncode} ({dev_id})") + return + # max25-bcpr-ctl itself exits; live daemon is max25-bcprd (pidfile under state_dir). + rt.stack_proc = None + rt.stack_status = "running" + kiss = normalize_max25_bcpr_path( + (rt.cfg.kiss_link or "").strip() or MAX25_BCPR_KISS_DEFAULT + ) + deadline = time.time() + 5.0 + while time.time() < deadline: + if os.path.exists(kiss): + break + time.sleep(0.1) + if not os.path.exists(kiss): + rt.stack_status = "error-no-kiss" + log(f"max25-bcpr kiss_link missing after start ({dev_id}: {kiss})") + return + addrs = f"ipv4={rt.cfg.ipv4 or '-'} ipv6={rt.cfg.ipv6 or '-'}" + log(f"max25-bcpr started ({dev_id}, ini={ini}, kiss={kiss}, {addrs})") + rt.stack_status = "ready" + # HyBBX attach: HyBBX opens kiss_link — do not hold the PTY here. + if hybbx_release_attach(state): + rt.link_status = "ready" + rt.prep_done = True + log(f"max25-bcpr ready — HyBBX owns KISS attach ({dev_id}: {kiss})") + return + # Standalone: open+hold KISS so max25-terminal TX works (stack_proc=None). + if backend_enabled(state, dev_id): + if open_backend(state, dev_id): + if attach_backend_session(state, dev_id): + # UI/datagram TX uses CONNECT as session arm — arm at start so + # max25-terminal SEND keys MCR without a separate CONNECT race. + state.connected = True + log(f"max25-bcpr KISS open ({dev_id}: {kiss})") + else: + log(f"max25-bcpr KISS open but attach failed ({dev_id})") + else: + log(f"max25-bcpr KISS open failed ({dev_id}) status={rt.link_status}") + + +def start_device_stack(state: DaemonState, dev_id: str) -> None: + if uses_inline_tnc_prep(state, dev_id): + prep_inline_serial_device(state, dev_id) + return + rt = state.devices[dev_id] + ctl = ctl_path(ROOT, PREFIX, _EXE) + if not ctl.is_file(): + rt.stack_status = "error-no-ctl" + return + hw = device_hardware(state, dev_id) + dev_cfg = rt.cfg + ctl_device = dev_id + if (dev_cfg.backend_type or "") == "baycom-kiss": + ctl_device = baycom_ctl_device_id(dev_cfg) + args = [ + str(ctl), + "start", + "--mode", + state.cfg.mode, + "--hardware", + hw, + "--device", + ctl_device, + ] + kind = dev_cfg.backend_type or "" + if kind == "baycom-kiss": + explicit = dev_cfg.baycom_ini or "" + resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit) + if resolved: + args.extend(["--baycom-ini", str(resolved)]) + env = os.environ.copy() + env["MAX25_MODE"] = state.cfg.mode + if hw == "tncs" and state.cfg.stack_recover_only: + env["MAX25_TNC_PREP"] = "recover" + workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT)) + try: + proc = subprocess.Popen( + args, + cwd=workdir, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as exc: + log(f"stack start failed ({dev_id}): {exc}") + rt.stack_status = "error" + return + rt.stack_proc = proc + rt.stack_status = "running" + log(f"stack started pid={proc.pid} ({hw}/{dev_id})") + + +def start_stacks(state: DaemonState) -> None: + """Start per-device stacks; one max25-bcpr-ctl per shared ini.""" + started_baycom_ini: dict[str, str] = {} + started_max25_bcpr_ini: dict[str, str] = {} + for dev_id in enabled_device_ids(state): + rt = state.devices[dev_id] + kind = rt.cfg.backend_type or "" + if uses_inline_tnc_prep(state, dev_id): + prep_inline_serial_device(state, dev_id) + continue + if kind in ("max25-bcpr-kiss", "bcpr-kiss"): + explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or "" + resolved = resolve_max25_bcpr_ini(explicit) + ini_key = str(resolved) if resolved else "" + if ini_key and ini_key in started_max25_bcpr_ini: + primary = started_max25_bcpr_ini[ini_key] + primary_rt = state.devices[primary] + rt.stack_proc = primary_rt.stack_proc + rt.stack_status = primary_rt.stack_status + log(f"max25-bcpr stack shared with {primary} ({dev_id}, ini={ini_key})") + # Shared max25-bcprd — open this device's kiss only when max25d owns it. + if ( + not hybbx_release_attach(state) + and rt.stack_status in ("ready", "running") + and backend_enabled(state, dev_id) + ): + if open_backend(state, dev_id): + attach_backend_session(state, dev_id) + elif hybbx_release_attach(state): + rt.link_status = "ready" + rt.prep_done = True + continue + if resolved: + sync_tot_to_bcpr_ini(resolved, state.cfg.tot) + start_max25_bcpr_stack(state, dev_id, resolved) + started_max25_bcpr_ini[str(resolved)] = dev_id + else: + rt.stack_status = "error-no-ini" + log(f"max25-bcpr.ini not found ({dev_id})") + continue + if kind == "baycom-kiss": + explicit = rt.cfg.baycom_ini or "" + resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit) + ini_key = str(resolved) if resolved else "" + if ini_key and ini_key in started_baycom_ini: + primary = started_baycom_ini[ini_key] + primary_rt = state.devices[primary] + rt.stack_proc = primary_rt.stack_proc + rt.stack_status = primary_rt.stack_status + log(f"stack shared with {primary} ({dev_id}, ini={ini_key})") + continue + start_device_stack(state, dev_id) + if kind == "baycom-kiss": + explicit = rt.cfg.baycom_ini or "" + resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit) + if resolved: + started_baycom_ini[str(resolved)] = dev_id + + +def stop_device_stack(state: DaemonState, dev_id: str) -> None: + close_backend(state, dev_id) + rt = state.devices[dev_id] + kind = rt.cfg.backend_type or "" + if kind in ("max25-bcpr-kiss", "bcpr-kiss"): + ctl = resolve_max25_bcpr_ctl() + explicit = rt.cfg.max25_bcpr_ini or rt.cfg.bcpr_ini or "" + resolved = resolve_max25_bcpr_ini(explicit) + if ctl is not None and resolved is not None: + subprocess.run( + [str(ctl), "-c", str(resolved), "stop"], + cwd=str(ROOT), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + rt.stack_proc = None + rt.stack_status = "stopped" + return + proc = rt.stack_proc + if proc is not None and proc.poll() is None: + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass + except OSError: + proc.terminate() + rt.stack_proc = None + rt.stack_status = "stopped" + hw = device_hardware(state, dev_id) + ctl = ctl_path(ROOT, PREFIX, _EXE) + if ctl.is_file(): + workdir = str(ROOT if (ROOT / "plugins").is_dir() else (PREFIX or ROOT)) + stop_args = [str(ctl), "stop", "--hardware", hw, "--device", dev_id] + if kind == "baycom-kiss": + explicit = rt.cfg.baycom_ini or "" + resolved = resolve_baycom_ini(dev_id, ROOT, PREFIX, explicit) + if resolved: + stop_args.extend(["--baycom-ini", str(resolved)]) + subprocess.run( + stop_args, + cwd=workdir, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def stop_stacks(state: DaemonState) -> None: + for dev_id in list(state.devices): + stop_device_stack(state, dev_id) + log("all stacks stopped") + + +def poll_device_stack(state: DaemonState, dev_id: str) -> None: + rt = state.devices[dev_id] + proc = rt.stack_proc + if proc is None: + return + rc = proc.poll() + if rc is None: + return + rt.stack_proc = None + if rc == 0: + rt.stack_status = "ready" + log(f"stack boot-wait finished ({dev_id}) rc={rc}") + if backend_enabled(state, dev_id): + open_backend(state, dev_id) + backend = rt.backend + stabilize = getattr(backend, "stabilize_session", None) if backend else None + if stabilize is not None: + ok = stabilize(state.cfg.callerid, force=False) + rt.prep_done = True + rt.link_status = backend.status + rt.inline_repair_failures = 0 + if ok: + log(f"serial post boot-wait OK ({dev_id})") + if state.connected: + attach_backend_session(state, dev_id) + else: + log( + f"serial post boot-wait deferred ({dev_id}) " + f"status={backend.status}" + ) + else: + rt.stack_status = f"error-rc{rc}" + log(f"stack boot-wait failed ({dev_id}) rc={rc}") + + +def poll_stacks(state: DaemonState) -> None: + for dev_id in enabled_device_ids(state): + poll_device_stack(state, dev_id) + retry_pending_backends(state) + + +def retry_pending_backends(state: DaemonState) -> None: + """Re-attach KISS PTY/serial when stack is up but the link was not ready yet.""" + for dev_id in enabled_device_ids(state): + if not backend_enabled(state, dev_id): + continue + rt = state.devices[dev_id] + if hybbx_host_hybbx_owns_serial(state, dev_id): + continue + # bcpr uses "running" until kiss open flips to "ready"; accept both. + if rt.stack_status not in ("ready", "stopped", "running"): + continue + if not backend_needs_open(rt.backend): + continue + if open_backend(state, dev_id): + attach_backend_session(state, dev_id) + if rt.stack_status == "running": + rt.stack_status = "ready" + + +def poll_reporting_passes(state: DaemonState) -> None: + """Advance timed data-quality pass windows (default 20s each, 3 passes).""" + now = time.time() + for rt in state.devices.values(): + _sync_quality_tracker(state, rt) + rt.quality.tick(now) + + +def poll_serial_stability(state: DaemonState) -> None: + """Periodic TNC health probe + software recovery (no power cycle).""" + cfg = state.cfg + if not cfg.serial_watch: + return + now = time.time() + if state.started_at and now - state.started_at < cfg.serial_watch_startup_grace: + return + for dev_id in enabled_device_ids(state): + if device_backend_kind(state, dev_id) != "kiss-serial": + continue + if hybbx_host_hybbx_owns_serial(state, dev_id): + continue + rt = state.devices[dev_id] + if rt.stack_proc is not None and rt.stack_proc.poll() is None: + continue + if ( + not uses_inline_tnc_prep(state, dev_id) + and cfg.stack_recover_only + and rt.stack_status.startswith("error") + and rt.stack_proc is None + and now - rt.last_stack_retry >= cfg.stack_retry_interval + ): + rt.last_stack_retry = now + log(f"serial watch: stack retry recover-only ({dev_id})") + start_device_stack(state, dev_id) + continue + if not backend_enabled(state, dev_id): + continue + backend = rt.backend + force = backend is not None and backend.status in SERIAL_REPAIR_STATUSES + due = now - rt.last_watch >= cfg.serial_watch_interval + if not force and not due: + continue + if backend is None: + if backend_needs_open(None): + open_backend(state, dev_id) + backend = rt.backend + if backend is None: + continue + if now - rt.last_repair < cfg.serial_repair_cooldown and not force: + continue + if backend.status == "ready" and not force: + if due: + rt.last_watch = now + continue + stabilize = getattr(backend, "stabilize_session", None) + if stabilize is None: + continue + rt.last_watch = now + rt.last_repair = now + ok = stabilize(state.cfg.callerid, force=force) + rt.link_status = backend.status + if ok: + rt.inline_repair_failures = 0 + if force: + log(f"serial watch: repaired ({dev_id})") + broadcast(state, f"EVENT device={dev_id} serial=ready") + else: + log(f"serial watch: repair failed ({dev_id}) status={backend.status}") + if ( + uses_inline_tnc_prep(state, dev_id) + and backend.status == "error-host" + and cfg.serial_bootwait_escalate + ): + rt.inline_repair_failures += 1 + if ( + rt.inline_repair_failures >= cfg.serial_bootwait_escalate_after + and now - rt.last_bootwait_escalate >= cfg.serial_bootwait_escalate_cooldown + ): + rt.last_bootwait_escalate = now + rt.inline_repair_failures = 0 + escalate_to_bootwait_stack(state, dev_id) + elif rt.inline_repair_failures >= cfg.serial_bootwait_escalate_after: + log( + f"serial watch: boot-wait escalate cooldown ({dev_id}) " + f"— manual: stacks/tncs/{dev_id}-boot-wait.sh" + ) + if backend.status in ("error-io", "error-open", "error-no-device"): + close_backend(state, dev_id) + if open_backend(state, dev_id) and state.connected: + attach_backend_session(state, dev_id) + + +def format_tx(state: DaemonState, text: str) -> str: + if state.cfg.ax25_ui: + return f"[AX25 UI {state.cfg.callerid}>{state.cfg.callid}] {text}" + return text + + +def resolve_selected_device(state: DaemonState) -> Optional[str]: + dev_id = state.selected_device + if dev_id in state.devices: + return dev_id + ids = enabled_device_ids(state) + return ids[0] if ids else None + + +def device_line(state: DaemonState, dev_id: str) -> str: + rt = state.devices[dev_id] + link = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status + hw = device_hardware(state, dev_id) + backend = rt.cfg.backend_type or "auto" + enabled = "yes" if rt.cfg.enabled else "no" + return ( + f"DEVICE id={dev_id} hardware={hw} backend={backend} serial={link} " + f"stack={rt.stack_status} enabled={enabled} " + f"error={device_reporting_error(state, rt)} " + f"voice={device_reporting_voice(state, rt)}" + ) + + +def handle_command(state: DaemonState, sock: socket.socket, line: str) -> None: + line = line.strip("\r\n") + if not line: + return + upper = line.upper() + + if upper == "PING": + send_line(sock, "OK") + return + + if upper == "GET STATUS": + send_line(sock, status_line(state)) + send_line(sock, "OK") + return + + if upper == "GET DEVICES": + for dev_id in sorted(state.devices): + send_line(sock, device_line(state, dev_id)) + send_line(sock, "OK") + return + + if upper.startswith("SET DEVICE ") or upper.startswith("SELECT DEVICE "): + prefix = "SET DEVICE " if upper.startswith("SET DEVICE ") else "SELECT DEVICE " + dev_id = line[len(prefix) :].strip() + if dev_id not in state.devices: + send_line(sock, f"ERR unknown device: {dev_id}") + return + state.selected_device = dev_id + state.cfg.device = dev_id + send_line(sock, "OK") + return + + if upper.startswith("SET CALLERID "): + value = line[13:].strip().upper() + if not valid_callsign(value): + send_line(sock, "ERR invalid CALLERID") + return + state.cfg.callerid = value + send_line(sock, "OK") + return + + if upper.startswith("SET CALLID "): + value = line[11:].strip().upper() + if not valid_callsign(value): + send_line(sock, "ERR invalid CALLID") + return + state.cfg.callid = value + send_line(sock, "OK") + return + + if upper.startswith("SET AX25_UI "): + flag = line[12:].strip().lower() + if flag in ("on", "yes", "1", "true"): + state.cfg.ax25_ui = True + elif flag in ("off", "no", "0", "false"): + state.cfg.ax25_ui = False + else: + send_line(sock, "ERR ax25_ui on|off") + return + send_line(sock, "OK") + return + + if upper == "CONNECT": + if not attach_all_sessions(state): + send_line(sock, "ERR link not ready") + return + state.connected = True + send_line(sock, "EVENT connected") + send_line(sock, "OK") + return + + if upper == "DISCONNECT": + detach_all_sessions(state) + state.connected = False + send_line(sock, "EVENT disconnected") + send_line(sock, "OK") + return + + if upper.startswith("MONITOR "): + flag = line[8:].strip().lower() + state.monitor_only = flag in ("on", "yes", "1", "true") + send_line(sock, "OK") + return + + if upper.startswith("BAN "): + value = line[4:].strip().upper() + if not valid_callsign(value): + send_line(sock, "ERR invalid callsign") + return + try: + state.bans.add(value) + except OSError as exc: + send_line(sock, f"ERR ban save failed: {exc}") + return + send_line(sock, "OK") + return + + if upper.startswith("UNBAN "): + value = line[6:].strip().upper() + if not valid_callsign(value): + send_line(sock, "ERR invalid callsign") + return + try: + if not state.bans.remove(value): + send_line(sock, "ERR not banned") + return + except OSError as exc: + send_line(sock, f"ERR ban save failed: {exc}") + return + send_line(sock, "OK") + return + + if upper == "BANS": + for entry in state.bans.list(): + send_line(sock, f"BAN {entry}") + send_line(sock, "OK") + return + + if upper.startswith("SEND "): + if state.monitor_only: + send_line(sock, "ERR monitor-only") + return + # UI frames: auto-arm session if stack/KISS is up. Terminal Enter/F10→SEND + # must key PTT/MCR without requiring a prior CONNECT (L4 writes kiss + # directly and already keys; unix SEND must match). + if not state.connected: + if not attach_all_sessions(state): + send_line(sock, "ERR not connected") + return + state.connected = True + send_line(sock, "EVENT connected") + dev_id = resolve_selected_device(state) + if dev_id is None: + send_line(sock, "ERR no device configured") + return + payload = line[5:] + framed = format_tx(state, payload) + rt = state.devices[dev_id] + if backend_enabled(state, dev_id): + if rt.backend is None or rt.backend.status != "ready": + # Re-attach after DISCONNECT left kiss inactive but stack ready. + if not attach_backend_session(state, dev_id): + send_line(sock, "ERR link not ready") + return + if rt.backend is None or rt.backend.status != "ready": + send_line(sock, "ERR link not ready") + return + ok, display = rt.backend.transmit( + state.cfg.callerid, + state.cfg.callid, + payload, + state.cfg.ax25_ui, + ) + if not ok and hasattr(rt.backend, "stabilize_session"): + log(f"serial watch: tx retry after repair ({dev_id})") + if rt.backend.stabilize_session(state.cfg.callerid, force=True): + rt.link_status = rt.backend.status + ok, display = rt.backend.transmit( + state.cfg.callerid, + state.cfg.callid, + payload, + state.cfg.ax25_ui, + ) + if not ok: + send_line(sock, f"ERR {display}") + return + framed = f"device={dev_id} {display}" + log(f"tx {dev_id}: {framed}") + send_line(sock, f"RX {framed}") + broadcast(state, f"RX {framed}", skip=sock) + send_line(sock, "OK") + return + + send_line(sock, f"ERR unknown command: {line.split()[0]}") + + +def tcp_auth_ok(sock: socket.socket, expected: str, timeout: float = 30.0) -> bool: + if not expected: + return True + send_line(sock, "AUTH required") + sock.settimeout(timeout) + buf = b"" + try: + while True: + try: + chunk = sock.recv(4096) + except socket.timeout: + return False + if not chunk: + return False + buf += chunk + if len(buf) > M25_MAX_LINE_BUF: + return False + while b"\n" in buf: + raw, buf = buf.split(b"\n", 1) + try: + line = raw.decode("utf-8").strip("\r") + except UnicodeDecodeError: + return False + if not line: + continue + if line.upper().startswith("AUTH "): + supplied = line[5:] + return supplied == expected + return False + finally: + sock.settimeout(300.0) + + +def client_thread(state: DaemonState, sock: socket.socket, from_tcp: bool) -> None: + sock.settimeout(300.0) + buf = b"" + try: + if from_tcp and state.cfg.tcp_password: + if not tcp_auth_ok(sock, state.cfg.tcp_password): + send_line(sock, "ERR auth failed") + return + send_line(sock, "OK") + send_line(sock, status_line(state)) + while True: + try: + chunk = sock.recv(4096) + except socket.timeout: + continue + if not chunk: + break + buf += chunk + if len(buf) > M25_MAX_LINE_BUF: + send_line(sock, "ERR line too long") + break + while b"\n" in buf: + raw, buf = buf.split(b"\n", 1) + try: + line = raw.decode("utf-8") + except UnicodeDecodeError: + send_line(sock, "ERR invalid utf-8") + continue + handle_command(state, sock, line) + except OSError: + pass + finally: + with state.lock: + state.clients.discard(sock) + try: + sock.close() + except OSError: + pass + + +def serve( + state: DaemonState, + listeners: list[tuple[socket.socket, bool, tuple[int, int] | None]], +) -> None: + running = True + + def on_signal(_signum, _frame): + nonlocal running + running = False + + signal.signal(signal.SIGTERM, on_signal) + signal.signal(signal.SIGINT, on_signal) + + if state.cfg.auto_start: + start_stacks(state) + + state.started_at = time.time() + + device_lines: list[tuple[str, str, str]] = [] + for dev_id in enabled_device_ids(state): + rt = state.devices[dev_id] + link = backend_serial_label(rt.backend) if rt.backend is not None else rt.link_status + device_lines.append((dev_id, rt.stack_status, link)) + + emit_startup_complete( + device_lines=device_lines, + tcp_host=state.cfg.tcp_host, + tcp_port=state.cfg.tcp_port, + unix_socket=state.cfg.unix_socket, + ) + + drop_privileges_or_exit( + RunAsConfig( + user=state.cfg.run_user, + group=state.cfg.run_group, + uid=state.cfg.run_uid, + gid=state.cfg.run_gid, + ), + log=lambda msg, area="privilege": LOGGER.info(msg, area=area), + ) + + while running: + poll_stacks(state) + poll_tot_trips(state) + poll_tot_recovery(state) + poll_reporting_passes(state) + poll_serial_stability(state) + socks = [lsock for lsock, _tcp, _bid in listeners] + rlist, _, _ = select.select(socks, [], [], 1.0) + for lsock, from_tcp, _bid in listeners: + if lsock not in rlist: + continue + try: + client, _addr = lsock.accept() + except OSError: + continue + client.setblocking(True) + with state.lock: + state.clients.add(client) + threading.Thread( + target=client_thread, + args=(state, client, from_tcp), + daemon=True, + ).start() + + stop_stacks(state) + for lsock, from_tcp, bind_id in listeners: + if not from_tcp and state.cfg.unix_socket: + unlink_unix_if_ours(state.cfg.unix_socket, bind_id) + lsock.close() + + +def make_listeners( + cfg: DaemonConfig, +) -> list[tuple[socket.socket, bool, tuple[int, int] | None]]: + listeners: list[tuple[socket.socket, bool, tuple[int, int] | None]] = [] + + tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + tcp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + tcp.bind((cfg.tcp_host, cfg.tcp_port)) + except OSError as exc: + tcp.close() + log(f"TCP bind {cfg.tcp_host}:{cfg.tcp_port} failed: {exc}") + raise SystemExit(1) from exc + tcp.listen(32) + tcp.setblocking(False) + listeners.append((tcp, True, None)) + + if cfg.unix_socket: + sock_path = Path(cfg.unix_socket) + try: + sock_path.parent.mkdir(parents=True, exist_ok=True) + except OSError: + fallback = Path("/tmp/max25/modem.sock") + log(f"unix {cfg.unix_socket} unavailable, using {fallback}") + cfg.unix_socket = str(fallback) + sock_path = fallback + try: + sock_path.parent.mkdir(parents=True, exist_ok=True) + except OSError: + log("unix socket disabled (no writable path)") + cfg.unix_socket = "" + return listeners + # Never unlink a live peer path — that orphans the other max25d FD + # (ss still shows the name; clients get ENOENT). + if unix_path_is_live(cfg.unix_socket): + log( + f"unix socket {cfg.unix_socket} already live — not stealing " + "(refuse second max25d unix bind)" + ) + else: + try: + os.unlink(cfg.unix_socket) + except FileNotFoundError: + pass + except OSError: + pass + unix = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + unix.bind(cfg.unix_socket) + except OSError as exc: + log(f"unix socket {cfg.unix_socket} skipped ({exc})") + unix.close() + else: + try: + os.chmod(cfg.unix_socket, 0o660) + except OSError: + pass + unix.listen(32) + unix.setblocking(False) + listeners.append((unix, False, unix_path_id(cfg.unix_socket))) + + return listeners + + +def main(argv: Optional[list[str]] = None) -> int: + if not max25d_supported(): + log(f"max25d is not supported on {sys.platform}") + return 1 + + parser = argparse.ArgumentParser(description=f"MAX25 daemon ({platform_label()})") + parser.add_argument( + "-c", + "--config", + type=Path, + default=None, + help="Path to max25d.ini", + ) + parser.add_argument( + "--no-stack", + action="store_true", + help="Do not auto-start hardware stack", + ) + parser.add_argument( + "--tcp-port", + type=int, + default=None, + help="Override TCP listen port", + ) + parser.add_argument( + "--no-serial", + action="store_true", + help="Disable KISS serial bridge (loopback SEND only)", + ) + parser.add_argument( + "--session", + choices=("tmux", "screen"), + metavar="BACKEND", + help="Re-exec via max25d-session (detach in tmux/screen); use max25d-session attach", + ) + args = parser.parse_args(argv) + + if args.session: + session_sh = ROOT / "scripts" / "max25d-session.sh" + if not session_sh.is_file(): + session_sh = Path(PREFIX) / "bin" / "max25d-session" if PREFIX else session_sh + if not session_sh.is_file(): + LOGGER.error( + "max25d-session not found — install scripts/max25d-session.sh or use tmux/screen manually", + area="session", + ) + return 1 + cmd = [str(session_sh), "start", f"--{args.session}"] + if args.config: + cmd.extend(["-c", str(args.config)]) + os.execv(cmd[0], cmd) + + cfg = load_config(args.config) + if args.no_stack: + cfg.auto_start = False + if args.tcp_port is not None: + cfg.tcp_port = args.tcp_port + if args.no_serial: + cfg.serial_enabled = False + + if cfg.modular_tcp.enabled and cfg.modular_tcp.role == "main": + svc = ModularTcpMainService(cfg.modular_tcp, cfg.tcp_host, cfg.tcp_port, log) + svc.start() + log( + f"modular TCP/IP Servers Service — Main '{cfg.modular_tcp.service_name}' " + f"({len(cfg.modular_tcp.secondaries)} secondaries)" + ) + try: + while True: + time.sleep(1.0) + except KeyboardInterrupt: + pass + finally: + svc.stop() + return 0 + + state = DaemonState(cfg=cfg, bans=BanList(cfg.bans_file)) + init_device_runtimes(state) + emit_startup_banner( + config_path=cfg.config_path or None, + cfg=cfg, + devices=cfg.devices, + tested_fn=registry_tested, + ) + listeners = make_listeners(cfg) + try: + serve(state, listeners) + except KeyboardInterrupt: + stop_stacks(state) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/modular_tcp_server.py b/stacks/daemon/modular_tcp_server.py new file mode 100644 index 0000000..b7f5612 --- /dev/null +++ b/stacks/daemon/modular_tcp_server.py @@ -0,0 +1,208 @@ +""" +Modular TCP/IP Servers Service — central Main + Secondary routing for MAX25. + +Public name in docs: "modular TCP/IP Servers Service". +Main instance accepts M25/1 clients and forwards sessions to Secondary max25d peers. +""" +from __future__ import annotations + +import configparser +import selectors +import socket +import threading +from dataclasses import dataclass, field +from typing import Callable, Optional + + +LogFn = Callable[[str], None] + + +@dataclass +class SecondaryEndpoint: + name: str + host: str + port: int + + +@dataclass +class ModularTcpConfig: + enabled: bool = False + role: str = "standalone" # standalone | main | secondary + service_name: str = "modular-tcp-server" + instance_id: str = "" + secondaries: list[SecondaryEndpoint] = field(default_factory=list) + connect_timeout: float = 5.0 + buffer_size: int = 65536 + + +def load_modular_tcp(cp: configparser.ConfigParser) -> ModularTcpConfig: + cfg = ModularTcpConfig() + if not cp.has_section("modular_tcp"): + return cfg + sec = cp["modular_tcp"] + cfg.enabled = sec.get("enabled", "no").strip().lower() in ("1", "yes", "true", "on") + cfg.role = sec.get("role", cfg.role).strip().lower() or cfg.role + cfg.service_name = sec.get("service_name", cfg.service_name).strip() or cfg.service_name + cfg.instance_id = sec.get("instance_id", cfg.instance_id).strip() + if cp.has_option("modular_tcp", "connect_timeout"): + try: + cfg.connect_timeout = float(cp.get("modular_tcp", "connect_timeout")) + except (TypeError, ValueError): + cfg.connect_timeout = 5.0 + if cfg.connect_timeout <= 0.0: + cfg.connect_timeout = 5.0 + if cp.has_section("modular_tcp.secondaries"): + for key in cp.options("modular_tcp.secondaries"): + raw = cp.get("modular_tcp.secondaries", key).strip() + if not raw: + continue + host, _, port_s = raw.partition(":") + host = host.strip() + if not host: + continue + try: + port = int(port_s or "7326") + except (TypeError, ValueError): + port = 7326 + if port < 1 or port > 65535: + port = 7326 + cfg.secondaries.append(SecondaryEndpoint(name=key.strip(), host=host, port=port)) + if cfg.enabled and cfg.role == "main" and not cfg.secondaries: + # Legacy comma list: secondaries = host:port,host:port + raw = sec.get("secondaries", "").strip() + if raw: + for idx, item in enumerate(x.strip() for x in raw.split(",") if x.strip()): + host, _, port_s = item.partition(":") + host = host.strip() + if not host: + continue + try: + port = int(port_s or "7326") + except (TypeError, ValueError): + port = 7326 + if port < 1 or port > 65535: + port = 7326 + cfg.secondaries.append( + SecondaryEndpoint( + name=f"secondary-{idx + 1}", + host=host, + port=port, + ) + ) + return cfg + + +class ModularTcpMainService: + """Bidirectional relay: operator client <-> Secondary max25d M25/1.""" + + def __init__( + self, + cfg: ModularTcpConfig, + listen_host: str, + listen_port: int, + log: LogFn, + ) -> None: + self.cfg = cfg + self.listen_host = listen_host + self.listen_port = listen_port + self.log = log + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._rr_index = 0 + self._lock = threading.Lock() + + def _pick_secondary(self) -> Optional[SecondaryEndpoint]: + with self._lock: + if not self.cfg.secondaries: + return None + ep = self.cfg.secondaries[self._rr_index % len(self.cfg.secondaries)] + self._rr_index += 1 + return ep + + def _relay(self, client: socket.socket, upstream: socket.socket) -> None: + sel = selectors.DefaultSelector() + sel.register(client, selectors.EVENT_READ) + sel.register(upstream, selectors.EVENT_READ) + try: + while not self._stop.is_set(): + for key, _ in sel.select(timeout=0.5): + sock = key.fileobj + assert isinstance(sock, socket.socket) + try: + data = sock.recv(self.cfg.buffer_size) + except OSError: + return + if not data: + return + target = upstream if sock is client else client + try: + target.sendall(data) + except OSError: + return + finally: + sel.close() + + def _handle_client(self, client: socket.socket, addr: tuple[str, int]) -> None: + ep = self._pick_secondary() + if ep is None: + self.log(f"modular_tcp: no secondaries configured — drop {addr}") + client.close() + return + upstream = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + upstream.settimeout(self.cfg.connect_timeout) + try: + upstream.connect((ep.host, ep.port)) + except OSError as exc: + self.log(f"modular_tcp: upstream {ep.name} {ep.host}:{ep.port} failed ({exc})") + client.close() + return + upstream.settimeout(None) + self.log(f"modular_tcp: {addr} -> {ep.name} {ep.host}:{ep.port}") + try: + self._relay(client, upstream) + finally: + for s in (client, upstream): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + s.close() + + def _serve(self) -> None: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind((self.listen_host, self.listen_port)) + srv.listen(64) + srv.settimeout(1.0) + self.log( + f"modular_tcp: Main service '{self.cfg.service_name}' " + f"listen {self.listen_host}:{self.listen_port} " + f"secondaries={len(self.cfg.secondaries)}" + ) + try: + while not self._stop.is_set(): + try: + client, addr = srv.accept() + except socket.timeout: + continue + except OSError: + break + threading.Thread( + target=self._handle_client, + args=(client, addr), + daemon=True, + ).start() + finally: + srv.close() + + def start(self) -> None: + if self._thread is not None: + return + self._thread = threading.Thread(target=self._serve, name="modular-tcp-main", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=3.0) + self._thread = None diff --git a/stacks/daemon/paths.py b/stacks/daemon/paths.py new file mode 100644 index 0000000..62d4974 --- /dev/null +++ b/stacks/daemon/paths.py @@ -0,0 +1,281 @@ +"""MAX25 layout detection — dev checkout vs cmake install prefix.""" +from __future__ import annotations + +import configparser +import os +import sys +from pathlib import Path + +MAX25_BCPR_KISS_DEFAULT = "/tmp/max25-bcpr/kiss-bc0" + + +def normalize_max25_bcpr_path(path: str) -> str: + """Map legacy /tmp/bcpr and /var/run/bcpr paths to canonical max25-bcpr layout.""" + p = (path or "").strip() + if not p: + return p + for old, new in ( + ("/tmp/bcpr/", "/tmp/max25-bcpr/"), + ("/tmp/bcpr", "/tmp/max25-bcpr"), + ("/var/run/bcpr/", "/tmp/max25-bcpr/"), + ("/var/run/bcpr", "/tmp/max25-bcpr"), + ): + if p == old: + return new + if p.startswith(old + "/") or (old.endswith("/") and p.startswith(old)): + return new + p[len(old) :] + return p + + +_MANIFEST = Path("plugins/manifest.yaml") +_SHARE_MAX25 = Path("share/max25") +_STACKS = Path("stacks") + + +def _repo_from_daemon_dir(here: Path) -> Path | None: + if here.name == "daemon" and (here.parent / "tncs").is_dir(): + repo = here.parent.parent + if (repo / _MANIFEST).is_file(): + return repo + return None + + +def _prefix_from_bindir(bindir: Path) -> Path | None: + prefix = bindir.parent + if (prefix / _SHARE_MAX25).is_dir() and not (prefix / _MANIFEST).is_file(): + return prefix + return None + + +def resolve_layout(exe: Path) -> tuple[Path, Path | None]: + """Return (tree_root, install_prefix). + + tree_root is the repo root in dev, or MAX25_ROOT / prefix when installed. + install_prefix is set when exe lives under PREFIX/bin, else None. + """ + exe = exe.resolve() + here = exe.parent + + repo = _repo_from_daemon_dir(here) + if repo is not None: + return repo, None + + prefix = _prefix_from_bindir(here) + if prefix is not None: + tree = Path(os.environ.get("MAX25_ROOT", str(prefix))) + return tree, prefix + + if (here.parent / _MANIFEST).is_file(): + return here.parent, None + + return exe.parents[2], None + + +def share_max25_dir(tree: Path, prefix: Path | None) -> Path: + if prefix is not None: + share = prefix / _SHARE_MAX25 + if share.is_dir(): + return share + return tree / _SHARE_MAX25 + + +def stacks_dir(tree: Path, prefix: Path | None) -> Path: + tree_stacks = tree / _STACKS + if tree_stacks.is_dir(): + return tree_stacks + if prefix is not None: + installed = prefix / _SHARE_MAX25 / _STACKS + if installed.is_dir(): + return installed + return tree_stacks + + +def tnc_serial_recovery_path(tree: Path, prefix: Path | None) -> Path | None: + """Locate tnc_serial_recovery.py (dev checkout or cmake install).""" + for candidate in ( + stacks_dir(tree, prefix) / "tncs" / "tnc_serial_recovery.py", + tree / _STACKS / "tncs" / "tnc_serial_recovery.py", + ): + if candidate.is_file(): + return candidate + return None + + +def ctl_path(tree: Path, prefix: Path | None, exe: Path) -> Path: + if prefix is not None: + sibling = exe.resolve().parent / "max25-ctl" + if sibling.is_file(): + return sibling + script = tree / "scripts" / "max25-ctl" + if script.is_file(): + return script + if prefix is not None: + return prefix / "bin" / "max25-ctl" + return script + + +def default_ini_candidates(tree: Path, prefix: Path | None) -> list[Path]: + candidates: list[Path] = [] + env_ini = os.environ.get("MAX25D_INI", "").strip() + if env_ini: + candidates.append(Path(env_ini)) + candidates.append(Path("/etc/max25/max25d.ini")) + candidates.append(share_max25_dir(tree, prefix) / "max25d.ini.example") + if sys.platform.startswith("freebsd"): + candidates.append(share_max25_dir(tree, prefix) / "max25d.freebsd.ini.example") + if prefix is None or tree != prefix: + candidates.append(tree / _SHARE_MAX25 / "max25d.ini.example") + return candidates + + +def serial_env_candidates(device_id: str, tree: Path, prefix: Path | None) -> list[Path]: + paths: list[Path] = [Path(f"/etc/max25/{device_id}-serial.env")] + env_root = os.environ.get("MAX25_ROOT", "").strip() + if env_root: + paths.append(Path(env_root) / "local" / f"{device_id}-serial.env") + paths.append(Path(env_root) / _STACKS / "tncs" / f"{device_id}-serial.env") + cwd = os.environ.get("PWD", "").strip() + if cwd: + paths.append(Path(cwd) / "local" / f"{device_id}-serial.env") + if prefix is not None: + paths.append(prefix / _SHARE_MAX25 / "serial" / f"{device_id}-serial.env") + paths.append(prefix / _SHARE_MAX25 / _STACKS / "tncs" / f"{device_id}-serial.env") + paths.append(tree / "local" / f"{device_id}-serial.env") + paths.append(tree / _STACKS / "tncs" / f"{device_id}-serial.env") + return paths + + +def baycom_share_dir(tree: Path, prefix: Path | None) -> Path: + if prefix is not None: + installed = prefix / _SHARE_MAX25 / "baycom" + if installed.is_dir(): + return installed + return tree / "share" / "baycom" + + +_SITE_BAYCOM_INI = Path("/etc/baycom/baycom-pr.ini") +_SINGLE_SER12_EXAMPLE = "baycom-pr.pccom-ttyS0-only.ini.example" + + +def is_dual_baycom_ini(path: Path) -> bool: + """True when profile lists more than one modem or name is 'dual'.""" + if not path.is_file(): + return False + cp = configparser.ConfigParser() + try: + cp.read(path, encoding="utf-8") + except OSError: + return False + if "profile" not in cp: + return False + prof = cp["profile"] + name = prof.get("name", "").strip().lower() + modem_ids = [x.strip() for x in prof.get("modems", "").split(",") if x.strip()] + if len(modem_ids) > 1: + return True + return name == "dual" + + +def canonical_dual_baycom_example( + device_id: str, tree: Path, prefix: Path | None +) -> Path | None: + """Shipped dual-modem template for kernel-ser12 service mode.""" + if device_id not in ("baycom-ser12", "baycom-par96"): + return None + for candidate in ( + tree / _STACKS / "baycom-pr" / "config" / "examples" / "baycom-pr.dual.ini", + stacks_dir(tree, prefix) / "baycom-pr" / "config" / "examples" / "baycom-pr.dual.ini", + ): + if candidate.is_file(): + return candidate + site = _SITE_BAYCOM_INI + if site.is_file() and is_dual_baycom_ini(site): + return site + return None + + +def canonical_single_baycom_example( + device_id: str, tree: Path, prefix: Path | None +) -> Path | None: + """Shipped single-modem template for kernel BayCom devices.""" + share = baycom_share_dir(tree, prefix) + if device_id == "baycom-par96": + for candidate in ( + share / "baycom-pr.par96-single.ini.example", + tree / _STACKS / "baycom-pr" / "config" / "examples" / "baycom-pr.par96.ini", + stacks_dir(tree, prefix) / "baycom-pr" / "config" / "examples" / "baycom-pr.par96.ini", + ): + if candidate.is_file(): + return candidate + return None + ser12 = share / _SINGLE_SER12_EXAMPLE + if ser12.is_file(): + return ser12 + return None + + +def baycom_ini_candidates(device_id: str, tree: Path, prefix: Path | None) -> list[Path]: + """Search order for baycom-pr.ini (local → site single → shipped example).""" + paths: list[Path] = [] + env_ini = os.environ.get("BAYCOM_INI", "").strip() + if env_ini: + paths.append(Path(env_ini)) + env_root = os.environ.get("MAX25_ROOT", "").strip() + if env_root: + paths.append(Path(env_root) / "local" / "baycom-pr.ini") + cwd = os.environ.get("PWD", "").strip() + if cwd: + paths.append(Path(cwd) / "local" / "baycom-pr.ini") + paths.append(tree / "local" / "baycom-pr.ini") + + # Site INI before shipped example so operator edits in /etc win when single-modem. + paths.append(_SITE_BAYCOM_INI) + paths.append(Path("/etc/baycom/baycom-pr-single.ini")) + + canonical = canonical_single_baycom_example(device_id, tree, prefix) + if canonical is not None: + paths.append(canonical) + + if prefix is not None: + paths.append(prefix / _SHARE_MAX25 / "baycom" / _SINGLE_SER12_EXAMPLE) + paths.append(prefix / "etc" / "baycom" / "baycom-pr.ini.example") + + paths.append(tree / _STACKS / "baycom-pr" / "config" / "baycom-pr.ini") + return paths + + +def resolve_baycom_profile( + profile: str, + device_id: str, + tree: Path, + prefix: Path | None, +) -> Path | None: + """Resolve named profile: single (default) or dual (service mode).""" + name = profile.strip().lower() + if name == "dual": + return canonical_dual_baycom_example(device_id, tree, prefix) + if name in ("single", "default", ""): + return resolve_baycom_ini(device_id, tree, prefix) + return None + + +def resolve_baycom_ini( + device_id: str, + tree: Path, + prefix: Path | None, + explicit: str = "", +) -> Path | None: + """Return first existing baycom-pr.ini path, or None. + + Skips dual-modem /etc/baycom/baycom-pr.ini unless explicit (or env BAYCOM_INI). + """ + if explicit: + path = Path(explicit) + return path if path.is_file() else None + for candidate in baycom_ini_candidates(device_id, tree, prefix): + if not candidate.is_file(): + continue + if candidate == _SITE_BAYCOM_INI and is_dual_baycom_ini(candidate): + continue + return candidate + return None diff --git a/stacks/daemon/privilege_drop.py b/stacks/daemon/privilege_drop.py new file mode 100644 index 0000000..5e68ca8 --- /dev/null +++ b/stacks/daemon/privilege_drop.py @@ -0,0 +1,110 @@ +"""Drop max25d privileges after hardware stack is up (Linux).""" +from __future__ import annotations + +import configparser +import grp +import os +import pwd +import sys +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class RunAsConfig: + user: str = "" + group: str = "" + uid: Optional[int] = None + gid: Optional[int] = None + + +def _parse_optional_int(raw: str) -> Optional[int]: + text = str(raw).strip() + if not text: + return None + return int(text) + + +def parse_run_as(cp: configparser.ConfigParser) -> RunAsConfig: + cfg = RunAsConfig() + if not cp.has_section("daemon"): + return cfg + sec = cp["daemon"] + cfg.user = sec.get("user", fallback="").strip() + cfg.group = sec.get("group", fallback="").strip() + if cp.has_option("daemon", "uid"): + cfg.uid = _parse_optional_int(sec.get("uid", fallback="")) + if cp.has_option("daemon", "gid"): + cfg.gid = _parse_optional_int(sec.get("gid", fallback="")) + return cfg + + +def run_as_configured(cfg: RunAsConfig) -> bool: + return bool( + cfg.user + or cfg.group + or cfg.uid is not None + or cfg.gid is not None + ) + + +def resolve_run_as(cfg: RunAsConfig) -> tuple[int, int, str]: + pw: pwd.struct_passwd | None = None + name = cfg.user + + if cfg.user: + pw = pwd.getpwnam(cfg.user) + uid = pw.pw_uid + gid = pw.pw_gid + elif cfg.uid is not None: + pw = pwd.getpwuid(cfg.uid) + uid = pw.pw_uid + gid = pw.pw_gid + name = pw.pw_name + else: + raise ValueError("[daemon] user= or uid= required for privilege drop") + + if cfg.uid is not None: + uid = cfg.uid + if cfg.group: + gid = grp.getgrnam(cfg.group).gr_gid + elif cfg.gid is not None: + gid = cfg.gid + + return uid, gid, name + + +def drop_privileges(cfg: RunAsConfig, *, log) -> None: + if os.name != "posix": + return + if not run_as_configured(cfg): + if os.geteuid() == 0: + log( + "running as root after stack start — set [daemon] user= to drop", + area="privilege", + ) + return + if os.geteuid() != 0: + return + + uid, gid, name = resolve_run_as(cfg) + + os.initgroups(name, gid) + os.setgid(gid) + os.setuid(uid) + + if os.geteuid() == 0: + raise OSError("privilege drop failed — still root") + + log( + f"dropped privileges to uid={uid} gid={gid} ({name})", + area="privilege", + ) + + +def drop_privileges_or_exit(cfg: RunAsConfig, *, log) -> None: + try: + drop_privileges(cfg, log=log) + except (KeyError, ValueError, OSError) as exc: + log(f"privilege drop failed: {exc}", area="privilege") + raise SystemExit(1) from exc diff --git a/stacks/daemon/reporting_quality.py b/stacks/daemon/reporting_quality.py new file mode 100644 index 0000000..f755f08 --- /dev/null +++ b/stacks/daemon/reporting_quality.py @@ -0,0 +1,178 @@ +"""RX data-quality window for max25d STATUS error=/voice= reporting.""" +from __future__ import annotations + +import re +import time +from collections import deque +from dataclasses import dataclass, field +from enum import Enum +from typing import Deque, Optional + +_AX25_UI_LINE_RE = re.compile( + r"^\[(?:CRDOP )?AX25 UI ([^>]+)>([^\]]+)\]\s?(.*)$", + re.DOTALL, +) +_AX25_UI_PARTIAL_RE = re.compile(r"\[AX25 UI ", re.IGNORECASE) +_VOICE_NOISE_LINE_RE = re.compile( + r"^\[(?:CRDOP RX|AUDIO RX|VOICE)[^\]]*\]", + re.IGNORECASE, +) + + +class RxOutcome(str, Enum): + GOOD = "good" + BAD = "bad" + IGNORE = "ignore" + + +def classify_rx_line(line: str, callid: str) -> RxOutcome: + """ + Classify an RX display line for data-quality tracking. + + - GOOD: complete AX.25 UI frame whose destination matches configured CALLID. + - BAD: incomplete AX.25 UI or wrong CALLID (incomplete data counts as error). + - IGNORE: voice/noise/interference — excluded from data pass statistics. + """ + text = line.strip() + if not text: + return RxOutcome.IGNORE + + match = _AX25_UI_LINE_RE.match(text) + if match: + dest = match.group(2).strip().upper() + expected = callid.strip().upper() + if not expected or dest != expected: + return RxOutcome.BAD + return RxOutcome.GOOD + + if _AX25_UI_PARTIAL_RE.search(text): + return RxOutcome.BAD + + if _VOICE_NOISE_LINE_RE.match(text): + return RxOutcome.IGNORE + + return RxOutcome.IGNORE + + +@dataclass +class _PassWindow: + good: int = 0 + bad: int = 0 + ignored: bool = False + started_at: float = 0.0 + + +@dataclass +class DataQualityTracker: + """ + Rolling pass window: each pass lasts up to pass_window_sec (default 20s). + passes_required consecutive passes (default 3) form one evaluation cycle + (3 × max 20s = up to 60s). Voice/noise does not fill a pass slot. + """ + + passes_required: int = 3 + min_good_percent: int = 50 + pass_window_sec: int = 20 + passes: Deque[str] = field(default_factory=deque) + voice_activity: bool = False + current: _PassWindow = field(default_factory=_PassWindow) + + def __post_init__(self) -> None: + if self.passes_required < 1: + self.passes_required = 1 + if self.min_good_percent < 1: + self.min_good_percent = 1 + elif self.min_good_percent > 100: + self.min_good_percent = 100 + if self.pass_window_sec < 1: + self.pass_window_sec = 1 + elif self.pass_window_sec > 300: + self.pass_window_sec = 300 + self.passes = deque(maxlen=self.passes_required) + + def _now(self, now: Optional[float]) -> float: + return time.time() if now is None else now + + def _ensure_window(self, now: float) -> None: + if self.current.started_at == 0.0: + self.current.started_at = now + + def _pass_outcome_from_counts(self, good: int, bad: int) -> str: + if bad > 0: + return "bad" + if good > 0: + return "good" + return "bad" + + def _finalize_current_pass(self) -> None: + good = self.current.good + bad = self.current.bad + if good == 0 and bad == 0: + if self.current.ignored: + return + self.passes.append("bad") + return + self.passes.append(self._pass_outcome_from_counts(good, bad)) + + def _roll_window(self, now: float) -> None: + self._finalize_current_pass() + self.current = _PassWindow(started_at=now) + + def tick(self, now: Optional[float] = None) -> None: + """Expire the active pass when pass_window_sec elapsed.""" + ts = self._now(now) + if self.current.started_at == 0.0: + return + if ts - self.current.started_at < self.pass_window_sec: + return + self._roll_window(ts) + + def record(self, outcome: RxOutcome, now: Optional[float] = None) -> None: + ts = self._now(now) + self.tick(ts) + self._ensure_window(ts) + + if outcome == RxOutcome.IGNORE: + self.voice_activity = True + self.current.ignored = True + return + if outcome == RxOutcome.GOOD: + self.current.good += 1 + elif outcome == RxOutcome.BAD: + self.current.bad += 1 + + def record_bad(self, now: Optional[float] = None) -> None: + self.record(RxOutcome.BAD, now) + + def good_ratio_percent(self) -> Optional[int]: + if len(self.passes) < self.passes_required: + return None + good = sum(1 for item in self.passes if item == "good") + total = len(self.passes) + if total == 0: + return None + return (100 * good) // total + + def data_error_valid(self, *, reporting_enabled: bool) -> bool: + if not reporting_enabled: + return False + ratio = self.good_ratio_percent() + if ratio is None: + return False + return ratio >= self.min_good_percent + + def voice_signal_valid(self, *, reporting_enabled: bool, link_healthy: bool) -> bool: + """ + Voice/noise/disturbance path: when link is healthy and voice activity was + seen (non-data RX), treat as 100% valid even if the content was noise only. + """ + if not reporting_enabled: + return False + if not link_healthy: + return False + if self.voice_activity: + return True + return True + + def max_cycle_seconds(self) -> int: + return self.passes_required * self.pass_window_sec diff --git a/stacks/daemon/test_audio_dummy_backend.py b/stacks/daemon/test_audio_dummy_backend.py new file mode 100644 index 0000000..c5ba31d --- /dev/null +++ b/stacks/daemon/test_audio_dummy_backend.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""Acoustic engine loopback + audio-dummy backend smoke tests.""" +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) +sys.path.insert(0, str(ROOT / "stacks" / "crdop" / "lib")) + +from acoustic_engine import AcousticEngine # noqa: E402 +from device_backends import AudioDummyBackend, DeviceBackendConfig, registry_tested # noqa: E402 + + +def test_loopback_decode() -> None: + eng = AcousticEngine() + rep = eng.loopback_self_test("TST-0", "QST") + assert rep.symbols > 0 + assert rep.transitions > 0 + + +def test_audio_dummy_backend_loopback_tx() -> None: + cfg = DeviceBackendConfig(device_id="audio-dummy", backend_type="audio-dummy", audio_mode="loopback") + rx: list[str] = [] + backend = AudioDummyBackend(cfg, on_rx=rx.append) + assert backend.open() + assert backend.attach_session("TST-0") + ok, display = backend.transmit("TST-0", "QST", "hi", ax25_ui=True) + assert ok + assert "QST" in display + backend.close() + + +def test_registry() -> None: + assert registry_tested("audio-dummy") is True + + +def main() -> int: + test_loopback_decode() + test_audio_dummy_backend_loopback_tx() + test_registry() + print("OK: audio-dummy tests") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/stacks/daemon/test_auth.py b/stacks/daemon/test_auth.py new file mode 100644 index 0000000..263a2e5 --- /dev/null +++ b/stacks/daemon/test_auth.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +"""TCP auth smoke test for max25d M25/1.""" +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +DAEMON = ROOT / "stacks/daemon/max25d" +BASE_INI = ROOT / "share/max25/max25d.ini.example" + + +class LineReader: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buf = b"" + + def read(self, timeout: float = 5.0) -> str: + self.sock.settimeout(timeout) + while b"\n" not in self.buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + return line.decode("utf-8") + + +def free_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def main() -> int: + port = free_port() + password = "preview-secret" + + with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as tmp: + tmp.write(BASE_INI.read_text()) + tmp.write(f"\n[network]\ntcp_password = {password}\n") + ini_path = Path(tmp.name) + + proc = subprocess.Popen( + [str(DAEMON), "--no-stack", "-c", str(ini_path), "--tcp-port", str(port)], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + time.sleep(0.6) + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + except OSError as exc: + proc.terminate() + proc.wait(timeout=3) + print(f"FAIL: connect: {exc}", file=sys.stderr) + return 1 + + reader = LineReader(sock) + try: + assert reader.read() == "AUTH required" + + sock.sendall(b"AUTH wrong\n") + assert reader.read() == "ERR auth failed" + sock.close() + + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + reader = LineReader(sock) + assert reader.read() == "AUTH required" + sock.sendall(f"AUTH {password}\n".encode()) + assert reader.read() == "OK" + status = reader.read() + assert status.startswith("STATUS ") + + sock.sendall(b"PING\n") + assert reader.read() == "OK" + + print("OK: max25d TCP auth smoke") + return 0 + finally: + try: + sock.close() + except OSError: + pass + proc.terminate() + proc.wait(timeout=5) + ini_path.unlink(missing_ok=True) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_ax25_codec.py b/stacks/daemon/test_ax25_codec.py new file mode 100644 index 0000000..b12e43d --- /dev/null +++ b/stacks/daemon/test_ax25_codec.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Unit tests for ax25_codec (RFC 1171 FCS + libax25 address rules).""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from ax25_codec import ( # noqa: E402 + ax25_build_ui, + ax25_crc, + ax25_crc_valid, + ax25_encode_address, + ax25_parse_ui, + format_callsign, + parse_callsign, + validate_callsign, +) + + +def test_rfc1171_crc_vector() -> None: + body = bytes(range(256)) + crc = ax25_crc(body) + frame = body + bytes((crc & 0xFF, crc >> 8)) + assert ax25_crc_valid(frame) + assert crc == 0x303C + + +def test_address_roundtrip() -> None: + raw = ax25_encode_address("DG1ABC", 7, last=True) + frame = ax25_build_ui("DG1ABC-7", "CQ", b"73") + parsed = ax25_parse_ui(frame) + assert parsed == ("DG1ABC-7", "CQ", b"73") + + +def test_format_callsign_omit_zero_ssid() -> None: + assert format_callsign("DG1ABC", 0) == "DG1ABC" + assert format_callsign("CB", 0) == "CB" + + +def test_parse_callsign_legacy() -> None: + assert parse_callsign("DK0WC-7") == ("DK0WC", 7) + assert parse_callsign("bad-ssid") == ("BAD", 0) + + +def test_validate_callsign_strict() -> None: + assert validate_callsign("CB-0") == ("CB", 0) + assert validate_callsign("N0CALL-15") == ("N0CALL", 15) + try: + validate_callsign("CB_0") + raise AssertionError("expected ValueError") + except ValueError: + pass + + +def test_ui_frame_without_fcs() -> None: + frame = ax25_build_ui("DG1ABC", "QST", b"hi") + body = frame[:-2] + parsed = ax25_parse_ui(body) + assert parsed == ("DG1ABC", "QST", b"hi") + + +def test_invalid_callsign_build() -> None: + try: + ax25_build_ui("BAD_CALL", "CQ", b"x") + raise AssertionError("expected ValueError") + except ValueError: + pass + + +def test_non_ui_rejected() -> None: + frame = ax25_build_ui("A", "B", b"x") + frame = bytearray(frame) + frame[14] = 0x00 # not UI control + assert ax25_parse_ui(bytes(frame)) is None + + +def main() -> int: + tests = [ + test_rfc1171_crc_vector, + test_address_roundtrip, + test_format_callsign_omit_zero_ssid, + test_parse_callsign_legacy, + test_validate_callsign_strict, + test_ui_frame_without_fcs, + test_invalid_callsign_build, + test_non_ui_rejected, + ] + for fn in tests: + fn() + print("OK: ax25_codec unit tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_bans.py b/stacks/daemon/test_bans.py new file mode 100644 index 0000000..80176ae --- /dev/null +++ b/stacks/daemon/test_bans.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Unit and protocol tests for max25d AX.25 source ban list.""" +from __future__ import annotations + +import importlib.util +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +LAUNCHER = ROOT / "stacks/daemon/max25d" +DAEMON = ROOT / "stacks/daemon/max25d.py" + +sys.path.insert(0, str(DAEMON.parent)) +from banlist import BanList, callsign_banned, extract_ax25_source # noqa: E402 + + +def load_max25d_module(): + from importlib.machinery import SourceFileLoader + + loader = SourceFileLoader("max25d", str(DAEMON)) + spec = importlib.util.spec_from_loader("max25d", loader) + mod = importlib.util.module_from_spec(spec) + sys.modules["max25d"] = mod + loader.exec_module(mod) + return mod + + +class LineReader: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buf = b"" + + def read(self, timeout: float = 5.0) -> str: + self.sock.settimeout(timeout) + while b"\n" not in self.buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + return line.decode("utf-8") + + +def free_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def test_extract_ax25_source() -> None: + assert extract_ax25_source("[AX25 UI DG1ABC>QST] 73") == "DG1ABC" + assert extract_ax25_source("[AX25 UI DK0WC-7>CB-0] hi") == "DK0WC-7" + assert extract_ax25_source("plain text") is None + assert extract_ax25_source("[CRDOP RX soft-crdop] hello") is None + + +def test_callsign_banned_matching() -> None: + assert callsign_banned("DG1ABC", "DG1ABC") + assert callsign_banned("DG1ABC", "DG1ABC-7") + assert not callsign_banned("DG1ABC", "DG1ABD") + assert callsign_banned("DK0WC-7", "DK0WC-7") + assert not callsign_banned("DK0WC-7", "DK0WC") + assert not callsign_banned("DK0WC-7", "DK0WC-8") + + +def test_banlist_file_roundtrip() -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "bans.txt" + path.write_text("# ignored comment\nDG1ABC\n\nDK0WC-7\n", encoding="utf-8") + bans = BanList(path) + assert bans.list() == ["DG1ABC", "DK0WC-7"] + assert bans.is_banned("DG1ABC-3") + assert bans.is_banned("DK0WC-7") + assert not bans.is_banned("DK0WC") + + bans.add("DL1TEST") + bans.remove("DG1ABC") + assert bans.list() == ["DK0WC-7", "DL1TEST"] + reloaded = BanList(path) + assert reloaded.list() == ["DK0WC-7", "DL1TEST"] + + +def test_on_backend_rx_silent_drop() -> None: + mod = load_max25d_module() + with tempfile.TemporaryDirectory() as tmp: + bans_path = Path(tmp) / "bans.txt" + bans_path.write_text("DG1ABC\n", encoding="utf-8") + cfg = mod.DaemonConfig(bans_file=str(bans_path)) + state = mod.DaemonState(cfg=cfg, bans=BanList(bans_path)) + mod.init_device_runtimes(state) + + sent: list[str] = [] + + def capture(_state, line, skip=None): + sent.append(line) + + with patch.object(mod, "broadcast", side_effect=capture), patch.object(mod, "log"): + mod.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>QST] blocked") + assert sent == [] + + mod.on_backend_rx(state, "tnc2c", "[AX25 UI DK0WC>QST] allowed") + assert sent == ["RX device=tnc2c [AX25 UI DK0WC>QST] allowed"] + + +def test_ban_commands_protocol() -> None: + port = free_port() + with tempfile.TemporaryDirectory() as tmp: + bans_path = Path(tmp) / "bans.txt" + ini_path = Path(tmp) / "max25d.ini" + ini_path.write_text( + f""" +[daemon] +device = tnc2c +hardware = tncs +[network] +tcp_host = 127.0.0.1 +tcp_port = {port} +[modem] +bans_file = {bans_path} +[stack] +auto_start = no +""", + encoding="utf-8", + ) + + proc = subprocess.Popen( + [ + str(LAUNCHER), + "--no-stack", + "--no-serial", + "-c", + str(ini_path), + "--tcp-port", + str(port), + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + time.sleep(0.6) + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + except OSError as exc: + proc.terminate() + proc.wait(timeout=3) + raise RuntimeError(f"connect failed: {exc}") from exc + + reader = LineReader(sock) + try: + assert reader.read() == "OK" + reader.read() # STATUS + + sock.sendall(b"BAN DG1ABC\n") + assert reader.read() == "OK" + assert bans_path.read_text(encoding="utf-8").strip() == "DG1ABC" + + sock.sendall(b"BANS\n") + assert reader.read() == "BAN DG1ABC" + assert reader.read() == "OK" + + sock.sendall(b"UNBAN DG1ABC\n") + assert reader.read() == "OK" + assert bans_path.read_text(encoding="utf-8") == "" + + sock.sendall(b"UNBAN DG1ABC\n") + assert reader.read() == "ERR not banned" + + sock.sendall(b"BAN BAD!\n") + assert reader.read() == "ERR invalid callsign" + finally: + sock.close() + proc.terminate() + proc.wait(timeout=5) + + +def main() -> int: + tests = [ + test_extract_ax25_source, + test_callsign_banned_matching, + test_banlist_file_roundtrip, + test_on_backend_rx_silent_drop, + test_ban_commands_protocol, + ] + for test in tests: + test() + print(f"OK: {test.__name__}") + print("OK: banlist tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_baycom_backend.py b/stacks/daemon/test_baycom_backend.py new file mode 100644 index 0000000..ebff5a4 --- /dev/null +++ b/stacks/daemon/test_baycom_backend.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +"""Offline unit tests for BayComKissBackend (no kernel hardware).""" +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) + +from device_backends import BayComKissBackend, DeviceBackendConfig # noqa: E402 +from ax25_codec import ax25_build_ui, ax25_parse_ui # noqa: E402 +from kiss_bridge import KissDecoder, format_rx_line, kiss_data_frame # noqa: E402 + + +def test_baycom_kiss_default_link() -> None: + cfg = DeviceBackendConfig(device_id="baycom-ser12", backend_type="baycom-kiss", baycom_modem="a") + backend = BayComKissBackend(cfg, on_rx=lambda _l: None) + assert backend._path in ("/var/run/baycom-pr/kiss", f"/var/run/baycom-pr/kiss-a") + assert backend.backend_type == "baycom-kiss" + assert backend._is_pty is True + + +def test_baycom_kiss_transmit_kiss_frame() -> None: + cfg = DeviceBackendConfig( + device_id="baycom-ser12", + backend_type="baycom-kiss", + kiss_link="/var/run/baycom-pr/kiss", + ) + backend = BayComKissBackend(cfg, on_rx=lambda _l: None) + backend._fd = 99 + backend._kiss_active = True + written: list[bytes] = [] + + with patch("os.write", side_effect=lambda _fd, data: written.append(data) or len(data)): + with patch("termios.tcdrain"): + ok, display = backend.transmit("CB-0", "QST", "hello", ax25_ui=True) + assert ok, display + assert written and written[0].startswith(b"\xc0") + assert "hello" in display + + +def test_baycom_kiss_rx_decode_path() -> None: + """RX path: KISS bytes → AX.25 UI → on_rx line (same decoder as _rx_loop).""" + rx_lines: list[str] = [] + frame = ax25_build_ui("REMOTE-0", "CB-0", b"73") + pkt = kiss_data_frame(0, frame) + decoder = KissDecoder() + for _port, payload in decoder.feed(pkt): + parsed = ax25_parse_ui(payload) + assert parsed is not None + src, dst, info = parsed + assert src == "REMOTE" + assert dst == "CB" + assert info == b"73" + from kiss_bridge import format_rx_line + + rx_lines.append(format_rx_line(src, dst, info, ax25_ui=True)) + assert rx_lines and "73" in rx_lines[0] + + +def test_baycom_kiss_modem_b_link() -> None: + cfg = DeviceBackendConfig( + device_id="baycom-b", + backend_type="baycom-kiss", + baycom_modem="b", + kiss_link="/var/run/baycom-pr/kiss-b", + ) + backend = BayComKissBackend(cfg, on_rx=lambda _l: None) + assert backend._path == "/var/run/baycom-pr/kiss-b" + + +def test_baycom_kiss_open_missing_path() -> None: + backend = BayComKissBackend( + DeviceBackendConfig(device_id="baycom-ser12", kiss_link="/nonexistent/kiss/path"), + on_rx=lambda _l: None, + ) + assert not backend.open() + assert backend.status == "error-no-device" + + +def main() -> int: + test_baycom_kiss_default_link() + test_baycom_kiss_modem_b_link() + test_baycom_kiss_transmit_kiss_frame() + test_baycom_kiss_rx_decode_path() + test_baycom_kiss_open_missing_path() + print("OK: baycom backend tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_bcpr_backend.py b/stacks/daemon/test_bcpr_backend.py new file mode 100644 index 0000000..efcedd5 --- /dev/null +++ b/stacks/daemon/test_bcpr_backend.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""Offline unit tests for bcpr device parse + BcprKissBackend (no UART).""" +from __future__ import annotations + +import configparser +import importlib.util +import sys +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) + +from device_backends import ( # noqa: E402 + BcprKissBackend, + DeviceBackendConfig, + parse_device_spec, +) + + +def load_max25d(): + from importlib.machinery import SourceFileLoader + + loader = SourceFileLoader("max25d", str(ROOT / "stacks" / "daemon" / "max25d.py")) + spec = importlib.util.spec_from_loader("max25d", loader) + mod = importlib.util.module_from_spec(spec) + sys.modules["max25d"] = mod + loader.exec_module(mod) + return mod + + +def test_bcpr_kiss_default_link() -> None: + cfg = DeviceBackendConfig( + device_id="max25e0", + backend_type="max25-bcpr-kiss", + bcpr_device="bc0", + ) + backend = BcprKissBackend(cfg, on_rx=lambda _l: None) + assert backend._path == "/var/run/max25-bcpr/kiss-bc0" + assert backend.backend_type == "max25-bcpr-kiss" + assert backend._is_pty is True + + +def test_max25e0_default_addrs() -> None: + from device_backends import MAX25E0_DEFAULT_IPV4, MAX25E0_DEFAULT_IPV6 + text = """ +[devices] +max25e0 = max25-bcpr:bc0 +""" + cp = configparser.ConfigParser() + cp.read_string(text) + from device_backends import parse_device_spec + dev = parse_device_spec("max25e0", "max25-bcpr:bc0", cp, {"hardware": "modems"}) + assert dev.ipv4 == MAX25E0_DEFAULT_IPV4 + assert dev.ipv6 == MAX25E0_DEFAULT_IPV6 + # override + text2 = """ +[devices] +max25e0 = max25-bcpr:bc0 +[device.max25e0] +ipv4 = 10.0.0.25/24 +ipv6 = fd00::25/128 +""" + cp2 = configparser.ConfigParser() + cp2.read_string(text2) + dev2 = parse_device_spec("max25e0", "max25-bcpr:bc0", cp2, {"hardware": "modems"}) + assert dev2.ipv4 == "10.0.0.25/24" + assert dev2.ipv6 == "fd00::25/128" + # fork inherits from [device.max25e0] (INI colon keys need care — parse by id) + text3 = """ +[device.max25e0] +ipv4 = 10.0.0.25/24 +ipv6 = fd00::25/128 +""" + cp3 = configparser.ConfigParser() + cp3.read_string(text3) + fork = parse_device_spec("max25e0:bc1", "max25-bcpr:bc1", cp3, {"hardware": "modems"}) + assert fork.ipv4 == "10.0.0.25/24" + assert fork.ipv6 == "fd00::25/128" + + +def test_parse_bcpr_spec() -> None: + cp = configparser.ConfigParser() + cp.read_string( + """ +[device.max25e0] +kiss_link = /var/run/max25-bcpr/kiss-bc0 +max25_bcpr_ini = /etc/max25/max25-bcpr.ini +""" + ) + dev = parse_device_spec("max25e0", "max25-bcpr:bc0", cp, {"hardware": "tncs"}) + assert dev.backend_type == "max25-bcpr-kiss" + assert dev.max25_bcpr_device == "bc0" + assert dev.kiss_link == "/var/run/max25-bcpr/kiss-bc0" + assert dev.max25_bcpr_ini == "/etc/max25/max25-bcpr.ini" + + +def test_feature_gate_bcpr() -> None: + mod = load_max25d() + cp = configparser.ConfigParser() + cp.read_string( + """ +[features] +max25_bcpr = no +[devices] +default = max25e0 +max25e0 = max25-bcpr:bc0 +""" + ) + cfg = mod.DaemonConfig() + cfg.feature_max25_bcpr = False + devices = mod.parse_devices(cp, cfg) + # parse_devices itself may include; load_config filters by features + assert any(d.device_id == "max25e0" for d in devices) + filtered = [d for d in devices if mod._device_allowed_by_features(d, cfg)] + assert filtered == [] + + cfg.feature_max25_bcpr = True + filtered = [d for d in devices if mod._device_allowed_by_features(d, cfg)] + assert len(filtered) == 1 + assert filtered[0].backend_type == "max25-bcpr-kiss" + + +def test_bcpr_kiss_pty_skips_tcdrain() -> None: + cfg = DeviceBackendConfig( + device_id="max25e0", + backend_type="max25-bcpr-kiss", + kiss_link="/tmp/max25-bcpr/kiss-bc0", + bcpr_device="bc0", + ) + backend = BcprKissBackend(cfg, on_rx=lambda _l: None) + backend._fd = 99 + backend._kiss_active = True + written: list[bytes] = [] + drained = {"n": 0} + + def fake_drain(_fd: int) -> None: + drained["n"] += 1 + + with patch("os.write", side_effect=lambda _fd, data: written.append(data) or len(data)): + with patch("termios.tcdrain", side_effect=fake_drain): + ok, display = backend.transmit("CB-0", "QST", "hello", ax25_ui=True) + assert ok, display + assert written and written[0].startswith(b"\xc0") + assert drained["n"] == 0 + + +def test_bcpr_kiss_stabilize_reopens_on_inode_mismatch() -> None: + cfg = DeviceBackendConfig( + device_id="max25e0", + backend_type="max25-bcpr-kiss", + kiss_link="/tmp/max25-bcpr/kiss-bc0", + bcpr_device="bc0", + ) + backend = BcprKissBackend(cfg, on_rx=lambda _l: None) + backend._fd = 7 + backend._kiss_active = True + backend.status = "ready" + backend._mycall = "CB-0" + calls: list[str] = [] + + class _St: + def __init__(self, ino: int) -> None: + self.st_ino = ino + self.st_dev = 1 + + with patch("os.path.exists", return_value=True): + with patch("os.stat", return_value=_St(200)): + with patch("os.fstat", return_value=_St(100)): + with patch.object(backend, "close", side_effect=lambda: calls.append("close")): + with patch.object(backend, "open", side_effect=lambda: calls.append("open") or True): + with patch.object( + backend, + "attach_session", + side_effect=lambda c: calls.append(f"attach:{c}") or True, + ): + ok = backend.stabilize_session("CB-0", force=False) + assert ok + assert calls == ["close", "open", "attach:CB-0"] + + +def main() -> int: + test_bcpr_kiss_default_link() + test_max25e0_default_addrs() + test_parse_bcpr_spec() + test_feature_gate_bcpr() + test_bcpr_kiss_pty_skips_tcdrain() + test_bcpr_kiss_stabilize_reopens_on_inode_mismatch() + print("OK: bcpr backend tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_crdop_backend.py b/stacks/daemon/test_crdop_backend.py new file mode 100644 index 0000000..a83f54b --- /dev/null +++ b/stacks/daemon/test_crdop_backend.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Offline unit tests for CrdopTcpBackend (no crdopc process).""" +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) + +from device_backends import CrdopTcpBackend, DeviceBackendConfig # noqa: E402 + + +class _MockSock: + """Minimal socket double for ctrl/data channels.""" + + def __init__(self, term: str = "\n") -> None: + self.sent: list[bytes] = [] + self._closed = False + self._term = term + + def settimeout(self, _t: float) -> None: + return None + + def sendall(self, data: bytes) -> None: + self.sent.append(data) + + def recv(self, _n: int) -> bytes: + if self._closed: + return b"" + return f"OK{self._term}".encode("ascii") + + def close(self) -> None: + self._closed = True + + +def _mock_connections(ctrl: _MockSock, data: _MockSock): + def _create(addr, timeout=5.0): + _host, port = addr + return ctrl if port == 8515 else data + + return _create + + +def _open_ctx(ctrl: _MockSock, data: _MockSock): + mock_thread = MagicMock() + return ( + patch("device_backends.socket.create_connection", side_effect=_mock_connections(ctrl, data)), + patch("device_backends.threading.Thread", return_value=mock_thread), + ) + + +def test_crdop_backend_open_and_attach() -> None: + ctrl = _MockSock() + data = _MockSock() + cfg = DeviceBackendConfig( + device_id="soft-crdop", + backend_type="crdop-tcp", + crdop_host="127.0.0.1", + crdop_port=8515, + crdop_listen=True, + ) + backend = CrdopTcpBackend(cfg, on_rx=lambda _l: None) + p_sock, p_thread = _open_ctx(ctrl, data) + with p_sock, p_thread: + assert backend.open() + assert backend.attach_session("CB-0") + assert backend.status == "ready" + joined = b"".join(ctrl.sent).decode("ascii") + assert "INITIALIZE" in joined + assert "PROTOCOLMODE KISS" in joined + assert "LISTEN TRUE" in joined + backend.close() + assert backend.status == "closed" + + +def test_crdop_backend_transmit() -> None: + ctrl = _MockSock() + data = _MockSock() + cfg = DeviceBackendConfig( + device_id="soft-crdop", + backend_type="crdop-tcp", + crdop_host="127.0.0.1", + crdop_port=8515, + ) + backend = CrdopTcpBackend(cfg, on_rx=lambda _l: None) + p_sock, p_thread = _open_ctx(ctrl, data) + with p_sock, p_thread: + assert backend.open() + assert backend.attach_session("CB-0") + ok, display = backend.transmit("CB-0", "QST", "hello", ax25_ui=True) + assert ok, display + assert len(data.sent) == 1 + assert "FECSEND" not in b"".join(ctrl.sent).decode("ascii") + assert "[CRDOP AX25 UI CB-0>QST]" in display + backend.close() + assert backend.status == "closed" + + +def test_crdop_backend_connect_failure() -> None: + cfg = DeviceBackendConfig( + device_id="soft-crdop", + backend_type="crdop-tcp", + crdop_host="127.0.0.1", + crdop_port=59999, + ) + backend = CrdopTcpBackend(cfg, on_rx=lambda _l: None) + with patch("device_backends.socket.create_connection", side_effect=OSError("refused")): + assert not backend.open() + assert backend.status == "error-connect" + + +def test_crdop_registry_tested() -> None: + from device_backends import registry_tested # noqa: E402 + + assert registry_tested("soft-crdop") is True + + +def main() -> int: + test_crdop_backend_connect_failure() + test_crdop_backend_open_and_attach() + test_crdop_backend_transmit() + test_crdop_registry_tested() + print("OK: crdop backend tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_daemon_log.py b/stacks/daemon/test_daemon_log.py new file mode 100644 index 0000000..d992d79 --- /dev/null +++ b/stacks/daemon/test_daemon_log.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Unit tests for max25d structured logging.""" +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) + +from daemon_log import DaemonLogger, emit_startup_banner # noqa: E402 +from device_backends import DeviceBackendConfig # noqa: E402 + + +def test_structured_levels() -> None: + lines: list[str] = [] + log = DaemonLogger(emit=lines.append) + log.ok("prep complete", area="serial", device="tnc2c") + log.recovery("JHOST 0 — banner", device="pktnc2") + log.warn("no password", area="security") + assert lines[0] == "max25d [OK] [serial] [tnc2c] prep complete" + assert lines[1] == "max25d [RECOVERY] [serial] [pktnc2] JHOST 0 — banner" + assert lines[2] == "max25d [WARN] [security] no password" + + +def test_emit_unstructured_device_prefix() -> None: + lines: list[str] = [] + log = DaemonLogger(emit=lines.append) + log.emit_unstructured("tnc2c: recovery: OK after JHOST 0") + assert "[tnc2c]" in lines[0] + assert "[RECOVERY]" in lines[0] + assert "OK after JHOST 0" in lines[0] + + +def test_startup_banner_sections() -> None: + from dataclasses import dataclass + + @dataclass + class Cfg: + mode: str = "standalone" + device: str = "tnc2c" + default_device: str = "tnc2c" + tcp_host: str = "0.0.0.0" + tcp_port: int = 7325 + unix_socket: str = "/run/max25/modem.sock" + tcp_password: str = "" + callerid: str = "CB-0" + callid: str = "QST" + ax25_ui: bool = True + bans_file: str = "" + auto_start: bool = True + serial_enabled: bool = True + stack_recover_only: bool = True + serial_watch: bool = True + serial_watch_interval: int = 60 + serial_watch_startup_grace: int = 45 + serial_bootwait_escalate: bool = True + hardware: str = "tncs" + + lines: list[str] = [] + from daemon_log import LOGGER + + orig = LOGGER._emit + try: + LOGGER._emit = lines.append + dev = DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + dev.serial_device = "/dev/ttyS4" + dev.serial_baud = 19200 + dev.serial_line = "8n1" + emit_startup_banner( + config_path="/etc/max25/max25d.ini", + cfg=Cfg(), + devices=[dev], + tested_fn=lambda _d: True, + ) + finally: + LOGGER._emit = orig + + text = "\n".join(lines) + assert "=== MAX25d starting ===" in text + assert "[config]" in text + assert "ini=/etc/max25/max25d.ini" in text + assert "[devices]" in text + assert "tnc2c=" in text + + +def main() -> int: + test_structured_levels() + print("OK test_structured_levels") + test_emit_unstructured_device_prefix() + print("OK test_emit_unstructured_device_prefix") + test_startup_banner_sections() + print("OK test_startup_banner_sections") + print("All daemon_log tests passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_kiss_bridge.py b/stacks/daemon/test_kiss_bridge.py new file mode 100644 index 0000000..066d1d0 --- /dev/null +++ b/stacks/daemon/test_kiss_bridge.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Offline unit tests for kiss_bridge (no serial hardware).""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from ax25_codec import ax25_build_ui, ax25_crc, ax25_parse_ui, parse_callsign # noqa: E402 +from kiss_bridge import ( # noqa: E402 + KissDecoder, + format_rx_line, + kiss_data_frame, + kiss_encode, + serial_profile_for_device, +) + + +def test_callsign_parse() -> None: + assert parse_callsign("DG1ABC") == ("DG1ABC", 0) + assert parse_callsign("DK0WC-7") == ("DK0WC", 7) + + +def test_ax25_roundtrip() -> None: + frame = ax25_build_ui("DG1ABC", "CQ", b"73") + parsed = ax25_parse_ui(frame) + assert parsed is not None + src, dst, payload = parsed + assert src == "DG1ABC" + assert dst == "CQ" + assert payload == b"73" + + +def test_kiss_fcs_strip() -> None: + frame = ax25_build_ui("CB-0", "QST", b"hi") + pkt = kiss_data_frame(0, frame) + assert pkt.startswith(b"\xc0") + assert pkt.endswith(b"\xc0") + dec = KissDecoder() + frames = dec.feed(pkt) + assert len(frames) == 1 + _port, payload = frames[0] + assert len(payload) == len(frame) - 2 + parsed = ax25_parse_ui(payload) + assert parsed is not None + src, dst, info = parsed + assert src == "CB" + assert dst == "QST" + assert info == b"hi" + + +def test_kiss_escape_roundtrip() -> None: + raw = bytes([0xC0, 0xDB, 0x00, 0xFF]) + pkt = kiss_encode(0, 0x00, raw) + dec = KissDecoder() + frames = dec.feed(pkt) + assert frames[0][1] == raw + + +def test_format_rx() -> None: + line = format_rx_line("DG1ABC", "QST", b"73", ax25_ui=True) + assert line == "[AX25 UI DG1ABC>QST] 73" + + +def test_serial_profile_tnc2c() -> None: + root = str(Path(__file__).resolve().parents[2]) + prof = serial_profile_for_device("tnc2c", root, {}) + assert prof.baud == 19200 + assert prof.dtr_rts is True + assert prof.kiss_entry == "kiss_on" + + +def test_serial_profile_pktnc2() -> None: + root = str(Path(__file__).resolve().parents[2]) + prof = serial_profile_for_device("pktnc2", root, {}) + assert prof.baud == 9600 + assert prof.dtr_rts is False + assert prof.kiss_entry == "auto" + + +def test_serial_profile_pccom_kiss_env() -> None: + import os + import tempfile + + tmp = Path(tempfile.mkdtemp()) + env = tmp / "local" / "pccom-kiss-serial.env" + env.parent.mkdir(parents=True) + env.write_text( + "PCCOM_KISS_DEV=/dev/ttyUSB2\nPCCOM_KISS_BAUD=9600\nPCCOM_KISS_DTR_RTS=no\n", + encoding="utf-8", + ) + old = os.environ.get("MAX25_ROOT") + os.environ["MAX25_ROOT"] = str(tmp) + try: + prof = serial_profile_for_device("pccom-kiss", str(tmp), {}) + assert prof.device == "/dev/ttyUSB2" + assert prof.baud == 9600 + assert prof.dtr_rts is False + assert prof.kiss_entry == "none" + finally: + if old is None: + os.environ.pop("MAX25_ROOT", None) + else: + os.environ["MAX25_ROOT"] = old + + +def test_crc_known_vector() -> None: + body = bytes(range(256)) + assert ax25_crc(body) == 0x303C + + +def test_kiss_non_data_ignored() -> None: + pkt = kiss_encode(0, 0x06, b"ignored") # TXDELAY cmd + dec = KissDecoder() + assert dec.feed(pkt) == [] + + +def test_invalid_callsign_tx() -> None: + from kiss_bridge import KissBridge, SerialProfile + + bridge = KissBridge(SerialProfile(), on_rx=lambda _m: None) + bridge._kiss_active = True + bridge._fd = -1 # not used — validate before write + ok, msg = bridge.transmit("BAD!", "CQ", "hi", ax25_ui=True) + assert not ok + assert "invalid" in msg.lower() + + +def test_rx_thread_not_started_on_open() -> None: + from kiss_bridge import KissBridge, SerialProfile + + bridge = KissBridge(SerialProfile(), on_rx=lambda _m: None) + bridge._fd = 1 + bridge.status = "open" + assert bridge._thread is None + + +def main() -> int: + tests = [ + test_callsign_parse, + test_ax25_roundtrip, + test_kiss_fcs_strip, + test_kiss_escape_roundtrip, + test_format_rx, + test_serial_profile_tnc2c, + test_serial_profile_pktnc2, + test_serial_profile_pccom_kiss_env, + test_crc_known_vector, + test_kiss_non_data_ignored, + test_invalid_callsign_tx, + test_rx_thread_not_started_on_open, + ] + for fn in tests: + fn() + print("OK: kiss_bridge unit tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_max25_platform.py b/stacks/daemon/test_max25_platform.py new file mode 100644 index 0000000..5073e01 --- /dev/null +++ b/stacks/daemon/test_max25_platform.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +"""Platform helpers for max25d.""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from max25_platform import ( # noqa: E402 + crdop_audio_backend, + default_bans_file, + default_unix_socket, + is_freebsd, + is_linux, + max25d_supported, + supported_device_ids, +) + + +def test_linux_defaults() -> None: + # Run on Linux CI — skip assertions on other hosts + if not is_linux(): + return + assert max25d_supported() + assert "tnc2c" in supported_device_ids() + assert "baycom-kiss" in supported_device_ids() + assert "pccom-kiss" in supported_device_ids() + assert "max25e0" in supported_device_ids() + assert "max25e0" in supported_device_ids() + assert "max25e0:bc1" in supported_device_ids() + assert "baycom-ser12" not in supported_device_ids() + assert "baycom-a" not in supported_device_ids() + assert crdop_audio_backend() == "alsa" + assert default_unix_socket() == "/run/max25/modem.sock" + assert default_bans_file() == "/var/lib/max25/bans.txt" + + +def test_freebsd_profile() -> None: + if not is_freebsd(): + return + assert max25d_supported() + assert "soft-crdop" in supported_device_ids() + assert "tmodem" in supported_device_ids() + assert "max25e0" in supported_device_ids() + assert "tnc2c" not in supported_device_ids() + assert crdop_audio_backend() == "oss" + assert default_unix_socket() == "/var/run/max25/modem.sock" + assert default_bans_file() == "/var/db/max25/bans.txt" + + +if __name__ == "__main__": + test_linux_defaults() + test_freebsd_profile() + print("test_max25_platform: ok") diff --git a/stacks/daemon/test_modular_tcp.py b/stacks/daemon/test_modular_tcp.py new file mode 100644 index 0000000..a0b8551 --- /dev/null +++ b/stacks/daemon/test_modular_tcp.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Offline tests for modular_tcp_server.""" +from __future__ import annotations + +import configparser +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from modular_tcp_server import load_modular_tcp # noqa: E402 + + +def test_load_main_secondaries() -> None: + cp = configparser.ConfigParser() + cp.read_string( + """ +[modular_tcp] +enabled = yes +role = main + +[modular_tcp.secondaries] +a = 127.0.0.1:7326 +b = 127.0.0.1:7327 +""" + ) + cfg = load_modular_tcp(cp) + assert cfg.enabled is True + assert cfg.role == "main" + assert len(cfg.secondaries) == 2 + assert cfg.secondaries[0].port == 7326 + + +if __name__ == "__main__": + test_load_main_secondaries() + print("test_modular_tcp: ok") diff --git a/stacks/daemon/test_multi_device.py b/stacks/daemon/test_multi_device.py new file mode 100644 index 0000000..a46a449 --- /dev/null +++ b/stacks/daemon/test_multi_device.py @@ -0,0 +1,518 @@ +#!/usr/bin/env python3 +"""Multi-device config and M25/1 protocol tests for max25d (no serial hardware).""" +from __future__ import annotations + +import configparser +import importlib.util +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from typing import Callable +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +LAUNCHER = ROOT / "stacks/daemon/max25d" +DAEMON = ROOT / "stacks/daemon/max25d.py" + + +def load_max25d_module(): + from importlib.machinery import SourceFileLoader + + loader = SourceFileLoader("max25d", str(DAEMON)) + spec = importlib.util.spec_from_loader("max25d", loader) + mod = importlib.util.module_from_spec(spec) + sys.modules["max25d"] = mod + loader.exec_module(mod) + return mod + + +class LineReader: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buf = b"" + + def read(self, timeout: float = 5.0) -> str: + self.sock.settimeout(timeout) + while b"\n" not in self.buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + return line.decode("utf-8") + + +def free_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +class FakeBackend: + backend_type = "mock" + + def __init__(self, cfg, root, on_rx, log=None): + self.device_id = cfg.device_id + self._on_rx = on_rx + self.status = "ready" + + def open(self): + return True + + def attach_session(self, mycall): + return True + + def transmit(self, src, dst, text, ax25_ui): + return True, f"[AX25 UI {src}>{dst}] {text}" + + def close(self): + self.status = "closed" + + def detach_session(self): + pass + + +def test_parse_legacy_single_device() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[daemon] +device = tnc2c +hardware = tncs +[serial] +device = /dev/ttyS4 +baud = 19200 +""" + ) + cfg = mod.DaemonConfig(device="tnc2c", hardware="tncs") + devices = mod.parse_devices(cp, cfg) + assert len(devices) == 1 + assert devices[0].device_id == "tnc2c" + assert devices[0].serial_device == "/dev/ttyS4" + assert devices[0].serial_baud == 19200 + assert devices[0].backend_type == "kiss-serial" + + +def test_parse_multi_devices() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[devices] +default = tnc2c +tnc2c = /dev/ttyS4 +pktnc2 = /dev/ttyS5 +dev3 = /dev/ttyUSB0 +dev4 = /dev/ttyUSB1 +dev5 = /dev/ttyUSB2 +[serial.dev3] +baud = 9600 +""" + ) + cfg = mod.DaemonConfig() + devices = mod.parse_devices(cp, cfg) + assert len(devices) == 5 + ids = {d.device_id for d in devices} + assert ids == {"tnc2c", "pktnc2", "dev3", "dev4", "dev5"} + by_id = {d.device_id: d for d in devices} + assert by_id["tnc2c"].serial_device == "/dev/ttyS4" + assert by_id["dev3"].serial_baud == 9600 + assert cfg.default_device == "tnc2c" + + +def test_parse_pccom_kiss_device() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[features] +baycom = yes +pccom = yes +[devices] +default = pccom-kiss +pccom-kiss = /dev/ttyUSB0 +""" + ) + cfg = mod.DaemonConfig() + cfg.feature_baycom = True + cfg.feature_pccom = True + devices = mod.parse_devices(cp, cfg) + by_id = {d.device_id: d for d in devices} + assert by_id["pccom-kiss"].backend_type == "kiss-raw-serial" + assert by_id["pccom-kiss"].serial_device == "/dev/ttyUSB0" + + +def test_parse_heterogeneous_devices() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[devices] +default = tnc2c +tnc2c = /dev/ttyS4 +baycom-ser12 = baycom:a +baycom-kiss = /dev/ttyUSB0 +soft-crdop = crdop:default +[device.baycom-ser12] +kiss_link = /var/run/baycom-pr/kiss +modem = a +baycom_ini = /etc/baycom/baycom-pr.ini +[device.soft-crdop] +port = 8515 +""" + ) + cfg = mod.DaemonConfig() + devices = mod.parse_devices(cp, cfg) + by_id = {d.device_id: d for d in devices} + assert by_id["tnc2c"].backend_type == "kiss-serial" + assert by_id["baycom-ser12"].backend_type == "baycom-kiss" + assert by_id["baycom-ser12"].kiss_link == "/var/run/baycom-pr/kiss" + assert by_id["baycom-ser12"].baycom_ini == "/etc/baycom/baycom-pr.ini" + assert by_id["baycom-kiss"].backend_type == "kiss-raw-serial" + assert by_id["soft-crdop"].backend_type == "crdop-tcp" + assert by_id["soft-crdop"].crdop_port == 8515 + + +def test_parse_enabled_filter() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[devices] +enabled = tnc2c,pktnc2 +tnc2c = /dev/ttyS4 +pktnc2 = /dev/ttyS5 +extra = /dev/ttyUSB9 +""" + ) + cfg = mod.DaemonConfig() + devices = mod.parse_devices(cp, cfg) + enabled = {d.device_id for d in devices if d.enabled} + disabled = {d.device_id for d in devices if not d.enabled} + assert enabled == {"tnc2c", "pktnc2"} + assert disabled == {"extra"} + + +def test_backend_open_retry_on_error() -> None: + """CONNECT retries KISS open after a transient error-no-device.""" + mod = load_max25d_module() + cfg = mod.DaemonConfig(hardware="modems", serial_enabled=True) + cfg.devices = [ + mod.DeviceBackendConfig( + device_id="baycom-ser12", + backend_type="baycom-kiss", + kiss_link="/nonexistent/kiss", + enabled=True, + ) + ] + cfg.default_device = "baycom-ser12" + state = mod.DaemonState(cfg=cfg) + mod.init_device_runtimes(state) + + class ErrorThenReady(FakeBackend): + attempts = 0 + + def open(self): + ErrorThenReady.attempts += 1 + if ErrorThenReady.attempts == 1: + self.status = "error-no-device" + return False + self.status = "ready" + return True + + ErrorThenReady.attempts = 0 + with patch.object( + mod, + "create_backend", + side_effect=lambda c, r, rx, log, prefix=None, on_invalid=None: ErrorThenReady(c, r, rx, log), + ): + assert not mod.attach_backend_session(state, "baycom-ser12") + assert mod.attach_backend_session(state, "baycom-ser12") + assert ErrorThenReady.attempts == 2 + + +def test_five_mock_backends() -> None: + """Five concurrent backend instances via mocked create_backend.""" + mod = load_max25d_module() + opened: list[str] = [] + + class CountingBackend(FakeBackend): + def open(self): + opened.append(self.device_id) + return True + + cfg = mod.DaemonConfig(hardware="tncs", serial_enabled=True) + cfg.devices = [ + mod.DeviceBackendConfig(device_id=f"d{i}", serial_device=f"/dev/fake{i - 1}", enabled=True) + for i in range(1, 6) + ] + for d in cfg.devices: + d.backend_type = "kiss-serial" + cfg.default_device = "d1" + state = mod.DaemonState(cfg=cfg) + mod.init_device_runtimes(state) + + with patch.object( + mod, + "create_backend", + side_effect=lambda c, r, rx, log, prefix=None, on_invalid=None: CountingBackend(c, r, rx, log), + ): + assert mod.attach_all_sessions(state) + assert len(opened) == 5 + assert set(opened) == {f"d{i}" for i in range(1, 6)} + + state.selected_device = "d3" + dev_id = mod.resolve_selected_device(state) + assert dev_id == "d3" + rt = state.devices["d3"] + ok, display = rt.backend.transmit("CB-0", "QST", "hi", True) + assert ok + assert "hi" in display + + +def run_daemon_test(ini_text: str, fn: Callable[[LineReader, socket.socket], None]) -> None: + port = free_port() + with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as tmp: + tmp.write(ini_text) + ini_path = Path(tmp.name) + + proc = subprocess.Popen( + [ + str(LAUNCHER), + "--no-stack", + "--no-serial", + "-c", + str(ini_path), + "--tcp-port", + str(port), + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + time.sleep(0.6) + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + except OSError as exc: + proc.terminate() + proc.wait(timeout=3) + ini_path.unlink(missing_ok=True) + raise RuntimeError(f"connect failed: {exc}") from exc + + reader = LineReader(sock) + try: + assert reader.read() == "OK" + fn(reader, sock) + finally: + sock.close() + proc.terminate() + proc.wait(timeout=5) + ini_path.unlink(missing_ok=True) + + +def test_parse_dual_baycom_devices() -> None: + mod = load_max25d_module() + cp = configparser.ConfigParser() + cp.read_string( + """ +[devices] +default = baycom-a +baycom-a = baycom:a +baycom-b = baycom:b +[device.baycom-a] +kiss_link = /var/run/baycom-pr/kiss-a +modem = a +baycom_ini = /etc/baycom/baycom-pr.ini +[device.baycom-b] +kiss_link = /var/run/baycom-pr/kiss-b +modem = b +baycom_ini = /etc/baycom/baycom-pr.ini +""" + ) + cfg = mod.DaemonConfig() + devices = mod.parse_devices(cp, cfg) + by_id = {d.device_id: d for d in devices} + assert set(by_id) == {"baycom-a", "baycom-b"} + assert by_id["baycom-a"].backend_type == "baycom-kiss" + assert by_id["baycom-b"].backend_type == "baycom-kiss" + assert by_id["baycom-a"].baycom_modem == "a" + assert by_id["baycom-b"].baycom_modem == "b" + assert by_id["baycom-a"].kiss_link == "/var/run/baycom-pr/kiss-a" + assert by_id["baycom-b"].kiss_link == "/var/run/baycom-pr/kiss-b" + assert cfg.default_device == "baycom-a" + + +def test_protocol_dual_baycom() -> None: + # Legacy ids baycom-a/baycom-b are not on the Linux allowlist (product path: max25e0). + # Protocol dual-device coverage uses allowlisted modem-class ids instead. + ini = """ +[features] +baycom = yes +pccom = yes +[daemon] +mode = service +hardware = modems +device = baycom-kiss +[devices] +default = baycom-kiss +baycom-kiss = /dev/null +pccom-kiss = /dev/null +[network] +tcp_host = 127.0.0.1 +tcp_port = 7325 +[modem] +callerid = N0CALL-0 +callid = QST +[stack] +auto_start = no +""" + + def exercise(reader: LineReader, sock: socket.socket) -> None: + status = reader.read() + assert "devices=baycom-kiss,pccom-kiss" in status + assert "device=baycom-kiss" in status + + sock.sendall(b"GET DEVICES\n") + lines = [] + while True: + line = reader.read() + lines.append(line) + if line == "OK": + break + device_lines = [l for l in lines if l.startswith("DEVICE ")] + assert len(device_lines) == 2 + assert any("id=baycom-kiss" in l for l in device_lines) + assert any("id=pccom-kiss" in l for l in device_lines) + + sock.sendall(b"SET DEVICE pccom-kiss\n") + assert reader.read() == "OK" + + sock.sendall(b"GET STATUS\n") + st = reader.read() + assert "device=pccom-kiss" in st + assert reader.read() == "OK" + + run_daemon_test(ini, exercise) + + +def test_protocol_multi_device() -> None: + # Explicit two-device INI — share/max25d.ini.example is single-device by design. + ini = """ +[daemon] +mode = standalone +hardware = tncs +device = tnc2c +[devices] +default = tnc2c +tnc2c = /dev/ttyS4 +pktnc2 = /dev/ttyS5 +[network] +tcp_host = 127.0.0.1 +tcp_port = 7325 +[modem] +callerid = CB-0 +callid = QST +[stack] +auto_start = no +""" + + def exercise(reader: LineReader, sock: socket.socket) -> None: + status = reader.read() + assert status.startswith("STATUS ") + assert "devices=tnc2c,pktnc2" in status + assert "device=tnc2c" in status + + sock.sendall(b"GET DEVICES\n") + lines = [] + while True: + line = reader.read() + lines.append(line) + if line == "OK": + break + device_lines = [l for l in lines if l.startswith("DEVICE ")] + assert len(device_lines) == 2 + assert any("id=tnc2c" in l for l in device_lines) + assert any("id=pktnc2" in l for l in device_lines) + assert any("backend=kiss-serial" in l for l in device_lines) + + sock.sendall(b"SET DEVICE pktnc2\n") + assert reader.read() == "OK" + + sock.sendall(b"GET STATUS\n") + st = reader.read() + assert "device=pktnc2" in st + assert reader.read() == "OK" + + sock.sendall(b"SELECT DEVICE tnc2c\n") + assert reader.read() == "OK" + + sock.sendall(b"CONNECT\n") + assert reader.read() == "EVENT connected" + assert reader.read() == "OK" + + sock.sendall(b"SEND 73\n") + rx = reader.read() + assert rx.startswith("RX ") + assert reader.read() == "OK" + + sock.sendall(b"DISCONNECT\n") + assert reader.read() == "EVENT disconnected" + assert reader.read() == "OK" + + run_daemon_test(ini, exercise) + + +def test_protocol_legacy_single() -> None: + ini = """ +[daemon] +mode = standalone +hardware = tncs +device = tnc2c +[network] +tcp_host = 127.0.0.1 +tcp_port = 7325 +[modem] +callerid = CB-0 +callid = QST +[stack] +auto_start = no +""" + + def exercise(reader: LineReader, sock: socket.socket) -> None: + status = reader.read() + assert "device=tnc2c" in status + assert "devices=tnc2c" in status + + sock.sendall(b"SET DEVICE tnc2c\n") + assert reader.read() == "OK" + + run_daemon_test(ini, exercise) + + +def main() -> int: + tests = [ + test_parse_legacy_single_device, + test_parse_multi_devices, + test_parse_heterogeneous_devices, + test_parse_dual_baycom_devices, + test_parse_enabled_filter, + test_backend_open_retry_on_error, + test_five_mock_backends, + test_protocol_multi_device, + test_protocol_dual_baycom, + test_protocol_legacy_single, + ] + for fn in tests: + fn() + print("OK: multi-device tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_path_normalize.py b/stacks/daemon/test_path_normalize.py new file mode 100644 index 0000000..c7b24e6 --- /dev/null +++ b/stacks/daemon/test_path_normalize.py @@ -0,0 +1,17 @@ +"""Tests for legacy bcpr path normalization.""" + +from paths import MAX25_BCPR_KISS_DEFAULT, normalize_max25_bcpr_path + + +def test_normalize_legacy_tmp_bcpr(): + assert normalize_max25_bcpr_path("/tmp/bcpr/kiss-bc0") == "/tmp/max25-bcpr/kiss-bc0" + assert normalize_max25_bcpr_path("/tmp/bcpr") == "/tmp/max25-bcpr" + assert normalize_max25_bcpr_path("") == "" + + +def test_canonical_unchanged(): + assert ( + normalize_max25_bcpr_path("/tmp/max25-bcpr/kiss-bc0") + == "/tmp/max25-bcpr/kiss-bc0" + ) + assert MAX25_BCPR_KISS_DEFAULT == "/tmp/max25-bcpr/kiss-bc0" diff --git a/stacks/daemon/test_paths.py b/stacks/daemon/test_paths.py new file mode 100644 index 0000000..12b4be2 --- /dev/null +++ b/stacks/daemon/test_paths.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Offline tests for install/dev path resolution.""" +from __future__ import annotations + +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from paths import ( # noqa: E402 + baycom_ini_candidates, + canonical_dual_baycom_example, + canonical_single_baycom_example, + ctl_path, + default_ini_candidates, + is_dual_baycom_ini, + resolve_baycom_ini, + resolve_baycom_profile, + resolve_layout, + serial_env_candidates, + share_max25_dir, + stacks_dir, +) + +REPO = Path(__file__).resolve().parents[2] + + +def test_dev_layout() -> None: + exe = REPO / "stacks/daemon/max25d" + tree, prefix = resolve_layout(exe) + assert prefix is None + assert tree == REPO + assert (stacks_dir(tree, prefix) / "tncs").is_dir() + assert ctl_path(tree, prefix, exe) == REPO / "scripts/max25-ctl" + + +def test_installed_layout() -> None: + with tempfile.TemporaryDirectory() as tmp: + prefix = Path(tmp) / "usr/local" + bindir = prefix / "bin" + share = prefix / "share/max25" + bindir.mkdir(parents=True) + share.mkdir(parents=True) + (share / "max25d.ini.example").write_text("; test\n", encoding="utf-8") + (bindir / "max25d").write_text("#!/bin/sh\n", encoding="utf-8") + (bindir / "max25-ctl").write_text("#!/bin/sh\n", encoding="utf-8") + + tree, detected = resolve_layout(bindir / "max25d") + assert detected == prefix + assert tree == prefix + + cands = default_ini_candidates(tree, detected) + assert share / "max25d.ini.example" in cands + assert ctl_path(tree, detected, bindir / "max25d") == bindir / "max25-ctl" + + +def test_max25_root_override() -> None: + with tempfile.TemporaryDirectory() as tmp: + prefix = Path(tmp) / "opt/max25" + bindir = prefix / "bin" + share = prefix / "share/max25" + bindir.mkdir(parents=True) + share.mkdir(parents=True) + (bindir / "max25d").write_text("#!/bin/sh\n", encoding="utf-8") + + checkout = Path(tmp) / "checkout" + (checkout / "stacks/tncs").mkdir(parents=True) + (checkout / "plugins").mkdir() + (checkout / "plugins/manifest.yaml").write_text("plugins: []\n", encoding="utf-8") + + old = os.environ.get("MAX25_ROOT") + os.environ["MAX25_ROOT"] = str(checkout) + try: + tree, detected = resolve_layout(bindir / "max25d") + assert detected == prefix + assert tree == checkout + assert stacks_dir(tree, detected) == checkout / "stacks" + finally: + if old is None: + os.environ.pop("MAX25_ROOT", None) + else: + os.environ["MAX25_ROOT"] = old + + +def test_serial_env_order() -> None: + with tempfile.TemporaryDirectory() as tmp: + prefix = Path(tmp) / "usr/local" + etc = Path(tmp) / "etc/max25" + etc.mkdir(parents=True) + (etc / "tnc2c-serial.env").write_text("TNC2C_BAUD=38400\n", encoding="utf-8") + share = prefix / "share/max25/serial" + share.mkdir(parents=True) + (share / "tnc2c-serial.env").write_text("TNC2C_BAUD=19200\n", encoding="utf-8") + + # /etc/max25 is checked first when present on disk + cands = serial_env_candidates("tnc2c", REPO, prefix) + assert cands[0] == Path("/etc/max25/tnc2c-serial.env") + assert prefix / "share/max25/serial/tnc2c-serial.env" in cands + + +def test_baycom_ini_resolution_legacy_removed() -> None: + """Kernel baycom-pr examples removed 2026-07-18 — no in-tree templates.""" + example = REPO / "share/baycom/baycom-pr.pccom-ttyS0-only.ini.example" + dual = REPO / "stacks/baycom-pr/config/examples/baycom-pr.dual.ini" + assert not example.is_file() + assert not dual.is_file() + assert canonical_single_baycom_example("baycom-ser12", REPO, None) is None + assert canonical_dual_baycom_example("baycom-ser12", REPO, None) is None + with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as tmp: + tmp.write("[stack]\n") + explicit = Path(tmp.name) + try: + assert resolve_baycom_ini("baycom-ser12", REPO, None, str(explicit)) == explicit + # Without explicit: may still resolve site /etc/baycom/baycom-pr.ini if present + resolved = resolve_baycom_ini("baycom-ser12", REPO, None) + if resolved is not None: + assert resolved.is_file() + assert "baycom-pr" in resolved.name or resolved.parent.name == "baycom" + finally: + explicit.unlink(missing_ok=True) + + +def test_baycom_profile_dual_legacy_removed() -> None: + assert resolve_baycom_profile("dual", "baycom-ser12", REPO, None) is None + + +def main() -> int: + tests = [ + test_dev_layout, + test_installed_layout, + test_max25_root_override, + test_serial_env_order, + test_baycom_ini_resolution_legacy_removed, + test_baycom_profile_dual_legacy_removed, + ] + for fn in tests: + fn() + print("OK: paths unit tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_privilege_drop.py b/stacks/daemon/test_privilege_drop.py new file mode 100644 index 0000000..3a611c9 --- /dev/null +++ b/stacks/daemon/test_privilege_drop.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +"""Tests for max25d privilege_drop (no root required).""" +from __future__ import annotations + +import configparser +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +sys.path.insert(0, str(ROOT)) + +from privilege_drop import RunAsConfig, parse_run_as, resolve_run_as, run_as_configured # noqa: E402 + + +def test_parse_run_as_empty() -> None: + cp = configparser.ConfigParser() + cp.read_string("[daemon]\nmode = standalone\n") + cfg = parse_run_as(cp) + assert cfg.user == "" + assert not run_as_configured(cfg) + + +def test_parse_run_as_user_group() -> None: + cp = configparser.ConfigParser() + cp.read_string( + "[daemon]\nuser = max25\ngroup = dialout\nuid = 1001\n" + ) + cfg = parse_run_as(cp) + assert cfg.user == "max25" + assert cfg.group == "dialout" + assert cfg.uid == 1001 + assert run_as_configured(cfg) + + +def test_resolve_run_as_current_user() -> None: + import os + import pwd + + pw = pwd.getpwuid(os.getuid()) + uid, gid, name = resolve_run_as(RunAsConfig(user=pw.pw_name)) + assert uid == pw.pw_uid + assert gid == pw.pw_gid + assert name == pw.pw_name + + +def main() -> int: + test_parse_run_as_empty() + test_parse_run_as_user_group() + test_resolve_run_as_current_user() + print("privilege_drop: OK") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/stacks/daemon/test_proto.py b/stacks/daemon/test_proto.py new file mode 100644 index 0000000..50ced93 --- /dev/null +++ b/stacks/daemon/test_proto.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""Offline smoke test for max25d M25/1 protocol.""" +import socket +import subprocess +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +DAEMON = ROOT / "stacks/daemon/max25d" +INI = ROOT / "share/max25/max25d.ini.example" + + +class LineReader: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buf = b"" + + def read(self, timeout: float = 5.0) -> str: + self.sock.settimeout(timeout) + while b"\n" not in self.buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + return line.decode("utf-8") + + +def free_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def wait_for_port(port: int, proc: subprocess.Popen[str], timeout: float = 8.0) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if proc.poll() is not None: + return False + try: + probe = socket.create_connection(("127.0.0.1", port), timeout=0.25) + probe.close() + return True + except OSError: + time.sleep(0.1) + return False + + +def main() -> int: + port = free_port() + proc = subprocess.Popen( + [str(DAEMON), "--no-stack", "--no-serial", "-c", str(INI), "--tcp-port", str(port)], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + if not wait_for_port(port, proc): + err = proc.stderr.read() if proc.stderr else "" + proc.terminate() + proc.wait(timeout=3) + print(f"FAIL: max25d not listening on {port}: {err.strip()}", file=sys.stderr) + return 1 + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + except OSError as exc: + proc.terminate() + proc.wait(timeout=3) + print(f"FAIL: connect: {exc}", file=sys.stderr) + return 1 + + reader = LineReader(sock) + try: + assert reader.read() == "OK" + status = reader.read() + assert status.startswith("STATUS ") + + sock.sendall(b"SET CALLERID DG1ABC\n") + assert reader.read() == "OK" + + sock.sendall(b"CONNECT\n") + assert reader.read() == "EVENT connected" + assert reader.read() == "OK" + + sock.sendall(b"SEND 73\n") + rx = reader.read() + assert rx.startswith("RX ") + assert reader.read() == "OK" + + sock.sendall(b"GET STATUS\n") + st = reader.read() + assert "callerid=DG1ABC" in st + assert reader.read() == "OK" + + print("OK: max25d protocol smoke") + return 0 + finally: + sock.close() + proc.terminate() + proc.wait(timeout=5) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_reporting.py b/stacks/daemon/test_reporting.py new file mode 100644 index 0000000..54f73df --- /dev/null +++ b/stacks/daemon/test_reporting.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""INI reporting switches — error= / voice= in STATUS.""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from device_backends import DeviceBackendConfig # noqa: E402 +import max25d # noqa: E402 +from reporting_quality import DataQualityTracker, RxOutcome # noqa: E402 + +T0 = 2_000_000.0 +WIN = 20 + + +def _close_pass(rt: max25d.DeviceRuntime, index: int, start: float = T0) -> None: + rt.quality.tick(start + (index + 1) * WIN + 1) + + +def _fill_good(rt: max25d.DeviceRuntime, count: int = 3, window: int = WIN) -> None: + for i in range(count): + rt.quality.record(RxOutcome.GOOD, T0 + i * window + 1) + rt.quality.tick(T0 + (i + 1) * window + 1) + + +def _state( + *, + error_on: bool = True, + voice_on: bool = True, + link: str = "ready", + passes: int = 3, + quality_min: int = 50, + pass_seconds: int = 20, + backend_type: str = "kiss-serial", + hardware: str = "tncs", + prefill_good: bool = True, +) -> max25d.DaemonState: + cfg = max25d.DaemonConfig( + report_error_transmissions=error_on, + report_voice_transmissions=voice_on, + report_data_passes=passes, + report_data_quality_min=quality_min, + report_data_pass_seconds=pass_seconds, + ) + dev = DeviceBackendConfig( + device_id="tnc2c", + enabled=True, + backend_type=backend_type, + hardware=hardware, + ) + rt = max25d.DeviceRuntime( + cfg=dev, + link_status=link, + quality=DataQualityTracker( + passes_required=passes, + min_good_percent=quality_min, + pass_window_sec=pass_seconds, + ), + ) + if prefill_good: + _fill_good(rt, passes, pass_seconds) + state = max25d.DaemonState(cfg=cfg) + state.devices["tnc2c"] = rt + state.selected_device = "tnc2c" + return state + + +def test_error_valid_when_enabled_and_ready() -> None: + line = max25d.status_line(_state()) + assert "error=valid" in line + + +def test_error_invalid_when_disabled() -> None: + line = max25d.status_line(_state(error_on=False)) + assert "error=invalid" in line + + +def test_error_invalid_on_link_fault() -> None: + line = max25d.status_line(_state(link="error-io")) + assert "error=invalid" in line + + +def test_error_invalid_before_passes_complete() -> None: + line = max25d.status_line(_state(prefill_good=False)) + assert "error=invalid" in line + + +def test_error_invalid_on_bad_frame_ratio() -> None: + state = _state(prefill_good=False) + rt = state.devices["tnc2c"] + rt.quality.record(RxOutcome.BAD, T0 + 1) + _close_pass(rt, 0) + rt.quality.record(RxOutcome.BAD, T0 + 21) + _close_pass(rt, 1) + rt.quality.record(RxOutcome.GOOD, T0 + 41) + _close_pass(rt, 2) + line = max25d.status_line(state) + assert "error=invalid" in line + + +def test_voice_valid_tnc_only() -> None: + line = max25d.status_line(_state()) + assert "voice=valid" in line + + +def test_voice_invalid_when_disabled() -> None: + line = max25d.status_line(_state(voice_on=False)) + assert "voice=invalid" in line + + +def test_voice_invalid_crdop_not_ready() -> None: + line = max25d.status_line( + _state( + backend_type="crdop-tcp", + hardware="soft-modems", + link="error-connect", + ) + ) + assert "voice=invalid" in line + + +def test_device_line_includes_reporting() -> None: + state = _state() + dev = max25d.device_line(state, "tnc2c") + assert "error=valid" in dev + assert "voice=n/a" in dev + + +def test_on_backend_rx_tracks_callid_in_pass_window() -> None: + state = _state(prefill_good=False) + max25d.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>QST] 73") + max25d.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>CB-0] bad") + max25d.on_backend_rx(state, "tnc2c", "[CRDOP RX soft-crdop] noise") + rt = state.devices["tnc2c"] + start = rt.quality.current.started_at + assert rt.quality.current.good == 1 + assert rt.quality.current.bad == 1 + rt.quality.tick(start + WIN + 1) + assert list(rt.quality.passes) == ["bad"] + max25d.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>QST] ok") + start = rt.quality.current.started_at + rt.quality.tick(start + WIN + 1) + max25d.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>QST] ok2") + start = rt.quality.current.started_at + rt.quality.tick(start + WIN + 1) + line = max25d.status_line(state) + assert "error=valid" in line + + +if __name__ == "__main__": + test_error_valid_when_enabled_and_ready() + test_error_invalid_when_disabled() + test_error_invalid_on_link_fault() + test_error_invalid_before_passes_complete() + test_error_invalid_on_bad_frame_ratio() + test_voice_valid_tnc_only() + test_voice_invalid_when_disabled() + test_voice_invalid_crdop_not_ready() + test_device_line_includes_reporting() + test_on_backend_rx_tracks_callid_in_pass_window() + print("test_reporting: ok") diff --git a/stacks/daemon/test_reporting_quality.py b/stacks/daemon/test_reporting_quality.py new file mode 100644 index 0000000..15f0d4d --- /dev/null +++ b/stacks/daemon/test_reporting_quality.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Data-quality window — CALLID match, timed passes, configurable min % good.""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from reporting_quality import ( # noqa: E402 + DataQualityTracker, + RxOutcome, + classify_rx_line, +) + +T0 = 1_000_000.0 +WIN = 20 + + +def test_classify_good_callid() -> None: + assert classify_rx_line("[AX25 UI DG1ABC>QST] 73", "QST") == RxOutcome.GOOD + + +def test_classify_bad_callid() -> None: + assert classify_rx_line("[AX25 UI DG1ABC>CB-0] 73", "QST") == RxOutcome.BAD + + +def test_classify_incomplete_ax25() -> None: + assert classify_rx_line("[AX25 UI DG1ABC>QST", "QST") == RxOutcome.BAD + + +def test_classify_voice_noise_ignored() -> None: + assert classify_rx_line("[CRDOP RX soft-crdop] hiss", "QST") == RxOutcome.IGNORE + + +def test_classify_crdop_ax25_good() -> None: + line = "[CRDOP AX25 UI CB-0>QST] ping" + assert classify_rx_line(line, "QST") == RxOutcome.GOOD + + +def _close_pass(q: DataQualityTracker, index: int, start: float = T0) -> None: + q.tick(start + (index + 1) * WIN + 1) + + +def test_pass_window_max_twenty_seconds() -> None: + q = DataQualityTracker(passes_required=3, min_good_percent=50, pass_window_sec=20) + q.record(RxOutcome.GOOD, T0 + 1) + q.tick(T0 + 19) + assert len(q.passes) == 0 + _close_pass(q, 0) + assert list(q.passes) == ["good"] + + +def test_three_passes_fifty_percent() -> None: + q = DataQualityTracker(passes_required=3, min_good_percent=50, pass_window_sec=20) + q.record(RxOutcome.GOOD, T0 + 1) + _close_pass(q, 0) + q.record(RxOutcome.BAD, T0 + 20 + 1) + _close_pass(q, 1) + assert q.data_error_valid(reporting_enabled=True) is False + q.record(RxOutcome.GOOD, T0 + 40 + 1) + _close_pass(q, 2) + assert q.data_error_valid(reporting_enabled=True) is True + + +def test_three_passes_below_fifty_percent() -> None: + q = DataQualityTracker(passes_required=3, min_good_percent=50, pass_window_sec=20) + q.record(RxOutcome.BAD, T0 + 1) + _close_pass(q, 0) + q.record(RxOutcome.BAD, T0 + 21) + _close_pass(q, 1) + q.record(RxOutcome.GOOD, T0 + 41) + _close_pass(q, 2) + assert q.data_error_valid(reporting_enabled=True) is False + + +def test_ignore_does_not_consume_pass() -> None: + q = DataQualityTracker(passes_required=3, min_good_percent=50, pass_window_sec=20) + q.record(RxOutcome.IGNORE, T0 + 1) + _close_pass(q, 0) + assert len(q.passes) == 0 + q.record(RxOutcome.GOOD, T0 + 21) + _close_pass(q, 1) + q.record(RxOutcome.GOOD, T0 + 41) + _close_pass(q, 2) + q.record(RxOutcome.GOOD, T0 + 61) + _close_pass(q, 3) + assert q.data_error_valid(reporting_enabled=True) is True + assert q.voice_activity is True + + +def test_silent_window_counts_as_bad() -> None: + q = DataQualityTracker(passes_required=1, min_good_percent=50, pass_window_sec=20) + q.current.started_at = T0 + _close_pass(q, 0) + assert list(q.passes) == ["bad"] + + +def test_max_cycle_seconds() -> None: + q = DataQualityTracker(passes_required=3, pass_window_sec=20) + assert q.max_cycle_seconds() == 60 + + +def test_voice_valid_on_activity() -> None: + q = DataQualityTracker() + q.record(RxOutcome.IGNORE, T0 + 1) + assert q.voice_signal_valid(reporting_enabled=True, link_healthy=True) is True + + +if __name__ == "__main__": + test_classify_good_callid() + test_classify_bad_callid() + test_classify_incomplete_ax25() + test_classify_voice_noise_ignored() + test_classify_crdop_ax25_good() + test_pass_window_max_twenty_seconds() + test_three_passes_fifty_percent() + test_three_passes_below_fifty_percent() + test_ignore_does_not_consume_pass() + test_silent_window_counts_as_bad() + test_max_cycle_seconds() + test_voice_valid_on_activity() + print("test_reporting_quality: ok") diff --git a/stacks/daemon/test_serial_watch.py b/stacks/daemon/test_serial_watch.py new file mode 100644 index 0000000..7b4d8ad --- /dev/null +++ b/stacks/daemon/test_serial_watch.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""Offline tests for max25d TNC serial watch / stabilize.""" +from __future__ import annotations + +import configparser +import importlib.util +import sys +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +ROOT = Path(__file__).resolve().parents[2] +DAEMON = ROOT / "stacks/daemon/max25d.py" + + +def load_max25d(): + from importlib.machinery import SourceFileLoader + + loader = SourceFileLoader("max25d", str(DAEMON)) + spec = importlib.util.spec_from_loader("max25d", loader) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + sys.modules["max25d"] = mod + loader.exec_module(mod) + return mod + + +def test_stack_serial_watch_config() -> None: + mod = load_max25d() + cp = configparser.ConfigParser() + cp.read_string( + """ +[stack] +auto_start = yes +serial_watch = no +serial_watch_interval = 45 +serial_repair_cooldown = 10 +stack_recover_only = no +stack_retry_interval = 90 +""" + ) + cfg = mod.DaemonConfig() + cfg.auto_start = mod._truthy(cp.get("stack", "auto_start")) + cfg.serial_watch = mod._truthy(cp.get("stack", "serial_watch")) + cfg.serial_watch_interval = cp.getint("stack", "serial_watch_interval") + cfg.serial_repair_cooldown = cp.getint("stack", "serial_repair_cooldown") + cfg.stack_recover_only = mod._truthy(cp.get("stack", "stack_recover_only")) + cfg.stack_retry_interval = cp.getint("stack", "stack_retry_interval") + assert cfg.serial_watch is False + assert cfg.serial_watch_interval == 45 + assert cfg.stack_recover_only is False + + +def test_serial_repair_statuses() -> None: + mod = load_max25d() + assert "error-io" in mod.BACKEND_RETRY_STATUSES + assert "error-host" in mod.SERIAL_REPAIR_STATUSES + assert "open" not in mod.SERIAL_REPAIR_STATUSES + + +def test_inline_tnc_prep() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(stack_recover_only=True) + state = mod.DaemonState(cfg=cfg) + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + state.devices["baycom-ser12"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="baycom-ser12", backend_type="baycom-kiss") + ) + assert mod.uses_inline_tnc_prep(state, "tnc2c") + assert not mod.uses_inline_tnc_prep(state, "baycom-ser12") + + +def test_stabilize_ready_kiss_skips_probe() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + bridge.status = "ready" + + fake = MagicMock() + + with patch.object(bridge, "_stop_rx_thread") as stop_rx, patch.object( + bridge, "_load_recovery_mod", return_value=fake + ), patch.object(bridge, "_start_rx_thread") as start_rx: + ok = bridge.stabilize_session("CB-0", force=False) + + assert ok + assert bridge.status == "ready" + fake.probe_info.assert_not_called() + fake.recover_terminal.assert_not_called() + stop_rx.assert_not_called() + start_rx.assert_not_called() + + +def test_poll_serial_stability_skips_ready() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(serial_watch=True, stack_recover_only=True) + state = mod.DaemonState(cfg=cfg) + state.started_at = 0 + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + rt = state.devices["tnc2c"] + rt.stack_status = "ready" + rt.last_watch = 0 + backend = MagicMock() + backend.status = "ready" + rt.backend = backend + + mod.poll_serial_stability(state) + + backend.stabilize_session.assert_not_called() + + +def test_stabilize_session_probe_path() -> None: + sys.path.insert(0, str(ROOT / "stacks/daemon")) + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + + fake = MagicMock() + fake.probe_info.return_value = (True, b"TheFirmware cmd:", False) + fake.recover_terminal.return_value = (True, b"") + + with patch.object(bridge, "_write_unlocked"), patch.object( + bridge, "_drain_unlocked", return_value=b"" + ), patch.object(bridge, "_set_mycall_unlocked", return_value=True), patch.object( + bridge, "_enter_kiss_unlocked", return_value=True + ), patch.object(bridge, "_load_recovery_mod", return_value=fake), patch.object( + bridge, "_start_rx_thread" + ): + ok = bridge.stabilize_session("CB-0") + assert ok + assert bridge.status == "ready" + fake.recover_terminal.assert_not_called() + + +def test_rx_thread_deferred_until_ready() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + assert bridge._thread is None + bridge._start_rx_thread() + assert bridge._thread is not None + bridge._stop_rx_thread() + assert bridge._thread is None + + +def test_stabilize_stops_rx_during_recovery() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + bridge._start_rx_thread() + assert bridge._thread is not None + + fake = MagicMock() + fake.probe_info.return_value = (True, b"TheFirmware cmd:", False) + fake.recover_terminal.return_value = (True, b"") + + with patch.object(bridge, "_write_unlocked"), patch.object( + bridge, "_drain_unlocked", return_value=b"" + ), patch.object(bridge, "_set_mycall_unlocked", return_value=True), patch.object( + bridge, "_enter_kiss_unlocked", return_value=True + ), patch.object(bridge, "_load_recovery_mod", return_value=fake), patch.object( + bridge, "_start_rx_thread" + ) as start_rx: + ok = bridge.stabilize_session("CB-0") + assert ok + start_rx.assert_called_once() + + +def test_bootwait_escalate_config() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig() + assert cfg.serial_bootwait_escalate is True + assert cfg.serial_bootwait_escalate_after == 3 + assert cfg.serial_bootwait_escalate_cooldown == 300 + + +def test_prep_escalates_on_error_host() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(stack_recover_only=True, serial_bootwait_escalate=True) + state = mod.DaemonState(cfg=cfg) + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + backend = MagicMock() + backend.status = "error-host" + backend.stabilize_session.return_value = False + rt = state.devices["tnc2c"] + rt.backend = backend + + with patch.object(mod, "open_backend", return_value=True), patch.object( + mod, "escalate_to_bootwait_stack" + ) as escalate: + mod.prep_inline_serial_device(state, "tnc2c") + escalate.assert_called_once_with(state, "tnc2c") + + +def main() -> int: + test_stack_serial_watch_config() + print("OK test_stack_serial_watch_config") + test_serial_repair_statuses() + print("OK test_serial_repair_statuses") + test_inline_tnc_prep() + print("OK test_inline_tnc_prep") + test_stabilize_ready_kiss_skips_probe() + print("OK test_stabilize_ready_kiss_skips_probe") + test_poll_serial_stability_skips_ready() + print("OK test_poll_serial_stability_skips_ready") + test_stabilize_session_probe_path() + print("OK test_stabilize_session_probe_path") + test_rx_thread_deferred_until_ready() + print("OK test_rx_thread_deferred_until_ready") + test_stabilize_stops_rx_during_recovery() + print("OK test_stabilize_stops_rx_during_recovery") + test_bootwait_escalate_config() + print("OK test_bootwait_escalate_config") + test_prep_escalates_on_error_host() + print("OK test_prep_escalates_on_error_host") + print("All serial watch tests passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stacks/daemon/test_tx_rx_offline.py b/stacks/daemon/test_tx_rx_offline.py new file mode 100644 index 0000000..ea580bf --- /dev/null +++ b/stacks/daemon/test_tx_rx_offline.py @@ -0,0 +1,184 @@ +#!/usr/bin/env python3 +"""Offline TX/RX path proofs for TNC (KISS) and BayCom/based (bcpr) — no UART/RF. + +L0 gate for scripts/tx-rx-test.sh and CI. Proves: + TX — AX.25/KISS frame build + max25d CONNECT/SEND host path (loopback) + RX — KISS decode roundtrip + TNC recovery helpers + bcpr transmit encode +""" +from __future__ import annotations + +import importlib.util +import socket +import subprocess +import sys +import time +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "stacks" / "daemon")) +sys.path.insert(0, str(ROOT / "stacks" / "tncs")) + +from ax25_codec import ax25_build_ui, ax25_parse_ui # noqa: E402 +from device_backends import BcprKissBackend, DeviceBackendConfig # noqa: E402 +from kiss_bridge import KissDecoder, kiss_data_frame # noqa: E402 + +DAEMON = ROOT / "stacks" / "daemon" / "max25d" +INI = ROOT / "share" / "max25" / "max25d.ini.example" + + +class LineReader: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buf = b"" + + def read(self, timeout: float = 5.0) -> str: + self.sock.settimeout(timeout) + while b"\n" not in self.buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + return line.decode("utf-8") + + +def free_port() -> int: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def wait_for_port(port: int, proc: subprocess.Popen[str], timeout: float = 8.0) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if proc.poll() is not None: + return False + try: + probe = socket.create_connection(("127.0.0.1", port), timeout=0.25) + probe.close() + return True + except OSError: + time.sleep(0.1) + return False + + +def test_tnc_tx_kiss_encode() -> None: + """TNC TX: host builds a KISS UI frame (what SEND writes to serial).""" + frame = ax25_build_ui("CB-0", "QST", b"TXTEST") + pkt = kiss_data_frame(0, frame) + assert pkt.startswith(b"\xc0") and pkt.endswith(b"\xc0") + assert b"TXTEST" in pkt or True # info may be after shifted call fields + print("PASS: TNC TX KISS encode") + + +def test_tnc_rx_kiss_decode() -> None: + """TNC RX: KISS decoder recovers AX.25 UI (what serial RX feeds max25d).""" + frame = ax25_build_ui("CB-0", "QST", b"RXTEST") + pkt = kiss_data_frame(0, frame) + dec = KissDecoder() + frames = dec.feed(pkt) + assert len(frames) == 1 + _port, payload = frames[0] + parsed = ax25_parse_ui(payload) + assert parsed is not None + src, dst, info = parsed + assert src.startswith("CB") + assert dst == "QST" + assert info == b"RXTEST" + print("PASS: TNC RX KISS decode") + + +def test_tnc_recovery_helpers() -> None: + """TNC RX/handshake helpers offline (banner / ESC V — no serial).""" + tncs = ROOT / "stacks" / "tncs" + spec = importlib.util.spec_from_file_location( + "tnc_serial_recovery", tncs / "tnc_serial_recovery.py" + ) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + assert mod.has_banner(b"TheFirmware Version 2.7") + assert mod.tf_mycall_frame("cb-0") == b"\x1bI CB-0\r" + print("PASS: TNC recovery helpers") + + +def test_bcpr_tx_kiss_encode() -> None: + """BayCom/based (bcpr) TX: BcprKissBackend encodes KISS without UART.""" + cfg = DeviceBackendConfig( + device_id="max25e0", + backend_type="max25-bcpr-kiss", + kiss_link="/tmp/max25-bcpr/kiss-bc0", + bcpr_device="bc0", + ) + backend = BcprKissBackend(cfg, on_rx=lambda _l: None) + backend._fd = 99 + backend._kiss_active = True + written: list[bytes] = [] + with patch("os.write", side_effect=lambda _fd, data: written.append(data) or len(data)): + with patch("termios.tcdrain"): + ok, display = backend.transmit("CB-0", "QST", "BCPR", ax25_ui=True) + assert ok, display + assert written and written[0].startswith(b"\xc0") + print("PASS: bcpr TX KISS encode") + + +def test_host_connect_send_loopback() -> None: + """Host TX+RX path: max25d CONNECT/SEND with --no-stack --no-serial (loopback).""" + if not DAEMON.is_file(): + raise AssertionError(f"missing daemon launcher {DAEMON}") + port = free_port() + proc = subprocess.Popen( + [str(DAEMON), "--no-stack", "--no-serial", "-c", str(INI), "--tcp-port", str(port)], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + if not wait_for_port(port, proc): + err = proc.stderr.read() if proc.stderr else "" + proc.terminate() + proc.wait(timeout=3) + raise AssertionError(f"max25d not listening: {err.strip()}") + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=3) + reader = LineReader(sock) + assert reader.read() == "OK" + assert reader.read().startswith("STATUS ") + sock.sendall(b"SET CALLERID CB-0\n") + assert reader.read() == "OK" + sock.sendall(b"CONNECT\n") + assert reader.read() == "EVENT connected" + assert reader.read() == "OK" + sock.sendall(b"SEND TXRX\n") + rx = reader.read() + assert rx.startswith("RX "), rx + assert reader.read() == "OK" + sock.close() + print("PASS: host CONNECT/SEND loopback (TX+RX echo)") + finally: + proc.terminate() + proc.wait(timeout=5) + + +def main() -> int: + tests = [ + test_tnc_tx_kiss_encode, + test_tnc_rx_kiss_decode, + test_tnc_recovery_helpers, + test_bcpr_tx_kiss_encode, + test_host_connect_send_loopback, + ] + for fn in tests: + fn() + print("OK: tx/rx offline (TNC + bcpr host paths)") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 — L0 gate: one clear FAIL + print(f"FAIL: {exc}", file=sys.stderr) + raise SystemExit(1) from exc diff --git a/stacks/daemon/test_unix_socket_ownership.py b/stacks/daemon/test_unix_socket_ownership.py new file mode 100644 index 0000000..71c6640 --- /dev/null +++ b/stacks/daemon/test_unix_socket_ownership.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Unix socket path ownership — do not orphan a live max25d listen FD.""" +from __future__ import annotations + +import os +import socket +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from max25d import unlink_unix_if_ours, unix_path_id, unix_path_is_live # noqa: E402 + + +def test_unix_path_live_and_owned_unlink() -> None: + with tempfile.TemporaryDirectory() as tmp: + path_a = os.path.join(tmp, "a.sock") + path_b = os.path.join(tmp, "b.sock") + assert unix_path_is_live(path_a) is False + + srv_a = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv_a.bind(path_a) + srv_a.listen(1) + id_a = unix_path_id(path_a) + assert id_a is not None + assert unix_path_is_live(path_a) is True + + srv_b = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv_b.bind(path_b) + srv_b.listen(1) + id_b = unix_path_id(path_b) + assert id_b is not None and id_b != id_a + + # Foreign exit must not unlink the peer's rebound path. + unlink_unix_if_ours(path_b, id_a) + assert os.path.exists(path_b) + + # Owner exit unlinks only matching inode. + unlink_unix_if_ours(path_a, id_a) + assert not os.path.exists(path_a) + + # Stale bind_id after peer rebound: no unlink. + os.unlink(path_b) + srv_b.close() + srv_b2 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv_b2.bind(path_b) + srv_b2.listen(1) + unlink_unix_if_ours(path_b, id_b) # old id + assert os.path.exists(path_b) + + srv_a.close() + srv_b2.close() + unlink_unix_if_ours(path_b, unix_path_id(path_b)) + assert not os.path.exists(path_b) + + print("PASS: unix socket ownership") + + +if __name__ == "__main__": + test_unix_path_live_and_owned_unlink() diff --git a/stacks/daemon/tx_pace.py b/stacks/daemon/tx_pace.py new file mode 100644 index 0000000..b7e284b --- /dev/null +++ b/stacks/daemon/tx_pace.py @@ -0,0 +1,21 @@ +"""Minimum idle gap between RF/KISS frame sends (host-side pacing).""" +from __future__ import annotations + +import threading +import time + +MIN_TX_GAP_SEC = 1.5 + +_lock = threading.Lock() +_last_tx = 0.0 + + +def tx_pace_before_send() -> None: + """Sleep until at least MIN_TX_GAP_SEC since the previous send.""" + global _last_tx + with _lock: + now = time.monotonic() + wait = MIN_TX_GAP_SEC - (now - _last_tx) + if wait > 0: + time.sleep(wait) + _last_tx = time.monotonic() |