From c3cf7a93e8c99ea981480f8530072e71bb9c11df Mon Sep 17 00:00:00 2001 From: mesh-taiwan Date: Mon, 21 Sep 2026 04:29:55 +0800 Subject: [PATCH] feat: firmware fingerprinting and Meshtastic 2.8/2.8.1 support - Decrypt AES-CCM (use_aead) channels from firmware 2.8.1, tried before AES-CTR - Estimate each node's firmware from on-air traits (public key, relay_node, is_unmessagable, CLIENT_BASE, CRC32-derived node id, XEdDSA signature, AEAD / PAGING_APP / soil-water telemetry); MapReport versions shown verbatim - Link node ids re-keyed by 2.8 through the MAC in NodeInfo ("Same Device") - Show it on the node page, /nodes and the hover card; dashboard card with a Firmware / Hardware toggle (version ladder or hardware models, "Show all") - Capture daemon accumulates evidence into node_fingerprint (flushed by the main loop) and backfills history once on first start (fingerprint_backfill_days, default 30) - Bump meshtastic to 2.7.11 (protobufs 2.8.0: xeddsa fields, hardware models up to 144) and name node_info hardware models stored as UNKNOWN_ - Fix get_node_details for nodes without packets; store MeshPacket.channel in channel_index (was always NULL) --- README.md | 3 +- config.sample.yaml | 4 + pyproject.toml | 1 + src/malla/config.py | 8 + src/malla/database/repositories.py | 301 +++++++- src/malla/database/schema.py | 22 + src/malla/fingerprint.py | 730 ++++++++++++++++++ src/malla/fingerprint_backfill.py | 346 +++++++++ src/malla/mqtt_capture.py | 316 +++++++- src/malla/routes/api_routes.py | 40 + src/malla/routes/node_routes.py | 6 + src/malla/services/node_service.py | 7 + src/malla/templates/dashboard.html | 282 +++++++ src/malla/templates/node_detail.html | 44 +- src/malla/templates/nodes.html | 16 + src/malla/utils/decryption.py | 47 +- .../test_firmware_fingerprint_routes.py | 242 ++++++ tests/unit/test_fingerprint.py | 723 +++++++++++++++++ tests/unit/test_fingerprint_backfill.py | 219 ++++++ tests/unit/test_hardware_model_rename.py | 60 ++ tests/unit/test_mqtt_capture_fingerprint.py | 254 ++++++ uv.lock | 4 +- 22 files changed, 3628 insertions(+), 47 deletions(-) create mode 100644 src/malla/fingerprint.py create mode 100644 src/malla/fingerprint_backfill.py create mode 100644 tests/integration/test_firmware_fingerprint_routes.py create mode 100644 tests/unit/test_fingerprint.py create mode 100644 tests/unit/test_fingerprint_backfill.py create mode 100644 tests/unit/test_hardware_model_rename.py create mode 100644 tests/unit/test_mqtt_capture_fingerprint.py diff --git a/README.md b/README.md index c333d870..bc10ee48 100644 --- a/README.md +++ b/README.md @@ -304,8 +304,9 @@ The following keys are recognised: | `mqtt_topic_prefix` | str | `"msh"` | MQTT topic prefix for Meshtastic messages. | `MALLA_MQTT_TOPIC_PREFIX` | | `mqtt_topic_suffix` | str | `"/+/+/+/#"` | MQTT topic suffix pattern. | `MALLA_MQTT_TOPIC_SUFFIX` | | `mqtt_client_id` | str | `""` | MQTT client ID. Leave empty for a randomly generated ID (recommended). | `MALLA_MQTT_CLIENT_ID` | -| `default_channel_key` | str | `"1PG7OiApB1nwvP+rz05pAQ=="` | Default channel key(s) for decryption (base64). Supports comma-separated list of keys - each will be tried in order until successful. | `MALLA_DEFAULT_CHANNEL_KEY` | +| `default_channel_key` | str | `"1PG7OiApB1nwvP+rz05pAQ=="` | Default channel key(s) for decryption (base64). Supports comma-separated list of keys - each will be tried in order until successful. . Both classic AES-CTR channels and the AES-CCM (`use_aead`) channels introduced in firmware 2.8.1 are handled| `MALLA_DEFAULT_CHANNEL_KEY` | | `data_retention_hours` | int | `0` | Number of hours after which to delete old data (0 = never delete). Automatically cleans up packet_history and node_info records older than specified hours. | `MALLA_DATA_RETENTION_HOURS` | +| `fingerprint_backfill_days` | int | `30` | On the capture daemon's first start after upgrading, fingerprint this many days of stored packets in the background so every node gets a firmware estimate right away (0 = skip; `malla-fingerprint-backfill` runs it by hand). | `MALLA_FINGERPRINT_BACKFILL_DAYS` | | `trusted_proxy_ips` | str | `null` | Comma-separated IPs of trusted reverse proxies. Malla only trusts the configured client-IP header from these exact peers. When running via Gunicorn, these same IPs are passed to `forwarded_allow_ips` so only those peers can forward headers. | `MALLA_TRUSTED_PROXY_IPS` | | `trusted_proxy_client_ip_header` | str | `"X-Forwarded-For"` | Header used to derive the real client IP from a trusted proxy. Recommended for Anubis: `X-Real-IP`. | `MALLA_TRUSTED_PROXY_CLIENT_IP_HEADER` | | `otlp_endpoint` | str | `null` | OpenTelemetry endpoint for sending traces (e.g. `http://localhost:4317`). | `MALLA_OTLP_ENDPOINT` | diff --git a/config.sample.yaml b/config.sample.yaml index 372f7af2..e29386d2 100644 --- a/config.sample.yaml +++ b/config.sample.yaml @@ -79,6 +79,10 @@ debug: false # Number of hours after which to delete old data (0 = never delete) # data_retention_hours: 0 +# Days of stored history the capture daemon fingerprints for firmware +# estimates on its first start after upgrading (0 = skip the automatic run) +# fingerprint_backfill_days: 30 + # OpenTelemetry settings # Endpoint for sending traces (e.g. "http://localhost:4317") # otlp_endpoint: "" diff --git a/pyproject.toml b/pyproject.toml index dc5b8b78..0be0fd92 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,6 +63,7 @@ Issues = "https://github.com/zenitraM/malla/issues" malla-web = "malla.web_ui:main" malla-web-gunicorn = "malla.wsgi:main" malla-capture = "malla.mqtt_capture:main" +malla-fingerprint-backfill = "malla.fingerprint_backfill:main" [project.optional-dependencies] dev = [ diff --git a/src/malla/config.py b/src/malla/config.py index 6ca012bb..3fe18ae7 100644 --- a/src/malla/config.py +++ b/src/malla/config.py @@ -61,6 +61,14 @@ class AppConfig: # Number of hours after which to delete old data (0 = never delete) data_retention_hours: int = 0 + # Firmware fingerprint backfill (capture daemon) + # On its first start after this feature is installed, the capture daemon + # computes firmware evidence for the packets already in the database, this + # many days back, in a background thread. 0 disables the automatic run; + # `malla-fingerprint-backfill` can always be run by hand. + # Corresponding env var: MALLA_FINGERPRINT_BACKFILL_DAYS + fingerprint_backfill_days: int = 30 + # Reverse proxy settings # Comma-separated IPs of trusted reverse proxies. When set, ProxyFix trusts # one trusted proto hop, and Gunicorn is configured to accept forwarded diff --git a/src/malla/database/repositories.py b/src/malla/database/repositories.py index 36a72126..b874d8d3 100644 --- a/src/malla/database/repositories.py +++ b/src/malla/database/repositories.py @@ -6,6 +6,7 @@ import json import logging +import sqlite3 import time from datetime import UTC, datetime from typing import Any, Literal @@ -14,6 +15,12 @@ from meshtastic.protobuf import mqtt_pb2 from ..config import get_config +from ..fingerprint import ( + EVIDENCE_COLUMNS, + FirmwareEvidence, + estimate, + summarize_distribution, +) from ..utils.decryption import try_decrypt_mesh_packet from ..utils.formatting import format_time_ago from ..utils.node_utils import convert_node_id, get_bulk_node_short_names @@ -167,6 +174,7 @@ def _aggregate( band.extend(run_band) return avg_line, band + RELAY_CANDIDATE_CACHE_TTL_SECONDS = 1800 RELAY_CANDIDATE_CACHE_MAX_ENTRIES = 4096 _relay_candidate_cache: dict[tuple[int, int], tuple[float, list[dict[str, Any]]]] = {} @@ -223,6 +231,9 @@ def _store_relay_candidates( # view, so a short TTL cache keeps the common case O(1) while the numbers stay # fresh to within a few seconds. Keyed by gateway_id (None => site-wide). DASHBOARD_STATS_CACHE_TTL_SECONDS = 30 +FIRMWARE_DISTRIBUTION_CACHE_TTL_SECONDS = 300 +_firmware_distribution_cache: dict[int, tuple[float, dict[str, Any]]] = {} +_hardware_distribution_cache: dict[tuple[int, int], tuple[float, dict[str, Any]]] = {} _dashboard_stats_cache: dict[str | None, tuple[float, dict[str, Any]]] = {} @@ -830,7 +841,9 @@ def get_packets( "hop_count", "relay_node", } - order_column = order_by if order_by in valid_order_columns else "timestamp" + order_column = ( + order_by if order_by in valid_order_columns else "timestamp" + ) order_dir_sql = "DESC" if order_dir.lower() == "desc" else "ASC" # Main query @@ -1351,8 +1364,22 @@ def get_nodes( COALESCE(stats.last_packet_time, ni.last_updated) as last_packet_time, datetime(COALESCE(stats.last_packet_time, ni.last_updated), 'unixepoch') as last_packet_str, stats.avg_rssi, - stats.avg_snr + stats.avg_snr, + nf.nodeinfo_count, + nf.has_public_key, + nf.has_unmessagable_field, + nf.id_from_public_key, + nf.mac_mismatch, + nf.relay_self_count, + nf.relay_none_count, + nf.xeddsa_signed_count, + nf.hop_start_zero_count, + nf.hop_start_set_count, + nf.hop_start_mask, + nf.firmware_version, + nf.firmware_version_at FROM node_info ni + LEFT JOIN node_fingerprint nf ON nf.node_id = ni.node_id LEFT JOIN ( SELECT from_node_id as node_id, @@ -1411,8 +1438,22 @@ def get_nodes( ni.last_updated as last_packet_time, datetime(ni.last_updated, 'unixepoch') as last_packet_str, NULL as avg_rssi, - NULL as avg_snr + NULL as avg_snr, + nf.nodeinfo_count, + nf.has_public_key, + nf.has_unmessagable_field, + nf.id_from_public_key, + nf.mac_mismatch, + nf.relay_self_count, + nf.relay_none_count, + nf.xeddsa_signed_count, + nf.hop_start_zero_count, + nf.hop_start_set_count, + nf.hop_start_mask, + nf.firmware_version, + nf.firmware_version_at FROM node_info ni + LEFT JOIN node_fingerprint nf ON nf.node_id = ni.node_id {where_clause} ORDER BY {order_column} {order_dir} LIMIT ? OFFSET ? @@ -1422,6 +1463,8 @@ def get_nodes( query_params = params + [limit, offset] cursor.execute(query, query_params) nodes = [dict(row) for row in cursor.fetchall()] + for node in nodes: + NodeRepository._attach_firmware_estimate(node) conn.close() @@ -1436,6 +1479,245 @@ def get_nodes( logger.error(f"Error getting nodes: {e}") raise + @staticmethod + def _attach_firmware_estimate(node: dict[str, Any]) -> None: + """Replace raw node_fingerprint columns on *node* with a verdict.""" + evidence = FirmwareEvidence.from_row(node) + verdict = estimate( + evidence, role=node.get("role"), hw_model=node.get("hw_model") + ) + for column in EVIDENCE_COLUMNS: + node.pop(column, None) + node["firmware_label"] = verdict["label"] + node["firmware_source"] = verdict["source"] + node["firmware_reasons"] = verdict["reasons"] + node["firmware_evidence"] = verdict["evidence"] + node["firmware_reported_version"] = verdict["reported_version"] + + @staticmethod + def get_node_fingerprint(node_id: int) -> dict[str, Any] | None: + """Raw accumulated fingerprint evidence for a node, if any.""" + conn = get_db_connection() + try: + cursor = conn.cursor() + cursor.execute( + "SELECT * FROM node_fingerprint WHERE node_id = ?", (node_id,) + ) + row = cursor.fetchone() + return dict(row) if row else None + except sqlite3.OperationalError as e: + # Databases written before the table existed simply have no evidence. + logger.debug(f"node_fingerprint unavailable: {e}") + return None + finally: + conn.close() + + @staticmethod + def get_firmware_estimate( + node_id: int, role: str | None = None, hw_model: str | None = None + ) -> dict[str, Any]: + """Firmware verdict (exact reported version or fingerprint band).""" + row = NodeRepository.get_node_fingerprint(node_id) + return estimate(FirmwareEvidence.from_row(row), role=role, hw_model=hw_model) + + @staticmethod + def get_firmware_distribution(days: int = 7) -> dict[str, Any]: + """How many recently-heard nodes fall into each firmware bucket. + + "Recently heard" is any node that sent a packet in the last ``days`` + days (the same covering index get_stats uses for nodes_seen_7d). + Verdicts are computed in Python from node_fingerprint, so this is + cached for a few minutes like the other dashboard aggregates. + """ + days = max(1, min(int(days), 90)) + now = time.time() + cached = _firmware_distribution_cache.get(days) + if ( + cached is not None + and now - cached[0] <= FIRMWARE_DISTRIBUTION_CACHE_TTL_SECONDS + ): + return dict(cached[1]) + + conn = get_db_connection() + try: + cursor = conn.cursor() + fp_columns = ", ".join(f"nf.{c}" for c in EVIDENCE_COLUMNS) + cursor.execute( + f""" + SELECT ni.role, ni.hw_model, {fp_columns} + FROM ( + SELECT DISTINCT from_node_id AS node_id + FROM packet_history + WHERE timestamp > ? AND from_node_id IS NOT NULL + ) seen + LEFT JOIN node_info ni ON ni.node_id = seen.node_id + LEFT JOIN node_fingerprint nf ON nf.node_id = seen.node_id + """, + (now - days * 86400,), + ) + rows = cursor.fetchall() + except sqlite3.OperationalError as e: + logger.debug(f"firmware distribution unavailable: {e}") + rows = [] + finally: + conn.close() + + verdicts = [ + estimate( + FirmwareEvidence.from_row(row), + role=row["role"], + hw_model=row["hw_model"], + now=now, + ) + for row in rows + ] + result = { + "days": days, + "total_nodes": len(verdicts), + "buckets": summarize_distribution(verdicts), + } + _firmware_distribution_cache[days] = (now, dict(result)) + return result + + @staticmethod + def get_hardware_distribution(days: int = 7, top: int = 12) -> dict[str, Any]: + """Hardware models of nodes heard in the last ``days`` days, most common first. + + The dashboard shows this next to the firmware distribution. Models past + the ``top`` most common are folded into one "Other" bucket so the chart + stays readable; nodes that never sent a NodeInfo count as Unknown. + """ + days = max(1, min(int(days), 90)) + top = max(1, min(int(top), 500)) + now = time.time() + cached = _hardware_distribution_cache.get((days, top)) + if ( + cached is not None + and now - cached[0] <= FIRMWARE_DISTRIBUTION_CACHE_TTL_SECONDS + ): + return dict(cached[1]) + + conn = get_db_connection() + try: + cursor = conn.cursor() + cursor.execute( + """ + SELECT COALESCE(NULLIF(ni.hw_model, ''), 'Unknown') AS hw_model, + COUNT(*) AS count + FROM ( + SELECT DISTINCT from_node_id AS node_id + FROM packet_history + WHERE timestamp > ? AND from_node_id IS NOT NULL + ) seen + LEFT JOIN node_info ni ON ni.node_id = seen.node_id + GROUP BY hw_model + ORDER BY count DESC, hw_model ASC + """, + (now - days * 86400,), + ) + rows = [dict(row) for row in cursor.fetchall()] + finally: + conn.close() + + known = [r for r in rows if r["hw_model"] != "Unknown"] + unknown = sum(r["count"] for r in rows if r["hw_model"] == "Unknown") + buckets: list[dict[str, Any]] = [ + {"label": r["hw_model"], "count": r["count"], "kind": "hardware"} + for r in known[:top] + ] + rest = known[top:] + if rest: + buckets.append( + { + "label": f"Other ({len(rest)} models)", + "count": sum(r["count"] for r in rest), + "kind": "other", + } + ) + if unknown: + buckets.append({"label": "Unknown", "count": unknown, "kind": "unknown"}) + result = { + "days": days, + "total_nodes": sum(r["count"] for r in rows), + "models": len(known), + "buckets": buckets, + } + _hardware_distribution_cache[(days, top)] = (now, dict(result)) + return result + + @staticmethod + def get_same_mac_nodes(node_id: int) -> list[dict[str, Any]]: + """Other node numbers that announced the same hardware MAC address. + + Firmware 2.8 derives the node number from the public key instead of + the MAC, so one radio shows up under a new id after upgrading (and + again whenever it re-keys). The MAC it reports in NodeInfo does not + change, which is what ties the identities together. + """ + conn = get_db_connection() + try: + cursor = conn.cursor() + cursor.execute( + "SELECT mac_address, first_seen FROM node_info WHERE node_id = ?", + (node_id,), + ) + me = cursor.fetchone() + if not me or not me["mac_address"]: + return [] + mac = str(me["mac_address"]).strip().lower() + # Empty and all-zero MACs are placeholders, not identities. + if not mac or not set(mac) - set("0:-"): + return [] + cursor.execute( + """ + SELECT + node_id, + printf('!%08x', node_id) AS hex_id, + long_name, + short_name, + hw_model, + role, + first_seen, + last_updated + FROM node_info + WHERE lower(mac_address) = ? AND node_id != ? + ORDER BY last_updated DESC + LIMIT 20 + """, + (mac, node_id), + ) + my_first_seen = me["first_seen"] or 0 + siblings = [] + for row in cursor.fetchall(): + sibling = dict(row) + sibling["node_name"] = ( + sibling["long_name"] or sibling["short_name"] or sibling["hex_id"] + ) + sibling["relation"] = ( + "newer" if (sibling["first_seen"] or 0) > my_first_seen else "older" + ) + sibling["first_seen_str"] = ( + datetime.fromtimestamp(sibling["first_seen"], UTC).strftime( + "%Y-%m-%d" + ) + if sibling["first_seen"] + else None + ) + sibling["last_updated_str"] = ( + datetime.fromtimestamp(sibling["last_updated"], UTC).strftime( + "%Y-%m-%d" + ) + if sibling["last_updated"] + else None + ) + siblings.append(sibling) + return siblings + except sqlite3.OperationalError as e: + logger.debug(f"same-MAC lookup unavailable: {e}") + return [] + finally: + conn.close() + @staticmethod def get_node_details(node_id: int) -> dict[str, Any] | None: """Get comprehensive details about a specific node.""" @@ -1495,7 +1777,8 @@ def get_node_details(node_id: int) -> dict[str, Any] | None: short_name, hw_model, role, - primary_channel + primary_channel, + mac_address FROM node_info WHERE node_id = ? """, @@ -1520,7 +1803,12 @@ def get_node_details(node_id: int) -> dict[str, Any] | None: "short_name": node_info_row["short_name"], "hw_model": node_info_row["hw_model"], "role": node_info_row["role"], - "primary_channel": node_info_row.get("primary_channel"), + "primary_channel": node_info_row["primary_channel"] + if "primary_channel" in node_info_row.keys() + else None, + "mac_address": node_info_row["mac_address"] + if "mac_address" in node_info_row.keys() + else None, "total_packets": 0, "last_seen": None, "first_seen": None, @@ -1562,6 +1850,9 @@ def get_node_details(node_id: int) -> dict[str, Any] | None: "primary_channel": node_info_row["primary_channel"] if node_info_row and "primary_channel" in node_info_row.keys() else None, + "mac_address": node_info_row["mac_address"] + if node_info_row and "mac_address" in node_info_row.keys() + else None, "total_packets": node_stats_row["total_packets"], "last_seen": last_seen.strftime("%Y-%m-%d %H:%M:%S UTC"), "last_seen_timestamp": last_seen.timestamp(), # Raw Unix timestamp for client-side formatting diff --git a/src/malla/database/schema.py b/src/malla/database/schema.py index c84fbfcd..4d6716bf 100644 --- a/src/malla/database/schema.py +++ b/src/malla/database/schema.py @@ -3,6 +3,8 @@ import logging import sqlite3 +from ..fingerprint import ensure_node_fingerprint_table + logger = logging.getLogger(__name__) @@ -128,6 +130,16 @@ # time), so each (tz_offset, day) is computed once and reused forever; only the # current day is recomputed live. Without this, the "All" range re-aggregates # the entire multi-million-row packet_history on every view. +# Small key/value store for one-off maintenance state (e.g. "the firmware +# fingerprint backfill already covered history up to time T"). +MALLA_META_TABLE_SQL = """ + CREATE TABLE IF NOT EXISTS malla_meta ( + key TEXT PRIMARY KEY, + value TEXT, + updated_at REAL NOT NULL + ) +""" + ACTIVITY_ROLLUP_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS activity_daily_rollup ( tz_offset_minutes INTEGER NOT NULL, @@ -207,6 +219,9 @@ def ensure_startup_schema( existing_indexes = _get_existing_indexes(cursor) cursor.execute(ACTIVITY_ROLLUP_TABLE_SQL) + for column in ensure_node_fingerprint_table(cursor): + logger.info("Added %s column to node_fingerprint table", column) + cursor.execute(MALLA_META_TABLE_SQL) if "node_info" in existing_tables: cursor.execute("PRAGMA table_info(node_info)") @@ -214,6 +229,13 @@ def ensure_startup_schema( if "primary_channel" not in node_info_columns: cursor.execute("ALTER TABLE node_info ADD COLUMN primary_channel TEXT") logger.info("Added primary_channel column to node_info table") + # Same-device lookups (firmware 2.8 re-keys node numbers) go through + # the MAC; very old databases predate the column, so guard on it. + if "mac_address" in node_info_columns: + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_node_mac_address ON node_info(mac_address) " + "WHERE mac_address IS NOT NULL AND mac_address != ''" + ) for index_name, table_name, sql in INDEX_SPECS: if table_name not in existing_tables or index_name in existing_indexes: diff --git a/src/malla/fingerprint.py b/src/malla/fingerprint.py new file mode 100644 index 00000000..cdf51372 --- /dev/null +++ b/src/malla/fingerprint.py @@ -0,0 +1,730 @@ +"""Firmware fingerprinting for Meshtastic nodes. + +Nodes never broadcast their firmware version, but each firmware generation +leaves traces in what it sends. This module turns those traces into a +version *band* ("≥ 2.8", "2.6.0 – 2.6.7", …). Markers, each verified +against the protobuf tag in which it first appeared: + +* ``hop_start`` set at all → 2.3+ +* NodeInfo ``User.public_key`` → 2.5.0+ +* own transmissions carry ``relay_node`` → 2.6.0+ (next-hop routing) +* NodeInfo ``User.is_unmessagable`` present → 2.6.8+ (firmware always + sets ``has_is_unmessagable`` once the field exists) +* role ``CLIENT_BASE`` → 2.7.8+ +* node number == CRC32(public key) → 2.8.0+ (key-derived ids) +* ``Data.xeddsa_signature`` (64 bytes) on packets → 2.8.0+ (packet signing) +* AES-CCM (``use_aead``) channel traffic, ``PAGING_APP`` (38) or + ``Telemetry.soil_water_metrics`` (tag 11) → 2.8.1+ + +Positive evidence always wins over absence: an old gateway that strips +``relay_node`` cannot make a 2.6+ node look older, it can only leave it +unclassified. ``MAP_REPORT_APP`` packets carry an exact +``firmware_version`` and take precedence when they are at least as new as +what the fingerprint implies. + +Only pure functions live here; the capture daemon and the web app supply the +storage. +""" + +from __future__ import annotations + +import sqlite3 +import time +import zlib +from collections.abc import Iterable +from dataclasses import asdict, dataclass, fields +from typing import Any + +BROADCAST_NODE_ID = 0xFFFFFFFF +XEDDSA_SIGNATURE_LEN = 64 +DATA_XEDDSA_SIGNATURE_TAG = 10 +TELEMETRY_SOIL_WATER_TAG = 11 # Telemetry.soil_water_metrics, protobufs after v2.8.0 +PAGING_APP_PORTNUM = 38 # PortNum.PAGING_APP, protobufs after v2.8.0 + +# Evidence thresholds: a handful of first-hop packets is enough to prove the +# relay_node behaviour, while claiming its *absence* needs more packets. +MIN_RELAY_SELF_PACKETS = 3 +MIN_RELAY_NONE_PACKETS = 10 +MIN_SOFTWARE_CLIENT_PACKETS = 10 +# A MapReport older than this may predate an upgrade; the fingerprint decides. +REPORTED_VERSION_MAX_AGE_SECONDS = 180 * 86400 + +SOFTWARE_CLIENT_HW_MODELS = frozenset({"PRIVATE_HW"}) + + +@dataclass +class FirmwareEvidence: + """Accumulated per-node observations. Counters add, flags OR, mask ORs.""" + + nodeinfo_count: int = 0 + has_public_key: bool = False + has_unmessagable_field: bool = False + id_from_public_key: bool = False + mac_mismatch: bool = False + relay_self_count: int = 0 + relay_none_count: int = 0 + xeddsa_signed_count: int = 0 + hop_start_zero_count: int = 0 + hop_start_set_count: int = 0 + hop_start_mask: int = 0 + aead_count: int = 0 + soil_water_count: int = 0 + paging_count: int = 0 + firmware_version: str | None = None + firmware_version_at: float | None = None + + def is_empty(self) -> bool: + return not any(v for v in asdict(self).values()) + + def merge(self, other: FirmwareEvidence) -> None: + """Fold *other* into this evidence (same semantics as the SQL upsert).""" + for name in COUNTER_FIELDS: + setattr(self, name, getattr(self, name) + getattr(other, name)) + for name in FLAG_FIELDS: + setattr(self, name, getattr(self, name) or getattr(other, name)) + self.hop_start_mask |= other.hop_start_mask + if other.firmware_version and ( + self.firmware_version_at is None + or (other.firmware_version_at or 0) >= self.firmware_version_at + ): + self.firmware_version = other.firmware_version + self.firmware_version_at = other.firmware_version_at + + @classmethod + def from_row(cls, row: Any) -> FirmwareEvidence: + """Build evidence from a ``node_fingerprint`` row (dict or sqlite3.Row).""" + if row is None: + return cls() + keys = row.keys() if hasattr(row, "keys") else () + values: dict[str, Any] = {} + for f in fields(cls): + if f.name not in keys: + continue + value = row[f.name] + if f.type == "bool": + value = bool(value) + elif f.type == "int": + value = int(value or 0) + values[f.name] = value + return cls(**values) + + +EVIDENCE_COLUMNS: tuple[str, ...] = tuple(f.name for f in fields(FirmwareEvidence)) + +COUNTER_FIELDS = ( + "nodeinfo_count", + "relay_self_count", + "relay_none_count", + "xeddsa_signed_count", + "hop_start_zero_count", + "hop_start_set_count", + "aead_count", + "soil_water_count", + "paging_count", +) +FLAG_FIELDS = ( + "has_public_key", + "has_unmessagable_field", + "id_from_public_key", + "mac_mismatch", +) + +NODE_FINGERPRINT_TABLE_SQL = """ + CREATE TABLE IF NOT EXISTS node_fingerprint ( + node_id INTEGER PRIMARY KEY, + nodeinfo_count INTEGER NOT NULL DEFAULT 0, + has_public_key INTEGER NOT NULL DEFAULT 0, + has_unmessagable_field INTEGER NOT NULL DEFAULT 0, + id_from_public_key INTEGER NOT NULL DEFAULT 0, + mac_mismatch INTEGER NOT NULL DEFAULT 0, + relay_self_count INTEGER NOT NULL DEFAULT 0, + relay_none_count INTEGER NOT NULL DEFAULT 0, + xeddsa_signed_count INTEGER NOT NULL DEFAULT 0, + hop_start_zero_count INTEGER NOT NULL DEFAULT 0, + hop_start_set_count INTEGER NOT NULL DEFAULT 0, + hop_start_mask INTEGER NOT NULL DEFAULT 0, + aead_count INTEGER NOT NULL DEFAULT 0, + soil_water_count INTEGER NOT NULL DEFAULT 0, + paging_count INTEGER NOT NULL DEFAULT 0, + firmware_version TEXT, + firmware_version_at REAL, + updated_at REAL NOT NULL + ) +""" + +# Merge-upsert: the capture daemon flushes *deltas*, so counters add up, +# flags stick once seen and the newest MapReport wins. +NODE_FINGERPRINT_UPSERT_SQL = """ + INSERT INTO node_fingerprint ( + node_id, nodeinfo_count, has_public_key, has_unmessagable_field, + id_from_public_key, mac_mismatch, relay_self_count, relay_none_count, + xeddsa_signed_count, hop_start_zero_count, hop_start_set_count, + hop_start_mask, aead_count, soil_water_count, paging_count, + firmware_version, firmware_version_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(node_id) DO UPDATE SET + nodeinfo_count = nodeinfo_count + excluded.nodeinfo_count, + has_public_key = MAX(has_public_key, excluded.has_public_key), + has_unmessagable_field = MAX(has_unmessagable_field, excluded.has_unmessagable_field), + id_from_public_key = MAX(id_from_public_key, excluded.id_from_public_key), + mac_mismatch = MAX(mac_mismatch, excluded.mac_mismatch), + relay_self_count = relay_self_count + excluded.relay_self_count, + relay_none_count = relay_none_count + excluded.relay_none_count, + xeddsa_signed_count = xeddsa_signed_count + excluded.xeddsa_signed_count, + hop_start_zero_count = hop_start_zero_count + excluded.hop_start_zero_count, + hop_start_set_count = hop_start_set_count + excluded.hop_start_set_count, + hop_start_mask = hop_start_mask | excluded.hop_start_mask, + aead_count = aead_count + excluded.aead_count, + soil_water_count = soil_water_count + excluded.soil_water_count, + paging_count = paging_count + excluded.paging_count, + firmware_version = CASE + WHEN excluded.firmware_version IS NOT NULL + AND (firmware_version_at IS NULL + OR COALESCE(excluded.firmware_version_at, 0) >= firmware_version_at) + THEN excluded.firmware_version ELSE firmware_version END, + firmware_version_at = CASE + WHEN excluded.firmware_version IS NOT NULL + AND (firmware_version_at IS NULL + OR COALESCE(excluded.firmware_version_at, 0) >= firmware_version_at) + THEN excluded.firmware_version_at ELSE firmware_version_at END, + updated_at = excluded.updated_at +""" + + +def evidence_row( + node_id: int, ev: FirmwareEvidence, updated_at: float +) -> tuple[Any, ...]: + """Parameter tuple for :data:`NODE_FINGERPRINT_UPSERT_SQL`.""" + return ( + node_id, + ev.nodeinfo_count, + int(ev.has_public_key), + int(ev.has_unmessagable_field), + int(ev.id_from_public_key), + int(ev.mac_mismatch), + ev.relay_self_count, + ev.relay_none_count, + ev.xeddsa_signed_count, + ev.hop_start_zero_count, + ev.hop_start_set_count, + ev.hop_start_mask, + ev.aead_count, + ev.soil_water_count, + ev.paging_count, + ev.firmware_version, + ev.firmware_version_at, + updated_at, + ) + + +# Columns added after the table first shipped; older databases get them via +# ALTER TABLE so upgrades need no manual migration. +NODE_FINGERPRINT_ADDED_COLUMNS: tuple[tuple[str, str], ...] = ( + ("aead_count", "INTEGER NOT NULL DEFAULT 0"), + ("soil_water_count", "INTEGER NOT NULL DEFAULT 0"), + ("paging_count", "INTEGER NOT NULL DEFAULT 0"), +) + + +def ensure_node_fingerprint_table(cursor: sqlite3.Cursor) -> list[str]: + """Create ``node_fingerprint`` if missing and add any newer columns. + + Returns the names of the columns that were added. + """ + cursor.execute(NODE_FINGERPRINT_TABLE_SQL) + cursor.execute("PRAGMA table_info(node_fingerprint)") + present = {row[1] for row in cursor.fetchall()} + added: list[str] = [] + for name, decl in NODE_FINGERPRINT_ADDED_COLUMNS: + if name not in present: + cursor.execute(f"ALTER TABLE node_fingerprint ADD COLUMN {name} {decl}") + added.append(name) + return added + + +# --------------------------------------------------------------------------- +# Wire-level helpers +# --------------------------------------------------------------------------- + + +def relay_marker_for(node_id: int) -> int: + """The ``relay_node`` byte a node writes for itself. + + Firmware stores the last byte of the node number, but 0 means "unset", + so nodes ending in 0x00 write 0xFF instead. + """ + last = node_id & 0xFF + return last if last else 0xFF + + +def _read_varint(buf: bytes, pos: int) -> tuple[int, int]: + result = shift = 0 + while True: + if pos >= len(buf): + raise ValueError("truncated varint") + byte = buf[pos] + pos += 1 + result |= (byte & 0x7F) << shift + shift += 7 + if not byte & 0x80: + return result, pos + if shift > 63: + raise ValueError("varint too long") + + +def _has_length_delimited_field( + buf: bytes | None, wanted_tag: int, wanted_length: int | None = None +) -> bool: + """True if a serialized protobuf message carries ``wanted_tag`` as a + length-delimited field (optionally with exactly ``wanted_length`` bytes). + + Works on the raw wire format so it does not depend on the installed + protobuf definitions knowing the field. + """ + if not buf: + return False + try: + pos = 0 + end = len(buf) + while pos < end: + key, pos = _read_varint(buf, pos) + tag, wire_type = key >> 3, key & 7 + if wire_type == 0: + _, pos = _read_varint(buf, pos) + elif wire_type == 1: + pos += 8 + elif wire_type == 2: + length, pos = _read_varint(buf, pos) + if tag == wanted_tag and ( + wanted_length is None or length == wanted_length + ): + return True + pos += length + elif wire_type == 5: + pos += 4 + else: + return False + return False + except (ValueError, IndexError): + return False + + +def data_has_xeddsa_signature(data_bytes: bytes | None) -> bool: + """True if a serialized ``Data`` carries a 64-byte XEdDSA signature (2.8+).""" + return _has_length_delimited_field( + data_bytes, DATA_XEDDSA_SIGNATURE_TAG, XEDDSA_SIGNATURE_LEN + ) + + +def telemetry_has_soil_water(telemetry_bytes: bytes | None) -> bool: + """True if a serialized ``Telemetry`` carries ``soil_water_metrics`` (2.8.1+).""" + return _has_length_delimited_field(telemetry_bytes, TELEMETRY_SOIL_WATER_TAG) + + +def _user_has_unmessagable_field(user: Any) -> bool: + descriptor = getattr(user, "DESCRIPTOR", None) + if descriptor is None: + return False + field = descriptor.fields_by_name.get("is_unmessagable") + if field is None or not getattr(field, "has_presence", False): + return False + try: + return bool(user.HasField("is_unmessagable")) + except ValueError: + return False + + +# --------------------------------------------------------------------------- +# Observers +# --------------------------------------------------------------------------- + + +def observe_packet( + ev: FirmwareEvidence, + node_id: int, + *, + to_node_id: int | None, + hop_start: int | None, + hop_limit: int | None, + relay_node: int | None, + data_bytes: bytes | None = None, + portnum: int | None = None, + telemetry_bytes: bytes | None = None, + aead: bool = False, +) -> None: + """Record what one packet *sent by* ``node_id`` reveals. + + ``data_bytes`` is the serialized ``Data`` (for the 2.8 signature), + ``telemetry_bytes`` the serialized ``Telemetry`` payload of a + TELEMETRY_APP packet, and ``aead`` whether the packet was decrypted + with AES-CCM (a 2.8.1 channel setting). + """ + hop_start = hop_start or 0 + if hop_start <= 0: + ev.hop_start_zero_count += 1 + else: + ev.hop_start_set_count += 1 + if to_node_id == BROADCAST_NODE_ID and hop_start <= 7: + ev.hop_start_mask |= 1 << hop_start + # First hop as heard by the gateway: relay_node is whatever the + # sender wrote (2.6+) or nothing at all (older firmware). + if hop_limit == hop_start: + if relay_node == relay_marker_for(node_id): + ev.relay_self_count += 1 + elif not relay_node: + ev.relay_none_count += 1 + if data_bytes and data_has_xeddsa_signature(data_bytes): + ev.xeddsa_signed_count += 1 + if aead: + ev.aead_count += 1 + if portnum == PAGING_APP_PORTNUM: + ev.paging_count += 1 + if telemetry_bytes and telemetry_has_soil_water(telemetry_bytes): + ev.soil_water_count += 1 + + +def observe_user(ev: FirmwareEvidence, node_id: int, user: Any) -> None: + """Record what a NodeInfo ``User`` payload from ``node_id`` reveals.""" + ev.nodeinfo_count += 1 + public_key = bytes(getattr(user, "public_key", b"") or b"") + if public_key: + ev.has_public_key = True + if (zlib.crc32(public_key) & 0xFFFFFFFF) == node_id: + ev.id_from_public_key = True + if _user_has_unmessagable_field(user): + ev.has_unmessagable_field = True + macaddr = bytes(getattr(user, "macaddr", b"") or b"") + if len(macaddr) >= 4 and any(macaddr): + if int.from_bytes(macaddr[-4:], "big") != node_id: + ev.mac_mismatch = True + + +def observe_map_report( + ev: FirmwareEvidence, report: Any, observed_at: float | None = None +) -> None: + """Record the exact firmware version a ``MapReport`` payload declares.""" + version = str(getattr(report, "firmware_version", "") or "").strip() + if not version: + return + ev.firmware_version = version[:40] + ev.firmware_version_at = observed_at if observed_at is not None else time.time() + + +# --------------------------------------------------------------------------- +# Estimation +# --------------------------------------------------------------------------- + + +def parse_version(version: str | None) -> tuple[int, int, int] | None: + """``"2.7.15.567b8ea"`` → ``(2, 7, 15)``; None if unparseable.""" + if not version: + return None + parts = version.strip().split(".") + nums: list[int] = [] + for part in parts[:3]: + digits = "" + for ch in part: + if ch.isdigit(): + digits += ch + else: + break + if not digits: + break + nums.append(int(digits)) + if len(nums) < 2: + return None + while len(nums) < 3: + nums.append(0) + return nums[0], nums[1], nums[2] + + +class _Trail: + """Collects, per finding, a full sentence plus a terse (trait, range) pair.""" + + def __init__(self) -> None: + self.reasons: list[str] = [] + self.items: list[dict[str, str]] = [] + + def add(self, reason: str, trait: str, version_range: str) -> None: + self.reasons.append(reason) + self.items.append({"trait": trait, "range": version_range}) + + +def _band(ev: FirmwareEvidence, role: str | None, hw_model: str | None): + """Return (label, band_min, trail) from fingerprint evidence alone.""" + t = _Trail() + if ( + hw_model in SOFTWARE_CLIENT_HW_MODELS + and ev.hop_start_set_count == 0 + and ev.hop_start_zero_count >= MIN_SOFTWARE_CLIENT_PACKETS + ): + t.add( + "PRIVATE_HW and hop_start never set: not a Meshtastic radio", + "PRIVATE_HW, no hop_start", + "not a radio", + ) + return "Software client", None, t + + if ev.aead_count > 0: + t.add( + f"{ev.aead_count} packet(s) used AES-CCM channel encryption (use_aead, 2.8.1+)", + f"AEAD channel ×{ev.aead_count}", + "2.8.1+", + ) + if ev.paging_count > 0: + t.add( + f"{ev.paging_count} PAGING_APP packet(s) (portnum added in 2.8.1)", + f"paging app ×{ev.paging_count}", + "2.8.1+", + ) + if ev.soil_water_count > 0: + t.add( + f"{ev.soil_water_count} soil/water telemetry packet(s) (2.8.1+)", + f"soil/water telemetry ×{ev.soil_water_count}", + "2.8.1+", + ) + newest = "≥ 2.8.1" if t.items else None + + if ev.xeddsa_signed_count > 0: + t.add( + f"{ev.xeddsa_signed_count} packet(s) carry an XEdDSA signature (2.8+)", + f"XEdDSA signature ×{ev.xeddsa_signed_count}", + "2.8+", + ) + if ev.id_from_public_key: + t.add( + "node number is CRC32 of its public key (2.8+)", + "node id = CRC32(public key)", + "2.8+", + ) + if newest: + return newest, (2, 8, 1), t + if t.items: + return "≥ 2.8", (2, 8, 0), t + + if role == "CLIENT_BASE": + t.add("role CLIENT_BASE exists since 2.7.8", "CLIENT_BASE role", "2.7.8+") + return "≥ 2.7.8", (2, 7, 8), t + + if ev.has_unmessagable_field: + t.add( + "NodeInfo carries is_unmessagable (field added in 2.6.8)", + "is_unmessagable field", + "2.6.8+", + ) + return "≥ 2.6.8", (2, 6, 8), t + + if ev.relay_self_count >= MIN_RELAY_SELF_PACKETS: + t.add( + f"{ev.relay_self_count} own packets carry relay_node (next-hop routing, 2.6+)", + f"relay_node ×{ev.relay_self_count}", + "2.6+", + ) + if ev.nodeinfo_count: + t.add( + "NodeInfo lacks is_unmessagable, so older than 2.6.8", + "no is_unmessagable", + "< 2.6.8", + ) + return "2.6.0 – 2.6.7", (2, 6, 0), t + return "≥ 2.6", (2, 6, 0), t + + if ev.has_public_key: + t.add("NodeInfo carries a public key (PKI, 2.5+)", "public key", "2.5+") + if ev.relay_none_count >= MIN_RELAY_NONE_PACKETS and not ev.relay_self_count: + t.add( + f"{ev.relay_none_count} own packets without relay_node, so older than 2.6", + f"no relay_node ×{ev.relay_none_count}", + "< 2.6", + ) + return "2.5.x", (2, 5, 0), t + return "2.5.0 – 2.6.7", (2, 5, 0), t + + if ev.nodeinfo_count: + t.add("NodeInfo has no public key", "no public key", "< 2.5 or PKI off") + if ev.relay_none_count >= MIN_RELAY_NONE_PACKETS and not ev.relay_self_count: + t.add( + f"{ev.relay_none_count} own packets without relay_node, so older than 2.6", + f"no relay_node ×{ev.relay_none_count}", + "< 2.6", + ) + return "< 2.5", (0, 0, 0), t + return "< 2.5 or PKI disabled", (0, 0, 0), t + + return None, None, t + + +def estimate( + ev: FirmwareEvidence, + *, + role: str | None = None, + hw_model: str | None = None, + now: float | None = None, +) -> dict[str, Any]: + """Combine fingerprint evidence and any reported version into one verdict. + + Returns ``label`` (what to show), ``source`` (``reported`` / + ``fingerprint`` / ``none``), the fingerprint-only ``estimated_label``, + the ``reported_version`` with its timestamp, full-sentence ``reasons`` + (for APIs and logs), ``evidence_items`` as ``{trait, range}`` pairs and + ``evidence`` strings ("trait (range)") for terse UIs. + """ + now = time.time() if now is None else now + label, band_min, trail = _band(ev, role, hw_model) + estimated_label = label + + if bin(ev.hop_start_mask).count("1") >= 3: + trail.add( + "broadcast hop limit keeps changing (automatic hop limits, a 2.8 trait)", + "variable hop limit", + "2.8 trait", + ) + + reported = ev.firmware_version + reported_tuple = parse_version(reported) + reported_at = ev.firmware_version_at + reported_at_str = ( + time.strftime("%Y-%m-%d", time.gmtime(reported_at)) if reported_at else None + ) + reported_fresh = reported_at is None or ( + now - reported_at <= REPORTED_VERSION_MAX_AGE_SECONDS + ) + + items = list(trail.items) + if reported and reported_tuple and label != "Software client": + report_item = { + "trait": "MapReport" + (f" {reported_at_str}" if reported_at_str else ""), + "range": "exact" if reported_fresh else "exact, old report", + } + if band_min is None or reported_tuple >= band_min: + label, source = reported, "reported" + reasons = [ + "version reported by the node itself (MapReport)" + + ("" if reported_fresh else ", but the report is old") + ] + trail.reasons + items = [report_item] + items + else: + source = "fingerprint" + reasons = trail.reasons + [ + f"last reported {reported}, but newer traits have been seen since" + ] + items = items + [ + {"trait": f"reported {reported} earlier", "range": "superseded"} + ] + else: + source = "fingerprint" if label else "none" + reasons = trail.reasons + if not items: + items = [{"trait": "no evidence yet", "range": ""}] + + return { + "label": label or "Unknown", + "source": source, + "estimated_label": estimated_label, + "reported_version": reported, + "reported_at": reported_at, + "reported_at_str": reported_at_str, + "reasons": reasons, + "evidence_items": items, + "evidence": [ + f"{item['trait']} ({item['range']})" if item["range"] else item["trait"] + for item in items + ], + } + + +# --------------------------------------------------------------------------- +# Distribution (dashboard) +# --------------------------------------------------------------------------- + +Version = tuple[int, int, int] + +# Fingerprints prove lower bounds ("at least 2.7.8"), and a node with weaker +# evidence cannot be told apart from an older one, so the dashboard buckets +# nodes by the highest version they are proven to have reached. That is a +# strict ladder: every node sits on exactly one rung and the rungs never +# overlap. Exact (MapReport) versions land on the same rungs. +_LADDER: tuple[tuple[str, Version], ...] = ( + ("2.8.1", (2, 8, 1)), + ("2.8", (2, 8, 0)), + ("2.7.8", (2, 7, 8)), + ("2.6.8", (2, 6, 8)), + ("2.6", (2, 6, 0)), + ("2.5", (2, 5, 0)), +) +_TAIL_SEGMENTS: tuple[tuple[str, str], ...] = ( + ("lt2.5", "< 2.5"), + ("software", "Software client"), + ("unknown", "Unknown"), +) +DISTRIBUTION_KEYS: tuple[str, ...] = tuple(k for k, _ in _LADDER) + tuple( + k for k, _ in _TAIL_SEGMENTS +) + +_ESTIMATE_TO_SEGMENT: dict[str, str] = { + "≥ 2.8.1": "2.8.1", + "≥ 2.8": "2.8", + "≥ 2.7.8": "2.7.8", + "≥ 2.6.8": "2.6.8", + "≥ 2.6": "2.6", + "2.6.0 – 2.6.7": "2.6", + "2.5.0 – 2.6.7": "2.5", + "2.5.x": "2.5", + "< 2.5": "lt2.5", + "< 2.5 or PKI disabled": "lt2.5", + "Software client": "software", +} + + +def _fmt(version: Version) -> str: + return ".".join(str(n) for n in version) + + +def segment_label(key: str) -> str: + """Human label for a segment key, e.g. ``"≥ 2.7.8"``.""" + for seg_key, low in _LADDER: + if seg_key == key: + return f"≥ {_fmt(low)}" + for seg_key, label in _TAIL_SEGMENTS: + if seg_key == key: + return label + return key + + +def distribution_segment(verdict: dict[str, Any]) -> tuple[str, str]: + """Map one verdict to ``(segment key, kind)``; kind is reported/estimated/other.""" + if verdict.get("source") == "reported": + parsed = parse_version(verdict.get("reported_version")) + if parsed: + for key, low in _LADDER: + if parsed >= low: + return key, "reported" + return "lt2.5", "reported" + label = verdict.get("label") or "Unknown" + segment = _ESTIMATE_TO_SEGMENT.get(label, "unknown") + if segment in ("software", "unknown"): + return segment, "other" + return segment, "estimated" + + +def summarize_distribution( + verdicts: Iterable[dict[str, Any]], +) -> list[dict[str, Any]]: + """Count verdicts per rung (newest first) with reported/estimated splits. + + Empty rungs are dropped. + """ + counts: dict[str, dict[str, int]] = { + key: {"count": 0, "reported": 0, "estimated": 0, "other": 0} + for key in DISTRIBUTION_KEYS + } + for verdict in verdicts: + segment, kind = distribution_segment(verdict) + entry = counts[segment] + entry["count"] += 1 + entry[kind] += 1 + return [ + {"key": key, "label": segment_label(key), **counts[key]} + for key in DISTRIBUTION_KEYS + if counts[key]["count"] + ] diff --git a/src/malla/fingerprint_backfill.py b/src/malla/fingerprint_backfill.py new file mode 100644 index 00000000..2ea19c29 --- /dev/null +++ b/src/malla/fingerprint_backfill.py @@ -0,0 +1,346 @@ +"""Backfill ``node_fingerprint`` from packets already stored in the database. + +The capture daemon accumulates firmware evidence as packets arrive and, on +its first start after this feature is installed, runs this backfill in the +background for the configured number of days (``fingerprint_backfill_days``). +A marker in ``malla_meta`` records the time the backfill covered, so later +restarts skip it and the daemon and the backfill never count a packet twice. + +The CLI exists for operators who want a longer window or a re-run:: + + malla-fingerprint-backfill --days 90 --until 1789000000 +""" + +from __future__ import annotations + +import argparse +import logging +import sqlite3 +import sys +import time +from collections.abc import Iterable +from contextlib import nullcontext +from typing import Any + +from meshtastic import mesh_pb2, mqtt_pb2 + +from .config import get_config +from .database.schema import MALLA_META_TABLE_SQL +from .fingerprint import ( + BROADCAST_NODE_ID, + NODE_FINGERPRINT_UPSERT_SQL, + FirmwareEvidence, + ensure_node_fingerprint_table, + evidence_row, + observe_map_report, + observe_packet, + observe_user, +) + +logger = logging.getLogger(__name__) + +BACKFILL_MARKER_KEY = "fingerprint_backfill_until" + + +def read_backfill_marker(db_path: str) -> float | None: + """Timestamp the last completed backfill covered, or None if never run.""" + try: + conn = sqlite3.connect(db_path, timeout=30) + try: + conn.execute(MALLA_META_TABLE_SQL) + row = conn.execute( + "SELECT value FROM malla_meta WHERE key = ?", (BACKFILL_MARKER_KEY,) + ).fetchone() + finally: + conn.close() + except sqlite3.Error as e: + logger.debug("Could not read backfill marker: %s", e) + return None + if not row or row[0] is None: + return None + try: + return float(row[0]) + except (TypeError, ValueError): + return None + + +# Only these apps can carry a signature or reveal an AEAD channel, so only +# their envelopes are worth parsing (and decrypting). +_SIGNABLE_PORTNUMS = frozenset( + { + "NODEINFO_APP", + "POSITION_APP", + "TELEMETRY_APP", + "TEXT_MESSAGE_APP", + "UNKNOWN_APP", + None, + } +) + + +def _decrypt_candidates(channel_name: str | None, keys: Iterable[str]) -> list[bytes]: + from .mqtt_capture import derive_key_from_channel_name + + candidates: list[bytes] = [] + for key in keys: + for name in ("", channel_name or ""): + try: + derived = derive_key_from_channel_name(name, key) + except Exception: # noqa: BLE001 + continue + if len(derived) in (16, 32) and derived not in candidates: + candidates.append(derived) + return candidates + + +def _looks_like_data(plain: bytes) -> bool: + data = mesh_pb2.Data() + try: + data.ParseFromString(plain) + except Exception: # noqa: BLE001 + return False + return bool(data.portnum) and data.portnum < 600 + + +def _data_bytes_from_envelope( + envelope_bytes: bytes | None, channel_name: str | None, keys: list[str] +) -> tuple[bytes | None, bool]: + """``(serialized Data, used_aead)`` for the packet in an envelope. + + Decrypts when needed, trying AES-CCM (2.8.1 ``use_aead`` channels) before + AES-CTR because the CCM tag rules out false positives. + """ + if not envelope_bytes: + return None, False + from .mqtt_capture import decrypt_packet, decrypt_packet_ccm + + envelope = mqtt_pb2.ServiceEnvelope() + envelope.ParseFromString(envelope_bytes) + packet = envelope.packet + if packet.HasField("decoded"): + return packet.decoded.SerializeToString(), False + if not packet.encrypted or getattr(packet, "pki_encrypted", False): + return None, False + sender = getattr(packet, "from") + encrypted = bytes(packet.encrypted) + for key in _decrypt_candidates(channel_name, keys): + plain = decrypt_packet_ccm(encrypted, packet.id, sender, packet.to, key) + if plain and _looks_like_data(plain): + return plain, True + plain = decrypt_packet(encrypted, packet.id, sender, key) + if plain and _looks_like_data(plain): + return plain, False + return None, False + + +def _log_label_histogram(db_path: str, evidence: dict[int, FirmwareEvidence]) -> None: + """Log how the evidence would be classified (a sanity check for operators).""" + from collections import Counter + + from .fingerprint import estimate + + meta: dict[int, tuple[str | None, str | None]] = {} + try: + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=60) + try: + for node_id, role, hw_model in conn.execute( + "SELECT node_id, role, hw_model FROM node_info" + ): + meta[node_id] = (role, hw_model) + finally: + conn.close() + except sqlite3.Error as e: # pragma: no cover - informational only + logger.debug("node_info unavailable for histogram: %s", e) + histogram: Counter[str] = Counter() + for node_id, ev in evidence.items(): + role, hw_model = meta.get(node_id, (None, None)) + verdict = estimate(ev, role=role, hw_model=hw_model) + key = ( + verdict["label"] if verdict["source"] != "reported" else "reported (exact)" + ) + histogram[key] += 1 + for label, count in sorted(histogram.items(), key=lambda kv: -kv[1]): + logger.info(" %5d %s", count, label) + + +def backfill( + db_path: str, + *, + days: float, + until: float, + keys: list[str], + dry_run: bool = False, + progress_every: int = 200_000, + lock: Any = None, +) -> dict[str, int]: + """Compute evidence for ``[until - days, until)`` and merge it into the DB. + + ``lock`` (a context manager such as the capture daemon's ``db_lock``) is + held only for the final write, which also records the completion marker + in the same transaction. + """ + since = until - days * 86400 + evidence: dict[int, FirmwareEvidence] = {} + stats = {"packets": 0, "signatures": 0, "map_reports": 0, "nodes": 0} + + read = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=60) + read.row_factory = sqlite3.Row + try: + logger.info("Scanning MapReport packets (all time)") + for row in read.execute( + """ + SELECT from_node_id, timestamp, raw_payload FROM packet_history + WHERE portnum_name = 'MAP_REPORT_APP' AND raw_payload IS NOT NULL + AND from_node_id IS NOT NULL AND timestamp < ? + ORDER BY timestamp + """, + (until,), + ): + try: + report = mqtt_pb2.MapReport() + report.ParseFromString(bytes(row["raw_payload"])) + except Exception: # noqa: BLE001 + continue + ev = evidence.setdefault(row["from_node_id"], FirmwareEvidence()) + observe_map_report(ev, report, row["timestamp"]) + stats["map_reports"] += 1 + + logger.info( + "Scanning packets from %s to %s", + time.strftime("%Y-%m-%d %H:%M", time.gmtime(since)), + time.strftime("%Y-%m-%d %H:%M", time.gmtime(until)), + ) + # Older databases may predate some columns; select what exists. + present = {r[1] for r in read.execute("PRAGMA table_info(packet_history)")} + wanted = ( + "from_node_id", + "to_node_id", + "hop_start", + "hop_limit", + "relay_node", + "portnum", + "portnum_name", + "raw_payload", + "raw_service_envelope", + "channel_id", + ) + select_list = ", ".join( + col if col in present else f"NULL AS {col}" for col in wanted + ) + cursor = read.execute( + f""" + SELECT {select_list} + FROM packet_history + WHERE timestamp >= ? AND timestamp < ? AND from_node_id IS NOT NULL + ORDER BY id + """, + (since, until), + ) + for row in cursor: + stats["packets"] += 1 + node_id = row["from_node_id"] + if node_id <= 0 or node_id == BROADCAST_NODE_ID: + continue + ev = evidence.setdefault(node_id, FirmwareEvidence()) + data_bytes = None + used_aead = False + if row["portnum_name"] in _SIGNABLE_PORTNUMS: + try: + data_bytes, used_aead = _data_bytes_from_envelope( + row["raw_service_envelope"], row["channel_id"], keys + ) + except Exception: # noqa: BLE001 + data_bytes, used_aead = None, False + before = ev.xeddsa_signed_count + observe_packet( + ev, + node_id, + to_node_id=row["to_node_id"], + hop_start=row["hop_start"], + hop_limit=row["hop_limit"], + relay_node=row["relay_node"], + data_bytes=data_bytes, + portnum=row["portnum"], + telemetry_bytes=( + bytes(row["raw_payload"]) + if row["portnum_name"] == "TELEMETRY_APP" and row["raw_payload"] + else None + ), + aead=used_aead, + ) + stats["signatures"] += ev.xeddsa_signed_count - before + if row["portnum_name"] == "NODEINFO_APP" and row["raw_payload"]: + try: + user = mesh_pb2.User() + user.ParseFromString(bytes(row["raw_payload"])) + observe_user(ev, node_id, user) + except Exception: # noqa: BLE001 + pass + if stats["packets"] % progress_every == 0: + logger.info("… %s packets scanned", stats["packets"]) + finally: + read.close() + + stats["nodes"] = len(evidence) + _log_label_histogram(db_path, evidence) + if dry_run: + logger.info("Dry run: %s", stats) + return stats + + now = time.time() + rows = [evidence_row(nid, ev, now) for nid, ev in evidence.items()] + with lock if lock is not None else nullcontext(): + write = sqlite3.connect(db_path, timeout=60) + try: + write.execute("PRAGMA busy_timeout=30000") + ensure_node_fingerprint_table(write.cursor()) + write.execute(MALLA_META_TABLE_SQL) + write.executemany(NODE_FINGERPRINT_UPSERT_SQL, rows) + write.execute( + "INSERT INTO malla_meta (key, value, updated_at) VALUES (?, ?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value, " + "updated_at = excluded.updated_at", + (BACKFILL_MARKER_KEY, repr(float(until)), now), + ) + write.commit() + finally: + write.close() + logger.info("Backfill done: %s", stats) + return stats + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=(__doc__ or "").split("\n\n")[0]) + parser.add_argument( + "--days", type=float, default=30, help="window length (default 30)" + ) + parser.add_argument( + "--until", + type=float, + default=None, + help="end of window as a Unix timestamp (default: now)", + ) + parser.add_argument( + "--database", default=None, help="override the configured DB path" + ) + parser.add_argument("--dry-run", action="store_true", help="scan but do not write") + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + cfg = get_config() + db_path = args.database or cfg.database_file + stats = backfill( + db_path, + days=args.days, + until=args.until or time.time(), + keys=cfg.get_decryption_keys(), + dry_run=args.dry_run, + ) + print(stats) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/malla/mqtt_capture.py b/src/malla/mqtt_capture.py index a9831877..f2d41ba1 100644 --- a/src/malla/mqtt_capture.py +++ b/src/malla/mqtt_capture.py @@ -43,6 +43,7 @@ import paho.mqtt.client as mqtt from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.ciphers.aead import AESCCM from meshtastic import ( config_pb2, mesh_pb2, @@ -59,6 +60,14 @@ from .database.connection import seed_query_planner_stats_async from .database.schema import ensure_startup_schema +from .fingerprint import ( + NODE_FINGERPRINT_UPSERT_SQL, + FirmwareEvidence, + evidence_row, + observe_map_report, + observe_packet, + observe_user, +) # Load the singleton configuration once at module import time. This ensures the # capture tool honours the same YAML + optional environment override mechanism @@ -84,6 +93,7 @@ # Data retention settings DATA_RETENTION_HOURS: int = _cfg.data_retention_hours +FINGERPRINT_BACKFILL_DAYS: int = int(_cfg.fingerprint_backfill_days or 0) # Logging configuration – falls back to INFO if an invalid level was supplied LOG_LEVEL = _cfg.log_level.upper() @@ -212,8 +222,57 @@ def decrypt_packet( return b"" +# Firmware 2.8.1 channels with ``use_aead`` encrypt with AES-CCM instead of +# AES-CTR: 12-byte tag appended to the ciphertext, 13-byte nonce +# [packet id (8, LE)][from node (4, LE)][0], and the sender/destination ids as +# associated data. The tag makes a wrong key fail loudly, so it is safe to try +# CCM before CTR on every packet. +AEAD_TAG_SIZE = 12 + + +def decrypt_packet_ccm( + encrypted_payload: bytes, + packet_id: int, + sender_id: int, + dest_id: int, + key: bytes, +) -> bytes: + """Decrypt an AES-CCM (``use_aead``) payload; empty bytes if the tag fails.""" + if len(encrypted_payload) <= AEAD_TAG_SIZE or len(key) not in (16, 32): + return b"" + try: + nonce = ( + (packet_id & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") + + (sender_id & 0xFFFFFFFF).to_bytes(4, "little") + + b"\x00" + ) + aad = (sender_id & 0xFFFFFFFF).to_bytes(4, "little") + ( + dest_id & 0xFFFFFFFF + ).to_bytes(4, "little") + return AESCCM(key, tag_length=AEAD_TAG_SIZE).decrypt( + nonce, bytes(encrypted_payload), aad + ) + except Exception: # noqa: BLE001 - InvalidTag or bad lengths: not our key + return b"" + + +def _parse_decrypted_data(decrypted_payload: bytes) -> Any | None: + """Parse plaintext as ``Data``; None if it does not look like one.""" + try: + decoded_data = mesh_pb2.Data() + decoded_data.ParseFromString(decrypted_payload) + except Exception: # noqa: BLE001 + return None + if decoded_data.portnum == portnums_pb2.PortNum.UNKNOWN_APP: + return None + return decoded_data + + def try_decrypt_mesh_packet( - mesh_packet: Any, channel_name: str = "", keys_base64: list[str] | None = None + mesh_packet: Any, + channel_name: str = "", + keys_base64: list[str] | None = None, + info: dict[str, Any] | None = None, ) -> bool: """ Try to decrypt an encrypted MeshPacket and update it with decoded content. @@ -224,6 +283,8 @@ def try_decrypt_mesh_packet( mesh_packet: The MeshPacket protobuf object channel_name: Channel name for key derivation (empty for primary channel) keys_base64: List of base64-encoded encryption keys to try (uses DECRYPTION_KEYS if None) + info: Optional dict that receives ``method`` ("ccm" for AES-CCM + ``use_aead`` channels, "ctr" otherwise) on success Returns: bool: True if decryption was successful and packet was updated @@ -264,43 +325,42 @@ def try_decrypt_mesh_packet( # Derive the decryption key key = derive_key_from_channel_name(channel_name, key_base64) - # Decrypt the payload - decrypted_payload = decrypt_packet( - encrypted_payload, packet_id, sender_id, key + # AES-CCM first: it authenticates, so a hit is never a false + # positive and a miss costs one small AES operation. + dest_id = getattr(mesh_packet, "to", 0) + ccm_plain = decrypt_packet_ccm( + encrypted_payload, packet_id, sender_id, dest_id, key ) - - if not decrypted_payload: - logging.debug( - f"Decryption with key {key_index + 1} returned empty payload" + decoded_data = _parse_decrypted_data(ccm_plain) if ccm_plain else None + method = "ccm" + if decoded_data is None: + decrypted_payload = decrypt_packet( + encrypted_payload, packet_id, sender_id, key ) - continue - - # Try to parse the decrypted payload as a Data protobuf - try: - decoded_data = mesh_pb2.Data() - decoded_data.ParseFromString(decrypted_payload) - - # Validate that we got a valid portnum (not UNKNOWN_APP) - if decoded_data.portnum == portnums_pb2.PortNum.UNKNOWN_APP: + if not decrypted_payload: logging.debug( - f"Key {key_index + 1} produced UNKNOWN_APP portnum, trying next key" + f"Decryption with key {key_index + 1} returned empty payload" ) continue - - # Update the mesh packet with decoded data - mesh_packet.decoded.CopyFrom(decoded_data) - - logging.info( - f"✅ Successfully decrypted packet {packet_id} from {sender_id} with key {key_index + 1}/{len(keys_to_try)}: {portnums_pb2.PortNum.Name(decoded_data.portnum)}" - ) - return True - - except Exception as parse_error: + decoded_data = _parse_decrypted_data(decrypted_payload) + method = "ctr" + if decoded_data is None: logging.debug( - f"Failed to parse decrypted payload with key {key_index + 1} as Data protobuf: {parse_error}" + f"Key {key_index + 1} did not yield a valid Data payload, trying next key" ) continue + # Update the mesh packet with decoded data + mesh_packet.decoded.CopyFrom(decoded_data) + if info is not None: + info["method"] = method + + logging.info( + f"✅ Successfully decrypted packet {packet_id} from {sender_id} with key {key_index + 1}/{len(keys_to_try)}" + f"{' (AEAD)' if method == 'ccm' else ''}: {get_enum_name(portnums_pb2.PortNum.DESCRIPTOR, decoded_data.portnum)}" + ) + return True + logging.debug( f"Failed to decrypt packet with any of the {len(keys_to_try)} provided keys" ) @@ -312,6 +372,33 @@ def try_decrypt_mesh_packet( # --- Database Functions --- +def rename_unknown_hardware_models(cursor: sqlite3.Cursor) -> int: + """Replace ``UNKNOWN_`` hw_model values that the current protobufs can name. + + ``get_enum_name`` falls back to ``UNKNOWN_`` when a node reports a + hardware model newer than the bundled protobufs. After a protobuf upgrade + those rows would only fix themselves on the node's next NodeInfo; this + names them right away. Returns the number of rows updated. + """ + cursor.execute( + "SELECT DISTINCT hw_model FROM node_info WHERE hw_model LIKE 'UNKNOWN\\_%' ESCAPE '\\'" + ) + renamed = 0 + for (hw_model,) in cursor.fetchall(): + suffix = hw_model[len("UNKNOWN_") :] + if not suffix.isdigit(): + continue + value = mesh_pb2.HardwareModel.DESCRIPTOR.values_by_number.get(int(suffix)) + if value is None or value.name == "UNSET": + continue + cursor.execute( + "UPDATE node_info SET hw_model = ? WHERE hw_model = ?", + (value.name, hw_model), + ) + renamed += cursor.rowcount + return renamed + + def init_database() -> None: """Initialize SQLite database with required tables.""" init_start = time.time() @@ -411,6 +498,17 @@ def init_database() -> None: ensure_startup_schema(cursor, drop_legacy_indexes=True) + # Hardware models captured before the bundled protobufs knew them were + # stored as UNKNOWN_; give them their name once it resolves. + try: + renamed = rename_unknown_hardware_models(cursor) + if renamed: + logging.info( + "Named %s node_info hardware models that were UNKNOWN_", renamed + ) + except Exception as e: # noqa: BLE001 + logging.warning(f"Could not rename UNKNOWN_ hardware models: {e}") + # Backfill primary_channel only when there are actually missing values. try: cursor.execute( @@ -688,6 +786,145 @@ def update_node_cache( conn.close() +# --------------------------------------------------------------------------- +# Firmware fingerprinting +# --------------------------------------------------------------------------- +# Every packet a node sends reveals a little about its firmware (see +# malla.fingerprint). Observations are accumulated in memory as deltas on the +# MQTT thread and merged into the node_fingerprint table by the main loop +# (once a minute, and on shutdown), so ingestion never pays an extra write. + +FINGERPRINT_FLUSH_INTERVAL_SECONDS: float = float( + os.environ.get("MALLA_FINGERPRINT_FLUSH_SECONDS", "30") +) +_fingerprint_pending: dict[int, FirmwareEvidence] = {} +_fingerprint_lock = threading.Lock() +_fingerprint_last_flush: float = time.time() + + +def record_fingerprint_packet( + service_envelope: Any, mesh_packet: Any, aead: bool = False +) -> None: + """Fold one parsed (and possibly decrypted) MeshPacket into the evidence.""" + try: + from_node_id = getattr(mesh_packet, "from", 0) or 0 + if from_node_id <= 0 or from_node_id == 0xFFFFFFFF: + return + decoded = mesh_packet.decoded + portnum = decoded.portnum + data_bytes = None + if portnum != portnums_pb2.PortNum.UNKNOWN_APP: + # Serialise the decoded Data so unknown fields (e.g. the 2.8 + # xeddsa_signature the bundled protobufs may not know) are kept. + data_bytes = decoded.SerializeToString() + + with _fingerprint_lock: + ev = _fingerprint_pending.setdefault(from_node_id, FirmwareEvidence()) + observe_packet( + ev, + from_node_id, + to_node_id=getattr(mesh_packet, "to", None), + hop_start=getattr(mesh_packet, "hop_start", None), + hop_limit=getattr(mesh_packet, "hop_limit", None), + relay_node=getattr(mesh_packet, "relay_node", None), + data_bytes=data_bytes, + portnum=portnum if data_bytes else None, + telemetry_bytes=decoded.payload + if portnum == portnums_pb2.PortNum.TELEMETRY_APP + else None, + aead=aead, + ) + if portnum == portnums_pb2.PortNum.NODEINFO_APP: + user = mesh_pb2.User() + user.ParseFromString(decoded.payload) + observe_user(ev, from_node_id, user) + elif portnum == portnums_pb2.PortNum.MAP_REPORT_APP: + report = mqtt_pb2.MapReport() + report.ParseFromString(decoded.payload) + observe_map_report(ev, report, time.time()) + except Exception as e: # noqa: BLE001 - fingerprinting must never break ingestion + logging.debug(f"Fingerprint observation skipped: {e}") + + +def _run_fingerprint_backfill(until: float) -> None: + from .fingerprint_backfill import backfill + + try: + stats = backfill( + DATABASE_FILE, + days=FINGERPRINT_BACKFILL_DAYS, + until=until, + keys=list(DECRYPTION_KEYS), + lock=db_lock, + ) + logging.info(f"Firmware fingerprint backfill finished: {stats}") + except Exception as e: # noqa: BLE001 + logging.warning(f"Firmware fingerprint backfill failed: {e}") + + +def start_fingerprint_backfill_if_needed() -> threading.Thread | None: + """First start after the feature is installed: fingerprint stored history. + + Runs in a daemon thread so ingestion starts immediately. The completion + marker written by the backfill makes later starts a no-op; an interrupted + run leaves no marker and simply repeats next time. + """ + from .fingerprint_backfill import read_backfill_marker + + if FINGERPRINT_BACKFILL_DAYS <= 0: + logging.info("Firmware fingerprint backfill disabled by configuration") + return None + marker = read_backfill_marker(DATABASE_FILE) + if marker is not None: + logging.debug( + "Firmware fingerprint backfill already covered history up to %s", + time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(marker)), + ) + return None + until = time.time() + logging.info( + f"Starting one-off firmware fingerprint backfill " + f"({FINGERPRINT_BACKFILL_DAYS} days of history) in the background" + ) + thread = threading.Thread( + target=_run_fingerprint_backfill, + args=(until,), + name="fingerprint-backfill", + daemon=True, + ) + thread.start() + return thread + + +def flush_fingerprints(force: bool = False) -> int: + """Merge pending evidence into node_fingerprint. Returns rows written.""" + global _fingerprint_last_flush + now = time.time() + if not force and now - _fingerprint_last_flush < FINGERPRINT_FLUSH_INTERVAL_SECONDS: + return 0 + with _fingerprint_lock: + rows = [evidence_row(nid, ev, now) for nid, ev in _fingerprint_pending.items()] + _fingerprint_pending.clear() + _fingerprint_last_flush = now + if not rows: + return 0 + try: + with db_lock: + conn = sqlite3.connect(DATABASE_FILE, timeout=30.0) + try: + conn.executemany(NODE_FINGERPRINT_UPSERT_SQL, rows) + conn.commit() + finally: + conn.close() + logging.debug(f"Flushed firmware fingerprints for {len(rows)} nodes") + return len(rows) + except Exception as e: # noqa: BLE001 + logging.warning( + f"Could not flush firmware fingerprints ({len(rows)} nodes): {e}" + ) + return 0 + + def hex_id_to_numeric(hex_id: str) -> int | None: """Convert hex node ID (like '!abcdef12') to numeric ID.""" if not hex_id or not isinstance(hex_id, str): @@ -839,7 +1076,9 @@ def log_packet_to_database( want_ack = getattr(mesh_packet, "want_ack", None) if mesh_packet else None priority = getattr(mesh_packet, "priority", None) if mesh_packet else None delayed = getattr(mesh_packet, "delayed", None) if mesh_packet else None - channel_index = getattr(mesh_packet, "channel_index", None) if mesh_packet else None + # MeshPacket.channel carries the channel hash on the wire (the field is + # named "channel", not "channel_index"; the latter was always NULL). + channel_index = getattr(mesh_packet, "channel", None) if mesh_packet else None rx_time = getattr(mesh_packet, "rx_time", None) if mesh_packet else None pki_encrypted = getattr(mesh_packet, "pki_encrypted", None) if mesh_packet else None next_hop = getattr(mesh_packet, "next_hop", None) if mesh_packet else None @@ -1155,6 +1394,7 @@ def on_message(client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage) -> Non and mesh_packet.encrypted ) + decryption_info: dict[str, Any] = {} if is_encrypted_packet: logging.debug( f"Attempting to decrypt UNKNOWN_APP packet {mesh_packet.id} from {from_node_id_numeric}" @@ -1177,7 +1417,7 @@ def on_message(client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage) -> Non # Try decryption with primary channel keys (most common case) decryption_successful = try_decrypt_mesh_packet( - mesh_packet, channel_name="" + mesh_packet, channel_name="", info=decryption_info ) # If primary channel decryption failed and we have a channel name, try with channel-specific keys @@ -1188,6 +1428,7 @@ def on_message(client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage) -> Non decryption_successful = try_decrypt_mesh_packet( mesh_packet, channel_name=channel_name, + info=decryption_info, ) if decryption_successful: @@ -1208,6 +1449,12 @@ def on_message(client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage) -> Non node_id=gateway_numeric_id, hex_id=service_envelope.gateway_id ) + record_fingerprint_packet( + service_envelope, + mesh_packet, + aead=bool(is_encrypted_packet) and decryption_info.get("method") == "ccm", + ) + # Process different packet types if mesh_packet.decoded.portnum == portnums_pb2.PortNum.TEXT_MESSAGE_APP: text_content = mesh_packet.decoded.payload.decode("utf-8", errors="replace") @@ -1500,6 +1747,10 @@ def main() -> None: "Node cache load step finished in %.3fs", time.time() - startup_step_start ) + # Packets from here on are fingerprinted live; history before this point + # is covered once by the background backfill. + start_fingerprint_backfill_if_needed() + # Initialize MQTT Client mqtt_client = mqtt.Client( CallbackAPIVersion.VERSION2, client_id=MQTT_CLIENT_ID or "" @@ -1560,6 +1811,7 @@ def main() -> None: last_optimize = time.time() while True: time.sleep(60) # Print stats every minute + flush_fingerprints() stats = get_node_statistics() logging.info( f"Stats: {stats['total_nodes']} nodes, {stats['total_packets']} packets, {stats['active_nodes_24h']} active (24h)" @@ -1597,6 +1849,8 @@ def main() -> None: if cleanup_thread.is_alive(): logging.warning("Cleanup thread did not finish gracefully") + flush_fingerprints(force=True) + logging.info("Stopping MQTT client loop...") mqtt_client.loop_stop() logging.info("Disconnecting from MQTT broker...") diff --git a/src/malla/routes/api_routes.py b/src/malla/routes/api_routes.py index 2e097e4c..9e2eed7d 100644 --- a/src/malla/routes/api_routes.py +++ b/src/malla/routes/api_routes.py @@ -195,6 +195,31 @@ def api_analytics(): return jsonify({"error": str(e)}), 500 +@api_bp.route("/firmware-distribution") +def api_firmware_distribution(): + """Dashboard: firmware version buckets of nodes heard in the last N days.""" + try: + days = request.args.get("days", default=7, type=int) or 7 + return safe_jsonify(NodeRepository.get_firmware_distribution(days=days)) + except Exception as e: + logger.error(f"Error in API firmware distribution: {e}") + return jsonify({"error": str(e)}), 500 + + +@api_bp.route("/hardware-distribution") +def api_hardware_distribution(): + """Dashboard: hardware models of nodes heard in the last N days.""" + try: + days = request.args.get("days", default=7, type=int) or 7 + top = request.args.get("top", default=12, type=int) or 12 + return safe_jsonify( + NodeRepository.get_hardware_distribution(days=days, top=top) + ) + except Exception as e: + logger.error(f"Error in API hardware distribution: {e}") + return jsonify({"error": str(e)}), 500 + + @api_bp.route("/activity-timeline") def api_activity_timeline(): """API endpoint for the dashboard activity timeline (24h / 7d / 30d / all).""" @@ -854,6 +879,16 @@ def api_node_info(node_id): if not node_info: return jsonify({"error": "Node not found"}), 404 + # The hover card shows a Firmware row whenever firmware_version is set. + verdict = NodeRepository.get_firmware_estimate( + node_id_int, + role=node_info.get("role"), + hw_model=node_info.get("hw_model"), + ) + if verdict["source"] != "none": + node_info["firmware_version"] = verdict["label"] + node_info["firmware_source"] = verdict["source"] + # Format the response to match what the frontend expects return jsonify({"node": node_info}) @@ -1789,6 +1824,11 @@ def api_nodes_data(): "last_packet_time": node.get("last_packet_time"), "packet_count_24h": node.get("packet_count_24h", 0), "status": status, + "firmware_label": node.get("firmware_label", "Unknown"), + "firmware_source": node.get("firmware_source", "none"), + "firmware_reasons": node.get("firmware_reasons", []), + "firmware_evidence": node.get("firmware_evidence", []), + "firmware_reported_version": node.get("firmware_reported_version"), } ) diff --git a/src/malla/routes/node_routes.py b/src/malla/routes/node_routes.py index c2292ec3..6c451e6a 100644 --- a/src/malla/routes/node_routes.py +++ b/src/malla/routes/node_routes.py @@ -46,6 +46,12 @@ def node_detail(node_id): if not node_details: return "Node not found", 404 + node = node_details.get("node") or {} + node_details["firmware"] = NodeRepository.get_firmware_estimate( + node_id_int, role=node.get("role"), hw_model=node.get("hw_model") + ) + node_details["same_mac_nodes"] = NodeRepository.get_same_mac_nodes(node_id_int) + logger.info("Node detail page rendered successfully") return render_template("node_detail.html", **node_details) except Exception as e: diff --git a/src/malla/services/node_service.py b/src/malla/services/node_service.py index ee9385e7..391e454f 100644 --- a/src/malla/services/node_service.py +++ b/src/malla/services/node_service.py @@ -65,12 +65,19 @@ def get_node_info(node_id) -> dict[str, Any]: node_id_int, max_distance_km=10.0 ) + firmware = NodeRepository.get_firmware_estimate( + node_id_int, role=node.get("role"), hw_model=node.get("hw_model") + ) + same_mac_nodes = NodeRepository.get_same_mac_nodes(node_id_int) + # Combine all data return { "node": node, "traceroute_stats": traceroute_stats, "location_history": location_history, "neighbors": neighbors, + "firmware": firmware, + "same_mac_nodes": same_mac_nodes, } @staticmethod diff --git a/src/malla/templates/dashboard.html b/src/malla/templates/dashboard.html index eedb8888..5b42abef 100644 --- a/src/malla/templates/dashboard.html +++ b/src/malla/templates/dashboard.html @@ -340,6 +340,48 @@
Protocol Usage (24h) + +
+
+
+
+
Firmware Versions
+
+ + +
+
+
+
+ +
+ + + +
+
+
+
+
@@ -544,6 +586,225 @@
Signal Quality
{ + button.classList.toggle('active', button.dataset.mode === distributionMode); + }); + const title = document.getElementById('distributionTitle'); + if (title) title.textContent = distributionMode === 'hardware' ? 'Hardware Models' : 'Firmware Versions'; +} + +async function loadDistribution(mode) { + if (mode !== 'hardware') mode = 'firmware'; + distributionMode = mode; + localStorage.setItem('mallaDistributionMode', mode); + syncDistributionMode(); + const slot = distributionSlot(mode); + if (distributionData[slot]) { + createDistributionChart(mode, distributionData[slot]); + return; + } + const loading = document.getElementById('firmwareDistributionChartLoading'); + const chartEl = document.getElementById('firmwareDistributionChart'); + const emptyEl = document.getElementById('firmwareDistributionEmpty'); + if (loading) loading.style.display = 'block'; + if (chartEl) chartEl.style.display = 'none'; + if (emptyEl) emptyEl.style.display = 'none'; + try { + const response = await fetch(distributionEndpoint(mode)); + if (!response.ok) throw new Error(`HTTP error! status: ${response.status}`); + distributionData[slot] = await response.json(); + if (distributionMode === mode && distributionSlot(mode) === slot) { + createDistributionChart(mode, distributionData[slot]); + } + } catch (error) { + console.error(`Error loading ${mode} distribution:`, error); + hideLoadingSpinner('firmwareDistributionChart'); + } +} + +// Kept for callers/tests that know the original name. +function loadFirmwareDistribution() { + return loadDistribution(distributionMode); +} + +function createDistributionChart(mode, data) { + hideLoadingSpinner('firmwareDistributionChart'); + const ctx = document.getElementById('firmwareDistributionCanvas'); + if (!ctx) return; + const isHardware = mode === 'hardware'; + const buckets = (data && data.buckets) || []; + const chartEl = document.getElementById('firmwareDistributionChart'); + const emptyEl = document.getElementById('firmwareDistributionEmpty'); + const emptyText = document.getElementById('firmwareDistributionEmptyText'); + if (chartInstances.firmwareDistributionChart) { + chartInstances.firmwareDistributionChart.destroy(); + delete chartInstances.firmwareDistributionChart; + } + if (!buckets.length) { + if (chartEl) chartEl.style.display = 'none'; + if (emptyText) emptyText.textContent = isHardware ? 'No hardware data yet' : 'No firmware evidence yet'; + if (emptyEl) emptyEl.style.display = 'block'; + return; + } + if (emptyEl) emptyEl.style.display = 'none'; + if (chartEl) chartEl.style.display = 'block'; + + const colors = getChartColors(); + const total = data.total_nodes || buckets.reduce((sum, b) => sum + b.count, 0); + + const wrap = document.getElementById('firmwareDistributionCanvasWrap'); + if (wrap) wrap.style.height = `${Math.max(150, 26 * buckets.length + (isHardware ? 16 : 40))}px`; + + const caption = document.getElementById('firmwareDistributionCaption'); + const expand = document.getElementById('firmwareDistributionExpand'); + if (caption && total) { + if (isHardware) { + const shown = buckets.filter(b => b.kind === 'hardware').length; + const models = data.models || shown; + caption.textContent = `${total.toLocaleString()} nodes · ${shown} of ${models} models`; + if (expand) { + // Offer the switch only when it changes something. + const collapsible = models > HARDWARE_TOP; + expand.style.display = collapsible ? '' : 'none'; + expand.textContent = hardwareShowAll ? `Top ${HARDWARE_TOP} only` : `Show all ${models} models`; + } + } else { + const known = buckets.reduce((sum, b) => sum + (b.reported || 0) + (b.estimated || 0), 0); + caption.textContent = `${total.toLocaleString()} nodes · ${(100 * known / total).toFixed(0)}% classified`; + if (expand) expand.style.display = 'none'; + } + } + + // Firmware: one stacked bar per version segment (reported vs estimated). + // Hardware: one bar per model, muted for the Other/Unknown rows. + const mutedColor = hexToRgba(colors.secondary, 0.45); + const datasets = isHardware + ? [{ + label: 'nodes', + data: buckets.map(b => b.count), + backgroundColor: buckets.map(b => b.kind === 'hardware' ? colors.primary : mutedColor), + maxBarThickness: 18 + }] + : [ + { key: 'reported', label: 'reported', color: colors.success }, + { key: 'estimated', label: 'estimated', color: colors.primary }, + { key: 'other', label: 'unclassified', color: mutedColor }, + ].map(s => ({ + label: s.label, + data: buckets.map(b => b[s.key] || 0), + backgroundColor: s.color, + maxBarThickness: 18 + })); + + // Node count and share printed just past the end of each bar. + const barTotalsPlugin = { + id: 'firmwareBarTotals', + afterDatasetsDraw(chart) { + const { ctx: c, chartArea } = chart; + const lastMeta = chart.getDatasetMeta(chart.data.datasets.length - 1); + if (!lastMeta || !lastMeta.data) return; + c.save(); + c.textBaseline = 'middle'; + c.textAlign = 'left'; + buckets.forEach((b, i) => { + // Right edge of the stack = max x over all datasets for this row. + let end = chartArea.left; + chart.data.datasets.forEach((_, di) => { + const el = chart.getDatasetMeta(di).data[i]; + if (el && el.x > end) end = el.x; + }); + const y = lastMeta.data[i] ? lastMeta.data[i].y : null; + if (y === null) return; + const pct = total ? (100 * b.count / total).toFixed(b.count / total < 0.1 ? 1 : 0) : '0'; + const x = Math.min(end, chartArea.right) + 6; + const countText = b.count.toLocaleString(); + c.font = '600 11px ' + (Chart.defaults.font.family || 'sans-serif'); + c.fillStyle = colors.textColor; + c.fillText(countText, x, y); + const countWidth = c.measureText(countText).width; + c.font = '400 11px ' + (Chart.defaults.font.family || 'sans-serif'); + c.fillStyle = colors.mutedTextColor; + c.fillText(` ${pct}%`, x + countWidth, y); + }); + c.restore(); + } + }; + const maxCount = Math.max(...buckets.map(b => b.count), 1); + + chartInstances.firmwareDistributionChart = new Chart(ctx, { + type: 'bar', + plugins: [barTotalsPlugin], + data: { labels: buckets.map(b => b.label), datasets }, + options: { + indexAxis: 'y', + responsive: true, + maintainAspectRatio: false, + layout: { padding: { right: 8 } }, + plugins: { + legend: { + display: !isHardware, + position: 'bottom', + labels: { color: colors.mutedTextColor, boxWidth: 10, usePointStyle: true } + }, + tooltip: { + callbacks: { + title: (items) => buckets[items[0].dataIndex].label, + label: (item) => { + const value = item.raw || 0; + if (!value) return null; + return isHardware ? null : `${item.dataset.label}: ${value.toLocaleString()}`; + }, + footer: (items) => { + const count = buckets[items[0].dataIndex].count; + const pct = total ? (100 * count / total).toFixed(1) : '0'; + return `${count.toLocaleString()} nodes (${pct}%)`; + } + } + } + }, + scales: { + x: { + stacked: true, + beginAtZero: true, + // leave room for the count label after the longest bar + suggestedMax: Math.ceil(maxCount * 1.2), + grid: { color: colors.gridColor }, + border: { display: false }, + ticks: { color: colors.mutedTextColor, maxTicksLimit: 6, precision: 0 } + }, + y: { + stacked: true, + grid: { display: false }, + ticks: { color: colors.textColor, autoSkip: false } + } + } + } + }); +} + // Update all charts when theme changes function updateChartsForTheme() { if (analyticsData) { @@ -560,6 +821,9 @@
Signal Quality
Signal Quality
{% endblock %} @@ -165,6 +167,23 @@
Node Information
{{ node.mac_address }} {% endif %} + {% if same_mac_nodes %} + + Same Device: + +
    + {% for sibling in same_mac_nodes %} +
  • + {{ sibling.hex_id }} + {{ sibling.node_name }} + {{ sibling.relation }} id + {% if sibling.first_seen_str %}first seen {{ sibling.first_seen_str }}{% if sibling.last_updated_str %}, last {{ sibling.last_updated_str }}{% endif %}{% endif %} +
    • + {% endfor %} +
+ + + {% endif %} First Seen: {{ node.first_seen }} @@ -212,11 +231,19 @@
Node Information
Firmware Version: - Not available - + {% if firmware and firmware.source == 'reported' %} + {{ firmware.label }} + reported{% if firmware.reported_at_str %} {{ firmware.reported_at_str }}{% endif %} + {% elif firmware and firmware.source == 'fingerprint' %} + {{ firmware.label }} + estimated + {% else %} + Unknown + {% endif %} + @@ -791,6 +818,13 @@
Quick Actions