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
Logan CusanoandClaude Opus 5.5 6bf2c9dd93 Merge feat/sdr-pins: pin each SDR service to a dongle by serial; OP25 always by serial (closes #11)
CI / lint (push) Successful in 10s
CI / test (push) Successful in 51s
Build edge-node / build (push) Successful in 58s
Build secondary-sdr / build (push) Successful in 2m39s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:41:28 -04:00
Logan CusanoandClaude Opus 5.5 575a99feb5 checkin: report an unreachable secondary-sdr as null, never keep stale 'Running'
QA blocker: when the secondary-sdr container was down, the checkin
omitted secondary_sdr_running / sdr_devices / op25_sdr_serial, and C2
only overwrites keys that are present, so the dashboard kept the last
'Running' forever. Send explicit nulls. The local card also says 'not
reported' rather than 'not plugged in' for a pin it can't check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:41:27 -04:00
Logan CusanoandClaude Opus 5.5 9b50ba8114 Pin each SDR service to a dongle by serial; OP25 always opens its SDR by serial
CI / lint (push) Successful in 6s
CI / test (push) Successful in 41s
Fixes node-26#11. OP25's generated config said "rtl" (= whichever dongle
enumerates first), so on 2-SDR nodes a decoder could take OP25's dongle
and stop recording. Now:

- sdr_pins: {op25|adsb|ais: serial}, absent = automatic. Every OP25
  config generation (op25_client.generate_config) rewrites the device to
  rtl=<serial>: the pin, else the first dongle's serial, which is what
  "rtl" always opened. Left as "rtl" only for unknown/shared serials.
- secondary-sdr: /secondary/devices lists dongles + serials via librtlsdr
  (works while claimed). apply(priority, pins, reserved) never touches
  OP25's dongle, gives a pinned service only its own dongle, lets a
  higher-priority service take a spare from a lower one, and still runs a
  pinned lower-priority service when the top pick has no dongle.
- sdr_settings.py replaces secondary_priority.py: one apply path for the
  local dashboard, the new set_sdr_config C2 command (set_secondary_priority
  kept as an alias) and config pushes. OP25 restarts only when its own
  dongle changes. Checkin reports sdr_devices, sdr_pins, op25_sdr_serial.
- Local dashboard: 'SDRs' card with an OP25 SDR dropdown and a per-service
  dongle dropdown, duplicate-serial and double-pin warnings.

