feat(ingestor): passive UDP transport (Mesh via UDP), primary-channel only (#838)

Adds an optional passive UDP transport (TRANSPORT=udp) that reads the node's
"Mesh via UDP" LAN multicast instead of holding the radio's single API/serial
slot, filters to the PRIMARY channel by channel-hash (fail-closed), decrypts
with the default PSK, and enriches payloads to match the API-path packet shape
so the collector receives identical records. Implemented as a MeshProtocol
provider; 100% unit coverage on new modules; real-capture fixtures included
(node IDs/GPS anonymized). See PR description for details.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
TJ Paulson
2026-07-05 15:24:26 -05:00
committed by GitHub
parent 436bf34de6
commit 1b934ee6f7
24 changed files with 3189 additions and 2 deletions
+33
View File
@@ -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
+40
View File
@@ -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:
+12
View File
@@ -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="" \
+14
View File
@@ -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",
]
+45
View File
@@ -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",
+4
View File
@@ -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
+10
View File
@@ -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)
+13 -2
View File
@@ -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",
]
@@ -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"]
@@ -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
@@ -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
+1
View File
@@ -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
+77
View File
@@ -0,0 +1,77 @@
<!-- Copyright © 2025-26 l5yth & contributors -->
<!-- Licensed under the Apache License, Version 2.0 (see LICENSE) -->
# `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=<n>, 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).
+254
View File
@@ -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 <your base64 psk>."
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+60
View File
@@ -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
+14
View File
@@ -0,0 +1,14 @@
<!-- Copyright © 2025-26 l5yth & contributors -->
<!-- Licensed under the Apache License, Version 2.0 (see LICENSE) -->
# 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.
@@ -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"}
+37
View File
@@ -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
+248
View File
@@ -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__
+145
View File
@@ -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
+575
View File
@@ -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"
+159
View File
@@ -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
+732
View File
@@ -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})]
+8
View File
@@ -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}