Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d07d312e92 | ||
|
|
1e1dab66b1 | ||
|
|
f46c7983ef | ||
|
|
32692030fb | ||
|
|
f22caf6259 | ||
|
|
c2c349043b |
@@ -19,6 +19,22 @@ jobs:
|
|||||||
- name: Run tests
|
- name: Run tests
|
||||||
run: python3 -m pytest tests -v
|
run: python3 -m pytest tests -v
|
||||||
|
|
||||||
|
test-op25:
|
||||||
|
# Config generation + tk_drb_conv state machine only (node-26#24) — no
|
||||||
|
# GNU Radio here; tests/test_conventional.py stubs OP25's own modules.
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
defaults:
|
||||||
|
run:
|
||||||
|
working-directory: op25-container
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- name: Install dependencies
|
||||||
|
run: python3 -m pip install --break-system-packages -r requirements.txt pytest httpx
|
||||||
|
|
||||||
|
- name: Run tests
|
||||||
|
run: python3 -m pytest tests -v
|
||||||
|
|
||||||
lint:
|
lint:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
|
|||||||
@@ -0,0 +1,243 @@
|
|||||||
|
"""
|
||||||
|
Attribution for CONVENTIONAL channels — P25 conventional (node-26#24) and
|
||||||
|
analog NBFM (node-26#20) share this one path.
|
||||||
|
|
||||||
|
metadata_watcher labels audio with whatever the OP25 console says was talking.
|
||||||
|
On a trunked system that is a talkgroup from a control-channel grant. On a
|
||||||
|
conventional channel there is no control channel, so:
|
||||||
|
|
||||||
|
* P25 conventional: link control in each LDU1 usually carries the talkgroup
|
||||||
|
and unit id, and our tk_drb_conv module adds `voice` (frames arriving) and
|
||||||
|
encryption. The talkgroup is often the P25 default (1) or "everyone"
|
||||||
|
(0xFFFF) and identical on every channel, which says nothing about WHICH
|
||||||
|
channel was keyed, and sometimes it is missing altogether.
|
||||||
|
* Analog: OP25 reports the channel with tgid None and nothing else — the
|
||||||
|
console never says anyone is talking.
|
||||||
|
|
||||||
|
So a conventional channel gets a stable SYNTHETIC talkgroup id derived from its
|
||||||
|
frequency (channel_tgid), used whenever link control gives nothing that
|
||||||
|
distinguishes the channel. A real, distinct link-control talkgroup is still
|
||||||
|
used when present. normalize() rewrites the console update so the rest of the
|
||||||
|
watcher sees ordinary "talkgroup X is active" rows and grants and needs no
|
||||||
|
conventional special cases, except one: with exactly ONE conventional channel
|
||||||
|
configured, every sound OP25 puts on drb_sink came from that channel, so audio
|
||||||
|
the console said nothing about is attributed to it (sole_channel) instead of
|
||||||
|
being discarded as orphan audio. With two channels there is no such certainty
|
||||||
|
and orphan audio is still discarded (DEFERRED.md, metadata_watcher row).
|
||||||
|
|
||||||
|
Encrypted conventional traffic is never a talker: the decoder mutes it, so no
|
||||||
|
audio follows, and letting it compete for attribution could mislabel a clear
|
||||||
|
call on the other channel. It is reported instead (encryption_report).
|
||||||
|
"""
|
||||||
|
import time
|
||||||
|
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||||
|
|
||||||
|
# P25 talkgroups that identify no particular group: TIA-102 default (1) and
|
||||||
|
# "everyone" (0xFFFF). 0 is "none" and already rejected by _as_int.
|
||||||
|
NON_DISTINCT_TGIDS = frozenset({1, 0xFFFF})
|
||||||
|
|
||||||
|
# Synthetic ids are frequency in 100 Hz units: 460.400 MHz -> 4_604_000. Any
|
||||||
|
# RTL-SDR frequency (>= 24 MHz) gives >= 240_000, clear of the 16-bit P25
|
||||||
|
# talkgroup space, and 6.25 kHz channel steps stay distinct. C2 resolves the
|
||||||
|
# same number back to the channel name (server talkgroups.name_from_system).
|
||||||
|
SYNTHETIC_TGID_MIN = 240_000
|
||||||
|
|
||||||
|
# How long a channel's last real talkgroup is remembered for a transmission
|
||||||
|
# whose link control hasn't decoded (yet). Covers the back-and-forth of one
|
||||||
|
# conversation; past it, an undecoded call is labelled with the channel.
|
||||||
|
REMEMBER_TGID_SECONDS = 30.0
|
||||||
|
|
||||||
|
|
||||||
|
def channel_tgid(freq: Any) -> Optional[int]:
|
||||||
|
"""Stable synthetic talkgroup id for a conventional channel frequency (Hz or MHz)."""
|
||||||
|
try:
|
||||||
|
f = float(freq)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
if f <= 0:
|
||||||
|
return None
|
||||||
|
hz = f * 1_000_000 if f < 1_000_000 else f
|
||||||
|
return int(round(hz / 100))
|
||||||
|
|
||||||
|
|
||||||
|
def _as_int(value: Any) -> Optional[int]:
|
||||||
|
try:
|
||||||
|
n = int(value)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
return n if n > 0 else None
|
||||||
|
|
||||||
|
|
||||||
|
def _truthy(value: Any) -> bool:
|
||||||
|
return bool(value) and value not in ("0", "false", "False")
|
||||||
|
|
||||||
|
|
||||||
|
ALGID_NAMES = {
|
||||||
|
0x81: "DES-OFB", 0x83: "Triple DES", 0x84: "AES-256", 0x85: "AES-128",
|
||||||
|
0x9F: "DES-XL", 0xA0: "DVI-XL", 0xA1: "DVP-XL", 0xAA: "ADP (RC4)",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def algid_name(algid: Any) -> str:
|
||||||
|
a = _as_int(algid)
|
||||||
|
if a is None:
|
||||||
|
return "unknown algorithm"
|
||||||
|
return f"{ALGID_NAMES.get(a, 'algorithm')} (ALGID 0x{a:02X})"
|
||||||
|
|
||||||
|
|
||||||
|
class ConventionalChannels:
|
||||||
|
"""Per-poll normalizer plus what it has learned about the channels."""
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
# msgqid -> latest raw conventional row. channel_update is not on every
|
||||||
|
# poll, so this outlives a single update.
|
||||||
|
self._rows: Dict[str, Dict[str, Any]] = {}
|
||||||
|
self._known = False # have we seen a channel_update at all
|
||||||
|
# synthetic channel id -> the talkgroup that channel last carried. A
|
||||||
|
# transmission's header frame sets `voice` ~180 ms before its first
|
||||||
|
# link control names the talkgroup; without this, every such poll
|
||||||
|
# would label the call with the synthetic id and the real talkgroup
|
||||||
|
# a poll later would force a bogus split (drb-audio-review, #24).
|
||||||
|
self._last_identity: Dict[int, Tuple[int, float]] = {}
|
||||||
|
self._clock: Callable[[], float] = time.time
|
||||||
|
|
||||||
|
# -- identity --------------------------------------------------------
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _name(row: Dict[str, Any]) -> str:
|
||||||
|
name = str(row.get("name") or "").strip()
|
||||||
|
if name and not name.startswith("["):
|
||||||
|
return name
|
||||||
|
f = row.get("freq")
|
||||||
|
try:
|
||||||
|
return f"{float(f) / 1e6:.4f} MHz"
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return "Conventional"
|
||||||
|
|
||||||
|
def identity(self, row: Dict[str, Any], lc_tgid: Optional[int]) -> Optional[int]:
|
||||||
|
"""Distinctive link-control talkgroup; else (no link control decoded
|
||||||
|
yet) what this channel last carried; else the channel's synthetic id."""
|
||||||
|
synthetic = channel_tgid(row.get("freq"))
|
||||||
|
now = self._clock()
|
||||||
|
if lc_tgid is not None and lc_tgid not in NON_DISTINCT_TGIDS:
|
||||||
|
if synthetic is not None:
|
||||||
|
self._last_identity[synthetic] = (lc_tgid, now)
|
||||||
|
return lc_tgid
|
||||||
|
remembered = self._last_identity.get(synthetic) if synthetic is not None else None
|
||||||
|
if lc_tgid is None and remembered and now - remembered[1] <= REMEMBER_TGID_SECONDS:
|
||||||
|
self._last_identity[synthetic] = (remembered[0], now) # still the same conversation
|
||||||
|
return remembered[0]
|
||||||
|
if synthetic is not None and lc_tgid is not None:
|
||||||
|
self._last_identity[synthetic] = (synthetic, now) # explicit default TG: channel identity
|
||||||
|
return synthetic
|
||||||
|
|
||||||
|
# -- normalization ---------------------------------------------------
|
||||||
|
|
||||||
|
def normalize(self, update: Any) -> Any:
|
||||||
|
"""Rewrite conventional rows and grants in place; trunked ones untouched."""
|
||||||
|
if update.channels:
|
||||||
|
self._known = True
|
||||||
|
self._rows = {}
|
||||||
|
for i, row in enumerate(update.channels):
|
||||||
|
if not _truthy(row.get("conventional")):
|
||||||
|
continue
|
||||||
|
key = str(row.get("msgqid", i))
|
||||||
|
self._rows[key] = row
|
||||||
|
update.channels[i] = self._normalize_row(row)
|
||||||
|
|
||||||
|
for i, entry in enumerate(update.call_log):
|
||||||
|
row = self._rows.get(str(entry.get("rcvr")))
|
||||||
|
if row is None:
|
||||||
|
continue
|
||||||
|
lc = _as_int(entry.get("tgid"))
|
||||||
|
tgid = self.identity(row, lc)
|
||||||
|
fixed = dict(entry)
|
||||||
|
fixed["lc_tgid"] = lc
|
||||||
|
fixed["tgid"] = tgid
|
||||||
|
fixed["tgtag"] = self._label(row, lc, entry.get("tgtag"))
|
||||||
|
fixed["freq"] = entry.get("freq") or row.get("freq")
|
||||||
|
update.call_log[i] = fixed
|
||||||
|
return update
|
||||||
|
|
||||||
|
def _label(self, row: Dict[str, Any], lc: Optional[int], tag: Any) -> str:
|
||||||
|
name = self._name(row)
|
||||||
|
tag = str(tag or "").strip()
|
||||||
|
if lc is not None and lc not in NON_DISTINCT_TGIDS and tag and tag not in ("Conventional", name):
|
||||||
|
return tag
|
||||||
|
return name
|
||||||
|
|
||||||
|
def _normalize_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
lc = _as_int(row.get("tgid"))
|
||||||
|
encrypted = _truthy(row.get("encrypted"))
|
||||||
|
# tgid is only ever set while a call is up (tk_p25 / tk_drb_conv clear
|
||||||
|
# it on TDU and sync timeout); `voice` is tk_drb_conv's own signal.
|
||||||
|
active = (lc is not None or _truthy(row.get("voice"))) and not encrypted
|
||||||
|
out = dict(row)
|
||||||
|
out["lc_tgid"] = lc
|
||||||
|
out["tgid"] = self.identity(row, lc) if active else None
|
||||||
|
out["tag"] = self._label(row, lc, row.get("tag"))
|
||||||
|
out["drb_conventional"] = True
|
||||||
|
out["drb_active"] = active
|
||||||
|
return out
|
||||||
|
|
||||||
|
# -- what the watcher asks -------------------------------------------
|
||||||
|
|
||||||
|
def same_channel_relabel(self, current_tgid: Optional[int], current_freq: Any,
|
||||||
|
new_tgid: Optional[int], new_freq: Any) -> Optional[str]:
|
||||||
|
"""Two labels for ONE conventional channel are not a talkgroup change.
|
||||||
|
|
||||||
|
"adopt": the open segment holds the channel's synthetic id and the
|
||||||
|
console now names the real talkgroup — relabel, don't split.
|
||||||
|
"keep": the segment holds a real talkgroup and the console offers the
|
||||||
|
synthetic id (link control not decoded on this frame) — ignore.
|
||||||
|
None: not the same conventional channel; the normal split rules apply.
|
||||||
|
"""
|
||||||
|
if current_tgid is None or new_tgid is None or current_tgid == new_tgid:
|
||||||
|
return None
|
||||||
|
synthetic = channel_tgid(new_freq)
|
||||||
|
if synthetic is None or channel_tgid(current_freq) != synthetic:
|
||||||
|
return None
|
||||||
|
if not any(channel_tgid(r.get("freq")) == synthetic for r in self._rows.values()):
|
||||||
|
return None # trunked: freq says nothing about identity
|
||||||
|
if current_tgid == synthetic:
|
||||||
|
return "adopt"
|
||||||
|
if new_tgid == synthetic:
|
||||||
|
return "keep"
|
||||||
|
return None
|
||||||
|
|
||||||
|
def sole_channel(self) -> Optional[Dict[str, Any]]:
|
||||||
|
"""The one conventional channel this node listens to, if exactly one
|
||||||
|
(and it is not encrypted). None on any trunked or multi-channel setup."""
|
||||||
|
if len(self._rows) != 1:
|
||||||
|
return None
|
||||||
|
row = next(iter(self._rows.values()))
|
||||||
|
tgid = channel_tgid(row.get("freq"))
|
||||||
|
if tgid is None:
|
||||||
|
return None
|
||||||
|
if _truthy(row.get("encrypted")):
|
||||||
|
return None
|
||||||
|
return {"tgid": tgid, "name": self._name(row), "freq": row.get("freq")}
|
||||||
|
|
||||||
|
def encryption_report(self) -> List[Dict[str, Any]]:
|
||||||
|
"""Per conventional channel: is it (or was its last call) encrypted."""
|
||||||
|
out = []
|
||||||
|
for row in self._rows.values():
|
||||||
|
enc_calls = _as_int(row.get("enc_calls")) or 0
|
||||||
|
clear_calls = _as_int(row.get("clear_calls")) or 0
|
||||||
|
now_enc = _truthy(row.get("encrypted"))
|
||||||
|
last_enc = row.get("last_enc_time") or 0
|
||||||
|
last_clear = row.get("last_clear_time") or 0
|
||||||
|
out.append({
|
||||||
|
"name": self._name(row),
|
||||||
|
"freq": row.get("freq"),
|
||||||
|
"nac": _as_int(row.get("nac")),
|
||||||
|
"encrypted_now": now_enc,
|
||||||
|
"enc_calls": enc_calls,
|
||||||
|
"clear_calls": clear_calls,
|
||||||
|
# The channel "is encrypted" when its most recent call was.
|
||||||
|
"encrypted": now_enc or (enc_calls > 0 and float(last_enc) >= float(last_clear)),
|
||||||
|
"algid": _as_int(row.get("algid")) or _as_int(row.get("last_enc_algid")),
|
||||||
|
"keyid": _as_int(row.get("keyid")) or _as_int(row.get("last_enc_keyid")),
|
||||||
|
"voice": _truthy(row.get("voice")),
|
||||||
|
})
|
||||||
|
return out
|
||||||
@@ -3,8 +3,12 @@ Manages the persisted node API key, plus the local-auth signing material used
|
|||||||
by app/internal/auth.py.
|
by app/internal/auth.py.
|
||||||
|
|
||||||
The API key is provisioned by the C2 server after an admin approves the node.
|
The API key is provisioned by the C2 server after an admin approves the node.
|
||||||
It arrives via MQTT and is saved to /configs/credentials.json so it survives
|
install.sh fetches it over HTTPS (GET /nodes/{id}/credentials + the pickup
|
||||||
container restarts.
|
secret it saved to /configs/pickup_secret) and writes /configs/credentials.json;
|
||||||
|
the running app can also fetch it itself (app/internal/key_pickup.py) and
|
||||||
|
re-read the file at any time (reload_api_key), so a key written after startup
|
||||||
|
is picked up without a container restart (node-26#14). The legacy MQTT
|
||||||
|
delivery (nodes/{id}/api_key) still saves here too.
|
||||||
|
|
||||||
The scrypt salt and session-signing secret are generated locally on first boot
|
The scrypt salt and session-signing secret are generated locally on first boot
|
||||||
(never provisioned externally) and persisted the same way, so dashboard
|
(never provisioned externally) and persisted the same way, so dashboard
|
||||||
@@ -12,12 +16,14 @@ sessions survive a container restart instead of forcing every operator to
|
|||||||
re-login whenever the node restarts.
|
re-login whenever the node restarts.
|
||||||
"""
|
"""
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import secrets
|
import secrets
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.internal.logger import logger
|
from app.internal.logger import logger
|
||||||
|
|
||||||
_CREDS_FILE = Path(settings.config_path) / "credentials.json"
|
_CREDS_FILE = Path(settings.config_path) / "credentials.json"
|
||||||
|
_PICKUP_FILE = Path(settings.config_path) / "pickup_secret"
|
||||||
_api_key: str | None = None
|
_api_key: str | None = None
|
||||||
_auth_salt: bytes | None = None
|
_auth_salt: bytes | None = None
|
||||||
_session_secret: bytes | None = None
|
_session_secret: bytes | None = None
|
||||||
@@ -74,6 +80,37 @@ def get_api_key() -> str | None:
|
|||||||
return _api_key
|
return _api_key
|
||||||
|
|
||||||
|
|
||||||
|
def reload_api_key() -> str | None:
|
||||||
|
"""Re-read the api_key from disk (install.sh, or an operator, may have
|
||||||
|
written it after startup). Returns the key now in effect. Only the api_key
|
||||||
|
is taken from the file — the auth material already in memory stays."""
|
||||||
|
global _api_key
|
||||||
|
try:
|
||||||
|
data = json.loads(_CREDS_FILE.read_text())
|
||||||
|
except FileNotFoundError:
|
||||||
|
return _api_key
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not re-read credentials file: {e}")
|
||||||
|
return _api_key
|
||||||
|
key = data.get("api_key")
|
||||||
|
if key and key != _api_key:
|
||||||
|
_api_key = key
|
||||||
|
logger.info("Node API key re-read from disk (changed since it was last loaded).")
|
||||||
|
return _api_key
|
||||||
|
|
||||||
|
|
||||||
|
def get_pickup_secret() -> str | None:
|
||||||
|
"""The one-time pickup secret install.sh saved at enrollment, if any."""
|
||||||
|
try:
|
||||||
|
secret = _PICKUP_FILE.read_text().strip()
|
||||||
|
except FileNotFoundError:
|
||||||
|
return None
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Could not read pickup secret: {e}")
|
||||||
|
return None
|
||||||
|
return secret or None
|
||||||
|
|
||||||
|
|
||||||
def save_api_key(key: str) -> None:
|
def save_api_key(key: str) -> None:
|
||||||
global _api_key
|
global _api_key
|
||||||
_api_key = key
|
_api_key = key
|
||||||
@@ -88,4 +125,10 @@ def _write() -> None:
|
|||||||
data["auth_salt"] = _auth_salt.hex()
|
data["auth_salt"] = _auth_salt.hex()
|
||||||
if _session_secret is not None:
|
if _session_secret is not None:
|
||||||
data["session_secret"] = _session_secret.hex()
|
data["session_secret"] = _session_secret.hex()
|
||||||
_CREDS_FILE.write_text(json.dumps(data))
|
# Atomic replace with 0600: install.sh on the host writes this same file,
|
||||||
|
# and a torn half-written JSON would lose the key AND the auth material.
|
||||||
|
tmp = _CREDS_FILE.with_name(_CREDS_FILE.name + ".tmp")
|
||||||
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
||||||
|
with os.fdopen(fd, "w") as f:
|
||||||
|
f.write(json.dumps(data))
|
||||||
|
os.replace(tmp, _CREDS_FILE)
|
||||||
|
|||||||
@@ -56,6 +56,51 @@ def _ago(seconds: float) -> str:
|
|||||||
return f"{seconds // 3600} h ago"
|
return f"{seconds // 3600} h ago"
|
||||||
|
|
||||||
|
|
||||||
|
def _channel_label(ch: Dict[str, Any]) -> str:
|
||||||
|
freq = _mhz(ch.get("freq"))
|
||||||
|
name = ch.get("name") or ""
|
||||||
|
if name and freq and freq not in name:
|
||||||
|
return f"{name} ({freq})"
|
||||||
|
return name or freq or "a channel"
|
||||||
|
|
||||||
|
|
||||||
|
def _conventional_decode_check(chans: List[Dict[str, Any]]) -> Dict[str, str]:
|
||||||
|
"""P25 conventional (node-26#24): no control channel to lock on to, so the
|
||||||
|
proof of decoding is P25 frames (a NAC) heard on the channels — and the one
|
||||||
|
failure the owner can't fix is encryption."""
|
||||||
|
if not chans:
|
||||||
|
return _check("decode", "Decoding", WARN, "Waiting for OP25 to report its channels…")
|
||||||
|
encrypted = [c for c in chans if c.get("encrypted")]
|
||||||
|
if encrypted:
|
||||||
|
from app.internal.conventional import algid_name
|
||||||
|
parts = []
|
||||||
|
for c in encrypted:
|
||||||
|
key = f", key 0x{c['keyid']:04X}" if c.get("keyid") is not None else ""
|
||||||
|
parts.append(f"{_channel_label(c)}: {algid_name(c.get('algid'))}{key}")
|
||||||
|
# "Everything is encrypted" only when no clear call was ever heard —
|
||||||
|
# `encrypted` alone means a channel's most recent call was.
|
||||||
|
every = len(encrypted) == len(chans) and not any(c.get("clear_calls") for c in chans)
|
||||||
|
return _check(
|
||||||
|
"decode", "Decoding", ERROR if every else WARN,
|
||||||
|
("Encrypted — " if every else "Partly encrypted — ") + "; ".join(parts)
|
||||||
|
+ ". Encrypted calls can't be decoded, so nothing from "
|
||||||
|
+ ("these channels" if every else "those channels") + " is recorded.",
|
||||||
|
"Nothing to fix on the node: the agency encrypts this traffic. "
|
||||||
|
+ ("Point the node at a different system." if every else "Clear calls are still recorded."),
|
||||||
|
)
|
||||||
|
heard = [c for c in chans if c.get("nac")]
|
||||||
|
listening = ", ".join(_channel_label(c) for c in chans)
|
||||||
|
if not heard:
|
||||||
|
return _check(
|
||||||
|
"decode", "Decoding", OK,
|
||||||
|
f"Listening on {listening} (P25 conventional) — no P25 traffic heard yet.",
|
||||||
|
"Normal on a quiet channel: conventional channels are silent until someone talks. If it stays "
|
||||||
|
"silent through a busy period, check the frequency, antenna and the SDR's PPM.",
|
||||||
|
)
|
||||||
|
nacs = ", ".join(sorted({f"0x{c['nac']:03X}" for c in heard}))
|
||||||
|
return _check("decode", "Decoding", OK, f"Listening on {listening} (P25 conventional) — decoding, NAC {nacs}.")
|
||||||
|
|
||||||
|
|
||||||
def active_config(node_cfg) -> Any:
|
def active_config(node_cfg) -> Any:
|
||||||
if node_cfg.override_system_id and node_cfg.override_config:
|
if node_cfg.override_system_id and node_cfg.override_config:
|
||||||
return node_cfg.override_config
|
return node_cfg.override_config
|
||||||
@@ -144,7 +189,8 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
|
|||||||
))
|
))
|
||||||
else:
|
else:
|
||||||
origin = "chosen on this dashboard" if src["local"] else "assigned by the server"
|
origin = "chosen on this dashboard" if src["local"] else "assigned by the server"
|
||||||
checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({src['type']}) — {origin}."))
|
kind = "P25 conventional" if src.get("mode") == "conventional" else src["type"]
|
||||||
|
checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({kind}) — {origin}."))
|
||||||
|
|
||||||
# 6. OP25
|
# 6. OP25
|
||||||
op25 = f.get("op25_status")
|
op25 = f.get("op25_status")
|
||||||
@@ -170,6 +216,7 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
|
|||||||
console = f.get("console") or {}
|
console = f.get("console") or {}
|
||||||
last_poll = console.get("last_poll_epoch")
|
last_poll = console.get("last_poll_epoch")
|
||||||
is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG")
|
is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG")
|
||||||
|
is_conventional = bool(src) and src.get("mode") == "conventional"
|
||||||
if not running:
|
if not running:
|
||||||
checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs."))
|
checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs."))
|
||||||
elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS:
|
elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS:
|
||||||
@@ -182,8 +229,9 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
|
|||||||
freq = _mhz(chans[0].get("freq")) if chans else None
|
freq = _mhz(chans[0].get("freq")) if chans else None
|
||||||
checks.append(_check(
|
checks.append(_check(
|
||||||
"decode", "Decoding", OK, f"Tuned to {freq or 'the analog frequency'} (analog FM — no decoding needed).",
|
"decode", "Decoding", OK, f"Tuned to {freq or 'the analog frequency'} (analog FM — no decoding needed).",
|
||||||
"Live listening only: analog calls are not recorded or uploaded yet (node-26#20).",
|
|
||||||
))
|
))
|
||||||
|
elif is_conventional:
|
||||||
|
checks.append(_conventional_decode_check(f.get("conventional") or []))
|
||||||
else:
|
else:
|
||||||
trunk = console.get("trunk") or []
|
trunk = console.get("trunk") or []
|
||||||
tsbk = [float(t.get("last_tsbk") or 0) for t in trunk]
|
tsbk = [float(t.get("last_tsbk") or 0) for t in trunk]
|
||||||
@@ -261,6 +309,7 @@ async def gather_facts() -> Dict[str, Any]:
|
|||||||
from app.internal.metadata_watcher import metadata_watcher
|
from app.internal.metadata_watcher import metadata_watcher
|
||||||
from app.internal.mqtt_manager import mqtt_manager
|
from app.internal.mqtt_manager import mqtt_manager
|
||||||
from app.internal.op25_client import op25_client
|
from app.internal.op25_client import op25_client
|
||||||
|
from app.internal.radio_config import system_mode
|
||||||
from app.internal.secondary_sdr_client import secondary_sdr_client
|
from app.internal.secondary_sdr_client import secondary_sdr_client
|
||||||
|
|
||||||
node_cfg = load_node_config()
|
node_cfg = load_node_config()
|
||||||
@@ -289,8 +338,10 @@ async def gather_facts() -> Dict[str, Any]:
|
|||||||
"sdr_count": sdr_count,
|
"sdr_count": sdr_count,
|
||||||
"source": None if cfg is None else {
|
"source": None if cfg is None else {
|
||||||
"name": cfg.name, "type": cfg.type, "system_id": cfg.system_id,
|
"name": cfg.name, "type": cfg.type, "system_id": cfg.system_id,
|
||||||
|
"mode": system_mode(cfg.type, cfg.config),
|
||||||
"local": bool(node_cfg.override_system_id and node_cfg.override_config),
|
"local": bool(node_cfg.override_system_id and node_cfg.override_config),
|
||||||
},
|
},
|
||||||
|
"conventional": metadata_watcher.conventional_channels,
|
||||||
"op25_status": (op25 or {}).get("status") if op25 is not None else None,
|
"op25_status": (op25 or {}).get("status") if op25 is not None else None,
|
||||||
"console": metadata_watcher.console_state,
|
"console": metadata_watcher.console_state,
|
||||||
"audio_capturing": call_recorder.is_capturing,
|
"audio_capturing": call_recorder.is_capturing,
|
||||||
|
|||||||
@@ -0,0 +1,73 @@
|
|||||||
|
"""
|
||||||
|
HTTPS api_key pickup — the node's own half of C2's enrollment flow
|
||||||
|
(Server/drb-c2-core/app/routers/enrollment.py, GET /nodes/{id}/credentials).
|
||||||
|
|
||||||
|
Why this exists (node-26#14): key delivery used to be MQTT-only
|
||||||
|
(nodes/{id}/api_key), and a dynsec broker refuses a node that has no key — or
|
||||||
|
a stale one — so a locked-out node could never receive the key that would let
|
||||||
|
it in. install.sh already did this pickup once at install time; doing it from
|
||||||
|
the running app too means a node approved after install, or whose key was
|
||||||
|
reissued, recovers on its own with no SSH and no container restart.
|
||||||
|
|
||||||
|
Auth is the pickup secret install.sh saved to /configs/pickup_secret at
|
||||||
|
enrollment (the same file, bind-mounted). Only ever sent to settings.c2_url,
|
||||||
|
over whatever scheme the operator configured (https in production).
|
||||||
|
"""
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal import credentials
|
||||||
|
from app.internal.logger import logger
|
||||||
|
|
||||||
|
_TIMEOUT = 15
|
||||||
|
|
||||||
|
|
||||||
|
async def fetch_api_key() -> Optional[str]:
|
||||||
|
"""Ask C2 for this node's api_key. Returns it, or None when there isn't one
|
||||||
|
to be had yet (not approved, no pickup secret, C2 unreachable, rejected).
|
||||||
|
Never raises — the caller retries on its own backoff."""
|
||||||
|
if not settings.c2_url:
|
||||||
|
logger.warning("Key pickup skipped: C2_URL is not configured.")
|
||||||
|
return None
|
||||||
|
secret = credentials.get_pickup_secret()
|
||||||
|
if not secret:
|
||||||
|
logger.warning(
|
||||||
|
f"Key pickup skipped: no {settings.config_path}/pickup_secret. This node was not enrolled by "
|
||||||
|
"install.sh on this machine — re-run install.sh, or have an admin Reissue key and write it in."
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
|
||||||
|
url = f"{settings.c2_url.rstrip('/')}/nodes/{settings.node_id}/credentials"
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
||||||
|
r = await client.get(url, headers={"X-Pickup-Secret": secret})
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Key pickup: could not reach C2 ({e}).")
|
||||||
|
return None
|
||||||
|
|
||||||
|
if r.status_code == 401:
|
||||||
|
logger.error(
|
||||||
|
"Key pickup: C2 rejected the saved pickup secret (401) — it was rotated by a re-enroll elsewhere. "
|
||||||
|
"Re-run install.sh with --token to recover."
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
if r.status_code == 404:
|
||||||
|
logger.error("Key pickup: C2 does not know this node (404) — it was deleted server-side. Re-run install.sh.")
|
||||||
|
return None
|
||||||
|
if r.status_code != 200:
|
||||||
|
logger.warning(f"Key pickup: unexpected HTTP {r.status_code} from C2.")
|
||||||
|
return None
|
||||||
|
|
||||||
|
try:
|
||||||
|
body = r.json()
|
||||||
|
except ValueError:
|
||||||
|
logger.warning("Key pickup: C2 returned a non-JSON body.")
|
||||||
|
return None
|
||||||
|
key = body.get("api_key") if isinstance(body, dict) else None
|
||||||
|
if not key or not isinstance(key, str):
|
||||||
|
status = body.get("approval_status") if isinstance(body, dict) else None
|
||||||
|
logger.info(f"Key pickup: no api_key yet (approval_status={status!r}) — waiting for an admin to approve.")
|
||||||
|
return None
|
||||||
|
return key
|
||||||
@@ -36,6 +36,19 @@ have seen the whole window and ask the rolling console history "what was active
|
|||||||
during this audio, give or take a few seconds" — see _attribute and the
|
during this audio, give or take a few seconds" — see _attribute and the
|
||||||
ATTRIBUTION_* constants.
|
ATTRIBUTION_* constants.
|
||||||
|
|
||||||
|
CONVENTIONAL CHANNELS (node-26#24 P25 conventional, node-26#20 analog). There
|
||||||
|
is no control channel, so internal/conventional.py rewrites each console update
|
||||||
|
first: a conventional channel that is keyed becomes an ordinary active row whose
|
||||||
|
talkgroup is the link-control TGID when it is distinctive, else a stable
|
||||||
|
synthetic id derived from the channel frequency. Such a row counts as a talker
|
||||||
|
without a radio id (`drb_active`) — link control does not always carry one and
|
||||||
|
analog never does. One more rule, applied only at CLOSE: with exactly one
|
||||||
|
conventional channel configured, audio the console said nothing about can only
|
||||||
|
have come from that channel, so it is attributed to it rather than discarded.
|
||||||
|
Not at open: on a channel whose link control carries a real talkgroup, the
|
||||||
|
grant may arrive a poll after the audio, and a premature synthetic label would
|
||||||
|
force a bogus split when it does.
|
||||||
|
|
||||||
ORPHAN AUDIO. If nothing in the console history overlaps the window, the audio
|
ORPHAN AUDIO. If nothing in the console history overlaps the window, the audio
|
||||||
is unattributed: Liquidsoap fallback, a test tone, stray noise, or a dropped
|
is unattributed: Liquidsoap fallback, a test tone, stray noise, or a dropped
|
||||||
`call_log`. Policy is DISCARD AND SHOUT — the recording is not uploaded and no
|
`call_log`. Policy is DISCARD AND SHOUT — the recording is not uploaded and no
|
||||||
@@ -74,6 +87,7 @@ from typing import Optional, Callable, Awaitable, Any, List, Dict
|
|||||||
|
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.internal.call_recorder import AudioActivity
|
from app.internal.call_recorder import AudioActivity
|
||||||
|
from app.internal.conventional import ConventionalChannels
|
||||||
from app.internal.op25_client import op25_client
|
from app.internal.op25_client import op25_client
|
||||||
from app.internal.logger import logger
|
from app.internal.logger import logger
|
||||||
|
|
||||||
@@ -177,6 +191,22 @@ def _as_float(value: Any) -> Optional[float]:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _talker(channel: Dict[str, Any]) -> Optional[tuple]:
|
||||||
|
"""(tgid, rid) if this channel row says someone is transmitting, else None.
|
||||||
|
|
||||||
|
Trunked rows need both a talkgroup and a source address — an idle channel
|
||||||
|
can keep its last tgid. A normalized conventional row (`drb_active`) only
|
||||||
|
has a tgid while a call is up, and may legitimately have no radio id.
|
||||||
|
"""
|
||||||
|
tgid = _as_int(channel.get("tgid"))
|
||||||
|
if tgid is None:
|
||||||
|
return None
|
||||||
|
rid = _as_int(channel.get("srcaddr"))
|
||||||
|
if rid is None and not channel.get("drb_active"):
|
||||||
|
return None
|
||||||
|
return tgid, rid
|
||||||
|
|
||||||
|
|
||||||
def _iso(epoch: Optional[float]) -> Optional[str]:
|
def _iso(epoch: Optional[float]) -> Optional[str]:
|
||||||
if epoch is None:
|
if epoch is None:
|
||||||
return None
|
return None
|
||||||
@@ -252,6 +282,10 @@ class MetadataWatcher:
|
|||||||
# Field-visible counter of discarded orphan audio.
|
# Field-visible counter of discarded orphan audio.
|
||||||
self._unattributed_segments: int = 0
|
self._unattributed_segments: int = 0
|
||||||
|
|
||||||
|
# Conventional-channel normalizer (node-26#24/#20); also remembers the
|
||||||
|
# channel list and their encryption state for the dashboard + checkin.
|
||||||
|
self._conventional = ConventionalChannels()
|
||||||
|
|
||||||
# Injectable for tests; production is always the host wall clock.
|
# Injectable for tests; production is always the host wall clock.
|
||||||
self._clock: Callable[[], float] = time.time
|
self._clock: Callable[[], float] = time.time
|
||||||
|
|
||||||
@@ -300,6 +334,7 @@ class MetadataWatcher:
|
|||||||
return
|
return
|
||||||
|
|
||||||
self._last_ok_poll = now
|
self._last_ok_poll = now
|
||||||
|
update = self._conventional.normalize(update)
|
||||||
self._record_console(update, now)
|
self._record_console(update, now)
|
||||||
self._note_terminal(update, now)
|
self._note_terminal(update, now)
|
||||||
|
|
||||||
@@ -341,10 +376,10 @@ class MetadataWatcher:
|
|||||||
self._console.append(event)
|
self._console.append(event)
|
||||||
|
|
||||||
for channel in update.channels:
|
for channel in update.channels:
|
||||||
tgid = _as_int(channel.get("tgid"))
|
talker = _talker(channel)
|
||||||
srcaddr = _as_int(channel.get("srcaddr"))
|
if talker is None:
|
||||||
if tgid is None or srcaddr is None:
|
|
||||||
continue # idle channel says nothing about who is talking
|
continue # idle channel says nothing about who is talking
|
||||||
|
tgid, srcaddr = talker
|
||||||
self._console.append(ConsoleEvent(
|
self._console.append(ConsoleEvent(
|
||||||
epoch=now,
|
epoch=now,
|
||||||
tgid=tgid,
|
tgid=tgid,
|
||||||
@@ -379,7 +414,7 @@ class MetadataWatcher:
|
|||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def _attribute(self, start: float, end: float) -> Optional[Attribution]:
|
def _attribute(self, start: float, end: float, allow_sole_channel: bool = False) -> Optional[Attribution]:
|
||||||
"""
|
"""
|
||||||
Resolve which talkgroup a stretch of audio belongs to.
|
Resolve which talkgroup a stretch of audio belongs to.
|
||||||
|
|
||||||
@@ -387,7 +422,8 @@ class MetadataWatcher:
|
|||||||
how well its console activity overlaps the audio itself, preferring
|
how well its console activity overlaps the audio itself, preferring
|
||||||
real overlap over merely being nearby, and grants over channel rows.
|
real overlap over merely being nearby, and grants over channel rows.
|
||||||
Returns None only when NOTHING was observed in that band at all — the
|
Returns None only when NOTHING was observed in that band at all — the
|
||||||
orphan-audio case.
|
orphan-audio case — unless allow_sole_channel and the node listens to
|
||||||
|
exactly one conventional channel (see module docstring).
|
||||||
"""
|
"""
|
||||||
low = start - ATTRIBUTION_LOOKBACK_SECONDS
|
low = start - ATTRIBUTION_LOOKBACK_SECONDS
|
||||||
high = end + ATTRIBUTION_LOOKAHEAD_SECONDS
|
high = end + ATTRIBUTION_LOOKAHEAD_SECONDS
|
||||||
@@ -415,7 +451,14 @@ class MetadataWatcher:
|
|||||||
found.rid = event.rid # most recent wins
|
found.rid = event.rid # most recent wins
|
||||||
|
|
||||||
if not candidates:
|
if not candidates:
|
||||||
|
sole = self._conventional.sole_channel() if allow_sole_channel else None
|
||||||
|
if sole is None:
|
||||||
return None
|
return None
|
||||||
|
logger.info(
|
||||||
|
f"No console activity near {start:.3f}-{end:.3f}; attributing to the only "
|
||||||
|
f"conventional channel, {sole['name']} (tgid {sole['tgid']})."
|
||||||
|
)
|
||||||
|
return Attribution(tgid=sole["tgid"], name=sole["name"], freq=sole["freq"])
|
||||||
|
|
||||||
best = max(
|
best = max(
|
||||||
candidates.values(),
|
candidates.values(),
|
||||||
@@ -530,6 +573,11 @@ class MetadataWatcher:
|
|||||||
self._refresh_meta_from_log(entry)
|
self._refresh_meta_from_log(entry)
|
||||||
return
|
return
|
||||||
|
|
||||||
|
if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""):
|
||||||
|
self._transmissions += 1
|
||||||
|
self._refresh_meta_from_log(entry)
|
||||||
|
return
|
||||||
|
|
||||||
# FORCED SPLIT. Two talkgroups can be back to back with no silence
|
# FORCED SPLIT. Two talkgroups can be back to back with no silence
|
||||||
# between them; pure audio segmentation would merge them into one file
|
# between them; pure audio segmentation would merge them into one file
|
||||||
# under one label, which is exactly the kind of wrong that corrupts
|
# under one label, which is exactly the kind of wrong that corrupts
|
||||||
@@ -546,6 +594,21 @@ class MetadataWatcher:
|
|||||||
transmissions=1,
|
transmissions=1,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def _same_channel(self, tgid: int, freq: Any, name: str) -> bool:
|
||||||
|
"""A conventional channel seen under its synthetic id and its real
|
||||||
|
talkgroup is one channel, not a split (see ConventionalChannels.
|
||||||
|
same_channel_relabel). Adopts the real talkgroup when offered."""
|
||||||
|
decision = self._conventional.same_channel_relabel(self._current_tgid, self._current_freq, tgid, freq)
|
||||||
|
if decision == "adopt":
|
||||||
|
logger.info(
|
||||||
|
f"Segment {self._active_call_id}: link control names tgid {tgid} for the channel "
|
||||||
|
f"labelled {self._current_tgid} — relabelling, not splitting."
|
||||||
|
)
|
||||||
|
self._current_tgid = tgid
|
||||||
|
if name:
|
||||||
|
self._current_tgid_name = name
|
||||||
|
return decision is not None
|
||||||
|
|
||||||
async def _scan_channels(self, channels: List[Dict[str, Any]], now: float) -> None:
|
async def _scan_channels(self, channels: List[Dict[str, Any]], now: float) -> None:
|
||||||
"""Channel rows in audio mode: refresh metadata, catch an unlogged split."""
|
"""Channel rows in audio mode: refresh metadata, catch an unlogged split."""
|
||||||
if self._active_call_id is None:
|
if self._active_call_id is None:
|
||||||
@@ -554,13 +617,14 @@ class MetadataWatcher:
|
|||||||
active: List[Dict[str, Any]] = []
|
active: List[Dict[str, Any]] = []
|
||||||
ours = False
|
ours = False
|
||||||
for channel in channels:
|
for channel in channels:
|
||||||
tgid = _as_int(channel.get("tgid"))
|
talker = _talker(channel)
|
||||||
srcaddr = _as_int(channel.get("srcaddr"))
|
if talker is None:
|
||||||
if tgid is None or srcaddr is None:
|
|
||||||
continue
|
continue
|
||||||
|
tgid, srcaddr = talker
|
||||||
active.append(channel)
|
active.append(channel)
|
||||||
if tgid == self._current_tgid:
|
if tgid == self._current_tgid or self._same_channel(tgid, channel.get("freq"), channel.get("tag") or ""):
|
||||||
ours = True
|
ours = True
|
||||||
|
if srcaddr is not None:
|
||||||
self._current_srcaddr = srcaddr
|
self._current_srcaddr = srcaddr
|
||||||
self._last_activity = now
|
self._last_activity = now
|
||||||
self._refresh_meta_from_channel(channel)
|
self._refresh_meta_from_channel(channel)
|
||||||
@@ -658,12 +722,13 @@ class MetadataWatcher:
|
|||||||
foreign_active_tgid: Optional[int] = None
|
foreign_active_tgid: Optional[int] = None
|
||||||
|
|
||||||
for channel in channels:
|
for channel in channels:
|
||||||
srcaddr = _as_int(channel.get("srcaddr"))
|
talker = _talker(channel)
|
||||||
chan_tgid = _as_int(channel.get("tgid"))
|
if talker is None:
|
||||||
if srcaddr is None:
|
|
||||||
continue
|
continue
|
||||||
|
chan_tgid, srcaddr = talker
|
||||||
if chan_tgid == self._current_tgid:
|
if chan_tgid == self._current_tgid:
|
||||||
tx_active = True
|
tx_active = True
|
||||||
|
if srcaddr is not None:
|
||||||
self._current_srcaddr = srcaddr
|
self._current_srcaddr = srcaddr
|
||||||
self._refresh_meta_from_channel(channel)
|
self._refresh_meta_from_channel(channel)
|
||||||
elif chan_tgid is not None:
|
elif chan_tgid is not None:
|
||||||
@@ -868,7 +933,7 @@ class MetadataWatcher:
|
|||||||
two talkgroups' console activity overlaps one recording, i.e. the split
|
two talkgroups' console activity overlaps one recording, i.e. the split
|
||||||
logic should have fired and did not.
|
logic should have fired and did not.
|
||||||
"""
|
"""
|
||||||
found = self._attribute(start, end)
|
found = self._attribute(start, end, allow_sole_channel=self._current_tgid is None)
|
||||||
if found is None:
|
if found is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -931,6 +996,12 @@ class MetadataWatcher:
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@property
|
||||||
|
def conventional_channels(self) -> List[Dict[str, Any]]:
|
||||||
|
"""Per conventional channel: name, freq, NAC heard, encryption. Empty on
|
||||||
|
a trunked system. Read by /api/health and the checkin."""
|
||||||
|
return self._conventional.encryption_report()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def unattributed_segments(self) -> int:
|
def unattributed_segments(self) -> int:
|
||||||
"""Orphan-audio segments discarded since start. Surfaced on /api/status."""
|
"""Orphan-audio segments discarded since start. Surfaced on /api/status."""
|
||||||
|
|||||||
@@ -6,12 +6,30 @@ from typing import Optional, Callable, Awaitable, Dict, Any
|
|||||||
import paho.mqtt.client as mqtt
|
import paho.mqtt.client as mqtt
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.internal.logger import logger
|
from app.internal.logger import logger
|
||||||
from app.internal import credentials
|
from app.internal import credentials, key_pickup
|
||||||
|
|
||||||
CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||||
ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||||
ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||||
|
|
||||||
|
# CONNACK codes meaning "your credentials are wrong", as paho reports them:
|
||||||
|
# MQTT 3.1.1 return codes 4/5, and the MQTT 5 reason codes paho maps them to
|
||||||
|
# (134 Bad user name or password, 135 Not authorized).
|
||||||
|
_AUTH_FAILURE_CODES = {4, 5, 134, 135}
|
||||||
|
|
||||||
|
# Credential-recovery backoff (node-26#14). The pickup endpoint is cheap, but a
|
||||||
|
# node that is simply not approved yet must not hammer C2 for days.
|
||||||
|
RECOVERY_MIN_DELAY = 30
|
||||||
|
RECOVERY_MAX_DELAY = 900
|
||||||
|
|
||||||
|
|
||||||
|
def _is_auth_failure(reason_code) -> bool:
|
||||||
|
value = getattr(reason_code, "value", reason_code)
|
||||||
|
if isinstance(value, int) and value in _AUTH_FAILURE_CODES:
|
||||||
|
return True
|
||||||
|
text = str(reason_code).lower()
|
||||||
|
return "not authori" in text or "bad user name or password" in text
|
||||||
|
|
||||||
|
|
||||||
class MQTTManager:
|
class MQTTManager:
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
@@ -19,6 +37,11 @@ class MQTTManager:
|
|||||||
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
||||||
self._connected = False
|
self._connected = False
|
||||||
self._connect_task: Optional[asyncio.Task] = None
|
self._connect_task: Optional[asyncio.Task] = None
|
||||||
|
# The api_key the CURRENT paho client was built with — not necessarily
|
||||||
|
# credentials.get_api_key(), which can move ahead of it (MQTT key push,
|
||||||
|
# install.sh writing the file). Recovery compares against this.
|
||||||
|
self._client_key: Optional[str] = None
|
||||||
|
self._recovery_task: Optional[asyncio.Task] = None
|
||||||
|
|
||||||
self.on_command: Optional[CommandCallback] = None
|
self.on_command: Optional[CommandCallback] = None
|
||||||
self.on_config_push: Optional[ConfigCallback] = None
|
self.on_config_push: Optional[ConfigCallback] = None
|
||||||
@@ -53,6 +76,7 @@ class MQTTManager:
|
|||||||
)
|
)
|
||||||
|
|
||||||
api_key = credentials.get_api_key()
|
api_key = credentials.get_api_key()
|
||||||
|
self._client_key = api_key
|
||||||
if api_key:
|
if api_key:
|
||||||
# Post-cutover auth: broker's dynsec plugin authenticates this
|
# Post-cutover auth: broker's dynsec plugin authenticates this
|
||||||
# exact (username, password) pair as this node's own client — see
|
# exact (username, password) pair as this node's own client — see
|
||||||
@@ -70,18 +94,14 @@ class MQTTManager:
|
|||||||
client.username_pw_set(settings.mqtt_user, settings.mqtt_pass)
|
client.username_pw_set(settings.mqtt_user, settings.mqtt_pass)
|
||||||
else:
|
else:
|
||||||
# No api_key on disk and no legacy shared login configured. A
|
# No api_key on disk and no legacy shared login configured. A
|
||||||
# dynsec broker (allow_anonymous false) refuses this outright —
|
# dynsec broker (allow_anonymous false) refuses this outright.
|
||||||
# expected, not a bug to route around here: this node hasn't been
|
# connect() starts _recover_credentials() in this case, which
|
||||||
# enrolled/approved yet, and the enrollment flow that would fix
|
# fetches the key over HTTPS (key_pickup.py) and rebuilds this
|
||||||
# that client-side is a later, separate pass (out of scope here;
|
# client with it — node-26#14. paho's reconnect_delay_set() below
|
||||||
# see MQTT-PUBLIC-AUTH-PLAN.md). paho's reconnect_delay_set()
|
# bounds the refusal loop meanwhile (2..60s backoff).
|
||||||
# below bounds the retry rate (2..60s exponential backoff), so
|
|
||||||
# this degrades to a slow, clearly-logged refusal loop via
|
|
||||||
# _on_connect's "MQTT connect refused" line — not a hot spin.
|
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"No API key on disk and no legacy MQTT_USER configured — "
|
"No API key on disk and no legacy MQTT_USER configured — "
|
||||||
"connecting without credentials; the broker is expected to "
|
"connecting without credentials while the key is fetched from C2."
|
||||||
"refuse this until the node is enrolled/approved."
|
|
||||||
)
|
)
|
||||||
|
|
||||||
if settings.mqtt_tls:
|
if settings.mqtt_tls:
|
||||||
@@ -120,6 +140,10 @@ class MQTTManager:
|
|||||||
asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
|
asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop)
|
||||||
else:
|
else:
|
||||||
logger.error(f"MQTT connect refused: {reason_code}")
|
logger.error(f"MQTT connect refused: {reason_code}")
|
||||||
|
if _is_auth_failure(reason_code) and self._loop is not None:
|
||||||
|
# paho thread -> event loop. A refused login never fixes itself
|
||||||
|
# by retrying the same password, so go get the right one.
|
||||||
|
self._loop.call_soon_threadsafe(self._start_recovery, f"broker refused login ({reason_code})")
|
||||||
|
|
||||||
def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
|
def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
|
||||||
self._connected = False
|
self._connected = False
|
||||||
@@ -144,6 +168,72 @@ class MQTTManager:
|
|||||||
self._loop = asyncio.get_event_loop()
|
self._loop = asyncio.get_event_loop()
|
||||||
self._client = self._build_client()
|
self._client = self._build_client()
|
||||||
self._connect_task = asyncio.create_task(self._connect_with_retry())
|
self._connect_task = asyncio.create_task(self._connect_with_retry())
|
||||||
|
if not self._client_key and not settings.mqtt_user:
|
||||||
|
self._start_recovery("no API key at startup")
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# Credential self-heal (node-26#14)
|
||||||
|
#
|
||||||
|
# A node with no key, or a stale one, used to sit in an MQTT refusal loop
|
||||||
|
# forever: key delivery was MQTT-only, and credentials were read once at
|
||||||
|
# startup. Now: re-read credentials.json (install.sh may have written it),
|
||||||
|
# else fetch from C2 over HTTPS with the pickup secret; if that yields a
|
||||||
|
# key different from the one the client is using, rebuild the client with
|
||||||
|
# it in place. No container restart.
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
def _start_recovery(self, why: str) -> None:
|
||||||
|
if self._recovery_task and not self._recovery_task.done():
|
||||||
|
return
|
||||||
|
logger.warning(f"MQTT credential recovery started: {why}")
|
||||||
|
self._recovery_task = asyncio.create_task(self._recover_credentials())
|
||||||
|
|
||||||
|
async def _recover_credentials(self) -> None:
|
||||||
|
delay = RECOVERY_MIN_DELAY
|
||||||
|
while True:
|
||||||
|
tried = self._client_key
|
||||||
|
key = credentials.reload_api_key()
|
||||||
|
if not key or key == tried:
|
||||||
|
fetched = await key_pickup.fetch_api_key()
|
||||||
|
if fetched and fetched != tried:
|
||||||
|
credentials.save_api_key(fetched)
|
||||||
|
key = fetched
|
||||||
|
if key and key != tried:
|
||||||
|
logger.info("MQTT credential recovery: new API key in hand — reconnecting with it.")
|
||||||
|
await self._rebuild_client()
|
||||||
|
return
|
||||||
|
if self._connected:
|
||||||
|
return
|
||||||
|
if key and key == tried:
|
||||||
|
logger.error(
|
||||||
|
"MQTT credential recovery: C2 returns the same key the broker refuses. The broker's login for "
|
||||||
|
"this node is out of sync with C2 (dynsec) — an admin re-approve re-syncs it without rotating."
|
||||||
|
)
|
||||||
|
logger.info(f"MQTT credential recovery: retrying in {delay}s.")
|
||||||
|
await asyncio.sleep(delay)
|
||||||
|
delay = min(delay * 2, RECOVERY_MAX_DELAY)
|
||||||
|
|
||||||
|
async def _rebuild_client(self) -> None:
|
||||||
|
"""Swap the paho client for one built from the current credentials."""
|
||||||
|
old = self._client
|
||||||
|
if self._connect_task and not self._connect_task.done():
|
||||||
|
self._connect_task.cancel()
|
||||||
|
if old is not None:
|
||||||
|
# loop_stop() joins paho's thread, which can be mid-backoff — off the event loop.
|
||||||
|
await asyncio.to_thread(old.loop_stop)
|
||||||
|
try:
|
||||||
|
old.disconnect()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self._connected = False
|
||||||
|
self._client = self._build_client()
|
||||||
|
self._connect_task = asyncio.create_task(self._connect_with_retry())
|
||||||
|
|
||||||
|
def update_api_key(self, key: str) -> None:
|
||||||
|
"""A key arrived while connected (legacy MQTT push / reissue): use it on
|
||||||
|
the next reconnect without dropping the current, working session."""
|
||||||
|
if self._client is not None and key != self._client_key:
|
||||||
|
self._client.username_pw_set(settings.node_id, key)
|
||||||
|
self._client_key = key
|
||||||
|
|
||||||
async def _connect_with_retry(self):
|
async def _connect_with_retry(self):
|
||||||
"""Attempt the initial TCP connect, retrying with backoff until it succeeds."""
|
"""Attempt the initial TCP connect, retrying with backoff until it succeeds."""
|
||||||
@@ -161,6 +251,8 @@ class MQTTManager:
|
|||||||
delay = min(delay * 2, 60)
|
delay = min(delay * 2, 60)
|
||||||
|
|
||||||
async def disconnect(self):
|
async def disconnect(self):
|
||||||
|
if self._recovery_task:
|
||||||
|
self._recovery_task.cancel()
|
||||||
if self._connect_task:
|
if self._connect_task:
|
||||||
self._connect_task.cancel()
|
self._connect_task.cancel()
|
||||||
if self._client:
|
if self._client:
|
||||||
@@ -235,6 +327,11 @@ class MQTTManager:
|
|||||||
"secondary_sdr_priority": config.secondary_sdr_priority,
|
"secondary_sdr_priority": config.secondary_sdr_priority,
|
||||||
}
|
}
|
||||||
payload["sdr_pins"] = config.sdr_pins
|
payload["sdr_pins"] = config.sdr_pins
|
||||||
|
# Conventional channels (node-26#24): name/freq/NAC and whether the
|
||||||
|
# agency encrypts them — the one decode failure nobody can fix on the
|
||||||
|
# node, so C2 has to be told. Empty list on a trunked system.
|
||||||
|
from app.internal.metadata_watcher import metadata_watcher
|
||||||
|
payload["radio_channels"] = metadata_watcher.conventional_channels
|
||||||
secondary = await secondary_sdr_client.status()
|
secondary = await secondary_sdr_client.status()
|
||||||
# Unreachable secondary-sdr: say so with explicit nulls. C2 only
|
# Unreachable secondary-sdr: say so with explicit nulls. C2 only
|
||||||
# overwrites keys present in the checkin, so omitting them would leave
|
# overwrites keys present in the checkin, so omitting them would leave
|
||||||
|
|||||||
@@ -0,0 +1,129 @@
|
|||||||
|
"""
|
||||||
|
What kind of radio system a SystemConfig describes, and the op25-container
|
||||||
|
generate-config payload for it.
|
||||||
|
|
||||||
|
C2 system documents (the `config` blob is passed through untouched):
|
||||||
|
|
||||||
|
P25 trunked type "P25", config.mode absent or "trunked",
|
||||||
|
config.control_channels = [MHz, ...]
|
||||||
|
P25 conventional type "P25", config.mode "conventional",
|
||||||
|
config.channels = [{freq: MHz, name, nac?}, ...] (node-26#24)
|
||||||
|
Analog type "NBFM", config.channels = [{freq, name}] — or, from
|
||||||
|
before #24, config.control_channels[0]
|
||||||
|
|
||||||
|
A missing mode is trunked, so every system saved before #24 behaves exactly as
|
||||||
|
it did. Pure functions; main._generate_op25_config does the I/O.
|
||||||
|
"""
|
||||||
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
|
TRUNKED, CONVENTIONAL, ANALOG, DMR = "trunked", "conventional", "analog", "dmr"
|
||||||
|
|
||||||
|
# Mirrors op25-container's MAX_CONVENTIONAL_CHANNELS: one audio.py carries two.
|
||||||
|
MAX_CONVENTIONAL_CHANNELS = 2
|
||||||
|
|
||||||
|
|
||||||
|
def to_hz(freq: Any) -> int:
|
||||||
|
"""MHz floats (< 1e6) or Hz ints -> Hz."""
|
||||||
|
f = float(freq)
|
||||||
|
return int(round(f * 1_000_000)) if f < 1_000_000 else int(f)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_nac(value: Any) -> Optional[int]:
|
||||||
|
"""NAC as typed by an admin ("293", "0x293") or stored as an int. Hex, like
|
||||||
|
RadioReference prints it. None when blank; ValueError when nonsense."""
|
||||||
|
if value is None or (isinstance(value, str) and not value.strip()):
|
||||||
|
return None
|
||||||
|
if isinstance(value, bool):
|
||||||
|
raise ValueError(f"NAC {value!r} isn't a number.")
|
||||||
|
if isinstance(value, int):
|
||||||
|
nac = value
|
||||||
|
else:
|
||||||
|
text = str(value).strip().lower()
|
||||||
|
text = text[2:] if text.startswith("0x") else text
|
||||||
|
try:
|
||||||
|
nac = int(text, 16)
|
||||||
|
except ValueError:
|
||||||
|
raise ValueError(f"NAC {value!r} isn't hex (e.g. 293 or 0x293).")
|
||||||
|
if not 0 < nac <= 0xFFF:
|
||||||
|
raise ValueError(f"NAC {value!r} is outside 001-FFF.")
|
||||||
|
return nac
|
||||||
|
|
||||||
|
|
||||||
|
def system_mode(system_type: str, config: Dict[str, Any]) -> str:
|
||||||
|
t = str(system_type or "").upper()
|
||||||
|
if t in ("NBFM", "ANALOG"):
|
||||||
|
return ANALOG
|
||||||
|
if t == "DMR":
|
||||||
|
return DMR
|
||||||
|
mode = str((config or {}).get("mode") or TRUNKED).lower()
|
||||||
|
return CONVENTIONAL if mode == CONVENTIONAL else TRUNKED
|
||||||
|
|
||||||
|
|
||||||
|
def system_kind(system_type: str, config: Dict[str, Any]) -> str:
|
||||||
|
"""Label for dashboards: "P25 trunked", "P25 conventional", "Analog FM", "DMR"."""
|
||||||
|
return {
|
||||||
|
TRUNKED: "P25 trunked", CONVENTIONAL: "P25 conventional", ANALOG: "Analog FM", DMR: "DMR",
|
||||||
|
}[system_mode(system_type, config)]
|
||||||
|
|
||||||
|
|
||||||
|
def conventional_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||||
|
"""config.channels -> [{frequency: Hz, name, nac}] (ValueError if malformed)."""
|
||||||
|
out = []
|
||||||
|
for raw in (config or {}).get("channels") or []:
|
||||||
|
if not isinstance(raw, dict) or raw.get("freq") in (None, ""):
|
||||||
|
raise ValueError("Every conventional channel needs a frequency.")
|
||||||
|
out.append({
|
||||||
|
"frequency": to_hz(raw["freq"]),
|
||||||
|
"name": str(raw.get("name") or "").strip(),
|
||||||
|
"nac": parse_nac(raw.get("nac")),
|
||||||
|
})
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def analog_frequency(config: Dict[str, Any]) -> Optional[int]:
|
||||||
|
chans = (config or {}).get("channels") or []
|
||||||
|
if chans and isinstance(chans[0], dict) and chans[0].get("freq") not in (None, ""):
|
||||||
|
return to_hz(chans[0]["freq"])
|
||||||
|
legacy = (config or {}).get("control_channels") or []
|
||||||
|
return to_hz(legacy[0]) if legacy else None
|
||||||
|
|
||||||
|
|
||||||
|
def build_generate_payload(
|
||||||
|
system_type: str,
|
||||||
|
name: str,
|
||||||
|
config: Dict[str, Any],
|
||||||
|
icecast: Dict[str, Any],
|
||||||
|
hardware_preset: str,
|
||||||
|
ppm_override: Optional[float],
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""SystemConfig fields -> op25-container /generate-config body."""
|
||||||
|
mode = system_mode(system_type, config)
|
||||||
|
payload: Dict[str, Any] = {
|
||||||
|
"type": system_type,
|
||||||
|
"systemName": name,
|
||||||
|
"icecastConfig": icecast,
|
||||||
|
"hardware_preset": hardware_preset,
|
||||||
|
}
|
||||||
|
if ppm_override is not None:
|
||||||
|
payload["ppm_override"] = ppm_override
|
||||||
|
|
||||||
|
if mode == CONVENTIONAL:
|
||||||
|
payload["type"] = "P25"
|
||||||
|
payload["p25_mode"] = CONVENTIONAL
|
||||||
|
payload["channels"] = []
|
||||||
|
payload["conv_channels"] = conventional_channels(config)
|
||||||
|
return payload
|
||||||
|
|
||||||
|
if mode == ANALOG:
|
||||||
|
freq = analog_frequency(config)
|
||||||
|
payload["channels"] = [freq] if freq else []
|
||||||
|
return payload
|
||||||
|
|
||||||
|
talkgroups = (config or {}).get("talkgroups", []) or []
|
||||||
|
payload["channels"] = [to_hz(ch) for ch in (config or {}).get("control_channels", []) or []]
|
||||||
|
payload["tags"] = [
|
||||||
|
{"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])}
|
||||||
|
for tg in talkgroups if tg.get("id") is not None
|
||||||
|
]
|
||||||
|
payload["whitelist"] = [int(tg["id"]) for tg in talkgroups if tg.get("id") is not None]
|
||||||
|
return payload
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
"""
|
||||||
|
Hand an MQTT `node_update` command to the host-level updater (node-26#15).
|
||||||
|
|
||||||
|
The edge-node container has no business updating its own host: it writes a
|
||||||
|
sentinel file into /configs (bind-mounted from <install dir>/configs), and the
|
||||||
|
host's systemd drb-update.path unit — installed by install.sh — triggers
|
||||||
|
drb-update.service, which runs Client/scripts/drb-update.sh. The host treats
|
||||||
|
the file as a bare "please update" flag and never reads its content as input,
|
||||||
|
so nothing C2 sends can pick the ref, image or command that runs.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal.logger import logger
|
||||||
|
|
||||||
|
SENTINEL_NAME = "update-requested"
|
||||||
|
|
||||||
|
|
||||||
|
def sentinel_path() -> Path:
|
||||||
|
return Path(settings.config_path) / SENTINEL_NAME
|
||||||
|
|
||||||
|
|
||||||
|
def request_host_update() -> bool:
|
||||||
|
"""Drop the sentinel. Returns False if it could not be written."""
|
||||||
|
path = sentinel_path()
|
||||||
|
try:
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
tmp = path.with_name(SENTINEL_NAME + ".tmp")
|
||||||
|
tmp.write_text(json.dumps({"requested_at": datetime.now(timezone.utc).isoformat()}))
|
||||||
|
os.replace(tmp, path)
|
||||||
|
except OSError as e:
|
||||||
|
logger.error(f"Node update requested but the sentinel could not be written ({e}).")
|
||||||
|
return False
|
||||||
|
logger.info(
|
||||||
|
f"Node update requested — handed to the host updater via {path}. If nothing happens within a minute, "
|
||||||
|
"this node predates the updater: re-run install.sh on it once."
|
||||||
|
)
|
||||||
|
return True
|
||||||
+26
-33
@@ -170,17 +170,15 @@ async def on_command(payload: dict):
|
|||||||
await asyncio.sleep(2)
|
await asyncio.sleep(2)
|
||||||
await op25_client.start()
|
await op25_client.start()
|
||||||
elif action == "node_update":
|
elif action == "node_update":
|
||||||
# TODO: Full OTA update — register a host-level systemd service (e.g. drb-update.service)
|
# node-26#15: the container cannot (and must not) update itself. It
|
||||||
# that stops all DRB containers, runs `docker compose pull`, then `docker compose up -d`.
|
# drops a sentinel in /configs; the host's drb-update.path unit
|
||||||
# The C2 server triggers it by sending this MQTT command; the host service watches for the
|
# (installed by install.sh) sees it and runs drb-update.service, which
|
||||||
# restart signal (e.g. via a Unix socket, a sentinel file, or a lightweight webhook).
|
# pulls the node's own pinned repo ref + images and recreates the
|
||||||
# Not implemented yet — for now, just restart the container so any pre-pulled image
|
# stack. The sentinel's content is ignored by the host — C2 cannot
|
||||||
# is picked up (requires a prior `docker compose pull` on the host).
|
# choose what gets installed. Offline/locked-out nodes get the same
|
||||||
logger.info("Node update requested — restarting container to pick up latest image.")
|
# request over HTTPS via the updater's timer instead.
|
||||||
await mqtt_manager.publish_status("offline")
|
from app.internal.update_request import request_host_update
|
||||||
await asyncio.sleep(1)
|
request_host_update()
|
||||||
import os
|
|
||||||
os._exit(0) # Docker restart=unless-stopped will bring the container back up
|
|
||||||
else:
|
else:
|
||||||
logger.warning(f"Unknown command: {action}")
|
logger.warning(f"Unknown command: {action}")
|
||||||
|
|
||||||
@@ -189,39 +187,34 @@ async def on_api_key(payload: dict):
|
|||||||
key = payload.get("api_key")
|
key = payload.get("api_key")
|
||||||
if key:
|
if key:
|
||||||
credentials.save_api_key(key)
|
credentials.save_api_key(key)
|
||||||
|
mqtt_manager.update_api_key(key)
|
||||||
logger.info("Node API key received and saved.")
|
logger.info("Node API key received and saved.")
|
||||||
|
|
||||||
|
|
||||||
def _to_hz(freq) -> int:
|
|
||||||
"""Convert a frequency to Hz. Accepts MHz floats (< 1e6) or Hz ints."""
|
|
||||||
f = float(freq)
|
|
||||||
return int(f * 1_000_000) if f < 1_000_000 else int(f)
|
|
||||||
|
|
||||||
|
|
||||||
async def _generate_op25_config(config: SystemConfig) -> bool:
|
async def _generate_op25_config(config: SystemConfig) -> bool:
|
||||||
"""Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq."""
|
"""Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq.
|
||||||
|
|
||||||
|
Trunked / conventional / analog are told apart in radio_config (node-26#24);
|
||||||
|
a system saved before that has no `mode` and is trunked, as it always was.
|
||||||
|
"""
|
||||||
from app.internal.op25_client import op25_client
|
from app.internal.op25_client import op25_client
|
||||||
|
from app.internal.radio_config import build_generate_payload
|
||||||
node_cfg = load_node_config()
|
node_cfg = load_node_config()
|
||||||
raw = config.config
|
try:
|
||||||
payload = {
|
payload = build_generate_payload(
|
||||||
"type": config.type,
|
config.type, config.name, config.config,
|
||||||
"systemName": config.name,
|
icecast={
|
||||||
"channels": [_to_hz(ch) for ch in raw.get("control_channels", [])],
|
|
||||||
"tags": [
|
|
||||||
{"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])}
|
|
||||||
for tg in raw.get("talkgroups", [])
|
|
||||||
if tg.get("id") is not None
|
|
||||||
],
|
|
||||||
"whitelist": [int(tg["id"]) for tg in raw.get("talkgroups", []) if tg.get("id") is not None],
|
|
||||||
"icecastConfig": {
|
|
||||||
"icecast_host": settings.icecast_host,
|
"icecast_host": settings.icecast_host,
|
||||||
"icecast_port": settings.icecast_port,
|
"icecast_port": settings.icecast_port,
|
||||||
"icecast_mountpoint": settings.icecast_mount,
|
"icecast_mountpoint": settings.icecast_mount,
|
||||||
"icecast_password": settings.icecast_source_password,
|
"icecast_password": settings.icecast_source_password,
|
||||||
},
|
},
|
||||||
"hardware_preset": node_cfg.hardware_preset,
|
hardware_preset=node_cfg.hardware_preset,
|
||||||
**({"ppm_override": node_cfg.ppm_override} if node_cfg.ppm_override is not None else {}),
|
ppm_override=node_cfg.ppm_override,
|
||||||
}
|
)
|
||||||
|
except ValueError as e:
|
||||||
|
logger.error(f"System {config.name!r} has an unusable channel list: {e}")
|
||||||
|
return False
|
||||||
return await op25_client.generate_config(payload)
|
return await op25_client.generate_config(payload)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -37,9 +37,12 @@ async def get_status():
|
|||||||
else node_cfg.system_config
|
else node_cfg.system_config
|
||||||
)
|
)
|
||||||
system_type = None
|
system_type = None
|
||||||
|
system_kind = None
|
||||||
if active_config:
|
if active_config:
|
||||||
|
from app.internal.radio_config import system_kind as _kind
|
||||||
system_name = active_config.name
|
system_name = active_config.name
|
||||||
system_type = active_config.type
|
system_type = active_config.type
|
||||||
|
system_kind = _kind(active_config.type, active_config.config)
|
||||||
if active_tgid:
|
if active_tgid:
|
||||||
# Reverse engineer the TGID to talkgroup name based on active system object
|
# Reverse engineer the TGID to talkgroup name based on active system object
|
||||||
tgs = active_config.config.get("talkgroups", [])
|
tgs = active_config.config.get("talkgroups", [])
|
||||||
@@ -61,6 +64,8 @@ async def get_status():
|
|||||||
"assigned_system_id": node_cfg.assigned_system_id,
|
"assigned_system_id": node_cfg.assigned_system_id,
|
||||||
"system_name": system_name,
|
"system_name": system_name,
|
||||||
"system_type": system_type,
|
"system_type": system_type,
|
||||||
|
# Human label that tells P25 trunked from P25 conventional (node-26#24).
|
||||||
|
"system_kind": system_kind,
|
||||||
# True when what OP25 is tuned to was chosen on this dashboard rather
|
# True when what OP25 is tuned to was chosen on this dashboard rather
|
||||||
# than assigned by C2 (a local override, revertible to C2's system).
|
# than assigned by C2 (a local override, revertible to C2's system).
|
||||||
"local_choice": bool(node_cfg.override_system_id and node_cfg.override_config),
|
"local_choice": bool(node_cfg.override_system_id and node_cfg.override_config),
|
||||||
@@ -192,11 +197,14 @@ class Talkgroup(BaseModel):
|
|||||||
|
|
||||||
class ListenBody(BaseModel):
|
class ListenBody(BaseModel):
|
||||||
"""What to listen to, chosen on the local dashboard."""
|
"""What to listen to, chosen on the local dashboard."""
|
||||||
mode: Literal["system", "p25", "analog"]
|
mode: Literal["system", "p25", "p25conv", "analog"]
|
||||||
system_id: Optional[str] = None # mode=system: a cached C2 system
|
system_id: Optional[str] = None # mode=system: a cached C2 system
|
||||||
name: Optional[str] = None
|
name: Optional[str] = None
|
||||||
frequencies: List[float] = [] # MHz; p25 = control channels, analog = exactly one
|
# MHz. p25 = control channels; p25conv = 1-2 voice channels (node-26#24);
|
||||||
|
# analog = exactly one.
|
||||||
|
frequencies: List[float] = []
|
||||||
talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup
|
talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup
|
||||||
|
nac: Optional[str] = None # p25conv only, hex; empty = any NAC
|
||||||
|
|
||||||
|
|
||||||
# RTL-SDR (R820T/R828D) tuning range.
|
# RTL-SDR (R820T/R828D) tuning range.
|
||||||
@@ -225,8 +233,11 @@ async def listen(body: ListenBody):
|
|||||||
)
|
)
|
||||||
if body.mode == "analog" and len(freqs) != 1:
|
if body.mode == "analog" and len(freqs) != 1:
|
||||||
raise HTTPException(400, "Analog listens to exactly one frequency.")
|
raise HTTPException(400, "Analog listens to exactly one frequency.")
|
||||||
is_p25 = body.mode == "p25"
|
|
||||||
first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".")
|
first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".")
|
||||||
|
if body.mode == "p25conv":
|
||||||
|
config = _conventional_listen_config(body, freqs, first)
|
||||||
|
else:
|
||||||
|
is_p25 = body.mode == "p25"
|
||||||
label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
|
label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
|
||||||
config = SystemConfig(
|
config = SystemConfig(
|
||||||
system_id=f"local-{body.mode}",
|
system_id=f"local-{body.mode}",
|
||||||
@@ -241,6 +252,43 @@ async def listen(body: ListenBody):
|
|||||||
return {"ok": True, "name": config.name}
|
return {"ok": True, "name": config.name}
|
||||||
|
|
||||||
|
|
||||||
|
# One SDR's usable passband, less a channel's width — op25-container makes the
|
||||||
|
# exact check; this one just turns the common mistake into a readable error.
|
||||||
|
_CONVENTIONAL_MAX_SPAN_MHZ = 1.5
|
||||||
|
|
||||||
|
|
||||||
|
def _conventional_listen_config(body: ListenBody, freqs: List[float], first: str) -> SystemConfig:
|
||||||
|
from app.internal.radio_config import MAX_CONVENTIONAL_CHANNELS, parse_nac
|
||||||
|
if len(freqs) > MAX_CONVENTIONAL_CHANNELS:
|
||||||
|
raise HTTPException(400, f"P25 conventional listens to at most {MAX_CONVENTIONAL_CHANNELS} channels.")
|
||||||
|
if len(set(freqs)) != len(freqs):
|
||||||
|
raise HTTPException(400, "The same frequency is entered twice.")
|
||||||
|
if max(freqs) - min(freqs) > _CONVENTIONAL_MAX_SPAN_MHZ:
|
||||||
|
raise HTTPException(
|
||||||
|
400, f"Those channels are {max(freqs) - min(freqs):.3f} MHz apart — one SDR covers about "
|
||||||
|
f"{_CONVENTIONAL_MAX_SPAN_MHZ:g} MHz. Pick channels closer together.",
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
nac = parse_nac(body.nac)
|
||||||
|
except ValueError as e:
|
||||||
|
raise HTTPException(400, str(e))
|
||||||
|
label = (body.name or "").strip() or f"P25 conventional {first} MHz"
|
||||||
|
return SystemConfig(
|
||||||
|
system_id="local-p25conv",
|
||||||
|
name=label,
|
||||||
|
type="P25",
|
||||||
|
config={
|
||||||
|
"mode": "conventional",
|
||||||
|
"channels": [
|
||||||
|
{"freq": f, "name": f"{label} {f:.4f}" if len(freqs) > 1 else label,
|
||||||
|
**({"nac": f"{nac:03X}"} if nac is not None else {})}
|
||||||
|
for f in freqs
|
||||||
|
],
|
||||||
|
"talkgroups": [],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@router.get("/systems")
|
@router.get("/systems")
|
||||||
async def get_systems():
|
async def get_systems():
|
||||||
from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems
|
from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems
|
||||||
|
|||||||
@@ -167,6 +167,7 @@
|
|||||||
|
|
||||||
<div class="tabs" role="radiogroup" aria-label="Source type">
|
<div class="tabs" role="radiogroup" aria-label="Source type">
|
||||||
<label><input type="radio" name="mode" value="p25" checked> P25 trunked</label>
|
<label><input type="radio" name="mode" value="p25" checked> P25 trunked</label>
|
||||||
|
<label><input type="radio" name="mode" value="p25conv"> P25 conventional</label>
|
||||||
<label><input type="radio" name="mode" value="analog"> Analog FM</label>
|
<label><input type="radio" name="mode" value="analog"> Analog FM</label>
|
||||||
<label><input type="radio" name="mode" value="system" id="mode-system"> Server's systems</label>
|
<label><input type="radio" name="mode" value="system" id="mode-system"> Server's systems</label>
|
||||||
</div>
|
</div>
|
||||||
@@ -188,6 +189,23 @@
|
|||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<div class="pane" data-pane="p25conv" hidden>
|
||||||
|
<div class="field">
|
||||||
|
<label for="pc-freqs">Channel frequencies (MHz)</label>
|
||||||
|
<input type="text" id="pc-freqs" class="mono" placeholder="460.4000, 460.4250" inputmode="decimal">
|
||||||
|
<span class="hint">One or two voice channels, comma-separated, within about 1.5 MHz of each other. No control channel — RadioReference lists these as conventional frequencies with mode P25 (or "DE").</span>
|
||||||
|
</div>
|
||||||
|
<div class="field">
|
||||||
|
<label for="pc-name">Name <span class="hint">(optional)</span></label>
|
||||||
|
<input type="text" id="pc-name" placeholder="City PD">
|
||||||
|
</div>
|
||||||
|
<div class="field">
|
||||||
|
<label for="pc-nac">NAC <span class="hint">(optional, hex)</span></label>
|
||||||
|
<input type="text" id="pc-nac" class="mono" placeholder="293">
|
||||||
|
<span class="hint">RadioReference's NAC for these channels. Leave empty to accept any.</span>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
<div class="pane" data-pane="analog" hidden>
|
<div class="pane" data-pane="analog" hidden>
|
||||||
<div class="field">
|
<div class="field">
|
||||||
<label for="an-freq">Frequency (MHz)</label>
|
<label for="an-freq">Frequency (MHz)</label>
|
||||||
@@ -307,7 +325,7 @@
|
|||||||
$('node-name').textContent = d.node_name || 'DRB Edge Node';
|
$('node-name').textContent = d.node_name || 'DRB Edge Node';
|
||||||
$('node-id').textContent = d.node_id || '';
|
$('node-id').textContent = d.node_id || '';
|
||||||
document.title = `${d.node_name || 'DRB Edge Node'} — DRB`;
|
document.title = `${d.node_name || 'DRB Edge Node'} — DRB`;
|
||||||
$('now-system').textContent = d.system_name ? `${d.system_name} (${d.system_type})` : 'Nothing chosen yet';
|
$('now-system').textContent = d.system_name ? `${d.system_name} (${d.system_kind || d.system_type})` : 'Nothing chosen yet';
|
||||||
$('now-tg').innerHTML = d.active_tgid
|
$('now-tg').innerHTML = d.active_tgid
|
||||||
? `<span class="live">${esc(d.active_tgid_name || 'Talkgroup')}</span> <span class="mono">TG ${esc(d.active_tgid)}</span>`
|
? `<span class="live">${esc(d.active_tgid_name || 'Talkgroup')}</span> <span class="mono">TG ${esc(d.active_tgid)}</span>`
|
||||||
: 'No call right now';
|
: 'No call right now';
|
||||||
@@ -335,7 +353,7 @@
|
|||||||
actions.hidden = true;
|
actions.hidden = true;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
cur.innerHTML = `Now: <strong>${esc(d.system_name)}</strong> (${esc(d.system_type)}) — ` +
|
cur.innerHTML = `Now: <strong>${esc(d.system_name)}</strong> (${esc(d.system_kind || d.system_type)}) — ` +
|
||||||
(d.local_choice ? 'chosen on this dashboard.' : 'assigned by the DRB server.');
|
(d.local_choice ? 'chosen on this dashboard.' : 'assigned by the DRB server.');
|
||||||
actions.hidden = !d.local_choice;
|
actions.hidden = !d.local_choice;
|
||||||
$('revert-btn').textContent = d.has_c2_system ? "Go back to the server's system" : 'Clear and stop';
|
$('revert-btn').textContent = d.has_c2_system ? "Go back to the server's system" : 'Clear and stop';
|
||||||
@@ -387,7 +405,9 @@
|
|||||||
try { systems = await api('/api/systems'); } catch (e) { systems = []; }
|
try { systems = await api('/api/systems'); } catch (e) { systems = []; }
|
||||||
const pick = $('sys-pick');
|
const pick = $('sys-pick');
|
||||||
const usable = systems.filter(s => String(s.type).toUpperCase() !== 'DMR');
|
const usable = systems.filter(s => String(s.type).toUpperCase() !== 'DMR');
|
||||||
pick.innerHTML = usable.map(s => `<option value="${esc(s.system_id)}">${esc(s.name)} (${esc(s.type)})</option>`).join('');
|
const kind = s => String(s.type).toUpperCase() === 'P25' && s.config && s.config.mode === 'conventional'
|
||||||
|
? 'P25 conventional' : s.type;
|
||||||
|
pick.innerHTML = usable.map(s => `<option value="${esc(s.system_id)}">${esc(s.name)} (${esc(kind(s))})</option>`).join('');
|
||||||
const none = usable.length === 0;
|
const none = usable.length === 0;
|
||||||
$('mode-system').disabled = none;
|
$('mode-system').disabled = none;
|
||||||
pick.disabled = none;
|
pick.disabled = none;
|
||||||
@@ -416,6 +436,10 @@
|
|||||||
tgs.push({ id: Number(mm[1]), name: mm[2] });
|
tgs.push({ id: Number(mm[1]), name: mm[2] });
|
||||||
}
|
}
|
||||||
body = { mode: 'p25', frequencies: freqs, name: $('p25-name').value, talkgroups: tgs };
|
body = { mode: 'p25', frequencies: freqs, name: $('p25-name').value, talkgroups: tgs };
|
||||||
|
} else if (m === 'p25conv') {
|
||||||
|
const freqs = parseFreqs($('pc-freqs').value);
|
||||||
|
if (!freqs.length || freqs.length > 2 || freqs.some(isNaN)) { msg.className = 'msg err'; msg.textContent = 'Enter one or two channel frequencies in MHz, e.g. 460.4000, 460.4250'; return; }
|
||||||
|
body = { mode: 'p25conv', frequencies: freqs, name: $('pc-name').value, nac: $('pc-nac').value.trim() || null };
|
||||||
} else {
|
} else {
|
||||||
const freqs = parseFreqs($('an-freq').value);
|
const freqs = parseFreqs($('an-freq').value);
|
||||||
if (freqs.length !== 1 || isNaN(freqs[0])) { msg.className = 'msg err'; msg.textContent = 'Enter one frequency in MHz, e.g. 154.4300'; return; }
|
if (freqs.length !== 1 || isNaN(freqs[0])) { msg.className = 'msg err'; msg.textContent = 'Enter one frequency in MHz, e.g. 154.4300'; return; }
|
||||||
|
|||||||
@@ -0,0 +1,402 @@
|
|||||||
|
"""
|
||||||
|
Conventional channels: P25 conventional (node-26#24) and analog (node-26#20).
|
||||||
|
|
||||||
|
Covers the generate-config payload (radio_config), the console normalizer
|
||||||
|
(conventional.py) and how the watcher attributes conventional audio. Channel
|
||||||
|
rows are shaped like tk_drb_conv's / tk_p25's conventional `channel_update`
|
||||||
|
entries; grants like tk_p25.log_call entries.
|
||||||
|
"""
|
||||||
|
import pytest
|
||||||
|
from unittest.mock import AsyncMock
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal import radio_config as rc
|
||||||
|
from app.internal.conventional import ConventionalChannels, channel_tgid, NON_DISTINCT_TGIDS
|
||||||
|
from app.internal.metadata_watcher import MetadataWatcher
|
||||||
|
from tests.test_metadata_watcher import FakeAudio, FakeClock, tick, update
|
||||||
|
|
||||||
|
PD1, PD3 = 460_400_000, 460_425_000
|
||||||
|
|
||||||
|
|
||||||
|
def conv_row(msgqid=0, freq=PD1, name="PD1 Dispatch", tgid=None, srcaddr=0, voice=False,
|
||||||
|
encrypted=0, **extra):
|
||||||
|
row = {
|
||||||
|
"freq": freq, "name": name, "tgid": tgid, "tag": name, "srcaddr": srcaddr,
|
||||||
|
"encrypted": encrypted, "conventional": True, "msgqid": msgqid, "voice": voice,
|
||||||
|
}
|
||||||
|
row.update(extra)
|
||||||
|
return row
|
||||||
|
|
||||||
|
|
||||||
|
def conv_grant(rcvr, tgid, at, freq=PD1, rid=4242):
|
||||||
|
return {"time": at, "sysid": 0, "rcvr": rcvr, "rcvrtag": "", "freq": freq, "slot": 0,
|
||||||
|
"prio": 0, "tgid": tgid, "tgtag": "", "rid": rid, "rtag": ""}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# radio_config: system document -> op25 generate-config payload
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
ICE = {"icecast_host": "localhost", "icecast_port": 8000, "icecast_mountpoint": "/n", "icecast_password": "p"}
|
||||||
|
|
||||||
|
WHITE_PLAINS = {
|
||||||
|
"mode": "conventional",
|
||||||
|
"channels": [
|
||||||
|
{"freq": 460.4, "name": "PD1 Dispatch"},
|
||||||
|
{"freq": 460.425, "name": "PD3 Tac", "nac": "293"},
|
||||||
|
],
|
||||||
|
"talkgroups": [],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_payload():
|
||||||
|
p = rc.build_generate_payload("P25", "White Plains", WHITE_PLAINS, ICE, "rtl-sdr-v3", None)
|
||||||
|
assert p["type"] == "P25" and p["p25_mode"] == "conventional"
|
||||||
|
assert p["channels"] == [] # nothing that could become a control channel
|
||||||
|
assert p["conv_channels"] == [
|
||||||
|
{"frequency": PD1, "name": "PD1 Dispatch", "nac": None},
|
||||||
|
{"frequency": PD3, "name": "PD3 Tac", "nac": 0x293},
|
||||||
|
]
|
||||||
|
assert "ppm_override" not in p
|
||||||
|
|
||||||
|
|
||||||
|
def test_trunked_payload_unchanged_when_mode_missing():
|
||||||
|
cfg = {"control_channels": [851.0125, 851.5125], "talkgroups": [{"id": 9048, "name": "Ossining PD"}]}
|
||||||
|
p = rc.build_generate_payload("P25", "County", cfg, ICE, "other", 1.5)
|
||||||
|
assert "p25_mode" not in p
|
||||||
|
assert p["channels"] == [851_012_500, 851_512_500]
|
||||||
|
assert p["tags"] == [{"talkgroup": "Ossining PD", "tagDec": 9048}]
|
||||||
|
assert p["whitelist"] == [9048]
|
||||||
|
assert p["ppm_override"] == 1.5
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("cfg", [
|
||||||
|
{"channels": [{"freq": 154.43, "name": "Fire"}]},
|
||||||
|
{"control_channels": [154.43]}, # saved before #24
|
||||||
|
])
|
||||||
|
def test_analog_payload_reads_channels_or_legacy(cfg):
|
||||||
|
p = rc.build_generate_payload("NBFM", "Fire", cfg, ICE, "rtl-sdr-v3", None)
|
||||||
|
assert p["channels"] == [154_430_000]
|
||||||
|
assert "p25_mode" not in p
|
||||||
|
|
||||||
|
|
||||||
|
def test_system_mode():
|
||||||
|
assert rc.system_mode("P25", {}) == rc.TRUNKED
|
||||||
|
assert rc.system_mode("P25", {"mode": "Conventional"}) == rc.CONVENTIONAL
|
||||||
|
assert rc.system_mode("NBFM", {"mode": "conventional"}) == rc.ANALOG
|
||||||
|
assert rc.system_mode("DMR", {}) == rc.DMR
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("raw, want", [("293", 0x293), ("0x293", 0x293), (0x293, 0x293), ("", None), (None, None)])
|
||||||
|
def test_parse_nac(raw, want):
|
||||||
|
assert rc.parse_nac(raw) == want
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("raw", ["xyz", "1000", 0, True])
|
||||||
|
def test_parse_nac_rejects(raw):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
rc.parse_nac(raw)
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_channel_without_freq_rejected():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
rc.conventional_channels({"channels": [{"name": "no freq"}]})
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Normalizer
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_synthetic_tgid_is_frequency_in_100hz_and_clear_of_p25_space():
|
||||||
|
assert channel_tgid(PD1) == 4_604_000
|
||||||
|
assert channel_tgid(460.4) == 4_604_000
|
||||||
|
assert channel_tgid(460_406_250) != channel_tgid(460_412_500) # 6.25 kHz steps stay distinct
|
||||||
|
assert channel_tgid(24_000_000) > 0xFFFF # lowest RTL-SDR frequency
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_distinct_link_control_tgid_becomes_the_channel():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
up = update(channels=[conv_row(0, PD1, tgid=1, srcaddr=7, voice=True),
|
||||||
|
conv_row(1, PD3, "PD3 Tac", tgid=1, srcaddr=8, voice=True)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert [r["tgid"] for r in up.channels] == [4_604_000, 4_604_250]
|
||||||
|
assert [r["lc_tgid"] for r in up.channels] == [1, 1]
|
||||||
|
assert [r["tag"] for r in up.channels] == ["PD1 Dispatch", "PD3 Tac"]
|
||||||
|
assert all(r["drb_active"] for r in up.channels)
|
||||||
|
assert 1 in NON_DISTINCT_TGIDS and 0xFFFF in NON_DISTINCT_TGIDS
|
||||||
|
|
||||||
|
|
||||||
|
def test_distinct_link_control_tgid_is_kept():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
up = update(channels=[conv_row(0, tgid=101, srcaddr=7)],
|
||||||
|
call_log=[conv_grant(0, 101, 1.0)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0]["tgid"] == 101
|
||||||
|
assert up.call_log[0]["tgid"] == 101
|
||||||
|
assert up.call_log[0]["tgtag"] == "PD1 Dispatch"
|
||||||
|
|
||||||
|
|
||||||
|
def test_grants_are_mapped_per_receiver():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
up = update(channels=[conv_row(0, PD1), conv_row(1, PD3, "PD3 Tac")],
|
||||||
|
call_log=[conv_grant(1, 1, 1.0, freq=PD3)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.call_log[0]["tgid"] == 4_604_250
|
||||||
|
assert up.call_log[0]["tgtag"] == "PD3 Tac"
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_without_link_control_is_still_a_talker():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
up = update(channels=[conv_row(0, voice=True)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0]["tgid"] == 4_604_000 and up.channels[0]["drb_active"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_idle_and_encrypted_channels_are_not_talkers():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
up = update(channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert [r["tgid"] for r in up.channels] == [None, None]
|
||||||
|
|
||||||
|
|
||||||
|
def test_trunked_rows_untouched():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
row = {"freq": 851_000_000, "tgid": 9048, "srcaddr": 5, "tag": "PD", "msgqid": 0}
|
||||||
|
up = update(channels=[dict(row)], call_log=[conv_grant(0, 9048, 1.0)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0] == row
|
||||||
|
assert up.call_log[0]["tgid"] == 9048
|
||||||
|
assert conv.sole_channel() is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_sole_channel_only_with_exactly_one_conventional_channel():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
conv.normalize(update(channels=[conv_row(0, 154_430_000, "Fire")]))
|
||||||
|
assert conv.sole_channel() == {"tgid": 1_544_300, "name": "Fire", "freq": 154_430_000}
|
||||||
|
conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
|
||||||
|
assert conv.sole_channel() is None
|
||||||
|
# A poll without channel_update does not forget the channel list.
|
||||||
|
conv.normalize(update())
|
||||||
|
assert conv.sole_channel() is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_encryption_report():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
conv.normalize(update(channels=[
|
||||||
|
conv_row(0, enc_calls=3, clear_calls=0, last_enc_time=50.0, last_enc_algid=0x84,
|
||||||
|
last_enc_keyid=0x1234, nac=0x293),
|
||||||
|
conv_row(1, PD3, "PD3 Tac", enc_calls=1, clear_calls=5, last_enc_time=10.0, last_clear_time=20.0),
|
||||||
|
]))
|
||||||
|
pd1, pd3 = conv.encryption_report()
|
||||||
|
assert pd1["encrypted"] is True and pd1["algid"] == 0x84 and pd1["keyid"] == 0x1234 and pd1["nac"] == 0x293
|
||||||
|
assert pd3["encrypted"] is False # its most recent call was clear
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Watcher attribution
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def clock():
|
||||||
|
return FakeClock()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def audio(clock):
|
||||||
|
return FakeAudio(clock)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def hearing(clock, audio):
|
||||||
|
w = MetadataWatcher()
|
||||||
|
w._clock = clock
|
||||||
|
w.on_call_start = AsyncMock()
|
||||||
|
w.on_call_end = AsyncMock()
|
||||||
|
w.audio_activity = audio
|
||||||
|
return w
|
||||||
|
|
||||||
|
|
||||||
|
async def _finish(hearing, clock, channels):
|
||||||
|
clock.advance(settings.call_silence_timeout + 0.5)
|
||||||
|
await tick(hearing, update(channels=channels))
|
||||||
|
return hearing.on_call_end.call_args[0][0]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_conventional_call_attributed_to_its_channel(hearing, clock, audio):
|
||||||
|
"""White Plains: both channels carry the default TGID 1; the keyed one wins."""
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(
|
||||||
|
channels=[conv_row(0, tgid=1, srcaddr=4242, voice=True), conv_row(1, PD3, "PD3 Tac")],
|
||||||
|
call_log=[conv_grant(0, 1, clock.now)],
|
||||||
|
))
|
||||||
|
assert hearing.is_active and hearing.current_tgid == 4_604_000
|
||||||
|
|
||||||
|
ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
|
||||||
|
assert ended["attributed"] is True
|
||||||
|
assert (ended["tgid"], ended["tgid_name"], ended["srcaddr"]) == (4_604_000, "PD1 Dispatch", 4242)
|
||||||
|
assert hearing.unattributed_segments == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_other_channel_keying_splits(hearing, clock, audio):
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(
|
||||||
|
channels=[conv_row(0, tgid=1, voice=True), conv_row(1, PD3, "PD3 Tac")],
|
||||||
|
call_log=[conv_grant(0, 1, clock.now)],
|
||||||
|
))
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(
|
||||||
|
channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True)],
|
||||||
|
call_log=[conv_grant(1, 1, clock.now, freq=PD3)],
|
||||||
|
))
|
||||||
|
first = hearing.on_call_end.call_args[0][0]
|
||||||
|
assert first["tgid"] == 4_604_000 and first["end_reason"] == "tgid_change"
|
||||||
|
assert hearing.current_tgid == 4_604_250
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_single_analog_channel_audio_is_kept_not_orphaned(hearing, clock, audio):
|
||||||
|
"""node-26#20: analog never reports a talker; the only channel is the answer."""
|
||||||
|
audio.speak()
|
||||||
|
rows = [conv_row(0, 154_430_000, "County Fire")]
|
||||||
|
await tick(hearing, update(channels=rows))
|
||||||
|
# Not labelled at open — only at close.
|
||||||
|
assert hearing.on_call_start.call_args[0][0]["tgid"] is None
|
||||||
|
|
||||||
|
ended = await _finish(hearing, clock, rows)
|
||||||
|
assert ended["attributed"] is True
|
||||||
|
assert (ended["tgid"], ended["tgid_name"]) == (1_544_300, "County Fire")
|
||||||
|
assert hearing.unattributed_segments == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_two_channels_with_no_evidence_is_still_orphan(hearing, clock, audio):
|
||||||
|
audio.speak()
|
||||||
|
rows = [conv_row(0), conv_row(1, PD3, "PD3 Tac")]
|
||||||
|
await tick(hearing, update(channels=rows))
|
||||||
|
ended = await _finish(hearing, clock, rows)
|
||||||
|
assert ended["attributed"] is False
|
||||||
|
assert hearing.unattributed_segments == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_late_real_talkgroup_on_single_channel_does_not_split(hearing, clock, audio):
|
||||||
|
"""Audio first, link-control grant a poll later: adopt it, don't split."""
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0)]))
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
|
||||||
|
call_log=[conv_grant(0, 101, clock.now - 0.3)]))
|
||||||
|
assert hearing.current_tgid == 101
|
||||||
|
hearing.on_call_end.assert_not_called()
|
||||||
|
ended = await _finish(hearing, clock, [conv_row(0)])
|
||||||
|
assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_encrypted_channel_does_not_label_clear_audio(hearing, clock, audio):
|
||||||
|
"""PD3 is keyed encrypted (muted) while PD1 talks in the clear."""
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[
|
||||||
|
conv_row(0, tgid=1, voice=True),
|
||||||
|
conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1, algid=0x84),
|
||||||
|
]))
|
||||||
|
ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
|
||||||
|
assert ended["tgid"] == 4_604_000
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_conventional_channels_property(hearing, clock):
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, encrypted=1, voice=True, algid=0x84, keyid=7)]))
|
||||||
|
(only,) = hearing.conventional_channels
|
||||||
|
assert only["encrypted"] and only["encrypted_now"] and only["algid"] == 0x84
|
||||||
|
|
||||||
|
|
||||||
|
# drb-audio-review findings on #24: a header frame (voice, no link control yet)
|
||||||
|
# must not label the call with the synthetic id and then split against the
|
||||||
|
# real talkgroup a poll later; nor may a reply's header split it back.
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_header_before_link_control_does_not_split(hearing, clock, audio):
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, voice=True)])) # HDU only
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
|
||||||
|
call_log=[conv_grant(0, 101, clock.now - 0.2)]))
|
||||||
|
hearing.on_call_end.assert_not_called()
|
||||||
|
assert hearing.current_tgid == 101
|
||||||
|
ended = await _finish(hearing, clock, [conv_row(0)])
|
||||||
|
assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
|
||||||
|
assert hearing.on_call_end.call_count == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reply_header_inside_conversation_does_not_ping_pong(hearing, clock, audio):
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
|
||||||
|
call_log=[conv_grant(0, 101, clock.now)]))
|
||||||
|
clock.advance(0.5)
|
||||||
|
await tick(hearing, update(channels=[conv_row(0)])) # TDU
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, voice=True)])) # reply's HDU
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=12, voice=True)],
|
||||||
|
call_log=[conv_grant(0, 101, clock.now)]))
|
||||||
|
hearing.on_call_end.assert_not_called()
|
||||||
|
ended = await _finish(hearing, clock, [conv_row(0)])
|
||||||
|
assert ended["tgid"] == 101 and hearing.on_call_end.call_count == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_never_decoded_link_control_then_real_tg_relabels(hearing, clock, audio):
|
||||||
|
"""First-ever call on a channel: synthetic label, then the real TG adopts it."""
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, voice=True)]))
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, voice=True)]))
|
||||||
|
assert hearing.current_tgid == 4_604_000
|
||||||
|
clock.advance(0.5)
|
||||||
|
audio.speak()
|
||||||
|
await tick(hearing, update(channels=[conv_row(0, tgid=101, voice=True)],
|
||||||
|
call_log=[conv_grant(0, 101, clock.now)]))
|
||||||
|
hearing.on_call_end.assert_not_called()
|
||||||
|
assert hearing.current_tgid == 101
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_without_link_control_uses_channels_last_talkgroup():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
|
||||||
|
up = update(channels=[conv_row(0, voice=True)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0]["tgid"] == 101
|
||||||
|
|
||||||
|
|
||||||
|
def test_same_channel_relabel_is_conventional_only():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) is None # no conventional rows known
|
||||||
|
conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
|
||||||
|
assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) == "adopt"
|
||||||
|
assert conv.same_channel_relabel(101, PD1, 4_604_000, PD1) == "keep"
|
||||||
|
assert conv.same_channel_relabel(4_604_000, PD1, 4_604_250, PD3) is None # other channel: split
|
||||||
|
|
||||||
|
|
||||||
|
def test_remembered_talkgroup_expires():
|
||||||
|
conv = ConventionalChannels()
|
||||||
|
t = [1000.0]
|
||||||
|
conv._clock = lambda: t[0]
|
||||||
|
conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
|
||||||
|
t[0] += 10
|
||||||
|
up = update(channels=[conv_row(0, voice=True)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0]["tgid"] == 101
|
||||||
|
t[0] += 3600
|
||||||
|
up = update(channels=[conv_row(0, voice=True)])
|
||||||
|
conv.normalize(up)
|
||||||
|
assert up.channels[0]["tgid"] == 4_604_000
|
||||||
@@ -0,0 +1,205 @@
|
|||||||
|
"""
|
||||||
|
node-26#14 — a node with no key, or a stale key, recovers on its own:
|
||||||
|
re-read credentials.json, else fetch over HTTPS with the pickup secret, then
|
||||||
|
rebuild the MQTT client in place. node-26#15 — node_update hands off to the
|
||||||
|
host updater via a sentinel file.
|
||||||
|
"""
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
from unittest.mock import AsyncMock
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal import credentials, key_pickup, mqtt_manager as mm_mod, update_request
|
||||||
|
from app.internal.mqtt_manager import MQTTManager, _is_auth_failure
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def isolated(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
|
||||||
|
monkeypatch.setattr(credentials, "_PICKUP_FILE", tmp_path / "pickup_secret")
|
||||||
|
monkeypatch.setattr(credentials, "_api_key", None)
|
||||||
|
monkeypatch.setattr(settings, "config_path", str(tmp_path))
|
||||||
|
monkeypatch.setattr(settings, "c2_url", "https://c2.example")
|
||||||
|
monkeypatch.setattr(settings, "mqtt_user", None)
|
||||||
|
monkeypatch.setattr(mm_mod, "RECOVERY_MIN_DELAY", 0)
|
||||||
|
yield tmp_path
|
||||||
|
|
||||||
|
|
||||||
|
def _manager(client_key=None):
|
||||||
|
m = MQTTManager()
|
||||||
|
m._client_key = client_key
|
||||||
|
m._rebuild_client = AsyncMock()
|
||||||
|
return m
|
||||||
|
|
||||||
|
|
||||||
|
# --- credentials.py -----------------------------------------------------------
|
||||||
|
|
||||||
|
def test_reload_picks_up_a_key_written_after_startup(isolated):
|
||||||
|
credentials.load()
|
||||||
|
assert credentials.get_api_key() is None
|
||||||
|
data = json.loads((isolated / "credentials.json").read_text())
|
||||||
|
data["api_key"] = "written-by-install-sh"
|
||||||
|
(isolated / "credentials.json").write_text(json.dumps(data))
|
||||||
|
assert credentials.reload_api_key() == "written-by-install-sh"
|
||||||
|
|
||||||
|
|
||||||
|
def test_write_is_atomic_and_private(isolated):
|
||||||
|
credentials.save_api_key("k1")
|
||||||
|
f = isolated / "credentials.json"
|
||||||
|
assert json.loads(f.read_text())["api_key"] == "k1"
|
||||||
|
assert (f.stat().st_mode & 0o777) == 0o600
|
||||||
|
assert not (isolated / "credentials.json.tmp").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_pickup_secret_read_and_absent(isolated):
|
||||||
|
assert credentials.get_pickup_secret() is None
|
||||||
|
(isolated / "pickup_secret").write_text("abc123\n")
|
||||||
|
assert credentials.get_pickup_secret() == "abc123"
|
||||||
|
|
||||||
|
|
||||||
|
# --- auth-failure classification -------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("code", [4, 5, 134, 135])
|
||||||
|
def test_auth_failure_codes(code):
|
||||||
|
assert _is_auth_failure(code)
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_auth_failures_do_not_trigger_recovery():
|
||||||
|
assert not _is_auth_failure(3) # server unavailable
|
||||||
|
assert not _is_auth_failure(136)
|
||||||
|
|
||||||
|
|
||||||
|
def test_paho_reason_code_objects_classified():
|
||||||
|
from paho.mqtt.reasoncodes import ReasonCode
|
||||||
|
from paho.mqtt.packettypes import PacketTypes
|
||||||
|
assert _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Not authorized"))
|
||||||
|
assert _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Bad user name or password"))
|
||||||
|
assert not _is_auth_failure(ReasonCode(PacketTypes.CONNACK, "Server unavailable"))
|
||||||
|
|
||||||
|
|
||||||
|
# --- recovery loop ----------------------------------------------------------------
|
||||||
|
|
||||||
|
async def test_recovery_uses_key_already_on_disk_without_calling_c2(isolated, monkeypatch):
|
||||||
|
credentials.save_api_key("disk-key")
|
||||||
|
monkeypatch.setattr(credentials, "_api_key", None) # process started before the write
|
||||||
|
fetch = AsyncMock()
|
||||||
|
monkeypatch.setattr(key_pickup, "fetch_api_key", fetch)
|
||||||
|
m = _manager(client_key=None)
|
||||||
|
await m._recover_credentials()
|
||||||
|
m._rebuild_client.assert_awaited_once()
|
||||||
|
fetch.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_recovery_fetches_over_https_and_persists(isolated, monkeypatch):
|
||||||
|
monkeypatch.setattr(key_pickup, "fetch_api_key", AsyncMock(return_value="fresh-key"))
|
||||||
|
m = _manager(client_key="stale-key")
|
||||||
|
monkeypatch.setattr(credentials, "_api_key", "stale-key")
|
||||||
|
await m._recover_credentials()
|
||||||
|
m._rebuild_client.assert_awaited_once()
|
||||||
|
assert credentials.get_api_key() == "fresh-key"
|
||||||
|
assert json.loads((isolated / "credentials.json").read_text())["api_key"] == "fresh-key"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_recovery_backs_off_until_approved(isolated, monkeypatch):
|
||||||
|
fetch = AsyncMock(side_effect=[None, None, "approved-key"])
|
||||||
|
monkeypatch.setattr(key_pickup, "fetch_api_key", fetch)
|
||||||
|
m = _manager()
|
||||||
|
await asyncio.wait_for(m._recover_credentials(), timeout=5)
|
||||||
|
assert fetch.await_count == 3
|
||||||
|
m._rebuild_client.assert_awaited_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_recovery_stops_if_connection_recovers_meanwhile(isolated, monkeypatch):
|
||||||
|
monkeypatch.setattr(key_pickup, "fetch_api_key", AsyncMock(return_value=None))
|
||||||
|
m = _manager()
|
||||||
|
m._connected = True
|
||||||
|
await asyncio.wait_for(m._recover_credentials(), timeout=5)
|
||||||
|
m._rebuild_client.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_start_recovery_is_single_flight(isolated, monkeypatch):
|
||||||
|
gate = asyncio.Event()
|
||||||
|
|
||||||
|
async def slow_fetch():
|
||||||
|
await gate.wait()
|
||||||
|
return "k"
|
||||||
|
monkeypatch.setattr(key_pickup, "fetch_api_key", slow_fetch)
|
||||||
|
m = _manager()
|
||||||
|
m._start_recovery("a")
|
||||||
|
first = m._recovery_task
|
||||||
|
m._start_recovery("b")
|
||||||
|
assert m._recovery_task is first
|
||||||
|
gate.set()
|
||||||
|
await first
|
||||||
|
|
||||||
|
|
||||||
|
async def test_connect_without_key_starts_recovery(isolated, monkeypatch):
|
||||||
|
m = MQTTManager()
|
||||||
|
started = []
|
||||||
|
monkeypatch.setattr(m, "_connect_with_retry", AsyncMock())
|
||||||
|
monkeypatch.setattr(m, "_start_recovery", lambda why: started.append(why))
|
||||||
|
await m.connect()
|
||||||
|
assert started == ["no API key at startup"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_update_api_key_changes_creds_for_next_reconnect(isolated):
|
||||||
|
m = MQTTManager()
|
||||||
|
m._client = m._build_client()
|
||||||
|
m.update_api_key("rotated")
|
||||||
|
assert m._client._password == b"rotated"
|
||||||
|
assert m._client_key == "rotated"
|
||||||
|
|
||||||
|
|
||||||
|
# --- key_pickup.fetch_api_key -----------------------------------------------------
|
||||||
|
|
||||||
|
def _mock_http(monkeypatch, handler):
|
||||||
|
real = httpx.AsyncClient
|
||||||
|
|
||||||
|
def factory(*a, **kw):
|
||||||
|
kw["transport"] = httpx.MockTransport(handler)
|
||||||
|
return real(*a, **kw)
|
||||||
|
monkeypatch.setattr(key_pickup.httpx, "AsyncClient", factory)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_fetch_sends_pickup_secret_and_returns_key(isolated, monkeypatch):
|
||||||
|
(isolated / "pickup_secret").write_text("the-secret")
|
||||||
|
seen = {}
|
||||||
|
|
||||||
|
def handler(request):
|
||||||
|
seen["url"] = str(request.url)
|
||||||
|
seen["secret"] = request.headers.get("X-Pickup-Secret")
|
||||||
|
return httpx.Response(200, json={"approval_status": "approved", "api_key": "K"})
|
||||||
|
_mock_http(monkeypatch, handler)
|
||||||
|
assert await key_pickup.fetch_api_key() == "K"
|
||||||
|
assert seen == {"url": f"https://c2.example/nodes/{settings.node_id}/credentials", "secret": "the-secret"}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("resp", [
|
||||||
|
httpx.Response(200, json={"approval_status": "pending", "api_key": None}),
|
||||||
|
httpx.Response(401, json={"detail": "Invalid pickup secret"}),
|
||||||
|
httpx.Response(404),
|
||||||
|
httpx.Response(500),
|
||||||
|
httpx.Response(200, text="not json"),
|
||||||
|
])
|
||||||
|
async def test_fetch_returns_none_when_no_key(isolated, monkeypatch, resp):
|
||||||
|
(isolated / "pickup_secret").write_text("s")
|
||||||
|
_mock_http(monkeypatch, lambda request: resp)
|
||||||
|
assert await key_pickup.fetch_api_key() is None
|
||||||
|
|
||||||
|
|
||||||
|
async def test_fetch_without_pickup_secret_does_not_call_c2(isolated, monkeypatch):
|
||||||
|
def handler(request):
|
||||||
|
raise AssertionError("must not call C2 without a pickup secret")
|
||||||
|
_mock_http(monkeypatch, handler)
|
||||||
|
assert await key_pickup.fetch_api_key() is None
|
||||||
|
|
||||||
|
|
||||||
|
# --- node_update -> host sentinel ---------------------------------------------------
|
||||||
|
|
||||||
|
def test_node_update_writes_sentinel(isolated):
|
||||||
|
assert update_request.request_host_update() is True
|
||||||
|
data = json.loads((isolated / "update-requested").read_text())
|
||||||
|
assert "requested_at" in data
|
||||||
@@ -0,0 +1,179 @@
|
|||||||
|
"""
|
||||||
|
node-26#15 — Client/scripts/drb-update.sh, the host-level updater.
|
||||||
|
|
||||||
|
Runs the real script with stub git/docker/curl on PATH, so what is checked is
|
||||||
|
the script's decision logic and its security guards: it only updates when
|
||||||
|
asked (MQTT sentinel or a new C2 request), never takes anything but "update
|
||||||
|
now" from C2, refuses a mismatched git origin or registry, and keeps node
|
||||||
|
credentials off curl's argv.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import stat
|
||||||
|
import subprocess
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "drb-update.sh"
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.skipif(
|
||||||
|
not (SCRIPT.exists() and shutil.which("bash") and shutil.which("jq") and shutil.which("flock")),
|
||||||
|
reason="needs the repo's scripts/ (absent inside the image) plus bash, jq and flock",
|
||||||
|
)
|
||||||
|
|
||||||
|
REPO = "https://git.example.net/logan/node-26.git"
|
||||||
|
OLD, NEW = "a" * 40, "b" * 40
|
||||||
|
|
||||||
|
STUB_GIT = r'''#!/usr/bin/env bash
|
||||||
|
echo "git $*" >> "$STUB/calls"
|
||||||
|
while [ "$1" = -c ]; do shift 2; done
|
||||||
|
case "$1" in
|
||||||
|
rev-parse) if [ "$2" = HEAD ]; then cat "$STUB/head"; else echo "$STUB_NEW"; fi ;;
|
||||||
|
remote) echo "$STUB_ORIGIN" ;;
|
||||||
|
fetch) exit 0 ;;
|
||||||
|
show-ref) [ "$4" = "refs/tags/v1" ] ;;
|
||||||
|
checkout) echo "${@: -1}" > "$STUB/head" ;;
|
||||||
|
*) exit 0 ;;
|
||||||
|
esac
|
||||||
|
'''
|
||||||
|
STUB_DOCKER = '#!/usr/bin/env bash\necho "docker $*" >> "$STUB/calls"\n'
|
||||||
|
STUB_CURL = r'''#!/usr/bin/env bash
|
||||||
|
echo "curl $*" >> "$STUB/calls"
|
||||||
|
for a in "$@"; do case "$a" in @*) cat "${a#@}" >> "$STUB/headers" ;; esac; done
|
||||||
|
cat "$STUB/response"
|
||||||
|
'''
|
||||||
|
|
||||||
|
|
||||||
|
def _exe(path: Path, body: str):
|
||||||
|
path.write_text(body)
|
||||||
|
path.chmod(path.stat().st_mode | stat.S_IEXEC)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def env(tmp_path):
|
||||||
|
stub = tmp_path / "stub"
|
||||||
|
bindir = stub / "bin"
|
||||||
|
bindir.mkdir(parents=True)
|
||||||
|
_exe(bindir / "git", STUB_GIT)
|
||||||
|
_exe(bindir / "docker", STUB_DOCKER)
|
||||||
|
_exe(bindir / "curl", STUB_CURL)
|
||||||
|
(stub / "head").write_text(OLD + "\n")
|
||||||
|
(stub / "calls").write_text("")
|
||||||
|
(stub / "response").write_text(json.dumps({"update_requested_at": None}))
|
||||||
|
|
||||||
|
inst = tmp_path / "node"
|
||||||
|
(inst / ".git").mkdir(parents=True)
|
||||||
|
(inst / "configs").mkdir()
|
||||||
|
(inst / ".env").write_text(
|
||||||
|
"NODE_ID=node-7\nC2_URL=https://api.example.net\nIMAGE_REGISTRY=git.example.net\n"
|
||||||
|
)
|
||||||
|
(inst / "configs" / "credentials.json").write_text(json.dumps({"api_key": "k" * 64}))
|
||||||
|
(inst / "configs" / "pickup_secret").write_text("p" * 48)
|
||||||
|
|
||||||
|
conf = tmp_path / "updater.conf"
|
||||||
|
conf.write_text(f"INSTALL_DIR={inst}\nREPO_URL={REPO}\nREF=v1\nREGISTRY=git.example.net\nBUILD=0\n")
|
||||||
|
|
||||||
|
e = dict(os.environ)
|
||||||
|
e.update({
|
||||||
|
"PATH": f"{bindir}:{os.environ['PATH']}",
|
||||||
|
"STUB": str(stub), "STUB_NEW": NEW, "STUB_ORIGIN": REPO,
|
||||||
|
"DRB_UPDATER_CONF": str(conf),
|
||||||
|
"DRB_UPDATER_STATE": str(tmp_path / "state"),
|
||||||
|
"DRB_UPDATER_LOCK": str(tmp_path / "lock"),
|
||||||
|
})
|
||||||
|
return {"env": e, "stub": stub, "inst": inst, "conf": conf, "state": tmp_path / "state"}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(env):
|
||||||
|
return subprocess.run(["bash", str(SCRIPT)], env=env["env"], capture_output=True, text=True, timeout=30)
|
||||||
|
|
||||||
|
|
||||||
|
def _calls(env):
|
||||||
|
return (env["stub"] / "calls").read_text()
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_request_means_checkin_only(env):
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode == 0, r.stderr
|
||||||
|
calls = _calls(env)
|
||||||
|
assert "update-check" in calls
|
||||||
|
assert "docker" not in calls
|
||||||
|
assert "fetch" not in calls
|
||||||
|
|
||||||
|
|
||||||
|
def test_c2_request_updates_once_and_reports(env):
|
||||||
|
(env["stub"] / "response").write_text(json.dumps({"update_requested_at": "2026-10-04T12:00:00+00:00"}))
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode == 0, r.stderr + r.stdout
|
||||||
|
calls = _calls(env)
|
||||||
|
assert "docker compose pull" in calls
|
||||||
|
assert "docker compose up -d --no-build --force-recreate --remove-orphans" in calls
|
||||||
|
assert (env["stub"] / "head").read_text().strip() == NEW
|
||||||
|
assert '"applied_request_at":"2026-10-04T12:00:00+00:00"' in calls
|
||||||
|
assert '"result":"ok"' in calls
|
||||||
|
# Same request again: nothing to do.
|
||||||
|
(env["stub"] / "calls").write_text("")
|
||||||
|
assert _run(env).returncode == 0
|
||||||
|
assert "docker" not in _calls(env)
|
||||||
|
|
||||||
|
|
||||||
|
def test_mqtt_sentinel_triggers_update_and_is_consumed(env):
|
||||||
|
sentinel = env["inst"] / "configs" / "update-requested"
|
||||||
|
sentinel.write_text('{"ref": "evil", "cmd": "rm -rf /"}')
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode == 0, r.stderr
|
||||||
|
assert not sentinel.exists()
|
||||||
|
calls = _calls(env)
|
||||||
|
assert "docker compose pull" in calls
|
||||||
|
assert "evil" not in calls and "rm -rf" not in calls
|
||||||
|
|
||||||
|
|
||||||
|
def test_c2_response_is_never_used_as_a_ref_or_command(env):
|
||||||
|
(env["stub"] / "response").write_text(json.dumps({"update_requested_at": "$(touch /tmp/pwned); main"}))
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode == 0
|
||||||
|
assert "docker" not in _calls(env) # malformed request is ignored, not executed
|
||||||
|
|
||||||
|
|
||||||
|
def test_refuses_when_git_origin_differs(env):
|
||||||
|
env["env"]["STUB_ORIGIN"] = "https://attacker.example/node-26.git"
|
||||||
|
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode != 0
|
||||||
|
assert "refusing" in r.stderr
|
||||||
|
assert "docker" not in _calls(env)
|
||||||
|
|
||||||
|
|
||||||
|
def test_refuses_when_env_registry_differs(env):
|
||||||
|
(env["inst"] / ".env").write_text(
|
||||||
|
"NODE_ID=node-7\nC2_URL=https://api.example.net\nIMAGE_REGISTRY=evil.example\n"
|
||||||
|
)
|
||||||
|
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode != 0
|
||||||
|
assert "docker compose pull" not in _calls(env)
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_https_repo_in_conf_is_rejected(env):
|
||||||
|
env["conf"].write_text(env["conf"].read_text().replace("https://", "http://"))
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode != 0
|
||||||
|
assert "git" not in _calls(env)
|
||||||
|
|
||||||
|
|
||||||
|
def test_credentials_go_in_a_header_file_never_argv(env):
|
||||||
|
_run(env)
|
||||||
|
calls = _calls(env)
|
||||||
|
assert "k" * 64 not in calls and "p" * 48 not in calls
|
||||||
|
headers = (env["stub"] / "headers").read_text()
|
||||||
|
assert f"Authorization: Bearer {'k' * 64}" in headers
|
||||||
|
assert f"X-Pickup-Secret: {'p' * 48}" in headers
|
||||||
|
|
||||||
|
|
||||||
|
def test_plain_http_c2_skips_checkin(env):
|
||||||
|
(env["inst"] / ".env").write_text("NODE_ID=node-7\nC2_URL=http://api.example.net\nIMAGE_REGISTRY=git.example.net\n")
|
||||||
|
r = _run(env)
|
||||||
|
assert r.returncode == 0
|
||||||
|
assert "curl" not in _calls(env)
|
||||||
@@ -126,10 +126,48 @@ def test_analog_needs_no_control_channel():
|
|||||||
console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
|
console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
|
||||||
)))
|
)))
|
||||||
assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
|
assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
|
||||||
assert "not recorded" in c["decode"]["action"]
|
# node-26#20: analog calls are recorded now (single-channel attribution).
|
||||||
|
assert c["decode"]["action"] == ""
|
||||||
assert "this dashboard" in c["source"]["summary"]
|
assert "this dashboard" in c["source"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
|
_CONV_SRC = {"name": "White Plains", "type": "P25", "mode": "conventional", "system_id": "wp", "local": False}
|
||||||
|
_PD1 = {"name": "PD1 Dispatch", "freq": 460400000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
|
||||||
|
_PD3 = {"name": "PD3 Tac", "freq": 460425000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
|
||||||
|
|
||||||
|
|
||||||
|
def _conv_facts(chans):
|
||||||
|
now = 1_000_000.0
|
||||||
|
return _facts(source=_CONV_SRC, conventional=chans,
|
||||||
|
console={"last_poll_epoch": now, "channels": [], "trunk": [], "last_heard": None})
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_quiet_channels_are_ok_not_a_control_channel_error():
|
||||||
|
c = _by_id(health.build_checks(_conv_facts([_PD1, _PD3])))
|
||||||
|
assert c["decode"]["state"] == "ok"
|
||||||
|
assert "PD1 Dispatch (460.4000 MHz)" in c["decode"]["summary"]
|
||||||
|
assert "control channel" not in c["decode"]["summary"]
|
||||||
|
assert "P25 conventional" in c["source"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_decoding_shows_nac():
|
||||||
|
c = _by_id(health.build_checks(_conv_facts([dict(_PD1, nac=0x293), _PD3])))
|
||||||
|
assert c["decode"]["state"] == "ok" and "NAC 0x293" in c["decode"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_all_encrypted_is_an_error_naming_the_algorithm():
|
||||||
|
enc = [dict(_PD1, encrypted=True, algid=0x84, keyid=0x1234), dict(_PD3, encrypted=True, algid=0x84)]
|
||||||
|
c = _by_id(health.build_checks(_conv_facts(enc)))
|
||||||
|
assert c["decode"]["state"] == "error"
|
||||||
|
assert "AES-256" in c["decode"]["summary"] and "0x1234" in c["decode"]["summary"]
|
||||||
|
assert "different system" in c["decode"]["action"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_conventional_partly_encrypted_warns():
|
||||||
|
c = _by_id(health.build_checks(_conv_facts([dict(_PD1, encrypted=True, algid=0x81), _PD3])))
|
||||||
|
assert c["decode"]["state"] == "warn" and "PD1 Dispatch" in c["decode"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
def test_op25_unreachable_and_icecast_down():
|
def test_op25_unreachable_and_icecast_down():
|
||||||
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
|
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
|
||||||
assert c["op25"]["state"] == "error"
|
assert c["op25"]["state"] == "error"
|
||||||
@@ -182,7 +220,26 @@ def test_listen_analog_single_frequency(client, op25):
|
|||||||
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
|
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
|
||||||
|
|
||||||
|
|
||||||
|
def test_listen_p25_conventional(client, op25):
|
||||||
|
r = client.post("/api/listen", json={
|
||||||
|
"mode": "p25conv", "frequencies": [460.4, 460.425], "name": "White Plains", "nac": "0x293",
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = config_manager.load_node_config().override_config
|
||||||
|
assert cfg.type == "P25" and cfg.config["mode"] == "conventional"
|
||||||
|
assert "control_channels" not in cfg.config
|
||||||
|
assert [c["freq"] for c in cfg.config["channels"]] == [460.4, 460.425]
|
||||||
|
assert {c["nac"] for c in cfg.config["channels"]} == {"293"}
|
||||||
|
op25["gen"].assert_awaited_once()
|
||||||
|
status = client.get("/api/status").json()
|
||||||
|
assert status["system_kind"] == "P25 conventional"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("body,needle", [
|
@pytest.mark.parametrize("body,needle", [
|
||||||
|
({"mode": "p25conv", "frequencies": [460.4, 460.425, 460.45]}, "at most 2"),
|
||||||
|
({"mode": "p25conv", "frequencies": [460.4, 462.6]}, "apart"),
|
||||||
|
({"mode": "p25conv", "frequencies": [460.4, 460.4]}, "twice"),
|
||||||
|
({"mode": "p25conv", "frequencies": [460.4], "nac": "zz"}, "hex"),
|
||||||
({"mode": "p25", "frequencies": []}, "at least one"),
|
({"mode": "p25", "frequencies": []}, "at least one"),
|
||||||
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
|
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
|
||||||
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
|
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
|
||||||
|
|||||||
+133
-55
@@ -17,9 +17,13 @@
|
|||||||
# 2. Install Docker + compose plugin + git + curl + jq
|
# 2. Install Docker + compose plugin + git + curl + jq
|
||||||
# 3. Clone logan/node-26 at a PINNED ref into $INSTALL_DIR
|
# 3. Clone logan/node-26 at a PINNED ref into $INSTALL_DIR
|
||||||
# 4. Write .env (non-interactive from flags/env, interactive fallback)
|
# 4. Write .env (non-interactive from flags/env, interactive fallback)
|
||||||
# 5. Enroll with C2 (POST /nodes/enroll) and poll for the api_key
|
# 5. Enroll with C2 (POST /nodes/enroll), then BLOCK until an admin approves
|
||||||
# 6. docker compose pull && up -d (prebuilt; --build opts into the ~1h build)
|
# and the api_key is written (node-26#14). Ctrl-C safe: re-run resumes
|
||||||
# 7. Print the approval step
|
# from configs/pickup_secret. A re-run also refreshes a stale key.
|
||||||
|
# 6. docker compose pull && up -d, ALWAYS recreating edge-node so it loads
|
||||||
|
# the key on disk (prebuilt; --build opts into the ~1h build)
|
||||||
|
# 7. Install the host updater (systemd timer + path unit, node-26#15)
|
||||||
|
# 8. Print a summary
|
||||||
#
|
#
|
||||||
# It does NOT set up WireGuard. WireGuard-per-node was evaluated and rejected
|
# It does NOT set up WireGuard. WireGuard-per-node was evaluated and rejected
|
||||||
# (Server/MQTT-PUBLIC-AUTH-PLAN.md, Server/infra/main.tf:67-73). A field node
|
# (Server/MQTT-PUBLIC-AUTH-PLAN.md, Server/infra/main.tf:67-73). A field node
|
||||||
@@ -85,7 +89,14 @@ REGISTRY_PASS="${DRB_REGISTRY_PASS:-}"
|
|||||||
DO_BUILD=0 # 0 = pull prebuilt images (default), 1 = build on the Pi (~1h for op25)
|
DO_BUILD=0 # 0 = pull prebuilt images (default), 1 = build on the Pi (~1h for op25)
|
||||||
DO_START=1
|
DO_START=1
|
||||||
ASSUME_YES=0
|
ASSUME_YES=0
|
||||||
ENROLL_WAIT="${DRB_ENROLL_WAIT:-0}" # seconds to block waiting for admin approval; 0 = don't block
|
# How long to block waiting for admin approval. Default: forever (node-26#14 —
|
||||||
|
# the old default of 0 started a keyless stack that never came online).
|
||||||
|
# N>0 = give up after N seconds and start anyway; 0 = don't wait at all. Either
|
||||||
|
# way the edge-node app keeps fetching the key itself and comes online on its
|
||||||
|
# own once approved.
|
||||||
|
ENROLL_WAIT="${DRB_ENROLL_WAIT:-forever}"
|
||||||
|
APP_URL="${DRB_APP_URL:-}"
|
||||||
|
INSTALL_UPDATER=1
|
||||||
|
|
||||||
C='\033[0;36m'; G='\033[0;32m'; Y='\033[1;33m'; R='\033[0;31m'; N='\033[0m'
|
C='\033[0;36m'; G='\033[0;32m'; Y='\033[1;33m'; R='\033[0;31m'; N='\033[0m'
|
||||||
say() { printf "${C}==>${N} %s\n" "$*"; }
|
say() { printf "${C}==>${N} %s\n" "$*"; }
|
||||||
@@ -111,7 +122,12 @@ Usage: install.sh [options]
|
|||||||
--dir PATH install location (default /opt/drb/node-26)
|
--dir PATH install location (default /opt/drb/node-26)
|
||||||
--build build images locally instead of pulling (~1h for op25)
|
--build build images locally instead of pulling (~1h for op25)
|
||||||
--no-start configure and enroll, but do not start containers
|
--no-start configure and enroll, but do not start containers
|
||||||
--wait-approval SEC block up to SEC seconds polling for admin approval
|
--wait-approval SEC stop waiting for admin approval after SEC seconds
|
||||||
|
(default: wait forever; 0 = don't wait). The node
|
||||||
|
still fetches its key by itself once approved.
|
||||||
|
--app-url URL admin UI base for the approval message
|
||||||
|
(default: C2 URL without its "api." prefix)
|
||||||
|
--no-updater don't install the host updater (no remote updates)
|
||||||
-y, --yes never prompt; fail instead of asking
|
-y, --yes never prompt; fail instead of asking
|
||||||
|
|
||||||
Every option also has an env var: DRB_NODE_ID, DRB_C2_URL, DRB_ENROLLMENT_TOKEN,
|
Every option also has an env var: DRB_NODE_ID, DRB_C2_URL, DRB_ENROLLMENT_TOKEN,
|
||||||
@@ -138,12 +154,17 @@ while [ $# -gt 0 ]; do
|
|||||||
--build) DO_BUILD=1; shift ;;
|
--build) DO_BUILD=1; shift ;;
|
||||||
--no-start) DO_START=0; shift ;;
|
--no-start) DO_START=0; shift ;;
|
||||||
--wait-approval) ENROLL_WAIT="$2"; shift 2 ;;
|
--wait-approval) ENROLL_WAIT="$2"; shift 2 ;;
|
||||||
|
--app-url) APP_URL="$2"; shift 2 ;;
|
||||||
|
--no-updater) INSTALL_UPDATER=0; shift ;;
|
||||||
-y|--yes) ASSUME_YES=1; shift ;;
|
-y|--yes) ASSUME_YES=1; shift ;;
|
||||||
-h|--help) usage; exit 0 ;;
|
-h|--help) usage; exit 0 ;;
|
||||||
*) die "unknown option: $1 (try --help)" ;;
|
*) die "unknown option: $1 (try --help)" ;;
|
||||||
esac
|
esac
|
||||||
done
|
done
|
||||||
|
|
||||||
|
[ "$ENROLL_WAIT" = forever ] || [[ "$ENROLL_WAIT" =~ ^[0-9]+$ ]] \
|
||||||
|
|| die "--wait-approval takes a number of seconds (or omit it to wait until approved)"
|
||||||
|
|
||||||
# Prompts must come from the terminal, not from the `curl |` pipe on stdin.
|
# Prompts must come from the terminal, not from the `curl |` pipe on stdin.
|
||||||
ask() { # ask VAR "prompt" "default"
|
ask() { # ask VAR "prompt" "default"
|
||||||
local __v="$1" __p="$2" __d="${3:-}" __r=""
|
local __v="$1" __p="$2" __d="${3:-}" __r=""
|
||||||
@@ -226,7 +247,9 @@ mkdir -p "$(dirname "$INSTALL_DIR")"
|
|||||||
if [ -d "$INSTALL_DIR/.git" ]; then
|
if [ -d "$INSTALL_DIR/.git" ]; then
|
||||||
ok "existing install at $INSTALL_DIR — updating in place (.env is preserved)"
|
ok "existing install at $INSTALL_DIR — updating in place (.env is preserved)"
|
||||||
git -C "$INSTALL_DIR" remote set-url origin "$REPO_URL"
|
git -C "$INSTALL_DIR" remote set-url origin "$REPO_URL"
|
||||||
git -C "$INSTALL_DIR" fetch --tags --prune origin
|
# --force: a moved release tag (v1 re-cut) must replace the local copy,
|
||||||
|
# otherwise fetch refuses with "would clobber existing tag" and set -e dies.
|
||||||
|
git -C "$INSTALL_DIR" fetch --force --tags --prune origin
|
||||||
else
|
else
|
||||||
git clone --no-checkout "$REPO_URL" "$INSTALL_DIR"
|
git clone --no-checkout "$REPO_URL" "$INSTALL_DIR"
|
||||||
fi
|
fi
|
||||||
@@ -345,39 +368,68 @@ PICKUP_FILE="$INSTALL_DIR/configs/pickup_secret"
|
|||||||
# GET /nodes/{id}/credentials. Sets CRED_HTTP + CRED_BODY (no -f: we need the
|
# GET /nodes/{id}/credentials. Sets CRED_HTTP + CRED_BODY (no -f: we need the
|
||||||
# body and status on a 4xx). One implementation so first-run and re-run agree.
|
# body and status on a 4xx). One implementation so first-run and re-run agree.
|
||||||
creds_pickup() { # creds_pickup PICKUP_SECRET
|
creds_pickup() { # creds_pickup PICKUP_SECRET
|
||||||
local _tmp; _tmp="$(mktemp)"
|
local _tmp _hdr; _tmp="$(mktemp)"; _hdr="$(mktemp)"; chmod 600 "$_hdr"
|
||||||
|
# Secret via a 0600 header file, never curl's argv (visible in `ps`).
|
||||||
|
printf 'X-Pickup-Secret: %s\n' "$1" > "$_hdr"
|
||||||
CRED_HTTP="$(curl -sS -o "$_tmp" -w '%{http_code}' \
|
CRED_HTTP="$(curl -sS -o "$_tmp" -w '%{http_code}' \
|
||||||
"$C2_URL/nodes/$NODE_ID/credentials" -H "X-Pickup-Secret: $1" 2>/dev/null || echo 000)"
|
"$C2_URL/nodes/$NODE_ID/credentials" -H "@$_hdr" 2>/dev/null || echo 000)"
|
||||||
CRED_BODY="$(cat "$_tmp" 2>/dev/null || true)"
|
CRED_BODY="$(cat "$_tmp" 2>/dev/null || true)"
|
||||||
rm -f "$_tmp"
|
rm -f "$_tmp" "$_hdr"
|
||||||
}
|
}
|
||||||
cred_field() { printf '%s' "${CRED_BODY:-}" | jq -r "$1 // empty" 2>/dev/null || true; }
|
cred_field() { printf '%s' "${CRED_BODY:-}" | jq -r "$1 // empty" 2>/dev/null || true; }
|
||||||
|
|
||||||
|
KEY_WRITTEN=0
|
||||||
|
# Merge into the existing file: the edge-node app keeps its dashboard auth
|
||||||
|
# salt + session secret in the same JSON, and clobbering them logs everyone
|
||||||
|
# out. Atomic (tmp + mv) so a Ctrl-C can never leave half a file.
|
||||||
write_creds() { # write_creds API_KEY
|
write_creds() { # write_creds API_KEY
|
||||||
umask 077; jq -n --arg k "$1" '{api_key:$k}' > "$CREDS"; umask 022
|
local _tmp="$CREDS.tmp"
|
||||||
ok "api_key received and written to configs/credentials.json"
|
umask 077
|
||||||
|
if [ -s "$CREDS" ] && jq -e . "$CREDS" >/dev/null 2>&1; then
|
||||||
|
jq --arg k "$1" '.api_key = $k' "$CREDS" > "$_tmp"
|
||||||
|
else
|
||||||
|
jq -n --arg k "$1" '{api_key:$k}' > "$_tmp"
|
||||||
|
fi
|
||||||
|
umask 022
|
||||||
|
mv -f "$_tmp" "$CREDS"
|
||||||
|
KEY_WRITTEN=1
|
||||||
|
ok "api_key written to configs/credentials.json"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[ -n "$APP_URL" ] || { APP_URL="${C2_URL/:\/\/api./://}"; }
|
||||||
|
APP_URL="${APP_URL%/}"
|
||||||
|
|
||||||
say_pending() { # say_pending APPROVAL_STATUS — not an error: node is enrolled, key not minted yet
|
say_pending() { # say_pending APPROVAL_STATUS — not an error: node is enrolled, key not minted yet
|
||||||
warn "not approved yet — C2 reports approval_status=${1:-pending}, no api_key minted."
|
warn "not approved yet — C2 reports approval_status=${1:-pending}, no api_key minted."
|
||||||
warn "an admin must, at <app-url>/settings/nodes : Approve '$NODE_ID' (assigning a system is separate)."
|
warn "an admin must approve '$NODE_ID' at ${APP_URL}/settings/nodes. The node keeps fetching"
|
||||||
warn "then re-run this installer — it reuses configs/pickup_secret — or fetch it directly:"
|
warn "its key by itself and comes online once approved; re-running this installer also works."
|
||||||
warn " curl -fsS $C2_URL/nodes/$NODE_ID/credentials -H \"X-Pickup-Secret: \$(cat $PICKUP_FILE)\" | jq -r .api_key"
|
|
||||||
}
|
}
|
||||||
|
|
||||||
# Poll creds_pickup for up to ENROLL_WAIT seconds while still pending. Result
|
# Block until the node is approved and C2 hands back an api_key (node-26#14).
|
||||||
# left in CRED_HTTP/CRED_BODY. --wait-approval is what would have avoided the
|
# Result left in CRED_HTTP/CRED_BODY. Returns 0 with a key, 1 on timeout/0-wait,
|
||||||
# original prod bug; the default (0) does not block, so the re-run path below
|
# dies on a credential error (401/404) that waiting cannot fix.
|
||||||
# must stand on its own.
|
|
||||||
wait_for_key() { # wait_for_key PICKUP_SECRET
|
wait_for_key() { # wait_for_key PICKUP_SECRET
|
||||||
[ "${ENROLL_WAIT:-0}" -gt 0 ] || return 0
|
[ "$ENROLL_WAIT" = 0 ] && return 1
|
||||||
local _end; _end=$(( $(date +%s) + ENROLL_WAIT ))
|
local _end=0 _shown=0 _delay=5
|
||||||
say "Waiting up to ${ENROLL_WAIT}s for an admin to approve '$NODE_ID'"
|
[ "$ENROLL_WAIT" = forever ] || _end=$(( $(date +%s) + ENROLL_WAIT ))
|
||||||
while [ "$(date +%s)" -lt "$_end" ]; do
|
trap 'printf "\n"; warn "interrupted — nothing is lost: re-run the same command to resume waiting (configs/pickup_secret is saved)."; exit 130' INT TERM
|
||||||
sleep 10
|
while :; do
|
||||||
|
case "$CRED_HTTP" in
|
||||||
|
200) [ -n "$(cred_field '.api_key')" ] && { trap - INT TERM; return 0; } ;;
|
||||||
|
401) die "C2 rejected the saved pickup secret (401) while waiting — re-run with --token DRB-… to re-enroll." ;;
|
||||||
|
404) die "C2 no longer knows node '$NODE_ID' (404) — it was deleted. Re-run with --token DRB-… to enroll again." ;;
|
||||||
|
*) : ;; # 000/5xx: transient, keep waiting
|
||||||
|
esac
|
||||||
|
if [ "$_shown" = 0 ]; then
|
||||||
|
printf "\n${Y}Please have an admin approve node '%s' at %s/settings/nodes to continue.${N}\n" \
|
||||||
|
"$NODE_ID" "$APP_URL"
|
||||||
|
printf "Waiting for approval%s… (Ctrl-C is safe — re-run the same command to resume)\n\n" \
|
||||||
|
"$([ "$_end" = 0 ] && echo "" || echo " for up to ${ENROLL_WAIT}s")"
|
||||||
|
_shown=1
|
||||||
|
fi
|
||||||
|
if [ "$_end" != 0 ] && [ "$(date +%s)" -ge "$_end" ]; then trap - INT TERM; return 1; fi
|
||||||
|
sleep "$_delay"; [ "$_delay" -lt 30 ] && _delay=$(( _delay + 5 ))
|
||||||
creds_pickup "$1"
|
creds_pickup "$1"
|
||||||
[ "$CRED_HTTP" = 200 ] || return 0
|
|
||||||
[ -z "$(cred_field '.api_key')" ] || return 0
|
|
||||||
done
|
done
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -390,12 +442,15 @@ do_fresh_enroll() {
|
|||||||
BODY="$(jq -nc --arg id "$NODE_ID" --arg n "${NODE_NAME:-$NODE_ID}" \
|
BODY="$(jq -nc --arg id "$NODE_ID" --arg n "${NODE_NAME:-$NODE_ID}" \
|
||||||
--argjson lat "${NODE_LAT:-0}" --argjson lon "${NODE_LON:-0}" \
|
--argjson lat "${NODE_LAT:-0}" --argjson lon "${NODE_LON:-0}" \
|
||||||
'{node_id:$id,name:$n,lat:$lat,lon:$lon}')"
|
'{node_id:$id,name:$n,lat:$lat,lon:$lon}')"
|
||||||
# Token goes in a header from a shell variable — never on a process command
|
# Token goes in a 0600 header file — never on a process command line (curl's
|
||||||
# line, never echoed.
|
# argv is visible in `ps`; CISO condition, server-26#61), never echoed.
|
||||||
|
local _ehdr; _ehdr="$(mktemp)"; chmod 600 "$_ehdr"
|
||||||
|
printf 'X-Enrollment-Token: %s\n' "$ENROLLMENT_TOKEN" > "$_ehdr"
|
||||||
if ! RESP="$(curl -fsS -X POST "$C2_URL/nodes/enroll" \
|
if ! RESP="$(curl -fsS -X POST "$C2_URL/nodes/enroll" \
|
||||||
-H "Content-Type: application/json" \
|
-H "Content-Type: application/json" \
|
||||||
-H "X-Enrollment-Token: $ENROLLMENT_TOKEN" \
|
-H "@$_ehdr" \
|
||||||
--data "$BODY" 2>&1)"; then
|
--data "$BODY" 2>&1)"; then
|
||||||
|
rm -f "$_ehdr"
|
||||||
case "$RESP" in
|
case "$RESP" in
|
||||||
*403*) die "enroll refused (403): node '$NODE_ID' is already approved on C2, and
|
*403*) die "enroll refused (403): node '$NODE_ID' is already approved on C2, and
|
||||||
this machine has no configs/pickup_secret to collect its key with. A token
|
this machine has no configs/pickup_secret to collect its key with. A token
|
||||||
@@ -409,6 +464,7 @@ do_fresh_enroll() {
|
|||||||
*) die "enroll failed: $RESP" ;;
|
*) die "enroll failed: $RESP" ;;
|
||||||
esac
|
esac
|
||||||
fi
|
fi
|
||||||
|
rm -f "$_ehdr"
|
||||||
PICKUP="$(printf '%s' "$RESP" | jq -r '.pickup_secret')"
|
PICKUP="$(printf '%s' "$RESP" | jq -r '.pickup_secret')"
|
||||||
STATUS="$(printf '%s' "$RESP" | jq -r '.approval_status')"
|
STATUS="$(printf '%s' "$RESP" | jq -r '.approval_status')"
|
||||||
[ -n "$PICKUP" ] && [ "$PICKUP" != null ] || die "enroll returned no pickup_secret: $RESP"
|
[ -n "$PICKUP" ] && [ "$PICKUP" != null ] || die "enroll returned no pickup_secret: $RESP"
|
||||||
@@ -418,21 +474,32 @@ do_fresh_enroll() {
|
|||||||
|
|
||||||
creds_pickup "$PICKUP"
|
creds_pickup "$PICKUP"
|
||||||
[ "$CRED_HTTP" = 200 ] || die "post-enroll credential fetch failed (HTTP $CRED_HTTP): ${CRED_BODY:-<no body>}"
|
[ "$CRED_HTTP" = 200 ] || die "post-enroll credential fetch failed (HTTP $CRED_HTTP): ${CRED_BODY:-<no body>}"
|
||||||
KEY="$(cred_field '.api_key')"
|
if wait_for_key "$PICKUP"; then
|
||||||
if [ -z "$KEY" ]; then
|
write_creds "$(cred_field '.api_key')"
|
||||||
wait_for_key "$PICKUP" || true
|
|
||||||
KEY="$(cred_field '.api_key')"
|
|
||||||
fi
|
|
||||||
if [ -n "$KEY" ]; then
|
|
||||||
write_creds "$KEY"
|
|
||||||
else
|
else
|
||||||
say_pending "$(cred_field '.approval_status')"
|
say_pending "$(cred_field '.approval_status')"
|
||||||
fi
|
fi
|
||||||
}
|
}
|
||||||
|
|
||||||
if [ -s "$CREDS" ] && jq -e '.api_key // empty' "$CREDS" >/dev/null 2>&1; then
|
DISK_KEY="$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null || true)"
|
||||||
|
if [ -n "$DISK_KEY" ] && [ -s "$PICKUP_FILE" ] && [ -n "${C2_URL:-}" ]; then
|
||||||
|
# Key on disk, but it may be STALE (an admin Reissue, or the old approve
|
||||||
|
# that rotated on every click). Ask C2 for the current one and replace it if
|
||||||
|
# different. Best-effort: on any error keep the key we have.
|
||||||
|
say "Enrollment — key on disk; checking it is still current"
|
||||||
|
creds_pickup "$(cat "$PICKUP_FILE")"
|
||||||
|
SERVER_KEY="$(cred_field '.api_key')"
|
||||||
|
if [ "$CRED_HTTP" = 200 ] && [ -n "$SERVER_KEY" ] && [ "$SERVER_KEY" != "$DISK_KEY" ]; then
|
||||||
|
warn "the key on disk is not the one C2 has on file (reissued?) — replacing it"
|
||||||
|
write_creds "$SERVER_KEY"
|
||||||
|
elif [ "$CRED_HTTP" = 200 ] && [ -n "$SERVER_KEY" ]; then
|
||||||
|
ok "api_key on disk matches C2"
|
||||||
|
else
|
||||||
|
warn "could not confirm the key with C2 (HTTP $CRED_HTTP) — keeping the key on disk"
|
||||||
|
fi
|
||||||
|
elif [ -n "$DISK_KEY" ]; then
|
||||||
say "Enrollment"
|
say "Enrollment"
|
||||||
ok "api_key already on disk — skipping enrollment"
|
ok "api_key already on disk (no pickup secret to re-check it with) — skipping enrollment"
|
||||||
elif [ -z "${C2_URL:-}" ]; then
|
elif [ -z "${C2_URL:-}" ]; then
|
||||||
say "Enrollment"
|
say "Enrollment"
|
||||||
warn "no C2_URL — skipping enrollment"
|
warn "no C2_URL — skipping enrollment"
|
||||||
@@ -446,16 +513,11 @@ elif [ -s "$PICKUP_FILE" ]; then
|
|||||||
creds_pickup "$PICKUP"
|
creds_pickup "$PICKUP"
|
||||||
case "$CRED_HTTP" in
|
case "$CRED_HTTP" in
|
||||||
200)
|
200)
|
||||||
API_KEY="$(cred_field '.api_key')"
|
if wait_for_key "$PICKUP"; then
|
||||||
if [ -z "$API_KEY" ]; then
|
write_creds "$(cred_field '.api_key')"
|
||||||
wait_for_key "$PICKUP" || true
|
|
||||||
API_KEY="$(cred_field '.api_key')"
|
|
||||||
fi
|
|
||||||
if [ -n "$API_KEY" ]; then
|
|
||||||
write_creds "$API_KEY"
|
|
||||||
else
|
else
|
||||||
# Still pending. Enrolled and idempotent — next run collects the key.
|
# Only with --wait-approval N/0. Enrolled and idempotent; the app
|
||||||
# Clean exit, fall through to start the stack. NOT a failure.
|
# fetches the key itself once approved. NOT a failure.
|
||||||
say_pending "$(cred_field '.approval_status')"
|
say_pending "$(cred_field '.approval_status')"
|
||||||
fi
|
fi
|
||||||
;;
|
;;
|
||||||
@@ -510,12 +572,30 @@ if [ "$DO_START" = 1 ]; then
|
|||||||
fi
|
fi
|
||||||
docker compose up --no-build -d
|
docker compose up --no-build -d
|
||||||
fi
|
fi
|
||||||
ok "stack started"
|
# ALWAYS recreate edge-node (node-26#14): `up -d` leaves an already-running
|
||||||
|
# container alone, and a process started before the key was written would
|
||||||
|
# otherwise keep running keyless. Cheap, and makes every re-run a fix.
|
||||||
|
docker compose up -d --no-build --force-recreate --no-deps edge-node
|
||||||
|
ok "stack started (edge-node recreated so it loads configs/credentials.json)"
|
||||||
else
|
else
|
||||||
say "Skipping start (--no-start). Run: cd $INSTALL_DIR && make up-prebuilt"
|
say "Skipping start (--no-start). Run: cd $INSTALL_DIR && make up-prebuilt"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
# ── 7. What the operator does next ──────────────────────────────────────────
|
# ── 7. Host updater (node-26#15) ───────────────────────────────────────────
|
||||||
|
UPDATER_STATE="not installed (--no-updater)"
|
||||||
|
if [ "$INSTALL_UPDATER" = 1 ]; then
|
||||||
|
say "Installing the host updater"
|
||||||
|
UPD_ARGS=(--dir "$INSTALL_DIR" --repo-url "$REPO_URL" --ref "$REF" --registry "$REGISTRY")
|
||||||
|
[ "$DO_BUILD" = 1 ] && UPD_ARGS+=(--build)
|
||||||
|
if bash "$INSTALL_DIR/scripts/install-updater.sh" "${UPD_ARGS[@]}"; then
|
||||||
|
UPDATER_STATE="enabled — checks C2 every ~15 min; 'Update Node Software' in the admin UI"
|
||||||
|
else
|
||||||
|
warn "host updater NOT installed — this node cannot be updated remotely"
|
||||||
|
UPDATER_STATE="FAILED to install (see above)"
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ── 8. Summary ──────────────────────────────────────────────────────────────
|
||||||
IP="$(hostname -I 2>/dev/null | awk '{print $1}')"
|
IP="$(hostname -I 2>/dev/null | awk '{print $1}')"
|
||||||
cat <<EOF
|
cat <<EOF
|
||||||
|
|
||||||
@@ -525,20 +605,18 @@ $(printf "${G}Node '%s' installed at %s${N}" "$NODE_ID" "$INSTALL_DIR")
|
|||||||
images $([ "$DO_BUILD" = 1 ] && echo "built locally" || echo "pulled from $REGISTRY")
|
images $([ "$DO_BUILD" = 1 ] && echo "built locally" || echo "pulled from $REGISTRY")
|
||||||
dashboard http://${IP:-<node-ip>}/ (user: ${DASHBOARD_USER})
|
dashboard http://${IP:-<node-ip>}/ (user: ${DASHBOARD_USER})
|
||||||
logs cd $INSTALL_DIR && docker compose logs -f edge-node
|
logs cd $INSTALL_DIR && docker compose logs -f edge-node
|
||||||
|
updater ${UPDATER_STATE}
|
||||||
|
api key $([ -n "$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null)" ] && echo "present" || echo "NOT YET — node fetches it once approved")
|
||||||
|
|
||||||
NEXT — an admin must approve this node before it can do anything:
|
NEXT — assign this node a radio system at ${APP_URL}/nodes/${NODE_ID}
|
||||||
|
|
||||||
1. Open <app-url>/settings/nodes
|
|
||||||
2. Approve "$NODE_ID"
|
|
||||||
3. Assign it a radio system
|
|
||||||
|
|
||||||
EOF
|
EOF
|
||||||
if [ "${GENERATED_DASH:-0}" = 1 ]; then
|
if [ "${GENERATED_DASH:-0}" = 1 ]; then
|
||||||
printf "${Y}Generated dashboard password (shown once): %s${N}\n\n" "$DASHBOARD_PASS"
|
printf "${Y}Generated dashboard password (shown once): %s${N}\n\n" "$DASHBOARD_PASS"
|
||||||
fi
|
fi
|
||||||
if [ ! -s "$CREDS" ]; then
|
if [ -z "$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null)" ]; then
|
||||||
if [ -s "$PICKUP_FILE" ]; then
|
if [ -s "$PICKUP_FILE" ]; then
|
||||||
printf "${Y}This node has no api_key yet. After an admin approves it, re-run the same\ninstall command — it reuses configs/pickup_secret and will collect the key\n(no --token needed for the re-run).${N}\n\n"
|
printf "${Y}This node has no api_key yet. Once an admin approves it at %s/settings/nodes\nthe node fetches the key and comes online by itself (no re-run needed).${N}\n\n" "$APP_URL"
|
||||||
else
|
else
|
||||||
printf "${Y}This node has no api_key and no saved pickup secret, so a plain re-run cannot\nfix it. Re-run with --token DRB-… to enroll; or, if the node is already\napproved, have an admin Reissue key and write it into\n%s by hand.${N}\n\n" "$CREDS"
|
printf "${Y}This node has no api_key and no saved pickup secret, so a plain re-run cannot\nfix it. Re-run with --token DRB-… to enroll; or, if the node is already\napproved, have an admin Reissue key and write it into\n%s by hand.${N}\n\n" "$CREDS"
|
||||||
fi
|
fi
|
||||||
|
|||||||
@@ -31,6 +31,10 @@ RUN ./install.sh -f
|
|||||||
COPY requirements.txt /tmp/requirements.txt
|
COPY requirements.txt /tmp/requirements.txt
|
||||||
RUN pip3 install --no-cache-dir -r /tmp/requirements.txt
|
RUN pip3 install --no-cache-dir -r /tmp/requirements.txt
|
||||||
|
|
||||||
|
# DRB's trunking module for P25 conventional channels (node-26#24): reports
|
||||||
|
# encryption + NAC and applies the NAC filter, which stock tk_p25 does not.
|
||||||
|
COPY op25_addons/tk_drb_conv.py /op25/op25/gr-op25_repeater/apps/tk_drb_conv.py
|
||||||
|
|
||||||
# Create the run_multi-rx_service.sh script
|
# Create the run_multi-rx_service.sh script
|
||||||
COPY run_multi-rx_service.sh /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
|
COPY run_multi-rx_service.sh /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
|
||||||
RUN chmod +x /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
|
RUN chmod +x /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
|
||||||
|
|||||||
@@ -0,0 +1,118 @@
|
|||||||
|
"""
|
||||||
|
P25 conventional OP25 config (node-26#24).
|
||||||
|
|
||||||
|
A conventional P25 channel is a fixed voice frequency with no control channel:
|
||||||
|
radios key up on it directly, C4FM (fsk4), and identify themselves in the link
|
||||||
|
control of each LDU1 (talkgroup + unit id) and the ESS of each LDU2
|
||||||
|
(encryption algorithm + key id). Building a `tk_p25.py` trunking config for one
|
||||||
|
— what "P25" always meant before — makes OP25 hunt forever for a control
|
||||||
|
channel that does not exist, so nothing is ever decoded.
|
||||||
|
|
||||||
|
Layout this produces, per boatbod OP25 gr310 `p25_conventional_example.json`:
|
||||||
|
|
||||||
|
* one non-tunable device parked so every channel sits inside its passband;
|
||||||
|
* one channel per frequency, `trunking_sysname: ""` (so the trunking module
|
||||||
|
treats it as conventional), `demod_type: fsk4`;
|
||||||
|
* trunking module `tk_drb_conv.py` with no chans — our thin subclass of
|
||||||
|
tk_p25.rx_ctl that reports encryption and the NAC for conventional
|
||||||
|
channels and applies the per-channel NAC filter (tk_p25 itself does
|
||||||
|
neither for FDMA conventional);
|
||||||
|
* channel N's audio to UDP 23456+N; with two channels audio.py runs in
|
||||||
|
two-channel mode (left = first channel, right = second).
|
||||||
|
|
||||||
|
Pure: no file I/O, so it is unit-testable outside the container.
|
||||||
|
"""
|
||||||
|
from typing import Any, Dict, List, Tuple
|
||||||
|
|
||||||
|
from models import ChannelConfig, ConventionalChannel, DeviceConfig, TerminalConfig
|
||||||
|
|
||||||
|
# Two is what one audio.py instance can carry (its A and B sockets). A third
|
||||||
|
# channel would need a second audio pipeline into PulseAudio.
|
||||||
|
MAX_CONVENTIONAL_CHANNELS = 2
|
||||||
|
|
||||||
|
AUDIO_UDP_BASE_PORT = 23456
|
||||||
|
|
||||||
|
DEVICE_RATE = 1_920_000
|
||||||
|
DEVICE_USABLE_BW_PCT = 0.85
|
||||||
|
# Half the occupied bandwidth of a 12.5 kHz P25 channel.
|
||||||
|
CHANNEL_HALF_BW_HZ = 6_250
|
||||||
|
# Keep the RTL-SDR's DC spike off the channels. A single channel (or two very
|
||||||
|
# close together) is offset from the centre; channels further apart than
|
||||||
|
# 2x this already straddle the spike, so the centre goes between them.
|
||||||
|
DC_SPIKE_OFFSET_HZ = 250_000
|
||||||
|
DC_SPIKE_CLEARANCE_HZ = 50_000
|
||||||
|
|
||||||
|
CONVENTIONAL_TRUNKING_MODULE = "tk_drb_conv.py"
|
||||||
|
|
||||||
|
|
||||||
|
def device_center(freqs: List[int]) -> int:
|
||||||
|
lo, hi = min(freqs), max(freqs)
|
||||||
|
mid = (lo + hi) // 2
|
||||||
|
if (hi - lo) // 2 >= DC_SPIKE_CLEARANCE_HZ:
|
||||||
|
return mid
|
||||||
|
return mid + DC_SPIKE_OFFSET_HZ
|
||||||
|
|
||||||
|
|
||||||
|
def validate_channels(channels: List[ConventionalChannel]) -> None:
|
||||||
|
"""Raise ValueError with an operator-readable reason."""
|
||||||
|
if not channels:
|
||||||
|
raise ValueError("P25 conventional needs at least one channel frequency.")
|
||||||
|
if len(channels) > MAX_CONVENTIONAL_CHANNELS:
|
||||||
|
raise ValueError(
|
||||||
|
f"P25 conventional supports up to {MAX_CONVENTIONAL_CHANNELS} channels on one SDR "
|
||||||
|
f"(got {len(channels)})."
|
||||||
|
)
|
||||||
|
freqs = [c.frequency for c in channels]
|
||||||
|
if len(set(freqs)) != len(freqs):
|
||||||
|
raise ValueError("The same conventional frequency is listed twice.")
|
||||||
|
for c in channels:
|
||||||
|
if c.nac is not None and not (0 < c.nac <= 0xFFF):
|
||||||
|
raise ValueError(f"NAC {c.nac:#x} on {c.frequency / 1e6:.5f} MHz is outside 0x001-0xFFF.")
|
||||||
|
center = device_center(freqs)
|
||||||
|
half = DEVICE_RATE * DEVICE_USABLE_BW_PCT / 2
|
||||||
|
for f in freqs:
|
||||||
|
if abs(f - center) + CHANNEL_HALF_BW_HZ > half:
|
||||||
|
span = (max(freqs) - min(freqs)) / 1e6
|
||||||
|
raise ValueError(
|
||||||
|
f"These channels are {span:.3f} MHz apart — too far for one SDR "
|
||||||
|
f"(max ~{(2 * (half - CHANNEL_HALF_BW_HZ)) / 1e6:.2f} MHz)."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def build_conventional_config(
|
||||||
|
system_name: str,
|
||||||
|
channels: List[ConventionalChannel],
|
||||||
|
gains: str,
|
||||||
|
ppm: float,
|
||||||
|
) -> Tuple[Dict[str, Any], bool]:
|
||||||
|
"""Return (OP25 config dict, two_channel_audio)."""
|
||||||
|
validate_channels(channels)
|
||||||
|
freqs = [c.frequency for c in channels]
|
||||||
|
center = device_center(freqs)
|
||||||
|
|
||||||
|
chans = []
|
||||||
|
for i, c in enumerate(channels):
|
||||||
|
chans.append(ChannelConfig(
|
||||||
|
name=c.name or f"{system_name} {c.frequency / 1e6:.4f}",
|
||||||
|
trunking_sysname="",
|
||||||
|
enable_analog="off",
|
||||||
|
demod_type="fsk4",
|
||||||
|
filter_type="rc",
|
||||||
|
frequency=c.frequency,
|
||||||
|
meta_stream_name="",
|
||||||
|
destination=f"udp://127.0.0.1:{AUDIO_UDP_BASE_PORT + i}",
|
||||||
|
nac=c.nac,
|
||||||
|
crypt_behavior=2,
|
||||||
|
))
|
||||||
|
|
||||||
|
device = DeviceConfig(
|
||||||
|
gains=gains, ppm=ppm, tunable=False, frequency=center,
|
||||||
|
rate=DEVICE_RATE, usable_bw_pct=DEVICE_USABLE_BW_PCT,
|
||||||
|
)
|
||||||
|
config = {
|
||||||
|
"channels": [ch.model_dump() for ch in chans],
|
||||||
|
"devices": [device.model_dump()],
|
||||||
|
"trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
|
||||||
|
"terminal": TerminalConfig().model_dump(),
|
||||||
|
}
|
||||||
|
return config, len(chans) > 1
|
||||||
@@ -2,7 +2,26 @@ import os
|
|||||||
from internal.op25_liq_template import liquidsoap_config_template
|
from internal.op25_liq_template import liquidsoap_config_template
|
||||||
from models import IcecastConfig
|
from models import IcecastConfig
|
||||||
|
|
||||||
def generate_liquid_script(config: IcecastConfig):
|
# P25 conventional with two channels (node-26#24): OP25 sends channel A's
|
||||||
|
# audio to UDP 23456 and channel B's to 23457, and audio.py -2 puts them on the
|
||||||
|
# left and right of the stereo stream. Every output below is mean(input) — mono
|
||||||
|
# — so a lone talker on one side would come out 6 dB down; doubling first keeps
|
||||||
|
# the common case (one channel keyed) at the same level as single-channel mode.
|
||||||
|
# Liquidsoap mixes in float, so the doubling cannot clip before the output.
|
||||||
|
_TWO_CHANNEL_AUDIO_ARGS = " -2"
|
||||||
|
_TWO_CHANNEL_MIX = "input = amplify(2.0, input) # two conventional channels, L=A R=B"
|
||||||
|
|
||||||
|
|
||||||
|
def render_liquid_script(config: IcecastConfig, two_channel: bool = False) -> str:
|
||||||
|
content = liquidsoap_config_template
|
||||||
|
for key, value in config.model_dump().items():
|
||||||
|
content = content.replace(f"${{{key}}}", str(value))
|
||||||
|
content = content.replace("${audio_py_args}", _TWO_CHANNEL_AUDIO_ARGS if two_channel else "")
|
||||||
|
content = content.replace("${two_channel_mix}", _TWO_CHANNEL_MIX if two_channel else "")
|
||||||
|
return content
|
||||||
|
|
||||||
|
|
||||||
|
def generate_liquid_script(config: IcecastConfig, two_channel: bool = False):
|
||||||
"""
|
"""
|
||||||
Generates the "*.liq" file that's run by OP25 on startup.
|
Generates the "*.liq" file that's run by OP25 on startup.
|
||||||
|
|
||||||
@@ -13,13 +32,7 @@ def generate_liquid_script(config: IcecastConfig):
|
|||||||
Keys should match the variable names in the template (e.g., 'icecast_host').
|
Keys should match the variable names in the template (e.g., 'icecast_host').
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
content = liquidsoap_config_template
|
content = render_liquid_script(config, two_channel)
|
||||||
# Replace variables
|
|
||||||
for key, value in config.model_dump().items():
|
|
||||||
placeholder = f"${{{key}}}"
|
|
||||||
# Ensure the value is converted to string for replacement
|
|
||||||
content = content.replace(placeholder, str(value))
|
|
||||||
print(f" - Replaced placeholder {placeholder}")
|
|
||||||
|
|
||||||
# Write the processed content to the output path
|
# Write the processed content to the output path
|
||||||
output_path = "/configs/op25.liq"
|
output_path = "/configs/op25.liq"
|
||||||
|
|||||||
@@ -17,8 +17,9 @@ ICE_DESCRIPTION = "${icecast_description}"
|
|||||||
ICE_GENRE = "${icecast_genre}"
|
ICE_GENRE = "${icecast_genre}"
|
||||||
# ==========================================================
|
# ==========================================================
|
||||||
|
|
||||||
input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s"))
|
input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s${audio_py_args}"))
|
||||||
# Consider increasing the buffer value on slow systems such as RPi3. e.g. buffer=0.25
|
# Consider increasing the buffer value on slow systems such as RPi3. e.g. buffer=0.25
|
||||||
|
${two_channel_mix}
|
||||||
|
|
||||||
# Compression
|
# Compression
|
||||||
input = compress(input, attack = 2.0, gain = 0.0, knee = 13.0, ratio = 2.0, release = 12.3, threshold = -18.0)
|
input = compress(input, attack = 2.0, gain = 0.0, knee = 13.0, ratio = 2.0, release = 12.3, threshold = -18.0)
|
||||||
|
|||||||
@@ -49,12 +49,28 @@ class IcecastConfig(BaseModel):
|
|||||||
icecast_genre: Optional[str] = "Public Safety"
|
icecast_genre: Optional[str] = "Public Safety"
|
||||||
|
|
||||||
|
|
||||||
|
class P25Mode(str, Enum):
|
||||||
|
TRUNKED = "trunked"
|
||||||
|
CONVENTIONAL = "conventional"
|
||||||
|
|
||||||
|
|
||||||
|
class ConventionalChannel(BaseModel):
|
||||||
|
"""One fixed P25 conventional voice channel (node-26#24)."""
|
||||||
|
frequency: int # Hz
|
||||||
|
name: str = ""
|
||||||
|
nac: Optional[int] = None # 0x001-0xFFF; None = accept any NAC
|
||||||
|
|
||||||
|
|
||||||
class ConfigGenerator(BaseModel):
|
class ConfigGenerator(BaseModel):
|
||||||
type: DecodeMode
|
type: DecodeMode
|
||||||
systemName: str
|
systemName: str
|
||||||
# P25: control channel list (Hz). NBFM: channels[0] is the one analog
|
# P25 trunked: control channel list (Hz). NBFM: channels[0] is the one
|
||||||
# frequency to listen to (Hz).
|
# analog frequency to listen to (Hz). P25 conventional: unused — see
|
||||||
channels: List[Union[int, str]]
|
# conv_channels.
|
||||||
|
channels: List[Union[int, str]] = []
|
||||||
|
# P25 only. Missing = trunked, so every existing caller is unchanged.
|
||||||
|
p25_mode: P25Mode = P25Mode.TRUNKED
|
||||||
|
conv_channels: List[ConventionalChannel] = []
|
||||||
tags: Optional[List[TalkgroupTag]] = None
|
tags: Optional[List[TalkgroupTag]] = None
|
||||||
whitelist: Optional[List[int]] = None
|
whitelist: Optional[List[int]] = None
|
||||||
icecastConfig: Optional[IcecastConfig] = None
|
icecastConfig: Optional[IcecastConfig] = None
|
||||||
@@ -84,6 +100,13 @@ class ChannelConfig(BaseModel):
|
|||||||
# Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
|
# Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
|
||||||
nbfm_squelch_threshold: Optional[float] = None
|
nbfm_squelch_threshold: Optional[float] = None
|
||||||
destination: Optional[str] = "udp://127.0.0.1:23456"
|
destination: Optional[str] = "udp://127.0.0.1:23456"
|
||||||
|
# P25 conventional only, read by tk_drb_conv.py (not by multi_rx): the NAC
|
||||||
|
# to lock the frame assembler to. multi_rx passes the whole channel dict
|
||||||
|
# through, so an extra key is harmless.
|
||||||
|
nac: Optional[int] = None
|
||||||
|
# Set on the channel itself for conventional (multi_rx forwards it to the
|
||||||
|
# frame assembler). 2 = drop encrypted voice instead of decoding noise.
|
||||||
|
crypt_behavior: Optional[int] = None
|
||||||
tracking_threshold: Optional[int] = 120
|
tracking_threshold: Optional[int] = 120
|
||||||
tracking_feedback: Optional[float] = 0.75
|
tracking_feedback: Optional[float] = 0.75
|
||||||
excess_bw: Optional[float] = 0.2
|
excess_bw: Optional[float] = 0.2
|
||||||
|
|||||||
@@ -8,7 +8,8 @@ import json
|
|||||||
import threading
|
import threading
|
||||||
from collections import deque
|
from collections import deque
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
|
from models import ConfigGenerator, DecodeMode, P25Mode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
|
||||||
|
from internal.conventional_config import build_conventional_config, CONVENTIONAL_TRUNKING_MODULE
|
||||||
from internal.logger import create_logger
|
from internal.logger import create_logger
|
||||||
from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
|
from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
|
||||||
from internal.liquidsoap_config_utils import generate_liquid_script
|
from internal.liquidsoap_config_utils import generate_liquid_script
|
||||||
@@ -178,7 +179,20 @@ def create_op25_router():
|
|||||||
@router.post("/generate-config")
|
@router.post("/generate-config")
|
||||||
async def generate_config(generator: ConfigGenerator):
|
async def generate_config(generator: ConfigGenerator):
|
||||||
try:
|
try:
|
||||||
if generator.type == DecodeMode.P25:
|
if generator.type == DecodeMode.P25 and generator.p25_mode == P25Mode.CONVENTIONAL:
|
||||||
|
# node-26#24: fixed voice channels, no control channel to hunt.
|
||||||
|
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
||||||
|
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
||||||
|
try:
|
||||||
|
config_dict, two_channel = build_conventional_config(
|
||||||
|
generator.systemName, generator.conv_channels, preset["gains"], ppm,
|
||||||
|
)
|
||||||
|
except ValueError as e:
|
||||||
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
if generator.icecastConfig:
|
||||||
|
generate_liquid_script(generator.icecastConfig, two_channel=two_channel)
|
||||||
|
|
||||||
|
elif generator.type == DecodeMode.P25:
|
||||||
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
||||||
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
||||||
channels = [ChannelConfig(
|
channels = [ChannelConfig(
|
||||||
@@ -236,9 +250,9 @@ def create_op25_router():
|
|||||||
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
||||||
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
||||||
freq = int(float(generator.channels[0]))
|
freq = int(float(generator.channels[0]))
|
||||||
# One fixed conventional channel: no trunking section, so
|
# One fixed conventional channel: a trunking module with no
|
||||||
# multi_rx adds its default tk_p25 with no chans and reports
|
# chans, so the channel is reported as `conventional` on the
|
||||||
# the channel as `conventional` on the HTTP terminal.
|
# HTTP terminal.
|
||||||
channels = [ChannelConfig(
|
channels = [ChannelConfig(
|
||||||
name=generator.systemName,
|
name=generator.systemName,
|
||||||
trunking_sysname="",
|
trunking_sysname="",
|
||||||
@@ -259,6 +273,11 @@ def create_op25_router():
|
|||||||
config_dict = {
|
config_dict = {
|
||||||
"channels": [channel.dict() for channel in channels],
|
"channels": [channel.dict() for channel in channels],
|
||||||
"devices": [device.dict() for device in devices],
|
"devices": [device.dict() for device in devices],
|
||||||
|
# Same module as P25 conventional (node-26#24): it reports
|
||||||
|
# the channel as `conventional` whatever tk_p25 version the
|
||||||
|
# image was built with, which is what the edge node's
|
||||||
|
# single-channel attribution keys on (node-26#20).
|
||||||
|
"trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
|
||||||
"terminal": TerminalConfig().dict(),
|
"terminal": TerminalConfig().dict(),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,263 @@
|
|||||||
|
"""
|
||||||
|
tk_drb_conv — DRB's trunking module for P25 CONVENTIONAL channels (node-26#24).
|
||||||
|
|
||||||
|
Copied into OP25's apps directory by the Dockerfile and named as the trunking
|
||||||
|
module by op25-container's conventional config, so multi_rx imports it in place
|
||||||
|
of tk_p25. It is tk_p25.rx_ctl with the conventional-channel handling replaced;
|
||||||
|
trunked receivers (none in a conventional config) still go to the base class.
|
||||||
|
|
||||||
|
Why not stock tk_p25: for an FDMA conventional channel it reports the talkgroup
|
||||||
|
and unit id from LDU1 link control, but
|
||||||
|
|
||||||
|
* never reports encryption — the encrypted/algid/keyid JSON OP25 sends after
|
||||||
|
every LDU2 carries no grpaddr, and tk_p25 only records `encrypted` when
|
||||||
|
grpaddr != 0 — so an encrypted channel looks exactly like a quiet one;
|
||||||
|
* never reports the NAC it is hearing, and has no per-channel NAC filter;
|
||||||
|
* has no "voice frames are arriving" signal of its own, so a transmission
|
||||||
|
whose link control is unreadable is invisible to the console.
|
||||||
|
|
||||||
|
All three matter to the edge node: it attributes audio to a channel from what
|
||||||
|
this module reports, and must tell the owner when a channel is encrypted rather
|
||||||
|
than silently recording nothing.
|
||||||
|
|
||||||
|
Each conventional channel row in `channel_update` keeps every key tk_p25 sends
|
||||||
|
(freq, tgid, srcaddr, tag, encrypted, conventional, ...) and adds:
|
||||||
|
|
||||||
|
voice True while voice frames are arriving (HDU/LDU1/LDU2 seen,
|
||||||
|
no TDU or sync timeout since)
|
||||||
|
nac last NAC heard on the channel (int) or None
|
||||||
|
nac_filter the configured NAC filter (int) or None
|
||||||
|
algid, keyid ALGID/key id of the current call (None when clear/idle)
|
||||||
|
enc_calls, clear_calls calls heard since start, by encryption
|
||||||
|
last_enc_algid, last_enc_keyid, last_enc_time, last_clear_time
|
||||||
|
|
||||||
|
Message types handled (lower 16 bits of msg.type(), P25 protocol only):
|
||||||
|
0 HDU -> voice start
|
||||||
|
3, 15 TDU / TDULC -> transmission end
|
||||||
|
19 LDU1 link control -> tgid, srcaddr, service options (E bit)
|
||||||
|
-3 LDU2 ESS JSON -> encrypted / algid / keyid
|
||||||
|
-1 sync timeout -> transmission end
|
||||||
|
"""
|
||||||
|
import ctypes
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
|
||||||
|
import tk_p25
|
||||||
|
from helper_funcs import from_dict, get_ordinals
|
||||||
|
from log_ts import log_ts
|
||||||
|
|
||||||
|
ALGID_CLEAR = 0x80
|
||||||
|
SVCOPT_ENCRYPTED = 0x40
|
||||||
|
LCW_PROTECTED = 0x80
|
||||||
|
LCO_GROUP_VOICE = 0x00
|
||||||
|
DUID_HDU, DUID_TDU, DUID_TDULC, DUID_LCW = 0, 3, 15, 19
|
||||||
|
M_TIMEOUT, M_JSON_DATA = -1, -3
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_nac(value):
|
||||||
|
if value is None or value == "":
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
nac = int(value, 16) if isinstance(value, str) else int(value)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
return nac if 0 < nac <= 0xFFF else None
|
||||||
|
|
||||||
|
|
||||||
|
class rx_ctl(tk_p25.rx_ctl):
|
||||||
|
def __init__(self, *args, **kwargs):
|
||||||
|
super().__init__(*args, **kwargs)
|
||||||
|
self.drb_conv = {} # msgq_id -> state dict
|
||||||
|
|
||||||
|
def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
|
||||||
|
super().add_receiver(msgq_id, config, meta_q=meta_q, freq=freq)
|
||||||
|
rcvr = self.receivers.get(msgq_id)
|
||||||
|
if rcvr is not None and rcvr.get('rx_rcvr') is not None:
|
||||||
|
return # trunked receiver: base class owns it
|
||||||
|
self.drb_conv[msgq_id] = {
|
||||||
|
'name': from_dict(config, 'name', str(msgq_id)),
|
||||||
|
'freq': freq,
|
||||||
|
'nac_filter': _parse_nac(config.get('nac')),
|
||||||
|
'nac': None,
|
||||||
|
'voice': False,
|
||||||
|
'tgid': None,
|
||||||
|
'srcaddr': 0,
|
||||||
|
'svc_encrypted': False,
|
||||||
|
'ess_encrypted': False,
|
||||||
|
'algid': None,
|
||||||
|
'keyid': None,
|
||||||
|
'call_start': None,
|
||||||
|
'call_encrypted': False,
|
||||||
|
'logged_tgid': None,
|
||||||
|
'prev_type': None,
|
||||||
|
'enc_calls': 0,
|
||||||
|
'clear_calls': 0,
|
||||||
|
'last_enc_algid': None,
|
||||||
|
'last_enc_keyid': None,
|
||||||
|
'last_enc_time': None,
|
||||||
|
'last_clear_time': None,
|
||||||
|
'last_voice_time': None,
|
||||||
|
}
|
||||||
|
|
||||||
|
def post_init(self):
|
||||||
|
super().post_init()
|
||||||
|
for msgq_id, cs in self.drb_conv.items():
|
||||||
|
if cs['nac_filter'] is not None and self.fa_ctrl is not None:
|
||||||
|
self.fa_ctrl({'tuner': msgq_id, 'cmd': 'set_nac', 'nac': cs['nac_filter']})
|
||||||
|
sys.stderr.write("%s [%d] conventional NAC filter 0x%03x\n" % (log_ts.get(), msgq_id, cs['nac_filter']))
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _encrypted(self, cs):
|
||||||
|
return cs['svc_encrypted'] or cs['ess_encrypted']
|
||||||
|
|
||||||
|
def _voice_start(self, cs, now):
|
||||||
|
if not cs['voice']:
|
||||||
|
cs['voice'] = True
|
||||||
|
cs['call_start'] = now
|
||||||
|
cs['call_encrypted'] = False
|
||||||
|
cs['logged_tgid'] = None
|
||||||
|
cs['last_voice_time'] = now
|
||||||
|
|
||||||
|
def _voice_end(self, cs, now):
|
||||||
|
if cs['voice']:
|
||||||
|
if cs['call_encrypted']:
|
||||||
|
cs['enc_calls'] += 1
|
||||||
|
cs['last_enc_time'] = now
|
||||||
|
else:
|
||||||
|
cs['clear_calls'] += 1
|
||||||
|
cs['last_clear_time'] = now
|
||||||
|
cs['voice'] = False
|
||||||
|
cs['tgid'] = None
|
||||||
|
cs['srcaddr'] = 0
|
||||||
|
cs['svc_encrypted'] = False
|
||||||
|
cs['ess_encrypted'] = False
|
||||||
|
cs['algid'] = None
|
||||||
|
cs['keyid'] = None
|
||||||
|
cs['call_start'] = None
|
||||||
|
cs['logged_tgid'] = None
|
||||||
|
|
||||||
|
def _note_encryption(self, cs, now):
|
||||||
|
if self._encrypted(cs):
|
||||||
|
cs['call_encrypted'] = True
|
||||||
|
if cs['algid'] is not None:
|
||||||
|
cs['last_enc_algid'] = cs['algid']
|
||||||
|
cs['last_enc_keyid'] = cs['keyid']
|
||||||
|
cs['last_enc_time'] = now
|
||||||
|
|
||||||
|
def _maybe_log_call(self, msgq_id, cs):
|
||||||
|
# One call_log entry per (call, talkgroup), and none for an encrypted
|
||||||
|
# call: it will produce no audio, so a "grant" would only mislead the
|
||||||
|
# edge node's attribution.
|
||||||
|
if cs['tgid'] is None or self._encrypted(cs) or cs['logged_tgid'] == cs['tgid']:
|
||||||
|
return
|
||||||
|
cs['logged_tgid'] = cs['tgid']
|
||||||
|
try:
|
||||||
|
self.log_call(0, msgq_id, cs['freq'], 0, 0, cs['tgid'], cs['name'], cs['srcaddr'], '')
|
||||||
|
except Exception as e: # older/newer tk_p25 log_call signature
|
||||||
|
sys.stderr.write("%s [%d] conventional call_log failed: %s\n" % (log_ts.get(), msgq_id, e))
|
||||||
|
|
||||||
|
def _process_conv(self, msgq_id, cs, m_type, msg, now):
|
||||||
|
if m_type == M_TIMEOUT or m_type in (DUID_TDU, DUID_TDULC):
|
||||||
|
self._voice_end(cs, now)
|
||||||
|
elif m_type == DUID_HDU:
|
||||||
|
s = msg.to_string()
|
||||||
|
cs['nac'] = get_ordinals(s[0:2]) if len(s) >= 2 else cs['nac']
|
||||||
|
self._voice_start(cs, now)
|
||||||
|
elif m_type == DUID_LCW:
|
||||||
|
s = msg.to_string()
|
||||||
|
if len(s) < 11:
|
||||||
|
return
|
||||||
|
cs['nac'] = get_ordinals(s[0:2])
|
||||||
|
if cs['prev_type'] in (DUID_TDU, DUID_TDULC):
|
||||||
|
return # TDULC's link control follows its TDU: the call is over
|
||||||
|
lcw = s[2:]
|
||||||
|
self._voice_start(cs, now)
|
||||||
|
pb_sf_lco = get_ordinals(lcw[0:1])
|
||||||
|
if pb_sf_lco & LCW_PROTECTED:
|
||||||
|
# Link control itself is encrypted — the call certainly is.
|
||||||
|
cs['svc_encrypted'] = True
|
||||||
|
elif (pb_sf_lco & 0x3f) == LCO_GROUP_VOICE:
|
||||||
|
svcopts = get_ordinals(lcw[2:3])
|
||||||
|
ga = get_ordinals(lcw[4:6])
|
||||||
|
sa = get_ordinals(lcw[6:9])
|
||||||
|
cs['svc_encrypted'] = bool(svcopts & SVCOPT_ENCRYPTED)
|
||||||
|
if ga:
|
||||||
|
cs['tgid'] = ga
|
||||||
|
if sa:
|
||||||
|
cs['srcaddr'] = sa
|
||||||
|
self._note_encryption(cs, now)
|
||||||
|
self._maybe_log_call(msgq_id, cs)
|
||||||
|
elif m_type == M_JSON_DATA:
|
||||||
|
try:
|
||||||
|
js = json.loads(msg.to_string())
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
return
|
||||||
|
if 'encrypted' in js:
|
||||||
|
self._voice_start(cs, now)
|
||||||
|
algid = from_dict(js, 'algid', None)
|
||||||
|
cs['ess_encrypted'] = bool(js.get('encrypted')) and algid != ALGID_CLEAR
|
||||||
|
cs['algid'] = algid if cs['ess_encrypted'] else None
|
||||||
|
cs['keyid'] = from_dict(js, 'keyid', None) if cs['ess_encrypted'] else None
|
||||||
|
self._note_encryption(cs, now)
|
||||||
|
grpaddr = from_dict(js, 'grpaddr', 0)
|
||||||
|
if grpaddr:
|
||||||
|
cs['tgid'] = grpaddr
|
||||||
|
cs['srcaddr'] = from_dict(js, 'srcaddr', 0) or cs['srcaddr']
|
||||||
|
self._maybe_log_call(msgq_id, cs)
|
||||||
|
|
||||||
|
def process_qmsg(self, msg):
|
||||||
|
m_proto = ctypes.c_int16(msg.type() >> 16).value
|
||||||
|
m_rxid = int(msg.arg1()) >> 1
|
||||||
|
if m_proto != 0 or m_rxid not in self.drb_conv:
|
||||||
|
return super().process_qmsg(msg)
|
||||||
|
m_type = ctypes.c_int16(msg.type() & 0xffff).value
|
||||||
|
cs = self.drb_conv[m_rxid]
|
||||||
|
try:
|
||||||
|
self._process_conv(m_rxid, cs, m_type, msg, time.time())
|
||||||
|
except Exception as e: # never take multi_rx down over a bad frame
|
||||||
|
sys.stderr.write("%s [%d] conventional message type %d failed: %s\n" % (log_ts.get(), m_rxid, m_type, e))
|
||||||
|
cs['prev_type'] = m_type
|
||||||
|
|
||||||
|
def get_chan_status(self):
|
||||||
|
d = json.loads(super().get_chan_status())
|
||||||
|
ids = list(d.get('channels') or [])
|
||||||
|
for msgq_id, cs in self.drb_conv.items():
|
||||||
|
key = str(msgq_id)
|
||||||
|
row = d.get(key) or {}
|
||||||
|
row.update({
|
||||||
|
'freq': cs['freq'],
|
||||||
|
'name': cs['name'],
|
||||||
|
'system': cs['name'],
|
||||||
|
'tdma': 0,
|
||||||
|
'tgid': cs['tgid'],
|
||||||
|
'tag': cs['name'],
|
||||||
|
'srcaddr': cs['srcaddr'],
|
||||||
|
'srctag': '',
|
||||||
|
'svcopts': 0,
|
||||||
|
'encrypted': 1 if self._encrypted(cs) else 0,
|
||||||
|
'emergency': 0,
|
||||||
|
'hold_tgid': 0,
|
||||||
|
'mode': None,
|
||||||
|
'stream': '',
|
||||||
|
'msgqid': msgq_id,
|
||||||
|
'conventional': True,
|
||||||
|
'voice': cs['voice'],
|
||||||
|
'nac': cs['nac'],
|
||||||
|
'nac_filter': cs['nac_filter'],
|
||||||
|
'algid': cs['algid'],
|
||||||
|
'keyid': cs['keyid'],
|
||||||
|
'enc_calls': cs['enc_calls'],
|
||||||
|
'clear_calls': cs['clear_calls'],
|
||||||
|
'last_enc_algid': cs['last_enc_algid'],
|
||||||
|
'last_enc_keyid': cs['last_enc_keyid'],
|
||||||
|
'last_enc_time': cs['last_enc_time'],
|
||||||
|
'last_clear_time': cs['last_clear_time'],
|
||||||
|
'last_voice_time': cs['last_voice_time'],
|
||||||
|
})
|
||||||
|
d[key] = row
|
||||||
|
if key not in ids:
|
||||||
|
ids.append(key)
|
||||||
|
d['channels'] = ids
|
||||||
|
return json.dumps(d)
|
||||||
@@ -0,0 +1,305 @@
|
|||||||
|
"""
|
||||||
|
P25 conventional (node-26#24): config generation, the liquidsoap two-channel
|
||||||
|
variant, and the tk_drb_conv state machine — all without OP25/GNU Radio.
|
||||||
|
|
||||||
|
tk_drb_conv imports tk_p25/helper_funcs/log_ts from OP25's apps dir; those are
|
||||||
|
stubbed here with the minimum of tk_p25.rx_ctl it relies on. The message
|
||||||
|
shapes fed in are the ones boatbod gr310 p25p1_fdma.cc sends: LCW = 2-byte NAC
|
||||||
|
+ 9 LC bytes (type 19), LDU2 ESS = JSON (type -3), TDU = 2-byte NAC (type 3/15),
|
||||||
|
sync timeout = type -1.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import types
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
HERE = os.path.dirname(__file__)
|
||||||
|
sys.path.insert(0, os.path.join(HERE, "..", "app"))
|
||||||
|
sys.path.insert(0, os.path.join(HERE, "..", "op25_addons"))
|
||||||
|
|
||||||
|
from fastapi.testclient import TestClient # noqa: E402
|
||||||
|
from fastapi import FastAPI # noqa: E402
|
||||||
|
|
||||||
|
from models import ConventionalChannel # noqa: E402
|
||||||
|
from internal import conventional_config as cc # noqa: E402
|
||||||
|
from internal.liquidsoap_config_utils import render_liquid_script # noqa: E402
|
||||||
|
from models import IcecastConfig # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Config generation
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
WP = [
|
||||||
|
ConventionalChannel(frequency=460_400_000, name="PD1 Dispatch"),
|
||||||
|
ConventionalChannel(frequency=460_425_000, name="PD3 Tac", nac=0x293),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_white_plains_two_channels_one_sdr():
|
||||||
|
config, two = cc.build_conventional_config("White Plains", WP, "lna:34", 0.0)
|
||||||
|
assert two is True
|
||||||
|
assert "trunking" in config and config["trunking"]["module"] == "tk_drb_conv.py"
|
||||||
|
assert config["trunking"]["chans"] == [] # nothing for tk_p25 to hunt
|
||||||
|
a, b = config["channels"]
|
||||||
|
for ch in (a, b):
|
||||||
|
assert ch["demod_type"] == "fsk4"
|
||||||
|
assert ch["trunking_sysname"] == ""
|
||||||
|
assert ch["enable_analog"] == "off"
|
||||||
|
assert ch["crypt_behavior"] == 2
|
||||||
|
assert "control_channel_list" not in ch
|
||||||
|
assert (a["frequency"], b["frequency"]) == (460_400_000, 460_425_000)
|
||||||
|
assert a["destination"].endswith(":23456") and b["destination"].endswith(":23457")
|
||||||
|
assert a["name"] == "PD1 Dispatch" and b["nac"] == 0x293
|
||||||
|
dev = config["devices"][0]
|
||||||
|
assert dev["tunable"] is False
|
||||||
|
# 25 kHz apart: centre is offset off both channels (DC spike), and both
|
||||||
|
# still sit inside the usable passband.
|
||||||
|
half = dev["rate"] * dev["usable_bw_pct"] / 2
|
||||||
|
for f in (460_400_000, 460_425_000):
|
||||||
|
assert 100_000 < abs(f - dev["frequency"]) < half
|
||||||
|
|
||||||
|
|
||||||
|
def test_single_channel_is_not_two_channel_audio():
|
||||||
|
config, two = cc.build_conventional_config("X", WP[:1], "lna:34", 0.0)
|
||||||
|
assert two is False
|
||||||
|
assert len(config["channels"]) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_wide_pair_centres_between_channels():
|
||||||
|
chans = [ConventionalChannel(frequency=453_000_000), ConventionalChannel(frequency=454_000_000)]
|
||||||
|
config, _ = cc.build_conventional_config("X", chans, "lna:34", 0.0)
|
||||||
|
assert config["devices"][0]["frequency"] == 453_500_000
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("chans, needle", [
|
||||||
|
([], "at least one"),
|
||||||
|
([ConventionalChannel(frequency=f) for f in (460_000_000, 460_025_000, 460_050_000)], "up to 2"),
|
||||||
|
([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=462_000_000)], "too far"),
|
||||||
|
([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=460_000_000)], "twice"),
|
||||||
|
([ConventionalChannel(frequency=460_000_000, nac=0x1000)], "NAC"),
|
||||||
|
])
|
||||||
|
def test_rejects_bad_channel_sets(chans, needle):
|
||||||
|
with pytest.raises(ValueError, match=needle):
|
||||||
|
cc.build_conventional_config("X", chans, "lna:34", 0.0)
|
||||||
|
|
||||||
|
|
||||||
|
ICE = IcecastConfig(icecast_host="localhost", icecast_port=8000, icecast_mountpoint="/n", icecast_password="p")
|
||||||
|
|
||||||
|
|
||||||
|
def test_liquidsoap_single_channel_unchanged_command():
|
||||||
|
liq = render_liquid_script(ICE, two_channel=False)
|
||||||
|
assert '"./audio.py -x 2.5 -s"' in liq
|
||||||
|
assert "amplify" not in liq and "${" not in liq
|
||||||
|
|
||||||
|
|
||||||
|
def test_liquidsoap_two_channel():
|
||||||
|
liq = render_liquid_script(ICE, two_channel=True)
|
||||||
|
assert '"./audio.py -x 2.5 -s -2"' in liq
|
||||||
|
assert "input = amplify(2.0, input)" in liq
|
||||||
|
assert "${" not in liq
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_config_route(tmp_path, monkeypatch):
|
||||||
|
from routers import op25_controller
|
||||||
|
monkeypatch.setattr(op25_controller, "_ACTIVE_CFG", str(tmp_path / "active.cfg.json"))
|
||||||
|
written = {}
|
||||||
|
monkeypatch.setattr(op25_controller, "generate_liquid_script",
|
||||||
|
lambda ice, two_channel=False: written.setdefault("two", two_channel))
|
||||||
|
monkeypatch.setattr(op25_controller, "save_talkgroup_tags", lambda tags: None)
|
||||||
|
monkeypatch.setattr(op25_controller, "save_whitelist", lambda wl: None)
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(op25_controller.create_op25_router(), prefix="/op25")
|
||||||
|
client = TestClient(app)
|
||||||
|
ice = ICE.model_dump()
|
||||||
|
r = client.post("/op25/generate-config", json={
|
||||||
|
"type": "P25", "systemName": "White Plains", "p25_mode": "conventional",
|
||||||
|
"conv_channels": [{"frequency": 460400000, "name": "PD1"}, {"frequency": 460425000, "name": "PD3"}],
|
||||||
|
"icecastConfig": ice,
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = json.loads((tmp_path / "active.cfg.json").read_text())
|
||||||
|
assert cfg["trunking"]["module"] == "tk_drb_conv.py"
|
||||||
|
assert [c["frequency"] for c in cfg["channels"]] == [460400000, 460425000]
|
||||||
|
assert written["two"] is True
|
||||||
|
|
||||||
|
# Trunked P25 without p25_mode is exactly what it was.
|
||||||
|
r = client.post("/op25/generate-config", json={
|
||||||
|
"type": "P25", "systemName": "County", "channels": [851012500], "icecastConfig": ice,
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = json.loads((tmp_path / "active.cfg.json").read_text())
|
||||||
|
assert cfg["trunking"]["module"] == "tk_p25.py"
|
||||||
|
assert cfg["trunking"]["chans"][0]["control_channel_list"] == "851012500"
|
||||||
|
assert cfg["channels"][0]["demod_type"] == "cqpsk"
|
||||||
|
|
||||||
|
r = client.post("/op25/generate-config", json={
|
||||||
|
"type": "P25", "systemName": "X", "p25_mode": "conventional", "conv_channels": [],
|
||||||
|
})
|
||||||
|
assert r.status_code == 400
|
||||||
|
|
||||||
|
# Analog uses the same module, so the edge node sees a `conventional` row
|
||||||
|
# whatever tk_p25 the image was built with (node-26#20).
|
||||||
|
r = client.post("/op25/generate-config", json={
|
||||||
|
"type": "NBFM", "systemName": "Fire", "channels": [154430000], "icecastConfig": ice,
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = json.loads((tmp_path / "active.cfg.json").read_text())
|
||||||
|
assert cfg["trunking"]["module"] == "tk_drb_conv.py"
|
||||||
|
assert cfg["channels"][0]["frequency"] == 154430000
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# tk_drb_conv state machine
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class _BaseRxCtl:
|
||||||
|
"""The slice of tk_p25.rx_ctl that tk_drb_conv relies on."""
|
||||||
|
|
||||||
|
def __init__(self, debug=0, frequency_set=None, nbfm_ctrl=None, fa_ctrl=None, chans={}):
|
||||||
|
self.fa_ctrl = fa_ctrl
|
||||||
|
self.receivers = {}
|
||||||
|
self.call_log = []
|
||||||
|
|
||||||
|
def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
|
||||||
|
self.receivers[msgq_id] = {'msgq_id': msgq_id, 'config': config, 'rx_rcvr': None,
|
||||||
|
'conv_state': {'tgid': None}}
|
||||||
|
|
||||||
|
def post_init(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def process_qmsg(self, msg):
|
||||||
|
self.base_saw = msg
|
||||||
|
|
||||||
|
def log_call(self, sysid, rcvr, freq, slot, prio, tgid, tgtag, rid, rtag):
|
||||||
|
self.call_log.append({"rcvr": rcvr, "freq": freq, "tgid": tgid, "tgtag": tgtag, "rid": rid})
|
||||||
|
|
||||||
|
def get_chan_status(self):
|
||||||
|
return json.dumps({'json_type': 'channel_update', 'channels': []})
|
||||||
|
|
||||||
|
|
||||||
|
def _stub_op25_modules():
|
||||||
|
tk = types.ModuleType("tk_p25")
|
||||||
|
tk.rx_ctl = _BaseRxCtl
|
||||||
|
hf = types.ModuleType("helper_funcs")
|
||||||
|
|
||||||
|
def from_dict(d, key, def_val):
|
||||||
|
return d[key] if key in d and d[key] != "" else def_val
|
||||||
|
|
||||||
|
def get_ordinals(s):
|
||||||
|
t = 0
|
||||||
|
for c in s:
|
||||||
|
t = (t << 8) + c
|
||||||
|
return t
|
||||||
|
hf.from_dict = from_dict
|
||||||
|
hf.get_ordinals = get_ordinals
|
||||||
|
lt = types.ModuleType("log_ts")
|
||||||
|
lt.log_ts = types.SimpleNamespace(get=lambda *a: "ts")
|
||||||
|
sys.modules.update({"tk_p25": tk, "helper_funcs": hf, "log_ts": lt})
|
||||||
|
|
||||||
|
|
||||||
|
class _Msg:
|
||||||
|
def __init__(self, m_type, rxid, payload=b""):
|
||||||
|
self._type = m_type & 0xffff # proto 0 in the upper 16 bits
|
||||||
|
self._arg1 = rxid << 1
|
||||||
|
self._payload = payload
|
||||||
|
|
||||||
|
def type(self):
|
||||||
|
return self._type
|
||||||
|
|
||||||
|
def arg1(self):
|
||||||
|
return self._arg1
|
||||||
|
|
||||||
|
def to_string(self):
|
||||||
|
return self._payload
|
||||||
|
|
||||||
|
|
||||||
|
def _lcw(nac, tgid, src, svcopts=0, lco=0x00):
|
||||||
|
lc = bytes([lco, 0x00, svcopts, 0x00]) + tgid.to_bytes(2, "big") + src.to_bytes(3, "big")
|
||||||
|
return nac.to_bytes(2, "big") + lc
|
||||||
|
|
||||||
|
|
||||||
|
def _ess(encrypted, algid, keyid):
|
||||||
|
return json.dumps({"encrypted": encrypted, "algid": algid, "keyid": keyid}).encode()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def tk():
|
||||||
|
_stub_op25_modules()
|
||||||
|
sys.modules.pop("tk_drb_conv", None)
|
||||||
|
import tk_drb_conv
|
||||||
|
sent = []
|
||||||
|
rx = tk_drb_conv.rx_ctl(fa_ctrl=sent.append, chans=[])
|
||||||
|
rx.add_receiver(0, {"name": "PD1 Dispatch", "nac": 0x293}, freq=460_400_000)
|
||||||
|
rx.add_receiver(1, {"name": "PD3 Tac"}, freq=460_425_000)
|
||||||
|
rx.post_init()
|
||||||
|
rx.sent = sent
|
||||||
|
return rx
|
||||||
|
|
||||||
|
|
||||||
|
def _rows(rx):
|
||||||
|
d = json.loads(rx.get_chan_status())
|
||||||
|
return {cid: d[cid] for cid in d["channels"]}
|
||||||
|
|
||||||
|
|
||||||
|
def test_nac_filter_applied_only_where_configured(tk):
|
||||||
|
assert tk.sent == [{'tuner': 0, 'cmd': 'set_nac', 'nac': 0x293}]
|
||||||
|
|
||||||
|
|
||||||
|
def test_clear_call_reports_tgid_unit_nac_and_logs_once(tk):
|
||||||
|
tk.process_qmsg(_Msg(0, 0, b"\x02\x93")) # HDU
|
||||||
|
tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242))) # LDU1
|
||||||
|
tk.process_qmsg(_Msg(-3, 0, _ess(0, 0x80, 0))) # LDU2 clear
|
||||||
|
tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242)))
|
||||||
|
row = _rows(tk)["0"]
|
||||||
|
assert row["conventional"] and row["voice"]
|
||||||
|
assert (row["tgid"], row["srcaddr"], row["nac"], row["encrypted"]) == (101, 4242, 0x293, 0)
|
||||||
|
assert row["algid"] is None
|
||||||
|
assert [e["tgid"] for e in tk.call_log] == [101]
|
||||||
|
assert tk.call_log[0]["tgtag"] == "PD1 Dispatch"
|
||||||
|
# The other channel is idle.
|
||||||
|
assert _rows(tk)["1"]["voice"] is False
|
||||||
|
|
||||||
|
tk.process_qmsg(_Msg(3, 0, b"\x02\x93")) # TDU
|
||||||
|
row = _rows(tk)["0"]
|
||||||
|
assert row["voice"] is False and row["tgid"] is None
|
||||||
|
assert (row["clear_calls"], row["enc_calls"]) == (1, 0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_tdulc_link_control_does_not_reopen_the_call(tk):
|
||||||
|
tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9)))
|
||||||
|
tk.process_qmsg(_Msg(15, 1, b"\x01\x11")) # TDULC
|
||||||
|
tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9))) # its LC
|
||||||
|
assert _rows(tk)["1"]["voice"] is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_encrypted_call_is_reported_and_never_logged(tk):
|
||||||
|
tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9, svcopts=0x40))) # E bit
|
||||||
|
tk.process_qmsg(_Msg(-3, 1, _ess(1, 0x84, 0x1234)))
|
||||||
|
row = _rows(tk)["1"]
|
||||||
|
assert row["encrypted"] == 1 and row["algid"] == 0x84 and row["keyid"] == 0x1234
|
||||||
|
assert tk.call_log == []
|
||||||
|
tk.process_qmsg(_Msg(-1, 1)) # sync timeout
|
||||||
|
row = _rows(tk)["1"]
|
||||||
|
assert row["encrypted"] == 0 and row["voice"] is False
|
||||||
|
assert (row["enc_calls"], row["last_enc_algid"], row["last_enc_keyid"]) == (1, 0x84, 0x1234)
|
||||||
|
|
||||||
|
|
||||||
|
def test_encryption_seen_only_in_ess_still_counts(tk):
|
||||||
|
tk.process_qmsg(_Msg(-3, 0, _ess(1, 0x81, 0x0001)))
|
||||||
|
assert _rows(tk)["0"]["encrypted"] == 1
|
||||||
|
tk.process_qmsg(_Msg(3, 0, b"\x02\x93"))
|
||||||
|
assert _rows(tk)["0"]["enc_calls"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_protected_link_control_marks_encrypted(tk):
|
||||||
|
tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 0, 0, lco=0x80)))
|
||||||
|
assert _rows(tk)["0"]["encrypted"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_trunked_receivers_and_other_protocols_go_to_base(tk):
|
||||||
|
msg = _Msg(7, 5)
|
||||||
|
tk.process_qmsg(msg)
|
||||||
|
assert tk.base_saw is msg
|
||||||
@@ -0,0 +1,172 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
#
|
||||||
|
# DRB node host updater (node-26#15). Runs as root from systemd:
|
||||||
|
# drb-update.timer -> every ~15 min: HTTPS check-in with C2; update if asked
|
||||||
|
# drb-update.path -> fires when the edge-node container drops
|
||||||
|
# <install>/configs/update-requested (MQTT node_update)
|
||||||
|
# Both start drb-update.service, which runs this script. install-updater.sh
|
||||||
|
# writes the units and /etc/drb/updater.conf; install.sh calls it.
|
||||||
|
#
|
||||||
|
# SECURITY MODEL — this is a remote-update path onto devices in the field:
|
||||||
|
# * What gets installed comes ONLY from /etc/drb/updater.conf (root-owned,
|
||||||
|
# written by install.sh): the repo URL, the ref, the image registry. C2's
|
||||||
|
# answer is a single timestamp meaning "update now"; it is never used as a
|
||||||
|
# ref, URL, path or command. The MQTT sentinel's content is never read.
|
||||||
|
# * git origin must equal the configured REPO_URL (https only) and the .env
|
||||||
|
# IMAGE_REGISTRY must equal the configured REGISTRY, or nothing runs.
|
||||||
|
# * Node credentials go to curl via a 0600 header file, never argv.
|
||||||
|
# * C2 must be https.
|
||||||
|
# What this does NOT defend against: a compromised git host or registry —
|
||||||
|
# refs and images are not signed (see the issue filed with node-26#15).
|
||||||
|
#
|
||||||
|
# The whole body is inside main(), called on the last line: bash parses the
|
||||||
|
# function completely before running it, so `git checkout` replacing this
|
||||||
|
# very file mid-run cannot change what is executing.
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
main() {
|
||||||
|
local CONF="${DRB_UPDATER_CONF:-/etc/drb/updater.conf}"
|
||||||
|
local STATE_DIR="${DRB_UPDATER_STATE:-/var/lib/drb-update}"
|
||||||
|
local LOCK="${DRB_UPDATER_LOCK:-/run/drb-update.lock}"
|
||||||
|
|
||||||
|
log() { printf 'drb-update: %s\n' "$*"; }
|
||||||
|
err() { printf 'drb-update: ERROR: %s\n' "$*" >&2; }
|
||||||
|
conf() { grep -E "^$1=" "$CONF" 2>/dev/null | head -1 | cut -d= -f2- || true; }
|
||||||
|
|
||||||
|
[ -r "$CONF" ] || { err "$CONF missing — re-run install.sh"; return 1; }
|
||||||
|
local INSTALL_DIR REPO_URL REF REGISTRY BUILD
|
||||||
|
INSTALL_DIR="$(conf INSTALL_DIR)"; REPO_URL="$(conf REPO_URL)"; REF="$(conf REF)"
|
||||||
|
REGISTRY="$(conf REGISTRY)"; BUILD="$(conf BUILD)"
|
||||||
|
|
||||||
|
[[ "$INSTALL_DIR" =~ ^/[A-Za-z0-9._/-]+$ ]] && [ -d "$INSTALL_DIR/.git" ] \
|
||||||
|
|| { err "bad INSTALL_DIR '$INSTALL_DIR' in $CONF"; return 1; }
|
||||||
|
[[ "$REPO_URL" =~ ^https://[A-Za-z0-9._/:-]+$ ]] || { err "REPO_URL must be https: '$REPO_URL'"; return 1; }
|
||||||
|
[[ "$REF" =~ ^[A-Za-z0-9._][A-Za-z0-9._/-]*$ ]] || { err "bad REF '$REF'"; return 1; }
|
||||||
|
[[ "$REGISTRY" =~ ^[A-Za-z0-9.-]+(:[0-9]+)?$ ]] || { err "bad REGISTRY '$REGISTRY'"; return 1; }
|
||||||
|
|
||||||
|
mkdir -p "$STATE_DIR"; chmod 700 "$STATE_DIR"
|
||||||
|
exec 9>"$LOCK"
|
||||||
|
flock -n 9 || { log "another update is already running"; return 0; }
|
||||||
|
|
||||||
|
cd "$INSTALL_DIR"
|
||||||
|
envget() { grep -E "^$1=" .env 2>/dev/null | head -1 | cut -d= -f2- | tr -d '"' || true; }
|
||||||
|
local NODE_ID C2_URL
|
||||||
|
NODE_ID="$(envget NODE_ID)"; C2_URL="$(envget C2_URL)"; C2_URL="${C2_URL%/}"
|
||||||
|
[[ "$NODE_ID" =~ ^[A-Za-z0-9_-]+$ ]] || { err "no valid NODE_ID in $INSTALL_DIR/.env"; return 1; }
|
||||||
|
|
||||||
|
local reason=""
|
||||||
|
local SENTINEL="$INSTALL_DIR/configs/update-requested"
|
||||||
|
if [ -e "$SENTINEL" ] || [ -L "$SENTINEL" ]; then
|
||||||
|
rm -f "$SENTINEL" # content deliberately ignored
|
||||||
|
reason="mqtt node_update"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ── HTTPS check-in ──────────────────────────────────────────────────────
|
||||||
|
local tmpd; tmpd="$(mktemp -d)"; chmod 700 "$tmpd"
|
||||||
|
# shellcheck disable=SC2064
|
||||||
|
trap "rm -rf '$tmpd'" RETURN
|
||||||
|
|
||||||
|
local sha; sha="$(git rev-parse HEAD 2>/dev/null || echo '')"
|
||||||
|
local requested=""
|
||||||
|
|
||||||
|
checkin() { # checkin APPLIED_AT RESULT -> stdout: response body
|
||||||
|
[[ "$C2_URL" == https://* ]] || return 1
|
||||||
|
local key="" pickup=""
|
||||||
|
key="$(jq -r '.api_key // empty' configs/credentials.json 2>/dev/null || true)"
|
||||||
|
pickup="$(cat configs/pickup_secret 2>/dev/null || true)"
|
||||||
|
: > "$tmpd/h"; chmod 600 "$tmpd/h"
|
||||||
|
if [[ "$key" =~ ^[A-Za-z0-9]+$ ]]; then
|
||||||
|
printf 'Authorization: Bearer %s\n' "$key" >> "$tmpd/h"
|
||||||
|
fi
|
||||||
|
if [[ "$pickup" =~ ^[A-Za-z0-9]+$ ]]; then
|
||||||
|
printf 'X-Pickup-Secret: %s\n' "$pickup" >> "$tmpd/h"
|
||||||
|
fi
|
||||||
|
[ -s "$tmpd/h" ] || return 1
|
||||||
|
local body
|
||||||
|
body="$(jq -nc --arg sha "$sha" --arg a "$1" --arg r "$2" \
|
||||||
|
'{sha:$sha} + (if $a != "" then {applied_request_at:$a} else {} end)
|
||||||
|
+ (if $r != "" then {result:$r} else {} end)')"
|
||||||
|
curl -fsS --proto '=https' --max-time 20 -X POST \
|
||||||
|
"$C2_URL/nodes/$NODE_ID/update-check" \
|
||||||
|
-H "Content-Type: application/json" -H "@$tmpd/h" --data "$body"
|
||||||
|
}
|
||||||
|
|
||||||
|
local resp
|
||||||
|
if [[ "$C2_URL" != https://* ]]; then
|
||||||
|
log "C2_URL is not https — skipping the HTTPS check-in"
|
||||||
|
elif resp="$(checkin "" "" 2>"$tmpd/err")"; then
|
||||||
|
requested="$(printf '%s' "$resp" | jq -r '.update_requested_at // empty' 2>/dev/null || true)"
|
||||||
|
if [ -n "$requested" ] && ! [[ "$requested" =~ ^[0-9T:.+Z-]{10,40}$ ]]; then
|
||||||
|
log "ignoring malformed update_requested_at from C2"
|
||||||
|
requested=""
|
||||||
|
fi
|
||||||
|
else
|
||||||
|
log "HTTPS check-in failed: $(tr -d '\n' < "$tmpd/err" | head -c 200)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
local last=""; last="$(cat "$STATE_DIR/last_request" 2>/dev/null || true)"
|
||||||
|
local new_request=""
|
||||||
|
if [ -n "$requested" ] && [ "$requested" != "$last" ]; then
|
||||||
|
new_request="$requested"
|
||||||
|
reason="${reason:-C2 request $requested}"
|
||||||
|
fi
|
||||||
|
[ -n "$reason" ] || { log "no update requested"; return 0; }
|
||||||
|
|
||||||
|
# ── Update ─────────────────────────────────────────────────────────────
|
||||||
|
log "updating ($reason) — repo $REPO_URL @ $REF"
|
||||||
|
local result=ok
|
||||||
|
if do_update; then
|
||||||
|
log "update complete: $(git rev-parse --short HEAD)"
|
||||||
|
else
|
||||||
|
result=failed
|
||||||
|
log "update FAILED — will retry on the next timer run"
|
||||||
|
fi
|
||||||
|
if [ -n "$new_request" ]; then
|
||||||
|
[ "$result" = ok ] && printf '%s' "$new_request" > "$STATE_DIR/last_request"
|
||||||
|
sha="$(git rev-parse HEAD 2>/dev/null || echo '')"
|
||||||
|
checkin "$new_request" "$result" >/dev/null 2>&1 || log "could not report the result to C2"
|
||||||
|
fi
|
||||||
|
[ "$result" = ok ]
|
||||||
|
}
|
||||||
|
|
||||||
|
do_update() {
|
||||||
|
local origin; origin="$(git remote get-url origin)"
|
||||||
|
[ "$origin" = "$REPO_URL" ] || { err "git origin '$origin' != configured '$REPO_URL' — refusing"; return 1; }
|
||||||
|
|
||||||
|
# --force: a moved release tag (v1) must replace the local copy.
|
||||||
|
git -c protocol.allow=never -c protocol.https.allow=always fetch --force --tags --prune origin || return 1
|
||||||
|
|
||||||
|
local target
|
||||||
|
if git show-ref --verify -q "refs/remotes/origin/$REF"; then
|
||||||
|
target="refs/remotes/origin/$REF"
|
||||||
|
elif git show-ref --verify -q "refs/tags/$REF"; then
|
||||||
|
target="refs/tags/$REF"
|
||||||
|
else
|
||||||
|
err "ref '$REF' not found on origin"; return 1
|
||||||
|
fi
|
||||||
|
local old new
|
||||||
|
old="$(git rev-parse HEAD)"; new="$(git rev-parse "$target^{commit}")"
|
||||||
|
if [ "$old" != "$new" ]; then
|
||||||
|
git -c advice.detachedHead=false checkout --force --detach "$new" || return 1
|
||||||
|
log "checked out $REF: ${old:0:12} -> ${new:0:12}"
|
||||||
|
# Units/conf may have changed with the new scripts.
|
||||||
|
bash scripts/install-updater.sh --refresh || log "unit refresh failed (continuing with the old units)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
local env_registry
|
||||||
|
env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
|
||||||
|
[ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
|
||||||
|
|| { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
|
||||||
|
|
||||||
|
if [ "$BUILD" = 1 ]; then
|
||||||
|
docker compose build || return 1
|
||||||
|
else
|
||||||
|
docker compose pull || return 1
|
||||||
|
fi
|
||||||
|
# --force-recreate: app code is bind-mounted from the checkout (install.sh
|
||||||
|
# D3), so a code-only change also needs every container restarted.
|
||||||
|
docker compose up -d --no-build --force-recreate --remove-orphans || return 1
|
||||||
|
docker image prune -f >/dev/null 2>&1 || true
|
||||||
|
}
|
||||||
|
|
||||||
|
main "$@"
|
||||||
@@ -0,0 +1,107 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
#
|
||||||
|
# Install (or refresh) the DRB host updater units (node-26#15). Run as root.
|
||||||
|
#
|
||||||
|
# install-updater.sh --dir /opt/drb/node-26 --repo-url URL --ref v1 \
|
||||||
|
# --registry git.vpn.cusano.net [--build]
|
||||||
|
# install-updater.sh --refresh # re-render units from /etc/drb/updater.conf
|
||||||
|
#
|
||||||
|
# Writes /etc/drb/updater.conf (root:root 0644 — the ONLY source of what the
|
||||||
|
# updater may install), drb-update.{service,timer,path}, and enables the timer
|
||||||
|
# and the path watcher. install.sh calls this; drb-update.sh calls --refresh
|
||||||
|
# after a checkout so unit changes ship with the repo.
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
CONF=/etc/drb/updater.conf
|
||||||
|
UNIT_DIR="${DRB_UNIT_DIR:-/etc/systemd/system}"
|
||||||
|
conf() { grep -E "^$1=" "$CONF" 2>/dev/null | head -1 | cut -d= -f2- || true; }
|
||||||
|
die() { printf 'install-updater: %s\n' "$*" >&2; exit 1; }
|
||||||
|
|
||||||
|
INSTALL_DIR="" REPO_URL="" REF="" REGISTRY="" BUILD=0 REFRESH=0
|
||||||
|
while [ $# -gt 0 ]; do
|
||||||
|
case "$1" in
|
||||||
|
--dir) INSTALL_DIR="$2"; shift 2 ;;
|
||||||
|
--repo-url) REPO_URL="$2"; shift 2 ;;
|
||||||
|
--ref) REF="$2"; shift 2 ;;
|
||||||
|
--registry) REGISTRY="$2"; shift 2 ;;
|
||||||
|
--build) BUILD=1; shift ;;
|
||||||
|
--refresh) REFRESH=1; shift ;;
|
||||||
|
*) die "unknown option: $1" ;;
|
||||||
|
esac
|
||||||
|
done
|
||||||
|
|
||||||
|
[ "$(id -u)" -eq 0 ] || die "must run as root"
|
||||||
|
if [ "$REFRESH" = 1 ]; then
|
||||||
|
[ -r "$CONF" ] || die "$CONF missing — run install.sh"
|
||||||
|
INSTALL_DIR="$(conf INSTALL_DIR)"; REPO_URL="$(conf REPO_URL)"; REF="$(conf REF)"
|
||||||
|
REGISTRY="$(conf REGISTRY)"; BUILD="$(conf BUILD)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
[[ "$INSTALL_DIR" =~ ^/[A-Za-z0-9._/-]+$ ]] || die "bad --dir '$INSTALL_DIR' (absolute, no spaces)"
|
||||||
|
[[ "$REPO_URL" =~ ^https://[A-Za-z0-9._/:-]+$ ]] || die "--repo-url must be https: '$REPO_URL'"
|
||||||
|
[[ "$REF" =~ ^[A-Za-z0-9._][A-Za-z0-9._/-]*$ ]] || die "bad --ref '$REF'"
|
||||||
|
[[ "$REGISTRY" =~ ^[A-Za-z0-9.-]+(:[0-9]+)?$ ]] || die "bad --registry '$REGISTRY'"
|
||||||
|
[ "$BUILD" = 1 ] || BUILD=0
|
||||||
|
|
||||||
|
if ! command -v systemctl >/dev/null 2>&1 || [ ! -d /run/systemd/system ]; then
|
||||||
|
die "systemd not running — the host updater needs systemd (remote updates disabled on this node)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
install -d -m 755 /etc/drb
|
||||||
|
umask 022
|
||||||
|
cat > "$CONF.tmp" <<EOF
|
||||||
|
# DRB host updater — written by install-updater.sh. The updater installs ONLY
|
||||||
|
# from these values; nothing C2 sends can change them. Edit = root only.
|
||||||
|
INSTALL_DIR=$INSTALL_DIR
|
||||||
|
REPO_URL=$REPO_URL
|
||||||
|
REF=$REF
|
||||||
|
REGISTRY=$REGISTRY
|
||||||
|
BUILD=$BUILD
|
||||||
|
EOF
|
||||||
|
chown root:root "$CONF.tmp"; chmod 644 "$CONF.tmp"; mv -f "$CONF.tmp" "$CONF"
|
||||||
|
|
||||||
|
cat > "$UNIT_DIR/drb-update.service" <<EOF
|
||||||
|
[Unit]
|
||||||
|
Description=DRB node updater (node-26#15)
|
||||||
|
After=network-online.target docker.service
|
||||||
|
Wants=network-online.target
|
||||||
|
Requires=docker.service
|
||||||
|
|
||||||
|
[Service]
|
||||||
|
Type=oneshot
|
||||||
|
ExecStart=/bin/bash $INSTALL_DIR/scripts/drb-update.sh
|
||||||
|
TimeoutStartSec=$([ "$BUILD" = 1 ] && echo 3h || echo 45min)
|
||||||
|
Nice=10
|
||||||
|
EOF
|
||||||
|
|
||||||
|
cat > "$UNIT_DIR/drb-update.timer" <<'EOF'
|
||||||
|
[Unit]
|
||||||
|
Description=DRB node updater — periodic HTTPS check-in with C2
|
||||||
|
|
||||||
|
[Timer]
|
||||||
|
OnBootSec=3min
|
||||||
|
OnUnitActiveSec=15min
|
||||||
|
RandomizedDelaySec=2min
|
||||||
|
Persistent=true
|
||||||
|
|
||||||
|
[Install]
|
||||||
|
WantedBy=timers.target
|
||||||
|
EOF
|
||||||
|
|
||||||
|
cat > "$UNIT_DIR/drb-update.path" <<EOF
|
||||||
|
[Unit]
|
||||||
|
Description=DRB node updater — MQTT node_update sentinel
|
||||||
|
|
||||||
|
[Path]
|
||||||
|
PathExists=$INSTALL_DIR/configs/update-requested
|
||||||
|
Unit=drb-update.service
|
||||||
|
|
||||||
|
[Install]
|
||||||
|
WantedBy=multi-user.target
|
||||||
|
EOF
|
||||||
|
|
||||||
|
systemctl daemon-reload
|
||||||
|
systemctl enable --now drb-update.timer drb-update.path >/dev/null 2>&1 \
|
||||||
|
|| die "could not enable drb-update.timer/path"
|
||||||
|
printf 'install-updater: host updater enabled (timer every ~15 min + MQTT sentinel), tracking %s @ %s\n' \
|
||||||
|
"$REPO_URL" "$REF"
|
||||||
Reference in New Issue
Block a user