diff --git a/README.md b/README.md index 0ddcfe8..e5ce95c 100644 --- a/README.md +++ b/README.md @@ -195,12 +195,23 @@ Hardening done beyond the plan: - [ ] Configuration drift detection (§48) - [ ] Last-known-connection tracking + connected-client view (§34) -### Post-MVP (§68.4 / §69 / §70) — NOT STARTED +### Post-MVP (§68.4 / §69 / §70) — PARTIAL -- [ ] Topic explorer / MQTT test client +- [x] Topic explorer (spec §35/§36): live message stream over SSE + (one MQTT v5 client per browser session, SUBACK-gated subscribe), + publish with QoS/retain (audited as TOPIC_PUBLISHED — metadata + only, never the payload), retained-message listing and clearing + (audited as RETAINED_CLEARED); RBAC: viewing = operator+, + publishing/clearing = broker-admin+ — verified live end-to-end +- [ ] Topic explorer: consolidated topic view (spec §35.1) — one row + per known topic; highlight differences between the new and the + previous message per topic (payload, QoS, retain, MQTT 5 + properties); briefly glow the row on new-message arrival; + browse the known history of each topic - [ ] Connection tester + client provisioning wizard - [ ] Bridge configuration -- [ ] Retained-message administration +- [ ] Retained-message administration (partially covered by the topic + explorer: listing + clearing; a dedicated §36 page can follow) - [ ] Certificate expiry alerts - [ ] Prometheus metrics + external notifications - [ ] Change approval (four-eyes) workflow diff --git a/backend/app/explorer.py b/backend/app/explorer.py new file mode 100644 index 0000000..e363d83 --- /dev/null +++ b/backend/app/explorer.py @@ -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() \ No newline at end of file diff --git a/backend/app/main.py b/backend/app/main.py index d2a04ee..e63b7e8 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -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) \ No newline at end of file +app.include_router(system.router) +app.include_router(topics.router) \ No newline at end of file diff --git a/backend/app/routers/topics.py b/backend/app/routers/topics.py new file mode 100644 index 0000000..d878cf9 --- /dev/null +++ b/backend/app/routers/topics.py @@ -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 \ No newline at end of file diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 966150b..a32f3a6 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -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 diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 8862d0a..ca0fcf2 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -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"}) diff --git a/frontend/nginx.conf b/frontend/nginx.conf index f385d82..140b070 100644 --- a/frontend/nginx.conf +++ b/frontend/nginx.conf @@ -17,4 +17,17 @@ server { proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Real-IP $remote_addr; } + + # Server-Sent Events (topic explorer): buffering would hold stream + # chunks until the buffer fills — disable it and allow long-lived + # connections. The backend also sends X-Accel-Buffering: no. + location /api/v1/topics/stream/ { + proxy_pass http://mqtt-backend:8000; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Real-IP $remote_addr; + proxy_buffering off; + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; + } } \ No newline at end of file diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index b773626..b5d5f0e 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -11,6 +11,7 @@ import { BrokerConfigPage } from "./pages/BrokerConfig"; import { CertificatesPage } from "./pages/Certificates"; import { SystemPage } from "./pages/System"; import { StatusPage } from "./pages/Status"; +import { TopicExplorerPage } from "./pages/TopicExplorer"; // Navigation structure per spec section 57. Entries with a `to` route are // implemented; entries without one are planned pages, rendered greyed out @@ -52,7 +53,7 @@ const NAV_SECTIONS: { { title: "Tools", items: [ - { label: "Topic Explorer", hint: "Test console (post-MVP)" }, + { label: "Topic Explorer", to: "/topics" }, { label: "Connection Test", hint: "Connectivity tester (post-MVP)" }, { label: "Client Provisioning", hint: "Provisioning wizard (post-MVP)" }, ], @@ -254,6 +255,14 @@ export function App() { } /> + + + + } + /> } /> ); diff --git a/frontend/src/api.ts b/frontend/src/api.ts index f8ee662..8121123 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -248,4 +248,29 @@ export interface Revision { export interface SystemStatus { broker_alive: boolean; restart_required: boolean; +} + +export interface SubscribeResponse { + session_id: string; +} + +export interface ExplorerMessage { + index: number; + topic: string; + qos: number; + retain: boolean; + received_at: number; + payload_b64: string; + payload_text: string | null; + properties: Record | null; +} + +export interface RetainedMessage { + index: number; + topic: string; + qos: number; + retain: boolean; + received_at: number; + payload_b64: string; + payload_text: string | null; } \ No newline at end of file diff --git a/frontend/src/pages/TopicExplorer.tsx b/frontend/src/pages/TopicExplorer.tsx new file mode 100644 index 0000000..2dfc553 --- /dev/null +++ b/frontend/src/pages/TopicExplorer.tsx @@ -0,0 +1,345 @@ +import { FormEvent, useEffect, useRef, useState } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { api, type ExplorerMessage, type RetainedMessage } from "../api"; + +const MAX_MESSAGES = 500; // UI-side cap; the backend buffers 1000 + +function toHex(b64: string): string { + const bytes = atob(b64); + return Array.from(bytes) + .map((b) => b.charCodeAt(0).toString(16).padStart(2, "0")) + .join(" "); +} + +function PayloadView({ msg }: { msg: ExplorerMessage | RetainedMessage }) { + const [mode, setMode] = useState<"text" | "hex">( + msg.payload_text !== null ? "text" : "hex", + ); + return ( +
+ + 
+        {mode === "text"
+          ? msg.payload_text ?? "(not valid UTF-8 — use hex view)"
+          : toHex(msg.payload_b64) || "(empty payload)"}
+
+ {"properties" in msg && msg.properties && ( +
+ MQTT 5 properties:{" "} + {JSON.stringify(msg.properties)} +
+ )} +
+ ); +} + +export function TopicExplorerPage() { + const queryClient = useQueryClient(); + const [filter, setFilter] = useState(""); + const [messages, setMessages] = useState([]); + const [selected, setSelected] = useState(null); + const [streamError, setStreamError] = useState(null); + const [streaming, setStreaming] = useState(false); + + // Publish form state + const [pubTopic, setPubTopic] = useState(""); + const [pubPayload, setPubPayload] = useState(""); + const [pubQos, setPubQos] = useState(0); + const [pubRetain, setPubRetain] = useState(false); + const [actionError, setActionError] = useState(null); + + // Retained panel + const [showRetained, setShowRetained] = useState(false); + const [retainedSelected, setRetainedSelected] = + useState(null); + + const eventSource = useRef(null); + + const stopStream = () => { + eventSource.current?.close(); + eventSource.current = null; + setStreaming(false); + }; + + const startStream = async (topicFilter: string) => { + stopStream(); + setMessages([]); + setSelected(null); + setStreamError(null); + try { + const { session_id } = await api.post<{ session_id: string }>( + "/topics/subscribe", + { topic_filter: topicFilter }, + ); + const es = new EventSource(`/api/v1/topics/stream/${session_id}`); + es.onopen = () => setStreaming(true); + es.onmessage = (event) => { + const msg: ExplorerMessage = JSON.parse(event.data); + setMessages((prev) => { + const next = [...prev, msg]; + return next.length > MAX_MESSAGES ? next.slice(-MAX_MESSAGES) : next; + }); + }; + es.onerror = () => { + setStreamError( + "Stream closed (session ended or broker unreachable). Restart the subscription to continue.", + ); + stopStream(); + }; + eventSource.current = es; + } catch (e) { + setStreamError((e as Error).message); + } + }; + + // Cleanup on unmount: closing the EventSource triggers the backend + // session teardown (stream generator's finally). + useEffect(() => () => eventSource.current?.close(), []); + + const publish = useMutation({ + mutationFn: () => + api.post("/topics/publish", { + topic: pubTopic, + payload_b64: btoa(pubPayload), + qos: pubQos, + retain: pubRetain, + }), + onSuccess: () => { + setActionError(null); + setPubPayload(""); + }, + onError: (e: Error) => setActionError(e.message), + }); + + const retainedQuery = useQuery({ + queryKey: ["retained"], + queryFn: () => api.get("/topics/retained"), + enabled: showRetained, + }); + + const clearRetained = useMutation({ + mutationFn: (topic: string) => + api.post("/topics/retained/clear", { topic }), + onSuccess: () => + queryClient.invalidateQueries({ queryKey: ["retained"] }), + onError: (e: Error) => setActionError(e.message), + }); + + const submitSubscribe = (e: FormEvent) => { + e.preventDefault(); + if (filter.trim()) void startStream(filter.trim()); + }; + + const submitPublish = (e: FormEvent) => { + e.preventDefault(); + if (pubTopic.trim()) publish.mutate(); + }; + + return ( +
+

Topic explorer

+

+ Live view of broker traffic (spec section 35). Subscribes with the + management identity — every topic the broker carries is visible. + Publishes are recorded in the audit log. +

+ + {streamError &&
{streamError}
} + {actionError &&
{actionError}
} + +
+ +
+ setFilter(e.target.value)} + placeholder="e.g. site/+/telemetry/#" + style={{ maxWidth: "300px" }} + /> + {streaming ? ( + + ) : null} + +
+
+ +

