Logan CusanoandClaude Opus 5.5
f18f7a49f6
POST /nodes/request-update-all (admin) does what Update Node does for each approved node: flags update_requested_at for the HTTPS updater and nudges it over MQTT. Pending and rejected nodes are skipped. Admin-only button on the Nodes page reports which nodes weren't reachable over MQTT. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
425 lines
18 KiB
Python
425 lines
18 KiB
Python
import secrets
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, List, Optional
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
from pydantic import BaseModel
|
|
from app.models import CommandPayload
|
|
from app.internal import firestore as fstore
|
|
from app.internal.mqtt_handler import mqtt_handler
|
|
from app.internal import dynsec
|
|
from app.internal.logger import logger
|
|
from app.internal.auth import (
|
|
require_admin_token,
|
|
require_service_key_or_admin,
|
|
require_service_or_firebase_token,
|
|
resolve_caller_org_id,
|
|
)
|
|
from app.internal import discord_streams
|
|
|
|
router = APIRouter(prefix="/nodes", tags=["nodes"])
|
|
|
|
|
|
@router.get("")
|
|
async def list_nodes(decoded: dict = Depends(require_service_or_firebase_token)):
|
|
org_id = await resolve_caller_org_id(decoded)
|
|
if org_id is None: # service key or platform admin — unrestricted, matches prior behaviour
|
|
return await fstore.collection_list("nodes")
|
|
return await fstore.collection_list("nodes", org_id=org_id)
|
|
|
|
|
|
@router.get("/{node_id}")
|
|
async def get_node(node_id: str, decoded: dict = Depends(require_service_or_firebase_token)):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
org_id = await resolve_caller_org_id(decoded)
|
|
if org_id is not None and node.get("org_id") != org_id:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
return node
|
|
|
|
|
|
@router.post("/{node_id}/approve")
|
|
async def approve_node(node_id: str, _: dict = Depends(require_admin_token)):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
# node-26#14: approve must NEVER rotate a key that already exists. The old
|
|
# behaviour minted a fresh key on every click and overwrote node_keys, so
|
|
# re-clicking Approve on a live node (or approving after a re-run) silently
|
|
# invalidated the key already on the device and locked it out of MQTT with
|
|
# no remote recovery. Rotation is an explicit, separate action
|
|
# (POST /nodes/{id}/reissue-key). An existing key is reused as-is; the
|
|
# dynsec upsert below still runs, so a second approve doubles as a safe
|
|
# "re-sync this node's broker login" without changing the credential.
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
existing_key = (key_doc or {}).get("api_key")
|
|
api_key = existing_key or secrets.token_hex(32)
|
|
|
|
# dynsec FIRST, Firestore second: if the broker rejects/never confirms
|
|
# the new client, we must not tell Firestore (and the admin UI) the
|
|
# node is approved with a key mosquitto doesn't actually recognise —
|
|
# that's exactly the silent-drift the two-sources-of-truth problem
|
|
# warns about. See app/internal/dynsec.py.
|
|
try:
|
|
await dynsec.upsert_node_client(node_id, api_key)
|
|
except dynsec.DynsecError as e:
|
|
logger.error(f"Approve {node_id!r}: dynsec upsert failed, NOT writing Firestore: {e}")
|
|
raise HTTPException(502, f"Could not provision MQTT credentials for node: {e}")
|
|
|
|
if not existing_key:
|
|
await fstore.doc_set("node_keys", node_id, {"node_id": node_id, "api_key": api_key}, merge=False)
|
|
await fstore.doc_update("nodes", node_id, {"approval_status": "approved"})
|
|
# TODO(mqtt-cutover): drop this MQTT push once nodes pull their key via
|
|
# GET /nodes/{id}/credentials (routers/enrollment.py) exclusively — see
|
|
# MQTT-PUBLIC-AUTH-PLAN.md "Rollout order" step 6. Kept for node-26.
|
|
mqtt_handler.publish_node_key(node_id, api_key)
|
|
if existing_key:
|
|
logger.info(f"Approve {node_id!r}: key already exists — reused, not rotated")
|
|
return {"ok": True, "key_rotated": False, "already_had_key": bool(existing_key)}
|
|
|
|
|
|
@router.post("/{node_id}/request-update")
|
|
async def request_node_update(node_id: str, _: dict = Depends(require_admin_token)):
|
|
"""Ask a node to run its host-level updater (node-26#15).
|
|
|
|
Two channels, so a node that is offline or locked out of MQTT still gets
|
|
it: (1) ``update_requested_at`` on the node doc, which the node's host
|
|
updater reads over HTTPS on its timer (POST /nodes/{id}/update-check in
|
|
routers/enrollment.py); (2) the MQTT ``node_update`` command, best-effort,
|
|
which makes an online node's updater run within seconds.
|
|
|
|
Deliberately carries NO payload the node acts on beyond "update now": no
|
|
ref, no image name, no command. The node pulls only the repo/registry and
|
|
ref it was installed with — C2 cannot steer what code a node runs.
|
|
"""
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
requested_at = datetime.now(timezone.utc).isoformat()
|
|
mqtt_sent = await _flag_update(node_id, requested_at)
|
|
return {"ok": True, "update_requested_at": requested_at, "mqtt_sent": mqtt_sent}
|
|
|
|
|
|
async def _flag_update(node_id: str, requested_at: str) -> bool:
|
|
await fstore.doc_update("nodes", node_id, {"update_requested_at": requested_at})
|
|
mqtt_sent = mqtt_handler.send_command(node_id, {"action": "node_update"})
|
|
logger.info(f"Update requested for {node_id!r} at {requested_at} (mqtt_sent={mqtt_sent})")
|
|
return mqtt_sent
|
|
|
|
|
|
@router.post("/request-update-all")
|
|
async def request_update_all_nodes(_: dict = Depends(require_admin_token)):
|
|
"""Fleet-wide "Update Node": the same two-channel request as
|
|
request_node_update, for every approved node. Pending/rejected nodes are
|
|
skipped — a rejected node must not be nudged (server-26#195), and a pending
|
|
one has no key to check in with."""
|
|
requested_at = datetime.now(timezone.utc).isoformat()
|
|
nodes = await fstore.collection_list("nodes")
|
|
requested, mqtt_sent, skipped = [], [], []
|
|
for n in nodes:
|
|
nid = n.get("node_id")
|
|
if not nid:
|
|
continue
|
|
if n.get("approval_status") != "approved":
|
|
skipped.append(nid)
|
|
continue
|
|
requested.append(nid)
|
|
if await _flag_update(nid, requested_at):
|
|
mqtt_sent.append(nid)
|
|
logger.info(f"Fleet update requested at {requested_at}: {len(requested)} nodes, "
|
|
f"{len(mqtt_sent)} via MQTT, {len(skipped)} skipped")
|
|
return {"ok": True, "update_requested_at": requested_at, "requested": requested,
|
|
"mqtt_sent": mqtt_sent, "skipped": skipped}
|
|
|
|
|
|
@router.delete("/{node_id}", status_code=204)
|
|
async def delete_node(node_id: str, _: dict = Depends(require_admin_token)):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
try:
|
|
await dynsec.delete_node_client(node_id)
|
|
except dynsec.DynsecError as e:
|
|
logger.error(f"Delete {node_id!r}: dynsec deleteClient failed, NOT deleting Firestore docs: {e}")
|
|
raise HTTPException(502, f"Could not revoke MQTT credentials for node: {e}")
|
|
await fstore.doc_delete("node_keys", node_id)
|
|
await fstore.doc_delete("nodes", node_id)
|
|
|
|
|
|
@router.post("/{node_id}/reject")
|
|
async def reject_node(node_id: str, _: dict = Depends(require_admin_token)):
|
|
"""Reject = revoke (server-26#195). Before this, reject only flipped
|
|
``approval_status``, so an approved-then-rejected node kept its MQTT login
|
|
(dynsec client), its ``/upload`` key and ``/systems`` read access (both
|
|
keyed off ``node_keys``).
|
|
|
|
Order is fail-closed, the opposite of approve's: the status flips to
|
|
``rejected`` FIRST, so even if the broker call fails, ``/upload`` (which
|
|
now checks approval_status) and ``GET /nodes/{id}/credentials`` already
|
|
refuse the node, and ``dynsec.reconcile_all()`` will no longer re-create
|
|
its login on the next c2-core restart (it also deletes rejected nodes'
|
|
logins there). Then the broker login is deleted, then the key. A dynsec
|
|
failure returns 502 with node_keys left intact, and the admin retries:
|
|
every step is idempotent. Re-approving later mints a fresh key, because
|
|
node_keys is gone; the revoked key is never handed back.
|
|
"""
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
await fstore.doc_update("nodes", node_id, {"approval_status": "rejected"})
|
|
try:
|
|
await dynsec.delete_node_client(node_id)
|
|
except dynsec.DynsecError as e:
|
|
logger.error(f"Reject {node_id!r}: dynsec deleteClient failed, node_keys NOT deleted: {e}")
|
|
raise HTTPException(
|
|
502,
|
|
f"Node marked rejected, but its MQTT login could not be revoked: {e}. Retry reject.",
|
|
)
|
|
await fstore.doc_delete("node_keys", node_id)
|
|
logger.info(f"Reject {node_id!r}: status rejected, MQTT login and API key revoked")
|
|
return {"ok": True, "revoked": True}
|
|
|
|
|
|
@router.post("/{node_id}/command")
|
|
async def send_command(
|
|
node_id: str,
|
|
cmd: CommandPayload,
|
|
_: dict = Depends(require_service_key_or_admin),
|
|
):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
payload = cmd.model_dump(exclude_none=True)
|
|
|
|
# Discord join/leave are desired state now (server-26#203/#204, node-26#25):
|
|
# they never go to the node as commands, and never carry a token string.
|
|
if cmd.action == "discord_join":
|
|
if not cmd.guild_id or not cmd.channel_id:
|
|
raise HTTPException(400, "guild_id and channel_id are required.")
|
|
# The raw-ID path never checked the server belongs to the node's org
|
|
# (server-26#213): one org's node could be streamed into another org's
|
|
# Discord server. The web picker uses /discord/nodes/{id}/streams now.
|
|
if await discord_streams.resolve_guild_org(cmd.guild_id) != node.get("org_id"):
|
|
raise HTTPException(404, "That Discord server is not linked to this node's organization.")
|
|
system_id = discord_streams._effective_system(node)
|
|
if not system_id:
|
|
raise HTTPException(409, "This node is not listening to a system yet.")
|
|
try:
|
|
doc = await discord_streams.start_stream(
|
|
org_id=node.get("org_id"), guild_id=cmd.guild_id, channel_id=cmd.channel_id,
|
|
system_id=system_id, node_id=node_id, preferred_token_id=cmd.preferred_token_id,
|
|
mode="adhoc",
|
|
)
|
|
except discord_streams.StreamError as e:
|
|
raise HTTPException(e.status, e.message)
|
|
return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"),
|
|
"delivered": doc.get("delivered")}
|
|
|
|
if cmd.action == "discord_leave":
|
|
stopped = await discord_streams.stop_node(node_id, "stopped_from_web")
|
|
return {"ok": True, "stopped": stopped}
|
|
|
|
if not mqtt_handler.send_command(node_id, payload):
|
|
raise HTTPException(503, "MQTT broker unavailable — command not delivered.")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/{node_id}/reissue-key")
|
|
async def reissue_node_key(node_id: str, _: dict = Depends(require_admin_token)):
|
|
"""Generate a new API key for the node and push it via MQTT (retained).
|
|
Use this to rotate a key or recover a node whose key was lost."""
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
api_key = secrets.token_hex(32)
|
|
try:
|
|
await dynsec.upsert_node_client(node_id, api_key)
|
|
except dynsec.DynsecError as e:
|
|
logger.error(f"Reissue {node_id!r}: dynsec upsert failed, NOT writing Firestore: {e}")
|
|
raise HTTPException(502, f"Could not update MQTT credentials for node: {e}")
|
|
await fstore.doc_set("node_keys", node_id, {"node_id": node_id, "api_key": api_key}, merge=False)
|
|
# TODO(mqtt-cutover): drop this MQTT push once nodes pull their key via
|
|
# GET /nodes/{id}/credentials (routers/enrollment.py) exclusively — see
|
|
# MQTT-PUBLIC-AUTH-PLAN.md "Rollout order" step 6. Kept for node-26.
|
|
mqtt_handler.publish_node_key(node_id, api_key)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/{node_id}/config/{system_id}")
|
|
async def assign_system(
|
|
node_id: str,
|
|
system_id: str,
|
|
hardware_preset: str = Query("rtl-sdr-v3"),
|
|
ppm_override: Optional[float] = Query(None),
|
|
_: dict = Depends(require_service_key_or_admin),
|
|
):
|
|
"""
|
|
Assign a system to a node. Fetches the system config from Firestore
|
|
and pushes it to the node via MQTT, then marks the node as configured.
|
|
"""
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
system = await fstore.doc_get("systems", system_id)
|
|
if not system:
|
|
raise HTTPException(404, f"System '{system_id}' not found.")
|
|
|
|
# Include hardware preset, node type, and enforce timeout in the push
|
|
push_payload = {
|
|
**system,
|
|
"hardware_preset": hardware_preset,
|
|
"node_type": node.get("node_type", "fixed"),
|
|
"enforce_override_timeout": node.get("enforce_override_timeout", True),
|
|
}
|
|
if ppm_override is not None:
|
|
push_payload["ppm_override"] = ppm_override
|
|
mqtt_handler.push_config(node_id, push_payload)
|
|
|
|
# Update Firestore
|
|
node_updates = {
|
|
"assigned_system_id": system_id,
|
|
"configured": True,
|
|
"hardware_preset": hardware_preset,
|
|
}
|
|
if ppm_override is not None:
|
|
node_updates["ppm_override"] = ppm_override
|
|
await fstore.doc_update("nodes", node_id, node_updates)
|
|
|
|
return {"ok": True}
|
|
|
|
|
|
SECONDARY_SDR_MODES = ("adsb", "ais")
|
|
SDR_PIN_KEYS = ("op25",) + SECONDARY_SDR_MODES
|
|
|
|
|
|
class NodeUpdateBody(BaseModel):
|
|
node_type: Optional[str] = None
|
|
enforce_override_timeout: Optional[bool] = None
|
|
secondary_sdr_mode: Optional[str] = None # legacy: none | adsb | ais
|
|
# Ordered, e.g. ["adsb", "ais"]: SDRs beyond op25's run these top-down.
|
|
secondary_sdr_priority: Optional[List[str]] = None
|
|
# node-26#11: service -> dongle serial; null/"" = automatic. Moving OP25's
|
|
# dongle restarts OP25 on the node; the other pins never do.
|
|
sdr_pins: Optional[Dict[str, Optional[str]]] = None
|
|
|
|
|
|
@router.patch("/{node_id}")
|
|
async def update_node(
|
|
node_id: str,
|
|
body: NodeUpdateBody,
|
|
_: dict = Depends(require_admin_token),
|
|
):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
updates = body.model_dump(exclude_unset=True)
|
|
if not updates:
|
|
return {"ok": True}
|
|
|
|
priority = updates.get("secondary_sdr_priority")
|
|
if priority is not None:
|
|
unknown = [m for m in priority if m not in SECONDARY_SDR_MODES]
|
|
if unknown or len(set(priority)) != len(priority):
|
|
raise HTTPException(400, f"secondary_sdr_priority must be distinct values from {SECONDARY_SDR_MODES}.")
|
|
updates["secondary_sdr_mode"] = priority[0] if priority else "none"
|
|
|
|
if "sdr_pins" in updates:
|
|
raw = updates["sdr_pins"] or {}
|
|
unknown = [k for k in raw if k not in SDR_PIN_KEYS]
|
|
if unknown:
|
|
raise HTTPException(400, f"sdr_pins keys must be from {SDR_PIN_KEYS}.")
|
|
pins = {k: str(v).strip() for k, v in raw.items() if v and str(v).strip()}
|
|
if len(set(pins.values())) != len(pins):
|
|
raise HTTPException(400, "Two services can't be pinned to the same SDR.")
|
|
updates["sdr_pins"] = pins
|
|
|
|
await fstore.doc_update("nodes", node_id, updates)
|
|
|
|
# SDR settings go as their own command: a config re-push restarts OP25, and
|
|
# changing what the spare dongles do must never interrupt P25 recording
|
|
# (only moving OP25's own dongle restarts it, on the node's side). The node
|
|
# applies it, then its checkin reports back what's really running.
|
|
sdr_keys = {"secondary_sdr_priority", "secondary_sdr_mode", "sdr_pins"}
|
|
if sdr_keys & set(updates):
|
|
command = {"action": "set_sdr_config"}
|
|
if priority is not None:
|
|
command["priority"] = priority
|
|
if "sdr_pins" in updates:
|
|
# Explicit nulls so a cleared pin reaches the node as "automatic".
|
|
command["pins"] = {k: updates["sdr_pins"].get(k) for k in SDR_PIN_KEYS}
|
|
mqtt_handler.send_command(node_id, command)
|
|
if set(updates) <= sdr_keys:
|
|
return {"ok": True}
|
|
|
|
# Re-push config to apply new node settings locally
|
|
updated_node = await fstore.doc_get("nodes", node_id)
|
|
assigned_system_id = updated_node.get("assigned_system_id")
|
|
if assigned_system_id:
|
|
system = await fstore.doc_get("systems", assigned_system_id)
|
|
if system:
|
|
push_payload = {
|
|
**system,
|
|
"hardware_preset": updated_node.get("hardware_preset", "rtl-sdr-v3"),
|
|
"node_type": updated_node.get("node_type", "fixed"),
|
|
"enforce_override_timeout": updated_node.get("enforce_override_timeout", True),
|
|
}
|
|
if updated_node.get("ppm_override") is not None:
|
|
push_payload["ppm_override"] = updated_node["ppm_override"]
|
|
if updated_node.get("secondary_sdr_priority") is not None:
|
|
push_payload["secondary_sdr_priority"] = updated_node["secondary_sdr_priority"]
|
|
elif updated_node.get("secondary_sdr_mode") is not None:
|
|
push_payload["secondary_sdr_mode"] = updated_node["secondary_sdr_mode"]
|
|
mqtt_handler.push_config(node_id, push_payload)
|
|
|
|
return {"ok": True}
|
|
|
|
|
|
class AckOverrideBody(BaseModel):
|
|
timeout_minutes: int = 1440
|
|
|
|
|
|
@router.post("/{node_id}/override/ack")
|
|
async def ack_override(
|
|
node_id: str,
|
|
body: AckOverrideBody,
|
|
_: dict = Depends(require_service_key_or_admin),
|
|
):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
new_timeout = datetime.now(timezone.utc) + timedelta(minutes=body.timeout_minutes)
|
|
|
|
await fstore.doc_update("nodes", node_id, {
|
|
"override_timeout_at": new_timeout.isoformat()
|
|
})
|
|
return {"ok": True, "override_timeout_at": new_timeout.isoformat()}
|
|
|
|
|
|
@router.post("/{node_id}/override/reset")
|
|
async def reset_override(
|
|
node_id: str,
|
|
_: dict = Depends(require_service_key_or_admin),
|
|
):
|
|
node = await fstore.doc_get("nodes", node_id)
|
|
if not node:
|
|
raise HTTPException(404, f"Node '{node_id}' not found.")
|
|
|
|
assigned_system_id = node.get("assigned_system_id")
|
|
if assigned_system_id:
|
|
system = await fstore.doc_get("systems", assigned_system_id)
|
|
if system:
|
|
mqtt_handler.push_config(node_id, system)
|
|
|
|
await fstore.doc_update("nodes", node_id, {
|
|
"is_overridden": False,
|
|
"override_system_id": None,
|
|
"override_timeout_at": None,
|
|
})
|
|
return {"ok": True}
|