Compare commits

..
Author SHA1 Message Date
Logan CusanoandClaude Opus 5.5 32692030fb install.sh: enrollment token via header file too — it was on curl argv despite the comment
CI / lint (push) Successful in 6s
CI / lint (pull_request) Successful in 4s
CI / test (push) Successful in 42s
CI / test (pull_request) Successful in 40s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 01:10:21 -04:00
Logan CusanoandClaude Opus 5.5 f22caf6259 install.sh: pickup secret via header file, not curl argv (security review)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 01:10:11 -04:00
Logan CusanoandClaude Opus 5.5 c2c349043b Node self-heals its API key; one-stop install.sh; real host-level OTA updater (#14, #15)
install.sh (#14): after enrolling it now BLOCKS until an admin approves,
printing "Please have an admin approve node '<id>' at <app>/settings/nodes
to continue", polling GET /nodes/{id}/credentials with the pickup secret
(no default timeout; Ctrl-C safe; re-run resumes from configs/pickup_secret).
A re-run with a key on disk re-checks it against C2 and replaces a stale
(reissued) key. credentials.json is merged atomically so the dashboard auth
salt/secret survive. edge-node is ALWAYS force-recreated at the end, so a key
written after the container started is loaded. --wait-approval N caps the
wait; 0 = don't wait. git fetch uses --force so a re-cut v1 tag is accepted.

edge-node (#14): with no key at startup, or on a broker auth refusal
(CONNACK 4/5/134/135), mqtt_manager re-reads credentials.json, else fetches
the key over HTTPS (key_pickup.py, pickup secret from /configs), persists it
and rebuilds the paho client in place. Single-flight, 30s..15min backoff.
A key pushed over MQTT while connected is applied for the next reconnect.

OTA (#15): node_update no longer os._exit()s. It drops
configs/update-requested; install.sh installs scripts/install-updater.sh's
drb-update.{service,timer,path}. The path unit fires on the sentinel, the
timer (~15 min) does an HTTPS check-in (POST /nodes/{id}/update-check,
server-26 PR) so offline/locked-out nodes still update. drb-update.sh only
installs the repo/ref/registry in root-owned /etc/drb/updater.conf; C2's
answer is just a timestamp; origin and registry mismatches abort; creds go
to curl via a 0600 header file.

Tests: test_credential_recovery.py, test_host_updater.py (runs the real
updater script with stub git/docker/curl).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 01:09:39 -04:00
20 changed files with 2077 additions and 1818 deletions
+46 -3
View File
@@ -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)
-300
View File
@@ -1,300 +0,0 @@
"""
Plain-language health checklist for the local dashboard (/api/health).
A first-time owner opens http://<node>/ and needs three answers: is this node
working, if not what is wrong, and what do I do about it. Each check below
answers one of those for one link in the chain
password -> C2 enrollment -> MQTT -> SDR -> what to listen to -> OP25
-> decoding -> audio (PulseAudio) -> local stream (Icecast)
`gather_facts()` does the I/O; `build_checks()` is a pure function of those
facts so every state (unapproved, no SDR, OP25 crashed, …) is unit-testable.
Read-only: nothing here changes node state, and it never polls the OP25
terminal itself (that drains OP25's call_log, which metadata_watcher owns) —
it reads metadata_watcher.console_state instead.
"""
import time
from typing import Any, Dict, List, Optional
import httpx
from app.config import settings
from app.internal.logger import logger
OK, WARN, ERROR, OFF = "ok", "warn", "error", "off"
# OP25's console answers within a second or two of a healthy multi_rx; past
# this the console is considered silent.
CONSOLE_STALE_SECONDS = 5.0
# P25 control channels send several messages a second; this long without one
# means OP25 is not decoding.
TSBK_STALE_SECONDS = 15.0
def _check(cid: str, label: str, state: str, summary: str, action: str = "") -> Dict[str, str]:
return {"id": cid, "label": label, "state": state, "summary": summary, "action": action}
def _mhz(freq: Any) -> Optional[str]:
try:
f = float(freq)
except (TypeError, ValueError):
return None
if f <= 0:
return None
return f"{f / 1e6:.4f} MHz" if f > 1e5 else f"{f:.4f} MHz"
def _ago(seconds: float) -> str:
seconds = max(0, int(seconds))
if seconds < 60:
return f"{seconds}s ago"
if seconds < 3600:
return f"{seconds // 60} min ago"
return f"{seconds // 3600} h ago"
def active_config(node_cfg) -> Any:
if node_cfg.override_system_id and node_cfg.override_config:
return node_cfg.override_config
return node_cfg.system_config
def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
"""Facts -> ordered checklist. See gather_facts() for the keys."""
now = f.get("now", time.time())
checks: List[Dict[str, str]] = []
# 1. Dashboard password
if f["default_password"]:
checks.append(_check(
"password", "Dashboard password", WARN,
"Still the factory default — anyone on this network can log in.",
"Set DASHBOARD_PASSWORD in the node's .env file, then restart the edge-node container.",
))
else:
checks.append(_check("password", "Dashboard password", OK, "Changed from the default."))
# 2. C2 enrollment / approval
c2_url, has_key = f["c2_url"], f["has_api_key"]
if not c2_url:
checks.append(_check(
"c2", "Server (C2) link", OFF,
"Standalone — no C2 server is set. Local listening below still works.",
"To join a DRB server, re-run install.sh with an enrollment token.",
))
elif not has_key:
checks.append(_check(
"c2", "Server (C2) link", WARN,
f"Waiting for approval — this node has no API key from {c2_url} yet, so nothing is uploaded.",
f"Ask a DRB admin to approve node \"{f['node_id']}\" on the Nodes page, then re-run "
"install.sh (or restart the edge-node container). Local listening below works meanwhile.",
))
else:
checks.append(_check("c2", "Server (C2) link", OK, f"Approved — API key on file for {c2_url}."))
# 3. MQTT
broker = f"{f['mqtt_broker']}:{f['mqtt_port']}"
if f["mqtt_connected"]:
checks.append(_check("mqtt", "Live link (MQTT)", OK, f"Connected to {broker}."))
elif not c2_url and not has_key:
checks.append(_check(
"mqtt", "Live link (MQTT)", OFF, f"Not connected to {broker} (expected for a standalone node).",
))
elif not has_key:
checks.append(_check(
"mqtt", "Live link (MQTT)", WARN,
f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).",
"Fixes itself once the node is approved and has its API key (see above).",
))
else:
checks.append(_check(
"mqtt", "Live link (MQTT)", ERROR,
f"Not connected to {broker} even though an API key is on file.",
"If the key was just installed or the node was re-approved, restart the edge-node container "
"so it loads the key. If it still fails, the key may be stale (every re-approval issues a new "
"one) — re-run install.sh, and check the node can reach the broker.",
))
# 4. SDR
sdr_count = f.get("sdr_count")
if sdr_count is None:
checks.append(_check(
"sdr", "SDR receiver", WARN, "Couldn't check — the SDR service isn't answering.",
"Check the secondary-sdr container is running.",
))
elif sdr_count == 0:
checks.append(_check(
"sdr", "SDR receiver", ERROR, "No RTL-SDR dongle detected.",
"Plug the SDR into a USB port (avoid unpowered hubs), then restart the stack.",
))
else:
checks.append(_check(
"sdr", "SDR receiver", OK, f"{sdr_count} SDR{'s' if sdr_count != 1 else ''} detected.",
))
# 5. What to listen to
src = f.get("source")
if not src:
checks.append(_check(
"source", "What to listen to", WARN, "Nothing chosen yet.",
"Pick a system or enter a frequency under \"Listen locally\" below.",
))
else:
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}."))
# 6. OP25
op25 = f.get("op25_status")
running = op25 == "running"
if op25 is None:
checks.append(_check(
"op25", "Radio decoder (OP25)", ERROR, "The op25 container isn't answering.",
"Check the op25 container is running; its log usually says why it stopped.",
))
elif running:
checks.append(_check("op25", "Radio decoder (OP25)", OK, "Running."))
elif op25 == "crashed":
checks.append(_check(
"op25", "Radio decoder (OP25)", ERROR, "Stopped unexpectedly.",
"Open \"OP25 log\" below for the reason — usually the SDR is unplugged or in use. Then press Start.",
))
elif not src:
checks.append(_check("op25", "Radio decoder (OP25)", OFF, "Stopped — choose what to listen to first."))
else:
checks.append(_check("op25", "Radio decoder (OP25)", WARN, "Stopped.", "Press Start."))
# 7. Decoding
console = f.get("console") or {}
last_poll = console.get("last_poll_epoch")
is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG")
if not running:
checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs."))
elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS:
checks.append(_check(
"decode", "Decoding", WARN, "OP25 is running but hasn't reported in yet.",
"Give it 20 seconds after a start. If this stays, open the OP25 log below.",
))
elif is_analog:
chans = console.get("channels") or []
freq = _mhz(chans[0].get("freq")) if chans else None
checks.append(_check(
"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).",
))
else:
trunk = console.get("trunk") or []
tsbk = [float(t.get("last_tsbk") or 0) for t in trunk]
if not trunk:
checks.append(_check("decode", "Decoding", WARN, "Waiting for OP25 to report control-channel status…"))
elif max(tsbk) > 0 and now - max(tsbk) < TSBK_STALE_SECONDS:
sys_ = trunk[tsbk.index(max(tsbk))]
cc = _mhz(sys_.get("rxchan"))
checks.append(_check(
"decode", "Decoding", OK, f"Locked on the control channel{' ' + cc if cc else ''}.",
))
else:
checks.append(_check(
"decode", "Decoding", ERROR, "Not decoding the control channel.",
"Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM). "
"Try another control channel for the same system.",
))
# 8. Audio (PulseAudio — recording + Discord path)
if not f["audio_capturing"]:
checks.append(_check(
"audio", "Audio", ERROR if running else OFF,
"No audio reaching the recorder." if running else "Nothing to hear until OP25 runs.",
"The op25 container owns the audio path (PulseAudio); restart it if this persists." if running else "",
))
else:
last_voice = f.get("last_voice_epoch")
if last_voice is None:
checks.append(_check("audio", "Audio", OK, "Audio path up — no voice heard yet."))
elif now - last_voice < 3:
checks.append(_check("audio", "Audio", OK, "Voice right now."))
else:
checks.append(_check("audio", "Audio", OK, f"Audio path up — last voice {_ago(now - last_voice)}."))
# 9. Local stream (Icecast)
ice = f.get("icecast_live")
if ice is None:
checks.append(_check(
"stream", "Local stream", ERROR, f"Icecast isn't answering on port {f['icecast_port']}.",
"Check the icecast container is running.",
))
elif ice:
checks.append(_check("stream", "Local stream", OK, "Live — press play in the player below."))
elif running:
checks.append(_check(
"stream", "Local stream", WARN, "OP25 is running but nothing is feeding the stream yet.",
"It normally appears within ~20 seconds of a start; if not, check the OP25 log below.",
))
else:
checks.append(_check("stream", "Local stream", OFF, "Goes live when OP25 runs."))
return checks
async def icecast_live() -> Optional[bool]:
"""True if the configured mount has a source, False if not, None if Icecast is down."""
url = f"http://{settings.icecast_host}:{settings.icecast_port}/status-json.xsl"
try:
async with httpx.AsyncClient(timeout=3) as client:
r = await client.get(url)
r.raise_for_status()
sources = (r.json().get("icestats") or {}).get("source") or []
except Exception as e:
logger.debug(f"Icecast status check failed: {e}")
return None
if isinstance(sources, dict):
sources = [sources]
return any(str(s.get("listenurl", "")).endswith(settings.icecast_mount) for s in sources)
async def gather_facts() -> Dict[str, Any]:
from app.internal import auth, credentials
from app.internal.call_recorder import call_recorder
from app.internal.config_manager import load_node_config
from app.internal.metadata_watcher import metadata_watcher
from app.internal.mqtt_manager import mqtt_manager
from app.internal.op25_client import op25_client
from app.internal.secondary_sdr_client import secondary_sdr_client
node_cfg = load_node_config()
cfg = active_config(node_cfg)
op25 = await op25_client.status()
devs = await secondary_sdr_client.devices()
sdr_count: Optional[int] = len(devs) if devs is not None else None
if not sdr_count:
# op25's own lsusb count — its :stable image may lack lsusb and say 0,
# so only trust a positive answer from it.
op25_devs = await op25_client.devices()
if op25_devs and op25_devs.get("count"):
sdr_count = op25_devs["count"]
activity = call_recorder.audio_activity()
return {
"now": time.time(),
"node_id": settings.node_id,
"default_password": auth.is_using_default_password(),
"c2_url": settings.c2_url,
"has_api_key": bool(credentials.get_api_key()),
"mqtt_connected": mqtt_manager.is_connected,
"mqtt_broker": settings.mqtt_broker,
"mqtt_port": settings.mqtt_port,
"sdr_count": sdr_count,
"source": None if cfg is None else {
"name": cfg.name, "type": cfg.type, "system_id": cfg.system_id,
"local": bool(node_cfg.override_system_id and node_cfg.override_config),
},
"op25_status": (op25 or {}).get("status") if op25 is not None else None,
"console": metadata_watcher.console_state,
"audio_capturing": call_recorder.is_capturing,
"last_voice_epoch": activity.last_voice_epoch,
"icecast_live": await icecast_live(),
"icecast_port": settings.icecast_port,
}
+73
View File
@@ -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
@@ -235,13 +235,6 @@ class MetadataWatcher:
self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge
self._last_ok_poll: float = 0.0 self._last_ok_poll: float = 0.0
# Dashboard-only view of what the OP25 console last said. Read by
# /api/health; never consulted by segmentation or attribution.
self._terminal_epoch: Optional[float] = None
self._terminal_channels: List[Dict[str, Any]] = []
self._terminal_trunk: List[Dict[str, Any]] = []
self._last_heard: Optional[ConsoleEvent] = None
# Rolling console history for close-time attribution. # Rolling console history for close-time attribution.
self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX) self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX)
@@ -301,7 +294,6 @@ class MetadataWatcher:
self._last_ok_poll = now self._last_ok_poll = now
self._record_console(update, now) self._record_console(update, now)
self._note_terminal(update, now)
activity = self._snapshot() activity = self._snapshot()
if activity is None or not activity.capturing: if activity is None or not activity.capturing:
@@ -358,18 +350,6 @@ class MetadataWatcher:
while self._console and self._console[0].epoch < cutoff: while self._console and self._console[0].epoch < cutoff:
self._console.popleft() self._console.popleft()
def _note_terminal(self, update: Any, now: float) -> None:
"""Keep the latest console snapshot for the dashboard. No side effects
on segment state — trunk_update/channel_update are not sent on every
poll, so the last non-empty one is kept."""
self._terminal_epoch = now
if update.channels:
self._terminal_channels = update.channels
if getattr(update, "trunk", None):
self._terminal_trunk = update.trunk
if self._console:
self._last_heard = self._console[-1]
def _is_duplicate_grant(self, event: ConsoleEvent) -> bool: def _is_duplicate_grant(self, event: ConsoleEvent) -> bool:
for index in range(len(self._console) - 1, -1, -1): for index in range(len(self._console) - 1, -1, -1):
if len(self._console) - index > _GRANT_DEDUPE_DEPTH: if len(self._console) - index > _GRANT_DEDUPE_DEPTH:
@@ -918,19 +898,6 @@ class MetadataWatcher:
def is_active(self) -> bool: def is_active(self) -> bool:
return self._active_call_id is not None return self._active_call_id is not None
@property
def console_state(self) -> Dict[str, Any]:
"""Last thing the OP25 console reported, for /api/health only."""
heard = self._last_heard
return {
"last_poll_epoch": self._terminal_epoch,
"channels": self._terminal_channels,
"trunk": self._terminal_trunk,
"last_heard": None if heard is None else {
"epoch": heard.epoch, "tgid": heard.tgid, "name": heard.name, "freq": heard.freq,
},
}
@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."""
+103 -11
View File
@@ -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:
+1 -40
View File
@@ -11,11 +11,6 @@ from app.internal.logger import logger
# each id is also a top-level key holding that channel's dict # each id is also a top-level key holding that channel's dict
# (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…). # (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…).
# #
# trunk_update — per-system trunking state keyed "0", "1", …; each carries
# `last_tsbk` (OP25's time.time() of the last control-channel
# message decoded) — the only direct proof OP25 is DECODING,
# not merely running. Surfaced on the dashboard only.
#
# call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended # call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended
# by tk_p25.log_call() at channel-grant time, each stamped with # by tk_p25.log_call() at channel-grant time, each stamped with
# OP25's own time.time(). get_call_log() DRAINS the deque, so # OP25's own time.time(). get_call_log() DRAINS the deque, so
@@ -23,7 +18,7 @@ from app.internal.logger import logger
# it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so # it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so
# the consumer must keep up. # the consumer must keep up.
# #
# Everything else (rx_update, terminal_config, …) is ignored. # Everything else (trunk_update, rx_update, terminal_config, …) is ignored.
TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}] TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}]
@@ -33,7 +28,6 @@ class TerminalUpdate:
channels: List[Dict[str, Any]] = field(default_factory=list) channels: List[Dict[str, Any]] = field(default_factory=list)
call_log: List[Dict[str, Any]] = field(default_factory=list) call_log: List[Dict[str, Any]] = field(default_factory=list)
trunk: List[Dict[str, Any]] = field(default_factory=list)
class OP25Client: class OP25Client:
@@ -51,23 +45,6 @@ class OP25Client:
logger.error(f"OP25 start failed: {e}") logger.error(f"OP25 start failed: {e}")
return False return False
async def start_with_reason(self) -> Optional[str]:
"""Start OP25; None on success, else a human-readable reason."""
try:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{self.api_url}/op25/start")
except Exception as e:
logger.error(f"OP25 start failed: {e}")
return "The op25 container isn't answering."
if r.is_success:
return None
try:
detail = r.json().get("detail")
except Exception:
detail = None
logger.error(f"OP25 start failed: HTTP {r.status_code} {detail or r.text[:200]}")
return detail or f"op25 container returned HTTP {r.status_code}."
async def stop(self) -> bool: async def stop(self) -> bool:
try: try:
async with httpx.AsyncClient(timeout=10) as client: async with httpx.AsyncClient(timeout=10) as client:
@@ -88,17 +65,6 @@ class OP25Client:
logger.error(f"OP25 status failed: {e}") logger.error(f"OP25 status failed: {e}")
return None return None
async def log(self, lines: int = 80) -> Optional[List[str]]:
"""Tail of multi_rx/liquidsoap output, kept by the op25 container."""
try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{self.api_url}/op25/log", params={"lines": lines})
r.raise_for_status()
return r.json().get("lines", [])
except Exception as e:
logger.error(f"OP25 log fetch failed: {e}")
return None
async def devices(self) -> Optional[Dict[str, Any]]: async def devices(self) -> Optional[Dict[str, Any]]:
try: try:
async with httpx.AsyncClient(timeout=5) as client: async with httpx.AsyncClient(timeout=5) as client:
@@ -178,11 +144,6 @@ def parse_terminal_messages(messages: Any) -> TerminalUpdate:
if isinstance(channel, dict): if isinstance(channel, dict):
update.channels.append(channel) update.channels.append(channel)
elif json_type == "trunk_update":
for key, system in msg.items():
if key != "json_type" and isinstance(system, dict):
update.trunk.append(system)
elif json_type == "call_log": elif json_type == "call_log":
for entry in msg.get("log") or []: for entry in msg.get("log") or []:
if isinstance(entry, dict): if isinstance(entry, dict):
@@ -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
+12 -16
View File
@@ -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,6 +187,7 @@ 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.")
@@ -327,12 +326,9 @@ async def lifespan(app: FastAPI):
if (node_cfg.override_system_id and node_cfg.override_config) if (node_cfg.override_system_id and node_cfg.override_config)
else node_cfg.system_config else node_cfg.system_config
) )
# Not gated on `configured` (set only by a C2 assignment): a standalone or if node_cfg.configured and active_config:
# not-yet-approved node listening to a system chosen on the local
# dashboard (an override) must come back on it after a reboot too.
if active_config:
from app.internal.op25_client import op25_client from app.internal.op25_client import op25_client
logger.info("Node has a system to listen to — waiting for OP25 API then generating config.") logger.info("Node is configured — waiting for OP25 API then generating config.")
for attempt in range(10): for attempt in range(10):
if await _generate_op25_config(active_config): if await _generate_op25_config(active_config):
await op25_client.start() await op25_client.start()
+33 -142
View File
@@ -1,12 +1,12 @@
from fastapi import APIRouter, Depends, HTTPException, Body from fastapi import APIRouter, Depends, HTTPException, Body
from pydantic import BaseModel from pydantic import BaseModel
from typing import Dict, List, Literal, Optional from typing import Dict, List, Optional
import asyncio import asyncio
import httpx import httpx
from app.config import settings from app.config import settings
from app.models import SystemConfig, SECONDARY_SDR_MODES from app.models import SystemConfig, SECONDARY_SDR_MODES
from app.internal.op25_client import op25_client from app.internal.op25_client import op25_client
from app.internal.config_manager import load_node_config, save_node_config from app.internal.config_manager import load_node_config, save_node_config, apply_system_config
from app.internal.call_recorder import call_recorder from app.internal.call_recorder import call_recorder
from app.internal.discord_radio import radio_bot from app.internal.discord_radio import radio_bot
from app.internal.metadata_watcher import metadata_watcher from app.internal.metadata_watcher import metadata_watcher
@@ -36,10 +36,8 @@ async def get_status():
if (node_cfg.override_system_id and node_cfg.override_config) if (node_cfg.override_system_id and node_cfg.override_config)
else node_cfg.system_config else node_cfg.system_config
) )
system_type = None
if active_config: if active_config:
system_name = active_config.name system_name = active_config.name
system_type = active_config.type
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", [])
@@ -60,11 +58,6 @@ async def get_status():
"configured": node_cfg.configured, "configured": node_cfg.configured,
"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,
# 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).
"local_choice": bool(node_cfg.override_system_id and node_cfg.override_config),
"has_c2_system": node_cfg.system_config is not None,
"is_recording": call_recorder.is_recording, "is_recording": call_recorder.is_recording,
# Health of the PulseAudio capture that feeds every recording — the single # Health of the PulseAudio capture that feeds every recording — the single
# most useful signal when recordings come back empty. Segment boundaries # most useful signal when recordings come back empty. Segment boundaries
@@ -84,27 +77,15 @@ async def get_status():
"icecast_url": ( "icecast_url": (
f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}" f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}"
), ),
# The browser builds its own URL from these + window.location.hostname;
# icecast_host is "localhost" from the node's point of view.
"icecast_port": settings.icecast_port,
"icecast_mount": settings.icecast_mount,
"op25": op25_status, "op25": op25_status,
} }
@router.get("/health")
async def get_health():
"""Plain-language setup/status checklist for the dashboard (see internal/health.py)."""
from app.internal import health
facts = await health.gather_facts()
return {"checks": health.build_checks(facts), "last_heard": facts["console"].get("last_heard")}
@router.post("/op25/start") @router.post("/op25/start")
async def start_op25(): async def start_op25():
err = await op25_client.start_with_reason() ok = await op25_client.start()
if err: if not ok:
raise HTTPException(502, err) raise HTTPException(500, "Failed to start OP25")
return {"ok": True} return {"ok": True}
@@ -112,135 +93,30 @@ async def start_op25():
async def stop_op25(): async def stop_op25():
ok = await op25_client.stop() ok = await op25_client.stop()
if not ok: if not ok:
raise HTTPException(502, "The op25 container did not answer the stop request.") raise HTTPException(500, "Failed to stop OP25")
return {"ok": True} return {"ok": True}
@router.get("/op25/log")
async def op25_log(lines: int = 80):
out = await op25_client.log(lines)
if out is None:
raise HTTPException(502, "The op25 container isn't answering.")
return {"lines": out}
@router.get("/config") @router.get("/config")
async def get_config(): async def get_config():
return load_node_config() return load_node_config()
async def _restart_op25_with(config: SystemConfig) -> None:
"""Generate OP25's config for `config` and restart OP25 on it."""
from app.main import _generate_op25_config
if not await _generate_op25_config(config):
raise HTTPException(
502, f"The op25 container did not accept the config for {config.name} — is it running?",
)
await op25_client.stop()
await asyncio.sleep(2)
err = await op25_client.start_with_reason()
if err:
raise HTTPException(502, f"Config saved, but OP25 did not start: {err}")
async def _apply_override(config: SystemConfig) -> None:
"""Make `config` what this node listens to, chosen locally.
Stored as an override so a C2-assigned system (if any) survives and can
be reverted to, and so C2 sees the node is overridden. A later C2 config
push replaces it.
"""
node_cfg = load_node_config()
node_cfg.override_system_id = config.system_id
node_cfg.override_config = config
save_node_config(node_cfg)
await _restart_op25_with(config)
if mqtt_manager.is_connected:
await mqtt_manager.publish_checkin()
@router.post("/config/system") @router.post("/config/system")
async def set_system_config(config: SystemConfig): async def set_system_config(config: SystemConfig):
""" """
Assign a system as if C2 had (persists as the node's own system, not an Apply a system config locally — called by the web UI or pushed by C2.
override) and restart OP25 on it. Writes the OP25 config and persists the node config.
Used to write the raw `config` blob straight to active.cfg.json, which is
not an OP25 config at all — OP25 could not start from it.
""" """
_validate_type(config.type)
node_cfg = load_node_config() node_cfg = load_node_config()
node_cfg.assigned_system_id = config.system_id node_cfg.assigned_system_id = config.system_id
node_cfg.system_config = config node_cfg.system_config = config
node_cfg.configured = True node_cfg.configured = True
save_node_config(node_cfg) save_node_config(node_cfg)
await _restart_op25_with(config) apply_system_config(config)
return {"ok": True} return {"ok": True}
def _validate_type(system_type: str) -> None:
if str(system_type).upper() == "DMR":
raise HTTPException(400, "DMR isn't supported by this node's decoder yet — use P25 or analog (NBFM).")
if str(system_type).upper() not in ("P25", "NBFM"):
raise HTTPException(400, f"Unknown system type {system_type!r} — use P25 or NBFM.")
class Talkgroup(BaseModel):
id: int
name: str = ""
class ListenBody(BaseModel):
"""What to listen to, chosen on the local dashboard."""
mode: Literal["system", "p25", "analog"]
system_id: Optional[str] = None # mode=system: a cached C2 system
name: Optional[str] = None
frequencies: List[float] = [] # MHz; p25 = control channels, analog = exactly one
talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup
# RTL-SDR (R820T/R828D) tuning range.
_MIN_MHZ, _MAX_MHZ = 24.0, 1766.0
@router.post("/listen")
async def listen(body: ListenBody):
"""Point OP25 at a system or frequency and start it — no C2 needed."""
if body.mode == "system":
from app.internal.system_cacher import get_cached_system
cached = get_cached_system(body.system_id or "")
if not cached:
raise HTTPException(404, "That system isn't in this node's saved list.")
config = SystemConfig(**cached)
_validate_type(config.type)
else:
freqs = body.frequencies
if not freqs:
raise HTTPException(400, "Enter at least one frequency in MHz.")
bad = [f for f in freqs if not (_MIN_MHZ <= f <= _MAX_MHZ)]
if bad:
raise HTTPException(
400, f"{bad[0]:g} isn't a frequency an RTL-SDR can tune — enter MHz between "
f"{_MIN_MHZ:g} and {_MAX_MHZ:g} (e.g. 851.0125).",
)
if body.mode == "analog" and len(freqs) != 1:
raise HTTPException(400, "Analog listens to exactly one frequency.")
is_p25 = body.mode == "p25"
first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".")
label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
config = SystemConfig(
system_id=f"local-{body.mode}",
name=label,
type="P25" if is_p25 else "NBFM",
config={
"control_channels": freqs,
"talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
},
)
await _apply_override(config)
return {"ok": True, "name": config.name}
@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
@@ -253,6 +129,9 @@ async def set_override(
system_id: Optional[str] = Body(None), system_id: Optional[str] = Body(None),
system_config: Optional[dict] = Body(None) system_config: Optional[dict] = Body(None)
): ):
node_cfg = load_node_config()
config = None
if system_id: if system_id:
from app.internal.system_cacher import get_cached_system from app.internal.system_cacher import get_cached_system
cached = get_cached_system(system_id) cached = get_cached_system(system_id)
@@ -266,8 +145,19 @@ async def set_override(
else: else:
raise HTTPException(400, "Must specify system_id or system_config.") raise HTTPException(400, "Must specify system_id or system_config.")
_validate_type(config.type) node_cfg.override_system_id = config.system_id
await _apply_override(config) node_cfg.override_config = config
save_node_config(node_cfg)
from app.main import _generate_op25_config
if not await _generate_op25_config(config):
raise HTTPException(500, f"Failed to generate OP25 config for override: {config.name}")
await op25_client.stop()
await asyncio.sleep(2)
await op25_client.start()
await mqtt_manager._publish_checkin()
return {"ok": True} return {"ok": True}
@@ -282,14 +172,15 @@ async def revert_config():
save_node_config(node_cfg) save_node_config(node_cfg)
if node_cfg.system_config: if node_cfg.system_config:
await _restart_op25_with(node_cfg.system_config) from app.main import _generate_op25_config
else: if not await _generate_op25_config(node_cfg.system_config):
# Nothing from C2 to go back to: stop rather than keep decoding a raise HTTPException(500, "Failed to regenerate original OP25 config.")
# system the dashboard now says is cleared.
await op25_client.stop()
if mqtt_manager.is_connected: await op25_client.stop()
await mqtt_manager.publish_checkin() await asyncio.sleep(2)
await op25_client.start()
await mqtt_manager._publish_checkin()
return {"ok": True} return {"ok": True}
File diff suppressed because it is too large Load Diff
+25 -14
View File
@@ -124,7 +124,7 @@
} }
.softkey:last-child { border-right: none; } .softkey:last-child { border-right: none; }
/* Overlay UI (Volume, status) */ /* Overlay UI (Volume, Hold) */
.overlay { .overlay {
position: absolute; position: absolute;
top: 50%; left: 50%; top: 50%; left: 50%;
@@ -153,6 +153,7 @@
<ul> <ul>
<li><span>Up Arrow</span> Volume Up</li> <li><span>Up Arrow</span> Volume Up</li>
<li><span>Down Arrow</span> Volume Down</li> <li><span>Down Arrow</span> Volume Down</li>
<li><span>Enter / H</span> Toggle Hold (UI mock)</li>
<li><span>P</span> Play/Pause Stream</li> <li><span>P</span> Play/Pause Stream</li>
<li><span>S</span> Select System</li> <li><span>S</span> Select System</li>
<li><span>R</span> Revert Override</li> <li><span>R</span> Revert Override</li>
@@ -170,14 +171,15 @@
<h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3> <h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3>
<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;"> <div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;">
<div> <div>
<label>Freq (MHz) — P25 control channel or analog:</label> <label>Control Freq (MHz/Hz):</label>
<input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;"> <input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
</div> </div>
<div> <div>
<label>System Type:</label> <label>System Type:</label>
<select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;"> <select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
<option value="P25">P25</option> <option value="P25">P25</option>
<option value="NBFM">Analog FM</option> <option value="DMR">DMR</option>
<option value="NBFM">NBFM</option>
</select> </select>
</div> </div>
</div> </div>
@@ -206,11 +208,11 @@
<div class="softkeys"> <div class="softkeys">
<div class="softkey">VOL (+/-)</div> <div class="softkey">VOL (+/-)</div>
<div class="softkey">SYSTEM (S)</div> <div class="softkey">HOLD (H)</div>
<div class="softkey">PLAY (P)</div> <div class="softkey">PLAY (P)</div>
</div> </div>
<div class="overlay" id="overlay"></div> <div class="overlay" id="overlay">HOLD</div>
<!-- Audio element for local playback --> <!-- Audio element for local playback -->
<audio id="audio-player" autoplay></audio> <audio id="audio-player" autoplay></audio>
@@ -229,6 +231,7 @@
const elOverlay = document.getElementById('overlay'); const elOverlay = document.getElementById('overlay');
let currentIcecastUrl = null; let currentIcecastUrl = null;
let isHolding = false;
let volume = 1.0; let volume = 1.0;
// Clock updater // Clock updater
@@ -306,7 +309,7 @@
const container = document.getElementById('system-list'); const container = document.getElementById('system-list');
container.innerHTML = ''; container.innerHTML = '';
if (systems.length === 0) { if (systems.length === 0) {
container.innerHTML = '<div style="padding: 10px; color: #888;">No server systems saved yet (they arrive once the node is approved). Enter a frequency below.</div>'; container.innerHTML = '<div style="padding: 10px; color: #888;">No cached systems found.</div>';
return; return;
} }
systems.forEach((sys, index) => { systems.forEach((sys, index) => {
@@ -370,20 +373,25 @@
} }
closeSystemModal(); closeSystemModal();
showOverlay("APPLYING MANUAL", 3000); showOverlay("APPLYING MANUAL", 3000);
// Same validated path as the dashboard's "Listen locally" (MHz range check). const manualConfig = {
const mode = typeVal === 'NBFM' ? 'analog' : 'p25'; system_id: "manual-override",
const freqs = freqVal.split(/[\s,;]+/).filter(Boolean).map(Number); name: `Freq ${freqVal} (${typeVal})`,
type: typeVal,
config: {
control_channels: [freqVal],
talkgroups: []
}
};
try { try {
const res = await fetch('/api/listen', { const res = await fetch('/api/config/override', {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ mode, frequencies: freqs }) body: JSON.stringify({ system_config: manualConfig })
}); });
if (res.ok) { if (res.ok) {
showOverlay('MANUAL OK', 1500); showOverlay('MANUAL OVERRIDE OK', 1500);
} else { } else {
const d = await res.json().catch(() => ({})); showOverlay('MANUAL OVERRIDE FAIL', 2000);
showOverlay((d.detail || 'MANUAL FAILED').slice(0, 60), 4000);
} }
} catch (err) { } catch (err) {
showOverlay('ERROR APPLYING', 2000); showOverlay('ERROR APPLYING', 2000);
@@ -436,6 +444,9 @@
audioPlayer.volume = volume; audioPlayer.volume = volume;
elVol.textContent = `VOL: ${Math.round(volume * 100)}%`; elVol.textContent = `VOL: ${Math.round(volume * 100)}%`;
showOverlay(`VOL: ${Math.round(volume * 100)}%`); showOverlay(`VOL: ${Math.round(volume * 100)}%`);
} else if (e.key === 'Enter' || e.key === 'h' || e.key === 'H') {
isHolding = !isHolding;
showOverlay(isHolding ? 'HOLD ON' : 'HOLD OFF');
} else if (e.key === 'p' || e.key === 'P') { } else if (e.key === 'p' || e.key === 'P') {
if (audioPlayer.paused) { if (audioPlayer.paused) {
audioPlayer.play(); audioPlayer.play();
@@ -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
+179
View File
@@ -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)
-248
View File
@@ -1,248 +0,0 @@
"""
Local dashboard: the health checklist (/api/health) and choosing what to
listen to without C2 (/api/listen, /api/config/system, /api/config/revert).
Same standalone-app pattern as test_auth_endpoints.py — the real routers,
none of app.main's lifespan. OP25/MQTT are mocked at the client boundary.
"""
import base64
import time
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.config import settings
from app.internal import config_manager, credentials, health
from app.internal.op25_client import parse_terminal_messages
from app.routers import api, ui
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
monkeypatch.setattr(credentials, "_api_key", None)
monkeypatch.setattr(credentials, "_auth_salt", None)
monkeypatch.setattr(credentials, "_session_secret", None)
monkeypatch.setattr(config_manager, "_CONFIG_FILE", tmp_path / "node_config.json")
monkeypatch.setattr(settings, "dashboard_username", "tester")
monkeypatch.setattr(settings, "dashboard_password", "s3cret-pass")
yield
@pytest.fixture
def client():
app = FastAPI()
app.include_router(api.router)
app.include_router(ui.router)
auth = base64.b64encode(b"tester:s3cret-pass").decode()
with TestClient(app, headers={"Authorization": f"Basic {auth}"}) as c:
yield c
@pytest.fixture
def op25():
"""Mock the OP25 side of an apply: generate, stop, start, checkin."""
with patch("app.main._generate_op25_config", AsyncMock(return_value=True)) as gen, \
patch("app.routers.api.op25_client.stop", AsyncMock(return_value=True)) as stop, \
patch("app.routers.api.op25_client.start_with_reason", AsyncMock(return_value=None)) as start, \
patch("app.routers.api.asyncio.sleep", AsyncMock()), \
patch("app.routers.api.mqtt_manager.publish_checkin", AsyncMock()) as checkin:
yield {"gen": gen, "stop": stop, "start": start, "checkin": checkin}
# ---------------------------------------------------------------------------
# Health checklist — pure function of facts
# ---------------------------------------------------------------------------
def _facts(**over):
now = 1_000_000.0
f = {
"now": now, "node_id": "toughny", "default_password": False,
"c2_url": "https://api.example", "has_api_key": True, "mqtt_connected": True,
"mqtt_broker": "mqtt.example", "mqtt_port": 8883, "sdr_count": 1,
"source": {"name": "County", "type": "P25", "system_id": "s1", "local": False},
"op25_status": "running",
"console": {"last_poll_epoch": now - 1, "channels": [], "trunk": [{"last_tsbk": now - 1, "rxchan": 851012500}],
"last_heard": None},
"audio_capturing": True, "last_voice_epoch": now - 30, "icecast_live": True, "icecast_port": 8000,
}
f.update(over)
return f
def _by_id(checks):
return {c["id"]: c for c in checks}
def test_healthy_node_is_all_ok():
checks = health.build_checks(_facts())
assert all(c["state"] == "ok" for c in checks), [c for c in checks if c["state"] != "ok"]
assert "851.0125 MHz" in _by_id(checks)["decode"]["summary"]
def test_unapproved_node_explains_wait_and_mqtt_refusal():
"""Toughny's state: C2 set, no API key, broker refuses, nothing chosen."""
c = _by_id(health.build_checks(_facts(has_api_key=False, mqtt_connected=False, source=None,
op25_status="stopped")))
assert c["c2"]["state"] == "warn" and "approve" in c["c2"]["action"]
assert "toughny" in c["c2"]["action"]
assert c["mqtt"]["state"] == "warn"
assert c["source"]["state"] == "warn" and "Listen locally" in c["source"]["action"]
assert c["op25"]["state"] == "off"
assert c["decode"]["state"] == "off"
def test_key_on_disk_but_mqtt_down_says_restart():
c = _by_id(health.build_checks(_facts(mqtt_connected=False)))
assert c["mqtt"]["state"] == "error"
assert "restart the edge-node container" in c["mqtt"]["action"]
def test_standalone_node_is_not_an_error():
c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
assert c["c2"]["state"] == "off"
assert c["mqtt"]["state"] == "off"
def test_no_sdr_and_crashed_op25():
c = _by_id(health.build_checks(_facts(sdr_count=0, op25_status="crashed")))
assert c["sdr"]["state"] == "error"
assert c["op25"]["state"] == "error" and "log" in c["op25"]["action"]
def test_running_but_no_control_channel():
now = 1_000_000.0
c = _by_id(health.build_checks(_facts(console={"last_poll_epoch": now, "channels": [],
"trunk": [{"last_tsbk": 0}], "last_heard": None})))
assert c["decode"]["state"] == "error"
def test_analog_needs_no_control_channel():
now = 1_000_000.0
c = _by_id(health.build_checks(_facts(
source={"name": "Fire", "type": "NBFM", "system_id": "local-analog", "local": True},
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 "not recorded" in c["decode"]["action"]
assert "this dashboard" in c["source"]["summary"]
def test_op25_unreachable_and_icecast_down():
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
assert c["op25"]["state"] == "error"
assert c["stream"]["state"] == "error"
def test_default_password_warns():
assert _by_id(health.build_checks(_facts(default_password=True)))["password"]["state"] == "warn"
def test_trunk_update_is_parsed():
update = parse_terminal_messages([
{"json_type": "trunk_update", "0": {"last_tsbk": 5.0, "rxchan": 851012500}, "nac": 0},
{"json_type": "channel_update", "channels": ["0"], "0": {"freq": 1, "tgid": None}},
])
assert update.trunk == [{"last_tsbk": 5.0, "rxchan": 851012500}]
assert len(update.channels) == 1
def test_health_route_returns_checks(client):
facts = _facts(now=time.time())
with patch("app.internal.health.gather_facts", AsyncMock(return_value=facts)):
r = client.get("/api/health")
assert r.status_code == 200
assert [c["id"] for c in r.json()["checks"]][:3] == ["password", "c2", "mqtt"]
# ---------------------------------------------------------------------------
# Choosing what to listen to
# ---------------------------------------------------------------------------
def test_listen_p25_saves_override_and_restarts_op25(client, op25):
r = client.post("/api/listen", json={
"mode": "p25", "frequencies": [851.0125, 851.5125], "talkgroups": [{"id": 1001, "name": "Fire"}],
})
assert r.status_code == 200, r.text
cfg = config_manager.load_node_config()
assert cfg.override_config.type == "P25"
assert cfg.override_config.config["control_channels"] == [851.0125, 851.5125]
assert cfg.override_config.config["talkgroups"] == [{"id": 1001, "name": "Fire"}]
op25["gen"].assert_awaited_once()
op25["start"].assert_awaited_once()
op25["checkin"].assert_not_awaited() # MQTT not connected in tests
def test_listen_analog_single_frequency(client, op25):
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43], "name": "Fire"})
assert r.status_code == 200, r.text
cfg = config_manager.load_node_config().override_config
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
@pytest.mark.parametrize("body,needle", [
({"mode": "p25", "frequencies": []}, "at least one"),
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
({"mode": "system", "system_id": "nope"}, "saved list"),
])
def test_listen_rejects_bad_input_with_a_readable_reason(client, op25, body, needle):
r = client.post("/api/listen", json=body)
assert r.status_code in (400, 404)
assert needle in r.json()["detail"]
op25["gen"].assert_not_awaited()
def test_listen_reports_op25_start_failure(client, op25):
op25["start"].return_value = "No OP25 config yet — choose a system or frequency first."
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
assert r.status_code == 502
assert "did not start" in r.json()["detail"]
def test_dmr_is_refused_up_front(client, op25):
r = client.post("/api/config/override", json={"system_config": {
"system_id": "x", "name": "x", "type": "DMR", "config": {"control_channels": [450.1]},
}})
assert r.status_code == 400 and "DMR" in r.json()["detail"]
op25["gen"].assert_not_awaited()
def test_config_system_goes_through_op25_generation(client, op25, tmp_path, monkeypatch):
"""It used to dump the raw blob into active.cfg.json — not an OP25 config."""
monkeypatch.setattr(settings, "config_path", str(tmp_path))
r = client.post("/api/config/system", json={
"system_id": "s1", "name": "County", "type": "P25", "config": {"control_channels": [851.0125]},
})
assert r.status_code == 200, r.text
op25["gen"].assert_awaited_once()
assert not (tmp_path / "active.cfg.json").exists()
cfg = config_manager.load_node_config()
assert cfg.configured and cfg.system_config.name == "County"
def test_revert_without_c2_system_stops_op25(client, op25):
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
op25["stop"].reset_mock()
r = client.post("/api/config/revert")
assert r.status_code == 200
assert config_manager.load_node_config().override_config is None
op25["stop"].assert_awaited_once()
def test_status_exposes_stream_parts_and_source(client, op25):
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
with patch("app.routers.api.op25_client.status", AsyncMock(return_value={"status": "running"})):
d = client.get("/api/status").json()
assert d["icecast_port"] == settings.icecast_port and d["icecast_mount"] == settings.icecast_mount
assert d["local_choice"] is True and d["has_c2_system"] is False
assert d["system_type"] == "NBFM"
def test_dashboard_page_has_no_dead_api_instruction(client):
client.post("/login", data={"username": "tester", "password": "s3cret-pass"}, follow_redirects=False)
html = client.get("/").text
assert "POST /api/config/system" not in html
assert "/api/listen" in html and "/api/health" in html
+133 -55
View File
@@ -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
+6 -14
View File
@@ -52,16 +52,12 @@ class IcecastConfig(BaseModel):
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
# frequency to listen to (Hz).
channels: List[Union[int, str]] channels: List[Union[int, str]]
tags: Optional[List[TalkgroupTag]] = None tags: Optional[List[TalkgroupTag]]
whitelist: Optional[List[int]] = None whitelist: Optional[List[int]]
icecastConfig: Optional[IcecastConfig] = None icecastConfig: Optional[IcecastConfig]
hardware_preset: str = "rtl-sdr-v3" hardware_preset: str = "rtl-sdr-v3"
ppm_override: Optional[float] = None # when set, overrides the preset's default PPM ppm_override: Optional[float] = None # when set, overrides the preset's default PPM
# NBFM only: power squelch threshold in dB (op25_nbfm default is -60).
nbfm_squelch: Optional[float] = None
class DemodType(str, Enum): class DemodType(str, Enum):
CQPSK = "cqpsk" CQPSK = "cqpsk"
@@ -73,16 +69,15 @@ class FilterType(str, Enum):
class ChannelConfig(BaseModel): class ChannelConfig(BaseModel):
name: str name: str
trunking_sysname: Optional[str] = None trunking_sysname: Optional[str]
enable_analog: str enable_analog: str
meta_stream_name: str meta_stream_name: str
demod_type: DemodType demod_type: DemodType
filter_type: FilterType filter_type: FilterType
device: Optional[str] = "sdr" device: Optional[str] = "sdr"
cqpsk_tracking: Optional[bool] = None cqpsk_tracking: Optional[bool] = None
frequency: Optional[int] = None frequency: Optional[float] = None
# Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)). nbfmSquelch: 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"
tracking_threshold: Optional[int] = 120 tracking_threshold: Optional[int] = 120
tracking_feedback: Optional[float] = 0.75 tracking_feedback: Optional[float] = 0.75
@@ -98,9 +93,6 @@ class DeviceConfig(BaseModel):
gains: Optional[str] = "lna:39" gains: Optional[str] = "lna:39"
gain_mode: Optional[bool] = False gain_mode: Optional[bool] = False
name: Optional[str] = "sdr" name: Optional[str] = "sdr"
# Centre frequency (Hz). Unset for trunked P25, where tk_p25 tunes the
# device; required for a fixed analog channel (tunable=False).
frequency: Optional[int] = None
offset: Optional[int] = 0 offset: Optional[int] = 0
ppm: Optional[float] = 0.0 ppm: Optional[float] = 0.0
rate: Optional[int] = 1920000 rate: Optional[int] = 1920000
+20 -125
View File
@@ -3,11 +3,7 @@ import subprocess
import os import os
import re import re
import signal import signal
import sys
import json import json
import threading
from collections import deque
from typing import Optional
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
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
@@ -18,16 +14,6 @@ LOGGER = create_logger(__name__)
OP25_PATH = "/op25/op25/gr-op25_repeater/apps/" OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
OP25_SCRIPT = "run_multi-rx_service.sh" OP25_SCRIPT = "run_multi-rx_service.sh"
_PGID_FILE = "/tmp/op25.pgid" _PGID_FILE = "/tmp/op25.pgid"
_ACTIVE_CFG = "/configs/active.cfg.json"
# Analog: park the dongle's centre this far from the channel so the channel
# does not sit on the RTL-SDR's DC spike. Well inside the 1.92 MS/s window.
ANALOG_CENTER_OFFSET_HZ = 250_000
# Last lines multi_rx/liquidsoap printed — the only place the reason OP25
# died (SDR busy, bad config, no lock) is visible without SSH + docker logs.
_log: deque = deque(maxlen=300)
_proc: Optional[subprocess.Popen] = None
def _save_pgid(pgid: int) -> None: def _save_pgid(pgid: int) -> None:
@@ -42,60 +28,15 @@ def _read_pgid():
return None return None
def _multi_rx_alive(pgid: int) -> bool:
"""True only if a live multi_rx.py process is in OP25's process group.
node-26#12: killpg(pgid, 0) succeeds while the wrapper shell or
liquidsoap outlives a dead multi_rx (and even for a zombie leader), so
the old check reported a dead scanner as "running".
"""
try:
pids = [d for d in os.listdir("/proc") if d.isdigit()]
except OSError:
return False
for pid in pids:
try:
with open(f"/proc/{pid}/stat") as f:
fields = f.read().rsplit(")", 1)[1].split()
state, pgrp = fields[0], int(fields[2])
if pgrp != pgid or state == "Z":
continue
with open(f"/proc/{pid}/cmdline", "rb") as f:
if b"multi_rx.py" in f.read():
return True
except (OSError, IndexError, ValueError):
continue
return False
def _reap() -> None:
if _proc is not None:
_proc.poll()
def _is_running() -> bool: def _is_running() -> bool:
pgid = _read_pgid() pgid = _read_pgid()
if pgid is None: if pgid is None:
return False return False
_reap()
return _multi_rx_alive(pgid)
def _kill_group(pgid: int) -> None:
try: try:
os.killpg(pgid, signal.SIGTERM) os.killpg(pgid, 0)
return True
except OSError: except OSError:
pass return False
def _pump(proc: subprocess.Popen) -> None:
"""Mirror OP25's output to the container log and keep the tail for /log."""
for line in proc.stdout:
sys.stdout.write(line)
sys.stdout.flush()
_log.append(line.rstrip("\n"))
rc = proc.wait()
_log.append(f"[OP25 exited with code {rc}]")
def create_op25_router(): def create_op25_router():
@@ -103,28 +44,11 @@ def create_op25_router():
@router.post("/start") @router.post("/start")
async def start_op25(): async def start_op25():
global _proc
if _is_running(): if _is_running():
return {"status": "OP25 already running"} return {"status": "OP25 already running"}
if not os.path.exists(_ACTIVE_CFG):
raise HTTPException(
status_code=409,
detail="No OP25 config yet — choose a system or frequency first.",
)
stale = _read_pgid()
if stale is not None:
# multi_rx died but liquidsoap/the wrapper may still hold the
# Icecast mount — clear the old group before starting a new one.
_kill_group(stale)
try: try:
_log.clear() proc = subprocess.Popen(os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH)
proc = subprocess.Popen(
os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
)
_proc = proc
_save_pgid(proc.pid) _save_pgid(proc.pid)
threading.Thread(target=_pump, args=(proc,), daemon=True).start()
LOGGER.debug(proc) LOGGER.debug(proc)
return {"status": "OP25 started"} return {"status": "OP25 started"}
except Exception as e: except Exception as e:
@@ -133,10 +57,10 @@ def create_op25_router():
@router.post("/stop") @router.post("/stop")
async def stop_op25(): async def stop_op25():
pgid = _read_pgid() pgid = _read_pgid()
if pgid is None: if pgid is None or not _is_running():
return {"status": "OP25 is not running"} return {"status": "OP25 is not running"}
try: try:
_kill_group(pgid) os.killpg(pgid, signal.SIGTERM)
os.remove(_PGID_FILE) os.remove(_PGID_FILE)
return {"status": "OP25 stopped"} return {"status": "OP25 stopped"}
except Exception as e: except Exception as e:
@@ -144,18 +68,7 @@ def create_op25_router():
@router.get("/status") @router.get("/status")
async def get_status(): async def get_status():
if _is_running(): return {"status": "running" if _is_running() else "stopped"}
return {"status": "running"}
# Started (pgid file present, never stopped via the API) but multi_rx
# is gone: it crashed. The tail of /log says why.
if _read_pgid() is not None:
return {"status": "crashed", "configured": os.path.exists(_ACTIVE_CFG)}
return {"status": "stopped", "configured": os.path.exists(_ACTIVE_CFG)}
@router.get("/log")
async def get_log(lines: int = 80):
lines = max(1, min(lines, _log.maxlen))
return {"lines": list(_log)[-lines:]}
@router.get("/devices") @router.get("/devices")
async def list_sdr_devices(): async def list_sdr_devices():
@@ -191,7 +104,7 @@ def create_op25_router():
meta_stream_name="stream_0" meta_stream_name="stream_0"
)] )]
devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)] devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)]
save_talkgroup_tags(generator.tags or []) save_talkgroup_tags(generator.tags)
save_whitelist(generator.whitelist) save_whitelist(generator.whitelist)
has_talkgroups = bool(generator.whitelist) has_talkgroups = bool(generator.whitelist)
trunking = TrunkingConfig( trunking = TrunkingConfig(
@@ -229,48 +142,30 @@ def create_op25_router():
"terminal": terminal.dict() "terminal": terminal.dict()
} }
elif generator.type == DecodeMode.ANALOG: elif generator.type == DecodeMode.ANALOG:
if not generator.channels: generator = generator.config
raise HTTPException(status_code=400, detail="Analog needs a frequency.")
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
freq = int(float(generator.channels[0]))
# One fixed conventional channel: no trunking section, so
# multi_rx adds its default tk_p25 with no chans and reports
# the channel as `conventional` on the HTTP terminal.
channels = [ChannelConfig( channels = [ChannelConfig(
name=generator.systemName, channelName=generator.systemName,
trunking_sysname="", enableAnalog="on",
enable_analog="on", demodType="fsk4",
demod_type="fsk4", frequency=generator.frequency,
filter_type="rc", filterType="widepulse",
frequency=freq, nbfmSquelch=generator.nbfmSquelch
meta_stream_name="",
nbfm_squelch_threshold=generator.nbfm_squelch,
)] )]
devices = [DeviceConfig( devices = [DeviceConfig(gain="LNA:32")]
gains=preset["gains"], ppm=ppm, tunable=False,
frequency=freq + ANALOG_CENTER_OFFSET_HZ,
)]
if generator.icecastConfig:
generate_liquid_script(generator.icecastConfig)
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]
"terminal": TerminalConfig().dict(),
} }
else: else:
raise HTTPException(status_code=400, detail="Unsupported system type. Must be 'P25' or 'NBFM' (DMR is not implemented).") raise HTTPException(status_code=400, detail="Invalid configuration type. Must be 'p25' or 'nbfm'.")
with open(_ACTIVE_CFG, 'w') as f: with open('/configs/active.cfg.json', 'w') as f:
json.dump(del_none_in_dict(config_dict), f, indent=2) json.dump(del_none_in_dict(config_dict), f, indent=2)
return {"message": f"Config exported to '{_ACTIVE_CFG}'"} return {"message": "Config exported to '/configs/active.cfg.json'"}
except HTTPException:
raise
except Exception as e: except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) raise HTTPException(status_code=500, detail=str(e))
-9
View File
@@ -20,12 +20,3 @@ LIQ_PID=$! # Store the PID of the op25.liq process
# The trap command ensures that SIGTERM is passed to the background jobs. # The trap command ensures that SIGTERM is passed to the background jobs.
trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT
wait $MULTI_RX_PID wait $MULTI_RX_PID
RC=$?
# multi_rx is gone (crashed, or SDR could not be opened). Take liquidsoap down
# with it so nothing in this process group outlives the decoder — a leftover
# liquidsoap kept /op25/status reporting "running" (node-26#12) and held the
# Icecast mount against the next start.
echo "multi_rx.py exited (code $RC) — stopping liquidsoap."
kill $LIQ_PID 2>/dev/null
exit $RC
+172
View File
@@ -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 "$@"
+107
View File
@@ -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"