From d3590f9228facf4956587bdb0cff6b64bcef6c9b Mon Sep 17 00:00:00 2001 From: MarekWo Date: Tue, 14 Jul 2026 19:00:37 +0200 Subject: [PATCH] feat(observer): MQTT publishing module (meshcore-packet-capture compatible) New app/observer.py: ObserverManager owns per-broker paho-mqtt clients (own network threads, QoS 0, LWT + retained status) and the packet fan-out. Packet decode/hash/format functions are ports of upstream meshcore-packet-capture including its legacy path fallbacks and unknown-version quirk; validated against 10 live packets from the reference observer network (hash/route/type/len all byte-identical). Adds paho-mqtt==2.1.0. Co-Authored-By: Claude Fable 5 --- app/observer.py | 571 +++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 3 + 2 files changed, 574 insertions(+) create mode 100644 app/observer.py diff --git a/app/observer.py b/app/observer.py new file mode 100644 index 0000000..01e1728 --- /dev/null +++ b/app/observer.py @@ -0,0 +1,571 @@ +"""Observer — publish captured mesh packets to MQTT brokers. + +mc-webui equivalent of the standalone meshcore-packet-capture project +(github.com/agessaman/meshcore-packet-capture): every packet the device +overhears (RX_LOG_DATA) is formatted in the exact same JSON shape and +published to all user-configured MQTT brokers, so letsmesh-style +analyzers and other existing consumers work unchanged. + +Threading model: +- DeviceManager's asyncio thread calls handle_raw_packet(): pure CPU + (header parse + SHA-256 + json.dumps) plus paho's non-blocking + publish() enqueue. No await, no network, no DB, no locks on this path. +- Each broker gets its own paho client with its own network thread + (loop_start), auto-reconnecting with backoff. QoS 0 everywhere, so a + dead broker never accumulates queued messages. +- Lifecycle (start/stop/reload) runs on Flask request threads or a + dedicated daemon thread, serialized by _lifecycle_lock. The client + list is swapped as a whole (atomic reference assignment), so the hot + path can read it lock-free. + +The packet parsing/formatting functions below are ports of the upstream +implementation (src/meshcore_packet_capture/packet_capture.py) including +its quirks — e.g. packets with an unknown payload version still get +published, with default route "U" / packet_type "0" — because downstream +consumers may rely on them. +""" + +import hashlib +import json +import logging +import re +import threading +import time +from datetime import datetime, timezone + +from app.version import VERSION_STRING + +logger = logging.getLogger(__name__) + +# Wire route_type (header & 0x03): TRANSPORT_FLOOD, FLOOD, DIRECT, TRANSPORT_DIRECT +_ROUTE_LETTERS = {0x00: "F", 0x01: "F", 0x02: "D", 0x03: "T"} +_TRANSPORT_ROUTES = (0x00, 0x03) # carry a 4-byte transport code after the header +_EMPTY_HASH = "0000000000000000" + + +def _decode_packed_path_length(path_len_byte: int, max_path_size: int = 64) -> tuple: + """Decode the packed path_len byte per MeshCore firmware. + + Low 6 bits = hop count, high 2 bits = bytes-per-hop minus 1. + Returns (path_byte_len, bytes_per_hop). Keeps upstream's two legacy + fallbacks (reserved 4-bytes/hop mode, oversized path) so decoded + output matches meshcore-packet-capture exactly. + """ + hop_count = path_len_byte & 0x3F + bytes_per_hop = (path_len_byte >> 6) + 1 + if bytes_per_hop == 4: # reserved mode -> legacy one-byte interpretation + return path_len_byte, 1 + path_byte_len = hop_count * bytes_per_hop + if path_byte_len > max_path_size: # invalid packed value -> legacy + return path_len_byte, 1 + return path_byte_len, bytes_per_hop + + +def _split_path_hops(path_bytes: bytes, bytes_per_hop: int) -> list: + """Split path bytes into per-hop hex tokens (1-3 bytes per hop).""" + path_hex = path_bytes.hex() + hop_hex_chars = max(bytes_per_hop, 1) * 2 + nodes = [path_hex[i:i + hop_hex_chars] for i in range(0, len(path_hex), hop_hex_chars)] + if (len(path_hex) % hop_hex_chars) != 0: + nodes = [path_hex[i:i + 2] for i in range(0, len(path_hex), 2)] + return nodes + + +def decode_packet(raw_hex: str): + """Decode packet header/path — port of upstream decode_and_publish_message(). + + Returns None for truncated packets and unknown payload versions + (those still get published, with default route/type). ADVERT payload + contents are not parsed: upstream never publishes them in the + packets payload. + """ + try: + byte_data = bytes.fromhex(raw_hex) + except ValueError: + return None + if len(byte_data) < 2: + return None + header = byte_data[0] + route_type = header & 0x03 + has_transport = route_type in _TRANSPORT_ROUTES + offset = 1 + if has_transport: + offset += 4 + if len(byte_data) <= offset: + return None + path_len_byte = byte_data[offset] + offset += 1 + path_byte_len, bytes_per_hop = _decode_packed_path_length(path_len_byte) + if len(byte_data) < offset + path_byte_len: + return None + path_bytes = byte_data[offset:offset + path_byte_len] + if (header >> 6) & 0x03 != 0: # only payload version VER_1 (0) is understood + return None + return { + 'route_type': route_type, + 'payload_type': (header >> 2) & 0x0F, + 'path': _split_path_hops(path_bytes, bytes_per_hop), + 'path_byte_len': path_byte_len, + 'has_transport': has_transport, + } + + +def calculate_packet_hash(raw_hex: str) -> str: + """Packet identity hash — port of firmware Packet::calculatePacketHash(). + + SHA256(payload_type byte || [packed path_len as uint16 LE, TRACE only] + || payload bytes), first 16 hex chars uppercase. + """ + try: + byte_data = bytes.fromhex(raw_hex) + header = byte_data[0] + payload_type = (header >> 2) & 0x0F + offset = 1 + if (header & 0x03) in _TRANSPORT_ROUTES: + offset += 4 + if len(byte_data) <= offset: + return _EMPTY_HASH + path_len_byte = byte_data[offset] + offset += 1 + path_byte_len, _ = _decode_packed_path_length(path_len_byte) + payload_start = offset + path_byte_len + if payload_start > len(byte_data): + return _EMPTY_HASH + hash_obj = hashlib.sha256() + hash_obj.update(bytes([payload_type])) + if payload_type == 9: # TRACE: firmware hashes the path_len as uint16 too + hash_obj.update(path_len_byte.to_bytes(2, byteorder='little')) + hash_obj.update(byte_data[payload_start:]) + return hash_obj.hexdigest()[:16].upper() + except Exception: + return _EMPTY_HASH + + +def format_packet_data(raw_hex: str, origin: str, origin_id: str, + snr=None, rssi=None, payload_length=None) -> dict: + """Build the letsmesh-compatible packet JSON — port of upstream (all values strings).""" + now = datetime.now(timezone.utc) + decoded = decode_packet(raw_hex) + packet_len = len(raw_hex) // 2 + route = "U" + packet_type = "0" + payload_len = "0" + if decoded: + route = _ROUTE_LETTERS.get(decoded['route_type'], "U") + packet_type = str(decoded['payload_type']) + if payload_length is not None: + payload_len = str(payload_length) + else: + transport_bytes = 4 if decoded['has_transport'] else 0 + payload_len = str(max(0, packet_len - 1 - transport_bytes - 1 - decoded['path_byte_len'])) + packet = { + "origin": origin, + "origin_id": (origin_id or '').upper(), + "timestamp": now.isoformat(), + "type": "PACKET", + "direction": "rx", + "time": now.strftime("%H:%M:%S"), + "date": now.strftime("%d/%m/%Y"), + "len": str(packet_len), + "packet_type": packet_type, + "route": route, + "payload_len": payload_len, + "raw": raw_hex.upper(), + "SNR": str(snr) if snr is not None else "Unknown", + "RSSI": str(rssi) if rssi is not None else "Unknown", + "hash": calculate_packet_hash(raw_hex), + } + if route == "D" and decoded: + packet["path"] = ",".join(decoded['path']) + return packet + + +class ObserverManager: + """Owns MQTT clients and the packet fan-out for the Observer feature.""" + + SETTINGS_KEY = 'observer_settings' + LAST_ADVERT_KEY = 'observer_last_advert_at' + DEFAULT_SETTINGS = {'enabled': False, 'iata': '', 'advert_interval_hours': 0} + STATUS_EMIT_MIN_INTERVAL = 2.0 # seconds between observer_status socket emits + + def __init__(self, db, socketio=None): + self.db = db + self.socketio = socketio + self.device_manager = None # set by main.py once DeviceManager exists + self._lifecycle_lock = threading.Lock() + self._clients = [] # per-broker dicts; list ref swapped atomically + self._started = False + self._enabled = False + self._iata = '' + self._origin = None + self._origin_id = None # uppercase device public key + self._status_extras = {'model': 'unknown', 'firmware_version': 'unknown', + 'radio': 'unknown'} + self._packets_seen = 0 + self._packets_published = 0 + self._last_status_emit = 0.0 + + # ---------------- settings ---------------- + + def get_settings(self) -> dict: + settings = dict(self.DEFAULT_SETTINGS) + stored = self.db.get_setting_json(self.SETTINGS_KEY, {}) or {} + for key in settings: + if key in stored: + settings[key] = stored[key] + return settings + + # ---------------- identity (from DeviceManager) ---------------- + + def set_identity(self, name, public_key, self_info=None, device_info=None): + """Record device identity; called from DeviceManager after connect.""" + new_origin_id = (public_key or '').upper() + identity_changed = self._started and new_origin_id != self._origin_id + self._origin = name or 'MeshCore Device' + self._origin_id = new_origin_id + if self_info: + self._status_extras['radio'] = ",".join( + str(self_info.get(key, 0)) + for key in ('radio_freq', 'radio_bw', 'radio_sf', 'radio_cr')) + if device_info: + # Mirrors upstream get_firmware_info(): fw ver >= 3 carries strings + fw_ver = device_info.get('fw ver', 0) or 0 + if fw_ver >= 3: + version = str(device_info.get('ver', 'Unknown')) + if version.startswith('v'): + version = version[1:] + self._status_extras['model'] = device_info.get('model', 'Unknown') + self._status_extras['firmware_version'] = ( + f"v{version} (Build: {device_info.get('fw_build', 'Unknown')})") + else: + self._status_extras['firmware_version'] = f"v{fw_ver}" + if identity_changed: + logger.info("Observer: device identity changed, reloading MQTT clients") + self.reload() + + # ---------------- lifecycle ---------------- + + def ensure_started(self): + """Start clients if enabled and not yet running (device just connected).""" + if self._started: + return + self.reload() + + def reload(self): + """Re-read config and rebuild MQTT clients on a daemon thread. + + loop_stop() joins paho's network thread and can take seconds when a + client is mid-reconnect, so this must never run on a Flask worker + or the DeviceManager loop. + """ + threading.Thread(target=self._reload_sync, name='observer-reload', + daemon=True).start() + + def stop(self, publish_offline=True): + with self._lifecycle_lock: + clients = self._clients + self._clients = [] + self._started = False + self._stop_clients(clients, publish_offline=publish_offline) + + def _reload_sync(self): + with self._lifecycle_lock: + try: + settings = self.get_settings() + self._enabled = bool(settings.get('enabled')) + self._iata = str(settings.get('iata') or '').strip() + old_clients = self._clients + self._clients = [] + self._started = False + self._stop_clients(old_clients, publish_offline=True) + if not self._enabled: + return + if not self._origin_id: + logger.info("Observer enabled, waiting for device connection " + "(identity unknown yet)") + return + new_clients = [] + for row in self.db.list_observer_brokers(): + if row.get('is_disabled'): + continue + try: + new_clients.append(self._build_client(row)) + except Exception as e: + logger.error(f"Observer: failed to set up MQTT client for " + f"broker '{row.get('name')}': {e}") + self._clients = new_clients + self._started = bool(new_clients) + if new_clients: + logger.info(f"Observer running with {len(new_clients)} MQTT client(s), " + f"packets topic: {self._topic('packets')}") + else: + logger.info("Observer enabled but no active brokers configured") + except Exception as e: + logger.error(f"Observer reload failed: {e}", exc_info=True) + finally: + self._emit_status(force=True) + + def _stop_clients(self, clients, publish_offline=True): + for ci in clients: + client = ci['client'] + try: + if publish_offline and client.is_connected(): + msg = client.publish(ci['status_topic'], + json.dumps(self._status_payload('offline')), + qos=0, retain=True) + try: + msg.wait_for_publish(timeout=2) + except Exception: + pass + client.disconnect() + client.loop_stop() + except Exception as e: + logger.debug(f"Observer: error stopping MQTT client " + f"'{ci.get('name')}': {e}") + + def _build_client(self, row) -> dict: + # Lazy import: a missing paho package must not break app startup — + # the observer just stays inert and reports the error. + import paho.mqtt.client as mqtt + + status_topic = self._topic('status') + base = re.sub(r"[^a-zA-Z0-9_-]", "", + f"meshcore_client_{(self._origin_id or 'unknown')[:8]}")[:23] + ci = { + 'broker_id': row['id'], + 'name': row['name'], + 'host': row['host'], + 'port': int(row['port'] or 1883), + 'status_topic': status_topic, + 'packets_topic': self._topic('packets'), + 'connected': False, + 'last_error': None, + } + client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, + client_id=f"{base}_{row['id']}", + clean_session=True) + ci['client'] = client + client.user_data_set(ci) + client.reconnect_delay_set(min_delay=1, max_delay=120) + if row.get('username'): + client.username_pw_set(row['username'], row.get('password') or None) + # LWT must be set before connect: retained offline status is the + # authoritative "observer gone" signal (the app has no graceful + # shutdown path — Docker SIGTERM just kills the process). + client.will_set(status_topic, json.dumps(self._lwt_payload()), + qos=0, retain=True) + if row.get('use_tls'): + import ssl + if row.get('tls_verify', 1): + client.tls_set(cert_reqs=ssl.CERT_REQUIRED) + else: + client.tls_set(cert_reqs=ssl.CERT_NONE) + client.tls_insecure_set(True) + client.on_connect = self._on_connect_cb + client.on_disconnect = self._on_disconnect_cb + client.on_connect_fail = self._on_connect_fail_cb + client.connect_async(ci['host'], ci['port'], keepalive=60) + client.loop_start() + return ci + + # ---------------- topics & payloads ---------------- + + def _topic(self, kind: str) -> str: + if self._iata: + return f"meshcore/{self._iata.upper()}/{self._origin_id}/{kind}" + return f"meshcore/{kind}" # upstream classic fallback when IATA unset + + def _status_payload(self, status: str) -> dict: + return { + 'status': status, + 'timestamp': datetime.now(timezone.utc).isoformat(), + 'origin': self._origin or 'MeshCore Device', + 'origin_id': self._origin_id or 'DEVICE', + 'model': self._status_extras.get('model', 'unknown'), + 'firmware_version': self._status_extras.get('firmware_version', 'unknown'), + 'radio': self._status_extras.get('radio', 'unknown'), + 'client_version': f"mc-webui/{VERSION_STRING}", + } + + def _lwt_payload(self) -> dict: + # Upstream's LWT is the minimal 4-field variant + return { + 'status': 'offline', + 'timestamp': datetime.now(timezone.utc).isoformat(), + 'origin': self._origin or 'MeshCore Device', + 'origin_id': self._origin_id or 'DEVICE', + } + + # ---------------- hot path (DeviceManager asyncio thread) ---------------- + + def handle_raw_packet(self, raw_hex, snr=None, rssi=None, payload_length=None): + """Format one overheard packet and publish it to all connected brokers.""" + if not self._started: + return + clients = self._clients # atomic ref read; never mutated in place + if not clients: + return + self._packets_seen += 1 + try: + payload = json.dumps(format_packet_data( + raw_hex, self._origin, self._origin_id, + snr=snr, rssi=rssi, payload_length=payload_length)) + except Exception as e: + logger.debug(f"Observer: packet formatting failed: {e}") + return + published = False + for ci in clients: + try: + if ci['client'].is_connected(): + result = ci['client'].publish(ci['packets_topic'], payload, + qos=0, retain=False) + if result.rc == 0: # mqtt.MQTT_ERR_SUCCESS + published = True + except Exception as e: + logger.debug(f"Observer: publish to '{ci['name']}' failed: {e}") + if published: + self._packets_published += 1 + self._emit_status() + + # ---------------- MQTT callbacks (paho network threads) ---------------- + + def _on_connect_cb(self, client, userdata, flags, reason_code, properties=None): + failed = getattr(reason_code, 'is_failure', None) + if failed is None: # MQTTv3 int fallback + failed = reason_code != 0 + if failed: + userdata['connected'] = False + userdata['last_error'] = str(reason_code) + logger.warning(f"Observer: MQTT broker '{userdata['name']}' refused " + f"connection: {reason_code}") + else: + userdata['connected'] = True + userdata['last_error'] = None + logger.info(f"Observer: connected to MQTT broker '{userdata['name']}' " + f"({userdata['host']}:{userdata['port']})") + try: + # Retained online status; re-published on every auto-reconnect + client.publish(userdata['status_topic'], + json.dumps(self._status_payload('online')), + qos=0, retain=True) + except Exception as e: + logger.debug(f"Observer: online status publish failed: {e}") + self._emit_status(force=True) + + def _on_disconnect_cb(self, client, userdata, disconnect_flags, reason_code, + properties=None): + userdata['connected'] = False + reason = str(reason_code) + if reason not in ('Success', '0'): + userdata['last_error'] = reason + logger.info(f"Observer: lost connection to MQTT broker " + f"'{userdata['name']}' ({reason}), auto-reconnecting") + self._emit_status(force=True) + + def _on_connect_fail_cb(self, client, userdata): + userdata['connected'] = False + if not userdata.get('last_error'): + userdata['last_error'] = 'connection failed' + self._emit_status() # throttled: fires on every paho retry + + # ---------------- status for UI ---------------- + + def _status_reason(self): + if not self._enabled: + return 'disabled' + if not self._origin_id: + return 'waiting for device connection' + if not self._clients: + return 'no active brokers configured' + return None + + def _live_status(self) -> dict: + """Status snapshot without DB access — safe for socket emits.""" + return { + 'enabled': self._enabled, + 'running': self._started, + 'reason': self._status_reason(), + 'origin': self._origin, + 'origin_id': self._origin_id, + 'iata': self._iata, + 'packets_seen': self._packets_seen, + 'packets_published': self._packets_published, + 'brokers': [{'id': ci['broker_id'], 'name': ci['name'], + 'connected': ci['connected'], + 'last_error': ci['last_error']} for ci in self._clients], + } + + def get_status(self) -> dict: + """Full status for the API: settings + DB broker rows merged with live state.""" + status = self._live_status() + status['settings'] = self.get_settings() + live_by_id = {b['id']: b for b in status.pop('brokers')} + brokers = [] + for row in self.db.list_observer_brokers(): + live = live_by_id.get(row['id'], {}) + brokers.append({ + 'id': row['id'], + 'name': row['name'], + 'host': row['host'], + 'port': row['port'], + 'username': row.get('username', ''), + 'has_password': bool(row.get('password')), + 'use_tls': bool(row.get('use_tls')), + 'tls_verify': bool(row.get('tls_verify', 1)), + 'is_disabled': bool(row.get('is_disabled')), + 'connected': live.get('connected', False), + 'last_error': live.get('last_error'), + }) + status['brokers'] = brokers + try: + status['last_advert_at'] = int(float( + self.db.get_setting(self.LAST_ADVERT_KEY) or 0)) or None + except (TypeError, ValueError): + status['last_advert_at'] = None + return status + + def _emit_status(self, force=False): + if not self.socketio: + return + now = time.time() + if not force and (now - self._last_status_emit) < self.STATUS_EMIT_MIN_INTERVAL: + return + self._last_status_emit = now + try: + self.socketio.emit('observer_status', self._live_status(), + namespace='/chat') + except Exception as e: + logger.debug(f"Observer: status emit failed: {e}") + + # ---------------- advert scheduler (APScheduler job) ---------------- + + def maybe_send_advert(self): + """Send a flood advert when the configured interval has elapsed. + + Called every few minutes by APScheduler; the last-advert timestamp + persists in app_settings so restarts don't re-advertise early. + """ + try: + settings = self.get_settings() + if not settings.get('enabled'): + return + interval_hours = int(settings.get('advert_interval_hours') or 0) + if interval_hours <= 0: + return + dm = self.device_manager + if not dm or not getattr(dm, 'is_connected', False): + return + last = float(self.db.get_setting(self.LAST_ADVERT_KEY) or 0) + now = time.time() + if now - last < interval_hours * 3600: + return + result = dm.send_advert(flood=True) + if result and result.get('success'): + self.db.set_setting(self.LAST_ADVERT_KEY, str(int(now))) + logger.info(f"Observer: sent scheduled flood advert " + f"(interval {interval_hours}h)") + self._emit_status(force=True) + else: + logger.warning(f"Observer: scheduled flood advert failed: {result}") + except Exception as e: + logger.error(f"Observer: advert scheduler error: {e}") diff --git a/requirements.txt b/requirements.txt index 2d7e583..1a73e02 100644 --- a/requirements.txt +++ b/requirements.txt @@ -28,6 +28,9 @@ pycryptodome==3.21.0 # WebSocket support for console (threading mode - no gevent needed) flask-socketio==5.3.6 + +# Observer: MQTT packet publishing (meshcore-packet-capture compatible) +paho-mqtt==2.1.0 python-socketio==5.10.0 python-engineio==4.8.1