Messages

+ + + + + + + + + + + + {messages.map((m) => ( + setSelected(m)} + style={{ cursor: "pointer" }} + > + + + + + + + ))} + +
TimeTopicQoSRetainPayload
+ {new Date(m.received_at * 1000).toLocaleTimeString()} + {m.topic}{m.qos}{m.retain ? "yes" : "—"} + {m.payload_text ?? `(binary, ${atob(m.payload_b64).length} B)`} +
+ {messages.length === 0 && !streamError && ( +

+ {streaming + ? "Waiting for messages matching the filter…" + : "Not subscribed — enter a filter above."} +

+ )} + + {selected && ( +
+

+ Message on {selected.topic} +

+ +
+ )} + +

Publish test message

+
+
+
+ + setPubTopic(e.target.value)} + placeholder="e.g. site/test/command" + /> +
+
+ + +
+
+ + +
+
+ + setPubPayload(e.target.value)} + placeholder='e.g. {"cmd": "restart"}' + /> +
+ +
+
+ +

Retained messages

+

+ {" "} + + Briefly subscribes to # on the broker (spec section 36). + +

+ {showRetained && ( + <> + {retainedQuery.error && ( +
+ {(retainedQuery.error as Error).message} +
+ )} + + + + + + + + + + + {(retainedQuery.data ?? []).map((m) => ( + + + + + + + ))} + +
TopicQoSPayload
{m.topic}{m.qos} setRetainedSelected(m)} + > + {m.payload_text ?? "(binary)"} + + +
+ {(retainedQuery.data ?? []).length === 0 && !retainedQuery.isLoading && ( +

No retained messages.

+ )} + + )} + {retainedSelected && ( +
+

+ Retained payload on {retainedSelected.topic} +

+ +
+ )} +
+ ); +} \ No newline at end of file diff --git a/specification.md b/specification.md index e4769a6..2622b61 100644 --- a/specification.md +++ b/specification.md @@ -1174,6 +1174,23 @@ set retained flag inspect MQTT 5 properties ``` +## 35.1 Consolidated topic view + +The live message view shall be organized as ONE ROW PER KNOWN TOPIC +(rather than one row per received message), so an active topic tree is +browsable at a glance. For each row: + +* on receiving a new message, the row shall HIGHLIGHT THE DIFFERENCES + between the new message and the previously known message for that + topic (payload, QoS, retain flag and MQTT 5 properties that changed) +* the row shall GLOW briefly when a new message arrives, so the user + notices activity without watching every row + +The full KNOWN HISTORY of each topic shall be browsable: selecting a +topic row opens its message history (most recent first), from which any +entry can be inspected in detail (payload as text/hex, QoS, retain +flag, MQTT 5 properties, arrival time). + Example: ```text