Secure the broker for public exposure: TLS and per-node credentials

Edge nodes are deployed to arbitrary locations by arbitrary people, so the
broker has to be reachable from the internet and secured on its own merits
rather than by a VPN.

Three defects made that impossible. The broker only had a plaintext 1883
listener; every node shared one drb-node password; and the ACL pattern used
%c, the client-supplied client id, so any holder of that shared password
could set client_id to another node and take over its namespace. The comment
claiming this cryptographically prevented cross-node access was wrong and is
gone.

Authentication now uses mosquitto 2.x's built-in dynamic-security plugin on
the stock eclipse-mosquitto image. c2-core administers it over the control
topic, creating each node's client on approval with username=<node_id> and
password=<its node_keys api_key>, attached to a role whose ACL is nodes/%u/#
against the authenticated username. One credential, one revocation point.
An HTTP-callback plugin was implemented first and rejected: that project is
archived upstream, which is not an acceptable dependency on an
internet-facing broker.

Because dynsec state is a second source of truth alongside Firestore,
approve/reissue/delete now write to the broker first and surface a 502
rather than drifting, and c2-core reconciles every approved node into dynsec
on startup.

Adds node self-enrollment (POST /nodes/enroll, GET /nodes/{id}/credentials)
so a new node can obtain its key over HTTPS without an operator handling
secrets by hand. Enrolling an already-approved node_id is refused on the
fleet token alone — otherwise a leaked token plus a guessable id would let
an attacker steal a live node's key before the real node asked for it.
Pickup secrets are stored hashed and returned once, and the endpoint is rate
limited per source IP.

