374 lines
15 KiB
Python
374 lines
15 KiB
Python
"""Dynamic Security client — the runtime control channel (spec sections 10-14).
|
|
|
|
The backend connects to Mosquitto as the seeded `admin` client and issues
|
|
commands on the $CONTROL/dynamic-security/v1 topic API.
|
|
|
|
Wire protocol (verified against Mosquitto 2.1 source: src/control_common.c,
|
|
plugins/dynamic-security/control.c and clients.c):
|
|
|
|
Request -> $CONTROL/dynamic-security/v1
|
|
{"commands": [{"command": "createClient", ...,
|
|
"correlationData": "<id>"}]}
|
|
Response <- $CONTROL/dynamic-security/v1/response (FIXED topic)
|
|
{"responses": [{"command": "createClient",
|
|
"correlationData": "<id>",
|
|
"error": "..." (only on failure),
|
|
"data": {...}}]}
|
|
|
|
Responses are matched to requests by the echoed correlationData in the
|
|
PAYLOAD, not by topic. Command results live under `data` (data.client,
|
|
data.clients, ...). Client objects report `disabled` (absent == enabled).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import queue
|
|
import threading
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import paho.mqtt.client as mqtt
|
|
from paho.mqtt.enums import CallbackAPIVersion
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CONTROL_TOPIC = "$CONTROL/dynamic-security/v1"
|
|
RESPONSE_TOPIC = "$CONTROL/dynamic-security/v1/response"
|
|
_RESPONSE_TIMEOUT_S = 10.0
|
|
|
|
|
|
class DynsecError(Exception):
|
|
"""The dynamic security API returned an error for a command."""
|
|
|
|
def __init__(self, message: str) -> None:
|
|
super().__init__(message)
|
|
self.message = message
|
|
|
|
|
|
class DynsecTimeout(DynsecError):
|
|
"""No response from the broker management channel."""
|
|
|
|
|
|
class DynsecClient:
|
|
"""Thin, thread-safe wrapper around the dynsec $CONTROL API."""
|
|
|
|
def __init__(self) -> None:
|
|
self._client: mqtt.Client | None = None
|
|
self._lock = threading.Lock()
|
|
self._responses: dict[str, queue.Queue[dict[str, Any]]] = {}
|
|
self._connected = threading.Event()
|
|
# Optional tap for $SYS telemetry (app.metrics registers itself
|
|
# at startup). Receives (topic, payload) for every $SYS message.
|
|
self._event_sink = None
|
|
|
|
# -- connection lifecycle ------------------------------------------------
|
|
def connect(self) -> None:
|
|
with self._lock:
|
|
if self._client is not None and self._connected.is_set():
|
|
return
|
|
if self._client is not None:
|
|
# The previous client lost its connection (broker restart,
|
|
# kicked credentials). paho's auto-reconnect retries with
|
|
# the SAME credentials forever, so it can never recover
|
|
# from a stale password — tear it down and start fresh,
|
|
# which also re-reads the current settings.
|
|
self._teardown_locked()
|
|
settings = get_settings()
|
|
client = mqtt.Client(
|
|
callback_api_version=CallbackAPIVersion.VERSION2,
|
|
client_id=f"mqtt-backend-{uuid.uuid4().hex[:8]}",
|
|
)
|
|
client.username_pw_set(
|
|
settings.mosquitto_dynsec_username, settings.mosquitto_dynsec_password
|
|
)
|
|
client.on_connect = self._on_connect
|
|
client.on_disconnect = self._on_disconnect
|
|
client.on_message = self._on_message
|
|
# Register BEFORE loop_start so the guard in the callbacks
|
|
# (client is not self._client) recognises this client from
|
|
# the very first callback onward.
|
|
self._client = client
|
|
client.connect_async(
|
|
settings.mosquitto_host, settings.mosquitto_mgmt_port, keepalive=30
|
|
)
|
|
client.loop_start()
|
|
if not self._connected.wait(timeout=_RESPONSE_TIMEOUT_S):
|
|
self._teardown_locked()
|
|
raise DynsecError("could not connect to broker management channel")
|
|
|
|
def _teardown_locked(self) -> None:
|
|
"""Stop and discard the current paho client. Call with _lock held."""
|
|
client = self._client
|
|
assert client is not None
|
|
# Null the reference FIRST: _on_disconnect's guard then treats any
|
|
# late callback from this client as stale and ignores it, so a
|
|
# deliberate teardown cannot clear a successor's connected flag.
|
|
self._client = None
|
|
self._connected.clear()
|
|
try:
|
|
client.disconnect()
|
|
except OSError:
|
|
pass # socket may already be gone
|
|
client.loop_stop()
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
if self._client is None:
|
|
return
|
|
self._teardown_locked()
|
|
|
|
def set_event_sink(self, sink) -> None:
|
|
"""Register a callable receiving every $SYS message (topic, payload).
|
|
|
|
Used by app.metrics for the broker dashboard (spec sections 33,
|
|
55). The sink must never raise: telemetry must not be able to
|
|
disturb the management channel.
|
|
"""
|
|
self._event_sink = sink
|
|
|
|
# -- paho callbacks -------------------------------------------------------
|
|
def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
|
|
if reason_code == 0:
|
|
# FIXED response topic — all commands reply here; requests are
|
|
# matched by the correlationData echo inside the payload.
|
|
client.subscribe(RESPONSE_TOPIC, qos=1)
|
|
# $SYS telemetry for the dashboard (spec sections 33, 55).
|
|
# qos 0: telemetry is periodically republished, losing one
|
|
# is harmless.
|
|
client.subscribe("$SYS/#", qos=0)
|
|
self._connected.set()
|
|
else:
|
|
logger.error("dynsec client connect failed: %s", reason_code)
|
|
|
|
def _on_disconnect(
|
|
self, client, userdata, disconnect_flags, reason_code, properties
|
|
) -> None:
|
|
# paho VERSION2 signature: all params are required even when unused.
|
|
if client is not self._client:
|
|
return # stale callback from a client already torn down
|
|
# Without this handler a connection loss went completely unnoticed:
|
|
# the flag stayed set, every command published into the void and
|
|
# burned its full response timeout (10 s per call). Clearing the
|
|
# flag makes health/ready report honestly and lets connect()
|
|
# rebuild the client on the next command. paho's auto-reconnect
|
|
# still fires _on_connect on success for transient drops.
|
|
self._connected.clear()
|
|
logger.warning(
|
|
"dynsec management channel disconnected (rc=%s); commands will "
|
|
"fail until the connection is re-established",
|
|
reason_code,
|
|
)
|
|
|
|
def _on_message(self, client, userdata, msg) -> None:
|
|
if msg.topic != RESPONSE_TOPIC:
|
|
# Not a dynsec reply: forward $SYS telemetry to the sink, if
|
|
# one is registered. Swallow sink errors — telemetry must
|
|
# never disturb the management channel.
|
|
if msg.topic.startswith("$SYS/") and self._event_sink is not None:
|
|
try:
|
|
self._event_sink(msg.topic, msg.payload)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("$SYS event sink raised; message dropped")
|
|
return
|
|
try:
|
|
payload = json.loads(msg.payload.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
logger.warning("unreadable dynsec response payload")
|
|
return
|
|
# Envelope is {"responses": [...]}; fan entries out to waiters by
|
|
# their echoed correlationData.
|
|
for entry in payload.get("responses", []):
|
|
correlation = entry.get("correlationData")
|
|
if correlation and correlation in self._responses:
|
|
self._responses[correlation].put_nowait(entry)
|
|
|
|
# -- command plumbing -----------------------------------------------------
|
|
def _command(self, command: str, **kwargs: Any) -> dict[str, Any]:
|
|
"""Send one command; return its `data` object on success.
|
|
|
|
Raises DynsecError when the broker replies with an error, and
|
|
DynsecTimeout when no response arrives in time.
|
|
"""
|
|
self.connect()
|
|
assert self._client is not None # for type checkers
|
|
correlation = uuid.uuid4().hex
|
|
reply_queue: queue.Queue[dict[str, Any]] = queue.Queue()
|
|
self._responses[correlation] = reply_queue
|
|
body = {
|
|
"commands": [
|
|
{"command": command, "correlationData": correlation, **kwargs}
|
|
]
|
|
}
|
|
try:
|
|
result = self._client.publish(
|
|
CONTROL_TOPIC, json.dumps(body).encode("utf-8"), qos=1
|
|
)
|
|
result.wait_for_publish(timeout=_RESPONSE_TIMEOUT_S)
|
|
try:
|
|
entry = reply_queue.get(timeout=_RESPONSE_TIMEOUT_S)
|
|
except queue.Empty as exc:
|
|
raise DynsecTimeout(f"dynsec command '{command}' timed out") from exc
|
|
finally:
|
|
self._responses.pop(correlation, None)
|
|
|
|
if "error" in entry:
|
|
raise DynsecError(f"{command}: {entry['error']}")
|
|
return entry.get("data") or {}
|
|
|
|
# -- Release 1 API surface (spec sections 11-14) ---------------------------
|
|
|
|
# Clients
|
|
def create_client(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
clientid: str | None = None,
|
|
text_name: str | None = None,
|
|
text_description: str | None = None,
|
|
groups: list[dict[str, Any]] | None = None,
|
|
roles: list[dict[str, Any]] | None = None,
|
|
) -> None:
|
|
params: dict[str, Any] = {"username": username, "password": password}
|
|
if clientid:
|
|
params["clientid"] = clientid
|
|
if text_name:
|
|
params["textname"] = text_name
|
|
if text_description:
|
|
params["textdescription"] = text_description
|
|
if groups:
|
|
params["groups"] = groups
|
|
if roles:
|
|
params["roles"] = roles
|
|
self._command("createClient", **params)
|
|
|
|
def delete_client(self, username: str) -> None:
|
|
self._command("deleteClient", username=username)
|
|
|
|
def enable_client(self, username: str) -> None:
|
|
self._command("enableClient", username=username)
|
|
|
|
def disable_client(self, username: str) -> None:
|
|
self._command("disableClient", username=username)
|
|
|
|
def set_client_password(self, username: str, password: str) -> None:
|
|
self._command("setClientPassword", username=username, password=password)
|
|
|
|
def set_client_id(self, username: str, clientid: str | None) -> None:
|
|
"""Bind (or clear) the expected client ID for a username.
|
|
|
|
With a bound client ID, connections must present exactly that
|
|
client id (username + password + client id must all match) — this
|
|
is what stops one device's credentials being reused by another
|
|
program (spec section 11: "bind an expected client ID").
|
|
"""
|
|
if clientid is None or clientid == "":
|
|
self._command("setClientId", username=username, clientid="")
|
|
else:
|
|
self._command("setClientId", username=username, clientid=clientid)
|
|
|
|
def get_client(self, username: str) -> dict[str, Any]:
|
|
return self._command("getClient", username=username).get("client", {})
|
|
|
|
def list_clients(self) -> list[dict[str, Any]]:
|
|
"""Full client objects (verbose=true gives groups/roles/disabled)."""
|
|
data = self._command("listClients", verbose=True)
|
|
return data.get("clients", [])
|
|
|
|
def set_client_group_memberships(
|
|
self, username: str, groups: list[dict[str, Any]]
|
|
) -> None:
|
|
"""Replace a client's complete group membership list."""
|
|
self._command("modifyClient", username=username, groups=groups)
|
|
|
|
def set_client_roles(self, username: str, roles: list[dict[str, Any]]) -> None:
|
|
"""Replace a client's direct role assignments."""
|
|
self._command("modifyClient", username=username, roles=roles)
|
|
|
|
def add_client_role(self, username: str, rolename: str, priority: int = -1) -> None:
|
|
"""Incrementally add one role to a client (spec section 11).
|
|
|
|
Deliberately incremental rather than modifyClient full-list
|
|
replacement: safer against accidental clobbering from the UI.
|
|
"""
|
|
self._command(
|
|
"addClientRole", username=username, rolename=rolename, priority=priority
|
|
)
|
|
|
|
def remove_client_role(self, username: str, rolename: str) -> None:
|
|
self._command("removeClientRole", username=username, rolename=rolename)
|
|
|
|
# Groups
|
|
def create_group(self, groupname: str, roles: list[dict[str, Any]] | None = None) -> None:
|
|
params: dict[str, Any] = {"groupname": groupname}
|
|
if roles:
|
|
params["roles"] = roles
|
|
self._command("createGroup", **params)
|
|
|
|
def delete_group(self, groupname: str) -> None:
|
|
self._command("deleteGroup", groupname=groupname)
|
|
|
|
def get_group(self, groupname: str) -> dict[str, Any]:
|
|
return self._command("getGroup", groupname=groupname).get("group", {})
|
|
|
|
def list_groups(self) -> list[dict[str, Any]]:
|
|
data = self._command("listGroups", verbose=True)
|
|
return data.get("groups", [])
|
|
|
|
def set_group_roles(self, groupname: str, roles: list[dict[str, Any]]) -> None:
|
|
self._command("modifyGroup", groupname=groupname, roles=roles)
|
|
|
|
def add_group_role(self, groupname: str, rolename: str, priority: int = -1) -> None:
|
|
"""Incrementally add one role to a group (spec section 12)."""
|
|
self._command(
|
|
"addGroupRole", groupname=groupname, rolename=rolename, priority=priority
|
|
)
|
|
|
|
def remove_group_role(self, groupname: str, rolename: str) -> None:
|
|
self._command("removeGroupRole", groupname=groupname, rolename=rolename)
|
|
|
|
def add_group_client(self, groupname: str, username: str, priority: int = 1) -> None:
|
|
self._command(
|
|
"addGroupClient", groupname=groupname, username=username, priority=priority
|
|
)
|
|
|
|
def remove_group_client(self, groupname: str, username: str) -> None:
|
|
self._command("removeGroupClient", groupname=groupname, username=username)
|
|
|
|
# Roles
|
|
def create_role(self, rolename: str, acls: list[dict[str, Any]] | None = None) -> None:
|
|
params: dict[str, Any] = {"rolename": rolename}
|
|
if acls:
|
|
params["acls"] = acls
|
|
self._command("createRole", **params)
|
|
|
|
def delete_role(self, rolename: str) -> None:
|
|
self._command("deleteRole", rolename=rolename)
|
|
|
|
def get_role(self, rolename: str) -> dict[str, Any]:
|
|
return self._command("getRole", rolename=rolename).get("role", {})
|
|
|
|
def list_roles(self) -> list[dict[str, Any]]:
|
|
data = self._command("listRoles", verbose=True)
|
|
return data.get("roles", [])
|
|
|
|
def set_role_acls(self, rolename: str, acls: list[dict[str, Any]]) -> None:
|
|
"""Replace the full ACL list of a role (spec section 14 editor)."""
|
|
self._command("modifyRole", rolename=rolename, acls=acls)
|
|
|
|
# Default ACL behaviour (spec section 58: deny by default)
|
|
def set_default_acl_access(self, acl_type: str, allow: bool) -> None:
|
|
self._command(
|
|
"setDefaultACLAccess",
|
|
acls=[{"acltype": acl_type, "allow": allow}],
|
|
)
|
|
|
|
def get_default_acl_access(self) -> list[dict[str, Any]]:
|
|
return self._command("getDefaultACLAccess").get("acls", [])
|
|
|
|
|
|
# Module-level singleton used by FastAPI dependencies.
|
|
dynsec_client = DynsecClient() |