jobe/mqtt-broker-stack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 Add topic explorer

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-09-20 09:23:04 +02:00
parent 1e491642f1
commit 6385a27411
11 changed files with 1184 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/explorer.py
+318
View File
@@ -0,0 +1,318 @@
"""Topic explorer sessions (spec sections 35, 36).
One MQTT v5 client per active browser stream. Sessions live only while
the Server-Sent-Events connection is open; the stream generator's
finally-clause stops the session, which disconnects the client — no
lingering subscriptions, no manual cleanup by the operator.
Why MQTT v5: the explorer must surface MQTT 5 publish properties (spec
section 35), which only arrive when the client itself speaks v5. The
dynsec management channel deliberately stays on its own connection.
Payloads are base64 with a looks-like-UTF-8 flag: messages can be binary,
and the UI offers text/hex views accordingly.
"""
from __future__ import annotations
import base64
import json
import logging
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
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__)
_BUFFER_SIZE = 1000 # ring buffer per session
_GRACE_SECONDS = 30.0 # page reload / SSE reconnect window
def _payload_info(payload: bytes) -> dict[str, Any]:
"""base64 + heuristic UTF-8 decoding for the UI."""
info: dict[str, Any] = {"payload_b64": base64.b64encode(payload).decode()}
try:
info["payload_text"] = payload.decode("utf-8")
except UnicodeDecodeError:
info["payload_text"] = None # binary — UI shows hex
return info
@dataclass
class _Message:
index: int
topic: str
qos: int
retain: bool
received_at: float
payload_b64: str
payload_text: str | None
properties: dict[str, Any] | None = None
def as_dict(self) -> dict[str, Any]:
return {
"index": self.index,
"topic": self.topic,
"qos": self.qos,
"retain": self.retain,
"received_at": self.received_at,
"payload_b64": self.payload_b64,
"payload_text": self.payload_text,
"properties": self.properties,
}
@dataclass
class ExplorerSession:
id: str
username: str
topic_filter: str
client: mqtt.Client = field(repr=False)
buffer: deque[_Message] = field(default_factory=lambda: deque(maxlen=_BUFFER_SIZE))
_next_index: int = 0
_cond: threading.Condition = field(default_factory=threading.Condition)
_subscribed: threading.Event = field(default_factory=threading.Event)
created_at: float = field(default_factory=time.time)
def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
if reason_code == 0:
client.subscribe(self.topic_filter, qos=0)
else:
logger.error("explorer session %s connect failed: %s", self.id, reason_code)
def _on_subscribe(self, client, userdata, mid, reason_codes, properties) -> None:
# SUBACK received: the subscription is now live on the broker.
self._subscribed.set()
def _on_message(self, client, userdata, msg) -> None:
# MQTT v5: surface publish properties (content type, response
# topic, user properties, ...). paho exposes them as msg.properties.
props: dict[str, Any] | None = None
raw_props = getattr(msg, "properties", None)
if raw_props is not None:
try:
props = {}
if getattr(raw_props, "ContentType", None):
props["content_type"] = raw_props.ContentType
if getattr(raw_props, "ResponseTopic", None):
props["response_topic"] = raw_props.ResponseTopic
if getattr(raw_props, "CorrelationData", None):
props["correlation_data"] = raw_props.CorrelationData.decode(
"utf-8", errors="replace"
)
if getattr(raw_props, "UserProperty", None):
props["user_properties"] = [
[k, v] for k, v in raw_props.UserProperty
]
except Exception: # noqa: BLE001 - properties must never break capture
props = None
info = _payload_info(msg.payload)
with self._cond:
message = _Message(
index=self._next_index,
topic=msg.topic,
qos=msg.qos,
retain=bool(getattr(msg, "retain", False)),
received_at=time.time(),
payload_b64=info["payload_b64"],
payload_text=info["payload_text"],
properties=props,
)
self._next_index += 1
self.buffer.append(message)
self._cond.notify_all()
class ExplorerSessions:
"""Thread-safe registry of live explorer sessions."""
def __init__(self) -> None:
self._lock = threading.Lock()
self._sessions: dict[str, ExplorerSession] = {}
def start(self, username: str, topic_filter: str) -> ExplorerSession:
settings = get_settings()
session_id = uuid.uuid4().hex
client = mqtt.Client(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=f"mqtt-explorer-{session_id[:8]}",
protocol=mqtt.MQTTv5,
)
client.username_pw_set(
settings.mosquitto_dynsec_username, settings.mosquitto_dynsec_password
)
session = ExplorerSession(
id=session_id,
username=username,
topic_filter=topic_filter,
client=client,
)
client.on_connect = session._on_connect
client.on_subscribe = session._on_subscribe
client.on_message = session._on_message
client.connect(
settings.mosquitto_host, settings.mosquitto_mgmt_port, keepalive=30
)
client.loop_start()
# Block until the broker acknowledges the subscription: without
# this, a message published immediately after subscribe returns
# can be missed (SUBACK arrives asynchronously on the network
# loop). 5 s covers a cold broker start; on timeout the session
# still proceeds — delivery may just start slightly later.
if not session._subscribed.wait(timeout=5.0):
logger.warning(
"explorer session %s: SUBACK not received within 5 s",
session_id,
)
with self._lock:
self._sessions[session_id] = session
return session
def get(self, session_id: str) -> ExplorerSession:
with self._lock:
session = self._sessions.get(session_id)
if session is None:
raise KeyError(session_id)
return session
def stop(self, session_id: str) -> None:
with self._lock:
session = self._sessions.pop(session_id, None)
if session is None:
return
try:
session.client.disconnect()
except OSError:
pass
session.client.loop_stop()
def sweep(self) -> None:
"""Drop sessions whose buffer has gone silent past the grace
window (browser tab closed without the SSE finally firing)."""
cutoff = time.time() - _GRACE_SECONDS
with self._lock:
stale = [
sid
for sid, s in self._sessions.items()
if s.buffer and s.buffer[-1].received_at < cutoff
and time.time() - s.created_at > _GRACE_SECONDS
]
for sid in stale:
self.stop(sid)
def stop_all(self) -> None:
with self._lock:
ids = list(self._sessions)
for sid in ids:
self.stop(sid)
sessions = ExplorerSessions()
def messages_since(session_id: str, index: int) -> list[dict[str, Any]]:
"""All buffered messages with index > the given one."""
session = sessions.get(session_id)
with session._cond:
return [m.as_dict() for m in session.buffer if m.index > index]
def list_retained(max_results: int = 200, timeout: float = 3.0) -> list[dict[str, Any]]:
"""Enumerate retained messages (spec section 36).
MQTT brokers expose no "list retained" API: the defined mechanism is
to subscribe to # — the broker immediately delivers every retained
message matching the filter (retain flag set on delivery) — collect
briefly, then unsubscribe. $SYS is filtered out of the listing.
"""
settings = get_settings()
collected: dict[str, _Message] = {}
done = threading.Event()
client_id = f"mqtt-explorer-retained-{uuid.uuid4().hex[:8]}"
client = mqtt.Client(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=client_id,
protocol=mqtt.MQTTv5,
)
client.username_pw_set(
settings.mosquitto_dynsec_username, settings.mosquitto_dynsec_password
)
def _on_connect(cl, userdata, flags, reason_code, properties) -> None:
if reason_code == 0:
cl.subscribe("#", qos=0)
else:
done.set() # nothing to collect without a subscription
def _on_message(cl, userdata, msg) -> None:
if not getattr(msg, "retain", False):
return # only retained deliveries belong in the listing
if msg.topic.startswith("$SYS"):
return # telemetry noise, not retained app data
if len(collected) >= max_results:
done.set()
return
info = _payload_info(msg.payload)
collected[msg.topic] = _Message(
index=len(collected),
topic=msg.topic,
qos=msg.qos,
retain=True,
received_at=time.time(),
payload_b64=info["payload_b64"],
payload_text=info["payload_text"],
)
client.on_connect = _on_connect
client.on_message = _on_message
client.connect(settings.mosquitto_host, settings.mosquitto_mgmt_port, keepalive=30)
client.loop_start()
try:
done.wait(timeout)
finally:
try:
client.disconnect()
except OSError:
pass
client.loop_stop()
return [m.as_dict() for _, m in sorted(collected.items())]
def publish_message(
topic: str, payload: bytes, qos: int, retain: bool, timeout: float = 10.0
) -> None:
"""Publish a test message as the management identity (spec section 35).
A short-lived dedicated client keeps the explorer's session clients
purely read-only.
"""
settings = get_settings()
client = mqtt.Client(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=f"mqtt-explorer-pub-{uuid.uuid4().hex[:8]}",
protocol=mqtt.MQTTv5,
)
client.username_pw_set(
settings.mosquitto_dynsec_username, settings.mosquitto_dynsec_password
)
client.connect(settings.mosquitto_host, settings.mosquitto_mgmt_port, keepalive=30)
client.loop_start()
try:
info = client.publish(topic, payload, qos=qos, retain=retain)
info.wait_for_publish(timeout=timeout)
if not info.is_published():
raise RuntimeError("publish did not complete in time")
finally:
try:
client.disconnect()
except OSError:
pass
client.loop_stop()
backend/app/main.py
+42 -2
View File
@@ -81,9 +81,48 @@ async def lifespan(app: FastAPI):
logger.info("sys-observe role granted to the management identity")
except Exception as exc: # noqa: BLE001
logger.warning("could not grant sys-observe: %s", exc)
# The topic explorer (spec section 35) publishes and subscribes
# to APPLICATION topics as the management identity. The 2.1
# auto-generated admin role set does NOT include publish/subscribe
# on # (deliberately: "the admin user does not have access to
# publish to normal application topics"). A dedicated role keeps
# the grant explicit and auditable; publishes through the
# explorer remain audited as TOPIC_PUBLISHED (spec section 35).
from app.dynsec import DynsecError
try:
try:
dynsec_client.create_role(
"topic-explorer",
acls=[
{
"acltype": "publishClientSend",
"topic": "#",
"allow": True,
"priority": 0,
},
{
"acltype": "subscribePattern",
"topic": "#",
"allow": True,
"priority": 0,
},
],
)
except DynsecError:
pass # role already exists (idempotent startup)
dynsec_client.add_client_role(
settings.mosquitto_dynsec_username, "topic-explorer", priority=1
)
logger.info("topic-explorer role granted to the management identity")
except Exception as exc: # noqa: BLE001
logger.warning("could not grant topic-explorer: %s", exc)
except Exception as exc: # noqa: BLE001 - startup should not crash the API
logger.warning("dynsec channel not ready yet: %s", exc)
yield
from app import explorer
explorer.sessions.stop_all()
dynsec_client.close()
@@ -140,11 +179,12 @@ def health_ready():
# Routers
from app.routers import auth, broker, mqtt, pki, system, users # noqa: E402
from app.routers import auth, broker, mqtt, pki, system, topics, users # noqa: E402
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(mqtt.router)
app.include_router(broker.router)
app.include_router(pki.router)
app.include_router(system.router)
app.include_router(system.router)
app.include_router(topics.router)
backend/app/routers/topics.py
+226
View File
@@ -0,0 +1,226 @@
"""Topic explorer API (spec sections 35, 36, 41).
Endpoints:
POST /topics/subscribe -> open an explorer session
GET /topics/stream/{id} -> Server-Sent Events message stream
POST /topics/publish -> publish a test message (audited)
GET /topics/retained -> list retained messages (spec 36)
POST /topics/retained/clear -> clear one retained message (audited)
Security model (spec section 35: access restricted through management
RBAC; all publishes audited):
* subscribe/stream/retained listing: OPERATOR and above (spec
section 9: "perform permitted diagnostic operations")
* publish / clear retained: BROKER_ADMIN and above (state-changing)
* every publish and clear writes an audit event with topic and QoS
metadata — NEVER the payload (audit rows are not a message store).
"""
from __future__ import annotations
import asyncio
import base64
import json
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app import audit, explorer
from app.db import get_db
from app.deps import CurrentUser, require_roles
from app.models import ManagementRole
router = APIRouter(prefix="/api/v1/topics", tags=["topics"])
OperatorRole = Annotated[
CurrentUser,
Depends(
require_roles(
ManagementRole.BROKER_ADMIN,
ManagementRole.OPERATOR,
)
),
]
BrokerAdmin = Annotated[CurrentUser, Depends(require_roles(ManagementRole.BROKER_ADMIN))]
DbDep = Annotated[Session, Depends(get_db)]
class SubscribeRequest(BaseModel):
topic_filter: str = Field(min_length=1, max_length=200)
class SubscribeResponse(BaseModel):
session_id: str
class PublishRequest(BaseModel):
topic: str = Field(min_length=1, max_length=200)
payload_b64: str = Field(default="", description="Empty clears retained")
qos: int = Field(default=0, ge=0, le=2)
retain: bool = False
class RetainedMessage(BaseModel):
index: int
topic: str
qos: int
retain: bool
received_at: float
payload_b64: str
payload_text: str | None
properties: dict | None = None
class ClearRetainedRequest(BaseModel):
topic: str = Field(min_length=1, max_length=200)
def _valid_topic_filter(value: str) -> None:
# Reject wildcard misuse the broker would reject anyway, with a
# clearer error than a generic MQTT rc.
if "#" in value and not value.endswith("#"):
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
"'#' is only valid as the last level of a filter",
)
@router.post("/subscribe", response_model=SubscribeResponse)
def subscribe(body: SubscribeRequest, user: OperatorRole):
"""Open an explorer session for one topic filter (spec section 35)."""
_valid_topic_filter(body.topic_filter)
explorer.sessions.sweep()
try:
session = explorer.sessions.start(user.username, body.topic_filter)
except OSError as exc:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"could not connect to the broker: {exc}",
) from exc
return SubscribeResponse(session_id=session.id)
@router.get("/stream/{session_id}")
async def stream(session_id: str, user: OperatorRole):
"""Server-Sent Events stream of live messages for the session.
Closing the EventSource stops the session (and its MQTT client) via
the generator's finally — the lifecycle is tied to the connection.
"""
try:
explorer.sessions.get(session_id)
except KeyError as exc:
raise HTTPException(status.HTTP_404_NOT_FOUND, "session not found") from exc
async def event_stream():
index = 0
heartbeat = 0.0
try:
while True:
for message in explorer.messages_since(session_id, index):
index = message["index"]
yield f"data: {json.dumps(message)}\n\n"
# Heartbeat comment keeps proxies from idling out.
heartbeat += 1
if heartbeat % 60 == 0: # ~15s at 0.25s cadence
yield ": keepalive\n\n"
await asyncio.sleep(0.25)
finally:
explorer.sessions.stop(session_id)
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no", # nginx: disable response buffering
},
)
@router.post("/publish", status_code=status.HTTP_204_NO_CONTENT)
def publish(
body: PublishRequest,
request: Request,
db: DbDep,
user: BrokerAdmin,
):
"""Publish a test message (spec section 35: all publishes audited).
An empty payload with retain=true is MQTT's clear-retained mechanism
(spec section 36) — the same call serves both purposes.
"""
_valid_topic_filter(body.topic)
if any(c in body.topic for c in ("#", "+")):
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
"wildcards cannot be published to — use a concrete topic",
)
try:
payload = base64.b64decode(body.payload_b64) if body.payload_b64 else b""
except Exception as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid base64 payload") from exc
try:
explorer.publish_message(body.topic, payload, body.qos, body.retain)
except (OSError, RuntimeError) as exc:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"publish failed: {exc}",
) from exc
audit.record(
db,
"TOPIC_PUBLISHED",
request=request,
user=user.username,
object_type="topic",
object_id=body.topic,
new_value={"qos": body.qos, "retain": body.retain, "bytes": len(payload)},
)
return None
@router.get("/retained", response_model=list[RetainedMessage])
def retained(user: OperatorRole):
"""List retained messages (spec section 36).
Brokers have no list-retained API: this subscribes briefly to #,
collects retain-flag deliveries, and unsubscribes. Capped; large
brokers deliver everything on subscribe.
"""
try:
return explorer.list_retained()
except OSError as exc:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"could not connect to the broker: {exc}",
) from exc
@router.post("/retained/clear", status_code=status.HTTP_204_NO_CONTENT)
def clear_retained(
body: ClearRetainedRequest,
request: Request,
db: DbDep,
user: BrokerAdmin,
):
"""Clear one retained message: publish a zero-length payload with the
retain flag (MQTT's defined mechanism, spec section 36)."""
try:
explorer.publish_message(body.topic, b"", qos=0, retain=True)
except (OSError, RuntimeError) as exc:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"clear failed: {exc}",
) from exc
audit.record(
db,
"RETAINED_CLEARED",
request=request,
user=user.username,
object_type="topic",
object_id=body.topic,
)
return None
backend/tests/conftest.py
+65
View File
@@ -245,6 +245,71 @@ def client(db_session, monkeypatch):
monkeypatch.setattr(broker_router, "_call_control", _fake_call_control)
monkeypatch.setattr(system_router, "_call_control", _fake_call_control)
# The topic explorer talks to the broker via MQTT. Fake the session
# registry + publish/retained so the endpoints can be exercised
# without a live broker.
from app.routers import topics as topics_router
from app import explorer as explorer_module
class FakeSession:
def __init__(self, id, username, topic_filter):
self.id = id
self.username = username
self.topic_filter = topic_filter
class FakeSessions:
def __init__(self):
self._sessions = {}
self._counter = 0
def sweep(self):
pass
def start(self, username, topic_filter):
self._counter += 1
sid = f"fake-{self._counter}"
self._sessions[sid] = FakeSession(sid, username, topic_filter)
return self._sessions[sid]
def get(self, session_id):
if session_id not in self._sessions:
raise KeyError(session_id)
return self._sessions[session_id]
def stop(self, session_id):
self._sessions.pop(session_id, None)
def stop_all(self):
self._sessions.clear()
fake_sessions = FakeSessions()
monkeypatch.setattr(explorer_module, "sessions", fake_sessions)
def fake_messages_since(session_id, index):
return []
def fake_list_retained(max_results=200, timeout=3.0):
return [
{
"index": 0,
"topic": "site/config/level",
"qos": 1,
"retain": True,
"received_at": 12345.0,
"payload_b64": "MTIz",
"payload_text": "123",
"properties": None,
}
]
def fake_publish(topic, payload, qos, retain, timeout=10.0):
fake_publish.calls.append((topic, payload, qos, retain))
fake_publish.calls = []
monkeypatch.setattr(explorer_module, "messages_since", fake_messages_since)
monkeypatch.setattr(explorer_module, "list_retained", fake_list_retained)
monkeypatch.setattr(explorer_module, "publish_message", fake_publish)
# The router imports names from the module: patch there too.
monkeypatch.setattr(topics_router.explorer, "sessions", fake_sessions)
monkeypatch.setattr(topics_router.explorer, "messages_since", fake_messages_since)
monkeypatch.setattr(topics_router.explorer, "list_retained", fake_list_retained)
monkeypatch.setattr(topics_router.explorer, "publish_message", fake_publish)
with TestClient(main_module.app) as test_client:
yield test_client
backend/tests/test_api.py
+109
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
import json
from tests.conftest import login
@@ -239,6 +241,113 @@ def test_broker_status_endpoint(admin_client, auditor_client):
metrics.reset()
# --- Topic explorer (spec sections 35, 36) ------------------------------------
def test_topic_explorer_rbac(auditor_client):
"""Subscribe/stream/retained need OPERATOR+; publish/clear need
BROKER_ADMIN (spec section 35: access restricted through
management RBAC)."""
# Auditor may not even subscribe (read-only role is not OPERATOR).
assert (
auditor_client.post("/api/v1/topics/subscribe", json={"topic_filter": "#"}).status_code
== 403
)
assert (
auditor_client.get("/api/v1/topics/retained").status_code == 403
)
assert (
auditor_client.post(
"/api/v1/topics/publish", json={"topic": "t", "payload_b64": ""}
).status_code
== 403
)
def test_topic_explorer_flow(admin_client):
"""Operator-level (admin is superuser) subscribe/stream/retained."""
sub = admin_client.post("/api/v1/topics/subscribe", json={"topic_filter": "#"})
assert sub.status_code == 200
session_id = sub.json()["session_id"]
# The stream endpoint exists and is SSE. NOTE: consuming the stream
# itself would hang (infinite generator by design) — verify the
# session's liveness through the subscribe/404 contract instead:
# an unknown session must 404, a live one was accepted above.
missing = admin_client.get("/api/v1/topics/stream/does-not-exist")
assert missing.status_code == 404
# Retained listing works.
retained = admin_client.get("/api/v1/topics/retained")
assert retained.status_code == 200
assert any(m["topic"] == "site/config/level" for m in retained.json())
def test_topic_explorer_publish_audited(admin_client, db_session):
"""Publishes through the test console are audited (spec section 35),
and wildcards are rejected."""
import base64
from sqlalchemy import select
from app.models import AuditEvent
payload = {"cmd": "restart"}
response = admin_client.post(
"/api/v1/topics/publish",
json={
"topic": "site/test/command",
"payload_b64": base64.b64encode(json.dumps(payload).encode()).decode(),
"qos": 1,
"retain": False,
},
)
assert response.status_code == 204
events = db_session.scalars(
select(AuditEvent).where(AuditEvent.action == "TOPIC_PUBLISHED")
).all()
assert any(e.object_id == "site/test/command" for e in events)
# The audit row records metadata, never the payload content.
for e in events:
if e.object_id == "site/test/command":
assert e.new_value["qos"] == 1
assert "payload" not in json.dumps(e.new_value).lower()
# Wildcards cannot be published to.
wildcard = admin_client.post(
"/api/v1/topics/publish",
json={"topic": "site/+/command", "payload_b64": "", "qos": 0},
)
assert wildcard.status_code == 400
# A mis-placed # is rejected on subscribe too.
bad = admin_client.post("/api/v1/topics/subscribe", json={"topic_filter": "#/deep"})
assert bad.status_code == 400
def test_topic_explorer_clear_retained(admin_client, db_session):
"""Clearing a retained message publishes a zero-length retained
payload and audits RETAINED_CLEARED (spec section 36)."""
from sqlalchemy import select
from app.models import AuditEvent
cleared = admin_client.post(
"/api/v1/topics/retained/clear", json={"topic": "site/config/level"}
)
assert cleared.status_code == 204
events = db_session.scalars(
select(AuditEvent).where(AuditEvent.action == "RETAINED_CLEARED")
).all()
assert any(e.object_id == "site/config/level" for e in events)
def test_topic_explorer_stream_404_for_unknown_session(admin_client):
missing = admin_client.get("/api/v1/topics/stream/does-not-exist")
assert missing.status_code == 404
def test_role_acl_lifecycle(admin_client):
admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})