diff --git a/.env.example b/.env.example index ba9409f..7c22f22 100644 --- a/.env.example +++ b/.env.example @@ -92,3 +92,36 @@ POTATOMESH_IMAGE_TAG="latest" # Set to "bridge" on Docker Desktop (macOS/Windows) if host networking # is unavailable. # COMPOSE_PROFILES="bridge" + +# ============================================================================= +# PASSIVE UDP TRANSPORT +# ============================================================================= +# Ingest by passively listening to the node's "Mesh via UDP" LAN multicast +# instead of holding the radio's single API/serial slot. Enable it on the node +# first: `meshtastic --set network.enabled_protocols 1`. Requires host +# networking (multicast can't reach a bridged container). See the +# "Passive UDP transport" section of README.md. + +# Transport: "api" (Meshtastic library over serial/TCP/BLE) or "udp" (passive). +# TRANSPORT=udp + +# Ingest only channel 0. In UDP mode filtering is unconditional; this flag only +# affects the api/serial transport. +# PRIMARY_CHANNEL_ONLY=1 + +# Base64 primary-channel PSK (Meshtastic default key shown). +# PRIMARY_CHANNEL_KEY=AQ== + +# Name of channel 0 — the preset name (MediumFast/LongFast/...) when the radio +# leaves the name blank, as shown by `meshtastic --info`. REQUIRED for UDP +# primary-only: it resolves the channel hash that separates the primary channel +# from a secondary channel sharing the default key. If unset, UDP mode drops ALL +# traffic (fail closed). +# PRIMARY_CHANNEL_NAME=MediumFast + +# Host node id used for the ingestor heartbeat (UDP can't auto-detect "self"). +# INGESTOR_NODE_ID=!xxxxxxxx + +# Multicast group/port for "Mesh via UDP" (defaults shown). +# MESH_UDP_GROUP=224.0.0.69 +# MESH_UDP_PORT=4403 diff --git a/README.md b/README.md index 2e03590..97403ad 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,13 @@ The web app can be configured with environment variables (defaults shown): | `DEBUG` | `0` | Set to `1` for verbose logging in the web and ingestor services. | | `ALLOWED_CHANNELS` | _unset_ | Comma-separated channel names the ingestor accepts; when set, all other channels are skipped before hidden filters. | | `HIDDEN_CHANNELS` | _unset_ | Comma-separated channel names the ingestor will ignore when forwarding packets. | +| `TRANSPORT` | `api` | Ingestor transport: `api` (Meshtastic library over serial/TCP/BLE) or `udp` (passive LAN multicast; see [Passive UDP transport](#passive-udp-transport)). | +| `PRIMARY_CHANNEL_ONLY` | `0` | Set to `1` to ingest only the primary channel (index 0) and drop all other channels. In UDP transport this requires `PRIMARY_CHANNEL_NAME`; without it, every packet is dropped (fail closed). | +| `PRIMARY_CHANNEL_KEY` | `AQ==` | Base64 PSK used to decrypt the primary channel in UDP transport (default = Meshtastic default key). | +| `PRIMARY_CHANNEL_NAME` | _unset_ | Name of channel 0 (e.g. `MediumFast`/`LongFast` — the preset name the firmware uses when the channel name is blank, as shown by `meshtastic --info`). Used to compute the channel hash that identifies primary traffic on the UDP multicast. Required by UDP `PRIMARY_CHANNEL_ONLY=1`, because a secondary channel can share the default `AQ==` key — only the per-channel hash of *(name, key)* distinguishes them. | +| `MESH_UDP_GROUP` | `224.0.0.69` | Multicast group joined in UDP transport. | +| `MESH_UDP_PORT` | `4403` | Multicast port joined in UDP transport. | +| `INGESTOR_NODE_ID` | _unset_ | `!xxxxxxxx` id used for the ingestor heartbeat in UDP transport (which cannot auto-detect "self"). | | `FEDERATION` | `1` | Set to `1` to announce your instance and crawl peers, or `0` to disable federation. Private mode overrides this. | | `PRIVATE` | `0` | Set to `1` to hide the chat UI, disable message APIs, and exclude hidden clients from public listings. | | `EVENTS` | `1` | Set to `0` to disable the live-update SSE stream (`GET /api/events`); clients then fall back to polling at the refresh interval. | @@ -267,6 +274,39 @@ example `ALLOWED_CHANNELS="Chat,Ops"`); packets on other channels are discarded. Use `HIDDEN_CHANNELS` to block specific channels from the web UI even when they appear in the allowlist. +### Passive UDP transport + +The Meshtastic node radio accepts only **one** API client at a time (serial or +TCP), so an ingestor connected over `CONNECTION` monopolizes the node — the +phone app, CLI, and message sending fight it for the single slot. Setting +`TRANSPORT=udp` switches the ingestor to a **passive** listener that never +connects to the node's API at all: it joins the node's LAN multicast group +(Meshtastic "Mesh via UDP", `224.0.0.69:4403`) and decodes packets off the wire, +leaving the node's API slot completely free. + +Enable "Mesh via UDP" on the node first (`meshtastic --set +network.enabled_protocols 1`). The ingestor decrypts the primary channel with +`PRIMARY_CHANNEL_KEY` (the Meshtastic default key `AQ==` by default). Private +channels with their own secret keys are cryptographically unreadable and +dropped. To guarantee **only** channel 0 is ingested, set +`PRIMARY_CHANNEL_ONLY=1` **and** `PRIMARY_CHANNEL_NAME` (e.g. `MediumFast`): each +packet advertises the hash of its channel *(name + key)*, and only packets whose +hash matches the primary channel's are accepted. This is stricter than decrypting +with the primary key — a secondary channel created with the default `AQ==` key +would decrypt too, but has a different name and therefore a different hash, so it +is dropped. If `PRIMARY_CHANNEL_NAME` is not set while `PRIMARY_CHANNEL_ONLY=1`, +the ingestor fails closed and drops everything. Because there is no API +connection, the node's bulk node database is not read — the node list rebuilds +over the air from observed packets, and payloads (position, telemetry, +traceroute, …) are decoded into the exact same shape the API/serial transport +produces, so the collector receives identical records. + +`TRANSPORT=udp` requires host networking so the container can receive LAN +multicast (`network_mode: host`). A ready-to-use Raspberry Pi (arm64) deployment +is provided in [`data/tools/compose.udp.pi.yml`](data/tools/compose.udp.pi.yml). +Capture live packets for testing with +[`data/tools/capture_udp_fixtures.py`](data/tools/capture_udp_fixtures.py). + ## Nix For the dev shell, run: diff --git a/data/Dockerfile b/data/Dockerfile index fe84a59..1d0f1b4 100644 --- a/data/Dockerfile +++ b/data/Dockerfile @@ -51,6 +51,12 @@ ENV CONNECTION=/dev/ttyACM0 \ CHANNEL_INDEX=0 \ DEBUG=0 \ PROTOCOL=meshtastic \ + TRANSPORT=api \ + PRIMARY_CHANNEL_ONLY=0 \ + PRIMARY_CHANNEL_KEY=AQ== \ + PRIMARY_CHANNEL_NAME="" \ + MESH_UDP_GROUP=224.0.0.69 \ + MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ INSTANCE_DOMAIN="" \ @@ -79,6 +85,12 @@ ENV CONNECTION=/dev/ttyACM0 \ CHANNEL_INDEX=0 \ DEBUG=0 \ PROTOCOL=meshtastic \ + TRANSPORT=api \ + PRIMARY_CHANNEL_ONLY=0 \ + PRIMARY_CHANNEL_KEY=AQ== \ + PRIMARY_CHANNEL_NAME="" \ + MESH_UDP_GROUP=224.0.0.69 \ + MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ INSTANCE_DOMAIN="" \ diff --git a/data/mesh_ingestor/channels.py b/data/mesh_ingestor/channels.py index 7547354..25408e9 100644 --- a/data/mesh_ingestor/channels.py +++ b/data/mesh_ingestor/channels.py @@ -310,6 +310,18 @@ def register_channel(channel_idx: int, channel_name_value: str) -> None: ) +def is_primary_channel(channel_index: int | None) -> bool: + """Return ``True`` when *channel_index* is the primary channel (index 0).""" + + return channel_index == 0 + + +def is_primary_only() -> bool: + """Return ``True`` when ingestion is restricted to the primary channel.""" + + return bool(getattr(config, "PRIMARY_CHANNEL_ONLY", False)) + + def _reset_channel_cache() -> None: """Clear cached channel data. Intended for use in tests only.""" @@ -327,5 +339,7 @@ __all__ = [ "hidden_channel_names", "is_allowed_channel", "is_hidden_channel", + "is_primary_channel", + "is_primary_only", "_reset_channel_cache", ] diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index 5cdd442..175b9ee 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -81,6 +81,44 @@ PROTOCOL = _raw_protocol Accepted values are ``meshtastic`` (default) and ``meshcore``. """ +_raw_transport = os.environ.get("TRANSPORT", "api").strip().lower() +if _raw_transport not in ("api", "udp"): + raise ValueError(f"Unknown TRANSPORT={_raw_transport!r}. Valid options: api, udp") +TRANSPORT = _raw_transport +"""Active ingestor transport: ``api`` (Meshtastic library) or ``udp`` (passive multicast).""" + +PRIMARY_CHANNEL_ONLY = os.environ.get("PRIMARY_CHANNEL_ONLY") == "1" +"""When ``True``, only channel index 0 (PRIMARY) is ingested; all else is dropped.""" + +PRIMARY_CHANNEL_KEY = os.environ.get("PRIMARY_CHANNEL_KEY", "AQ==").strip() or "AQ==" +"""Base64 PSK used to decrypt the primary channel; defaults to the Meshtastic default key.""" + +PRIMARY_CHANNEL_NAME = os.environ.get("PRIMARY_CHANNEL_NAME", "").strip() +"""Name of the primary channel (e.g. ``"MediumFast"``), used to compute the +channel hash that identifies primary-channel traffic on the UDP multicast. + +For a channel whose name is left blank in the radio config, this is the LoRa +modem-preset name the firmware substitutes when hashing (``"LongFast"``, +``"MediumFast"``, ``"ShortFast"``, ...) -- i.e. the name shown for channel 0 by +``meshtastic --info``. Required for UDP primary-channel filtering: two channels +can share the default ``AQ==`` key (a SECONDARY channel added with the default +PSK), so decryptability alone cannot distinguish PRIMARY from SECONDARY -- only +the per-channel hash of *(name, key)* can. When blank, UDP primary-only mode +fails closed (drops every packet) rather than risk leaking a secondary channel.""" + +MESH_UDP_GROUP = os.environ.get("MESH_UDP_GROUP", "224.0.0.69").strip() or "224.0.0.69" +"""IPv4 multicast group joined in UDP transport mode.""" + +MESH_UDP_PORT = int(os.environ.get("MESH_UDP_PORT", "4403").strip() or "4403") +"""UDP port for the Mesh-via-UDP multicast group. + +The value is stripped and falls back to ``4403`` when blank, matching the other +UDP env vars, so a whitespace/empty ``MESH_UDP_PORT`` in a ``.env`` file does not +raise ``ValueError`` at import and prevent the service from starting.""" + +INGESTOR_NODE_ID = os.environ.get("INGESTOR_NODE_ID", "").strip() or None +"""Optional ``!xxxxxxxx`` host node id used for the ingestor heartbeat in UDP mode.""" + def _parse_lora_freq_env(raw: str | None) -> float | int | None: """Parse the ``FREQUENCY`` environment variable into a numeric LoRa frequency. @@ -322,6 +360,13 @@ __all__ = [ "ENERGY_SAVING", "LORA_FREQ", "MODEM_PRESET", + "TRANSPORT", + "PRIMARY_CHANNEL_ONLY", + "PRIMARY_CHANNEL_KEY", + "PRIMARY_CHANNEL_NAME", + "MESH_UDP_GROUP", + "MESH_UDP_PORT", + "INGESTOR_NODE_ID", "_RECONNECT_INITIAL_DELAY_SECS", "_RECONNECT_MAX_DELAY_SECS", "_CLOSE_TIMEOUT_SECS", diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index b233f0c..7f2febc 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -627,6 +627,10 @@ def main(*, provider: MeshProtocol | None = None) -> None: from .protocols.meshcore import MeshcoreProvider provider = MeshcoreProvider() + elif config.TRANSPORT == "udp": + from .protocols.meshtastic_udp import MeshtasticUdpProvider + + provider = MeshtasticUdpProvider() else: from .protocols.meshtastic import MeshtasticProvider diff --git a/data/mesh_ingestor/handlers/generic.py b/data/mesh_ingestor/handlers/generic.py index 012c84a..87a7c2b 100644 --- a/data/mesh_ingestor/handlers/generic.py +++ b/data/mesh_ingestor/handlers/generic.py @@ -447,6 +447,16 @@ def store_packet_dict(packet: Mapping) -> None: except Exception: channel = 0 + if channels.is_primary_only() and not channels.is_primary_channel(channel): + _ignored_mod._record_ignored_packet(packet, reason="non-primary-channel") + if config.DEBUG: + config._debug_log( + "Ignored packet on non-primary channel", + context="handlers.store_packet_dict", + channel=channel, + ) + return + channel_name_value = channels.channel_name(channel) pkt_id = _first(packet, "id", "packet_id", "packetId", default=None) diff --git a/data/mesh_ingestor/protocols/__init__.py b/data/mesh_ingestor/protocols/__init__.py index 06f3841..3986ae1 100644 --- a/data/mesh_ingestor/protocols/__init__.py +++ b/data/mesh_ingestor/protocols/__init__.py @@ -28,7 +28,9 @@ def __getattr__(name: str) -> object: ``MeshcoreProvider`` and ``ClosedBeforeConnectedError`` are imported on demand so that the MeshCore library (once wired in) is not loaded at - startup when ``PROTOCOL=meshtastic``. + startup when ``PROTOCOL=meshtastic``. ``MeshtasticUdpProvider`` is + likewise lazy so its ``cryptography``/protobuf imports are not paid for + unless ``TRANSPORT=udp`` is actually selected. """ if name == "MeshcoreProvider": from .meshcore import MeshcoreProvider @@ -38,7 +40,16 @@ def __getattr__(name: str) -> object: from .meshcore import ClosedBeforeConnectedError return ClosedBeforeConnectedError + if name == "MeshtasticUdpProvider": + from .meshtastic_udp import MeshtasticUdpProvider + + return MeshtasticUdpProvider raise AttributeError(f"module {__name__!r} has no attribute {name!r}") -__all__ = ["MeshtasticProvider", "MeshcoreProvider", "ClosedBeforeConnectedError"] +__all__ = [ + "MeshtasticProvider", + "MeshcoreProvider", + "ClosedBeforeConnectedError", + "MeshtasticUdpProvider", +] diff --git a/data/mesh_ingestor/protocols/meshtastic_udp.py b/data/mesh_ingestor/protocols/meshtastic_udp.py new file mode 100644 index 0000000..56d93ef --- /dev/null +++ b/data/mesh_ingestor/protocols/meshtastic_udp.py @@ -0,0 +1,327 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Passive UDP ``MeshProtocol`` provider. + +Wires the pure decrypt/mapping logic in +:mod:`data.mesh_ingestor.protocols.meshtastic_udp_decode` and the socket +plumbing in :mod:`data.mesh_ingestor.protocols.meshtastic_udp_socket` into a +:class:`~data.mesh_ingestor.mesh_protocol.MeshProtocol` implementation, so the +daemon can ingest Meshtastic's "Mesh via UDP" LAN multicast broadcasts +instead of holding the node's single API/serial connection slot. + +Unlike :class:`~data.mesh_ingestor.protocols.meshtastic.MeshtasticProvider` +(pubsub-driven) this provider has no async callback registration: a single +background thread reads datagrams off a multicast socket and calls +:func:`~data.mesh_ingestor.handlers.on_receive` directly for every +primary-channel packet. + +Primary-channel membership is decided by the packet's channel *hash*, not by +decryptability, and the gate is UNCONDITIONAL: a datagram is accepted only when +its ``channel`` hash equals the hash of the configured primary channel (see +:func:`~data.mesh_ingestor.protocols.meshtastic_udp_decode.channel_hash`). This +is deliberately stricter than "decrypts with :data:`config.PRIMARY_CHANNEL_KEY`" +because a SECONDARY channel created with the default key would also decrypt -- +so the hash, which folds in the channel *name*, is what keeps secondary/private +channels out. Because this transport stamps channel index 0 on everything it +emits, it can only faithfully represent the primary channel, so filtering is not +optional: when the primary hash cannot be resolved (no +:data:`config.PRIMARY_CHANNEL_NAME`) the provider FAILS CLOSED and drops every +packet. (:data:`config.PRIMARY_CHANNEL_ONLY` still governs the separate +API/serial transport; it does not weaken this gate.) Accepted packets must be +channel-encrypted -- already-decoded (plaintext) packets are dropped to close a +no-key LAN spoofing path -- then decrypted with +:data:`config.PRIMARY_CHANNEL_KEY` and enriched to match the API/serial +transport's packet shape. +""" + +from __future__ import annotations + +import socket +import threading + +from meshtastic.protobuf import mesh_pb2 + +from .. import config, handlers +from .meshtastic_udp_decode import ( + channel_hash, + decrypt_meshpacket, + meshpacket_to_packet_dict, +) +from .meshtastic_udp_socket import open_multicast_socket + + +class _UdpInterface: + """Minimal interface object standing in for a Meshtastic library interface. + + The rest of the ingestor pipeline (daemon loop, heartbeat, snapshot code) + expects an "interface" object with a ``nodes`` mapping, an + ``isConnected`` event, and a ``close()`` method; this class supplies just + that surface for the UDP transport; it does not otherwise track node + state (:meth:`MeshtasticUdpProvider.node_snapshot_items` accordingly + reads an always-empty dict). + """ + + def __init__(self) -> None: + """Initialise an unconnected interface with no known nodes.""" + self.nodes: dict = {} + self.isConnected = threading.Event() + self._sock: socket.socket | None = None + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + def close(self) -> None: + """Stop the receive thread and release the socket. + + Signals :attr:`_stop` first so the receive loop's next timeout (or + the socket close below, whichever comes first) causes it to exit, + then closes the socket (best-effort -- close errors are not + actionable here) and joins the thread with a bounded timeout so + shutdown can never hang indefinitely. + """ + self._stop.set() + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + if self._thread is not None: + self._thread.join(timeout=2.0) + self.isConnected.clear() + + +class MeshtasticUdpProvider: + """Passive Meshtastic "Mesh via UDP" ``MeshProtocol`` implementation.""" + + name = "meshtastic-udp" + + def __init__(self) -> None: + """Initialise the provider with no topics subscribed yet.""" + self._subscribed: list[str] = [] + + def _primary_channel_hash(self) -> int | None: + """Return the channel hash that identifies primary-channel traffic. + + Computed from :data:`config.PRIMARY_CHANNEL_NAME` and + :data:`config.PRIMARY_CHANNEL_KEY` via + :func:`~data.mesh_ingestor.protocols.meshtastic_udp_decode.channel_hash`. + Read fresh each call so a test (or a live config reload) that changes + the environment is honoured without reconstructing the provider. + + Returns: + The primary channel's hash byte, or ``None`` when + :data:`config.PRIMARY_CHANNEL_NAME` is blank -- in which case the + primary channel cannot be identified and primary-only filtering + must fail closed (drop everything) rather than risk leaking a + secondary channel that happens to share the primary key. + """ + name = config.PRIMARY_CHANNEL_NAME + if not name: + return None + return channel_hash(name, config.PRIMARY_CHANNEL_KEY) + + def subscribe(self) -> list[str]: + """Return an empty topic list. + + This provider has no pubsub callbacks to register -- the receive + thread started in :meth:`connect` calls + :func:`~data.mesh_ingestor.handlers.on_receive` directly for every + decoded packet. The method is still idempotent and side-effect-free + so it mirrors the shape of + :meth:`~data.mesh_ingestor.protocols.meshtastic.MeshtasticProvider.subscribe`. + + Returns: + An empty list, always. + """ + return list(self._subscribed) + + def connect( + self, *, active_candidate: str | None + ) -> tuple[object, str | None, str | None]: + """Open the multicast socket and start the background receive thread. + + Parameters: + active_candidate: Ignored (there is no serial/BLE candidate + concept for a multicast listener); passed through unchanged + as the returned "next active candidate" to satisfy the + :class:`~data.mesh_ingestor.mesh_protocol.MeshProtocol` + contract. + + Returns: + A ``(iface, resolved_target, next_active_candidate)`` tuple: the + live :class:`_UdpInterface`, a ``udp://group:port`` string + describing the joined group, and *active_candidate* unchanged. + """ + iface = _UdpInterface() + iface._sock = open_multicast_socket(config.MESH_UDP_GROUP, config.MESH_UDP_PORT) + # Surface the resolved primary-channel filter so operators can verify at + # a glance that ingestion is pinned to the intended channel 0 (e.g. + # "primary_channel_name='MediumFast' primary_channel_hash=31"). Filtering + # is unconditional; a warn severity flags the FAIL-CLOSED state where no + # PRIMARY_CHANNEL_NAME is configured, in which every packet is dropped. + primary_hash = self._primary_channel_hash() + config._debug_log( + "UDP primary-channel filter", + context="udp.connect", + severity="warn" if primary_hash is None else "info", + always=True, + primary_channel_name=config.PRIMARY_CHANNEL_NAME or None, + primary_channel_hash=primary_hash, + ) + # Mark connected BEFORE starting the reader thread so the thread's + # finally-clause always has the last word on clearing it. If the thread + # were started first and hit an immediate socket error, its + # ``finally: isConnected.clear()`` could run before this line, leaving + # the interface wrongly marked connected over a dead reader. + iface.isConnected.set() + iface._thread = threading.Thread( + target=self._recv_loop, args=(iface,), daemon=True + ) + iface._thread.start() + target = f"udp://{config.MESH_UDP_GROUP}:{config.MESH_UDP_PORT}" + return iface, target, active_candidate + + def _recv_loop(self, iface: _UdpInterface) -> None: + """Poll *iface*'s socket for datagrams until told to stop. + + Runs on the background thread started by :meth:`connect`. A + ``socket.timeout`` (the socket has a 1-second timeout, see + :func:`~data.mesh_ingestor.protocols.meshtastic_udp_socket.open_multicast_socket`) + is expected and simply re-checks the stop flag; any other + ``OSError`` (e.g. the socket was closed out from under this thread by + :meth:`_UdpInterface.close`) ends the loop. Per-datagram handling is + wrapped so a malformed or hostile packet is dropped rather than + propagating and killing the thread, and :attr:`_UdpInterface.isConnected` + is cleared on every exit path so a dead reader is detectable. + + Parameters: + iface: The interface whose socket to read and stop flag to + honour. + """ + try: + while not iface._stop.is_set(): + try: + raw, _addr = iface._sock.recvfrom(65535) + except socket.timeout: + continue + except OSError: + break + try: + self._handle_datagram(raw, iface) + except Exception: + # A single malformed or hostile datagram must never kill + # the reader thread. Drop it and continue. Logged at debug + # severity only, so a flood of bad datagrams cannot amplify + # into a log-volume DoS. + config._debug_log( + "Dropped malformed UDP datagram", + context="udp.recv", + severity="debug", + ) + finally: + # Any loop exit -- stop flag, socket error, or an unexpected error + # -- marks the interface disconnected so the daemon can notice a + # dead reader and reconnect, instead of believing a crashed thread + # is still healthy (isConnected was previously only cleared on + # OSError, so a thread death left the daemon wedged). + iface.isConnected.clear() + + def _handle_datagram(self, raw: bytes, iface: _UdpInterface) -> None: + """Parse, filter, decrypt, and dispatch one raw UDP datagram. + + Parses *raw* as a ``MeshPacket`` and dispatches it to + :func:`~data.mesh_ingestor.handlers.on_receive` only when it passes + every gate below; anything else is silently dropped: + + 1. **Parse** -- unparseable bytes are dropped. + 2. **Primary-channel hash** -- the packet's ``channel`` hash must equal + the configured primary channel's hash (see + :meth:`_primary_channel_hash`). This gate is UNCONDITIONAL: the UDP + transport can only faithfully represent the primary channel (it + stamps channel index 0), so it must never emit anything else. When + the primary hash cannot be resolved (no + :data:`config.PRIMARY_CHANNEL_NAME`) the gate FAILS CLOSED and drops + everything, rather than risk leaking a secondary channel. + 3. **Encrypted-only** -- the packet must carry ``encrypted`` bytes; + already-``decoded`` (plaintext) packets are dropped, closing a + no-key LAN spoofing path. + 4. **Decrypt** -- decryption with :data:`config.PRIMARY_CHANNEL_KEY` + must succeed (a private channel this key cannot open decrypts to + ``None`` and is dropped). + + Parameters: + raw: The raw datagram bytes read from the multicast socket. + iface: The interface to report as the packet's origin. + """ + mp = mesh_pb2.MeshPacket() + try: + mp.ParseFromString(raw) + except Exception: + return + # Channel-0-only enforcement (unconditional -- fail closed). A + # ``MeshPacket`` advertises the hash of its channel (a fold of channel + # name + key); accept only when that hash matches the PRIMARY channel's. + # This is stricter than "decrypts with the primary key" -- a SECONDARY + # channel created with the default AQ== key would decrypt too, but has a + # different name and therefore a different hash. + primary_hash = self._primary_channel_hash() + if primary_hash is None or mp.channel != primary_hash: + return + # Require channel-encrypted traffic. Real primary-channel packets on the + # multicast feed are always encrypted with the channel key; dropping + # packets that arrive already-``decoded`` (plaintext) closes a no-key + # LAN spoofing path and avoids forwarding unauthenticated records. + if not mp.HasField("encrypted"): + return + data = decrypt_meshpacket(mp, config.PRIMARY_CHANNEL_KEY) + if data is None: + # Private channel (or noise) this key cannot open -- drop. + return + mp.decoded.CopyFrom(data) + handlers.on_receive(packet=meshpacket_to_packet_dict(mp), interface=iface) + + def extract_host_node_id(self, iface: object) -> str | None: + """Return the configured host node id. + + Unlike the API/serial transport, a passive multicast listener has no + protocol-level handshake that reveals "our" node id, so this simply + surfaces the operator-supplied :data:`config.INGESTOR_NODE_ID`. + + Parameters: + iface: Unused; accepted for + :class:`~data.mesh_ingestor.mesh_protocol.MeshProtocol` + signature compatibility. + + Returns: + :data:`config.INGESTOR_NODE_ID`, or ``None`` when unset. + """ + return config.INGESTOR_NODE_ID + + def node_snapshot_items(self, iface: object) -> list[tuple[str, object]]: + """Return a snapshot of known nodes. + + This provider does not track a node roster (it only relays decoded + packets), so the snapshot reflects whatever (typically empty) + ``nodes`` mapping the interface carries. + + Parameters: + iface: The interface whose ``nodes`` mapping to snapshot. + + Returns: + A list of ``(node_id, node_obj)`` tuples; empty when *iface* has + no ``nodes`` attribute or an empty one. + """ + return list(getattr(iface, "nodes", {}).items()) + + +__all__ = ["MeshtasticUdpProvider"] diff --git a/data/mesh_ingestor/protocols/meshtastic_udp_decode.py b/data/mesh_ingestor/protocols/meshtastic_udp_decode.py new file mode 100644 index 0000000..c1ca593 --- /dev/null +++ b/data/mesh_ingestor/protocols/meshtastic_udp_decode.py @@ -0,0 +1,283 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Decrypt raw Meshtastic ``MeshPacket`` datagrams and map them to dicts. + +This module is the pure-logic core of the passive UDP transport: it has no +socket, threading, or daemon dependencies, so it can be unit tested (and +100%-covered) in complete isolation from the network. + +Meshtastic's default/primary channel uses a small, publicly documented set +of 1-byte "default" PSKs (``0x01``..``0x07``) that every stock node ships +with, specifically so default-channel traffic is decodable by any compatible +client. This module implements that well-known key expansion plus the +AES-CTR decrypt used by the Meshtastic firmware, and maps a decoded packet +into the same dict shape the rest of this ingestor's pipeline already +consumes (see :mod:`data.mesh_ingestor.handlers`). +""" + +from __future__ import annotations + +import base64 +import time + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from google.protobuf.json_format import MessageToDict +from meshtastic import protocols as _PROTOCOLS +from meshtastic.protobuf import mesh_pb2, portnums_pb2 + +_ONE_BYTE_PSK_PREFIX = bytes.fromhex("d4f1bb3a20290759f0bcffabcf4e69") +"""15-byte prefix Meshtastic firmware prepends to a 1-byte default PSK. + +Concatenating this prefix with the raw 1-byte key (``0x01``..``0x07``) +reconstructs the 16-byte AES-128 key used for the corresponding default +channel, per the Meshtastic firmware's crypto implementation. +""" + + +def expand_default_key(key_b64: str) -> bytes: + """Return the 16-byte AES key encoded by *key_b64*. + + Meshtastic channel PSKs are base64-encoded. A single decoded byte in the + range ``0x01``..``0x07`` denotes one of the firmware's built-in "default" + keys and must be expanded to 16 bytes via :data:`_ONE_BYTE_PSK_PREFIX` + before use with AES-128. Any other decoded length (typically a full + 16-byte or 32-byte channel PSK) is returned unchanged. + + Args: + key_b64: Base64-encoded channel PSK, e.g. ``"AQ=="``. + + Returns: + The raw AES key bytes. + + Raises: + binascii.Error: If *key_b64* is not valid base64. + """ + raw = base64.b64decode(key_b64.encode("ascii"), validate=True) + if len(raw) == 1 and 0x01 <= raw[0] <= 0x07: + return _ONE_BYTE_PSK_PREFIX + raw + return raw + + +def _xor_hash(data: bytes) -> int: + """Return the XOR of every byte in *data* (Meshtastic's ``xorHash``). + + Args: + data: The bytes to fold together. + + Returns: + A single byte (``0``..``255``): all of *data* XOR-ed into one value, + or ``0`` for empty input. + """ + result = 0 + for byte in data: + result ^= byte + return result + + +def channel_hash(channel_name: str, key_b64: str) -> int: + """Return the 1-byte Meshtastic channel hash for *(channel_name, key)*. + + This mirrors the firmware's ``Channels::generateHash``: the XOR-fold of the + UTF-8 channel name XOR-ed with the XOR-fold of the (expanded) channel key. + The hash is what a Meshtastic ``MeshPacket`` carries in its ``channel`` + field so receivers can pick the matching channel/key. + + Because the hash mixes in the channel *name*, two channels that share the + same PSK -- e.g. the PRIMARY channel and a SECONDARY channel both left on + the default ``AQ==`` key -- still produce different hashes. That distinction + is exactly what lets the passive UDP transport keep only PRIMARY traffic: + decrypting with the default key is not sufficient (a default-key SECONDARY + channel would decrypt too), so we additionally require the packet's channel + hash to equal the PRIMARY channel's hash. + + Args: + channel_name: The channel name used by the firmware when hashing (the + configured name, or the modem-preset name when the name is blank). + key_b64: Base64-encoded channel PSK (see :func:`expand_default_key`). + + Returns: + The channel hash byte (``0``..``255``). + + Raises: + binascii.Error: If *key_b64* is not valid base64. + """ + return _xor_hash(channel_name.encode("utf-8")) ^ _xor_hash( + expand_default_key(key_b64) + ) + + +def decrypt_meshpacket( + mp: "mesh_pb2.MeshPacket", key_b64: str +) -> "mesh_pb2.Data | None": + """Decrypt ``mp.encrypted`` and return the parsed :class:`~mesh_pb2.Data`. + + Uses AES-CTR with the key derived from *key_b64* (see + :func:`expand_default_key`) and a nonce built from the packet's ``id`` + and ``from`` fields (both little-endian, 8 bytes each), matching the + Meshtastic firmware's construction. + + Any failure along the way -- a malformed *key_b64*, an undersized/absent + ``mp.encrypted`` payload, or ciphertext that fails to decode as a valid + ``Data`` protobuf -- is treated as "this packet was not encrypted with + this key" and reported as ``None`` rather than raised, since the same + code path is used to probe packets on channels this process has no key + for (e.g. private channels captured alongside the primary channel). + + A packet that decodes cleanly but carries the default/unknown portnum + (``0``) with an empty payload is also treated as a decrypt failure: in + practice this is what "wrong key, coincidentally valid protobuf" garbage + looks like, and real Meshtastic application payloads always set a + portnum, a payload, or both. + + Args: + mp: A parsed ``MeshPacket`` whose ``encrypted`` field holds + ciphertext (as opposed to an already-``decoded`` packet). + key_b64: Base64-encoded channel PSK to decrypt with. + + Returns: + The decrypted :class:`~mesh_pb2.Data`, or ``None`` if decryption or + parsing failed, or the result looks like private-channel noise. + """ + try: + key = expand_default_key(key_b64) + nonce = mp.id.to_bytes(8, "little") + getattr(mp, "from").to_bytes(8, "little") + decryptor = Cipher(algorithms.AES(key), modes.CTR(nonce)).decryptor() + clear = decryptor.update(mp.encrypted) + decryptor.finalize() + data = mesh_pb2.Data() + data.ParseFromString(clear) + except Exception: + # Any decode/parse failure means this key does not open this packet. + return None + # A wrong key usually yields non-parseable bytes; a parse that produced + # an unknown/zero portnum with no payload is treated as failure (i.e. + # traffic on a channel this key does not decrypt). + if data.portnum == 0 and not data.payload: + return None + return data + + +def _node_id(num: int) -> str: + """Return the canonical Meshtastic node id string for *num*. + + Args: + num: A 32-bit (or wider, masked down) node number. + + Returns: + ``"^all"`` for the reserved broadcast address + (``0xFFFFFFFF``), otherwise the canonical ``"!xxxxxxxx"`` hex form. + """ + num &= 0xFFFFFFFF + return "^all" if num == 0xFFFFFFFF else "!%08x" % num + + +def _enrich_decoded(decoded: dict, portnum: int, payload: bytes) -> None: + """Populate *decoded* with the protobuf section for *portnum*, in place. + + The passive UDP transport only recovers ``portnum`` + raw ``payload`` from a + ``MeshPacket``, but the ingestor's handlers (position, telemetry, + traceroute, neighborinfo, ...) read the *decoded application message* from a + named sub-dict -- ``decoded["position"]``, ``decoded["telemetry"]``, etc. -- + exactly as the Meshtastic Python library populates it on the API/serial + path. This reproduces that step so a UDP-sourced packet yields byte-for-byte + the same POST payloads as a library-sourced one. + + The decode table (``meshtastic.protocols``) and ``MessageToDict`` call are + the same ones the library uses in ``_handlePacketFromRadio``, so field names + (camelCase) and value formats match. Portnums with no protobuf factory + (e.g. ``TEXT_MESSAGE_APP``) are left untouched. A malformed sub-payload is + swallowed -- the packet still flows with its ``portnum``/``payload`` intact. + + Args: + decoded: The decoded dict to mutate (already holds ``portnum`` and + base64 ``payload``). + portnum: The integer application portnum from the packet. + payload: The raw application-payload bytes to parse. + """ + handler = _PROTOCOLS.get(portnum) + factory = getattr(handler, "protobufFactory", None) if handler else None + if factory is None: + return + try: + message = factory() + message.ParseFromString(payload) + decoded[handler.name] = MessageToDict(message) + except Exception: + # A wrong-length or malformed sub-payload is non-fatal: the handler for + # this portnum will simply find its section absent, exactly as it would + # for a library packet the firmware could not decode. + return + + +def meshpacket_to_packet_dict(mp: "mesh_pb2.MeshPacket") -> dict: + """Map a ``MeshPacket`` with a populated ``decoded`` field to a packet dict. + + Produces the same dict shape the rest of the ingestor pipeline already + consumes from the Meshtastic library's pubsub callbacks (see + :mod:`data.mesh_ingestor.handlers`), so a UDP-sourced packet can be fed + into ``handlers.on_receive`` unchanged. + + Args: + mp: A ``MeshPacket`` whose ``decoded`` field is already populated + (typically via :func:`decrypt_meshpacket` followed by + ``mp.decoded.CopyFrom(data)``, or a packet that was never + encrypted in the first place). + + Returns: + A dict with ``from``, ``fromId``, ``to``, ``toId``, ``id``, + ``channel``, ``rxTime``, and ``decoded`` always present; ``rxSnr``, + ``rxRssi``, and ``hopLimit`` present only when the corresponding + source field is non-zero. The ``decoded`` sub-dict is enriched with the + same protobuf-derived sections the Meshtastic library populates (see + :func:`_enrich_decoded`) so downstream handlers behave identically to + the API/serial transport. + """ + try: + portnum_name = portnums_pb2.PortNum.Name(mp.decoded.portnum) + except ValueError: + # A portnum newer than the installed protobufs (or attacker-supplied + # garbage on the LAN multicast) has no enum name. Map it to a stable + # sentinel that no handler dispatches on, rather than letting the + # ValueError escape and kill the receive thread (a single such packet + # would otherwise permanently stop ingestion). + portnum_name = "UNKNOWN_APP" + decoded: dict = { + "portnum": portnum_name, + "payload": base64.b64encode(mp.decoded.payload).decode("ascii"), + } + if portnum_name == "TEXT_MESSAGE_APP": + decoded["text"] = mp.decoded.payload.decode("utf-8", errors="replace") + _enrich_decoded(decoded, mp.decoded.portnum, mp.decoded.payload) + + packet = { + "from": getattr(mp, "from"), + "fromId": _node_id(getattr(mp, "from")), + "to": mp.to, + "toId": _node_id(mp.to), + "id": mp.id, + # The caller (MeshtasticUdpProvider) only dispatches packets whose + # channel hash equals the PRIMARY channel's hash (see + # channel_hash / MeshtasticUdpProvider._handle_datagram), so a mapped + # packet is always primary -- channel index 0. + "channel": 0, + "rxTime": int(mp.rx_time) if mp.rx_time else int(time.time()), + "decoded": decoded, + } + if mp.rx_snr: + packet["rxSnr"] = float(mp.rx_snr) + if mp.rx_rssi: + packet["rxRssi"] = int(mp.rx_rssi) + if mp.hop_limit: + packet["hopLimit"] = int(mp.hop_limit) + return packet diff --git a/data/mesh_ingestor/protocols/meshtastic_udp_socket.py b/data/mesh_ingestor/protocols/meshtastic_udp_socket.py new file mode 100644 index 0000000..dda8321 --- /dev/null +++ b/data/mesh_ingestor/protocols/meshtastic_udp_socket.py @@ -0,0 +1,66 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Join the Meshtastic "Mesh via UDP" LAN multicast group. + +This module is the pure socket-plumbing half of the passive UDP transport: it +has no protobuf, crypto, or daemon dependencies, so the option-setting and +group-join logic can be unit tested (and 100%-covered) with a fake socket, +independent of any real network stack. + +This logic intentionally mirrors ``data/tools/capture_udp_fixtures.py``'s +``open_multicast_socket``, which has been exercised against a live +Station G2 on macOS and Linux; keep the two in sync if either changes. +""" + +from __future__ import annotations + +import socket + + +def open_multicast_socket(group: str, port: int) -> socket.socket: + """Open, configure, and join *group* on *port* for passive reception. + + Creates an IPv4 UDP socket suitable for receive-only Meshtastic "Mesh via + UDP" multicast traffic: address reuse is enabled (so multiple local + listeners, or quick restarts, don't collide on the port), the socket is + bound to the wildcard address, and it joins *group* via + ``IP_ADD_MEMBERSHIP`` on the default interface (``0.0.0.0``). + + Args: + group: IPv4 multicast group address to join, e.g. ``"224.0.0.69"``. + port: UDP port the group publishes on, e.g. ``4403``. + + Returns: + A bound, group-joined datagram socket with a 1-second receive + timeout, ready for ``recvfrom`` polling. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + # SO_REUSEPORT is not available on every platform (e.g. some Windows + # builds) and, even where the constant exists, some kernels reject it; + # both cases are non-fatal since SO_REUSEADDR above already covers the + # common "restart while the old socket lingers" case. + if hasattr(socket, "SO_REUSEPORT"): + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except OSError: + pass + sock.bind(("", port)) + # Joining with the wildcard interface (0.0.0.0) lets the kernel pick + # the receiving interface, matching how the capture tool joins the group. + mreq = socket.inet_aton(group) + socket.inet_aton("0.0.0.0") + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + sock.settimeout(1.0) + return sock diff --git a/data/requirements.txt b/data/requirements.txt index 6aadee5..564464e 100644 --- a/data/requirements.txt +++ b/data/requirements.txt @@ -3,6 +3,7 @@ meshtastic>=2.5.0 meshcore>=2.3.5 bleak>=0.21.0 protobuf>=5.27.2 +cryptography>=42.0.0 # AES-CTR decryption for the passive UDP transport # Development dependencies (optional) black>=24.8.0 diff --git a/data/tools/README.md b/data/tools/README.md new file mode 100644 index 0000000..b32b090 --- /dev/null +++ b/data/tools/README.md @@ -0,0 +1,77 @@ + + + +# `data/tools/` — passive UDP operator & dev tools + +Helpers for the passive UDP transport (`TRANSPORT=udp`, see the +[Passive UDP transport](../../README.md#passive-udp-transport) section of the +README). Neither file is part of the ingestor runtime. + +## `capture_udp_fixtures.py` — capture real datagrams for testing + +A **receive-only** diagnostic that joins the node's "Mesh via UDP" multicast +group and writes each raw datagram to a JSONL file (base64 in `raw_b64`). Used to +produce the real-traffic fixtures under +[`../../tests/fixtures/mesh_udp/`](../../tests/fixtures/mesh_udp/). + +```bash +# Prereq: "Mesh via UDP" enabled on the node +# meshtastic --set network.enabled_protocols 1 +# Run on a host on the node's LAN (host networking; multicast can't cross a NAT): +python data/tools/capture_udp_fixtures.py --out capture.jsonl --count 40 +# Optional live decode summary of primary-channel packets: +python data/tools/capture_udp_fixtures.py --out capture.jsonl --primary-only +``` + +It never transmits and never connects to the radio API, so it is safe to run +alongside a live ingestor or the phone app. + +**Coverage note:** this is an operator-run diagnostic that needs a live LAN +socket, so it is intentionally exempt from the ingestor package's 100%-unit-test +gate. The runtime decode/crypto it exercises *is* fully covered by +`tests/test_meshtastic_udp_decode_unit.py` against the captured fixtures. + +## `compose.udp.pi.yml` — Raspberry Pi (arm64) deployment + +A Docker Compose file for running the ingestor in passive UDP mode on a Pi 5. It +requires `network_mode: host` (multicast `224.0.0.69` cannot reach a bridged +container) and reads the same `.env` as the standard deployment. + +### `.env` keys the UDP deployment reads + +```dotenv +TRANSPORT=udp +PRIMARY_CHANNEL_ONLY=1 +PRIMARY_CHANNEL_KEY=AQ== # base64 primary PSK (Meshtastic default) +PRIMARY_CHANNEL_NAME=MediumFast # REQUIRED: name of channel 0 (or the preset + # name if blank on the radio); resolves the + # channel hash. If unset, primary-only mode + # drops ALL traffic (fail closed). +INGESTOR_NODE_ID=!xxxxxxxx # host node id for the ingestor heartbeat +MESH_UDP_GROUP=224.0.0.69 +MESH_UDP_PORT=4403 +# plus the standard API_TOKEN / INSTANCE_DOMAIN +``` + +### Build → ship → verify + +The image is built **natively on an arm64 Pi** (no QEMU) and copied to the +target: + +```bash +# 1. On a build Pi, from the source tree: +docker build -f data/Dockerfile -t potato-mesh-ingestor:udp . +docker save potato-mesh-ingestor:udp | gzip > potato-mesh-ingestor-udp.tar.gz + +# 2. Copy the image tarball to the target Pi, then: +docker load < potato-mesh-ingestor-udp.tar.gz +cp data/tools/compose.udp.pi.yml compose.yml # first time only; add .env keys above +docker compose up -d + +# 3. Verify: the startup log pins the channel, and no secondary names appear. +docker compose logs ingestor | grep "UDP primary-channel filter" # primary_channel_hash=, severity=info +docker compose logs ingestor | grep "POST request failed" # (expect nothing) +``` + +A `primary_channel_hash` of `null` / `severity=warn` means `PRIMARY_CHANNEL_NAME` +is unset and the ingestor is dropping everything (fail closed). diff --git a/data/tools/capture_udp_fixtures.py b/data/tools/capture_udp_fixtures.py new file mode 100755 index 0000000..0771f55 --- /dev/null +++ b/data/tools/capture_udp_fixtures.py @@ -0,0 +1,254 @@ +#!/usr/bin/env python3 +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Capture Meshtastic "Mesh via UDP" multicast datagrams as replayable test fixtures. + +Passive and receive-only: it never sends anything and never connects to the +node's API — it just joins the LAN multicast group the node broadcasts to and +records raw datagram bytes. It saves each datagram as base64 in a JSONL file so +the exact bytes can be replayed in unit tests, and — when ``meshtastic`` and +``cryptography`` are importable — prints a live decoded summary so the operator +can watch the primary channel decode with the default key in real time. + +This is an operator/developer tool, not part of the shipped ingestor runtime. +Its socket-join and decrypt logic intentionally mirror +``data/mesh_ingestor/protocols/meshtastic_udp_socket.py`` and +``meshtastic_udp_decode.py`` so a live capture doubles as a validation of those +modules; once they exist this tool may be refactored to import them directly. + +Privacy: raw bytes of private-channel packets stay encrypted (their keys are +not available, so nothing readable is captured). Use ``--primary-only`` to save +ONLY packets that decrypt with the default key (i.e. the public/primary +channel). + +Usage (run on any host on the same LAN as the node, e.g. the gateway Pi):: + + python3 data/tools/capture_udp_fixtures.py --seconds 120 --out fixtures.jsonl + python3 data/tools/capture_udp_fixtures.py --seconds 120 --primary-only --out fixtures.jsonl + +The live summary and ``--primary-only`` require ``pip install meshtastic +cryptography``; raw capture works with the standard library alone. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import socket +import sys +import time + +DEFAULT_GROUP = "224.0.0.69" +DEFAULT_PORT = 4403 +DEFAULT_KEY_B64 = "AQ==" +#: 15-byte prefix Meshtastic prepends to a 1-byte PSK (0x01..0x07) to form the key. +_ONE_BYTE_PSK_PREFIX = bytes.fromhex("d4f1bb3a20290759f0bcffabcf4e69") + +# Optional decode support ------------------------------------------------------- +try: + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from meshtastic.protobuf import mesh_pb2, portnums_pb2 + + HAVE_DECODE = True +except Exception: # pragma: no cover - environment dependent + HAVE_DECODE = False + + +def open_multicast_socket(group: str, port: int) -> socket.socket: + """Join *group* on *port* for passive multicast reception (receive-only). + + Parameters: + group: IPv4 multicast group address to join. + port: UDP port the group publishes on. + + Returns: + A bound, group-joined datagram socket with a 1-second receive timeout. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if hasattr(socket, "SO_REUSEPORT"): + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except OSError: + pass + sock.bind(("", port)) + mreq = socket.inet_aton(group) + socket.inet_aton("0.0.0.0") + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + sock.settimeout(1.0) + return sock + + +def _expand_key(key_b64: str) -> bytes: + """Return the 16-byte AES key for *key_b64*, expanding 1-byte Meshtastic PSKs.""" + raw = base64.b64decode(key_b64) + if len(raw) == 1 and 0x01 <= raw[0] <= 0x07: + return _ONE_BYTE_PSK_PREFIX + raw + return raw + + +def _portnum_name(portnum: int) -> str: + """Return the ``PortNum`` enum name, or ``"UNKNOWN_APP"`` when out of range. + + ``PortNum`` is a proto3 open enum, so a datagram may carry a portnum with no + registered name (newer firmware, or garbage on the multicast group). + ``PortNum.Name`` raises ``ValueError`` on those; this maps them to a stable + sentinel so the capture tool never crashes on an unexpected packet. + """ + try: + return portnums_pb2.PortNum.Name(portnum) + except ValueError: + return "UNKNOWN_APP" + + +def summarize(raw: bytes, key_b64: str) -> dict | None: + """Return a human-readable summary of a datagram, decrypting the primary channel. + + Parameters: + raw: Raw datagram bytes as received from the multicast socket. + key_b64: Base64 PSK used to attempt decryption of encrypted packets. + + Returns: + A summary dict, or ``None`` when decode support is unavailable. + """ + if not HAVE_DECODE: + return None + mp = mesh_pb2.MeshPacket() + try: + mp.ParseFromString(raw) + except Exception: + return {"parse": "FAILED (not a MeshPacket?)"} + portnum = None + decoded_ok = False + if mp.HasField("decoded"): + portnum = _portnum_name(mp.decoded.portnum) + decoded_ok = True + elif mp.HasField("encrypted"): + try: + key = _expand_key(key_b64) + nonce = mp.id.to_bytes(8, "little") + getattr(mp, "from").to_bytes( + 8, "little" + ) + dec = Cipher(algorithms.AES(key), modes.CTR(nonce)).decryptor() + clear = dec.update(mp.encrypted) + dec.finalize() + data = mesh_pb2.Data() + data.ParseFromString(clear) + if data.portnum or data.payload: + portnum = _portnum_name(data.portnum) + decoded_ok = True + except Exception: + portnum = None + return { + "id": mp.id, + "from": "!%08x" % (getattr(mp, "from") & 0xFFFFFFFF), + "to": "!%08x" % (mp.to & 0xFFFFFFFF), + "chan_hash": mp.channel, + "encrypted": mp.HasField("encrypted"), + "portnum": portnum, + "primary_decodable": decoded_ok, + } + + +def main() -> int: + """Parse arguments, capture datagrams, and write JSONL fixtures. + + Returns: + Process exit code: ``0`` on success, ``2`` for invalid option combos. + """ + ap = argparse.ArgumentParser(description="Capture Mesh-via-UDP fixtures.") + ap.add_argument("--group", default=DEFAULT_GROUP) + ap.add_argument("--port", type=int, default=DEFAULT_PORT) + ap.add_argument( + "--key", default=DEFAULT_KEY_B64, help="primary channel PSK (base64)" + ) + ap.add_argument("--seconds", type=float, default=120.0, help="capture duration") + ap.add_argument("--max", type=int, default=500, help="max datagrams to save") + ap.add_argument( + "--primary-only", + action="store_true", + help="save only packets that decrypt with the default key", + ) + ap.add_argument("--out", default="fixtures.jsonl") + args = ap.parse_args() + + if args.primary_only and not HAVE_DECODE: + print( + "--primary-only needs meshtastic + cryptography installed", file=sys.stderr + ) + return 2 + + sock = open_multicast_socket(args.group, args.port) + print( + f"Listening on {args.group}:{args.port} for {args.seconds:.0f}s " + f"(decode={'on' if HAVE_DECODE else 'off'}) — Ctrl-C to stop early" + ) + + saved = 0 + seen = 0 + primary = 0 + deadline = time.monotonic() + args.seconds + with open(args.out, "w") as fh: + try: + while time.monotonic() < deadline and saved < args.max: + try: + raw, addr = sock.recvfrom(65535) + except socket.timeout: + continue + seen += 1 + info = summarize(raw, args.key) + is_primary = bool(info and info.get("primary_decodable")) + if is_primary: + primary += 1 + if info is not None: + print( + f" #{seen:<4} {info.get('portnum') or '?':<22} " + f"from={info.get('from')} enc={info.get('encrypted')} " + f"primary={is_primary}" + ) + if args.primary_only and not is_primary: + continue + fh.write( + json.dumps( + { + "raw_b64": base64.b64encode(raw).decode("ascii"), + "len": len(raw), + "src": addr[0], + } + ) + + "\n" + ) + saved += 1 + except KeyboardInterrupt: + print("\nstopped") + finally: + try: + sock.close() + except Exception: + pass + + print( + f"\nDone. datagrams seen={seen}, primary-decodable={primary}, " + f"saved={saved} -> {args.out}" + ) + if HAVE_DECODE and seen and primary == 0: + print( + "WARNING: nothing decoded with the default key. Is your primary " + "channel using a custom PSK? Re-run with --key ." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data/tools/compose.udp.pi.yml b/data/tools/compose.udp.pi.yml new file mode 100644 index 0000000..f978199 --- /dev/null +++ b/data/tools/compose.udp.pi.yml @@ -0,0 +1,60 @@ +# Copyright © 2025-26 l5yth & contributors +# Licensed under the Apache License, Version 2.0 (see LICENSE) +# +# Passive UDP ingestor deployment for a Raspberry Pi 5 (arm64). +# +# Runs the ingestor in TRANSPORT=udp mode. It reads the same .env as the +# standard deployment (API_TOKEN etc. are untouched) and only swaps the node +# connection from single-client serial/TCP to passive LAN multicast. +# +# Prerequisites on the Pi (once): +# 1. Get the source onto the Pi (arm64 images are built natively — no QEMU). +# 2. Build the arm64 image from it: +# docker build -f data/Dockerfile -t potato-mesh-ingestor:udp . +# 3. Copy this file to your compose.yml and add PRIMARY_CHANNEL_NAME + +# INGESTOR_NODE_ID to .env. +# 4. docker compose up -d +# +# `network_mode: host` is REQUIRED — multicast (224.0.0.69) cannot reach a +# bridged container. + +services: + ingestor: + image: potato-mesh-ingestor:udp # built locally from source (step 2) + network_mode: host # REQUIRED for LAN multicast reception + environment: + TRANSPORT: ${TRANSPORT:-udp} + PRIMARY_CHANNEL_ONLY: ${PRIMARY_CHANNEL_ONLY:-1} + PRIMARY_CHANNEL_KEY: ${PRIMARY_CHANNEL_KEY:-AQ==} + # Name of channel 0 (the preset name when the radio leaves it blank, e.g. + # MediumFast / LongFast). REQUIRED for primary-only filtering: it is what + # distinguishes the primary channel from a secondary channel that shares + # the default AQ== key. If unset, primary-only mode drops ALL traffic. + PRIMARY_CHANNEL_NAME: ${PRIMARY_CHANNEL_NAME:-} + MESH_UDP_GROUP: ${MESH_UDP_GROUP:-224.0.0.69} + MESH_UDP_PORT: ${MESH_UDP_PORT:-4403} + INGESTOR_NODE_ID: ${INGESTOR_NODE_ID:-} + API_TOKEN: ${API_TOKEN} + INSTANCE_DOMAIN: ${INSTANCE_DOMAIN} + POTATOMESH_INSTANCE: ${POTATOMESH_INSTANCE} + DEBUG: ${DEBUG:-0} + FEDERATION: ${FEDERATION:-1} + PRIVATE: ${PRIVATE:-0} + volumes: + - potatomesh_data:/app/.local/share/potato-mesh + - potatomesh_config:/app/.config/potato-mesh + - potatomesh_logs:/app/logs + restart: unless-stopped + deploy: + resources: + limits: + memory: 256M + cpus: '0.25' + +volumes: + potatomesh_data: + driver: local + potatomesh_config: + driver: local + potatomesh_logs: + driver: local diff --git a/tests/fixtures/mesh_udp/README.md b/tests/fixtures/mesh_udp/README.md new file mode 100644 index 0000000..c942c08 --- /dev/null +++ b/tests/fixtures/mesh_udp/README.md @@ -0,0 +1,14 @@ + + + +# Mesh-via-UDP capture fixtures + +`primary_and_private_capture.jsonl` — 32 real Meshtastic multicast datagrams +captured from a live Station G2 with `data/tools/capture_udp_fixtures.py` +(no filter — all channels). Each line: `{"raw_b64", "len", "src"}` where +`raw_b64` is the raw `MeshPacket` protobuf datagram. + +Composition (validated): 21 primary-channel packets (channel hash 31) that +decrypt with the default key `AQ==`; 11 packets on 5 private channels that do +NOT decrypt with the default key (used to prove the drop path). Portnums on +primary: POSITION, TELEMETRY, TEXT_MESSAGE, TRACEROUTE, NODEINFO, ROUTING. diff --git a/tests/fixtures/mesh_udp/primary_and_private_capture.jsonl b/tests/fixtures/mesh_udp/primary_and_private_capture.jsonl new file mode 100644 index 0000000..23142d7 --- /dev/null +++ b/tests/fixtures/mesh_udp/primary_and_private_capture.jsonl @@ -0,0 +1,32 @@ +{"raw_b64": "DQ8Ard4V/////xgfKinilD7xl2lp9hEtGUFIF80R6Sv/4br7j/FWwBxDDCjQHXtILazxbNBWLTWEoj3bPdJwRmpFAABwwEgCWEBgqv//////////AXgHmAH8AagBAQ==", "len": 94, "src": "192.0.2.1"} +{"raw_b64": "DQYArd4V/////xgfKiPH39dudDGi9vkeNRaI+bJGnLnFhOfH5dE4mwWG3vSl8TDsHzX+L7TOPdhwRmpFAAAAQEgCWEBgqf//////////AXgFmAH8AagBAQ==", "len": 88, "src": "192.0.2.1"} +{"raw_b64": "DQMArd4V/////xgfKipGzYMBG4tQaH6GaPWW64SasabmbOn3Xzl7TvNyDql9bKtzHrjHPjOXaIM1CoLw8D3kcEZqRQAA+EBYQGCu//////////8BeAOYAfwBqAEB", "len": 93, "src": "192.0.2.1"} +{"raw_b64": "DQYArd4V/////xgfKiEKaAmkdOoIhG24PFiN2zZSTZHFw6Ab9vENveuCGyjje441/5czkD3ycEZqRQAAgEBIA1hAYKn//////////wF4BZgB/AGoAQE=", "len": 86, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xgfKg4SiqhGzidk76GPxfhUuTWa/dHHPQJxRmpFAADwQEgCWGRgsf//////////AXgDmAH8AagBAQ==", "len": 67, "src": "192.0.2.1"} +{"raw_b64": "DQgArd4V/////xgfKiGMABfwy9Rb6b1mwfl3AJQA3Cm7MvgKlQEPGLXBZDGA83A1VfnniD0OcUZqRQAAUEBIAVhAYKn//////////wF4BZgB/AGoAQE=", "len": 86, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xhzKhGUuauOnmmvlL8egWG9ao8T4TVzI9HwPR9xRmpFAAAEQUgCWEBgsP//////////AXgDmAH8AagBAQ==", "len": 70, "src": "192.0.2.1"} +{"raw_b64": "DQoArd4V/////xhzKhRQWtudloibkBSfAFVmWVeEm19lVzWsNEJBPS9xRmpFAAAwQUgEWEBgxv//////////AXgHmAH8AagBAQ==", "len": 73, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xjPASoT4ClpfBOBKuWUTJ1H2myZr4BXBzXpD5HDPTBxRmpFAAAQQEgCWEBgpf//////////AXgDmAH8AagBAQ==", "len": 73, "src": "192.0.2.1"} +{"raw_b64": "DQoArd4V/////xhzKhTD3dUcYzqJCY3XzxL67aTsfTh2ATWtNEJBPTFxRmpFAACIwEgGWEBgof//////////AXgHmAH8AagBAQ==", "len": 73, "src": "192.0.2.1"} +{"raw_b64": "DQoArd4V/////xhzKhTEyuI/f8vJbBQpE+6rMAmMjD4IfTWuNEJBPTRxRmpFAAAAQEgGWEBgqf//////////AXgHmAH8AagBAQ==", "len": 73, "src": "192.0.2.1"} +{"raw_b64": "DQkArd4V/////xgfKhghji4yAS+H9Z0039juSUJEs254ZbFFMP01JsKjIj04cUZqRQAAoEBIAlhkYK///////////wF4B5gB/AGoAQE=", "len": 77, "src": "192.0.2.1"} +{"raw_b64": "DRAArd4V/wCt3iot68wgOEetNwfvdIj054uHAwlHntR+8537dvoWkoAzWWNl/C9skWbIKfxoSIkrNX+fQV89OnFGakUAAEC/SANQAVhGYKr//////////wF4BpgB/AGoAQE=", "len": 98, "src": "192.0.2.1"} +{"raw_b64": "DQUArd4V/////xgfKinZxYvIFj9ZKKNM0R5FraY+fn3YRDNjIDLhGCWlEvS2d4K/ZfXbXZHNgTV5r+3RPT5xRmpFAAAAQEgBWEBgrv//////////AXgDmAH8AagBAQ==", "len": 94, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4VCwCt3hgfKhrCb4Cm7s9diCdB1s6lK0tc8anB1SE4GcQ6LzWz9ggnPZ9xRmpFAAC4wEgCUAFYRmCk//////////8BeAOYAfwBqAEB", "len": 81, "src": "192.0.2.1"} +{"raw_b64": "DQsArd4VEgCt3hgfKlo6IbUx22AodbE2P3riKyw2NJkHih+49q32I9nEP8+gTAWJXuMSZ2FUlA7iCVzHcCGrYdAQoNJLtZ2Vx4Ss6QKXVDZ/F8L1jN89PnlEmtcM/FKi4gSMaTUVgx81u3MB8D2kcUZqRQAAwD9QAVhQYLH//////////wF4A5gB/AGoAQE=", "len": 143, "src": "192.0.2.1"} +{"raw_b64": "DQcArd4V/////xgfKhfoOKOfoabMkzmyhk7OPtikxvBDLw5GNDUnSMyMPbFxRmpFAABQwEgEWEBgrv//////////AXgFmAH8AagBAQ==", "len": 76, "src": "192.0.2.1"} +{"raw_b64": "DQ4Ard4V/////xgfKhxKGzOS/9v3ZVqRRYg2kRhdbmtIp8vglOXfDIMYNWNrNvg9xHFGakUAAOA/SAJYQGCq//////////8BeAOYAfwBqAEB", "len": 81, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xhzKiodDICkxFuQX0xBiXjwAHlb8CFCh01/4WegFxjjH48roY8nR1KrB7JL5uo1j/pVaz3QcUZqRQAA4MBIAlhAYKP//////////wF4A5gB/AGoAQE=", "len": 95, "src": "192.0.2.1"} +{"raw_b64": "DQEArd4V/////xgfKiPe12p9IFCJgZINuBQiuZqhIV45Lbz4u5kK2RvrFoVKpqNPaDV9Uo+yPdZxRmpFAADwQFhAYK///////////wF4BZgB/AGoAQE=", "len": 86, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xgfKg5KJWBB/9B3hOy+dm6s9TXASWWOPdhxRmpFAACAPkgCWGRgpP//////////AXgDmAH8AagBAQ==", "len": 67, "src": "192.0.2.1"} +{"raw_b64": "DQ4Ard4V/////xgfKilXPF9+0n9NHWwLDeZu/nEaZQj+pPCc5+ZYJ5/drhqedVmrwqijeH3XGzVkf95APelxRmpFAADgv0gBWEBgqv//////////AXgDmAH8AagBAQ==", "len": 94, "src": "192.0.2.1"} +{"raw_b64": "DQwArd4V/////xgfKi9SFtTr2/+6XSpK8hL5lO8CXVJqct8xqpSNhyFrSxQXsPZw/4c6LPpJERITbYh16DVwIK5MPetxRmpFAAAwQUgGWEBgyP//////////AXgHmAH8AagBAQ==", "len": 100, "src": "192.0.2.1"} +{"raw_b64": "DRMArd4V/////xhVKhexhxwn45yjeVIHjDKFMFYun4rJFGOBKjWKwZqaPexxRmpFAAAMQUgBWEBgrv//////////AXgFmAH8AagBAQ==", "len": 76, "src": "192.0.2.1"} +{"raw_b64": "DQIArd4VEgCt3hgfKg2PpFJ69rP4mU0JfXziNZYdJxY9+3FGakgCWHh4ApgB/AE=", "len": 47, "src": "192.0.2.1"} +{"raw_b64": "DQQArd4V/////xgfKid59EhiXlgN85zMy5AuclZLizzKPE00algvWbjrKoh+3WnWpeZG8/Y1fKUV4D0EckZqRQAAAEFIAlhAYKn//////////wF4B5gB/AGoAQE=", "len": 92, "src": "192.0.2.1"} +{"raw_b64": "DREArd4V/////xgfKiGeYuULCMqaQV/UYmnxUDQShnK0xH0dGRngd5hK6bB0OsI1lnDBHD0RckZqRQAANMFYQGCi//////////8BeAWYAfwBqAEB", "len": 84, "src": "192.0.2.1"} +{"raw_b64": "DQ0Ard4V/////xgfKiF+UJpOYE8AB6LwY6hCsjnBOc2qVDsQ+uranmicrNTRzi41/CGehT0rckZqRQAALEFIAVhAYMf//////////wF4BZgB/AGoAQE=", "len": 86, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4V/////xhNKg4IARIITm9tIG5vbSBIATXlJnBIPTByRmpFAAAEQUgCWEBgsP//////////AXgDmAH8AagBAQ==", "len": 67, "src": "192.0.2.1"} +{"raw_b64": "DQoArd4V/////xhzKi05DsjRDrnwWyTxvRYSBqMKtvSVj6gXU5QPwGsYBbyFEP+oTVHIf4o8MG+C7Go1BH/7PT01ckZqRQAAwL9IBFhAYLD//////////wF4B5gB/AGoAQE=", "len": 98, "src": "192.0.2.1"} +{"raw_b64": "DQoArd4V/////xhzKieXFUPT6IxsKTHoBw1SE2kVYA6IxV5NkO2zXJnVWy8b5rL691HWWEg1l+UTAz1SckZqRQAAMMBIBlhAYKL//////////wF4B5gB/AGoAQE=", "len": 92, "src": "192.0.2.1"} +{"raw_b64": "DRIArd4VDgCt3hgfKhonoa21HTPcULSQ2Ad3Psi+SQGt+ldoZoVPCTU4b3e+PVJyRmpFAAAswUgCUAFYRmCv//////////8BeAOYAfwBqAEB", "len": 81, "src": "192.0.2.1"} diff --git a/tests/test_channels_unit.py b/tests/test_channels_unit.py index 5428136..6281026 100644 --- a/tests/test_channels_unit.py +++ b/tests/test_channels_unit.py @@ -472,3 +472,40 @@ class TestRegisterChannel: channels.register_channel(1, "Chat") assert channels.channel_name(0) == "LongFast" assert channels.channel_name(1) == "Chat" + + +# --------------------------------------------------------------------------- +# is_primary_channel / is_primary_only +# --------------------------------------------------------------------------- + + +class TestIsPrimaryChannel: + """Tests for :func:`channels.is_primary_channel`.""" + + def test_is_primary_channel_true_for_zero(self): + """Channel index 0 (PRIMARY) is reported as primary.""" + assert channels.is_primary_channel(0) is True + + def test_is_primary_channel_false_for_nonzero(self): + """Any non-zero channel index is not primary.""" + assert channels.is_primary_channel(1) is False + + def test_is_primary_channel_false_for_none(self): + """A missing channel index is not primary.""" + assert channels.is_primary_channel(None) is False + + +class TestIsPrimaryOnly: + """Tests for :func:`channels.is_primary_only`.""" + + def test_is_primary_only_reads_config(self, monkeypatch): + """Reflects the live value of ``config.PRIMARY_CHANNEL_ONLY``.""" + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", True) + assert channels.is_primary_only() is True + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", False) + assert channels.is_primary_only() is False + + def test_is_primary_only_defaults_false_when_attr_missing(self, monkeypatch): + """Falls back to ``False`` if ``config.PRIMARY_CHANNEL_ONLY`` is absent.""" + monkeypatch.delattr(config, "PRIMARY_CHANNEL_ONLY", raising=False) + assert channels.is_primary_only() is False diff --git a/tests/test_config_unit.py b/tests/test_config_unit.py index c7ed9af..5047fa3 100644 --- a/tests/test_config_unit.py +++ b/tests/test_config_unit.py @@ -369,3 +369,251 @@ class TestParseLoraFreqEnv: monkeypatch.delenv("FREQUENCY", raising=False) importlib.reload(config) assert config.LORA_FREQ is None + + +# --------------------------------------------------------------------------- +# TRANSPORT / PRIMARY_CHANNEL_ONLY / PRIMARY_CHANNEL_KEY / MESH_UDP_* / +# INGESTOR_NODE_ID +# --------------------------------------------------------------------------- + +# Every new env-driven config name introduced for the passive UDP transport. +_UDP_ENV_VARS = ( + "TRANSPORT", + "PRIMARY_CHANNEL_ONLY", + "PRIMARY_CHANNEL_KEY", + "PRIMARY_CHANNEL_NAME", + "MESH_UDP_GROUP", + "MESH_UDP_PORT", + "INGESTOR_NODE_ID", +) + + +def _clear_udp_env(monkeypatch) -> None: + """Delete every UDP-transport env var so tests start from a clean slate.""" + for name in _UDP_ENV_VARS: + monkeypatch.delenv(name, raising=False) + + +class TestTransportConfig: + """Tests for :data:`config.TRANSPORT`.""" + + @pytest.fixture(autouse=True) + def _isolate(self, monkeypatch): + """Clear UDP-transport env vars and reload config to defaults after each test. + + Prevents state set by one test (or leaked into later test modules + sharing this process) from affecting subsequent tests, since + ``config`` attributes set via :func:`importlib.reload` persist in + ``sys.modules`` beyond the env vars that produced them. + """ + import importlib + + _clear_udp_env(monkeypatch) + yield + _clear_udp_env(monkeypatch) + importlib.reload(config) + + def test_transport_defaults_to_api(self, monkeypatch): + """TRANSPORT defaults to 'api' when unset.""" + import importlib + + monkeypatch.delenv("TRANSPORT", raising=False) + importlib.reload(config) + assert config.TRANSPORT == "api" + + def test_transport_udp_lowercased(self, monkeypatch): + """TRANSPORT values are lower-cased; 'UDP' becomes 'udp'.""" + import importlib + + monkeypatch.setenv("TRANSPORT", "UDP") + importlib.reload(config) + assert config.TRANSPORT == "udp" + + def test_transport_invalid_raises_value_error(self, monkeypatch): + """An unrecognised TRANSPORT value raises ValueError at import time.""" + import importlib + + monkeypatch.setenv("TRANSPORT", "bogus") + with pytest.raises(ValueError, match="Unknown TRANSPORT"): + importlib.reload(config) + + +class TestPrimaryChannelOnlyConfig: + """Tests for :data:`config.PRIMARY_CHANNEL_ONLY`.""" + + @pytest.fixture(autouse=True) + def _isolate(self, monkeypatch): + """Clear UDP-transport env vars and reload config to defaults after each test.""" + import importlib + + _clear_udp_env(monkeypatch) + yield + _clear_udp_env(monkeypatch) + importlib.reload(config) + + def test_primary_channel_only_flag(self, monkeypatch): + """PRIMARY_CHANNEL_ONLY is True when the env var is exactly '1'.""" + import importlib + + monkeypatch.setenv("PRIMARY_CHANNEL_ONLY", "1") + importlib.reload(config) + assert config.PRIMARY_CHANNEL_ONLY is True + + def test_primary_channel_only_defaults_false(self, monkeypatch): + """PRIMARY_CHANNEL_ONLY defaults to False when unset.""" + import importlib + + monkeypatch.delenv("PRIMARY_CHANNEL_ONLY", raising=False) + importlib.reload(config) + assert config.PRIMARY_CHANNEL_ONLY is False + + def test_primary_channel_only_false_for_non_one_values(self, monkeypatch): + """Any value other than the literal '1' leaves the flag False.""" + import importlib + + monkeypatch.setenv("PRIMARY_CHANNEL_ONLY", "true") + importlib.reload(config) + assert config.PRIMARY_CHANNEL_ONLY is False + + +class TestPrimaryChannelNameConfig: + """Tests for :data:`config.PRIMARY_CHANNEL_NAME`.""" + + @pytest.fixture(autouse=True) + def _isolate(self, monkeypatch): + """Clear UDP-transport env vars and reload config to defaults after each test.""" + import importlib + + _clear_udp_env(monkeypatch) + yield + _clear_udp_env(monkeypatch) + importlib.reload(config) + + def test_defaults_to_empty_string(self, monkeypatch): + """PRIMARY_CHANNEL_NAME defaults to '' (fail-closed sentinel) when unset.""" + import importlib + + monkeypatch.delenv("PRIMARY_CHANNEL_NAME", raising=False) + importlib.reload(config) + assert config.PRIMARY_CHANNEL_NAME == "" + + def test_reads_and_strips_env_value(self, monkeypatch): + """A configured name is read and surrounding whitespace stripped.""" + import importlib + + monkeypatch.setenv("PRIMARY_CHANNEL_NAME", " MediumFast ") + importlib.reload(config) + assert config.PRIMARY_CHANNEL_NAME == "MediumFast" + + def test_exported_in_all(self): + """PRIMARY_CHANNEL_NAME is part of the module's public surface.""" + assert "PRIMARY_CHANNEL_NAME" in config.__all__ + + +class TestUdpTransportDefaults: + """Tests for PRIMARY_CHANNEL_KEY, MESH_UDP_GROUP, MESH_UDP_PORT, INGESTOR_NODE_ID.""" + + @pytest.fixture(autouse=True) + def _isolate(self, monkeypatch): + """Clear UDP-transport env vars and reload config to defaults after each test.""" + import importlib + + _clear_udp_env(monkeypatch) + yield + _clear_udp_env(monkeypatch) + importlib.reload(config) + + def test_udp_defaults(self, monkeypatch): + """All four vars fall back to their documented defaults when unset.""" + import importlib + + for k in ( + "PRIMARY_CHANNEL_KEY", + "MESH_UDP_GROUP", + "MESH_UDP_PORT", + "INGESTOR_NODE_ID", + ): + monkeypatch.delenv(k, raising=False) + importlib.reload(config) + assert config.PRIMARY_CHANNEL_KEY == "AQ==" + assert config.MESH_UDP_GROUP == "224.0.0.69" + assert config.MESH_UDP_PORT == 4403 + assert config.INGESTOR_NODE_ID is None + + def test_primary_channel_key_custom_value(self, monkeypatch): + """A custom PRIMARY_CHANNEL_KEY overrides the default.""" + import importlib + + monkeypatch.setenv("PRIMARY_CHANNEL_KEY", "c3VwZXJzZWNyZXQ=") + importlib.reload(config) + assert config.PRIMARY_CHANNEL_KEY == "c3VwZXJzZWNyZXQ=" + + def test_primary_channel_key_blank_falls_back_to_default(self, monkeypatch): + """A whitespace-only PRIMARY_CHANNEL_KEY falls back to the default.""" + import importlib + + monkeypatch.setenv("PRIMARY_CHANNEL_KEY", " ") + importlib.reload(config) + assert config.PRIMARY_CHANNEL_KEY == "AQ==" + + def test_mesh_udp_group_custom_value(self, monkeypatch): + """A custom MESH_UDP_GROUP overrides the default.""" + import importlib + + monkeypatch.setenv("MESH_UDP_GROUP", "239.1.2.3") + importlib.reload(config) + assert config.MESH_UDP_GROUP == "239.1.2.3" + + def test_mesh_udp_group_blank_falls_back_to_default(self, monkeypatch): + """A whitespace-only MESH_UDP_GROUP falls back to the default.""" + import importlib + + monkeypatch.setenv("MESH_UDP_GROUP", " ") + importlib.reload(config) + assert config.MESH_UDP_GROUP == "224.0.0.69" + + def test_mesh_udp_port_custom_value(self, monkeypatch): + """A custom MESH_UDP_PORT is parsed to an int.""" + import importlib + + monkeypatch.setenv("MESH_UDP_PORT", "5000") + importlib.reload(config) + assert config.MESH_UDP_PORT == 5000 + assert isinstance(config.MESH_UDP_PORT, int) + + def test_mesh_udp_port_blank_falls_back_to_default(self, monkeypatch): + """A whitespace/empty MESH_UDP_PORT falls back to 4403 without raising. + + A blank value is common in ``.env`` files; parsing it with a bare + ``int()`` would raise ``ValueError`` at import and stop the service. + """ + import importlib + + monkeypatch.setenv("MESH_UDP_PORT", " ") + importlib.reload(config) # must not raise + assert config.MESH_UDP_PORT == 4403 + + def test_ingestor_node_id_custom_value(self, monkeypatch): + """A custom INGESTOR_NODE_ID is returned stripped of surrounding whitespace.""" + import importlib + + monkeypatch.setenv("INGESTOR_NODE_ID", " !deadbeef ") + importlib.reload(config) + assert config.INGESTOR_NODE_ID == "!deadbeef" + + def test_ingestor_node_id_blank_is_none(self, monkeypatch): + """A whitespace-only INGESTOR_NODE_ID normalises to None.""" + import importlib + + monkeypatch.setenv("INGESTOR_NODE_ID", " ") + importlib.reload(config) + assert config.INGESTOR_NODE_ID is None + + +class TestUdpTransportAllExports: + """Tests that the new config names are exported via ``__all__``.""" + + def test_new_names_in_all(self): + """TRANSPORT, PRIMARY_CHANNEL_ONLY, and the UDP vars are in __all__.""" + for name in _UDP_ENV_VARS: + assert name in config.__all__ diff --git a/tests/test_handlers_unit.py b/tests/test_handlers_unit.py index b283811..57b5082 100644 --- a/tests/test_handlers_unit.py +++ b/tests/test_handlers_unit.py @@ -1331,3 +1331,148 @@ class TestCoerceEmojiStringFailure: raise RuntimeError("boom") assert generic_mod._coerce_emoji_codepoint(Boom()) is None + + +# --------------------------------------------------------------------------- +# store_packet_dict — PRIMARY_CHANNEL_ONLY guard +# --------------------------------------------------------------------------- + + +class TestStorePacketDictPrimaryChannelGuard: + """Tests for the ``PRIMARY_CHANNEL_ONLY`` guard in + :func:`handlers.store_packet_dict`. + + Mirrors the existing disallowed-channel / hidden-channel guard tests: + build a message packet on a non-primary channel and confirm it is + dropped (no queue POST, ``_record_ignored_packet`` called with + ``reason="non-primary-channel"``) only when + ``config.PRIMARY_CHANNEL_ONLY`` is enabled. + """ + + def _make_packet(self, *, channel: int, pkt_id: int = 4242) -> dict: + return { + "id": pkt_id, + "rxTime": 1_700_000_000, + "from": "!sender", + "to": "^all", + "channel": channel, + "decoded": {"text": "secondary channel msg", "portnum": 1}, + } + + def test_drops_secondary_channel_when_flag_enabled(self, monkeypatch): + """Non-primary channel packet is dropped when the flag is on.""" + import data.mesh_ingestor.queue as q + + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", True) + monkeypatch.setattr(config, "ALLOWED_CHANNELS", ()) + monkeypatch.setattr(config, "HIDDEN_CHANNELS", ()) + monkeypatch.setattr(config, "DEBUG", False) + + sent = [] + ignored = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append( + (path, payload) + ) + monkeypatch.setattr( + ignored_mod, + "_record_ignored_packet", + lambda packet, *, reason: ignored.append(reason), + ) + try: + handlers.store_packet_dict(self._make_packet(channel=3)) + finally: + q._queue_post_json = original + + assert sent == [] + assert ignored == ["non-primary-channel"] + + def test_drops_secondary_channel_logs_when_debug_enabled(self, monkeypatch, capsys): + """The debug log branch fires (and is skipped when DEBUG is off, per + the sibling test above) so both sides of the ``if config.DEBUG`` + branch are covered.""" + import data.mesh_ingestor.queue as q + + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", True) + monkeypatch.setattr(config, "ALLOWED_CHANNELS", ()) + monkeypatch.setattr(config, "HIDDEN_CHANNELS", ()) + monkeypatch.setattr(config, "DEBUG", True) + + sent = [] + ignored = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append( + (path, payload) + ) + monkeypatch.setattr( + ignored_mod, + "_record_ignored_packet", + lambda packet, *, reason: ignored.append(reason), + ) + capsys.readouterr() + try: + handlers.store_packet_dict(self._make_packet(channel=3)) + finally: + q._queue_post_json = original + + assert sent == [] + assert ignored == ["non-primary-channel"] + assert "Ignored packet on non-primary channel" in capsys.readouterr().out + + def test_allows_primary_channel_when_flag_enabled(self, monkeypatch): + """Channel 0 (PRIMARY) is never dropped by this guard, flag on or + off.""" + import data.mesh_ingestor.queue as q + + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", True) + monkeypatch.setattr(config, "ALLOWED_CHANNELS", ()) + monkeypatch.setattr(config, "HIDDEN_CHANNELS", ()) + monkeypatch.setattr(config, "DEBUG", False) + + sent = [] + ignored = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append( + (path, payload) + ) + monkeypatch.setattr( + ignored_mod, + "_record_ignored_packet", + lambda packet, *, reason: ignored.append(reason), + ) + try: + handlers.store_packet_dict(self._make_packet(channel=0)) + finally: + q._queue_post_json = original + + assert any(path == "/api/messages" for path, _ in sent) + assert "non-primary-channel" not in ignored + + def test_does_not_drop_secondary_channel_when_flag_disabled(self, monkeypatch): + """With the flag off, a secondary-channel packet is NOT dropped by + this guard (it proceeds to be queued).""" + import data.mesh_ingestor.queue as q + + monkeypatch.setattr(config, "PRIMARY_CHANNEL_ONLY", False) + monkeypatch.setattr(config, "ALLOWED_CHANNELS", ()) + monkeypatch.setattr(config, "HIDDEN_CHANNELS", ()) + monkeypatch.setattr(config, "DEBUG", False) + + sent = [] + ignored = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append( + (path, payload) + ) + monkeypatch.setattr( + ignored_mod, + "_record_ignored_packet", + lambda packet, *, reason: ignored.append(reason), + ) + try: + handlers.store_packet_dict(self._make_packet(channel=3)) + finally: + q._queue_post_json = original + + assert any(path == "/api/messages" for path, _ in sent) + assert "non-primary-channel" not in ignored diff --git a/tests/test_meshtastic_udp_decode_unit.py b/tests/test_meshtastic_udp_decode_unit.py new file mode 100644 index 0000000..70ffe7c --- /dev/null +++ b/tests/test_meshtastic_udp_decode_unit.py @@ -0,0 +1,575 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for :mod:`data.mesh_ingestor.protocols.meshtastic_udp_decode`. + +Two layers of coverage: + +1. **Real-fixture tests** replay 32 genuine Meshtastic multicast datagrams + captured from a live Station G2 (``tests/fixtures/mesh_udp``) and assert + the decrypt heuristic accepts exactly the 21 primary-channel packets and + drops exactly the 11 private-channel packets, with the mapping producing + sane, round-trippable output for every accepted packet. +2. **Synthetic tests** build encrypted packets in-process to exercise every + line and branch of the module (key expansion edge cases, decrypt failure + paths, and every optional field in the packet-dict mapping). +""" + +from __future__ import annotations + +import base64 +import binascii +import json +import os +import sys +import time +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from meshtastic.protobuf import mesh_pb2, portnums_pb2 + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from data.mesh_ingestor.protocols import meshtastic_udp_decode as udp + +DEFAULT_KEY = "AQ==" +"""Base64 form of the Meshtastic default 1-byte primary-channel PSK.""" + +FIXTURE_PATH = os.path.join( + os.path.dirname(__file__), + "fixtures", + "mesh_udp", + "primary_and_private_capture.jsonl", +) +"""Path to the real captured-datagram fixture, resolved relative to this file.""" + +EXPECTED_PORTNUMS = { + "POSITION_APP", + "TELEMETRY_APP", + "TEXT_MESSAGE_APP", + "TRACEROUTE_APP", + "NODEINFO_APP", + "ROUTING_APP", +} +"""Portnums documented in the fixture README as present on the primary channel.""" + + +def _encrypt(data: "mesh_pb2.Data", mp: "mesh_pb2.MeshPacket", key_b64: str) -> bytes: + """Encrypt *data* the same way a Meshtastic node would for *mp*. + + Mirrors the production nonce construction (``id`` then ``from``, both + little-endian 8-byte) so tests can build round-trippable fixtures without + importing any private module internals. + """ + key = udp.expand_default_key(key_b64) + nonce = mp.id.to_bytes(8, "little") + getattr(mp, "from").to_bytes(8, "little") + enc = Cipher(algorithms.AES(key), modes.CTR(nonce)).encryptor() + return enc.update(data.SerializeToString()) + enc.finalize() + + +def _load_fixture_packets() -> list["mesh_pb2.MeshPacket"]: + """Parse every line of the real-capture fixture into a ``MeshPacket``.""" + packets = [] + with open(FIXTURE_PATH, "r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + record = json.loads(line) + raw = base64.b64decode(record["raw_b64"]) + mp = mesh_pb2.MeshPacket() + mp.ParseFromString(raw) + packets.append(mp) + return packets + + +# --------------------------------------------------------------------------- +# Real-fixture tests +# --------------------------------------------------------------------------- + + +class TestRealCaptureFixture: + """Replays genuine captured datagrams through the decode pipeline.""" + + def test_all_datagrams_parse_as_meshpacket(self): + """All 32 captured datagrams parse cleanly as ``MeshPacket``.""" + packets = _load_fixture_packets() + assert len(packets) == 32 + + def test_decrypt_accepts_primary_and_drops_private(self): + """Decrypt accepts exactly the 21 channel-31 packets, drops the rest.""" + packets = _load_fixture_packets() + accepted = 0 + dropped = 0 + for mp in packets: + data = udp.decrypt_meshpacket(mp, DEFAULT_KEY) + if mp.channel == 31: + assert data is not None + accepted += 1 + else: + assert data is None + dropped += 1 + assert accepted == 21 + assert dropped == 11 + + def test_accepted_packets_map_to_known_portnums_and_roundtrip_payload(self): + """Every accepted packet maps to a known portnum with a round-trippable payload.""" + packets = _load_fixture_packets() + seen_portnums = set() + accepted_count = 0 + for mp in packets: + data = udp.decrypt_meshpacket(mp, DEFAULT_KEY) + if data is None: + continue + accepted_count += 1 + mp.decoded.CopyFrom(data) + packet_dict = udp.meshpacket_to_packet_dict(mp) + portnum_name = packet_dict["decoded"]["portnum"] + assert isinstance(portnum_name, str) and portnum_name + assert base64.b64decode(packet_dict["decoded"]["payload"]) == data.payload + seen_portnums.add(portnum_name) + assert accepted_count == 21 + assert seen_portnums <= EXPECTED_PORTNUMS + + +# --------------------------------------------------------------------------- +# expand_default_key +# --------------------------------------------------------------------------- + + +class TestExpandDefaultKey: + """Tests for :func:`expand_default_key`.""" + + def test_one_byte_psk_expands_to_16_bytes(self): + """A 1-byte PSK in range 0x01..0x07 is prefixed to a 16-byte AES key.""" + key = udp.expand_default_key(DEFAULT_KEY) + assert len(key) == 16 + assert key.hex().endswith("01") + assert key == bytes.fromhex("d4f1bb3a20290759f0bcffabcf4e69") + b"\x01" + + def test_multi_byte_key_passthrough(self): + """A key that already decodes to more than 1 byte is returned as-is.""" + raw = b"\x00" * 16 + key_b64 = base64.b64encode(raw).decode() + assert udp.expand_default_key(key_b64) == raw + + def test_one_byte_out_of_range_passthrough(self): + """A 1-byte value outside 0x01..0x07 is NOT treated as a default PSK.""" + raw = b"\x08" + key_b64 = base64.b64encode(raw).decode() + assert udp.expand_default_key(key_b64) == raw + assert len(udp.expand_default_key(key_b64)) == 1 + + def test_bad_base64_raises(self): + """Malformed base64 propagates a decode error to the caller.""" + with pytest.raises(binascii.Error): + udp.expand_default_key("not-valid-base64!!!") + + +# --------------------------------------------------------------------------- +# decrypt_meshpacket +# --------------------------------------------------------------------------- + + +class TestDecryptMeshpacket: + """Tests for :func:`decrypt_meshpacket`.""" + + def test_round_trip_text(self): + """A packet encrypted with the default key decrypts back to its payload.""" + mp = mesh_pb2.MeshPacket() + mp.id = 0x1234 + setattr(mp, "from", 0x849B7154) + data = mesh_pb2.Data( + portnum=portnums_pb2.PortNum.TEXT_MESSAGE_APP, payload=b"hi" + ) + mp.encrypted = _encrypt(data, mp, DEFAULT_KEY) + + out = udp.decrypt_meshpacket(mp, DEFAULT_KEY) + + assert out is not None + assert out.payload == b"hi" + assert out.portnum == portnums_pb2.PortNum.TEXT_MESSAGE_APP + + def test_wrong_key_returns_none_deterministically(self): + """Decrypting with an incorrect (but well-formed) key returns ``None``. + + Both the plaintext and the wrong key are fixed constants, so AES-CTR + (a deterministic stream cipher) always produces the same garbage + bytes on every run/platform -- this test carries no random or + time-based inputs and is fully reproducible. + """ + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 2) + data = mesh_pb2.Data( + portnum=portnums_pb2.PortNum.NODEINFO_APP, payload=b"\x08\x01" + ) + mp.encrypted = _encrypt(data, mp, DEFAULT_KEY) + + other_key = base64.b64encode(b"\x00" * 16).decode() + + # Re-run to demonstrate the result is stable, not flaky. + for _ in range(3): + assert udp.decrypt_meshpacket(mp, other_key) is None + + def test_decrypt_failure_path_returns_none_on_bad_key(self): + """An invalid key string is caught internally and yields ``None``.""" + mp = mesh_pb2.MeshPacket() + mp.id = 7 + setattr(mp, "from", 8) + data = mesh_pb2.Data( + portnum=portnums_pb2.PortNum.TEXT_MESSAGE_APP, payload=b"hi" + ) + mp.encrypted = _encrypt(data, mp, DEFAULT_KEY) + + assert udp.decrypt_meshpacket(mp, "not-valid-base64!!!") is None + + def test_empty_decoded_data_is_treated_as_private_channel(self): + """A cleanly-parsed but empty ``Data`` (portnum 0, no payload) is dropped.""" + mp = mesh_pb2.MeshPacket() + mp.id = 55 + setattr(mp, "from", 66) + data = mesh_pb2.Data() # all defaults: portnum == 0, payload == b"" + mp.encrypted = _encrypt(data, mp, DEFAULT_KEY) + + assert udp.decrypt_meshpacket(mp, DEFAULT_KEY) is None + + def test_zero_portnum_with_payload_is_not_dropped(self): + """Portnum 0 with a non-empty payload is NOT treated as private-channel noise.""" + mp = mesh_pb2.MeshPacket() + mp.id = 42 + setattr(mp, "from", 99) + data = mesh_pb2.Data(portnum=0, payload=b"x") + mp.encrypted = _encrypt(data, mp, DEFAULT_KEY) + + out = udp.decrypt_meshpacket(mp, DEFAULT_KEY) + + assert out is not None + assert out.payload == b"x" + + +# --------------------------------------------------------------------------- +# _node_id +# --------------------------------------------------------------------------- + + +class TestNodeId: + """Tests for the private :func:`_node_id` helper.""" + + def test_broadcast_num_maps_to_all(self): + """The reserved broadcast address maps to ``^all``.""" + assert udp._node_id(0xFFFFFFFF) == "^all" + + def test_unicast_num_maps_to_bang_hex(self): + """A regular node number maps to canonical ``!xxxxxxxx`` form.""" + assert udp._node_id(0x849B7154) == "!849b7154" + + def test_masks_to_32_bits(self): + """Values outside the 32-bit range are masked before formatting.""" + assert udp._node_id(0x1_849B7154) == "!849b7154" + + +# --------------------------------------------------------------------------- +# meshpacket_to_packet_dict +# --------------------------------------------------------------------------- + + +class TestMeshpacketToPacketDict: + """Tests for :func:`meshpacket_to_packet_dict`.""" + + def test_text_sets_decoded_text_and_broadcast_to(self): + """A text packet gets a decoded ``text`` field and broadcast ``toId``.""" + mp = mesh_pb2.MeshPacket() + mp.id = 9 + setattr(mp, "from", 0x849B7154) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.TEXT_MESSAGE_APP + mp.decoded.payload = "Guten Morgen!".encode("utf-8") + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["from"] == 0x849B7154 + assert d["fromId"] == "!849b7154" + assert d["to"] == 0xFFFFFFFF + assert d["toId"] == "^all" + assert d["id"] == 9 + assert d["channel"] == 0 + assert d["decoded"]["portnum"] == "TEXT_MESSAGE_APP" + assert d["decoded"]["text"] == "Guten Morgen!" + assert base64.b64decode(d["decoded"]["payload"]) == "Guten Morgen!".encode( + "utf-8" + ) + + def test_text_payload_invalid_utf8_is_replaced_not_raised(self): + """Non-UTF-8 bytes in a text payload are decoded with replacement, not raised.""" + mp = mesh_pb2.MeshPacket() + mp.id = 10 + setattr(mp, "from", 1) + mp.to = 2 + mp.decoded.portnum = portnums_pb2.PortNum.TEXT_MESSAGE_APP + mp.decoded.payload = b"\xff\xfe" + + d = udp.meshpacket_to_packet_dict(mp) + + assert "�" in d["decoded"]["text"] + + def test_non_text_sets_base64_payload_and_unicast_to(self): + """A non-text portnum does not get a ``text`` field; ``toId`` is unicast.""" + mp = mesh_pb2.MeshPacket() + mp.id = 5 + setattr(mp, "from", 0x111) + mp.to = 0x222 + mp.decoded.portnum = portnums_pb2.PortNum.NODEINFO_APP + mp.decoded.payload = b"\x08\x01" + + d = udp.meshpacket_to_packet_dict(mp) + + assert base64.b64decode(d["decoded"]["payload"]) == b"\x08\x01" + assert d["toId"] == "!00000222" + assert "text" not in d["decoded"] + + def test_optional_fields_absent_when_falsy(self): + """``rxSnr``/``rxRssi``/``hopLimit`` are omitted when the source field is falsy.""" + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.to = 2 + mp.decoded.portnum = portnums_pb2.PortNum.ROUTING_APP + mp.decoded.payload = b"" + + d = udp.meshpacket_to_packet_dict(mp) + + assert "rxSnr" not in d + assert "rxRssi" not in d + assert "hopLimit" not in d + + def test_optional_fields_present_when_truthy(self): + """``rxSnr``/``rxRssi``/``hopLimit`` are included when the source field is set.""" + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.to = 2 + mp.decoded.portnum = portnums_pb2.PortNum.ROUTING_APP + mp.decoded.payload = b"" + mp.rx_snr = 7.5 + mp.rx_rssi = -42 + mp.hop_limit = 3 + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["rxSnr"] == pytest.approx(7.5) + assert d["rxRssi"] == -42 + assert d["hopLimit"] == 3 + + def test_rx_time_present_uses_packet_value(self): + """A non-zero ``rx_time`` on the packet is used verbatim.""" + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.to = 2 + mp.decoded.portnum = portnums_pb2.PortNum.ROUTING_APP + mp.decoded.payload = b"" + mp.rx_time = 1717000000 + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["rxTime"] == 1717000000 + + def test_rx_time_absent_falls_back_to_now(self, monkeypatch): + """A zero (unset) ``rx_time`` falls back to the current wall-clock time.""" + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.to = 2 + mp.decoded.portnum = portnums_pb2.PortNum.ROUTING_APP + mp.decoded.payload = b"" + assert mp.rx_time == 0 + + monkeypatch.setattr(udp.time, "time", lambda: 1234567890.0) + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["rxTime"] == 1234567890 + + +class TestXorHash: + """Tests for the private :func:`_xor_hash` byte-fold helper.""" + + def test_empty_is_zero(self): + """The XOR-fold of no bytes is 0.""" + assert udp._xor_hash(b"") == 0 + + def test_single_byte_is_itself(self): + """The XOR-fold of one byte is that byte.""" + assert udp._xor_hash(b"\x2a") == 0x2A + + def test_multiple_bytes_fold(self): + """0x01 ^ 0x02 ^ 0x04 == 0x07.""" + assert udp._xor_hash(b"\x01\x02\x04") == 0x07 + + def test_repeated_byte_cancels(self): + """A byte XOR-ed with itself cancels to 0.""" + assert udp._xor_hash(b"\xab\xab") == 0 + + +class TestChannelHash: + """Tests for :func:`channel_hash`, the Meshtastic ``generateHash`` mirror.""" + + def test_mediumfast_default_key_is_31(self): + """The real RGW1 primary (MediumFast + AQ==) hashes to 31 (0x1F). + + This value is cross-checked against the 21 primary-channel datagrams in + the real-capture fixture, every one of which carries channel hash 31. + """ + assert udp.channel_hash("MediumFast", "AQ==") == 31 + + def test_longfast_default_key_is_8(self): + """The Meshtastic global default (LongFast + AQ==) hashes to 8.""" + assert udp.channel_hash("LongFast", "AQ==") == 8 + + def test_empty_name_is_key_hash_only(self): + """A blank name contributes 0, so the hash is the key's XOR-fold alone.""" + key_hash = udp._xor_hash(udp.expand_default_key("AQ==")) + assert udp.channel_hash("", "AQ==") == key_hash + + def test_same_key_different_names_differ(self): + """Two channels sharing the default key still hash differently by name. + + This is the property that lets the UDP transport separate a PRIMARY + channel from a SECONDARY channel that was created with the same default + ``AQ==`` key: decryptability is identical, but the hashes differ. + """ + assert udp.channel_hash("MediumFast", "AQ==") != udp.channel_hash( + "Private", "AQ==" + ) + + def test_matches_manual_xor_formula(self): + """channel_hash == xorHash(name) ^ xorHash(expanded_key).""" + name, key = "ShortFast", "AQ==" + expected = udp._xor_hash(name.encode("utf-8")) ^ udp._xor_hash( + udp.expand_default_key(key) + ) + assert udp.channel_hash(name, key) == expected + + +class TestEnrichDecoded: + """Tests that :func:`meshpacket_to_packet_dict` reproduces the library's + decoded sub-dicts so downstream handlers behave identically to the API path. + """ + + def test_position_payload_enriched_with_coordinates(self): + """A POSITION packet gains a ``decoded['position']`` with scaled coords.""" + from meshtastic.protobuf import mesh_pb2 as m + + pos = m.Position(latitude_i=449052672, longitude_i=-932446208, altitude=265) + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 0xABCCBB6C) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.POSITION_APP + mp.decoded.payload = pos.SerializeToString() + + d = udp.meshpacket_to_packet_dict(mp) + + assert "position" in d["decoded"] + assert d["decoded"]["position"]["latitudeI"] == 449052672 + assert d["decoded"]["position"]["longitudeI"] == -932446208 + + def test_telemetry_payload_enriched(self): + """A TELEMETRY packet gains a ``decoded['telemetry']`` section.""" + from meshtastic.protobuf import telemetry_pb2 + + tel = telemetry_pb2.Telemetry( + device_metrics=telemetry_pb2.DeviceMetrics(battery_level=87, voltage=4.1) + ) + mp = mesh_pb2.MeshPacket() + mp.id = 2 + setattr(mp, "from", 1) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.TELEMETRY_APP + mp.decoded.payload = tel.SerializeToString() + + d = udp.meshpacket_to_packet_dict(mp) + + assert "telemetry" in d["decoded"] + assert d["decoded"]["telemetry"]["deviceMetrics"]["batteryLevel"] == 87 + + def test_text_packet_not_enriched_with_factory_section(self): + """TEXT_MESSAGE_APP has no protobuf factory: only ``text`` is added.""" + mp = mesh_pb2.MeshPacket() + mp.id = 3 + setattr(mp, "from", 1) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.TEXT_MESSAGE_APP + mp.decoded.payload = b"hello" + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["decoded"]["text"] == "hello" + # No factory-derived section keys beyond portnum/payload/text. + assert set(d["decoded"]) == {"portnum", "payload", "text"} + + def test_malformed_subpayload_is_swallowed(self): + """A POSITION portnum with a non-Position payload must not raise. + + The packet still flows with its ``portnum``/``payload``; the ``position`` + section is simply absent, matching how a library packet behaves when the + firmware could not decode the sub-message. + """ + mp = mesh_pb2.MeshPacket() + mp.id = 4 + setattr(mp, "from", 1) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.POSITION_APP + # A single 0xFF byte is not a valid Position wire message. + mp.decoded.payload = b"\xff" + + d = udp.meshpacket_to_packet_dict(mp) + + assert d["decoded"]["portnum"] == "POSITION_APP" + assert "position" not in d["decoded"] + + def test_unknown_portnum_without_factory_is_left_alone(self): + """An UNKNOWN_APP (portnum 0) payload adds no factory section.""" + mp = mesh_pb2.MeshPacket() + mp.id = 5 + setattr(mp, "from", 1) + mp.to = 0xFFFFFFFF + mp.decoded.portnum = portnums_pb2.PortNum.UNKNOWN_APP + mp.decoded.payload = b"\x01\x02" + + d = udp.meshpacket_to_packet_dict(mp) + + assert set(d["decoded"]) == {"portnum", "payload"} + + def test_out_of_enum_portnum_maps_to_sentinel_without_raising(self): + """A portnum with no enum name yields ``UNKNOWN_APP`` instead of raising. + + Guards the DoS where ``PortNum.Name()`` raised ``ValueError`` on an + out-of-range portnum (newer firmware, or attacker garbage) and killed + the receive thread. + """ + mp = mesh_pb2.MeshPacket() + mp.id = 6 + setattr(mp, "from", 1) + mp.to = 0xFFFFFFFF + # proto3 open enums accept arbitrary int32 values on the wire. + mp.decoded.portnum = 99999 + mp.decoded.payload = b"\x01" + + d = udp.meshpacket_to_packet_dict(mp) # must not raise + + assert d["decoded"]["portnum"] == "UNKNOWN_APP" diff --git a/tests/test_meshtastic_udp_socket_unit.py b/tests/test_meshtastic_udp_socket_unit.py new file mode 100644 index 0000000..86144ab --- /dev/null +++ b/tests/test_meshtastic_udp_socket_unit.py @@ -0,0 +1,159 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for :mod:`data.mesh_ingestor.protocols.meshtastic_udp_socket`. + +Every test replaces ``socket.socket`` with an in-process fake that records +calls instead of touching the network, so the suite never opens a real +socket or requires multicast-capable hardware/CI sandboxing. +""" + +from __future__ import annotations + +import socket as real_socket +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from data.mesh_ingestor.protocols import meshtastic_udp_socket as udp_socket + + +class FakeSock: + """Records socket calls instead of touching a real network socket.""" + + def __init__(self, raise_on_reuseport: bool = False) -> None: + """Initialize the fake with an empty call log. + + Args: + raise_on_reuseport: When ``True``, calling ``setsockopt`` with + ``SO_REUSEPORT`` raises ``OSError`` (simulating a platform + that advertises the constant but rejects the option). + """ + self.calls: list[tuple] = [] + self._raise_on_reuseport = raise_on_reuseport + + def setsockopt(self, *args): + """Record a ``setsockopt`` call, optionally raising for SO_REUSEPORT.""" + if ( + self._raise_on_reuseport + and hasattr(real_socket, "SO_REUSEPORT") + and args[1] == real_socket.SO_REUSEPORT + ): + self.calls.append(("setsockopt", args)) + raise OSError("SO_REUSEPORT not supported") + self.calls.append(("setsockopt", args)) + + def bind(self, addr): + """Record a ``bind`` call.""" + self.calls.append(("bind", addr)) + + def settimeout(self, timeout): + """Record a ``settimeout`` call.""" + self.calls.append(("settimeout", timeout)) + + +class TestOpenMulticastSocket: + """Tests for :func:`open_multicast_socket`.""" + + def test_sets_options_binds_joins_and_times_out(self, monkeypatch): + """Happy path: reuseaddr, bind, IP_ADD_MEMBERSHIP, and a 1s timeout.""" + fake = FakeSock() + monkeypatch.setattr(udp_socket.socket, "socket", lambda *a, **k: fake) + + sock = udp_socket.open_multicast_socket("224.0.0.69", 4403) + + assert sock is fake + assert ("bind", ("", 4403)) in fake.calls + assert ( + "setsockopt", + (real_socket.SOL_SOCKET, real_socket.SO_REUSEADDR, 1), + ) in fake.calls + expected_mreq = real_socket.inet_aton("224.0.0.69") + real_socket.inet_aton( + "0.0.0.0" + ) + assert ( + "setsockopt", + (real_socket.IPPROTO_IP, real_socket.IP_ADD_MEMBERSHIP, expected_mreq), + ) in fake.calls + assert ("settimeout", 1.0) in fake.calls + + def test_reuseport_set_when_available(self, monkeypatch): + """When ``SO_REUSEPORT`` exists on the platform, it is set to 1.""" + monkeypatch.setattr(real_socket, "SO_REUSEPORT", 15, raising=False) + fake = FakeSock() + monkeypatch.setattr(udp_socket.socket, "socket", lambda *a, **k: fake) + + udp_socket.open_multicast_socket("224.0.0.69", 4403) + + assert ( + "setsockopt", + (real_socket.SOL_SOCKET, real_socket.SO_REUSEPORT, 1), + ) in fake.calls + + def test_reuseport_absent_is_skipped(self, monkeypatch): + """When the platform has no ``SO_REUSEPORT``, no such call is made.""" + monkeypatch.delattr(real_socket, "SO_REUSEPORT", raising=False) + fake = FakeSock() + monkeypatch.setattr(udp_socket.socket, "socket", lambda *a, **k: fake) + + udp_socket.open_multicast_socket("224.0.0.69", 4403) + + assert not hasattr(udp_socket.socket, "SO_REUSEPORT") + assert all( + call[0] != "setsockopt" or len(call[1]) < 2 or call[1][1] != 15 + for call in fake.calls + ) + # No SO_REUSEPORT option key can appear at all when the attribute + # is absent, since the code cannot reference it. + reuseport_calls = [ + call + for call in fake.calls + if call[0] == "setsockopt" and call[1][0] == real_socket.SOL_SOCKET + ] + assert len(reuseport_calls) == 1 # only SO_REUSEADDR + + def test_reuseport_oserror_is_tolerated(self, monkeypatch): + """An ``OSError`` while setting ``SO_REUSEPORT`` does not propagate.""" + monkeypatch.setattr(real_socket, "SO_REUSEPORT", 15, raising=False) + fake = FakeSock(raise_on_reuseport=True) + monkeypatch.setattr(udp_socket.socket, "socket", lambda *a, **k: fake) + + sock = udp_socket.open_multicast_socket("224.0.0.69", 4403) + + assert sock is fake + assert ( + "setsockopt", + (real_socket.SOL_SOCKET, real_socket.SO_REUSEPORT, 1), + ) in fake.calls + # Despite the OSError, bind/join/timeout still happened. + assert ("bind", ("", 4403)) in fake.calls + assert ("settimeout", 1.0) in fake.calls + + def test_join_uses_requested_group_and_port(self, monkeypatch): + """A different group/port is threaded through to bind and the mreq.""" + fake = FakeSock() + monkeypatch.setattr(udp_socket.socket, "socket", lambda *a, **k: fake) + + udp_socket.open_multicast_socket("239.1.2.3", 5000) + + assert ("bind", ("", 5000)) in fake.calls + expected_mreq = real_socket.inet_aton("239.1.2.3") + real_socket.inet_aton( + "0.0.0.0" + ) + assert ( + "setsockopt", + (real_socket.IPPROTO_IP, real_socket.IP_ADD_MEMBERSHIP, expected_mreq), + ) in fake.calls diff --git a/tests/test_meshtastic_udp_unit.py b/tests/test_meshtastic_udp_unit.py new file mode 100644 index 0000000..fac2842 --- /dev/null +++ b/tests/test_meshtastic_udp_unit.py @@ -0,0 +1,732 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for :mod:`data.mesh_ingestor.protocols.meshtastic_udp`. + +Coverage strategy mirrors ``tests/test_meshtastic_udp_decode_unit.py``: + +1. **Real-fixture tests** replay genuine captured datagrams (see + ``tests/fixtures/mesh_udp``) through :meth:`MeshtasticUdpProvider._handle_datagram` + to prove the primary/private split works end-to-end against real traffic. +2. **Synthetic tests** exercise every remaining line/branch (parse failures, + the no-``decoded`` drop path, the receive loop's timeout/OSError/dispatch + branches, and the lifecycle of :class:`_UdpInterface`) with fakes so no + real socket or long-lived thread is ever involved. + +No test opens a real network socket or a real ``socket.timeout``-driven +sleep loop of more than a few milliseconds: every fake socket either raises +immediately or sets the interface's stop flag as a side effect of being +called, so a hung test is not possible. +""" + +from __future__ import annotations + +import base64 +import json +import os +import socket +import sys +import threading +import time +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from meshtastic.protobuf import mesh_pb2, portnums_pb2 + +from data.mesh_ingestor.protocols import meshtastic_udp as udp_mod +from data.mesh_ingestor.protocols import meshtastic_udp_decode as udp_decode +from data.mesh_ingestor.protocols.meshtastic_udp import ( + MeshtasticUdpProvider, + _UdpInterface, +) + + +def _encrypt_packet( + channel_hash: int, + *, + portnum=portnums_pb2.PortNum.TEXT_MESSAGE_APP, + text: bytes = b"hi", + key_b64: str = "AQ==", + packet_id: int = 0x1111, + node_from: int = 0x2222, +) -> bytes: + """Build a raw encrypted ``MeshPacket`` carrying *channel_hash*. + + The application payload is AES-CTR-encrypted with *key_b64* using the same + id/from nonce the firmware uses, so the packet is genuinely decryptable with + that key. The ``channel`` field is set independently to *channel_hash* -- + this lets a test build a packet that *decrypts* with the default key yet + carries a non-primary channel hash (i.e. a default-key SECONDARY channel). + """ + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + data = mesh_pb2.Data(portnum=portnum, payload=text) + key = udp_decode.expand_default_key(key_b64) + nonce = packet_id.to_bytes(8, "little") + node_from.to_bytes(8, "little") + encryptor = Cipher(algorithms.AES(key), modes.CTR(nonce)).encryptor() + ciphertext = encryptor.update(data.SerializeToString()) + encryptor.finalize() + + mp = mesh_pb2.MeshPacket() + mp.id = packet_id + setattr(mp, "from", node_from) + mp.to = 0xFFFFFFFF + mp.channel = channel_hash + mp.encrypted = ciphertext + return mp.SerializeToString() + + +FIXTURE_PATH = os.path.join( + os.path.dirname(__file__), + "fixtures", + "mesh_udp", + "primary_and_private_capture.jsonl", +) +"""Path to the real captured-datagram fixture, resolved relative to this file.""" + + +def _load_fixture_raw() -> tuple[bytes, bytes]: + """Return ``(primary_raw, private_raw)`` from the real-capture fixture. + + Scans the fixture for the first datagram whose ``MeshPacket.channel`` is + 31 (the primary channel's channel hash, per the fixture README) and the + first whose channel is anything else, and returns their raw bytes. + """ + primary_raw = None + private_raw = None + with open(FIXTURE_PATH, "r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + record = json.loads(line) + raw = base64.b64decode(record["raw_b64"]) + mp = mesh_pb2.MeshPacket() + mp.ParseFromString(raw) + if mp.channel == 31 and primary_raw is None: + primary_raw = raw + elif mp.channel != 31 and private_raw is None: + private_raw = raw + if primary_raw is not None and private_raw is not None: + break + assert primary_raw is not None, "fixture must contain a primary-channel datagram" + assert private_raw is not None, "fixture must contain a private-channel datagram" + return primary_raw, private_raw + + +# --------------------------------------------------------------------------- +# Real-fixture integration tests +# --------------------------------------------------------------------------- + + +class TestHandleDatagramRealFixture: + """Replays real captured datagrams through ``_handle_datagram``.""" + + def test_primary_datagram_dispatches_exactly_once(self, monkeypatch): + """A real primary-channel datagram decrypts and reaches on_receive once.""" + primary_raw, _private_raw = _load_fixture_raw() + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(primary_raw, iface) + + assert len(received) == 1 + packet = received[0] + assert packet["channel"] == 0 + portnum = packet["decoded"]["portnum"] + assert isinstance(portnum, str) and portnum + + def test_private_datagram_is_dropped(self, monkeypatch): + """A real private-channel datagram never reaches on_receive.""" + _primary_raw, private_raw = _load_fixture_raw() + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(private_raw, iface) + + assert received == [] + + +# --------------------------------------------------------------------------- +# _handle_datagram synthetic drop paths +# --------------------------------------------------------------------------- + + +class TestHandleDatagramDropPaths: + """Exercises the parse-failure and no-``decoded`` drop branches.""" + + def test_unparseable_bytes_are_dropped(self, monkeypatch): + """Bytes that fail protobuf parsing never reach on_receive.""" + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(b"\xff\xff", iface) + + assert received == [] + + def test_packet_without_encrypted_or_decoded_is_dropped(self, monkeypatch): + """A parsed MeshPacket with neither payload_variant field is dropped.""" + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + mp = mesh_pb2.MeshPacket() + mp.id = 42 + setattr(mp, "from", 7) + raw = mp.SerializeToString() + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(raw, iface) + + assert received == [] + + def test_plaintext_decoded_packet_is_dropped(self, monkeypatch): + """A packet arriving already-``decoded`` (unencrypted) is dropped. + + Even when it carries the correct primary channel hash, a plaintext + packet is rejected: real primary traffic is channel-encrypted, and + accepting plaintext would let a keyless LAN attacker inject spoofed + records. + """ + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.to = 2 + mp.channel = udp_decode.channel_hash("MediumFast", "AQ==") # passes hash gate + mp.decoded.portnum = 3 # POSITION_APP, but plaintext -> must be dropped + raw = mp.SerializeToString() + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(raw, iface) + + assert received == [] + + def test_encrypted_with_wrong_key_is_dropped(self, monkeypatch): + """An encrypted packet that fails to decrypt with the configured key is dropped.""" + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + mp = mesh_pb2.MeshPacket() + mp.id = 1 + setattr(mp, "from", 1) + mp.channel = udp_decode.channel_hash("MediumFast", "AQ==") # passes hash gate + # Garbage ciphertext under the default key never parses to a + # non-empty Data, so decrypt_meshpacket returns None. + mp.encrypted = b"\x00" * 16 + raw = mp.SerializeToString() + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(raw, iface) + + assert received == [] + + def test_unknown_portnum_does_not_crash_the_reader(self, monkeypatch): + """A packet decrypting to an unknown portnum is handled, never raised. + + Regression for the DoS where ``PortNum.Name()`` raised ``ValueError`` + on an out-of-enum portnum and killed the receive thread. It must be + mapped to a sentinel and dispatched (a handler-less portnum is simply + ignored downstream), not crash. + """ + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + primary_hash = udp_decode.channel_hash("MediumFast", "AQ==") + # portnum 99999 is not in the PortNum enum (proto3 open enums accept it). + raw = _encrypt_packet(primary_hash, portnum=99999, text=b"x") + + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + provider._handle_datagram(raw, iface) # must not raise + + assert len(received) == 1 + assert received[0]["decoded"]["portnum"] == "UNKNOWN_APP" + + +# --------------------------------------------------------------------------- +# _recv_loop +# --------------------------------------------------------------------------- + + +class TestRecvLoop: + """Directly exercises ``_recv_loop``'s branches without a real thread.""" + + def test_timeout_then_stop(self): + """A socket.timeout is swallowed (continue) and the loop exits on stop.""" + iface = _UdpInterface() + + calls = {"n": 0} + + class FakeSock: + def recvfrom(self, bufsize): + calls["n"] += 1 + if calls["n"] >= 2: + iface._stop.set() + raise socket.timeout() + + iface._sock = FakeSock() + provider = MeshtasticUdpProvider() + provider._recv_loop(iface) + + assert calls["n"] == 2 + + def test_oserror_breaks_and_clears_connected(self): + """An OSError from recvfrom clears isConnected and exits the loop.""" + iface = _UdpInterface() + iface.isConnected.set() + + class FakeSock: + def recvfrom(self, bufsize): + raise OSError("socket closed") + + iface._sock = FakeSock() + provider = MeshtasticUdpProvider() + provider._recv_loop(iface) + + assert not iface.isConnected.is_set() + + def test_dispatches_datagram_then_stops(self, monkeypatch): + """A successfully received datagram is routed through _handle_datagram.""" + iface = _UdpInterface() + handled: list[bytes] = [] + + mp = mesh_pb2.MeshPacket() + mp.id = 9 + setattr(mp, "from", 9) + mp.decoded.portnum = 1 + raw = mp.SerializeToString() + + calls = {"n": 0} + + class FakeSock: + def recvfrom(self, bufsize): + calls["n"] += 1 + if calls["n"] == 1: + return raw, ("192.0.2.1", 4403) + iface._stop.set() + raise socket.timeout() + + iface._sock = FakeSock() + provider = MeshtasticUdpProvider() + monkeypatch.setattr( + provider, "_handle_datagram", lambda r, i: handled.append(r) + ) + provider._recv_loop(iface) + + assert handled == [raw] + + def test_handle_datagram_exception_is_swallowed_loop_survives(self, monkeypatch): + """An exception from _handle_datagram is caught; the loop keeps running. + + Regression for the DoS where one bad datagram propagated out of + _handle_datagram and killed the reader thread. Here _handle_datagram + raises on the first datagram; the loop must continue to the second and + exit cleanly on the stop flag rather than propagating. + """ + iface = _UdpInterface() + iface.isConnected.set() + calls = {"n": 0} + + class FakeSock: + def recvfrom(self, bufsize): + calls["n"] += 1 + if calls["n"] == 1: + return b"anything", ("192.0.2.1", 4403) + iface._stop.set() + raise socket.timeout() + + def boom(_raw, _iface): + raise ValueError("simulated bad datagram") + + iface._sock = FakeSock() + provider = MeshtasticUdpProvider() + monkeypatch.setattr(provider, "_handle_datagram", boom) + provider._recv_loop(iface) # must not raise + + assert calls["n"] == 2 + # Loop exit clears isConnected so a dead reader is detectable. + assert not iface.isConnected.is_set() + + +# --------------------------------------------------------------------------- +# _UdpInterface lifecycle +# --------------------------------------------------------------------------- + + +class TestUdpInterfaceLifecycle: + """Tests for :class:`_UdpInterface`.""" + + def test_init_defaults(self): + """A fresh interface has no nodes, is not connected, and has no thread/sock.""" + iface = _UdpInterface() + assert iface.nodes == {} + assert isinstance(iface.isConnected, threading.Event) + assert not iface.isConnected.is_set() + assert iface._sock is None + assert iface._thread is None + assert not iface._stop.is_set() + + def test_close_with_no_sock_or_thread_is_safe(self): + """close() must not raise when _sock and _thread were never set.""" + iface = _UdpInterface() + iface.isConnected.set() + iface.close() + assert iface._stop.is_set() + assert not iface.isConnected.is_set() + + def test_close_closes_socket_and_swallows_oserror(self): + """close() swallows an OSError raised by the socket's close().""" + iface = _UdpInterface() + + class RaisingSock: + def close(self): + raise OSError("already closed") + + iface._sock = RaisingSock() + iface.close() # must not raise + assert iface._stop.is_set() + + def test_close_joins_thread(self): + """close() joins the receive thread with a bounded timeout.""" + iface = _UdpInterface() + joined = {"timeout": None} + + class FakeThread: + def join(self, timeout=None): + joined["timeout"] = timeout + + iface._thread = FakeThread() + iface.close() + + assert joined["timeout"] == 2.0 + + def test_close_is_idempotent(self): + """Calling close() twice must not raise.""" + iface = _UdpInterface() + iface.close() + iface.close() + + +# --------------------------------------------------------------------------- +# MeshtasticUdpProvider.connect (full lifecycle through a fake socket) +# --------------------------------------------------------------------------- + + +class TestConnectLifecycle: + """Exercises connect() end-to-end with a fake socket and a real thread.""" + + def test_connect_returns_triple_and_receives_then_closes(self, monkeypatch): + """connect() starts the receive thread; close() stops it cleanly.""" + primary_raw, _private_raw = _load_fixture_raw() + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + monkeypatch.setattr(udp_mod.config, "MESH_UDP_GROUP", "224.0.0.69") + monkeypatch.setattr(udp_mod.config, "MESH_UDP_PORT", 4403) + + received: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: received.append(packet), + ) + + class FakeSock: + def __init__(self): + self._served = False + self.closed = False + + def recvfrom(self, bufsize): + if not self._served: + self._served = True + return primary_raw, ("192.0.2.1", 4403) + # Small sleep keeps the background thread from busy-spinning + # at full CPU while the test asserts and calls close(). + time.sleep(0.005) + raise socket.timeout() + + def close(self): + self.closed = True + + fake_sock = FakeSock() + monkeypatch.setattr( + udp_mod, "open_multicast_socket", lambda group, port: fake_sock + ) + + provider = MeshtasticUdpProvider() + iface, target, next_candidate = provider.connect(active_candidate="ignored") + + assert target == "udp://224.0.0.69:4403" + assert next_candidate == "ignored" + assert iface.isConnected.is_set() + + deadline = time.monotonic() + 2.0 + while not received and time.monotonic() < deadline: + time.sleep(0.01) + assert len(received) == 1 + + iface.close() + + assert not iface._thread.is_alive() + assert not iface.isConnected.is_set() + assert fake_sock.closed + + +# --------------------------------------------------------------------------- +# subscribe / extract_host_node_id / node_snapshot_items +# --------------------------------------------------------------------------- + + +PRIMARY_HASH = udp_decode.channel_hash("MediumFast", "AQ==") # 31, per the fixture +SECONDARY_HASH = udp_decode.channel_hash("Private", "AQ==") # default-key secondary + + +class TestPrimaryChannelHashHelper: + """Tests for :meth:`MeshtasticUdpProvider._primary_channel_hash`.""" + + def test_returns_hash_when_name_set(self, monkeypatch): + """With a configured name, the helper returns the computed channel hash.""" + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + assert MeshtasticUdpProvider()._primary_channel_hash() == 31 + + def test_returns_none_when_name_blank(self, monkeypatch): + """A blank name yields None so primary-only mode can fail closed.""" + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "") + assert MeshtasticUdpProvider()._primary_channel_hash() is None + + +class TestPrimaryChannelFilter: + """The channel-hash gate: only channel-0 (primary) traffic is dispatched.""" + + @pytest.fixture + def received(self, monkeypatch): + """Capture packets that reach on_receive; default env to the RGW1 setup.""" + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_ONLY", True) + captured: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: captured.append(packet), + ) + return captured + + def test_primary_hash_packet_is_dispatched(self, received): + """A packet whose channel hash matches the primary channel is delivered.""" + raw = _encrypt_packet(PRIMARY_HASH) + MeshtasticUdpProvider()._handle_datagram(raw, _UdpInterface()) + assert len(received) == 1 + assert received[0]["channel"] == 0 + + def test_default_key_secondary_channel_is_dropped(self, received): + """A default-key SECONDARY channel that DECRYPTS is still dropped by hash. + + This is the core privacy guarantee: the packet is encrypted with the + very same ``AQ==`` key as the primary channel and would decrypt cleanly, + but its channel hash is not the primary's, so it must never reach the + collector. + """ + raw = _encrypt_packet(SECONDARY_HASH) + # Sanity: prove the packet really does decrypt with the primary key, so + # the drop is attributable to the hash gate and not a decrypt failure. + mp = mesh_pb2.MeshPacket() + mp.ParseFromString(raw) + assert udp_decode.decrypt_meshpacket(mp, "AQ==") is not None + assert mp.channel != PRIMARY_HASH + + MeshtasticUdpProvider()._handle_datagram(raw, _UdpInterface()) + assert received == [] + + def test_blank_name_fails_closed(self, received, monkeypatch): + """primary-only with no configured name drops even a valid primary packet.""" + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "") + raw = _encrypt_packet(PRIMARY_HASH) + MeshtasticUdpProvider()._handle_datagram(raw, _UdpInterface()) + assert received == [] + + def test_filtering_is_unconditional_of_primary_channel_only(self, monkeypatch): + """The hash gate applies even when PRIMARY_CHANNEL_ONLY is False. + + PRIMARY_CHANNEL_ONLY governs only the API/serial transport; the UDP + transport can never represent a non-primary channel (it stamps index 0), + so it filters unconditionally. A secondary-channel packet is dropped and + a primary-channel packet is accepted regardless of the flag. + """ + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_ONLY", False) + captured: list[dict] = [] + monkeypatch.setattr( + udp_mod.handlers, + "on_receive", + lambda packet, interface: captured.append(packet), + ) + + provider = MeshtasticUdpProvider() + provider._handle_datagram(_encrypt_packet(SECONDARY_HASH), _UdpInterface()) + assert captured == [] # secondary dropped despite the flag being off + + provider._handle_datagram(_encrypt_packet(PRIMARY_HASH), _UdpInterface()) + assert len(captured) == 1 # primary still accepted + + +class TestConnectLogsPrimaryFilter: + """connect() emits a startup log describing the resolved primary filter.""" + + def _fake_socket(self, monkeypatch): + """Install a fake multicast socket that only times out (no traffic).""" + + class FakeSock: + def recvfrom(self, bufsize): + time.sleep(0.005) + raise socket.timeout() + + def close(self): + pass + + monkeypatch.setattr( + udp_mod, "open_multicast_socket", lambda group, port: FakeSock() + ) + + def test_logs_resolved_hash_info(self, monkeypatch): + """A configured name logs the resolved hash at info severity.""" + self._fake_socket(monkeypatch) + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "MediumFast") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_KEY", "AQ==") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_ONLY", True) + logs: list[dict] = [] + monkeypatch.setattr( + udp_mod.config, + "_debug_log", + lambda *a, **k: logs.append(k), + ) + provider = MeshtasticUdpProvider() + iface, _target, _c = provider.connect(active_candidate=None) + iface.close() + + assert any(k.get("primary_channel_hash") == 31 for k in logs) + entry = next(k for k in logs if "primary_channel_hash" in k) + assert entry["severity"] == "info" + + def test_logs_warn_when_fail_closed(self, monkeypatch): + """primary-only with no name logs at warn severity (fail-closed).""" + self._fake_socket(monkeypatch) + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_NAME", "") + monkeypatch.setattr(udp_mod.config, "PRIMARY_CHANNEL_ONLY", True) + logs: list[dict] = [] + monkeypatch.setattr( + udp_mod.config, + "_debug_log", + lambda *a, **k: logs.append(k), + ) + provider = MeshtasticUdpProvider() + iface, _target, _c = provider.connect(active_candidate=None) + iface.close() + + entry = next(k for k in logs if "primary_channel_hash" in k) + assert entry["primary_channel_hash"] is None + assert entry["severity"] == "warn" + + +class TestProviderMisc: + """Tests for the remaining small provider methods.""" + + def test_subscribe_returns_empty_list_and_is_idempotent(self): + """subscribe() always returns [] and calling it twice is harmless.""" + provider = MeshtasticUdpProvider() + first = provider.subscribe() + second = provider.subscribe() + assert first == [] + assert second == [] + + def test_extract_host_node_id_returns_config_value(self, monkeypatch): + """extract_host_node_id surfaces config.INGESTOR_NODE_ID verbatim.""" + monkeypatch.setattr(udp_mod.config, "INGESTOR_NODE_ID", "!deadbeef") + provider = MeshtasticUdpProvider() + assert provider.extract_host_node_id(object()) == "!deadbeef" + + def test_extract_host_node_id_none_by_default(self, monkeypatch): + """extract_host_node_id returns None when unset.""" + monkeypatch.setattr(udp_mod.config, "INGESTOR_NODE_ID", None) + provider = MeshtasticUdpProvider() + assert provider.extract_host_node_id(object()) is None + + def test_node_snapshot_items_empty(self): + """node_snapshot_items returns [] for a fresh interface.""" + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + assert provider.node_snapshot_items(iface) == [] + + def test_node_snapshot_items_populated(self): + """node_snapshot_items reflects a non-empty nodes mapping.""" + provider = MeshtasticUdpProvider() + iface = _UdpInterface() + iface.nodes["!aabbccdd"] = {"num": 1} + items = provider.node_snapshot_items(iface) + assert items == [("!aabbccdd", {"num": 1})] diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index f1342ba..16a2c16 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -52,6 +52,9 @@ from data.mesh_ingestor.mesh_protocol import MeshProtocol # noqa: E402 - path s from data.mesh_ingestor.protocols.meshtastic import ( # noqa: E402 - path setup MeshtasticProvider, ) +from data.mesh_ingestor.protocols.meshtastic_udp import ( # noqa: E402 - path setup + MeshtasticUdpProvider, +) from data.mesh_ingestor.connection import parse_tcp_target # noqa: E402 - path setup from data.mesh_ingestor.protocols.meshcore import ( # noqa: E402 - path setup EventType, @@ -87,6 +90,11 @@ def test_meshtastic_provider_satisfies_protocol(): assert isinstance(MeshtasticProvider(), MeshProtocol) +def test_meshtastic_udp_provider_satisfies_protocol(): + """MeshtasticUdpProvider must structurally satisfy the Provider Protocol.""" + assert isinstance(MeshtasticUdpProvider(), MeshProtocol) + + def test_daemon_main_uses_provider_connect(monkeypatch): calls = {"connect": 0}