Logan CusanoandClaude Opus 5.5
268470b6c0
Checkin now carries disk_free_bytes/disk_total_bytes. Stored on the node doc (not overwritten by older nodes that don't send them). Node page shows a Disk row; node cards warn amber under 15% free, red under 5% or 2 GB. A full disk stopped node-002 recording for 17 days while it showed Online. Frontend unverified locally (no Node, #207); CI build typechecks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
413 lines
19 KiB
Python
413 lines
19 KiB
Python
import asyncio
|
|
import json
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional
|
|
import paho.mqtt.client as mqtt
|
|
from app.config import settings
|
|
from app.internal.logger import logger
|
|
from app.internal import firestore as fstore
|
|
from app.internal import talkgroups
|
|
from app.internal.tenancy import FOUNDING_ORG_ID
|
|
|
|
|
|
class MQTTHandler:
|
|
def __init__(self):
|
|
self._client: Optional[mqtt.Client] = None
|
|
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
|
self._connected = False
|
|
|
|
def _build_client(self) -> mqtt.Client:
|
|
client = mqtt.Client(
|
|
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
|
|
client_id="drb-c2-core",
|
|
)
|
|
if settings.mqtt_user:
|
|
client.username_pw_set(settings.mqtt_user, settings.mqtt_pass)
|
|
|
|
client.on_connect = self._on_connect
|
|
client.on_disconnect = self._on_disconnect
|
|
client.on_message = self._on_message
|
|
return client
|
|
|
|
def _on_connect(self, client, userdata, flags, reason_code, properties):
|
|
if reason_code == 0:
|
|
self._connected = True
|
|
client.subscribe("nodes/+/checkin", qos=1)
|
|
client.subscribe("nodes/+/status", qos=1)
|
|
client.subscribe("nodes/+/metadata", qos=1)
|
|
# Edge -> C2 actual state of its Discord radio streams (node-26#25).
|
|
client.subscribe("nodes/+/discord/actual", qos=1)
|
|
# TODO(mqtt-cutover): drop this subscribe once the enrollment/HTTP
|
|
# credentials flow (routers/enrollment.py) is stable in prod and
|
|
# node-26 (the one live node) has been migrated. See
|
|
# MQTT-PUBLIC-AUTH-PLAN.md "Rollout order" step 6.
|
|
client.subscribe("nodes/+/key_request", qos=1)
|
|
logger.info("MQTT connected — subscribed to node topics.")
|
|
else:
|
|
logger.error(f"MQTT connect refused: {reason_code}")
|
|
|
|
def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
|
|
self._connected = False
|
|
logger.warning(f"MQTT disconnected: {reason_code}")
|
|
|
|
def _on_message(self, client, userdata, msg):
|
|
try:
|
|
payload = json.loads(msg.payload.decode())
|
|
except Exception:
|
|
logger.warning(f"Non-JSON MQTT message on {msg.topic}")
|
|
return
|
|
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._dispatch(msg.topic, payload), self._loop
|
|
)
|
|
|
|
async def _dispatch(self, topic: str, payload: dict):
|
|
parts = topic.split("/")
|
|
# nodes/{node_id}/discord/actual — the one four-level node topic.
|
|
if len(parts) == 4 and parts[0] == "nodes" and parts[2:] == ["discord", "actual"]:
|
|
from app.internal import discord_streams
|
|
try:
|
|
await discord_streams.handle_actual(parts[1], payload)
|
|
except Exception as e:
|
|
logger.error(f"MQTT dispatch error [discord/actual] from {parts[1]}: {e}")
|
|
return
|
|
# Expected: nodes/{node_id}/{type}
|
|
if len(parts) != 3 or parts[0] != "nodes":
|
|
return
|
|
|
|
node_id = parts[1]
|
|
msg_type = parts[2]
|
|
|
|
try:
|
|
if msg_type == "checkin":
|
|
await self._handle_checkin(node_id, payload)
|
|
elif msg_type == "status":
|
|
await self._handle_status(node_id, payload)
|
|
elif msg_type == "metadata":
|
|
await self._handle_metadata(node_id, payload)
|
|
elif msg_type == "key_request":
|
|
await self._handle_key_request(node_id)
|
|
except Exception as e:
|
|
logger.error(f"MQTT dispatch error [{msg_type}] from {node_id}: {e}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Checkin — upsert node; flag new unconfigured nodes
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _handle_checkin(self, node_id: str, payload: dict):
|
|
existing = await fstore.doc_get("nodes", node_id)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if not existing:
|
|
# First time we've seen this node — create it as unconfigured, pending approval.
|
|
# This branch only fires for a node_id that has never gone through
|
|
# POST /nodes/enroll (routers/enrollment.py) — a properly-enrolled
|
|
# node already has a Firestore doc, with its real org_id, by the time
|
|
# its first checkin arrives, so `existing` would be truthy and this
|
|
# branch wouldn't run. What's left is the legacy shared-MQTT-password
|
|
# path (node-26 — see the TODO(mqtt-cutover) notes in this file),
|
|
# which has no enrollment token to resolve org_id from at all.
|
|
# Default it to FOUNDING_ORG_ID, same as enrollment.py's own
|
|
# legacy-token fallback.
|
|
doc = {
|
|
"node_id": node_id,
|
|
"org_id": FOUNDING_ORG_ID,
|
|
"name": payload.get("name", node_id),
|
|
"lat": payload.get("lat", 0.0),
|
|
"lon": payload.get("lon", 0.0),
|
|
"status": "unconfigured",
|
|
"configured": False,
|
|
"last_seen": now.isoformat(),
|
|
"assigned_system_id": None,
|
|
"approval_status": "pending",
|
|
"node_type": payload.get("node_type", "fixed"),
|
|
"secondary_sdr_mode": payload.get("secondary_sdr_mode", "none"),
|
|
"secondary_sdr_priority": payload.get("secondary_sdr_priority", []),
|
|
"secondary_sdr_running": payload.get("secondary_sdr_running"),
|
|
"sdr_pins": payload.get("sdr_pins", {}),
|
|
"sdr_devices": payload.get("sdr_devices"),
|
|
"op25_sdr_serial": payload.get("op25_sdr_serial"),
|
|
"sdr_count": payload.get("sdr_count"), # None until reported, never a guessed 1
|
|
"disk_free_bytes": payload.get("disk_free_bytes"),
|
|
"disk_total_bytes": payload.get("disk_total_bytes"),
|
|
"enforce_override_timeout": payload.get("enforce_override_timeout", True),
|
|
"is_overridden": False,
|
|
"override_system_id": None,
|
|
"override_timeout_at": None,
|
|
}
|
|
await fstore.doc_set("nodes", node_id, doc, merge=False)
|
|
logger.info(f"New node registered: {node_id} — pending admin approval.")
|
|
else:
|
|
updates = {
|
|
"last_seen": now.isoformat(),
|
|
"name": payload.get("name", existing.get("name", node_id)),
|
|
"lat": payload.get("lat", existing.get("lat", 0.0)),
|
|
"lon": payload.get("lon", existing.get("lon", 0.0)),
|
|
}
|
|
# Update status on checkin (don't clobber an active recording)
|
|
if existing.get("status") not in ("recording",):
|
|
if existing.get("configured"):
|
|
updates["status"] = "online"
|
|
elif existing.get("approval_status") == "approved":
|
|
# Approved but not yet configured — restore reachable status after reboot
|
|
updates["status"] = "unconfigured"
|
|
|
|
node_type = payload.get("node_type", existing.get("node_type", "fixed"))
|
|
enforce_timeout = payload.get("enforce_override_timeout", existing.get("enforce_override_timeout", True))
|
|
is_overridden = payload.get("is_overridden", False)
|
|
override_system_id = payload.get("override_system_id")
|
|
|
|
updates["node_type"] = node_type
|
|
updates["enforce_override_timeout"] = enforce_timeout
|
|
|
|
# radio_channels (node-26#24): a conventional node's channels and
|
|
# whether each is encrypted — the one decode failure the owner has
|
|
# to hear about from C2, since nothing on the node can fix it.
|
|
# disk_* (node-26#41): free/total bytes of the host disk the node
|
|
# records to — a full disk silently stopped node-002 for 17 days.
|
|
for key in ("secondary_sdr_mode", "secondary_sdr_priority", "secondary_sdr_running",
|
|
"sdr_pins", "sdr_devices", "op25_sdr_serial", "radio_channels",
|
|
"disk_free_bytes", "disk_total_bytes"):
|
|
if key in payload:
|
|
updates[key] = payload[key]
|
|
if "sdr_count" in payload:
|
|
updates["sdr_count"] = payload["sdr_count"]
|
|
|
|
if node_type == "portable":
|
|
updates["is_overridden"] = False
|
|
updates["override_system_id"] = None
|
|
updates["override_timeout_at"] = None
|
|
else:
|
|
updates["is_overridden"] = is_overridden
|
|
updates["override_system_id"] = override_system_id
|
|
|
|
if is_overridden:
|
|
existing_timeout = existing.get("override_timeout_at")
|
|
existing_override_id = existing.get("override_system_id")
|
|
if enforce_timeout:
|
|
if not existing_timeout or existing_override_id != override_system_id:
|
|
updates["override_timeout_at"] = (now + timedelta(hours=24)).isoformat()
|
|
else:
|
|
updates["override_timeout_at"] = None
|
|
else:
|
|
updates["override_timeout_at"] = None
|
|
|
|
await fstore.doc_update("nodes", node_id, updates)
|
|
|
|
# NOTE: discord_connected in checkins is informational only — never release a
|
|
# token from it. Radio-bot tokens are freed only when the node acks (over
|
|
# nodes/{id}/discord/actual) that it logged the bot out, or by the
|
|
# reconciler once the node has been offline past the lease grace — see
|
|
# app/internal/discord_leases.py.
|
|
|
|
# ------------------------------------------------------------------
|
|
# Status update
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _handle_status(self, node_id: str, payload: dict):
|
|
status = payload.get("status")
|
|
if not status:
|
|
return
|
|
try:
|
|
await fstore.doc_update("nodes", node_id, {
|
|
"status": status,
|
|
"last_seen": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
except Exception as e:
|
|
if "No document to update" in str(e):
|
|
logger.info(f"Status from deleted/unknown node {node_id} — ignoring (no Firestore doc)")
|
|
else:
|
|
raise
|
|
|
|
# ------------------------------------------------------------------
|
|
# Metadata — call_start / call_end events
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _handle_metadata(self, node_id: str, payload: dict):
|
|
event = payload.get("event")
|
|
if event == "call_start":
|
|
await self._on_call_start(node_id, payload)
|
|
elif event == "call_end":
|
|
await self._on_call_end(node_id, payload)
|
|
|
|
async def _on_call_start(self, node_id: str, payload: dict):
|
|
call_id = payload.get("call_id")
|
|
if not call_id:
|
|
return
|
|
|
|
# Look up assigned system for this node (cached — assignment rarely changes)
|
|
node = await fstore.doc_get_cached("nodes", node_id)
|
|
system_id = node.get("assigned_system_id") if node else None
|
|
# org_id is inherited from the node, not carried in the MQTT payload —
|
|
# this is the load-bearing tenancy stamp (SAAS_PLAN.md B2b): every call
|
|
# and, downstream, every incident correlated from it, traces back to
|
|
# this. None only for a call from a node that predates tenancy and
|
|
# hasn't been through scripts/backfill_org_id.py yet.
|
|
org_id = node.get("org_id") if node else None
|
|
|
|
started_at_raw = payload.get("started_at")
|
|
started_at = (
|
|
datetime.fromisoformat(started_at_raw)
|
|
if started_at_raw
|
|
else datetime.now(timezone.utc)
|
|
)
|
|
|
|
# Prefer the name from OP25 metadata; fall back to the system config.
|
|
# The lookup lives in internal/talkgroups.py because /upload needs the
|
|
# identical resolution and used to go without it — see server-26#34.
|
|
tgid_name = await talkgroups.resolve(
|
|
system_id, payload.get("tgid"), hint=payload.get("tgid_name") or None
|
|
) or ""
|
|
|
|
doc = {
|
|
"call_id": call_id,
|
|
"node_id": node_id,
|
|
"org_id": org_id,
|
|
"system_id": system_id,
|
|
"talkgroup_id": payload.get("tgid"),
|
|
"talkgroup_name": tgid_name,
|
|
"freq": payload.get("freq"),
|
|
"srcaddr": payload.get("srcaddr"),
|
|
"started_at": started_at,
|
|
"ended_at": None,
|
|
"audio_url": None,
|
|
"transcript": None,
|
|
"incident_id": None,
|
|
"location": None,
|
|
"tags": [],
|
|
"status": "active",
|
|
}
|
|
await fstore.doc_set("calls", call_id, doc, merge=False)
|
|
logger.info(f"Call start: {call_id} (node={node_id}, tgid={payload.get('tgid')})")
|
|
|
|
async def _on_call_end(self, node_id: str, payload: dict):
|
|
call_id = payload.get("call_id")
|
|
if not call_id:
|
|
return
|
|
|
|
ended_at_raw = payload.get("ended_at")
|
|
ended_at = (
|
|
datetime.fromisoformat(ended_at_raw)
|
|
if ended_at_raw
|
|
else datetime.now(timezone.utc)
|
|
)
|
|
|
|
updates = {
|
|
"ended_at": ended_at,
|
|
"status": "ended",
|
|
}
|
|
# doc_set below is a merge, so if call_start already wrote org_id this
|
|
# is a no-op write of the same value. But DEFERRED.md notes call_end
|
|
# can in principle arrive before call_start (ordering relies on MQTT
|
|
# preserving per-topic order, which holds in practice but isn't
|
|
# guaranteed) — in that case doc_set would CREATE the calls doc here
|
|
# with no org_id at all unless it's resolved independently.
|
|
node = await fstore.doc_get_cached("nodes", node_id)
|
|
if node and node.get("org_id"):
|
|
updates["org_id"] = node["org_id"]
|
|
if payload.get("audio_url"):
|
|
updates["audio_url"] = payload["audio_url"]
|
|
|
|
await fstore.doc_set("calls", call_id, updates)
|
|
logger.info(f"Call end: {call_id}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Key request — re-deliver an existing approved key to a node that
|
|
# lost its credentials (e.g. after a directory move / fresh volume)
|
|
# TODO(mqtt-cutover): remove this handler + publish_node_key() below,
|
|
# and the key_request subscribe above, in the separate post-cutover
|
|
# pass called out in MQTT-PUBLIC-AUTH-PLAN.md. Left in place for now so
|
|
# node-26 (currently live, using the shared-password MQTT path) keeps
|
|
# working until the enrollment flow has replaced it in prod.
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _handle_key_request(self, node_id: str):
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
if not key_doc or not key_doc.get("api_key"):
|
|
logger.warning(f"Key request from {node_id} but no key found in Firestore — node may not be approved yet.")
|
|
return
|
|
self.publish_node_key(node_id, key_doc["api_key"])
|
|
logger.info(f"Re-delivered API key to {node_id} on request.")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Outbound — send a command to a specific node
|
|
# ------------------------------------------------------------------
|
|
|
|
def send_command(self, node_id: str, payload: dict) -> bool:
|
|
topic = f"nodes/{node_id}/commands"
|
|
if self._client and self._connected:
|
|
self._client.publish(topic, json.dumps(payload), qos=1)
|
|
logger.info(f"Command sent to {node_id}: {payload.get('action')}")
|
|
return True
|
|
logger.warning(f"MQTT not connected — could not send command to {node_id}")
|
|
return False
|
|
|
|
def push_config(self, node_id: str, system_config: dict):
|
|
topic = f"nodes/{node_id}/config"
|
|
if self._client and self._connected:
|
|
self._client.publish(topic, json.dumps(system_config), qos=1)
|
|
logger.info(f"Config pushed to {node_id}")
|
|
else:
|
|
logger.warning(f"MQTT not connected — could not push config to {node_id}")
|
|
|
|
def publish_discord_desired(self, node_id: str, payload: dict) -> bool:
|
|
"""Retained full list of the Discord streams this node should carry
|
|
(app/internal/discord_streams.py). Carries token IDs, never token
|
|
strings — the broker never stores a bot credential."""
|
|
topic = f"nodes/{node_id}/discord/desired"
|
|
if self._client and self._connected:
|
|
info = self._client.publish(topic, json.dumps(payload), qos=1, retain=True)
|
|
if getattr(info, "rc", 0) == mqtt.MQTT_ERR_SUCCESS:
|
|
logger.info(f"Discord desired state seq {payload.get('seq')} -> {node_id}")
|
|
return True
|
|
logger.warning(f"MQTT not connected — Discord desired state for {node_id} not published")
|
|
return False
|
|
|
|
def publish_node_key(self, node_id: str, api_key: str):
|
|
"""Publish the provisioned API key to the node (retained so it survives reconnects).
|
|
TODO(mqtt-cutover): dead once nodes.py's callers switch to the HTTP
|
|
credentials poll (routers/enrollment.py) exclusively. See note above
|
|
_handle_key_request."""
|
|
topic = f"nodes/{node_id}/api_key"
|
|
if self._client and self._connected:
|
|
self._client.publish(topic, json.dumps({"api_key": api_key}), qos=2, retain=True)
|
|
logger.info(f"API key provisioned to {node_id}")
|
|
else:
|
|
logger.warning(f"MQTT not connected — could not provision key to {node_id}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
async def connect(self):
|
|
self._loop = asyncio.get_event_loop()
|
|
self._client = self._build_client()
|
|
# Start the paho network loop first so it drives reconnects automatically,
|
|
# then keep attempting the initial TCP connect until it succeeds.
|
|
self._client.loop_start()
|
|
asyncio.create_task(self._connect_with_retry())
|
|
|
|
async def _connect_with_retry(self):
|
|
delay = 5
|
|
logger.info(f"MQTT connecting to {settings.mqtt_broker}:{settings.mqtt_port}")
|
|
while True:
|
|
try:
|
|
self._client.connect(settings.mqtt_broker, settings.mqtt_port, keepalive=60)
|
|
return # paho loop_start + reconnect_delay_set handles the rest
|
|
except Exception as e:
|
|
logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s")
|
|
await asyncio.sleep(delay)
|
|
delay = min(delay * 2, 60)
|
|
|
|
async def disconnect(self):
|
|
if self._client:
|
|
self._client.loop_stop()
|
|
self._client.disconnect()
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return self._connected
|
|
|
|
|
|
mqtt_handler = MQTTHandler()
|