Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
797e14481b | ||
|
|
10f7db0a2c |
@@ -3,12 +3,8 @@ 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.
|
||||||
install.sh fetches it over HTTPS (GET /nodes/{id}/credentials + the pickup
|
It arrives via MQTT and is saved to /configs/credentials.json so it survives
|
||||||
secret it saved to /configs/pickup_secret) and writes /configs/credentials.json;
|
container restarts.
|
||||||
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
|
||||||
@@ -16,14 +12,12 @@ 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
|
||||||
@@ -80,37 +74,6 @@ 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
|
||||||
@@ -125,10 +88,4 @@ 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()
|
||||||
# Atomic replace with 0600: install.sh on the host writes this same file,
|
_CREDS_FILE.write_text(json.dumps(data))
|
||||||
# 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)
|
|
||||||
|
|||||||
@@ -0,0 +1,300 @@
|
|||||||
|
"""
|
||||||
|
Plain-language health checklist for the local dashboard (/api/health).
|
||||||
|
|
||||||
|
A first-time owner opens http://<node>/ and needs three answers: is this node
|
||||||
|
working, if not what is wrong, and what do I do about it. Each check below
|
||||||
|
answers one of those for one link in the chain
|
||||||
|
|
||||||
|
password -> C2 enrollment -> MQTT -> SDR -> what to listen to -> OP25
|
||||||
|
-> decoding -> audio (PulseAudio) -> local stream (Icecast)
|
||||||
|
|
||||||
|
`gather_facts()` does the I/O; `build_checks()` is a pure function of those
|
||||||
|
facts so every state (unapproved, no SDR, OP25 crashed, …) is unit-testable.
|
||||||
|
|
||||||
|
Read-only: nothing here changes node state, and it never polls the OP25
|
||||||
|
terminal itself (that drains OP25's call_log, which metadata_watcher owns) —
|
||||||
|
it reads metadata_watcher.console_state instead.
|
||||||
|
"""
|
||||||
|
import time
|
||||||
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal.logger import logger
|
||||||
|
|
||||||
|
OK, WARN, ERROR, OFF = "ok", "warn", "error", "off"
|
||||||
|
|
||||||
|
# OP25's console answers within a second or two of a healthy multi_rx; past
|
||||||
|
# this the console is considered silent.
|
||||||
|
CONSOLE_STALE_SECONDS = 5.0
|
||||||
|
# P25 control channels send several messages a second; this long without one
|
||||||
|
# means OP25 is not decoding.
|
||||||
|
TSBK_STALE_SECONDS = 15.0
|
||||||
|
|
||||||
|
|
||||||
|
def _check(cid: str, label: str, state: str, summary: str, action: str = "") -> Dict[str, str]:
|
||||||
|
return {"id": cid, "label": label, "state": state, "summary": summary, "action": action}
|
||||||
|
|
||||||
|
|
||||||
|
def _mhz(freq: Any) -> Optional[str]:
|
||||||
|
try:
|
||||||
|
f = float(freq)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
if f <= 0:
|
||||||
|
return None
|
||||||
|
return f"{f / 1e6:.4f} MHz" if f > 1e5 else f"{f:.4f} MHz"
|
||||||
|
|
||||||
|
|
||||||
|
def _ago(seconds: float) -> str:
|
||||||
|
seconds = max(0, int(seconds))
|
||||||
|
if seconds < 60:
|
||||||
|
return f"{seconds}s ago"
|
||||||
|
if seconds < 3600:
|
||||||
|
return f"{seconds // 60} min ago"
|
||||||
|
return f"{seconds // 3600} h ago"
|
||||||
|
|
||||||
|
|
||||||
|
def active_config(node_cfg) -> Any:
|
||||||
|
if node_cfg.override_system_id and node_cfg.override_config:
|
||||||
|
return node_cfg.override_config
|
||||||
|
return node_cfg.system_config
|
||||||
|
|
||||||
|
|
||||||
|
def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]:
|
||||||
|
"""Facts -> ordered checklist. See gather_facts() for the keys."""
|
||||||
|
now = f.get("now", time.time())
|
||||||
|
checks: List[Dict[str, str]] = []
|
||||||
|
|
||||||
|
# 1. Dashboard password
|
||||||
|
if f["default_password"]:
|
||||||
|
checks.append(_check(
|
||||||
|
"password", "Dashboard password", WARN,
|
||||||
|
"Still the factory default — anyone on this network can log in.",
|
||||||
|
"Set DASHBOARD_PASSWORD in the node's .env file, then restart the edge-node container.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check("password", "Dashboard password", OK, "Changed from the default."))
|
||||||
|
|
||||||
|
# 2. C2 enrollment / approval
|
||||||
|
c2_url, has_key = f["c2_url"], f["has_api_key"]
|
||||||
|
if not c2_url:
|
||||||
|
checks.append(_check(
|
||||||
|
"c2", "Server (C2) link", OFF,
|
||||||
|
"Standalone — no C2 server is set. Local listening below still works.",
|
||||||
|
"To join a DRB server, re-run install.sh with an enrollment token.",
|
||||||
|
))
|
||||||
|
elif not has_key:
|
||||||
|
checks.append(_check(
|
||||||
|
"c2", "Server (C2) link", WARN,
|
||||||
|
f"Waiting for approval — this node has no API key from {c2_url} yet, so nothing is uploaded.",
|
||||||
|
f"Ask a DRB admin to approve node \"{f['node_id']}\" on the Nodes page, then re-run "
|
||||||
|
"install.sh (or restart the edge-node container). Local listening below works meanwhile.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check("c2", "Server (C2) link", OK, f"Approved — API key on file for {c2_url}."))
|
||||||
|
|
||||||
|
# 3. MQTT
|
||||||
|
broker = f"{f['mqtt_broker']}:{f['mqtt_port']}"
|
||||||
|
if f["mqtt_connected"]:
|
||||||
|
checks.append(_check("mqtt", "Live link (MQTT)", OK, f"Connected to {broker}."))
|
||||||
|
elif not c2_url and not has_key:
|
||||||
|
checks.append(_check(
|
||||||
|
"mqtt", "Live link (MQTT)", OFF, f"Not connected to {broker} (expected for a standalone node).",
|
||||||
|
))
|
||||||
|
elif not has_key:
|
||||||
|
checks.append(_check(
|
||||||
|
"mqtt", "Live link (MQTT)", WARN,
|
||||||
|
f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).",
|
||||||
|
"Fixes itself once the node is approved and has its API key (see above).",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check(
|
||||||
|
"mqtt", "Live link (MQTT)", ERROR,
|
||||||
|
f"Not connected to {broker} even though an API key is on file.",
|
||||||
|
"If the key was just installed or the node was re-approved, restart the edge-node container "
|
||||||
|
"so it loads the key. If it still fails, the key may be stale (every re-approval issues a new "
|
||||||
|
"one) — re-run install.sh, and check the node can reach the broker.",
|
||||||
|
))
|
||||||
|
|
||||||
|
# 4. SDR
|
||||||
|
sdr_count = f.get("sdr_count")
|
||||||
|
if sdr_count is None:
|
||||||
|
checks.append(_check(
|
||||||
|
"sdr", "SDR receiver", WARN, "Couldn't check — the SDR service isn't answering.",
|
||||||
|
"Check the secondary-sdr container is running.",
|
||||||
|
))
|
||||||
|
elif sdr_count == 0:
|
||||||
|
checks.append(_check(
|
||||||
|
"sdr", "SDR receiver", ERROR, "No RTL-SDR dongle detected.",
|
||||||
|
"Plug the SDR into a USB port (avoid unpowered hubs), then restart the stack.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check(
|
||||||
|
"sdr", "SDR receiver", OK, f"{sdr_count} SDR{'s' if sdr_count != 1 else ''} detected.",
|
||||||
|
))
|
||||||
|
|
||||||
|
# 5. What to listen to
|
||||||
|
src = f.get("source")
|
||||||
|
if not src:
|
||||||
|
checks.append(_check(
|
||||||
|
"source", "What to listen to", WARN, "Nothing chosen yet.",
|
||||||
|
"Pick a system or enter a frequency under \"Listen locally\" below.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
origin = "chosen on this dashboard" if src["local"] else "assigned by the server"
|
||||||
|
checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({src['type']}) — {origin}."))
|
||||||
|
|
||||||
|
# 6. OP25
|
||||||
|
op25 = f.get("op25_status")
|
||||||
|
running = op25 == "running"
|
||||||
|
if op25 is None:
|
||||||
|
checks.append(_check(
|
||||||
|
"op25", "Radio decoder (OP25)", ERROR, "The op25 container isn't answering.",
|
||||||
|
"Check the op25 container is running; its log usually says why it stopped.",
|
||||||
|
))
|
||||||
|
elif running:
|
||||||
|
checks.append(_check("op25", "Radio decoder (OP25)", OK, "Running."))
|
||||||
|
elif op25 == "crashed":
|
||||||
|
checks.append(_check(
|
||||||
|
"op25", "Radio decoder (OP25)", ERROR, "Stopped unexpectedly.",
|
||||||
|
"Open \"OP25 log\" below for the reason — usually the SDR is unplugged or in use. Then press Start.",
|
||||||
|
))
|
||||||
|
elif not src:
|
||||||
|
checks.append(_check("op25", "Radio decoder (OP25)", OFF, "Stopped — choose what to listen to first."))
|
||||||
|
else:
|
||||||
|
checks.append(_check("op25", "Radio decoder (OP25)", WARN, "Stopped.", "Press Start."))
|
||||||
|
|
||||||
|
# 7. Decoding
|
||||||
|
console = f.get("console") or {}
|
||||||
|
last_poll = console.get("last_poll_epoch")
|
||||||
|
is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG")
|
||||||
|
if not running:
|
||||||
|
checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs."))
|
||||||
|
elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS:
|
||||||
|
checks.append(_check(
|
||||||
|
"decode", "Decoding", WARN, "OP25 is running but hasn't reported in yet.",
|
||||||
|
"Give it 20 seconds after a start. If this stays, open the OP25 log below.",
|
||||||
|
))
|
||||||
|
elif is_analog:
|
||||||
|
chans = console.get("channels") or []
|
||||||
|
freq = _mhz(chans[0].get("freq")) if chans else None
|
||||||
|
checks.append(_check(
|
||||||
|
"decode", "Decoding", OK, f"Tuned to {freq or 'the analog frequency'} (analog FM — no decoding needed).",
|
||||||
|
"Live listening only: analog calls are not recorded or uploaded yet (node-26#20).",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
trunk = console.get("trunk") or []
|
||||||
|
tsbk = [float(t.get("last_tsbk") or 0) for t in trunk]
|
||||||
|
if not trunk:
|
||||||
|
checks.append(_check("decode", "Decoding", WARN, "Waiting for OP25 to report control-channel status…"))
|
||||||
|
elif max(tsbk) > 0 and now - max(tsbk) < TSBK_STALE_SECONDS:
|
||||||
|
sys_ = trunk[tsbk.index(max(tsbk))]
|
||||||
|
cc = _mhz(sys_.get("rxchan"))
|
||||||
|
checks.append(_check(
|
||||||
|
"decode", "Decoding", OK, f"Locked on the control channel{' ' + cc if cc else ''}.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check(
|
||||||
|
"decode", "Decoding", ERROR, "Not decoding the control channel.",
|
||||||
|
"Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM). "
|
||||||
|
"Try another control channel for the same system.",
|
||||||
|
))
|
||||||
|
|
||||||
|
# 8. Audio (PulseAudio — recording + Discord path)
|
||||||
|
if not f["audio_capturing"]:
|
||||||
|
checks.append(_check(
|
||||||
|
"audio", "Audio", ERROR if running else OFF,
|
||||||
|
"No audio reaching the recorder." if running else "Nothing to hear until OP25 runs.",
|
||||||
|
"The op25 container owns the audio path (PulseAudio); restart it if this persists." if running else "",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
last_voice = f.get("last_voice_epoch")
|
||||||
|
if last_voice is None:
|
||||||
|
checks.append(_check("audio", "Audio", OK, "Audio path up — no voice heard yet."))
|
||||||
|
elif now - last_voice < 3:
|
||||||
|
checks.append(_check("audio", "Audio", OK, "Voice right now."))
|
||||||
|
else:
|
||||||
|
checks.append(_check("audio", "Audio", OK, f"Audio path up — last voice {_ago(now - last_voice)}."))
|
||||||
|
|
||||||
|
# 9. Local stream (Icecast)
|
||||||
|
ice = f.get("icecast_live")
|
||||||
|
if ice is None:
|
||||||
|
checks.append(_check(
|
||||||
|
"stream", "Local stream", ERROR, f"Icecast isn't answering on port {f['icecast_port']}.",
|
||||||
|
"Check the icecast container is running.",
|
||||||
|
))
|
||||||
|
elif ice:
|
||||||
|
checks.append(_check("stream", "Local stream", OK, "Live — press play in the player below."))
|
||||||
|
elif running:
|
||||||
|
checks.append(_check(
|
||||||
|
"stream", "Local stream", WARN, "OP25 is running but nothing is feeding the stream yet.",
|
||||||
|
"It normally appears within ~20 seconds of a start; if not, check the OP25 log below.",
|
||||||
|
))
|
||||||
|
else:
|
||||||
|
checks.append(_check("stream", "Local stream", OFF, "Goes live when OP25 runs."))
|
||||||
|
|
||||||
|
return checks
|
||||||
|
|
||||||
|
|
||||||
|
async def icecast_live() -> Optional[bool]:
|
||||||
|
"""True if the configured mount has a source, False if not, None if Icecast is down."""
|
||||||
|
url = f"http://{settings.icecast_host}:{settings.icecast_port}/status-json.xsl"
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(timeout=3) as client:
|
||||||
|
r = await client.get(url)
|
||||||
|
r.raise_for_status()
|
||||||
|
sources = (r.json().get("icestats") or {}).get("source") or []
|
||||||
|
except Exception as e:
|
||||||
|
logger.debug(f"Icecast status check failed: {e}")
|
||||||
|
return None
|
||||||
|
if isinstance(sources, dict):
|
||||||
|
sources = [sources]
|
||||||
|
return any(str(s.get("listenurl", "")).endswith(settings.icecast_mount) for s in sources)
|
||||||
|
|
||||||
|
|
||||||
|
async def gather_facts() -> Dict[str, Any]:
|
||||||
|
from app.internal import auth, credentials
|
||||||
|
from app.internal.call_recorder import call_recorder
|
||||||
|
from app.internal.config_manager import load_node_config
|
||||||
|
from app.internal.metadata_watcher import metadata_watcher
|
||||||
|
from app.internal.mqtt_manager import mqtt_manager
|
||||||
|
from app.internal.op25_client import op25_client
|
||||||
|
from app.internal.secondary_sdr_client import secondary_sdr_client
|
||||||
|
|
||||||
|
node_cfg = load_node_config()
|
||||||
|
cfg = active_config(node_cfg)
|
||||||
|
op25 = await op25_client.status()
|
||||||
|
|
||||||
|
devs = await secondary_sdr_client.devices()
|
||||||
|
sdr_count: Optional[int] = len(devs) if devs is not None else None
|
||||||
|
if not sdr_count:
|
||||||
|
# op25's own lsusb count — its :stable image may lack lsusb and say 0,
|
||||||
|
# so only trust a positive answer from it.
|
||||||
|
op25_devs = await op25_client.devices()
|
||||||
|
if op25_devs and op25_devs.get("count"):
|
||||||
|
sdr_count = op25_devs["count"]
|
||||||
|
|
||||||
|
activity = call_recorder.audio_activity()
|
||||||
|
return {
|
||||||
|
"now": time.time(),
|
||||||
|
"node_id": settings.node_id,
|
||||||
|
"default_password": auth.is_using_default_password(),
|
||||||
|
"c2_url": settings.c2_url,
|
||||||
|
"has_api_key": bool(credentials.get_api_key()),
|
||||||
|
"mqtt_connected": mqtt_manager.is_connected,
|
||||||
|
"mqtt_broker": settings.mqtt_broker,
|
||||||
|
"mqtt_port": settings.mqtt_port,
|
||||||
|
"sdr_count": sdr_count,
|
||||||
|
"source": None if cfg is None else {
|
||||||
|
"name": cfg.name, "type": cfg.type, "system_id": cfg.system_id,
|
||||||
|
"local": bool(node_cfg.override_system_id and node_cfg.override_config),
|
||||||
|
},
|
||||||
|
"op25_status": (op25 or {}).get("status") if op25 is not None else None,
|
||||||
|
"console": metadata_watcher.console_state,
|
||||||
|
"audio_capturing": call_recorder.is_capturing,
|
||||||
|
"last_voice_epoch": activity.last_voice_epoch,
|
||||||
|
"icecast_live": await icecast_live(),
|
||||||
|
"icecast_port": settings.icecast_port,
|
||||||
|
}
|
||||||
@@ -1,73 +0,0 @@
|
|||||||
"""
|
|
||||||
HTTPS api_key pickup — the node's own half of C2's enrollment flow
|
|
||||||
(Server/drb-c2-core/app/routers/enrollment.py, GET /nodes/{id}/credentials).
|
|
||||||
|
|
||||||
Why this exists (node-26#14): key delivery used to be MQTT-only
|
|
||||||
(nodes/{id}/api_key), and a dynsec broker refuses a node that has no key — or
|
|
||||||
a stale one — so a locked-out node could never receive the key that would let
|
|
||||||
it in. install.sh already did this pickup once at install time; doing it from
|
|
||||||
the running app too means a node approved after install, or whose key was
|
|
||||||
reissued, recovers on its own with no SSH and no container restart.
|
|
||||||
|
|
||||||
Auth is the pickup secret install.sh saved to /configs/pickup_secret at
|
|
||||||
enrollment (the same file, bind-mounted). Only ever sent to settings.c2_url,
|
|
||||||
over whatever scheme the operator configured (https in production).
|
|
||||||
"""
|
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
import httpx
|
|
||||||
|
|
||||||
from app.config import settings
|
|
||||||
from app.internal import credentials
|
|
||||||
from app.internal.logger import logger
|
|
||||||
|
|
||||||
_TIMEOUT = 15
|
|
||||||
|
|
||||||
|
|
||||||
async def fetch_api_key() -> Optional[str]:
|
|
||||||
"""Ask C2 for this node's api_key. Returns it, or None when there isn't one
|
|
||||||
to be had yet (not approved, no pickup secret, C2 unreachable, rejected).
|
|
||||||
Never raises — the caller retries on its own backoff."""
|
|
||||||
if not settings.c2_url:
|
|
||||||
logger.warning("Key pickup skipped: C2_URL is not configured.")
|
|
||||||
return None
|
|
||||||
secret = credentials.get_pickup_secret()
|
|
||||||
if not secret:
|
|
||||||
logger.warning(
|
|
||||||
f"Key pickup skipped: no {settings.config_path}/pickup_secret. This node was not enrolled by "
|
|
||||||
"install.sh on this machine — re-run install.sh, or have an admin Reissue key and write it in."
|
|
||||||
)
|
|
||||||
return None
|
|
||||||
|
|
||||||
url = f"{settings.c2_url.rstrip('/')}/nodes/{settings.node_id}/credentials"
|
|
||||||
try:
|
|
||||||
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
|
||||||
r = await client.get(url, headers={"X-Pickup-Secret": secret})
|
|
||||||
except Exception as e:
|
|
||||||
logger.warning(f"Key pickup: could not reach C2 ({e}).")
|
|
||||||
return None
|
|
||||||
|
|
||||||
if r.status_code == 401:
|
|
||||||
logger.error(
|
|
||||||
"Key pickup: C2 rejected the saved pickup secret (401) — it was rotated by a re-enroll elsewhere. "
|
|
||||||
"Re-run install.sh with --token to recover."
|
|
||||||
)
|
|
||||||
return None
|
|
||||||
if r.status_code == 404:
|
|
||||||
logger.error("Key pickup: C2 does not know this node (404) — it was deleted server-side. Re-run install.sh.")
|
|
||||||
return None
|
|
||||||
if r.status_code != 200:
|
|
||||||
logger.warning(f"Key pickup: unexpected HTTP {r.status_code} from C2.")
|
|
||||||
return None
|
|
||||||
|
|
||||||
try:
|
|
||||||
body = r.json()
|
|
||||||
except ValueError:
|
|
||||||
logger.warning("Key pickup: C2 returned a non-JSON body.")
|
|
||||||
return None
|
|
||||||
key = body.get("api_key") if isinstance(body, dict) else None
|
|
||||||
if not key or not isinstance(key, str):
|
|
||||||
status = body.get("approval_status") if isinstance(body, dict) else None
|
|
||||||
logger.info(f"Key pickup: no api_key yet (approval_status={status!r}) — waiting for an admin to approve.")
|
|
||||||
return None
|
|
||||||
return key
|
|
||||||
@@ -235,6 +235,13 @@ class MetadataWatcher:
|
|||||||
self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge
|
self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge
|
||||||
self._last_ok_poll: float = 0.0
|
self._last_ok_poll: float = 0.0
|
||||||
|
|
||||||
|
# Dashboard-only view of what the OP25 console last said. Read by
|
||||||
|
# /api/health; never consulted by segmentation or attribution.
|
||||||
|
self._terminal_epoch: Optional[float] = None
|
||||||
|
self._terminal_channels: List[Dict[str, Any]] = []
|
||||||
|
self._terminal_trunk: List[Dict[str, Any]] = []
|
||||||
|
self._last_heard: Optional[ConsoleEvent] = None
|
||||||
|
|
||||||
# Rolling console history for close-time attribution.
|
# Rolling console history for close-time attribution.
|
||||||
self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX)
|
self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX)
|
||||||
|
|
||||||
@@ -294,6 +301,7 @@ class MetadataWatcher:
|
|||||||
|
|
||||||
self._last_ok_poll = now
|
self._last_ok_poll = now
|
||||||
self._record_console(update, now)
|
self._record_console(update, now)
|
||||||
|
self._note_terminal(update, now)
|
||||||
|
|
||||||
activity = self._snapshot()
|
activity = self._snapshot()
|
||||||
if activity is None or not activity.capturing:
|
if activity is None or not activity.capturing:
|
||||||
@@ -350,6 +358,18 @@ class MetadataWatcher:
|
|||||||
while self._console and self._console[0].epoch < cutoff:
|
while self._console and self._console[0].epoch < cutoff:
|
||||||
self._console.popleft()
|
self._console.popleft()
|
||||||
|
|
||||||
|
def _note_terminal(self, update: Any, now: float) -> None:
|
||||||
|
"""Keep the latest console snapshot for the dashboard. No side effects
|
||||||
|
on segment state — trunk_update/channel_update are not sent on every
|
||||||
|
poll, so the last non-empty one is kept."""
|
||||||
|
self._terminal_epoch = now
|
||||||
|
if update.channels:
|
||||||
|
self._terminal_channels = update.channels
|
||||||
|
if getattr(update, "trunk", None):
|
||||||
|
self._terminal_trunk = update.trunk
|
||||||
|
if self._console:
|
||||||
|
self._last_heard = self._console[-1]
|
||||||
|
|
||||||
def _is_duplicate_grant(self, event: ConsoleEvent) -> bool:
|
def _is_duplicate_grant(self, event: ConsoleEvent) -> bool:
|
||||||
for index in range(len(self._console) - 1, -1, -1):
|
for index in range(len(self._console) - 1, -1, -1):
|
||||||
if len(self._console) - index > _GRANT_DEDUPE_DEPTH:
|
if len(self._console) - index > _GRANT_DEDUPE_DEPTH:
|
||||||
@@ -898,6 +918,19 @@ class MetadataWatcher:
|
|||||||
def is_active(self) -> bool:
|
def is_active(self) -> bool:
|
||||||
return self._active_call_id is not None
|
return self._active_call_id is not None
|
||||||
|
|
||||||
|
@property
|
||||||
|
def console_state(self) -> Dict[str, Any]:
|
||||||
|
"""Last thing the OP25 console reported, for /api/health only."""
|
||||||
|
heard = self._last_heard
|
||||||
|
return {
|
||||||
|
"last_poll_epoch": self._terminal_epoch,
|
||||||
|
"channels": self._terminal_channels,
|
||||||
|
"trunk": self._terminal_trunk,
|
||||||
|
"last_heard": None if heard is None else {
|
||||||
|
"epoch": heard.epoch, "tgid": heard.tgid, "name": heard.name, "freq": heard.freq,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def unattributed_segments(self) -> int:
|
def unattributed_segments(self) -> int:
|
||||||
"""Orphan-audio segments discarded since start. Surfaced on /api/status."""
|
"""Orphan-audio segments discarded since start. Surfaced on /api/status."""
|
||||||
|
|||||||
@@ -6,30 +6,12 @@ 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, key_pickup
|
from app.internal import credentials
|
||||||
|
|
||||||
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):
|
||||||
@@ -37,11 +19,6 @@ 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
|
||||||
@@ -76,7 +53,6 @@ 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
|
||||||
@@ -94,14 +70,18 @@ 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 —
|
||||||
# connect() starts _recover_credentials() in this case, which
|
# expected, not a bug to route around here: this node hasn't been
|
||||||
# fetches the key over HTTPS (key_pickup.py) and rebuilds this
|
# enrolled/approved yet, and the enrollment flow that would fix
|
||||||
# client with it — node-26#14. paho's reconnect_delay_set() below
|
# that client-side is a later, separate pass (out of scope here;
|
||||||
# bounds the refusal loop meanwhile (2..60s backoff).
|
# see MQTT-PUBLIC-AUTH-PLAN.md). paho's reconnect_delay_set()
|
||||||
|
# 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 while the key is fetched from C2."
|
"connecting without credentials; the broker is expected to "
|
||||||
|
"refuse this until the node is enrolled/approved."
|
||||||
)
|
)
|
||||||
|
|
||||||
if settings.mqtt_tls:
|
if settings.mqtt_tls:
|
||||||
@@ -140,10 +120,6 @@ 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
|
||||||
@@ -168,72 +144,6 @@ 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."""
|
||||||
@@ -251,8 +161,6 @@ 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:
|
||||||
|
|||||||
@@ -11,6 +11,11 @@ from app.internal.logger import logger
|
|||||||
# each id is also a top-level key holding that channel's dict
|
# each id is also a top-level key holding that channel's dict
|
||||||
# (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…).
|
# (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…).
|
||||||
#
|
#
|
||||||
|
# trunk_update — per-system trunking state keyed "0", "1", …; each carries
|
||||||
|
# `last_tsbk` (OP25's time.time() of the last control-channel
|
||||||
|
# message decoded) — the only direct proof OP25 is DECODING,
|
||||||
|
# not merely running. Surfaced on the dashboard only.
|
||||||
|
#
|
||||||
# call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended
|
# call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended
|
||||||
# by tk_p25.log_call() at channel-grant time, each stamped with
|
# by tk_p25.log_call() at channel-grant time, each stamped with
|
||||||
# OP25's own time.time(). get_call_log() DRAINS the deque, so
|
# OP25's own time.time(). get_call_log() DRAINS the deque, so
|
||||||
@@ -18,7 +23,7 @@ from app.internal.logger import logger
|
|||||||
# it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so
|
# it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so
|
||||||
# the consumer must keep up.
|
# the consumer must keep up.
|
||||||
#
|
#
|
||||||
# Everything else (trunk_update, rx_update, terminal_config, …) is ignored.
|
# Everything else (rx_update, terminal_config, …) is ignored.
|
||||||
TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}]
|
TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}]
|
||||||
|
|
||||||
|
|
||||||
@@ -28,6 +33,7 @@ class TerminalUpdate:
|
|||||||
|
|
||||||
channels: List[Dict[str, Any]] = field(default_factory=list)
|
channels: List[Dict[str, Any]] = field(default_factory=list)
|
||||||
call_log: List[Dict[str, Any]] = field(default_factory=list)
|
call_log: List[Dict[str, Any]] = field(default_factory=list)
|
||||||
|
trunk: List[Dict[str, Any]] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
class OP25Client:
|
class OP25Client:
|
||||||
@@ -45,6 +51,23 @@ class OP25Client:
|
|||||||
logger.error(f"OP25 start failed: {e}")
|
logger.error(f"OP25 start failed: {e}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
async def start_with_reason(self) -> Optional[str]:
|
||||||
|
"""Start OP25; None on success, else a human-readable reason."""
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(timeout=10) as client:
|
||||||
|
r = await client.post(f"{self.api_url}/op25/start")
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"OP25 start failed: {e}")
|
||||||
|
return "The op25 container isn't answering."
|
||||||
|
if r.is_success:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
detail = r.json().get("detail")
|
||||||
|
except Exception:
|
||||||
|
detail = None
|
||||||
|
logger.error(f"OP25 start failed: HTTP {r.status_code} {detail or r.text[:200]}")
|
||||||
|
return detail or f"op25 container returned HTTP {r.status_code}."
|
||||||
|
|
||||||
async def stop(self) -> bool:
|
async def stop(self) -> bool:
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=10) as client:
|
async with httpx.AsyncClient(timeout=10) as client:
|
||||||
@@ -65,6 +88,17 @@ class OP25Client:
|
|||||||
logger.error(f"OP25 status failed: {e}")
|
logger.error(f"OP25 status failed: {e}")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
async def log(self, lines: int = 80) -> Optional[List[str]]:
|
||||||
|
"""Tail of multi_rx/liquidsoap output, kept by the op25 container."""
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(timeout=5) as client:
|
||||||
|
r = await client.get(f"{self.api_url}/op25/log", params={"lines": lines})
|
||||||
|
r.raise_for_status()
|
||||||
|
return r.json().get("lines", [])
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"OP25 log fetch failed: {e}")
|
||||||
|
return None
|
||||||
|
|
||||||
async def devices(self) -> Optional[Dict[str, Any]]:
|
async def devices(self) -> Optional[Dict[str, Any]]:
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=5) as client:
|
async with httpx.AsyncClient(timeout=5) as client:
|
||||||
@@ -144,6 +178,11 @@ def parse_terminal_messages(messages: Any) -> TerminalUpdate:
|
|||||||
if isinstance(channel, dict):
|
if isinstance(channel, dict):
|
||||||
update.channels.append(channel)
|
update.channels.append(channel)
|
||||||
|
|
||||||
|
elif json_type == "trunk_update":
|
||||||
|
for key, system in msg.items():
|
||||||
|
if key != "json_type" and isinstance(system, dict):
|
||||||
|
update.trunk.append(system)
|
||||||
|
|
||||||
elif json_type == "call_log":
|
elif json_type == "call_log":
|
||||||
for entry in msg.get("log") or []:
|
for entry in msg.get("log") or []:
|
||||||
if isinstance(entry, dict):
|
if isinstance(entry, dict):
|
||||||
|
|||||||
@@ -1,41 +0,0 @@
|
|||||||
"""
|
|
||||||
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
|
|
||||||
+16
-12
@@ -170,15 +170,17 @@ async def on_command(payload: dict):
|
|||||||
await asyncio.sleep(2)
|
await asyncio.sleep(2)
|
||||||
await op25_client.start()
|
await op25_client.start()
|
||||||
elif action == "node_update":
|
elif action == "node_update":
|
||||||
# node-26#15: the container cannot (and must not) update itself. It
|
# TODO: Full OTA update — register a host-level systemd service (e.g. drb-update.service)
|
||||||
# drops a sentinel in /configs; the host's drb-update.path unit
|
# that stops all DRB containers, runs `docker compose pull`, then `docker compose up -d`.
|
||||||
# (installed by install.sh) sees it and runs drb-update.service, which
|
# The C2 server triggers it by sending this MQTT command; the host service watches for the
|
||||||
# pulls the node's own pinned repo ref + images and recreates the
|
# restart signal (e.g. via a Unix socket, a sentinel file, or a lightweight webhook).
|
||||||
# stack. The sentinel's content is ignored by the host — C2 cannot
|
# Not implemented yet — for now, just restart the container so any pre-pulled image
|
||||||
# choose what gets installed. Offline/locked-out nodes get the same
|
# is picked up (requires a prior `docker compose pull` on the host).
|
||||||
# request over HTTPS via the updater's timer instead.
|
logger.info("Node update requested — restarting container to pick up latest image.")
|
||||||
from app.internal.update_request import request_host_update
|
await mqtt_manager.publish_status("offline")
|
||||||
request_host_update()
|
await asyncio.sleep(1)
|
||||||
|
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}")
|
||||||
|
|
||||||
@@ -187,7 +189,6 @@ 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.")
|
||||||
|
|
||||||
|
|
||||||
@@ -326,9 +327,12 @@ async def lifespan(app: FastAPI):
|
|||||||
if (node_cfg.override_system_id and node_cfg.override_config)
|
if (node_cfg.override_system_id and node_cfg.override_config)
|
||||||
else node_cfg.system_config
|
else node_cfg.system_config
|
||||||
)
|
)
|
||||||
if node_cfg.configured and active_config:
|
# Not gated on `configured` (set only by a C2 assignment): a standalone or
|
||||||
|
# not-yet-approved node listening to a system chosen on the local
|
||||||
|
# dashboard (an override) must come back on it after a reboot too.
|
||||||
|
if active_config:
|
||||||
from app.internal.op25_client import op25_client
|
from app.internal.op25_client import op25_client
|
||||||
logger.info("Node is configured — waiting for OP25 API then generating config.")
|
logger.info("Node has a system to listen to — waiting for OP25 API then generating config.")
|
||||||
for attempt in range(10):
|
for attempt in range(10):
|
||||||
if await _generate_op25_config(active_config):
|
if await _generate_op25_config(active_config):
|
||||||
await op25_client.start()
|
await op25_client.start()
|
||||||
|
|||||||
@@ -1,12 +1,12 @@
|
|||||||
from fastapi import APIRouter, Depends, HTTPException, Body
|
from fastapi import APIRouter, Depends, HTTPException, Body
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
from typing import Dict, List, Optional
|
from typing import Dict, List, Literal, Optional
|
||||||
import asyncio
|
import asyncio
|
||||||
import httpx
|
import httpx
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.models import SystemConfig, SECONDARY_SDR_MODES
|
from app.models import SystemConfig, SECONDARY_SDR_MODES
|
||||||
from app.internal.op25_client import op25_client
|
from app.internal.op25_client import op25_client
|
||||||
from app.internal.config_manager import load_node_config, save_node_config, apply_system_config
|
from app.internal.config_manager import load_node_config, save_node_config
|
||||||
from app.internal.call_recorder import call_recorder
|
from app.internal.call_recorder import call_recorder
|
||||||
from app.internal.discord_radio import radio_bot
|
from app.internal.discord_radio import radio_bot
|
||||||
from app.internal.metadata_watcher import metadata_watcher
|
from app.internal.metadata_watcher import metadata_watcher
|
||||||
@@ -36,8 +36,10 @@ async def get_status():
|
|||||||
if (node_cfg.override_system_id and node_cfg.override_config)
|
if (node_cfg.override_system_id and node_cfg.override_config)
|
||||||
else node_cfg.system_config
|
else node_cfg.system_config
|
||||||
)
|
)
|
||||||
|
system_type = None
|
||||||
if active_config:
|
if active_config:
|
||||||
system_name = active_config.name
|
system_name = active_config.name
|
||||||
|
system_type = active_config.type
|
||||||
if active_tgid:
|
if active_tgid:
|
||||||
# Reverse engineer the TGID to talkgroup name based on active system object
|
# Reverse engineer the TGID to talkgroup name based on active system object
|
||||||
tgs = active_config.config.get("talkgroups", [])
|
tgs = active_config.config.get("talkgroups", [])
|
||||||
@@ -58,6 +60,11 @@ async def get_status():
|
|||||||
"configured": node_cfg.configured,
|
"configured": node_cfg.configured,
|
||||||
"assigned_system_id": node_cfg.assigned_system_id,
|
"assigned_system_id": node_cfg.assigned_system_id,
|
||||||
"system_name": system_name,
|
"system_name": system_name,
|
||||||
|
"system_type": system_type,
|
||||||
|
# True when what OP25 is tuned to was chosen on this dashboard rather
|
||||||
|
# than assigned by C2 (a local override, revertible to C2's system).
|
||||||
|
"local_choice": bool(node_cfg.override_system_id and node_cfg.override_config),
|
||||||
|
"has_c2_system": node_cfg.system_config is not None,
|
||||||
"is_recording": call_recorder.is_recording,
|
"is_recording": call_recorder.is_recording,
|
||||||
# Health of the PulseAudio capture that feeds every recording — the single
|
# Health of the PulseAudio capture that feeds every recording — the single
|
||||||
# most useful signal when recordings come back empty. Segment boundaries
|
# most useful signal when recordings come back empty. Segment boundaries
|
||||||
@@ -77,15 +84,27 @@ async def get_status():
|
|||||||
"icecast_url": (
|
"icecast_url": (
|
||||||
f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}"
|
f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}"
|
||||||
),
|
),
|
||||||
|
# The browser builds its own URL from these + window.location.hostname;
|
||||||
|
# icecast_host is "localhost" from the node's point of view.
|
||||||
|
"icecast_port": settings.icecast_port,
|
||||||
|
"icecast_mount": settings.icecast_mount,
|
||||||
"op25": op25_status,
|
"op25": op25_status,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/health")
|
||||||
|
async def get_health():
|
||||||
|
"""Plain-language setup/status checklist for the dashboard (see internal/health.py)."""
|
||||||
|
from app.internal import health
|
||||||
|
facts = await health.gather_facts()
|
||||||
|
return {"checks": health.build_checks(facts), "last_heard": facts["console"].get("last_heard")}
|
||||||
|
|
||||||
|
|
||||||
@router.post("/op25/start")
|
@router.post("/op25/start")
|
||||||
async def start_op25():
|
async def start_op25():
|
||||||
ok = await op25_client.start()
|
err = await op25_client.start_with_reason()
|
||||||
if not ok:
|
if err:
|
||||||
raise HTTPException(500, "Failed to start OP25")
|
raise HTTPException(502, err)
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
@@ -93,30 +112,135 @@ async def start_op25():
|
|||||||
async def stop_op25():
|
async def stop_op25():
|
||||||
ok = await op25_client.stop()
|
ok = await op25_client.stop()
|
||||||
if not ok:
|
if not ok:
|
||||||
raise HTTPException(500, "Failed to stop OP25")
|
raise HTTPException(502, "The op25 container did not answer the stop request.")
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/op25/log")
|
||||||
|
async def op25_log(lines: int = 80):
|
||||||
|
out = await op25_client.log(lines)
|
||||||
|
if out is None:
|
||||||
|
raise HTTPException(502, "The op25 container isn't answering.")
|
||||||
|
return {"lines": out}
|
||||||
|
|
||||||
|
|
||||||
@router.get("/config")
|
@router.get("/config")
|
||||||
async def get_config():
|
async def get_config():
|
||||||
return load_node_config()
|
return load_node_config()
|
||||||
|
|
||||||
|
|
||||||
|
async def _restart_op25_with(config: SystemConfig) -> None:
|
||||||
|
"""Generate OP25's config for `config` and restart OP25 on it."""
|
||||||
|
from app.main import _generate_op25_config
|
||||||
|
if not await _generate_op25_config(config):
|
||||||
|
raise HTTPException(
|
||||||
|
502, f"The op25 container did not accept the config for {config.name} — is it running?",
|
||||||
|
)
|
||||||
|
await op25_client.stop()
|
||||||
|
await asyncio.sleep(2)
|
||||||
|
err = await op25_client.start_with_reason()
|
||||||
|
if err:
|
||||||
|
raise HTTPException(502, f"Config saved, but OP25 did not start: {err}")
|
||||||
|
|
||||||
|
|
||||||
|
async def _apply_override(config: SystemConfig) -> None:
|
||||||
|
"""Make `config` what this node listens to, chosen locally.
|
||||||
|
|
||||||
|
Stored as an override so a C2-assigned system (if any) survives and can
|
||||||
|
be reverted to, and so C2 sees the node is overridden. A later C2 config
|
||||||
|
push replaces it.
|
||||||
|
"""
|
||||||
|
node_cfg = load_node_config()
|
||||||
|
node_cfg.override_system_id = config.system_id
|
||||||
|
node_cfg.override_config = config
|
||||||
|
save_node_config(node_cfg)
|
||||||
|
await _restart_op25_with(config)
|
||||||
|
if mqtt_manager.is_connected:
|
||||||
|
await mqtt_manager.publish_checkin()
|
||||||
|
|
||||||
|
|
||||||
@router.post("/config/system")
|
@router.post("/config/system")
|
||||||
async def set_system_config(config: SystemConfig):
|
async def set_system_config(config: SystemConfig):
|
||||||
"""
|
"""
|
||||||
Apply a system config locally — called by the web UI or pushed by C2.
|
Assign a system as if C2 had (persists as the node's own system, not an
|
||||||
Writes the OP25 config and persists the node config.
|
override) and restart OP25 on it.
|
||||||
|
|
||||||
|
Used to write the raw `config` blob straight to active.cfg.json, which is
|
||||||
|
not an OP25 config at all — OP25 could not start from it.
|
||||||
"""
|
"""
|
||||||
|
_validate_type(config.type)
|
||||||
node_cfg = load_node_config()
|
node_cfg = load_node_config()
|
||||||
node_cfg.assigned_system_id = config.system_id
|
node_cfg.assigned_system_id = config.system_id
|
||||||
node_cfg.system_config = config
|
node_cfg.system_config = config
|
||||||
node_cfg.configured = True
|
node_cfg.configured = True
|
||||||
save_node_config(node_cfg)
|
save_node_config(node_cfg)
|
||||||
apply_system_config(config)
|
await _restart_op25_with(config)
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_type(system_type: str) -> None:
|
||||||
|
if str(system_type).upper() == "DMR":
|
||||||
|
raise HTTPException(400, "DMR isn't supported by this node's decoder yet — use P25 or analog (NBFM).")
|
||||||
|
if str(system_type).upper() not in ("P25", "NBFM"):
|
||||||
|
raise HTTPException(400, f"Unknown system type {system_type!r} — use P25 or NBFM.")
|
||||||
|
|
||||||
|
|
||||||
|
class Talkgroup(BaseModel):
|
||||||
|
id: int
|
||||||
|
name: str = ""
|
||||||
|
|
||||||
|
|
||||||
|
class ListenBody(BaseModel):
|
||||||
|
"""What to listen to, chosen on the local dashboard."""
|
||||||
|
mode: Literal["system", "p25", "analog"]
|
||||||
|
system_id: Optional[str] = None # mode=system: a cached C2 system
|
||||||
|
name: Optional[str] = None
|
||||||
|
frequencies: List[float] = [] # MHz; p25 = control channels, analog = exactly one
|
||||||
|
talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup
|
||||||
|
|
||||||
|
|
||||||
|
# RTL-SDR (R820T/R828D) tuning range.
|
||||||
|
_MIN_MHZ, _MAX_MHZ = 24.0, 1766.0
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/listen")
|
||||||
|
async def listen(body: ListenBody):
|
||||||
|
"""Point OP25 at a system or frequency and start it — no C2 needed."""
|
||||||
|
if body.mode == "system":
|
||||||
|
from app.internal.system_cacher import get_cached_system
|
||||||
|
cached = get_cached_system(body.system_id or "")
|
||||||
|
if not cached:
|
||||||
|
raise HTTPException(404, "That system isn't in this node's saved list.")
|
||||||
|
config = SystemConfig(**cached)
|
||||||
|
_validate_type(config.type)
|
||||||
|
else:
|
||||||
|
freqs = body.frequencies
|
||||||
|
if not freqs:
|
||||||
|
raise HTTPException(400, "Enter at least one frequency in MHz.")
|
||||||
|
bad = [f for f in freqs if not (_MIN_MHZ <= f <= _MAX_MHZ)]
|
||||||
|
if bad:
|
||||||
|
raise HTTPException(
|
||||||
|
400, f"{bad[0]:g} isn't a frequency an RTL-SDR can tune — enter MHz between "
|
||||||
|
f"{_MIN_MHZ:g} and {_MAX_MHZ:g} (e.g. 851.0125).",
|
||||||
|
)
|
||||||
|
if body.mode == "analog" and len(freqs) != 1:
|
||||||
|
raise HTTPException(400, "Analog listens to exactly one frequency.")
|
||||||
|
is_p25 = body.mode == "p25"
|
||||||
|
first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".")
|
||||||
|
label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz")
|
||||||
|
config = SystemConfig(
|
||||||
|
system_id=f"local-{body.mode}",
|
||||||
|
name=label,
|
||||||
|
type="P25" if is_p25 else "NBFM",
|
||||||
|
config={
|
||||||
|
"control_channels": freqs,
|
||||||
|
"talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await _apply_override(config)
|
||||||
|
return {"ok": True, "name": config.name}
|
||||||
|
|
||||||
|
|
||||||
@router.get("/systems")
|
@router.get("/systems")
|
||||||
async def get_systems():
|
async def get_systems():
|
||||||
from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems
|
from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems
|
||||||
@@ -129,9 +253,6 @@ async def set_override(
|
|||||||
system_id: Optional[str] = Body(None),
|
system_id: Optional[str] = Body(None),
|
||||||
system_config: Optional[dict] = Body(None)
|
system_config: Optional[dict] = Body(None)
|
||||||
):
|
):
|
||||||
node_cfg = load_node_config()
|
|
||||||
config = None
|
|
||||||
|
|
||||||
if system_id:
|
if system_id:
|
||||||
from app.internal.system_cacher import get_cached_system
|
from app.internal.system_cacher import get_cached_system
|
||||||
cached = get_cached_system(system_id)
|
cached = get_cached_system(system_id)
|
||||||
@@ -145,19 +266,8 @@ async def set_override(
|
|||||||
else:
|
else:
|
||||||
raise HTTPException(400, "Must specify system_id or system_config.")
|
raise HTTPException(400, "Must specify system_id or system_config.")
|
||||||
|
|
||||||
node_cfg.override_system_id = config.system_id
|
_validate_type(config.type)
|
||||||
node_cfg.override_config = config
|
await _apply_override(config)
|
||||||
save_node_config(node_cfg)
|
|
||||||
|
|
||||||
from app.main import _generate_op25_config
|
|
||||||
if not await _generate_op25_config(config):
|
|
||||||
raise HTTPException(500, f"Failed to generate OP25 config for override: {config.name}")
|
|
||||||
|
|
||||||
await op25_client.stop()
|
|
||||||
await asyncio.sleep(2)
|
|
||||||
await op25_client.start()
|
|
||||||
|
|
||||||
await mqtt_manager._publish_checkin()
|
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
@@ -172,15 +282,14 @@ async def revert_config():
|
|||||||
save_node_config(node_cfg)
|
save_node_config(node_cfg)
|
||||||
|
|
||||||
if node_cfg.system_config:
|
if node_cfg.system_config:
|
||||||
from app.main import _generate_op25_config
|
await _restart_op25_with(node_cfg.system_config)
|
||||||
if not await _generate_op25_config(node_cfg.system_config):
|
else:
|
||||||
raise HTTPException(500, "Failed to regenerate original OP25 config.")
|
# Nothing from C2 to go back to: stop rather than keep decoding a
|
||||||
|
# system the dashboard now says is cleared.
|
||||||
await op25_client.stop()
|
await op25_client.stop()
|
||||||
await asyncio.sleep(2)
|
|
||||||
await op25_client.start()
|
|
||||||
|
|
||||||
await mqtt_manager._publish_checkin()
|
if mqtt_manager.is_connected:
|
||||||
|
await mqtt_manager.publish_checkin()
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
@@ -124,7 +124,7 @@
|
|||||||
}
|
}
|
||||||
.softkey:last-child { border-right: none; }
|
.softkey:last-child { border-right: none; }
|
||||||
|
|
||||||
/* Overlay UI (Volume, Hold) */
|
/* Overlay UI (Volume, status) */
|
||||||
.overlay {
|
.overlay {
|
||||||
position: absolute;
|
position: absolute;
|
||||||
top: 50%; left: 50%;
|
top: 50%; left: 50%;
|
||||||
@@ -153,7 +153,6 @@
|
|||||||
<ul>
|
<ul>
|
||||||
<li><span>Up Arrow</span> Volume Up</li>
|
<li><span>Up Arrow</span> Volume Up</li>
|
||||||
<li><span>Down Arrow</span> Volume Down</li>
|
<li><span>Down Arrow</span> Volume Down</li>
|
||||||
<li><span>Enter / H</span> Toggle Hold (UI mock)</li>
|
|
||||||
<li><span>P</span> Play/Pause Stream</li>
|
<li><span>P</span> Play/Pause Stream</li>
|
||||||
<li><span>S</span> Select System</li>
|
<li><span>S</span> Select System</li>
|
||||||
<li><span>R</span> Revert Override</li>
|
<li><span>R</span> Revert Override</li>
|
||||||
@@ -171,15 +170,14 @@
|
|||||||
<h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3>
|
<h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3>
|
||||||
<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;">
|
<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;">
|
||||||
<div>
|
<div>
|
||||||
<label>Control Freq (MHz/Hz):</label>
|
<label>Freq (MHz) — P25 control channel or analog:</label>
|
||||||
<input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
|
<input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
|
||||||
</div>
|
</div>
|
||||||
<div>
|
<div>
|
||||||
<label>System Type:</label>
|
<label>System Type:</label>
|
||||||
<select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
|
<select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
|
||||||
<option value="P25">P25</option>
|
<option value="P25">P25</option>
|
||||||
<option value="DMR">DMR</option>
|
<option value="NBFM">Analog FM</option>
|
||||||
<option value="NBFM">NBFM</option>
|
|
||||||
</select>
|
</select>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
@@ -208,11 +206,11 @@
|
|||||||
|
|
||||||
<div class="softkeys">
|
<div class="softkeys">
|
||||||
<div class="softkey">VOL (+/-)</div>
|
<div class="softkey">VOL (+/-)</div>
|
||||||
<div class="softkey">HOLD (H)</div>
|
<div class="softkey">SYSTEM (S)</div>
|
||||||
<div class="softkey">PLAY (P)</div>
|
<div class="softkey">PLAY (P)</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<div class="overlay" id="overlay">HOLD</div>
|
<div class="overlay" id="overlay"></div>
|
||||||
|
|
||||||
<!-- Audio element for local playback -->
|
<!-- Audio element for local playback -->
|
||||||
<audio id="audio-player" autoplay></audio>
|
<audio id="audio-player" autoplay></audio>
|
||||||
@@ -231,7 +229,6 @@
|
|||||||
const elOverlay = document.getElementById('overlay');
|
const elOverlay = document.getElementById('overlay');
|
||||||
|
|
||||||
let currentIcecastUrl = null;
|
let currentIcecastUrl = null;
|
||||||
let isHolding = false;
|
|
||||||
let volume = 1.0;
|
let volume = 1.0;
|
||||||
|
|
||||||
// Clock updater
|
// Clock updater
|
||||||
@@ -309,7 +306,7 @@
|
|||||||
const container = document.getElementById('system-list');
|
const container = document.getElementById('system-list');
|
||||||
container.innerHTML = '';
|
container.innerHTML = '';
|
||||||
if (systems.length === 0) {
|
if (systems.length === 0) {
|
||||||
container.innerHTML = '<div style="padding: 10px; color: #888;">No cached systems found.</div>';
|
container.innerHTML = '<div style="padding: 10px; color: #888;">No server systems saved yet (they arrive once the node is approved). Enter a frequency below.</div>';
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
systems.forEach((sys, index) => {
|
systems.forEach((sys, index) => {
|
||||||
@@ -373,25 +370,20 @@
|
|||||||
}
|
}
|
||||||
closeSystemModal();
|
closeSystemModal();
|
||||||
showOverlay("APPLYING MANUAL", 3000);
|
showOverlay("APPLYING MANUAL", 3000);
|
||||||
const manualConfig = {
|
// Same validated path as the dashboard's "Listen locally" (MHz range check).
|
||||||
system_id: "manual-override",
|
const mode = typeVal === 'NBFM' ? 'analog' : 'p25';
|
||||||
name: `Freq ${freqVal} (${typeVal})`,
|
const freqs = freqVal.split(/[\s,;]+/).filter(Boolean).map(Number);
|
||||||
type: typeVal,
|
|
||||||
config: {
|
|
||||||
control_channels: [freqVal],
|
|
||||||
talkgroups: []
|
|
||||||
}
|
|
||||||
};
|
|
||||||
try {
|
try {
|
||||||
const res = await fetch('/api/config/override', {
|
const res = await fetch('/api/listen', {
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers: { 'Content-Type': 'application/json' },
|
headers: { 'Content-Type': 'application/json' },
|
||||||
body: JSON.stringify({ system_config: manualConfig })
|
body: JSON.stringify({ mode, frequencies: freqs })
|
||||||
});
|
});
|
||||||
if (res.ok) {
|
if (res.ok) {
|
||||||
showOverlay('MANUAL OVERRIDE OK', 1500);
|
showOverlay('MANUAL OK', 1500);
|
||||||
} else {
|
} else {
|
||||||
showOverlay('MANUAL OVERRIDE FAIL', 2000);
|
const d = await res.json().catch(() => ({}));
|
||||||
|
showOverlay((d.detail || 'MANUAL FAILED').slice(0, 60), 4000);
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
showOverlay('ERROR APPLYING', 2000);
|
showOverlay('ERROR APPLYING', 2000);
|
||||||
@@ -444,9 +436,6 @@
|
|||||||
audioPlayer.volume = volume;
|
audioPlayer.volume = volume;
|
||||||
elVol.textContent = `VOL: ${Math.round(volume * 100)}%`;
|
elVol.textContent = `VOL: ${Math.round(volume * 100)}%`;
|
||||||
showOverlay(`VOL: ${Math.round(volume * 100)}%`);
|
showOverlay(`VOL: ${Math.round(volume * 100)}%`);
|
||||||
} else if (e.key === 'Enter' || e.key === 'h' || e.key === 'H') {
|
|
||||||
isHolding = !isHolding;
|
|
||||||
showOverlay(isHolding ? 'HOLD ON' : 'HOLD OFF');
|
|
||||||
} else if (e.key === 'p' || e.key === 'P') {
|
} else if (e.key === 'p' || e.key === 'P') {
|
||||||
if (audioPlayer.paused) {
|
if (audioPlayer.paused) {
|
||||||
audioPlayer.play();
|
audioPlayer.play();
|
||||||
|
|||||||
@@ -1,205 +0,0 @@
|
|||||||
"""
|
|
||||||
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
|
|
||||||
@@ -1,179 +0,0 @@
|
|||||||
"""
|
|
||||||
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)
|
|
||||||
@@ -0,0 +1,248 @@
|
|||||||
|
"""
|
||||||
|
Local dashboard: the health checklist (/api/health) and choosing what to
|
||||||
|
listen to without C2 (/api/listen, /api/config/system, /api/config/revert).
|
||||||
|
|
||||||
|
Same standalone-app pattern as test_auth_endpoints.py — the real routers,
|
||||||
|
none of app.main's lifespan. OP25/MQTT are mocked at the client boundary.
|
||||||
|
"""
|
||||||
|
import base64
|
||||||
|
import time
|
||||||
|
from unittest.mock import AsyncMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from app.config import settings
|
||||||
|
from app.internal import config_manager, credentials, health
|
||||||
|
from app.internal.op25_client import parse_terminal_messages
|
||||||
|
from app.routers import api, ui
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def isolated(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
|
||||||
|
monkeypatch.setattr(credentials, "_api_key", None)
|
||||||
|
monkeypatch.setattr(credentials, "_auth_salt", None)
|
||||||
|
monkeypatch.setattr(credentials, "_session_secret", None)
|
||||||
|
monkeypatch.setattr(config_manager, "_CONFIG_FILE", tmp_path / "node_config.json")
|
||||||
|
monkeypatch.setattr(settings, "dashboard_username", "tester")
|
||||||
|
monkeypatch.setattr(settings, "dashboard_password", "s3cret-pass")
|
||||||
|
yield
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def client():
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(api.router)
|
||||||
|
app.include_router(ui.router)
|
||||||
|
auth = base64.b64encode(b"tester:s3cret-pass").decode()
|
||||||
|
with TestClient(app, headers={"Authorization": f"Basic {auth}"}) as c:
|
||||||
|
yield c
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def op25():
|
||||||
|
"""Mock the OP25 side of an apply: generate, stop, start, checkin."""
|
||||||
|
with patch("app.main._generate_op25_config", AsyncMock(return_value=True)) as gen, \
|
||||||
|
patch("app.routers.api.op25_client.stop", AsyncMock(return_value=True)) as stop, \
|
||||||
|
patch("app.routers.api.op25_client.start_with_reason", AsyncMock(return_value=None)) as start, \
|
||||||
|
patch("app.routers.api.asyncio.sleep", AsyncMock()), \
|
||||||
|
patch("app.routers.api.mqtt_manager.publish_checkin", AsyncMock()) as checkin:
|
||||||
|
yield {"gen": gen, "stop": stop, "start": start, "checkin": checkin}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Health checklist — pure function of facts
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _facts(**over):
|
||||||
|
now = 1_000_000.0
|
||||||
|
f = {
|
||||||
|
"now": now, "node_id": "toughny", "default_password": False,
|
||||||
|
"c2_url": "https://api.example", "has_api_key": True, "mqtt_connected": True,
|
||||||
|
"mqtt_broker": "mqtt.example", "mqtt_port": 8883, "sdr_count": 1,
|
||||||
|
"source": {"name": "County", "type": "P25", "system_id": "s1", "local": False},
|
||||||
|
"op25_status": "running",
|
||||||
|
"console": {"last_poll_epoch": now - 1, "channels": [], "trunk": [{"last_tsbk": now - 1, "rxchan": 851012500}],
|
||||||
|
"last_heard": None},
|
||||||
|
"audio_capturing": True, "last_voice_epoch": now - 30, "icecast_live": True, "icecast_port": 8000,
|
||||||
|
}
|
||||||
|
f.update(over)
|
||||||
|
return f
|
||||||
|
|
||||||
|
|
||||||
|
def _by_id(checks):
|
||||||
|
return {c["id"]: c for c in checks}
|
||||||
|
|
||||||
|
|
||||||
|
def test_healthy_node_is_all_ok():
|
||||||
|
checks = health.build_checks(_facts())
|
||||||
|
assert all(c["state"] == "ok" for c in checks), [c for c in checks if c["state"] != "ok"]
|
||||||
|
assert "851.0125 MHz" in _by_id(checks)["decode"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_unapproved_node_explains_wait_and_mqtt_refusal():
|
||||||
|
"""Toughny's state: C2 set, no API key, broker refuses, nothing chosen."""
|
||||||
|
c = _by_id(health.build_checks(_facts(has_api_key=False, mqtt_connected=False, source=None,
|
||||||
|
op25_status="stopped")))
|
||||||
|
assert c["c2"]["state"] == "warn" and "approve" in c["c2"]["action"]
|
||||||
|
assert "toughny" in c["c2"]["action"]
|
||||||
|
assert c["mqtt"]["state"] == "warn"
|
||||||
|
assert c["source"]["state"] == "warn" and "Listen locally" in c["source"]["action"]
|
||||||
|
assert c["op25"]["state"] == "off"
|
||||||
|
assert c["decode"]["state"] == "off"
|
||||||
|
|
||||||
|
|
||||||
|
def test_key_on_disk_but_mqtt_down_says_restart():
|
||||||
|
c = _by_id(health.build_checks(_facts(mqtt_connected=False)))
|
||||||
|
assert c["mqtt"]["state"] == "error"
|
||||||
|
assert "restart the edge-node container" in c["mqtt"]["action"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_standalone_node_is_not_an_error():
|
||||||
|
c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
|
||||||
|
assert c["c2"]["state"] == "off"
|
||||||
|
assert c["mqtt"]["state"] == "off"
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_sdr_and_crashed_op25():
|
||||||
|
c = _by_id(health.build_checks(_facts(sdr_count=0, op25_status="crashed")))
|
||||||
|
assert c["sdr"]["state"] == "error"
|
||||||
|
assert c["op25"]["state"] == "error" and "log" in c["op25"]["action"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_running_but_no_control_channel():
|
||||||
|
now = 1_000_000.0
|
||||||
|
c = _by_id(health.build_checks(_facts(console={"last_poll_epoch": now, "channels": [],
|
||||||
|
"trunk": [{"last_tsbk": 0}], "last_heard": None})))
|
||||||
|
assert c["decode"]["state"] == "error"
|
||||||
|
|
||||||
|
|
||||||
|
def test_analog_needs_no_control_channel():
|
||||||
|
now = 1_000_000.0
|
||||||
|
c = _by_id(health.build_checks(_facts(
|
||||||
|
source={"name": "Fire", "type": "NBFM", "system_id": "local-analog", "local": True},
|
||||||
|
console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
|
||||||
|
)))
|
||||||
|
assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
|
||||||
|
assert "not recorded" in c["decode"]["action"]
|
||||||
|
assert "this dashboard" in c["source"]["summary"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_op25_unreachable_and_icecast_down():
|
||||||
|
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
|
||||||
|
assert c["op25"]["state"] == "error"
|
||||||
|
assert c["stream"]["state"] == "error"
|
||||||
|
|
||||||
|
|
||||||
|
def test_default_password_warns():
|
||||||
|
assert _by_id(health.build_checks(_facts(default_password=True)))["password"]["state"] == "warn"
|
||||||
|
|
||||||
|
|
||||||
|
def test_trunk_update_is_parsed():
|
||||||
|
update = parse_terminal_messages([
|
||||||
|
{"json_type": "trunk_update", "0": {"last_tsbk": 5.0, "rxchan": 851012500}, "nac": 0},
|
||||||
|
{"json_type": "channel_update", "channels": ["0"], "0": {"freq": 1, "tgid": None}},
|
||||||
|
])
|
||||||
|
assert update.trunk == [{"last_tsbk": 5.0, "rxchan": 851012500}]
|
||||||
|
assert len(update.channels) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_health_route_returns_checks(client):
|
||||||
|
facts = _facts(now=time.time())
|
||||||
|
with patch("app.internal.health.gather_facts", AsyncMock(return_value=facts)):
|
||||||
|
r = client.get("/api/health")
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert [c["id"] for c in r.json()["checks"]][:3] == ["password", "c2", "mqtt"]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Choosing what to listen to
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_listen_p25_saves_override_and_restarts_op25(client, op25):
|
||||||
|
r = client.post("/api/listen", json={
|
||||||
|
"mode": "p25", "frequencies": [851.0125, 851.5125], "talkgroups": [{"id": 1001, "name": "Fire"}],
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = config_manager.load_node_config()
|
||||||
|
assert cfg.override_config.type == "P25"
|
||||||
|
assert cfg.override_config.config["control_channels"] == [851.0125, 851.5125]
|
||||||
|
assert cfg.override_config.config["talkgroups"] == [{"id": 1001, "name": "Fire"}]
|
||||||
|
op25["gen"].assert_awaited_once()
|
||||||
|
op25["start"].assert_awaited_once()
|
||||||
|
op25["checkin"].assert_not_awaited() # MQTT not connected in tests
|
||||||
|
|
||||||
|
|
||||||
|
def test_listen_analog_single_frequency(client, op25):
|
||||||
|
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43], "name": "Fire"})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
cfg = config_manager.load_node_config().override_config
|
||||||
|
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("body,needle", [
|
||||||
|
({"mode": "p25", "frequencies": []}, "at least one"),
|
||||||
|
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
|
||||||
|
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
|
||||||
|
({"mode": "system", "system_id": "nope"}, "saved list"),
|
||||||
|
])
|
||||||
|
def test_listen_rejects_bad_input_with_a_readable_reason(client, op25, body, needle):
|
||||||
|
r = client.post("/api/listen", json=body)
|
||||||
|
assert r.status_code in (400, 404)
|
||||||
|
assert needle in r.json()["detail"]
|
||||||
|
op25["gen"].assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
def test_listen_reports_op25_start_failure(client, op25):
|
||||||
|
op25["start"].return_value = "No OP25 config yet — choose a system or frequency first."
|
||||||
|
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
||||||
|
assert r.status_code == 502
|
||||||
|
assert "did not start" in r.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_dmr_is_refused_up_front(client, op25):
|
||||||
|
r = client.post("/api/config/override", json={"system_config": {
|
||||||
|
"system_id": "x", "name": "x", "type": "DMR", "config": {"control_channels": [450.1]},
|
||||||
|
}})
|
||||||
|
assert r.status_code == 400 and "DMR" in r.json()["detail"]
|
||||||
|
op25["gen"].assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
def test_config_system_goes_through_op25_generation(client, op25, tmp_path, monkeypatch):
|
||||||
|
"""It used to dump the raw blob into active.cfg.json — not an OP25 config."""
|
||||||
|
monkeypatch.setattr(settings, "config_path", str(tmp_path))
|
||||||
|
r = client.post("/api/config/system", json={
|
||||||
|
"system_id": "s1", "name": "County", "type": "P25", "config": {"control_channels": [851.0125]},
|
||||||
|
})
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
op25["gen"].assert_awaited_once()
|
||||||
|
assert not (tmp_path / "active.cfg.json").exists()
|
||||||
|
cfg = config_manager.load_node_config()
|
||||||
|
assert cfg.configured and cfg.system_config.name == "County"
|
||||||
|
|
||||||
|
|
||||||
|
def test_revert_without_c2_system_stops_op25(client, op25):
|
||||||
|
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
||||||
|
op25["stop"].reset_mock()
|
||||||
|
r = client.post("/api/config/revert")
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert config_manager.load_node_config().override_config is None
|
||||||
|
op25["stop"].assert_awaited_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_status_exposes_stream_parts_and_source(client, op25):
|
||||||
|
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
||||||
|
with patch("app.routers.api.op25_client.status", AsyncMock(return_value={"status": "running"})):
|
||||||
|
d = client.get("/api/status").json()
|
||||||
|
assert d["icecast_port"] == settings.icecast_port and d["icecast_mount"] == settings.icecast_mount
|
||||||
|
assert d["local_choice"] is True and d["has_c2_system"] is False
|
||||||
|
assert d["system_type"] == "NBFM"
|
||||||
|
|
||||||
|
|
||||||
|
def test_dashboard_page_has_no_dead_api_instruction(client):
|
||||||
|
client.post("/login", data={"username": "tester", "password": "s3cret-pass"}, follow_redirects=False)
|
||||||
|
html = client.get("/").text
|
||||||
|
assert "POST /api/config/system" not in html
|
||||||
|
assert "/api/listen" in html and "/api/health" in html
|
||||||
+55
-133
@@ -17,13 +17,9 @@
|
|||||||
# 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), then BLOCK until an admin approves
|
# 5. Enroll with C2 (POST /nodes/enroll) and poll for the api_key
|
||||||
# and the api_key is written (node-26#14). Ctrl-C safe: re-run resumes
|
# 6. docker compose pull && up -d (prebuilt; --build opts into the ~1h build)
|
||||||
# from configs/pickup_secret. A re-run also refreshes a stale key.
|
# 7. Print the approval step
|
||||||
# 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
|
||||||
@@ -89,14 +85,7 @@ 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
|
||||||
# How long to block waiting for admin approval. Default: forever (node-26#14 —
|
ENROLL_WAIT="${DRB_ENROLL_WAIT:-0}" # seconds to block waiting for admin approval; 0 = don't block
|
||||||
# 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" "$*"; }
|
||||||
@@ -122,12 +111,7 @@ 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 stop waiting for admin approval after SEC seconds
|
--wait-approval SEC block up to SEC seconds polling for admin approval
|
||||||
(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,
|
||||||
@@ -154,17 +138,12 @@ 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=""
|
||||||
@@ -247,9 +226,7 @@ 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"
|
||||||
# --force: a moved release tag (v1 re-cut) must replace the local copy,
|
git -C "$INSTALL_DIR" fetch --tags --prune origin
|
||||||
# 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
|
||||||
@@ -368,68 +345,39 @@ 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 _hdr; _tmp="$(mktemp)"; _hdr="$(mktemp)"; chmod 600 "$_hdr"
|
local _tmp; _tmp="$(mktemp)"
|
||||||
# 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 "@$_hdr" 2>/dev/null || echo 000)"
|
"$C2_URL/nodes/$NODE_ID/credentials" -H "X-Pickup-Secret: $1" 2>/dev/null || echo 000)"
|
||||||
CRED_BODY="$(cat "$_tmp" 2>/dev/null || true)"
|
CRED_BODY="$(cat "$_tmp" 2>/dev/null || true)"
|
||||||
rm -f "$_tmp" "$_hdr"
|
rm -f "$_tmp"
|
||||||
}
|
}
|
||||||
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
|
||||||
local _tmp="$CREDS.tmp"
|
umask 077; jq -n --arg k "$1" '{api_key:$k}' > "$CREDS"; umask 022
|
||||||
umask 077
|
ok "api_key received and written to configs/credentials.json"
|
||||||
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 approve '$NODE_ID' at ${APP_URL}/settings/nodes. The node keeps fetching"
|
warn "an admin must, at <app-url>/settings/nodes : Approve '$NODE_ID' (assigning a system is separate)."
|
||||||
warn "its key by itself and comes online once approved; re-running this installer also works."
|
warn "then re-run this installer — it reuses configs/pickup_secret — or fetch it directly:"
|
||||||
|
warn " curl -fsS $C2_URL/nodes/$NODE_ID/credentials -H \"X-Pickup-Secret: \$(cat $PICKUP_FILE)\" | jq -r .api_key"
|
||||||
}
|
}
|
||||||
|
|
||||||
# Block until the node is approved and C2 hands back an api_key (node-26#14).
|
# Poll creds_pickup for up to ENROLL_WAIT seconds while still pending. Result
|
||||||
# Result left in CRED_HTTP/CRED_BODY. Returns 0 with a key, 1 on timeout/0-wait,
|
# left in CRED_HTTP/CRED_BODY. --wait-approval is what would have avoided the
|
||||||
# dies on a credential error (401/404) that waiting cannot fix.
|
# original prod bug; the default (0) does not block, so the re-run path below
|
||||||
|
# must stand on its own.
|
||||||
wait_for_key() { # wait_for_key PICKUP_SECRET
|
wait_for_key() { # wait_for_key PICKUP_SECRET
|
||||||
[ "$ENROLL_WAIT" = 0 ] && return 1
|
[ "${ENROLL_WAIT:-0}" -gt 0 ] || return 0
|
||||||
local _end=0 _shown=0 _delay=5
|
local _end; _end=$(( $(date +%s) + ENROLL_WAIT ))
|
||||||
[ "$ENROLL_WAIT" = forever ] || _end=$(( $(date +%s) + ENROLL_WAIT ))
|
say "Waiting up to ${ENROLL_WAIT}s for an admin to approve '$NODE_ID'"
|
||||||
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
|
while [ "$(date +%s)" -lt "$_end" ]; do
|
||||||
while :; do
|
sleep 10
|
||||||
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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -442,15 +390,12 @@ 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 0600 header file — never on a process command line (curl's
|
# Token goes in a header from a shell variable — never on a process command
|
||||||
# argv is visible in `ps`; CISO condition, server-26#61), never echoed.
|
# line, 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 "@$_ehdr" \
|
-H "X-Enrollment-Token: $ENROLLMENT_TOKEN" \
|
||||||
--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
|
||||||
@@ -464,7 +409,6 @@ 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"
|
||||||
@@ -474,32 +418,21 @@ 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>}"
|
||||||
if wait_for_key "$PICKUP"; then
|
KEY="$(cred_field '.api_key')"
|
||||||
write_creds "$(cred_field '.api_key')"
|
if [ -z "$KEY" ]; then
|
||||||
|
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
|
||||||
}
|
}
|
||||||
|
|
||||||
DISK_KEY="$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null || true)"
|
if [ -s "$CREDS" ] && jq -e '.api_key // empty' "$CREDS" >/dev/null 2>&1; then
|
||||||
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 (no pickup secret to re-check it with) — skipping enrollment"
|
ok "api_key already on disk — 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"
|
||||||
@@ -513,11 +446,16 @@ elif [ -s "$PICKUP_FILE" ]; then
|
|||||||
creds_pickup "$PICKUP"
|
creds_pickup "$PICKUP"
|
||||||
case "$CRED_HTTP" in
|
case "$CRED_HTTP" in
|
||||||
200)
|
200)
|
||||||
if wait_for_key "$PICKUP"; then
|
API_KEY="$(cred_field '.api_key')"
|
||||||
write_creds "$(cred_field '.api_key')"
|
if [ -z "$API_KEY" ]; then
|
||||||
|
wait_for_key "$PICKUP" || true
|
||||||
|
API_KEY="$(cred_field '.api_key')"
|
||||||
|
fi
|
||||||
|
if [ -n "$API_KEY" ]; then
|
||||||
|
write_creds "$API_KEY"
|
||||||
else
|
else
|
||||||
# Only with --wait-approval N/0. Enrolled and idempotent; the app
|
# Still pending. Enrolled and idempotent — next run collects the key.
|
||||||
# fetches the key itself once approved. NOT a failure.
|
# Clean exit, fall through to start the stack. NOT a failure.
|
||||||
say_pending "$(cred_field '.approval_status')"
|
say_pending "$(cred_field '.approval_status')"
|
||||||
fi
|
fi
|
||||||
;;
|
;;
|
||||||
@@ -572,30 +510,12 @@ if [ "$DO_START" = 1 ]; then
|
|||||||
fi
|
fi
|
||||||
docker compose up --no-build -d
|
docker compose up --no-build -d
|
||||||
fi
|
fi
|
||||||
# ALWAYS recreate edge-node (node-26#14): `up -d` leaves an already-running
|
ok "stack started"
|
||||||
# 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. Host updater (node-26#15) ───────────────────────────────────────────
|
# ── 7. What the operator does next ──────────────────────────────────────────
|
||||||
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
|
||||||
|
|
||||||
@@ -605,18 +525,20 @@ $(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 — assign this node a radio system at ${APP_URL}/nodes/${NODE_ID}
|
NEXT — an admin must approve this node before it can do anything:
|
||||||
|
|
||||||
|
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 [ -z "$(jq -r '.api_key // empty' "$CREDS" 2>/dev/null)" ]; then
|
if [ ! -s "$CREDS" ]; then
|
||||||
if [ -s "$PICKUP_FILE" ]; then
|
if [ -s "$PICKUP_FILE" ]; then
|
||||||
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"
|
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"
|
||||||
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
|
||||||
|
|||||||
@@ -52,12 +52,16 @@ class IcecastConfig(BaseModel):
|
|||||||
class ConfigGenerator(BaseModel):
|
class ConfigGenerator(BaseModel):
|
||||||
type: DecodeMode
|
type: DecodeMode
|
||||||
systemName: str
|
systemName: str
|
||||||
|
# P25: control channel list (Hz). NBFM: channels[0] is the one analog
|
||||||
|
# frequency to listen to (Hz).
|
||||||
channels: List[Union[int, str]]
|
channels: List[Union[int, str]]
|
||||||
tags: Optional[List[TalkgroupTag]]
|
tags: Optional[List[TalkgroupTag]] = None
|
||||||
whitelist: Optional[List[int]]
|
whitelist: Optional[List[int]] = None
|
||||||
icecastConfig: Optional[IcecastConfig]
|
icecastConfig: Optional[IcecastConfig] = None
|
||||||
hardware_preset: str = "rtl-sdr-v3"
|
hardware_preset: str = "rtl-sdr-v3"
|
||||||
ppm_override: Optional[float] = None # when set, overrides the preset's default PPM
|
ppm_override: Optional[float] = None # when set, overrides the preset's default PPM
|
||||||
|
# NBFM only: power squelch threshold in dB (op25_nbfm default is -60).
|
||||||
|
nbfm_squelch: Optional[float] = None
|
||||||
|
|
||||||
class DemodType(str, Enum):
|
class DemodType(str, Enum):
|
||||||
CQPSK = "cqpsk"
|
CQPSK = "cqpsk"
|
||||||
@@ -69,15 +73,16 @@ class FilterType(str, Enum):
|
|||||||
|
|
||||||
class ChannelConfig(BaseModel):
|
class ChannelConfig(BaseModel):
|
||||||
name: str
|
name: str
|
||||||
trunking_sysname: Optional[str]
|
trunking_sysname: Optional[str] = None
|
||||||
enable_analog: str
|
enable_analog: str
|
||||||
meta_stream_name: str
|
meta_stream_name: str
|
||||||
demod_type: DemodType
|
demod_type: DemodType
|
||||||
filter_type: FilterType
|
filter_type: FilterType
|
||||||
device: Optional[str] = "sdr"
|
device: Optional[str] = "sdr"
|
||||||
cqpsk_tracking: Optional[bool] = None
|
cqpsk_tracking: Optional[bool] = None
|
||||||
frequency: Optional[float] = None
|
frequency: Optional[int] = None
|
||||||
nbfmSquelch: Optional[float] = None
|
# Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
|
||||||
|
nbfm_squelch_threshold: Optional[float] = None
|
||||||
destination: Optional[str] = "udp://127.0.0.1:23456"
|
destination: Optional[str] = "udp://127.0.0.1:23456"
|
||||||
tracking_threshold: Optional[int] = 120
|
tracking_threshold: Optional[int] = 120
|
||||||
tracking_feedback: Optional[float] = 0.75
|
tracking_feedback: Optional[float] = 0.75
|
||||||
@@ -93,6 +98,9 @@ class DeviceConfig(BaseModel):
|
|||||||
gains: Optional[str] = "lna:39"
|
gains: Optional[str] = "lna:39"
|
||||||
gain_mode: Optional[bool] = False
|
gain_mode: Optional[bool] = False
|
||||||
name: Optional[str] = "sdr"
|
name: Optional[str] = "sdr"
|
||||||
|
# Centre frequency (Hz). Unset for trunked P25, where tk_p25 tunes the
|
||||||
|
# device; required for a fixed analog channel (tunable=False).
|
||||||
|
frequency: Optional[int] = None
|
||||||
offset: Optional[int] = 0
|
offset: Optional[int] = 0
|
||||||
ppm: Optional[float] = 0.0
|
ppm: Optional[float] = 0.0
|
||||||
rate: Optional[int] = 1920000
|
rate: Optional[int] = 1920000
|
||||||
|
|||||||
@@ -3,7 +3,11 @@ import subprocess
|
|||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import signal
|
import signal
|
||||||
|
import sys
|
||||||
import json
|
import json
|
||||||
|
import threading
|
||||||
|
from collections import deque
|
||||||
|
from typing import Optional
|
||||||
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
|
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
|
||||||
from internal.logger import create_logger
|
from internal.logger import create_logger
|
||||||
from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
|
from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
|
||||||
@@ -14,6 +18,16 @@ LOGGER = create_logger(__name__)
|
|||||||
OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
|
OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
|
||||||
OP25_SCRIPT = "run_multi-rx_service.sh"
|
OP25_SCRIPT = "run_multi-rx_service.sh"
|
||||||
_PGID_FILE = "/tmp/op25.pgid"
|
_PGID_FILE = "/tmp/op25.pgid"
|
||||||
|
_ACTIVE_CFG = "/configs/active.cfg.json"
|
||||||
|
|
||||||
|
# Analog: park the dongle's centre this far from the channel so the channel
|
||||||
|
# does not sit on the RTL-SDR's DC spike. Well inside the 1.92 MS/s window.
|
||||||
|
ANALOG_CENTER_OFFSET_HZ = 250_000
|
||||||
|
|
||||||
|
# Last lines multi_rx/liquidsoap printed — the only place the reason OP25
|
||||||
|
# died (SDR busy, bad config, no lock) is visible without SSH + docker logs.
|
||||||
|
_log: deque = deque(maxlen=300)
|
||||||
|
_proc: Optional[subprocess.Popen] = None
|
||||||
|
|
||||||
|
|
||||||
def _save_pgid(pgid: int) -> None:
|
def _save_pgid(pgid: int) -> None:
|
||||||
@@ -28,15 +42,60 @@ def _read_pgid():
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _multi_rx_alive(pgid: int) -> bool:
|
||||||
|
"""True only if a live multi_rx.py process is in OP25's process group.
|
||||||
|
|
||||||
|
node-26#12: killpg(pgid, 0) succeeds while the wrapper shell or
|
||||||
|
liquidsoap outlives a dead multi_rx (and even for a zombie leader), so
|
||||||
|
the old check reported a dead scanner as "running".
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
pids = [d for d in os.listdir("/proc") if d.isdigit()]
|
||||||
|
except OSError:
|
||||||
|
return False
|
||||||
|
for pid in pids:
|
||||||
|
try:
|
||||||
|
with open(f"/proc/{pid}/stat") as f:
|
||||||
|
fields = f.read().rsplit(")", 1)[1].split()
|
||||||
|
state, pgrp = fields[0], int(fields[2])
|
||||||
|
if pgrp != pgid or state == "Z":
|
||||||
|
continue
|
||||||
|
with open(f"/proc/{pid}/cmdline", "rb") as f:
|
||||||
|
if b"multi_rx.py" in f.read():
|
||||||
|
return True
|
||||||
|
except (OSError, IndexError, ValueError):
|
||||||
|
continue
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _reap() -> None:
|
||||||
|
if _proc is not None:
|
||||||
|
_proc.poll()
|
||||||
|
|
||||||
|
|
||||||
def _is_running() -> bool:
|
def _is_running() -> bool:
|
||||||
pgid = _read_pgid()
|
pgid = _read_pgid()
|
||||||
if pgid is None:
|
if pgid is None:
|
||||||
return False
|
return False
|
||||||
|
_reap()
|
||||||
|
return _multi_rx_alive(pgid)
|
||||||
|
|
||||||
|
|
||||||
|
def _kill_group(pgid: int) -> None:
|
||||||
try:
|
try:
|
||||||
os.killpg(pgid, 0)
|
os.killpg(pgid, signal.SIGTERM)
|
||||||
return True
|
|
||||||
except OSError:
|
except OSError:
|
||||||
return False
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _pump(proc: subprocess.Popen) -> None:
|
||||||
|
"""Mirror OP25's output to the container log and keep the tail for /log."""
|
||||||
|
for line in proc.stdout:
|
||||||
|
sys.stdout.write(line)
|
||||||
|
sys.stdout.flush()
|
||||||
|
_log.append(line.rstrip("\n"))
|
||||||
|
rc = proc.wait()
|
||||||
|
_log.append(f"[OP25 exited with code {rc}]")
|
||||||
|
|
||||||
|
|
||||||
def create_op25_router():
|
def create_op25_router():
|
||||||
@@ -44,11 +103,28 @@ def create_op25_router():
|
|||||||
|
|
||||||
@router.post("/start")
|
@router.post("/start")
|
||||||
async def start_op25():
|
async def start_op25():
|
||||||
|
global _proc
|
||||||
if _is_running():
|
if _is_running():
|
||||||
return {"status": "OP25 already running"}
|
return {"status": "OP25 already running"}
|
||||||
|
if not os.path.exists(_ACTIVE_CFG):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=409,
|
||||||
|
detail="No OP25 config yet — choose a system or frequency first.",
|
||||||
|
)
|
||||||
|
stale = _read_pgid()
|
||||||
|
if stale is not None:
|
||||||
|
# multi_rx died but liquidsoap/the wrapper may still hold the
|
||||||
|
# Icecast mount — clear the old group before starting a new one.
|
||||||
|
_kill_group(stale)
|
||||||
try:
|
try:
|
||||||
proc = subprocess.Popen(os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH)
|
_log.clear()
|
||||||
|
proc = subprocess.Popen(
|
||||||
|
os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH,
|
||||||
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
|
||||||
|
)
|
||||||
|
_proc = proc
|
||||||
_save_pgid(proc.pid)
|
_save_pgid(proc.pid)
|
||||||
|
threading.Thread(target=_pump, args=(proc,), daemon=True).start()
|
||||||
LOGGER.debug(proc)
|
LOGGER.debug(proc)
|
||||||
return {"status": "OP25 started"}
|
return {"status": "OP25 started"}
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -57,10 +133,10 @@ def create_op25_router():
|
|||||||
@router.post("/stop")
|
@router.post("/stop")
|
||||||
async def stop_op25():
|
async def stop_op25():
|
||||||
pgid = _read_pgid()
|
pgid = _read_pgid()
|
||||||
if pgid is None or not _is_running():
|
if pgid is None:
|
||||||
return {"status": "OP25 is not running"}
|
return {"status": "OP25 is not running"}
|
||||||
try:
|
try:
|
||||||
os.killpg(pgid, signal.SIGTERM)
|
_kill_group(pgid)
|
||||||
os.remove(_PGID_FILE)
|
os.remove(_PGID_FILE)
|
||||||
return {"status": "OP25 stopped"}
|
return {"status": "OP25 stopped"}
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -68,7 +144,18 @@ def create_op25_router():
|
|||||||
|
|
||||||
@router.get("/status")
|
@router.get("/status")
|
||||||
async def get_status():
|
async def get_status():
|
||||||
return {"status": "running" if _is_running() else "stopped"}
|
if _is_running():
|
||||||
|
return {"status": "running"}
|
||||||
|
# Started (pgid file present, never stopped via the API) but multi_rx
|
||||||
|
# is gone: it crashed. The tail of /log says why.
|
||||||
|
if _read_pgid() is not None:
|
||||||
|
return {"status": "crashed", "configured": os.path.exists(_ACTIVE_CFG)}
|
||||||
|
return {"status": "stopped", "configured": os.path.exists(_ACTIVE_CFG)}
|
||||||
|
|
||||||
|
@router.get("/log")
|
||||||
|
async def get_log(lines: int = 80):
|
||||||
|
lines = max(1, min(lines, _log.maxlen))
|
||||||
|
return {"lines": list(_log)[-lines:]}
|
||||||
|
|
||||||
@router.get("/devices")
|
@router.get("/devices")
|
||||||
async def list_sdr_devices():
|
async def list_sdr_devices():
|
||||||
@@ -104,7 +191,7 @@ def create_op25_router():
|
|||||||
meta_stream_name="stream_0"
|
meta_stream_name="stream_0"
|
||||||
)]
|
)]
|
||||||
devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)]
|
devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)]
|
||||||
save_talkgroup_tags(generator.tags)
|
save_talkgroup_tags(generator.tags or [])
|
||||||
save_whitelist(generator.whitelist)
|
save_whitelist(generator.whitelist)
|
||||||
has_talkgroups = bool(generator.whitelist)
|
has_talkgroups = bool(generator.whitelist)
|
||||||
trunking = TrunkingConfig(
|
trunking = TrunkingConfig(
|
||||||
@@ -142,30 +229,48 @@ def create_op25_router():
|
|||||||
"terminal": terminal.dict()
|
"terminal": terminal.dict()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
elif generator.type == DecodeMode.ANALOG:
|
elif generator.type == DecodeMode.ANALOG:
|
||||||
generator = generator.config
|
if not generator.channels:
|
||||||
|
raise HTTPException(status_code=400, detail="Analog needs a frequency.")
|
||||||
|
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
|
||||||
|
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
|
||||||
|
freq = int(float(generator.channels[0]))
|
||||||
|
# One fixed conventional channel: no trunking section, so
|
||||||
|
# multi_rx adds its default tk_p25 with no chans and reports
|
||||||
|
# the channel as `conventional` on the HTTP terminal.
|
||||||
channels = [ChannelConfig(
|
channels = [ChannelConfig(
|
||||||
channelName=generator.systemName,
|
name=generator.systemName,
|
||||||
enableAnalog="on",
|
trunking_sysname="",
|
||||||
demodType="fsk4",
|
enable_analog="on",
|
||||||
frequency=generator.frequency,
|
demod_type="fsk4",
|
||||||
filterType="widepulse",
|
filter_type="rc",
|
||||||
nbfmSquelch=generator.nbfmSquelch
|
frequency=freq,
|
||||||
|
meta_stream_name="",
|
||||||
|
nbfm_squelch_threshold=generator.nbfm_squelch,
|
||||||
)]
|
)]
|
||||||
devices = [DeviceConfig(gain="LNA:32")]
|
devices = [DeviceConfig(
|
||||||
|
gains=preset["gains"], ppm=ppm, tunable=False,
|
||||||
|
frequency=freq + ANALOG_CENTER_OFFSET_HZ,
|
||||||
|
)]
|
||||||
|
if generator.icecastConfig:
|
||||||
|
generate_liquid_script(generator.icecastConfig)
|
||||||
|
|
||||||
config_dict = {
|
config_dict = {
|
||||||
"channels": [channel.dict() for channel in channels],
|
"channels": [channel.dict() for channel in channels],
|
||||||
"devices": [device.dict() for device in devices]
|
"devices": [device.dict() for device in devices],
|
||||||
|
"terminal": TerminalConfig().dict(),
|
||||||
}
|
}
|
||||||
|
|
||||||
else:
|
else:
|
||||||
raise HTTPException(status_code=400, detail="Invalid configuration type. Must be 'p25' or 'nbfm'.")
|
raise HTTPException(status_code=400, detail="Unsupported system type. Must be 'P25' or 'NBFM' (DMR is not implemented).")
|
||||||
|
|
||||||
with open('/configs/active.cfg.json', 'w') as f:
|
with open(_ACTIVE_CFG, 'w') as f:
|
||||||
json.dump(del_none_in_dict(config_dict), f, indent=2)
|
json.dump(del_none_in_dict(config_dict), f, indent=2)
|
||||||
|
|
||||||
return {"message": "Config exported to '/configs/active.cfg.json'"}
|
return {"message": f"Config exported to '{_ACTIVE_CFG}'"}
|
||||||
|
except HTTPException:
|
||||||
|
raise
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise HTTPException(status_code=500, detail=str(e))
|
raise HTTPException(status_code=500, detail=str(e))
|
||||||
|
|
||||||
|
|||||||
@@ -20,3 +20,12 @@ LIQ_PID=$! # Store the PID of the op25.liq process
|
|||||||
# The trap command ensures that SIGTERM is passed to the background jobs.
|
# The trap command ensures that SIGTERM is passed to the background jobs.
|
||||||
trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT
|
trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT
|
||||||
wait $MULTI_RX_PID
|
wait $MULTI_RX_PID
|
||||||
|
RC=$?
|
||||||
|
|
||||||
|
# multi_rx is gone (crashed, or SDR could not be opened). Take liquidsoap down
|
||||||
|
# with it so nothing in this process group outlives the decoder — a leftover
|
||||||
|
# liquidsoap kept /op25/status reporting "running" (node-26#12) and held the
|
||||||
|
# Icecast mount against the next start.
|
||||||
|
echo "multi_rx.py exited (code $RC) — stopping liquidsoap."
|
||||||
|
kill $LIQ_PID 2>/dev/null
|
||||||
|
exit $RC
|
||||||
@@ -1,172 +0,0 @@
|
|||||||
#!/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 "$@"
|
|
||||||
@@ -1,107 +0,0 @@
|
|||||||
#!/usr/bin/env bash
|
|
||||||
#
|
|
||||||
# Install (or refresh) the DRB host updater units (node-26#15). Run as root.
|
|
||||||
#
|
|
||||||
# install-updater.sh --dir /opt/drb/node-26 --repo-url URL --ref v1 \
|
|
||||||
# --registry git.vpn.cusano.net [--build]
|
|
||||||
# install-updater.sh --refresh # re-render units from /etc/drb/updater.conf
|
|
||||||
#
|
|
||||||
# Writes /etc/drb/updater.conf (root:root 0644 — the ONLY source of what the
|
|
||||||
# updater may install), drb-update.{service,timer,path}, and enables the timer
|
|
||||||
# and the path watcher. install.sh calls this; drb-update.sh calls --refresh
|
|
||||||
# after a checkout so unit changes ship with the repo.
|
|
||||||
set -euo pipefail
|
|
||||||
|
|
||||||
CONF=/etc/drb/updater.conf
|
|
||||||
UNIT_DIR="${DRB_UNIT_DIR:-/etc/systemd/system}"
|
|
||||||
conf() { grep -E "^$1=" "$CONF" 2>/dev/null | head -1 | cut -d= -f2- || true; }
|
|
||||||
die() { printf 'install-updater: %s\n' "$*" >&2; exit 1; }
|
|
||||||
|
|
||||||
INSTALL_DIR="" REPO_URL="" REF="" REGISTRY="" BUILD=0 REFRESH=0
|
|
||||||
while [ $# -gt 0 ]; do
|
|
||||||
case "$1" in
|
|
||||||
--dir) INSTALL_DIR="$2"; shift 2 ;;
|
|
||||||
--repo-url) REPO_URL="$2"; shift 2 ;;
|
|
||||||
--ref) REF="$2"; shift 2 ;;
|
|
||||||
--registry) REGISTRY="$2"; shift 2 ;;
|
|
||||||
--build) BUILD=1; shift ;;
|
|
||||||
--refresh) REFRESH=1; shift ;;
|
|
||||||
*) die "unknown option: $1" ;;
|
|
||||||
esac
|
|
||||||
done
|
|
||||||
|
|
||||||
[ "$(id -u)" -eq 0 ] || die "must run as root"
|
|
||||||
if [ "$REFRESH" = 1 ]; then
|
|
||||||
[ -r "$CONF" ] || die "$CONF missing — run install.sh"
|
|
||||||
INSTALL_DIR="$(conf INSTALL_DIR)"; REPO_URL="$(conf REPO_URL)"; REF="$(conf REF)"
|
|
||||||
REGISTRY="$(conf REGISTRY)"; BUILD="$(conf BUILD)"
|
|
||||||
fi
|
|
||||||
|
|
||||||
[[ "$INSTALL_DIR" =~ ^/[A-Za-z0-9._/-]+$ ]] || die "bad --dir '$INSTALL_DIR' (absolute, no spaces)"
|
|
||||||
[[ "$REPO_URL" =~ ^https://[A-Za-z0-9._/:-]+$ ]] || die "--repo-url must be https: '$REPO_URL'"
|
|
||||||
[[ "$REF" =~ ^[A-Za-z0-9._][A-Za-z0-9._/-]*$ ]] || die "bad --ref '$REF'"
|
|
||||||
[[ "$REGISTRY" =~ ^[A-Za-z0-9.-]+(:[0-9]+)?$ ]] || die "bad --registry '$REGISTRY'"
|
|
||||||
[ "$BUILD" = 1 ] || BUILD=0
|
|
||||||
|
|
||||||
if ! command -v systemctl >/dev/null 2>&1 || [ ! -d /run/systemd/system ]; then
|
|
||||||
die "systemd not running — the host updater needs systemd (remote updates disabled on this node)"
|
|
||||||
fi
|
|
||||||
|
|
||||||
install -d -m 755 /etc/drb
|
|
||||||
umask 022
|
|
||||||
cat > "$CONF.tmp" <<EOF
|
|
||||||
# DRB host updater — written by install-updater.sh. The updater installs ONLY
|
|
||||||
# from these values; nothing C2 sends can change them. Edit = root only.
|
|
||||||
INSTALL_DIR=$INSTALL_DIR
|
|
||||||
REPO_URL=$REPO_URL
|
|
||||||
REF=$REF
|
|
||||||
REGISTRY=$REGISTRY
|
|
||||||
BUILD=$BUILD
|
|
||||||
EOF
|
|
||||||
chown root:root "$CONF.tmp"; chmod 644 "$CONF.tmp"; mv -f "$CONF.tmp" "$CONF"
|
|
||||||
|
|
||||||
cat > "$UNIT_DIR/drb-update.service" <<EOF
|
|
||||||
[Unit]
|
|
||||||
Description=DRB node updater (node-26#15)
|
|
||||||
After=network-online.target docker.service
|
|
||||||
Wants=network-online.target
|
|
||||||
Requires=docker.service
|
|
||||||
|
|
||||||
[Service]
|
|
||||||
Type=oneshot
|
|
||||||
ExecStart=/bin/bash $INSTALL_DIR/scripts/drb-update.sh
|
|
||||||
TimeoutStartSec=$([ "$BUILD" = 1 ] && echo 3h || echo 45min)
|
|
||||||
Nice=10
|
|
||||||
EOF
|
|
||||||
|
|
||||||
cat > "$UNIT_DIR/drb-update.timer" <<'EOF'
|
|
||||||
[Unit]
|
|
||||||
Description=DRB node updater — periodic HTTPS check-in with C2
|
|
||||||
|
|
||||||
[Timer]
|
|
||||||
OnBootSec=3min
|
|
||||||
OnUnitActiveSec=15min
|
|
||||||
RandomizedDelaySec=2min
|
|
||||||
Persistent=true
|
|
||||||
|
|
||||||
[Install]
|
|
||||||
WantedBy=timers.target
|
|
||||||
EOF
|
|
||||||
|
|
||||||
cat > "$UNIT_DIR/drb-update.path" <<EOF
|
|
||||||
[Unit]
|
|
||||||
Description=DRB node updater — MQTT node_update sentinel
|
|
||||||
|
|
||||||
[Path]
|
|
||||||
PathExists=$INSTALL_DIR/configs/update-requested
|
|
||||||
Unit=drb-update.service
|
|
||||||
|
|
||||||
[Install]
|
|
||||||
WantedBy=multi-user.target
|
|
||||||
EOF
|
|
||||||
|
|
||||||
systemctl daemon-reload
|
|
||||||
systemctl enable --now drb-update.timer drb-update.path >/dev/null 2>&1 \
|
|
||||||
|| die "could not enable drb-update.timer/path"
|
|
||||||
printf 'install-updater: host updater enabled (timer every ~15 min + MQTT sentinel), tracking %s @ %s\n' \
|
|
||||||
"$REPO_URL" "$REF"
|
|
||||||
Reference in New Issue
Block a user