"""The MQTT link to the 8883 device plane (AGENTS.md §10 rule 6, FR-6). The client is one device account on the TLS listener. Its ACL lets it read only ``restrict//config`` and write only its own status and telemetry, so the broker itself stops a child publishing a fake "unlocked" (TRANSPORT_DESIGN.md "Security"). Presence is the broker's Last Will and Testament: the client publishes a retained ``{"online": true}`` when it connects and registers the will ``{"online": false}`` on the same topic, so the Control Service learns the device is gone the moment the connection drops without a clean goodbye (FR-6). A broker that is down is not fatal: the cached config stays in force (FR-14), paho retries on a backoff, and the retained config is delivered again on reconnect. """ from __future__ import annotations import json import logging import ssl from datetime import datetime, timezone from typing import Callable, Mapping from .settings import Settings log = logging.getLogger("webrestrict.mqtt") QOS = 1 KEEPALIVE_SECONDS = 30 def iso_now(moment: datetime | None = None) -> str: """RFC 3339 in UTC, as the Control Service writes timestamps.""" current = moment or datetime.now(timezone.utc) if current.tzinfo is None: current = current.replace(tzinfo=timezone.utc) return current.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def status_payload(locked: bool, app: str = "", tamper: dict[str, str] | None = None) -> str: """The status report the dashboard reads (FR-6, FR-9, FR-17).""" body: dict[str, object] = {"online": True, "locked": locked, "app": app, "at": iso_now()} if tamper: body["tamper"] = {**tamper, "at": iso_now()} return json.dumps(body, sort_keys=True) def telemetry_payload(day: str, apps: Mapping[str, int]) -> str: """A day's per-app totals. Not retained: it is a report, not a rule.""" return json.dumps( { "day": day, "apps": [ {"name": name, "seconds": int(seconds)} for name, seconds in sorted(apps.items()) if name and int(seconds) > 0 ], "domains": [], }, sort_keys=True, ) class MqttLink: """A reconnecting paho client wrapper with a small, testable surface.""" def __init__( self, settings: Settings, on_config: Callable[[str], None], on_connected: Callable[[], None] | None = None, on_lost: Callable[[str], None] | None = None, ): self.settings = settings self.on_config = on_config self.on_connected = on_connected or (lambda: None) self.on_lost = on_lost or (lambda reason: None) self.connected = False self.last_error = "" self._client = None self._stopped = False self._unreachable_logged = False # -- lifecycle --------------------------------------------------------- def start(self) -> None: """Create the client and begin connecting in the background.""" import paho.mqtt.client as mqtt # noqa: PLC0415 - keep the import local client = self._new_client(mqtt) self._client = client client.on_connect = self._handle_connect client.on_connect_fail = self._handle_connect_fail client.on_disconnect = self._handle_disconnect client.on_message = self._handle_message client.username_pw_set(self.settings.mqtt_username, self.settings.mqtt_password or None) if self.settings.ca_file: client.tls_set( ca_certs=self.settings.ca_file, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS_CLIENT, ) else: # pragma: no cover - the lab always pins the CA client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS_CLIENT) # Presence (FR-6): the broker publishes this the moment the link drops # without a clean DISCONNECT. client.will_set( self.settings.status_topic, json.dumps({"online": False}), qos=QOS, retain=True, ) client.reconnect_delay_set(min_delay=1, max_delay=30) log.info( "mqtt connecting to %s:%s as %s (topic %s)", self.settings.broker_host, self.settings.broker_port, self.settings.mqtt_username, self.settings.device_topic, ) client.connect_async( self.settings.broker_host, self.settings.broker_port, keepalive=KEEPALIVE_SECONDS, ) client.loop_start() def stop(self) -> None: """Disconnect cleanly, which suppresses the will.""" self._stopped = True client = self._client if client is None: return try: client.disconnect() client.loop_stop() except Exception as exc: # noqa: BLE001 - shutting down anyway log.debug("mqtt stop: %s", exc) # -- publishing -------------------------------------------------------- def publish_status(self, locked: bool, app: str = "", tamper: dict[str, str] | None = None) -> bool: return self._publish(self.settings.status_topic, status_payload(locked, app, tamper), retain=True) def publish_telemetry(self, day: str, apps: Mapping[str, int]) -> bool: if not apps: return False return self._publish(self.settings.telemetry_topic, telemetry_payload(day, apps), retain=False) def publish_offline(self) -> bool: """Retained offline status, published on a clean shutdown (FR-6).""" return self._publish(self.settings.status_topic, json.dumps({"online": False}), retain=True) def _publish(self, topic: str, payload: str, *, retain: bool) -> bool: client = self._client if client is None or not self.connected: return False try: info = client.publish(topic, payload, qos=QOS, retain=retain) if info.rc != 0: log.warning("publish to %s rejected (rc=%s)", topic, info.rc) return False return True except Exception as exc: # noqa: BLE001 - a lost link retries log.warning("publish to %s failed: %s", topic, exc) return False # -- paho plumbing ----------------------------------------------------- def _new_client(self, mqtt): """paho 2.x takes an explicit callback API version; 1.x does not.""" try: return mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=self.settings.effective_client_id, clean_session=True, protocol=mqtt.MQTTv311, ) except TypeError: # pragma: no cover - paho 1.x return mqtt.Client( client_id=self.settings.effective_client_id, clean_session=True, protocol=mqtt.MQTTv311, ) def _handle_connect(self, client, _userdata, _flags, reason_code, _properties=None): code = getattr(reason_code, "value", reason_code) try: code = int(code) except (TypeError, ValueError): code = 1 if code != 0: self.connected = False self.last_error = f"connect refused: {reason_code}" log.warning("mqtt connect refused: %s", reason_code) return self.connected = True self.last_error = "" self._unreachable_logged = False log.info("mqtt connected, subscribing to %s", self.settings.device_topic) client.subscribe(self.settings.device_topic, qos=QOS) # on_connected publishes the retained online status with the real # locked state, and flushes any telemetry held while offline. self.on_connected() def _handle_connect_fail(self, *_args): """The broker could not be reached at all (FR-14). Logged once per outage so the record shows the client is running on its cached config rather than sitting idle, and not flooded by paho's retry. """ self.connected = False self.last_error = "broker unreachable" if not self._unreachable_logged: self._unreachable_logged = True log.warning( "broker %s:%s is unreachable; enforcing the cached config only (FR-14)", self.settings.broker_host, self.settings.broker_port, ) self.on_lost("broker unreachable") def _handle_disconnect(self, _client, _userdata, _flags, reason_code, _properties=None): self.connected = False message = str(reason_code) self.last_error = message if self._stopped: log.info("mqtt disconnected (client stopped): %s", message) return log.warning("mqtt connection lost: %s — keeping the last config (FR-14)", message) self.on_lost(message) def _handle_message(self, _client, _userdata, message): if message.topic != self.settings.device_topic: return payload = message.payload.decode("utf-8", "replace") try: self.on_config(payload) except Exception as exc: # noqa: BLE001 - a bad config must not kill the link log.warning("config rejected, keeping the last one: %s", exc)