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

"""
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")