Infrastructure: an 8883 TLS listener fed by Caddy's certificate via a
systemd path unit, a firewall rule for it, and Caddy now 404s /internal/*
so the api vhost cannot proxy internal routes.

Also fixes CORS, which allowed https://app.<domain> while the frontend is
served on the bare domain — every call from the portal would have failed —
and widens the vault gitignore to a glob, since ansible-vault leaves
backup siblings that the exact-name rule left committable.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

drb-c2-core/app/config.py

@@ -9,6 +9,15 @@ class Settings(BaseSettings):
     mqtt_user: Optional[str] = None
     mqtt_pass: Optional[str] = None
 
+    # mosquitto's built-in dynamic-security plugin (see app/internal/dynsec.py).
+    # "admin" is hardcoded by the plugin itself on first boot — not actually
+    # configurable — kept as a named setting rather than a literal for
+    # readability. mqtt_dynsec_admin_pass must equal the mosquitto
+    # container's own MOSQUITTO_DYNSEC_PASSWORD env var (root .env /
+    # root.env.j2) or c2-core can't administer node credentials at all.
+    mqtt_dynsec_admin_user: str = "admin"
+    mqtt_dynsec_admin_pass: Optional[str] = None
+
     # GCP
     gcp_credentials_path: Optional[str] = None  # None → uses ADC
     gcs_bucket: Optional[str] = None             # None → audio upload disabled
@@ -51,6 +60,11 @@ class Settings(BaseSettings):
     # Internal service key — allows server-side services (discord bot) to call C2 without Firebase
     service_key: Optional[str] = None
 
+    # Fleet-wide token edge nodes present to POST /nodes/enroll on first boot.
+    # Not a per-node secret — see routers/enrollment.py for why a leaked copy
+    # of this alone can't steal an already-approved node's key.
+    enrollment_token: Optional[str] = None
+
     # Upload size limit — reject audio files larger than this (bytes). Default 100 MB.
     upload_max_bytes: int = 100 * 1024 * 1024
 

drb-c2-core/app/internal/dynsec.py

@@ -0,0 +1,341 @@
+"""
+Client for mosquitto's built-in dynamic-security plugin.
+
+WHY THIS EXISTS: MQTT-PUBLIC-AUTH-PLAN.md originally specced the
+mosquitto-go-auth plugin (HTTP backend). That project was archived by its
+maintainer 2025-08-06 ("no more changes") — unacceptable for a broker that's
+about to be reachable from the public internet, no way to get a CVE fix.
+Replaced with mosquitto 2.x's own `dynamic-security` plugin, which ships in
+and is maintained alongside the official eclipse-mosquitto image itself.
+
+HOW DYNSEC WORKS (verified against plugin source on
+github.com/eclipse-mosquitto/mosquitto, 2026-08-16 — see citations inline;
+NOT verified by running anything, per instruction not to execute/deploy
+anything from this machine):
+
+  - The broker persists clients/roles/ACLs in a JSON file at
+    `plugin_opt_config_file` (we point this at /mosquitto/data/, the same
+    volume `persistence_location` already uses — one durable volume for all
+    broker state, see docker-compose.yml).
+  - Admin commands are plain MQTT publishes: JSON `{"commands": [...]}` to
+    `$CONTROL/dynamic-security/v1` (source: plugin.c,
+    `mosquitto_callback_register(plg_id, MOSQ_EVT_CONTROL,
+    dynsec_control_callback, "$CONTROL/dynamic-security/v1", ...)`).
+    Replies come back on `$CONTROL/dynamic-security/v1/response`
+    (source: control.c, `#define RESPONSE_TOPIC
+    "$CONTROL/dynamic-security/v1/response"`).
+  - Per-command JSON fields (verified against clients.c / roles.c handlers
+    and the plugin README):
+      createClient: username, password, clientid, textname, textdescription,
+                    roles: [{rolename, priority}], groups: [...]
+      modifyClient: same fields, username identifies the existing client
+      deleteClient: username
+      createRole:   rolename, textname, textdescription,
+                    acls: [{acltype, topic, priority, allow}]
+    acltype values: publishClientSend, publishClientReceive,
+    subscribeLiteral, subscribePattern, unsubscribeLiteral,
+    unsubscribePattern. %u (username) and %c (clientid) are valid
+    substitutions in `topic` for every type except the two *Literal ones.
+  - On first boot, if `plugin_opt_config_file` doesn't exist, the plugin
+    bootstraps itself (config_init.c): reads env var
+    `MOSQUITTO_DYNSEC_PASSWORD` (or `plugin_opt_password_init_file`) and
+    creates a client literally named "admin" (hardcoded string, NOT
+    configurable — verified in config_init.c's `client_add_admin()`) with
+    three roles: `super-admin` (full pub/sub on `$CONTROL/#` — i.e. this is
+    what makes a client capable of issuing further dynsec commands, and
+    it's an ordinary role, nothing hardcoded beyond the initial grant),
+    `sys-observe` ($SYS/# read-only), `topic-observe` (# read-only, NOT
+    read-write). If MOSQUITTO_DYNSEC_PASSWORD is set (we always set it),
+    no `democlient` demo account gets created — that only happens in the
+    "no password provided, generate one randomly" path.
+  - This is a genuine backend swap, not just config: `allow_anonymous
+    false` plus the *absence* of `password_file`/`acl_file` directives
+    means dynsec is the only auth backend registered — nothing else is
+    there to conflict with it. (Inferred from plugin architecture — every
+    mosquitto auth backend, built-in or plugin, registers the same
+    basic-auth/ACL callback hooks; there's no "layering" mechanism, so
+    without password_file/acl_file directives there is nothing else to
+    check credentials or topics.)
+
+WHAT COULD NOT BE VERIFIED (see also the plan doc + final report):
+  - The exact JSON envelope of a *response* message (only individual
+    command outcomes were confirmed: `mosquitto_control_command_reply(cmd,
+    NULL)` for success, `mosquitto_control_command_reply(cmd, "error
+    string")` for failure — the wrapping object shape, e.g. whether it's
+    `{"responses": [{"command": ..., "error": ...}]}`, was not directly
+    read from source). This client parses defensively: it treats ANY
+    dict containing a non-null "error"/"Error" key anywhere in the
+    top-level response payload as failure, presence of "already exists" in
+    that string as an idempotent success, and a response with no such key
+    within the timeout as success. A response timeout is always a hard
+    failure (never assumed to mean success).
+  - "Client already exists" was confirmed verbatim as createClient's
+    error string; "already exists" for createRole is assumed analogous,
+    not directly confirmed.
+
+TWO-SOURCES-OF-TRUTH: Firestore's `node_keys` collection is the source of
+truth for node credentials (nothing changes there); dynamic-security.json
+is a derived cache the broker uses to authenticate. `reconcile_all()`
+rebuilds every approved node's dynsec client from Firestore and is called
+on every c2-core startup — so a lost/corrupted dynamic-security.json (e.g.
+volume wiped) self-heals on the next restart instead of silently locking
+out every node. `upsert_node_client()`/`delete_node_client()` are also
+called synchronously from routers/nodes.py's approve/reissue/delete
+handlers and raise on failure — those endpoints now fail loudly (502)
+instead of updating Firestore while dynsec silently didn't get the memo.
+"""
+import asyncio
+import json
+import time
+import uuid
+import paho.mqtt.client as mqtt
+from app.config import settings
+from app.internal.logger import logger
+from app.internal import firestore as fstore
+
+CONTROL_TOPIC = "$CONTROL/dynamic-security/v1"
+RESPONSE_TOPIC = "$CONTROL/dynamic-security/v1/response"
+_RESPONSE_TIMEOUT_SECONDS = 10
+
+# Role every approved node's dynsec client is attached to. %u = the
+# authenticated username (the node_id) — this is the fixed version of the
+# old `pattern readwrite nodes/%c/#`, where %c was the client-supplied,
+# spoofable client ID.
+NODE_ROLE = "node"
+# Role c2-core's own login gets, in addition to being handed the plugin's
+# built-in `super-admin` role (see grant_c2core_admin()). Mirrors the old
+# `topic readwrite #` superuser line.
+C2CORE_ROLE = "c2core"
+
+
+class DynsecError(Exception):
+    """A dynsec command was rejected, or no response arrived in time."""
+
+
+def _run_commands_sync(commands: list[dict], username: str, password: str) -> list[dict]:
+    """
+    Blocking: open a short-lived MQTT connection, publish one or more dynsec
+    commands, wait for the matching replies, disconnect. Always called via
+    asyncio.to_thread — see the async wrappers below. A fresh connection per
+    call (rather than reusing mqtt_handler's long-lived client) keeps this
+    request/response exchange simple and isolated from that client's
+    async-callback-driven subscribe state.
+    """
+    responses: list[dict] = []
+    done = {"got": False, "error": None}
+
+    def _on_connect(client, userdata, flags, reason_code, properties):
+        if reason_code != 0:
+            done["error"] = f"connect refused: {reason_code}"
+            done["got"] = True
+            return
+        client.subscribe(RESPONSE_TOPIC, qos=1)
+        client.publish(CONTROL_TOPIC, json.dumps({"commands": commands}), qos=1)
+
+    def _on_message(client, userdata, msg):
+        try:
+            payload = json.loads(msg.payload.decode())
+        except Exception:
+            return
+        responses.append(payload)
+        done["got"] = True
+
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id=f"drb-c2-core-dynsec-{uuid.uuid4().hex[:8]}",
+    )
+    client.username_pw_set(username, password)
+    client.on_connect = _on_connect
+    client.on_message = _on_message
+
+    try:
+        client.connect(settings.mqtt_broker, settings.mqtt_port, keepalive=30)
+    except Exception as e:
+        raise DynsecError(f"could not connect to mosquitto for dynsec command: {e}")
+
+    client.loop_start()
+    deadline = time.monotonic() + _RESPONSE_TIMEOUT_SECONDS
+    try:
+        while not done["got"] and time.monotonic() < deadline:
+            time.sleep(0.05)
+    finally:
+        client.loop_stop()
+        client.disconnect()
+
+    if done["error"]:
+        raise DynsecError(str(done["error"]))
+    if not responses:
+        raise DynsecError(
+            f"no response on {RESPONSE_TOPIC} within {_RESPONSE_TIMEOUT_SECONDS}s for commands: "
+            f"{[c.get('command') for c in commands]}"
+        )
+    return responses
+
+
+def _check_responses_ok(responses: list[dict], tolerate_already_exists: bool = False) -> None:
+    """Raise DynsecError unless every response payload is error-free (or,
+    when tolerate_already_exists, only contains an 'already exists'-style
+    error — see the module docstring's "could not verify" note on why this
+    is a substring match rather than a structured error code check)."""
+    for payload in responses:
+        # Defensive: walk the payload looking for any *-cased "error" key
+        # with a non-empty value, since the exact envelope shape wasn't
+        # confirmed from source. Covers both a flat {"error": "..."} and a
+        # {"responses": [{"error": "..."}]}-style wrapper.
+        errors = _find_error_strings(payload)
+        for err in errors:
+            if tolerate_already_exists and "already exist" in err.lower():
+                continue
+            raise DynsecError(f"dynsec command failed: {err}")
+
+
+def _find_error_strings(obj) -> list[str]:
+    found = []
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            if k.lower() == "error" and v:
+                found.append(str(v))
+            else:
+                found.extend(_find_error_strings(v))
+    elif isinstance(obj, list):
+        for item in obj:
+            found.extend(_find_error_strings(item))
+    return found
+
+
+# ---------------------------------------------------------------------------
+# Async wrappers (all real work happens in the thread pool)
+# ---------------------------------------------------------------------------
+
+async def _admin_publish(commands: list[dict], tolerate_already_exists: bool = False) -> list[dict]:
+    if not settings.mqtt_dynsec_admin_pass:
+        raise DynsecError("MQTT_DYNSEC_ADMIN_PASS / mqtt_dynsec_admin_pass is not configured")
+    responses = await asyncio.to_thread(
+        _run_commands_sync, commands, settings.mqtt_dynsec_admin_user, settings.mqtt_dynsec_admin_pass
+    )
+    _check_responses_ok(responses, tolerate_already_exists=tolerate_already_exists)
+    return responses
+
+
+async def ensure_roles_and_c2core_grant() -> None:
+    """
+    Idempotent, safe to run on every startup:
+      1. createRole "node"    — nodes/%u/# publish+subscribe (both directions)
+      2. createRole "c2core"  — full "#" publish+subscribe, same reach the
+         old `topic readwrite #` superuser line gave c2-core
+      3. createClient/modifyClient drb-c2-core (settings.mqtt_user) with
+         BOTH roles above AND the plugin's built-in "super-admin" role —
+         i.e. c2-core's existing login is handed the actual dynsec admin
+         role, not a separate identity, per the design decision.
+    Runs over the dedicated "admin" bootstrap login (step 3 assigns
+    super-admin to c2-core's own login for the record / future use, but
+    THIS module still authenticates its own ongoing calls as "admin" — see
+    the module docstring for why: it's the one identity guaranteed by
+    mosquitto's own source to hold super-admin, so control-plane calls
+    don't depend on step 3's grant having actually landed).
+    """
+    node_acl_types = ["publishClientSend", "publishClientReceive", "subscribePattern", "unsubscribePattern"]
+    await _admin_publish([{
+        "command": "createRole",
+        "rolename": NODE_ROLE,
+        "textname": "DRB edge node — own namespace only",
+        "acls": [{"acltype": t, "topic": "nodes/%u/#", "priority": 0, "allow": True} for t in node_acl_types],
+    }], tolerate_already_exists=True)
+
+    c2core_acl_types = ["publishClientSend", "publishClientReceive", "subscribePattern", "unsubscribePattern"]
+    await _admin_publish([{
+        "command": "createRole",
+        "rolename": C2CORE_ROLE,
+        "textname": "DRB c2-core — full broker access",
+        "acls": [{"acltype": t, "topic": "#", "priority": 0, "allow": True} for t in c2core_acl_types],
+    }], tolerate_already_exists=True)
+
+    if not settings.mqtt_user or not settings.mqtt_pass:
+        logger.warning("dynsec: MQTT_USER/MQTT_PASS not configured — skipping c2-core client grant")
+        return
+
+    roles = [{"rolename": C2CORE_ROLE, "priority": 1}, {"rolename": "super-admin", "priority": 2}]
+    try:
+        await _admin_publish([{
+            "command": "createClient",
+            "username": settings.mqtt_user,
+            "password": settings.mqtt_pass,
+            "roles": roles,
+        }])
+        logger.info(f"dynsec: created client {settings.mqtt_user!r} with roles {C2CORE_ROLE}, super-admin")
+    except DynsecError as e:
+        if "already exist" in str(e).lower():
+            await _admin_publish([{
+                "command": "modifyClient",
+                "username": settings.mqtt_user,
+                "password": settings.mqtt_pass,
+                "roles": roles,
+            }])
+            logger.info(f"dynsec: updated existing client {settings.mqtt_user!r} with roles {C2CORE_ROLE}, super-admin")
+        else:
+            raise
+
+
+async def upsert_node_client(node_id: str, api_key: str) -> None:
+    """Create or update a node's dynsec client — called from
+    routers/nodes.py approve_node()/reissue_node_key(), and from
+    reconcile_all() on startup. Raises DynsecError on failure; callers
+    must not write Firestore as if this succeeded when it didn't."""
+    try:
+        await _admin_publish([{
+            "command": "createClient",
+            "username": node_id,
+            "password": api_key,
+            "roles": [{"rolename": NODE_ROLE, "priority": 1}],
+        }])
+    except DynsecError as e:
+        if "already exist" not in str(e).lower():
+            raise
+        await _admin_publish([{
+            "command": "modifyClient",
+            "username": node_id,
+            "password": api_key,
+            "roles": [{"rolename": NODE_ROLE, "priority": 1}],
+        }])
+
+
+async def delete_node_client(node_id: str) -> None:
+    """Best-effort: a node that was never enrolled in dynsec (or already
+    removed) is treated as already-deleted, not an error."""
+    try:
+        await _admin_publish([{"command": "deleteClient", "username": node_id}])
+    except DynsecError as e:
+        if "not found" not in str(e).lower():
+            raise
+
+
+async def reconcile_all() -> None:
+    """
+    Rebuild dynsec state for every approved node from Firestore
+    (node_keys is the source of truth). Called once at c2-core startup,
+    after ensure_roles_and_c2core_grant(). Self-heals a lost/corrupted
+    dynamic-security.json (e.g. volume wiped, or a prior approve/reissue's
+    dynsec publish silently failed to persist for some other reason) —
+    without this, a broker restart with an intact Firestore but an empty
+    dynsec store would lock out every previously-approved node until
+    someone noticed and manually re-approved each one.
+    """
+    nodes = await fstore.collection_list("nodes", approval_status="approved")
+    if not nodes:
+        return
+    ok, failed = 0, 0
+    for node in nodes:
+        node_id = node.get("node_id")
+        if not node_id:
+            continue
+        key_doc = await fstore.doc_get("node_keys", node_id)
+        if not key_doc or not key_doc.get("api_key"):
+            logger.warning(f"dynsec reconcile: node {node_id!r} is approved but has no node_keys entry — skipping")
+            continue
+        try:
+            await upsert_node_client(node_id, key_doc["api_key"])
+            ok += 1
+        except DynsecError as e:
+            failed += 1
+            logger.error(f"dynsec reconcile: failed to sync node {node_id!r}: {e}")
+    logger.info(f"dynsec reconcile: {ok} node(s) synced, {failed} failed")

drb-c2-core/app/internal/mqtt_handler.py

@@ -33,6 +33,10 @@ class MQTTHandler:
             client.subscribe("nodes/+/checkin",     qos=1)
             client.subscribe("nodes/+/status",      qos=1)
             client.subscribe("nodes/+/metadata",    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:
@@ -117,17 +121,14 @@ class MQTTHandler:
                     # Approved but not yet configured — restore reachable status after reboot
                     updates["status"] = "unconfigured"
             
-            node_type = payload.get("node_type") or existing.get("node_type") or "fixed"
-            enforce_timeout = payload.get("enforce_override_timeout")
-            if enforce_timeout is None:
-                enforce_timeout = existing.get("enforce_override_timeout", True)
-            
-            updates["node_type"] = node_type
-            updates["enforce_override_timeout"] = enforce_timeout
-            
+            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
+
             if node_type == "portable":
                 updates["is_overridden"] = False
                 updates["override_system_id"] = None
@@ -257,6 +258,11 @@ class MQTTHandler:
     # ------------------------------------------------------------------
     # 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):
@@ -289,7 +295,10 @@ class MQTTHandler:
             logger.warning(f"MQTT not connected — could not push config to {node_id}")
 
     def publish_node_key(self, node_id: str, api_key: str):
-        """Publish the provisioned API key to the node (retained so it survives reconnects)."""
+        """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)

drb-c2-core/app/main.py

@@ -11,6 +11,8 @@ from app.internal.recorrelation_sweep import recorrelation_loop
 from app.config import settings
 from app.internal.auth import require_firebase_token, require_service_or_firebase_token
 from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
+from app.routers import enrollment
+from app.internal import dynsec
 from app.internal import firestore as fstore
 
 
@@ -36,6 +38,22 @@ async def lifespan(app: FastAPI):
     logger.info("DRB C2 Core starting.")
     await _release_orphaned_tokens()
 
+    # dynsec bootstrap + reconcile — must happen before mqtt_handler.connect()
+    # so that by the time the app is serving requests, c2-core's own dynsec
+    # client/roles exist and every already-approved node's dynsec client
+    # matches Firestore (see app/internal/dynsec.py "TWO-SOURCES-OF-TRUTH").
+    # Non-fatal by design: if the broker or MQTT_DYNSEC_ADMIN_PASS isn't
+    # reachable/configured yet (e.g. first-ever deploy, mosquitto still
+    # starting), log loudly and keep booting rather than crash-looping
+    # c2-core itself — mqtt_handler.connect() below has its own retry loop
+    # and node approval/reissue endpoints fail loudly on their own if dynsec
+    # calls fail later, so nothing here is silently swallowed forever.
+    try:
+        await dynsec.ensure_roles_and_c2core_grant()
+        await dynsec.reconcile_all()
+    except dynsec.DynsecError as e:
+        logger.error(f"dynsec bootstrap/reconcile failed — node approval/reissue will fail until this is resolved: {e}")
+
     await mqtt_handler.connect()
     sweeper_task      = asyncio.create_task(sweeper_loop())
     summarizer_task   = asyncio.create_task(summarizer_loop())
@@ -74,6 +92,16 @@ app.include_router(upload.router)    # auth is per-node, handled inline
 app.include_router(admin.router)     # auth is per-endpoint (read: firebase, write: admin)
 app.include_router(users.router)     # auth: admin only
 app.include_router(links.router)     # auth is per-endpoint (generate: firebase, resolve: service key)
+app.include_router(enrollment.router)  # public; auth is the enrollment/pickup-secret tokens, checked inline
+# NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP
+# backend for the mosquitto-go-auth plugin). That plugin's upstream project
+# is archived (no CVE patches) and was rejected for an internet-facing
+# broker — see MQTT-PUBLIC-AUTH-PLAN.md. MQTT auth is now mosquitto's own
+# built-in dynamic-security plugin (app/internal/dynsec.py talks to it over
+# MQTT control topics, not HTTP), so there is nothing at /internal/mqtt/*
+# anymore. Caddy's Caddyfile.j2 still 404s /internal/* on api.<domain> as
+# defence in depth even though nothing calls it today — cheap insurance
+# against a future /internal/* route being added and forgotten there.
 
 
 @app.get("/health")

drb-c2-core/app/routers/enrollment.py

@@ -0,0 +1,184 @@
+"""
+Node self-enrollment — the public replacement for the old shared-MQTT-
+password flow (see MQTT-PUBLIC-AUTH-PLAN.md "Enrollment flow").
+
+  1. POST /nodes/enroll          (X-Enrollment-Token: <fleet-wide token>)
+     First-boot node upserts itself as `approval_status: pending` and gets
+     back a one-time pickup_secret. Only its hash is persisted.
+  2. GET  /nodes/{id}/credentials (X-Pickup-Secret: <secret from step 1>)
+     Node polls this with backoff until an admin approves it in the
+     frontend (existing nodes.py approve_node() flow — unchanged, still
+     writes node_keys/{id}.api_key) and then reads its api_key back.
+
+These two endpoints are meant to be public (unlike the dynsec control-plane
+traffic in app/internal/dynsec.py, which never leaves the docker-internal
+MQTT bridge) — that's the whole point of moving off WireGuard-per-node.
+Auth is the token headers checked inline below, not the app-wide
+Firebase/service-key dependency the rest of routers/nodes.py uses.
+"""
+import hashlib
+import secrets
+import time
+from typing import Optional
+from fastapi import APIRouter, HTTPException, Header, Request
+from pydantic import BaseModel
+from app.config import settings
+from app.internal import firestore as fstore
+from app.internal.logger import logger
+
+router = APIRouter(prefix="/nodes", tags=["enrollment"])
+
+
+# ---------------------------------------------------------------------------
+# Per-source-IP token bucket for /nodes/enroll.
+#
+# c2-core has no rate-limiting dependency anywhere today (see
+# MQTT-PUBLIC-AUTH-PLAN.md); this is a deliberately small (~30 line)
+# in-memory limiter rather than a new library. Known limitations:
+#   - per-process: with more than one c2-core instance, each has its own
+#     bucket, so real throughput is (limit x instance count). Fine today —
+#     there is exactly one instance.
+#   - resets on every restart/redeploy — not persisted anywhere.
+# Good enough to blunt casual guessing of node_ids against the fleet token;
+# not a substitute for a real edge/WAF rate limiter if this endpoint is
+# ever seriously targeted.
+# ---------------------------------------------------------------------------
+class _TokenBucket:
+    def __init__(self, capacity: int, refill_per_sec: float):
+        self.capacity = capacity
+        self.refill_per_sec = refill_per_sec
+        self._buckets: dict[str, tuple[float, float]] = {}  # key -> (tokens, last_refill_ts)
+
+    def allow(self, key: str) -> bool:
+        now = time.monotonic()
+        tokens, last_ts = self._buckets.get(key, (float(self.capacity), now))
+        tokens = min(self.capacity, tokens + (now - last_ts) * self.refill_per_sec)
+        if tokens < 1:
+            self._buckets[key] = (tokens, now)
+            return False
+        self._buckets[key] = (tokens - 1, now)
+        return True
+
+
+# Burst of 5, refilling 1/minute — enrollment is a first-boot, once-per-node
+# event, so a legitimate node never needs more than a handful of attempts.
+_enroll_limiter = _TokenBucket(capacity=5, refill_per_sec=1 / 60)
+
+
+def _hash_secret(secret: str) -> str:
+    return hashlib.sha256(secret.encode()).hexdigest()
+
+
+class EnrollRequest(BaseModel):
+    node_id: str
+    name: Optional[str] = None
+    lat: float = 0.0
+    lon: float = 0.0
+
+
+class EnrollResponse(BaseModel):
+    node_id: str
+    pickup_secret: str
+    approval_status: str
+
+
+@router.post("/enroll", response_model=EnrollResponse)
+async def enroll_node(
+    body: EnrollRequest,
+    request: Request,
+    x_enrollment_token: Optional[str] = Header(None),
+):
+    client_ip = request.client.host if request.client else "unknown"
+    if not _enroll_limiter.allow(client_ip):
+        raise HTTPException(429, "Too many enrollment attempts. Try again later.")
+
+    if not settings.enrollment_token:
+        raise HTTPException(503, "Enrollment is not configured on this server.")
+    if not x_enrollment_token or not secrets.compare_digest(x_enrollment_token, settings.enrollment_token):
+        logger.warning(f"Enroll 401: bad/missing enrollment token from {client_ip} for node_id={body.node_id!r}")
+        raise HTTPException(401, "Invalid or missing X-Enrollment-Token")
+
+    node_id = body.node_id.strip()
+    if not node_id:
+        raise HTTPException(400, "node_id is required")
+
+    existing = await fstore.doc_get("nodes", node_id)
+
+    # -------------------------------------------------------------------
+    # CRITICAL GUARD — do not remove or weaken this check.
+    #
+    # An already-approved node_id must NEVER get a fresh pickup_secret off
+    # the fleet-wide enrollment token alone. The fleet token is shared by
+    # every node (it ships in every node's .env / setup.sh prompt), so it
+    # is the credential most likely to leak. Without this guard, a leaked
+    # fleet token plus a guessable node_id (node-001, node-002, ...) would
+    # let an attacker "re-enroll" a live, already-approved node and race
+    # the real node to GET /nodes/{id}/credentials — stealing its actual
+    # api_key before the legitimate device ever asks.
+    #
+    # Recovery for an approved node goes through the existing admin-only
+    # POST /nodes/{id}/reissue-key instead (routers/nodes.py), which
+    # requires a Firebase admin token, not the fleet token.
+    # -------------------------------------------------------------------
+    if existing and existing.get("approval_status") == "approved":
+        logger.warning(
+            f"Enroll refused: node_id={node_id!r} is already approved — "
+            f"refusing to issue a new pickup_secret from the fleet token alone "
+            f"(source_ip={client_ip})"
+        )
+        raise HTTPException(
+            403,
+            "Node is already approved. This endpoint cannot re-issue credentials "
+            "for an approved node from the enrollment token alone — use admin "
+            "key reissue.",
+        )
+
+    pickup_secret = secrets.token_hex(24)
+    doc = {
+        "node_id": node_id,
+        "name": body.name or (existing or {}).get("name") or node_id,
+        "lat": body.lat or (existing or {}).get("lat", 0.0),
+        "lon": body.lon or (existing or {}).get("lon", 0.0),
+        "approval_status": (existing or {}).get("approval_status", "pending"),
+        "pickup_secret_hash": _hash_secret(pickup_secret),
+    }
+    # approval_status stays "pending" for a brand-new node; if it's an
+    # existing "pending" or "rejected" node re-enrolling (e.g. lost its
+    # pickup_secret before an admin ever approved it), leave whatever
+    # status it already has rather than silently flipping "rejected" back
+    # to "pending" — that decision belongs to an admin, not this endpoint.
+    await fstore.doc_set("nodes", node_id, doc, merge=True)
+    logger.info(f"Node enrolled: {node_id} (status={doc['approval_status']}, source_ip={client_ip})")
+
+    return EnrollResponse(node_id=node_id, pickup_secret=pickup_secret, approval_status=doc["approval_status"])
+
+
+class CredentialsResponse(BaseModel):
+    approval_status: str
+    api_key: Optional[str] = None
+
+
+@router.get("/{node_id}/credentials", response_model=CredentialsResponse)
+async def get_node_credentials(node_id: str, x_pickup_secret: Optional[str] = Header(None)):
+    if not x_pickup_secret:
+        raise HTTPException(401, "Missing X-Pickup-Secret header")
+
+    node = await fstore.doc_get("nodes", node_id)
+    if not node or not node.get("pickup_secret_hash"):
+        raise HTTPException(404, "Unknown node, or node was never enrolled via POST /nodes/enroll")
+
+    if not secrets.compare_digest(_hash_secret(x_pickup_secret), node["pickup_secret_hash"]):
+        raise HTTPException(401, "Invalid pickup secret")
+
+    approval_status = node.get("approval_status", "pending")
+    if approval_status != "approved":
+        return CredentialsResponse(approval_status=approval_status)
+
+    key_doc = await fstore.doc_get("node_keys", node_id)
+    if not key_doc or not key_doc.get("api_key"):
+        # Approved but no key yet — shouldn't normally happen, approve_node()
+        # always writes node_keys in the same call that sets approved. Treat
+        # it as "keep polling" rather than erroring the node's retry loop.
+        return CredentialsResponse(approval_status=approval_status)
+
+    return CredentialsResponse(approval_status=approval_status, api_key=key_doc["api_key"])

drb-c2-core/app/routers/nodes.py

@@ -5,6 +5,8 @@ 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
 from app.routers.tokens import assign_token, release_token
 
@@ -31,8 +33,23 @@ async def approve_node(node_id: str, _: dict = Depends(require_admin_token)):
         raise HTTPException(404, f"Node '{node_id}' not found.")
 
     api_key = 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}")
+
     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)
     return {"ok": True}
 
@@ -42,6 +59,11 @@ 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)
 
@@ -102,7 +124,15 @@ async def reissue_node_key(node_id: str, _: dict = Depends(require_admin_token))
     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}