Verified: edge-node pytest 194 passed; secondary-sdr tests 7 passed;
flake8 clean; page JS passes node --check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:35:22 -04:00
Logan CusanoandClaude Opus 5.5 3ecc7eff1a checkin: take sdr_count from secondary-sdr first; op25's 0 is never real
CI / lint (push) Successful in 7s
Build edge-node / build (push) Successful in 44s
CI / test (push) Successful in 48s
op25's :stable image has no lsusb, and /op25/devices answers count 0
instead of unknown, so the dashboard said radio-box 'reports 0 SDRs'
with 2 plugged in. Prefer the secondary-sdr container's lsusb count;
treat 0 from op25 as unknown.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:23:30 -04:00
Logan CusanoandClaude Opus 5.5 358766d88b Merge feat/secondary-sdr-priority: every spare SDR runs the next decoder in priority order
CI / lint (push) Successful in 11s
CI / test (push) Successful in 55s
Build edge-node / build (push) Successful in 1m12s
Build secondary-sdr / build (push) Successful in 3m29s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:14:53 -04:00
Logan CusanoandClaude Opus 5.5 c82fc61910 Local Secondary SDRs card: 'Unsaved' outranks 'Running'; unknown when unreachable
QA blocker: a reordered but unsaved list still showed the old order's
decoder as 'Running'. Also shows 'Unknown' rather than 'Waiting for SDR'
when the secondary-sdr service doesn't answer (server-26#187).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:14:49 -04:00
22 changed files with 1630 additions and 267 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)
+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
+124 -16
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:
@@ -234,14 +326,30 @@ class MQTTManager:
"secondary_sdr_mode": config.secondary_sdr_mode, "secondary_sdr_mode": config.secondary_sdr_mode,
"secondary_sdr_priority": config.secondary_sdr_priority, "secondary_sdr_priority": config.secondary_sdr_priority,
} }
payload["sdr_pins"] = config.sdr_pins
secondary = await secondary_sdr_client.status() secondary = await secondary_sdr_client.status()
# Unreachable secondary-sdr: say so with explicit nulls. C2 only
# overwrites keys present in the checkin, so omitting them would leave
# a dead decoder showing "Running" forever.
payload["secondary_sdr_running"] = None
payload["sdr_devices"] = None
payload["op25_sdr_serial"] = None
if secondary is not None: if secondary is not None:
payload["secondary_sdr_running"] = [r["mode"] for r in secondary.get("running", [])] payload["secondary_sdr_running"] = [r["mode"] for r in secondary.get("running", [])]
# Best-effort hardware report — omit rather than guess. op25's :stable if secondary.get("devices") is not None:
# image has no lsusb, so fall back to the secondary-sdr container's count. payload["sdr_devices"] = [
count = devices.get("count") if devices else None {k: d.get(k) for k in ("index", "serial", "name", "duplicate_serial")}
if count is None and secondary is not None: for d in secondary["devices"]
count = secondary.get("sdr_count") ]
from app.internal.sdr_settings import op25_serial
payload["op25_sdr_serial"] = op25_serial(config, secondary["devices"])
# Best-effort hardware report — omit rather than guess. Prefer the
# secondary-sdr container's count: op25's :stable image has no lsusb and
# its /op25/devices answers 0 rather than "unknown" when that fails. A
# node running op25 has at least one SDR, so 0 is never a real reading.
count = secondary.get("sdr_count") if secondary is not None else None
if not count and devices:
count = devices.get("count") or None
if count is not None: if count is not None:
payload["sdr_count"] = count payload["sdr_count"] = count
self._publish(self._t_checkin, payload, qos=1) self._publish(self._t_checkin, payload, qos=1)
+5 -1
View File
@@ -80,10 +80,14 @@ class OP25Client:
async with httpx.AsyncClient(timeout=10) as client: async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{self.api_url}/op25/generate-config", json=config) r = await client.post(f"{self.api_url}/op25/generate-config", json=config)
r.raise_for_status() r.raise_for_status()
return True
except Exception as e: except Exception as e:
logger.error(f"OP25 generate-config failed: {e}") logger.error(f"OP25 generate-config failed: {e}")
return False return False
# Every generated config opens its dongle by serial, never "first
# found" (node-26#11) — here so no generation path can skip it.
from app.internal.sdr_settings import pin_op25_device
await pin_op25_device()
return True
async def poll_terminal(self) -> Optional[TerminalUpdate]: async def poll_terminal(self) -> Optional[TerminalUpdate]:
""" """
+102
View File
@@ -0,0 +1,102 @@
"""
Which SDR does what on this node (node-26#9, node-26#11).
- OP25 always has exactly one dongle. `sdr_pins["op25"]` names it by serial;
unpinned, OP25 keeps its historical "first dongle" — but that dongle is now
named by serial too, so the decoders can never take it out from under OP25.
- Every other dongle runs the next enabled service in `secondary_sdr_priority`.
`sdr_pins[mode]` optionally binds a service to the dongle carrying its
antenna; unpinned services take any spare.
One apply path for the local dashboard, C2 commands and config pushes.
"""
import asyncio
import json
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional
from app.config import settings
from app.internal.config_manager import load_node_config, save_node_config
from app.internal.logger import logger
from app.internal.secondary_sdr_client import secondary_sdr_client
from app.models import NodeConfig, normalize_sdr_pins, normalize_secondary_priority
_OP25_CONFIG = Path(settings.config_path) / "active.cfg.json"
def op25_serial(cfg: NodeConfig, devs: Optional[List[Dict[str, Any]]]) -> Optional[str]:
"""The serial of OP25's dongle: the pin, else the first detected dongle
(what OP25's plain "rtl" device string has always opened)."""
if cfg.sdr_pins.get("op25"):
return cfg.sdr_pins["op25"]
return devs[0]["serial"] if devs else None
async def pin_op25_device() -> Optional[str]:
"""Rewrite OP25's generated config to open its dongle by serial.
Runs after every generate-config (op25_client.generate_config). A plain
"rtl" means "device 0", and which dongle is device 0 depends on who opened
what first — that is how OP25 lost its SDR to readsb on 2026-09-27.
Left as "rtl" only when the serial is unknown or shared by two dongles.
"""
devs = await secondary_sdr_client.devices()
serial = op25_serial(load_node_config(), devs)
if not serial or (devs and sum(d["serial"] == serial for d in devs) > 1):
return None
try:
cfg = json.loads(_OP25_CONFIG.read_text())
for dev in cfg.get("devices", []):
dev["args"] = f"rtl={serial}"
_OP25_CONFIG.write_text(json.dumps(cfg, indent=2))
except Exception as e:
logger.error(f"Could not pin OP25 to SDR {serial}: {e}")
return None
return serial
async def apply_secondaries() -> Optional[List[str]]:
"""Start/stop decoders to match the saved priority and pins, never on OP25's dongle."""
cfg = load_node_config()
if not cfg.secondary_sdr_priority:
return await secondary_sdr_client.apply([], {}, [])
devs = await secondary_sdr_client.devices()
reserved = [s for s in [op25_serial(cfg, devs)] if s]
pins = {m: s for m, s in cfg.sdr_pins.items() if m != "op25"}
return await secondary_sdr_client.apply(cfg.secondary_sdr_priority, pins, reserved)
async def set_sdr_settings(priority: Optional[Iterable[str]] = None,
pins: Optional[Dict[str, Optional[str]]] = None) -> Optional[List[str]]:
"""Persist and apply priority and/or pins. OP25 restarts only if its own
dongle changed; reordering the spare dongles never interrupts P25.
Returns the decoders now running (None if the container is unreachable).
"""
cfg = load_node_config()
old_op25 = cfg.sdr_pins.get("op25")
if priority is not None:
ordered = normalize_secondary_priority(priority)
cfg.secondary_sdr_priority = ordered
cfg.secondary_sdr_mode = ordered[0] if ordered else "none"
if pins is not None:
cfg.sdr_pins = normalize_sdr_pins(pins)
save_node_config(cfg)
if cfg.sdr_pins.get("op25") != old_op25:
from app.internal.op25_client import op25_client
# Free the new dongle first if a decoder holds it, then move OP25.
await secondary_sdr_client.apply([], {}, [])
serial = await pin_op25_device()
await op25_client.stop()
await asyncio.sleep(2)
await op25_client.start()
logger.info(f"OP25 moved to SDR {serial or 'rtl (first dongle)'}")
running = await apply_secondaries()
logger.info(f"SDR settings: priority={cfg.secondary_sdr_priority!r} pins={cfg.sdr_pins!r}; running {running!r}")
# Report straight away so C2's view doesn't wait for the next heartbeat.
from app.internal.mqtt_manager import mqtt_manager
asyncio.create_task(mqtt_manager.publish_checkin())
return running
@@ -1,29 +0,0 @@
import asyncio
from typing import Iterable, List, Optional
from app.internal.config_manager import load_node_config, save_node_config
from app.internal.logger import logger
from app.internal.secondary_sdr_client import secondary_sdr_client
from app.models import normalize_secondary_priority
async def set_secondary_priority(priority: Iterable[str]) -> Optional[List[str]]:
"""Persist and apply the node's secondary SDR priority — one path for the
local dashboard, a C2 command and a config push alike. Never touches op25:
changing what the spare dongles do must not interrupt P25 recording.
Returns the decoders now running (None if the container is unreachable).
"""
ordered = normalize_secondary_priority(priority)
cfg = load_node_config()
cfg.secondary_sdr_priority = ordered
cfg.secondary_sdr_mode = ordered[0] if ordered else "none"
save_node_config(cfg)
running = await secondary_sdr_client.apply(ordered)
logger.info(f"Secondary SDR priority set to {ordered!r}; running {running!r}")
# Report straight away so C2's view doesn't wait for the next heartbeat.
from app.internal.mqtt_manager import mqtt_manager
asyncio.create_task(mqtt_manager.publish_checkin())
return running
@@ -15,12 +15,25 @@ class SecondarySdrClient:
def __init__(self): def __init__(self):
self.api_url = settings.secondary_sdr_api_url self.api_url = settings.secondary_sdr_api_url
async def apply(self, priority: List[str]) -> Optional[List[str]]: async def devices(self) -> Optional[List[Dict[str, Any]]]:
"""Run decoders down the priority list until SDRs run out. Returns the """Every RTL-SDR on the node with its serial, or None if unreachable."""
modes actually running, or None if the container is unreachable.""" try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{self.api_url}/secondary/devices")
r.raise_for_status()
return r.json().get("devices", [])
except Exception as e:
logger.error(f"Secondary SDR device list failed: {e}")
return None
async def apply(self, priority: List[str], pins: Dict[str, str], reserved: List[str]) -> Optional[List[str]]:
"""Run decoders down the priority list until SDRs run out, honouring
pins and never touching `reserved` (op25's) dongles. Returns the modes
actually running, or None if the container is unreachable."""
body = {"priority": priority, "pins": pins, "reserved": reserved}
try: try:
async with httpx.AsyncClient(timeout=30) as client: async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{self.api_url}/secondary/apply", json={"priority": priority}) r = await client.post(f"{self.api_url}/secondary/apply", json=body)
r.raise_for_status() r.raise_for_status()
return r.json().get("running", []) return r.json().get("running", [])
except Exception as e: except Exception as e:
@@ -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
+36 -23
View File
@@ -5,7 +5,7 @@ from typing import Optional
from fastapi import FastAPI from fastapi import FastAPI
from app.config import settings from app.config import settings
from app.models import SystemConfig from app.models import SystemConfig, normalize_sdr_pins, normalize_secondary_priority
from app.internal.logger import logger from app.internal.logger import logger
from app.internal.mqtt_manager import mqtt_manager from app.internal.mqtt_manager import mqtt_manager
from app.internal import credentials from app.internal import credentials
@@ -158,26 +158,27 @@ async def on_command(payload: dict):
) )
elif action == "discord_leave": elif action == "discord_leave":
await radio_bot.leave() await radio_bot.leave()
elif action == "set_secondary_priority": elif action in ("set_sdr_config", "set_secondary_priority"):
from app.internal.secondary_priority import set_secondary_priority from app.internal.sdr_settings import set_sdr_settings
await set_secondary_priority(payload.get("priority") or []) try:
await set_sdr_settings(payload.get("priority"), payload.get("pins"))
except ValueError as e:
logger.error(f"Rejected SDR settings from C2: {e}")
elif action == "op25_restart": elif action == "op25_restart":
from app.internal.op25_client import op25_client from app.internal.op25_client import op25_client
await op25_client.stop() await op25_client.stop()
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}")
@@ -186,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.")
@@ -228,6 +230,7 @@ async def on_config_push(payload: dict):
ppm_override = payload.pop("ppm_override", None) ppm_override = payload.pop("ppm_override", None)
node_type = payload.pop("node_type", None) node_type = payload.pop("node_type", None)
secondary_sdr_priority = payload.pop("secondary_sdr_priority", None) secondary_sdr_priority = payload.pop("secondary_sdr_priority", None)
sdr_pins = payload.pop("sdr_pins", None)
secondary_sdr_mode = payload.pop("secondary_sdr_mode", None) # legacy single-mode C2 secondary_sdr_mode = payload.pop("secondary_sdr_mode", None) # legacy single-mode C2
enforce_override_timeout = payload.pop("enforce_override_timeout", None) enforce_override_timeout = payload.pop("enforce_override_timeout", None)
try: try:
@@ -250,6 +253,16 @@ async def on_config_push(payload: dict):
node_cfg.node_type = node_type node_cfg.node_type = node_type
if enforce_override_timeout is not None: if enforce_override_timeout is not None:
node_cfg.enforce_override_timeout = bool(enforce_override_timeout) node_cfg.enforce_override_timeout = bool(enforce_override_timeout)
if secondary_sdr_priority is None and secondary_sdr_mode is not None:
secondary_sdr_priority = [secondary_sdr_mode]
if secondary_sdr_priority is not None:
node_cfg.secondary_sdr_priority = normalize_secondary_priority(secondary_sdr_priority)
node_cfg.secondary_sdr_mode = (node_cfg.secondary_sdr_priority or ["none"])[0]
if sdr_pins is not None:
try:
node_cfg.sdr_pins = normalize_sdr_pins(sdr_pins)
except ValueError as e:
logger.error(f"Ignoring invalid sdr_pins in config push: {e}")
save_node_config(node_cfg) save_node_config(node_cfg)
from app.internal.op25_client import op25_client from app.internal.op25_client import op25_client
@@ -257,16 +270,16 @@ async def on_config_push(payload: dict):
logger.error(f"Failed to generate OP25 config for {config.name}") logger.error(f"Failed to generate OP25 config for {config.name}")
return return
# OP25's (pinned) dongle may be one a decoder holds right now: free the
# spares, restart OP25 on its own dongle, then hand the rest back out.
from app.internal.secondary_sdr_client import secondary_sdr_client
from app.internal.sdr_settings import apply_secondaries
await secondary_sdr_client.apply([], {}, [])
await op25_client.stop() await op25_client.stop()
await asyncio.sleep(2) await asyncio.sleep(2)
await op25_client.start() await op25_client.start()
logger.info(f"Config push applied: {config.name}") logger.info(f"Config push applied: {config.name}")
await apply_secondaries()
if secondary_sdr_priority is None and secondary_sdr_mode is not None:
secondary_sdr_priority = [secondary_sdr_mode]
if secondary_sdr_priority is not None:
from app.internal.secondary_priority import set_secondary_priority
await set_secondary_priority(secondary_sdr_priority)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -325,9 +338,9 @@ async def lifespan(app: FastAPI):
# After op25 has claimed its dongle, so the decoders only get the spares. # After op25 has claimed its dongle, so the decoders only get the spares.
if node_cfg.secondary_sdr_priority: if node_cfg.secondary_sdr_priority:
from app.internal.secondary_sdr_client import secondary_sdr_client from app.internal.sdr_settings import apply_secondaries
logger.info(f"Resuming secondary SDRs (priority={node_cfg.secondary_sdr_priority!r}) after restart.") logger.info(f"Resuming secondary SDRs (priority={node_cfg.secondary_sdr_priority!r}) after restart.")
await secondary_sdr_client.apply(node_cfg.secondary_sdr_priority) await apply_secondaries()
heartbeat_task = asyncio.create_task(mqtt_manager.heartbeat_loop()) heartbeat_task = asyncio.create_task(mqtt_manager.heartbeat_loop())
from app.internal.telemetry_uplink import telemetry_uplink_loop from app.internal.telemetry_uplink import telemetry_uplink_loop
+14
View File
@@ -38,6 +38,19 @@ def normalize_secondary_priority(items: Iterable[str]) -> List[str]:
return out return out
# Services that can be bound to a specific dongle by its USB serial.
SDR_PIN_KEYS = ("op25",) + SECONDARY_SDR_MODES
def normalize_sdr_pins(pins: Dict[str, Optional[str]]) -> Dict[str, str]:
"""Known services only, blank = automatic (dropped). Two services can't
share a dongle, so a serial claimed twice raises."""
out = {k: str(v).strip() for k, v in (pins or {}).items() if k in SDR_PIN_KEYS and v and str(v).strip()}
if len(set(out.values())) != len(out):
raise ValueError("Two services can't be pinned to the same SDR.")
return out
class NodeConfig(BaseModel): class NodeConfig(BaseModel):
node_id: str node_id: str
node_name: str node_name: str
@@ -50,6 +63,7 @@ class NodeConfig(BaseModel):
ppm_override: Optional[float] = None ppm_override: Optional[float] = None
node_type: str = "fixed" # fixed or portable node_type: str = "fixed" # fixed or portable
secondary_sdr_priority: List[str] = [] # ordered; SDRs beyond op25's run these top-down secondary_sdr_priority: List[str] = [] # ordered; SDRs beyond op25's run these top-down
sdr_pins: Dict[str, str] = {} # service (op25/adsb/ais) -> dongle serial; absent = automatic
# Legacy single-mode field (pre-priority). Still written as priority[0] so # Legacy single-mode field (pre-priority). Still written as priority[0] so
# an older C2 reading checkins sees something sensible; read only to # an older C2 reading checkins sees something sensible; read only to
# migrate a node_config.json saved before priority existed. # migrate a node_config.json saved before priority existed.
+30 -15
View File
@@ -1,5 +1,6 @@
from fastapi import APIRouter, Depends, HTTPException, Body from fastapi import APIRouter, Depends, HTTPException, Body
from typing import List, Optional from pydantic import BaseModel
from typing import Dict, List, Optional
import asyncio import asyncio
import httpx import httpx
from app.config import settings from app.config import settings
@@ -204,28 +205,42 @@ async def ack_override(timeout_minutes: int = Body(1440)):
raise HTTPException(500, f"Failed to contact C2: {e}") raise HTTPException(500, f"Failed to contact C2: {e}")
@router.get("/secondary") @router.get("/sdr")
async def get_secondary(): async def get_sdr():
"""Secondary SDR priority plus what's actually running (node-26#9).""" """Which SDR does what: detected dongles, pins, priority, what's running."""
from app.internal.secondary_sdr_client import secondary_sdr_client from app.internal.secondary_sdr_client import secondary_sdr_client
from app.internal.sdr_settings import op25_serial
cfg = load_node_config()
devs = await secondary_sdr_client.devices()
status = await secondary_sdr_client.status() status = await secondary_sdr_client.status()
return { return {
"priority": load_node_config().secondary_sdr_priority, "devices": devs, # None = secondary-sdr service unreachable
"op25_serial": op25_serial(cfg, devs),
"pins": cfg.sdr_pins,
"priority": cfg.secondary_sdr_priority,
"modes": list(SECONDARY_SDR_MODES), "modes": list(SECONDARY_SDR_MODES),
"running": [r["mode"] for r in status.get("running", [])] if status else None, "running": [r["mode"] for r in status.get("running", [])] if status else None,
"sdr_count": status.get("sdr_count") if status else None,
} }
@router.post("/secondary/priority") class SdrSettingsBody(BaseModel):
async def set_secondary_priority(priority: List[str] = Body(..., embed=True)): priority: Optional[List[str]] = None
"""Set the ordered list the spare SDRs work through. Never restarts op25.""" pins: Optional[Dict[str, Optional[str]]] = None # service -> serial; null/"" = automatic
from app.internal.secondary_priority import set_secondary_priority as apply_priority
unknown = [m for m in priority if m not in SECONDARY_SDR_MODES]
if unknown: @router.post("/sdr")
raise HTTPException(400, f"Unknown secondary SDR mode(s): {unknown}") async def set_sdr(body: SdrSettingsBody):
running = await apply_priority(priority) """Set priority and/or pins. Restarts OP25 only if OP25's own SDR changes."""
return {"ok": True, "priority": load_node_config().secondary_sdr_priority, "running": running} from app.internal.sdr_settings import set_sdr_settings
if body.priority is not None:
unknown = [m for m in body.priority if m not in SECONDARY_SDR_MODES]
if unknown:
raise HTTPException(400, f"Unknown secondary SDR mode(s): {unknown}")
try:
running = await set_sdr_settings(body.priority, body.pins)
except ValueError as e:
raise HTTPException(400, str(e))
return {"ok": True, "running": running}
@router.post("/discord/join") @router.post("/discord/join")
+83 -32
View File
@@ -128,6 +128,16 @@
.sdr-move:disabled { opacity: 0.3; cursor: default; } .sdr-move:disabled { opacity: 0.3; cursor: default; }
.sdr-state { font-size: 0.75rem; min-width: 6.5rem; text-align: right; color: var(--text-muted); } .sdr-state { font-size: 0.75rem; min-width: 6.5rem; text-align: right; color: var(--text-muted); }
.sdr-state.on { color: var(--success); } .sdr-state.on { color: var(--success); }
.sdr-select {
background: rgba(255, 255, 255, 0.05);
border: 1px solid var(--glass-border);
color: var(--text-main);
border-radius: 6px;
padding: 0.3rem 0.4rem;
font-size: 0.8rem;
max-width: 13rem;
}
.sdr-select option { background: #111827; }
.grid { .grid {
display: grid; display: grid;
@@ -384,18 +394,24 @@
</div> </div>
</div> </div>
<!-- Secondary SDRs Card (node-26#9) --> <!-- SDRs Card (node-26#9, #11) -->
<div class="glass-card"> <div class="glass-card">
<div class="card-header"> <div class="card-header">
<div class="card-title">Secondary SDRs</div> <div class="card-title">SDRs</div>
</div> </div>
<div class="data-row" style="align-items:center;">
<span class="data-label">OP25 SDR</span>
<select id="sdr-op25" class="sdr-select" aria-label="OP25 SDR"></select>
</div>
<p id="sdr-op25-note" style="color: var(--text-muted); font-size: 0.75rem; margin: 0.25rem 0 0.75rem;"></p>
<p style="color: var(--text-muted); font-size: 0.8rem; margin: 0 0 0.5rem;"> <p style="color: var(--text-muted); font-size: 0.8rem; margin: 0 0 0.5rem;">
OP25 always keeps its own SDR. Every other SDR runs the next enabled item below, top first. Every other SDR runs the next enabled service, top first. Pin a service to the SDR
<span id="sdr-summary"></span> that has its antenna, or leave it on "Any spare SDR".
</p> </p>
<div id="sdr-list"></div> <div id="sdr-list"></div>
<p id="sdr-warning" style="color: var(--danger); font-size: 0.8rem; margin: 0.5rem 0 0; display:none;"></p>
<div style="display:flex; gap:0.75rem; align-items:center; margin-top:0.75rem;"> <div style="display:flex; gap:0.75rem; align-items:center; margin-top:0.75rem;">
<button id="sdr-save" class="btn btn-primary" style="padding:0.5rem 1rem;" onclick="saveSdrPriority()" disabled>Save priority</button> <button id="sdr-save" class="btn btn-primary" style="padding:0.5rem 1rem;" onclick="saveSdr()" disabled>Save</button>
<span id="sdr-msg" style="color: var(--text-muted); font-size: 0.8rem;"></span> <span id="sdr-msg" style="color: var(--text-muted); font-size: 0.8rem;"></span>
</div> </div>
</div> </div>
@@ -532,40 +548,81 @@
} }
} }
// ── Secondary SDR priority ──────────────────────────────────────────── // ── SDRs: OP25's dongle, secondary priority and per-service pins ────────
const SDR_LABELS = { const SDR_LABELS = {
adsb: ['ADS-B', 'Aircraft · 1090 MHz'], adsb: ['ADS-B', 'Aircraft · 1090 MHz'],
ais: ['AIS', 'Vessels · 162 MHz'], ais: ['AIS', 'Vessels · 162 MHz'],
}; };
let sdrRows = []; // [{mode, enabled}] in display order let sdr = null; // last GET /api/sdr
let sdrRunning = []; let sdrRows = []; // [{mode, enabled, pin}] in display order
let sdrOp25Pin = ''; // '' = automatic
let sdrDirty = false; let sdrDirty = false;
function esc(v) {
return String(v ?? '').replace(/[&<>"']/g, c => ({'&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;'}[c]));
}
function deviceOptions(selected, autoLabel) {
const devs = sdr.devices || [];
const known = devs.some(d => d.serial === selected);
return `<option value="">${esc(autoLabel)}</option>` +
devs.map(d => `<option value="${esc(d.serial)}" ${d.serial === selected ? 'selected' : ''}>` +
`SDR ${d.index + 1} · serial ${esc(d.serial)}${d.duplicate_serial ? ' (shared serial!)' : ''}</option>`).join('') +
(selected && !known ? `<option value="${esc(selected)}" selected>serial ${esc(selected)} ` +
`(${sdr.devices === null ? 'not reported' : 'not plugged in'})</option>` : '');
}
function renderSdr() { function renderSdr() {
const op25 = document.getElementById('sdr-op25');
op25.innerHTML = deviceOptions(sdrOp25Pin, 'Automatic (first SDR)');
op25.onchange = () => { sdrOp25Pin = op25.value; markSdrDirty(); };
document.getElementById('sdr-op25-note').textContent = sdrOp25Pin !== (sdr.pins.op25 || '')
? 'Saving will restart OP25 on the selected SDR.'
: (sdr.op25_serial ? `OP25 is using serial ${sdr.op25_serial}.` : '');
const list = document.getElementById('sdr-list'); const list = document.getElementById('sdr-list');
list.innerHTML = ''; list.innerHTML = '';
let rank = 0; let rank = 0;
const running = sdr.running || [];
sdrRows.forEach((row, i) => { sdrRows.forEach((row, i) => {
const [name, hint] = SDR_LABELS[row.mode] || [row.mode, '']; const [name, hint] = SDR_LABELS[row.mode] || [row.mode, ''];
const running = sdrRunning.includes(row.mode); const pinMissing = row.pin && sdr.devices !== null && !sdr.devices.some(d => d.serial === row.pin);
const state = !row.enabled ? 'Off' : running ? 'Running' : (sdrDirty ? 'Unsaved' : 'Waiting for SDR'); const state = !row.enabled ? 'Off'
: sdrDirty ? 'Unsaved'
: sdr.running === null ? 'Unknown'
: running.includes(row.mode) ? 'Running'
: pinMissing ? 'Pinned SDR missing' : 'Waiting for SDR';
const el = document.createElement('div'); const el = document.createElement('div');
el.className = 'sdr-row'; el.className = 'sdr-row';
el.innerHTML = ` el.innerHTML = `
<span class="sdr-rank">${row.enabled ? ++rank : ''}</span> <span class="sdr-rank">${row.enabled ? ++rank : ''}</span>
<input type="checkbox" ${row.enabled ? 'checked' : ''} aria-label="Enable ${name}"> <input type="checkbox" ${row.enabled ? 'checked' : ''} aria-label="Enable ${name}">
<span class="sdr-name">${name}<small>${hint}</small></span> <span class="sdr-name">${name}<small>${hint}</small></span>
<select class="sdr-select" aria-label="${name} SDR">${deviceOptions(row.pin, 'Any spare SDR')}</select>
<button class="sdr-move" aria-label="Move ${name} up" ${i === 0 ? 'disabled' : ''}>▲</button> <button class="sdr-move" aria-label="Move ${name} up" ${i === 0 ? 'disabled' : ''}>▲</button>
<button class="sdr-move" aria-label="Move ${name} down" ${i === sdrRows.length - 1 ? 'disabled' : ''}>▼</button> <button class="sdr-move" aria-label="Move ${name} down" ${i === sdrRows.length - 1 ? 'disabled' : ''}>▼</button>
<span class="sdr-state ${running && row.enabled ? 'on' : ''}">${state}</span>`; <span class="sdr-state ${state === 'Running' ? 'on' : ''}">${state}</span>`;
const [box] = el.getElementsByTagName('input'); const [box] = el.getElementsByTagName('input');
const [pick] = el.getElementsByTagName('select');
const [up, down] = el.getElementsByTagName('button'); const [up, down] = el.getElementsByTagName('button');
box.onchange = () => { row.enabled = box.checked; markSdrDirty(); }; box.onchange = () => { row.enabled = box.checked; markSdrDirty(); };
pick.onchange = () => { row.pin = pick.value; markSdrDirty(); };
up.onclick = () => { [sdrRows[i - 1], sdrRows[i]] = [sdrRows[i], sdrRows[i - 1]]; markSdrDirty(); }; up.onclick = () => { [sdrRows[i - 1], sdrRows[i]] = [sdrRows[i], sdrRows[i - 1]]; markSdrDirty(); };
down.onclick = () => { [sdrRows[i + 1], sdrRows[i]] = [sdrRows[i], sdrRows[i + 1]]; markSdrDirty(); }; down.onclick = () => { [sdrRows[i + 1], sdrRows[i]] = [sdrRows[i], sdrRows[i + 1]]; markSdrDirty(); };
list.appendChild(el); list.appendChild(el);
}); });
document.getElementById('sdr-save').disabled = !sdrDirty;
const warn = document.getElementById('sdr-warning');
const pinned = [sdrOp25Pin, ...sdrRows.map(r => r.pin)].filter(Boolean);
const problems = [];
if (sdr.devices === null) problems.push('The secondary SDR service is not responding, so SDRs can\'t be listed.');
if ((sdr.devices || []).some(d => d.duplicate_serial)) {
problems.push('Two SDRs share a serial number, so they can\'t be told apart. Give each a unique serial (rtl_eeprom -s) before pinning.');
}
if (new Set(pinned).size !== pinned.length) problems.push('Two services are pinned to the same SDR.');
warn.textContent = problems.join(' ');
warn.style.display = problems.length ? 'block' : 'none';
document.getElementById('sdr-save').disabled = !sdrDirty || new Set(pinned).size !== pinned.length;
} }
function markSdrDirty() { function markSdrDirty() {
@@ -577,47 +634,41 @@
async function loadSdr() { async function loadSdr() {
if (sdrDirty) return; // never clobber unsaved edits with a poll if (sdrDirty) return; // never clobber unsaved edits with a poll
try { try {
const r = await fetch('/api/secondary'); const r = await fetch('/api/sdr');
if (!r.ok) return; if (!r.ok) return;
const d = await r.json(); sdr = await r.json();
sdrRunning = d.running || []; sdrOp25Pin = sdr.pins.op25 || '';
sdrRows = [ sdrRows = [
...d.priority.map(mode => ({ mode, enabled: true })), ...sdr.priority.map(mode => ({ mode, enabled: true, pin: sdr.pins[mode] || '' })),
...d.modes.filter(m => !d.priority.includes(m)).map(mode => ({ mode, enabled: false })), ...sdr.modes.filter(m => !sdr.priority.includes(m)).map(mode => ({ mode, enabled: false, pin: sdr.pins[mode] || '' })),
]; ];
const summary = document.getElementById('sdr-summary');
if (d.running === null) {
summary.textContent = 'Secondary SDR service is not responding.';
} else if (d.sdr_count != null) {
const spare = Math.max(d.sdr_count - 1, 0);
summary.textContent = `This node has ${d.sdr_count} SDR${d.sdr_count === 1 ? '' : 's'}, so ${spare} spare.`;
}
renderSdr(); renderSdr();
} catch (e) { } catch (e) {
console.error('Secondary SDR load failed:', e); console.error('SDR load failed:', e);
} }
} }
async function saveSdrPriority() { async function saveSdr() {
const priority = sdrRows.filter(r => r.enabled).map(r => r.mode); const pins = { op25: sdrOp25Pin || null };
sdrRows.forEach(r => { pins[r.mode] = r.pin || null; });
const btn = document.getElementById('sdr-save'); const btn = document.getElementById('sdr-save');
const msg = document.getElementById('sdr-msg'); const msg = document.getElementById('sdr-msg');
btn.disabled = true; btn.disabled = true;
msg.textContent = 'Applying…'; msg.textContent = 'Applying…';
try { try {
const r = await fetch('/api/secondary/priority', { const r = await fetch('/api/sdr', {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ priority }), body: JSON.stringify({ priority: sdrRows.filter(r => r.enabled).map(r => r.mode), pins }),
}); });
if (!r.ok) throw new Error(await r.text()); if (!r.ok) throw new Error((await r.json().catch(() => ({}))).detail || r.statusText);
const d = await r.json(); const d = await r.json();
sdrDirty = false; sdrDirty = false;
msg.textContent = d.running === null ? 'Saved, but the secondary SDR service did not respond.' : 'Saved.'; msg.textContent = d.running === null ? 'Saved, but the secondary SDR service did not respond.' : 'Saved.';
await loadSdr(); await loadSdr();
} catch (e) { } catch (e) {
console.error('Secondary SDR save failed:', e); console.error('SDR save failed:', e);
msg.textContent = 'Save failed.'; msg.textContent = `Save failed: ${e.message}`;
btn.disabled = false; btn.disabled = false;
} }
} }
@@ -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)
+91
View File
@@ -0,0 +1,91 @@
"""
node-26#9 / #11 — which SDR does what: secondary priority, per-service pins,
and OP25 always opening its dongle by serial.
"""
import asyncio
import json
from unittest.mock import AsyncMock, patch
import pytest
from app.models import NodeConfig, normalize_sdr_pins, normalize_secondary_priority
DEVS = [
{"index": 0, "serial": "69420", "name": "RTL", "duplicate_serial": False},
{"index": 1, "serial": "00000001", "name": "RTL", "duplicate_serial": False},
]
def _cfg(**kw) -> NodeConfig:
return NodeConfig(node_id="n1", node_name="N1", lat=0.0, lon=0.0, **kw)
def test_normalize_priority_keeps_order_drops_unknown_and_duplicates():
assert normalize_secondary_priority(["ais", "bogus", "adsb", "ais"]) == ["ais", "adsb"]
def test_legacy_single_mode_migrates_to_priority():
assert _cfg(secondary_sdr_mode="adsb").secondary_sdr_priority == ["adsb"]
assert _cfg(secondary_sdr_mode="adsb", secondary_sdr_priority=["ais"]).secondary_sdr_priority == ["ais"]
def test_pins_drop_blanks_and_unknown_services():
assert normalize_sdr_pins({"op25": "00000001", "adsb": "", "ais": None, "x": "1"}) == {"op25": "00000001"}
def test_two_services_cannot_share_a_dongle():
with pytest.raises(ValueError):
normalize_sdr_pins({"op25": "69420", "adsb": "69420"})
@pytest.fixture
def node(tmp_path):
"""Isolated node_config.json + OP25 active.cfg.json, mocked decoders/op25/mqtt."""
import app.internal.config_manager as cm
from app.internal import sdr_settings as ss
op25_cfg = tmp_path / "active.cfg.json"
op25_cfg.write_text(json.dumps({"devices": [{"args": "rtl", "name": "sdr"}]}))
with patch.object(cm, "_CONFIG_FILE", tmp_path / "node_config.json"), \
patch.object(ss, "_OP25_CONFIG", op25_cfg), \
patch.object(ss.secondary_sdr_client, "devices", AsyncMock(return_value=DEVS)), \
patch.object(ss.secondary_sdr_client, "apply", AsyncMock(return_value=["adsb"])) as apply, \
patch("app.internal.op25_client.op25_client.stop", AsyncMock()) as op25_stop, \
patch("app.internal.op25_client.op25_client.start", AsyncMock()), \
patch("app.internal.mqtt_manager.mqtt_manager.publish_checkin", AsyncMock()), \
patch("asyncio.sleep", AsyncMock()):
cm.save_node_config(_cfg())
yield {"cm": cm, "ss": ss, "apply": apply, "op25_stop": op25_stop,
"op25_args": lambda: json.loads(op25_cfg.read_text())["devices"][0]["args"]}
def test_unpinned_op25_is_still_opened_by_serial_of_the_first_dongle(node):
assert asyncio.run(node["ss"].pin_op25_device()) == "69420"
assert node["op25_args"]() == "rtl=69420"
def test_pinned_op25_opens_its_pinned_dongle(node):
node["cm"].save_node_config(_cfg(sdr_pins={"op25": "00000001"}))
asyncio.run(node["ss"].pin_op25_device())
assert node["op25_args"]() == "rtl=00000001"
def test_shared_serial_leaves_op25_on_plain_rtl(node):
dup = [dict(d, serial="00000001") for d in DEVS]
with patch.object(node["ss"].secondary_sdr_client, "devices", AsyncMock(return_value=dup)):
assert asyncio.run(node["ss"].pin_op25_device()) is None
assert node["op25_args"]() == "rtl"
def test_priority_change_never_restarts_op25_and_reserves_its_dongle(node):
node["cm"].save_node_config(_cfg(sdr_pins={"op25": "00000001"}))
asyncio.run(node["ss"].set_sdr_settings(priority=["adsb", "ais"]))
node["op25_stop"].assert_not_awaited()
node["apply"].assert_awaited_with(["adsb", "ais"], {}, ["00000001"])
def test_moving_op25_restarts_it_on_the_new_dongle(node):
asyncio.run(node["ss"].set_sdr_settings(priority=["adsb"], pins={"op25": "00000001", "adsb": "69420"}))
node["op25_stop"].assert_awaited_once()
assert node["op25_args"]() == "rtl=00000001"
node["apply"].assert_awaited_with(["adsb"], {"adsb": "69420"}, ["00000001"])
assert node["cm"].load_node_config().sdr_pins == {"op25": "00000001", "adsb": "69420"}
@@ -1,58 +0,0 @@
"""
node-26#9 — secondary SDR priority: an ordered list the SDRs beyond op25's work
through. Normalisation, migration from the legacy single-mode field, and the one
apply path shared by the dashboard, C2 commands and config pushes.
"""
import asyncio
from unittest.mock import AsyncMock, patch
from app.models import NodeConfig, normalize_secondary_priority
def _cfg(**kw) -> NodeConfig:
return NodeConfig(node_id="n1", node_name="N1", lat=0.0, lon=0.0, **kw)
def test_normalize_keeps_order_drops_unknown_and_duplicates():
assert normalize_secondary_priority(["ais", "bogus", "adsb", "ais"]) == ["ais", "adsb"]
assert normalize_secondary_priority([]) == []
def test_legacy_single_mode_migrates_to_priority():
assert _cfg(secondary_sdr_mode="adsb").secondary_sdr_priority == ["adsb"]
assert _cfg(secondary_sdr_mode="none").secondary_sdr_priority == []
def test_explicit_priority_wins_over_legacy_mode():
assert _cfg(secondary_sdr_mode="adsb", secondary_sdr_priority=["ais"]).secondary_sdr_priority == ["ais"]
def test_set_priority_persists_applies_and_never_touches_op25(tmp_path):
config_file = tmp_path / "node_config.json"
import app.internal.config_manager as cm
with patch.object(cm, "_CONFIG_FILE", config_file):
cm.save_node_config(_cfg(secondary_sdr_mode="adsb"))
from app.internal import secondary_priority as sp
with patch.object(sp.secondary_sdr_client, "apply", AsyncMock(return_value=["ais"])) as apply, \
patch("app.internal.mqtt_manager.mqtt_manager.publish_checkin", AsyncMock()), \
patch("app.internal.op25_client.op25_client.stop", AsyncMock()) as op25_stop:
running = asyncio.run(sp.set_secondary_priority(["ais", "adsb", "ais"]))
saved = cm.load_node_config()
assert running == ["ais"]
apply.assert_awaited_once_with(["ais", "adsb"])
op25_stop.assert_not_awaited()
assert saved.secondary_sdr_priority == ["ais", "adsb"]
assert saved.secondary_sdr_mode == "ais"
def test_clearing_priority_is_not_undone_by_legacy_migration(tmp_path):
config_file = tmp_path / "node_config.json"
import app.internal.config_manager as cm
with patch.object(cm, "_CONFIG_FILE", config_file):
cm.save_node_config(_cfg(secondary_sdr_mode="adsb"))
from app.internal import secondary_priority as sp
with patch.object(sp.secondary_sdr_client, "apply", AsyncMock(return_value=[])), \
patch("app.internal.mqtt_manager.mqtt_manager.publish_checkin", AsyncMock()):
asyncio.run(sp.set_secondary_priority([]))
assert cm.load_node_config().secondary_sdr_priority == []
+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
+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"
@@ -1,3 +1,5 @@
import ctypes
import ctypes.util
import json import json
import os import os
import signal import signal
@@ -141,8 +143,9 @@ def running() -> List[str]:
return [m for m in MODES if is_running(m)] return [m for m in MODES if is_running(m)]
def _start_one(mode: str) -> bool: def _start_one(mode: str, candidates: Optional[List[int]] = None) -> bool:
"""Start one decoder on the first free SDR. False when none is free.""" """Start one decoder on the first free SDR among `candidates` (default:
every index). False when none is free."""
if mode not in _COMMANDS: if mode not in _COMMANDS:
raise ValueError(f"Unknown secondary SDR mode: {mode!r}") raise ValueError(f"Unknown secondary SDR mode: {mode!r}")
if is_running(mode): if is_running(mode):
@@ -152,7 +155,7 @@ def _start_one(mode: str) -> bool:
_ais_vessels.clear() _ais_vessels.clear()
needs_stdout = mode == "ais" needs_stdout = mode == "ais"
for index in range(MAX_SDR_INDEX): for index in candidates if candidates is not None else range(MAX_SDR_INDEX):
if index in {_indices[m] for m in running() if m in _indices}: if index in {_indices[m] for m in running() if m in _indices}:
continue continue
try: try:
@@ -211,47 +214,95 @@ def stop(mode: Optional[str] = None) -> None:
_stop_one(m) _stop_one(m)
def apply(priority: List[str]) -> List[str]: def _candidates(mode: str, pins: Dict[str, str], reserved: List[str], devs: List[Dict[str, Any]]) -> List[int]:
"""SDR indices `mode` may use. A pinned mode gets exactly its dongle; an
unpinned one gets any dongle that isn't op25's (reserved) or pinned to
another service. Without enumeration, fall back to probing every index."""
if not devs:
return [] if pins.get(mode) else list(range(MAX_SDR_INDEX))
if pins.get(mode):
idx = _index_of(pins[mode], devs)
return [] if idx is None else [idx]
taken = set(reserved) | {s for m, s in pins.items() if m != mode and s}
return [d["index"] for d in devs if d["serial"] not in taken]
def apply(priority: List[str], pins: Optional[Dict[str, str]] = None,
reserved: Optional[List[str]] = None) -> List[str]:
"""Run decoders in priority order until SDRs run out; stop everything else. """Run decoders in priority order until SDRs run out; stop everything else.
Unchanged when the running set is already the right prefix of the list, so `pins` maps a mode to the serial of the dongle carrying its antenna;
re-applying the same priority (every restart/config push) is a no-op `reserved` lists serials no decoder may touch (op25's). Pins only bind
rather than a decoder restart. enabled modes — a disabled service's dongle is free for the others.
Unchanged when the right decoders already run on allowed dongles, so
re-applying the same settings is a no-op rather than a restart.
""" """
for m in priority: for m in priority:
if m not in _COMMANDS: if m not in _COMMANDS:
raise ValueError(f"Unknown secondary SDR mode: {m!r}") raise ValueError(f"Unknown secondary SDR mode: {m!r}")
pins = {m: s for m, s in (pins or {}).items() if m in priority and s}
reserved = [s for s in (reserved or []) if s]
devs = devices()
with _lock: with _lock:
live = set(running()) for m in running():
if live and live == set(priority[:len(live)]): if m not in priority or _indices.get(m) not in _candidates(m, pins, reserved, devs):
# Already running the head of the list; only try to extend it. _stop_one(m)
for m in priority[len(live):]: for i, m in enumerate(priority):
if not _start_one(m): if m in running():
break continue
return running() cands = _candidates(m, pins, reserved, devs)
for m in list(_procs): if _start_one(m, cands):
_stop_one(m) continue
for m in priority: # Out of free dongles: take one from the lowest-priority decoder
if not _start_one(m): # holding a dongle this mode may use, which then gets its own turn.
break lower = [x for x in priority[i + 1:] if x in running() and _indices.get(x) in cands]
if lower:
_stop_one(lower[-1])
_start_one(m, cands)
return running() return running()
def sdr_count() -> Optional[int]: def devices() -> List[Dict[str, Any]]:
"""RTL-SDR dongles on USB, same heuristic as install.sh. This container has """Every RTL-SDR on the node with its USB serial — readable even while a
usbutils; op25's :stable image doesn't, so its /op25/devices can't answer.""" dongle is claimed (op25's included), since it doesn't open the device.
Cheap dongles often ship with the same serial (00000001); those are
flagged, because pinning a service to a shared serial is ambiguous."""
try: try:
out = subprocess.run(["lsusb"], capture_output=True, text=True, timeout=5).stdout lib = ctypes.CDLL(ctypes.util.find_library("rtlsdr") or "librtlsdr.so.0")
except Exception: lib.rtlsdr_get_device_name.restype = ctypes.c_char_p
return None out = []
return sum(1 for line in out.splitlines() if "0bda:2838" in line or "0bda:2832" in line) for i in range(lib.rtlsdr_get_device_count()):
manufact, product, serial = (ctypes.create_string_buffer(256) for _ in range(3))
lib.rtlsdr_get_device_usb_strings(i, manufact, product, serial)
out.append({
"index": i,
"serial": serial.value.decode(errors="replace") or None,
"name": (lib.rtlsdr_get_device_name(i) or b"").decode(errors="replace"),
})
except Exception as e:
LOGGER.warning(f"SDR enumeration failed: {e}")
return []
serials = [d["serial"] for d in out]
for d in out:
d["duplicate_serial"] = d["serial"] is not None and serials.count(d["serial"]) > 1
return out
def _index_of(serial: str, devs: List[Dict[str, Any]]) -> Optional[int]:
"""Index of a uniquely-identified serial; None if absent or ambiguous."""
matches = [d["index"] for d in devs if d["serial"] == serial]
return matches[0] if len(matches) == 1 else None
def status() -> Dict[str, Any]: def status() -> Dict[str, Any]:
live = running() devs = devices()
by_index = {d["index"]: d["serial"] for d in devs}
return { return {
"running": [{"mode": m, "sdr_index": _indices.get(m)} for m in live], "running": [
"sdr_count": sdr_count(), {"mode": m, "sdr_index": _indices.get(m), "serial": by_index.get(_indices.get(m))} for m in running()
],
"sdr_count": len(devs) if devs else None,
"devices": devs,
} }
@@ -1,4 +1,4 @@
from typing import List, Optional from typing import Dict, List, Optional
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, HTTPException
from pydantic import BaseModel from pydantic import BaseModel
@@ -19,6 +19,8 @@ class StopBody(BaseModel):
class ApplyBody(BaseModel): class ApplyBody(BaseModel):
priority: List[str] # ordered, e.g. ["adsb", "ais"] priority: List[str] # ordered, e.g. ["adsb", "ais"]
pins: Dict[str, str] = {} # mode -> serial of the dongle with its antenna
reserved: List[str] = [] # serials no decoder may touch (op25's)
def create_secondary_router(): def create_secondary_router():
@@ -27,7 +29,7 @@ def create_secondary_router():
@router.post("/apply") @router.post("/apply")
async def apply(body: ApplyBody): async def apply(body: ApplyBody):
try: try:
live = decoder_control.apply(body.priority) live = decoder_control.apply(body.priority, body.pins, body.reserved)
except ValueError as e: except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) raise HTTPException(status_code=400, detail=str(e))
return {"running": live} return {"running": live}
@@ -51,6 +53,10 @@ def create_secondary_router():
async def get_status(): async def get_status():
return decoder_control.status() return decoder_control.status()
@router.get("/devices")
async def get_devices():
return {"devices": decoder_control.devices()}
@router.get("/data") @router.get("/data")
async def get_data(): async def get_data():
return decoder_control.data() return decoder_control.data()
@@ -0,0 +1,84 @@
"""
node-26#11 — which dongle each secondary decoder may use.
Run from secondary-sdr-container/: PYTHONPATH=app python -m pytest -q tests
"""
from unittest.mock import patch
import pytest
from internal import decoder_control as dc
# radio-box's real pair: op25's dongle and the one with the 1090 antenna.
DEVS = [
{"index": 0, "serial": "69420", "name": "RTL", "duplicate_serial": False},
{"index": 1, "serial": "00000001", "name": "RTL", "duplicate_serial": False},
]
def test_unpinned_mode_never_gets_op25s_dongle():
assert dc._candidates("adsb", {}, ["00000001"], DEVS) == [0]
def test_pinned_mode_gets_exactly_its_dongle():
assert dc._candidates("ais", {"ais": "69420"}, ["00000001"], DEVS) == [0]
def test_unpinned_mode_skips_a_dongle_pinned_to_another_service():
assert dc._candidates("ais", {"adsb": "69420"}, ["00000001"], DEVS) == []
def test_missing_or_ambiguous_pin_gets_nothing():
assert dc._candidates("adsb", {"adsb": "nope"}, [], DEVS) == []
dup = [dict(d, serial="00000001") for d in DEVS]
assert dc._candidates("adsb", {"adsb": "00000001"}, [], dup) == []
class FakeDecoders:
"""Stands in for real processes: one decoder per free index."""
def __init__(self):
self.live = {}
def start(self, mode, candidates=None):
free = [i for i in candidates if i not in self.live.values()]
if not free:
return False
self.live[mode] = free[0]
return True
def stop(self, mode):
self.live.pop(mode, None)
@pytest.fixture
def fake():
f = FakeDecoders()
with patch.object(dc, "devices", return_value=DEVS), \
patch.object(dc, "_start_one", side_effect=f.start), \
patch.object(dc, "_stop_one", side_effect=f.stop), \
patch.object(dc, "running", side_effect=lambda: [m for m in dc.MODES if m in f.live]), \
patch.dict(dc._indices, clear=True):
dc._indices.update(f.live)
yield f
def _apply(fake, priority, pins=None):
dc.apply(priority, pins, ["00000001"])
dc._indices.clear()
dc._indices.update(fake.live)
return fake.live
def test_one_spare_goes_to_the_top_pick(fake):
assert _apply(fake, ["ais", "adsb"]) == {"ais": 0}
def test_reordering_hands_the_spare_to_the_new_top_pick(fake):
_apply(fake, ["adsb", "ais"])
assert _apply(fake, ["ais", "adsb"]) == {"ais": 0}
def test_pinned_lower_priority_still_runs_when_top_pick_has_no_dongle(fake):
# AIS ranked first but its only candidate is ADS-B's pinned antenna dongle.
assert _apply(fake, ["ais", "adsb"], {"adsb": "69420"}) == {"adsb": 0}