From 44fab19bf33c6c00da5d87c3aed060634f1f30cb Mon Sep 17 00:00:00 2001 From: Afri <220195275+l5yth@users.noreply.github.com> Date: Sun, 4 Oct 2026 15:51:58 +0200 Subject: [PATCH] data: listen on both Meshtastic UDP multicast groups by default (#903) (#909) --- .env.example | 5 +- CHANGELOG.md | 1 + README.md | 9 +- data/Dockerfile | 4 +- data/mesh_ingestor/config.py | 71 ++- .../mesh_ingestor/protocols/meshtastic_udp.py | 128 +++-- data/tools/README.md | 9 +- data/tools/compose.udp.pi.yml | 6 +- flake.nix | 4 +- tests/test_config_udp_groups_unit.py | 203 ++++++++ tests/test_config_unit.py | 4 +- tests/test_meshtastic_udp_unit.py | 481 ++++++++++++------ 12 files changed, 717 insertions(+), 208 deletions(-) create mode 100644 tests/test_config_udp_groups_unit.py diff --git a/.env.example b/.env.example index 68c13e4a..66d8c2cf 100644 --- a/.env.example +++ b/.env.example @@ -226,8 +226,9 @@ POTATOMESH_IMAGE_TAG="latest" # its own id from your primary identity; set it there only to override that. # INGESTOR_NODE_ID=!xxxxxxxx -# Multicast group/port for "Mesh via UDP" (defaults shown). -# MESH_UDP_GROUP=224.0.0.69 +# Multicast groups/port for "Mesh via UDP" (defaults shown). The default listens +# on both groups; set one address to listen on that group only. +# MESH_UDP_GROUP=239.0.0.69,224.0.0.69 # MESH_UDP_PORT=4403 # Frequency and preset shown for Reticulum nodes. Both are read from the first # RNodeInterface in your RNS config; set these only to override that. diff --git a/CHANGELOG.md b/CHANGELOG.md index 25a8f0b7..09933e74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ * Web: the federation table renders a Reticulum zero count as a dash instead of a tile (SPEC RD8) ### Fixes +* Data: `TRANSPORT=udp` listens on both Meshtastic multicast groups by default, `239.0.0.69` (firmware 2.8+) and `224.0.0.69`; `MESH_UDP_GROUP` takes a comma-separated list. If you set `MESH_UDP_GROUP=224.0.0.69` before, remove it to also hear firmware 2.8+ nodes (#903) * Data: `MESHCORE_TELEMETRY_POLL_24H_EXEMPT` reaches every packaged deployment - `docker-compose.yml`, both `data/Dockerfile` stages and the NixOS module (`meshcoreTelemetryPoll24hExempt`); setting it in `.env` had no effect under Compose or NixOS (SPEC PX5) * Docs: the README documents `MESHTASTIC_PSK_B64` (SPEC PX7) * CI: the env-surface check runs in pytest whenever a deployment surface changes, and also checks names with digits, names read through helpers, and retired names (SPEC PX6/PX8) diff --git a/README.md b/README.md index fb049fc8..2e8e8d6f 100644 --- a/README.md +++ b/README.md @@ -288,7 +288,7 @@ Configure with the environment variables below. | `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`); find it with `meshtastic --info` if blank on the radio. Required by UDP `PRIMARY_CHANNEL_ONLY=1`. | -| `MESH_UDP_GROUP` | `224.0.0.69` | Multicast group joined in UDP transport. | +| `MESH_UDP_GROUP` | `239.0.0.69,224.0.0.69` | Comma-separated multicast groups joined in UDP transport. The default listens on both; set one address to listen on that group only. | | `MESH_UDP_PORT` | `4403` | Multicast port joined in UDP transport. | | `INGESTOR_NODE_ID` | _unset_ | `!xxxxxxxx` id used for the ingestor heartbeat. Required for the UDP transport, which cannot auto-detect "self". Optional for `PROTOCOL=reticulum`, which derives one from your primary announced identity; set it there only to override that. | | `RETICULUM_CONFIG_DIR` | `~/.reticulum` | Which RNS config the ingestor uses, and so which interfaces it can see. Point it at the directory your `rnsd` uses. See [Reticulum](#reticulum). | @@ -427,12 +427,13 @@ Reticulum nodes stay off the map. ### Passive UDP transport Meshtastic's node API accepts only one client at a time. Set `TRANSPORT=udp` -to run the ingestor as a passive listener on the node's LAN multicast group -(`224.0.0.69:4403`) instead of connecting to the API - leaving the API slot -free for the phone app or CLI. +to run the ingestor as a passive listener on the node's LAN multicast groups +(port `4403`) instead of connecting to the API - leaving the API slot free for +the phone app or CLI. Enable "Mesh via UDP" on the node first: `meshtastic --set network.enabled_protocols 1`. +- Listens on `239.0.0.69` (firmware 2.8 and later) and `224.0.0.69` (earlier firmware). Set `MESH_UDP_GROUP` to one of them to listen on that group only. - Decrypts the primary channel with `PRIMARY_CHANNEL_KEY` (default `AQ==`). Channels with other keys are dropped undecrypted. - Set `PRIMARY_CHANNEL_ONLY=1` and `PRIMARY_CHANNEL_NAME` to ingest only channel 0. Without `PRIMARY_CHANNEL_NAME` set, `PRIMARY_CHANNEL_ONLY=1` drops every packet (fail closed). - The node list rebuilds from observed packets - the node's own database is not read. Decoded payloads (position, telemetry, traceroute, …) match the API/serial transport shape. diff --git a/data/Dockerfile b/data/Dockerfile index daf889d4..9277031c 100644 --- a/data/Dockerfile +++ b/data/Dockerfile @@ -83,7 +83,7 @@ ENV CONNECTION=/dev/ttyACM0 \ PRIMARY_CHANNEL_ONLY=0 \ PRIMARY_CHANNEL_KEY=AQ== \ PRIMARY_CHANNEL_NAME="" \ - MESH_UDP_GROUP=224.0.0.69 \ + MESH_UDP_GROUP=239.0.0.69,224.0.0.69 \ MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ @@ -132,7 +132,7 @@ ENV CONNECTION=/dev/ttyACM0 \ PRIMARY_CHANNEL_ONLY=0 \ PRIMARY_CHANNEL_KEY=AQ== \ PRIMARY_CHANNEL_NAME="" \ - MESH_UDP_GROUP=224.0.0.69 \ + MESH_UDP_GROUP=239.0.0.69,224.0.0.69 \ MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index 4879a68f..357765b7 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -17,6 +17,7 @@ from __future__ import annotations import base64 +import ipaddress import math import os from datetime import datetime, timezone @@ -335,8 +336,73 @@ 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.""" +DEFAULT_MESH_UDP_GROUPS = ("239.0.0.69", "224.0.0.69") +"""Multicast groups joined when :envvar:`MESH_UDP_GROUP` is unset or blank. + +Meshtastic firmware 2.8 and later sends "Mesh via UDP" to ``239.0.0.69``; +earlier firmware sends to ``224.0.0.69``. Neither falls back to the other, so +the default joins both (SPEC UG1).""" + + +def _parse_mesh_udp_groups(raw: str) -> tuple[str, ...]: + """Parse :envvar:`MESH_UDP_GROUP` into the multicast groups to join. + + The value is a comma-separated list of IPv4 multicast addresses. Each + fragment is trimmed (a quote-only fragment counts as blank, see + :func:`_clean_env_fragment`), blank fragments are skipped, and a repeated + group collapses onto its first position. + + Validated at import time for the reason :data:`PRIMARY_CHANNEL_KEY` is: an + entry that is not a multicast group otherwise fails only inside + ``connect()``, where the daemon's generic reconnect handler logs it as a + connection failure and retries forever. + + Parameters: + raw: Raw :envvar:`MESH_UDP_GROUP` value. + + Returns: + The groups in configured order, or :data:`DEFAULT_MESH_UDP_GROUPS` + when no fragment carries an address. + + Raises: + ValueError: When a fragment is not an IPv4 address inside + ``224.0.0.0/4``. The message names the variable and the entry. + """ + groups: list[str] = [] + for part in raw.split(","): + entry = _clean_env_fragment(part) + if not entry: + continue + try: + is_multicast = ipaddress.IPv4Address(entry).is_multicast + except ValueError: # not an IPv4 address at all + is_multicast = False + if not is_multicast: + raise ValueError( + f"MESH_UDP_GROUP entry {entry!r} is not an IPv4 multicast address " + "(224.0.0.0/4). Set one address or a comma-separated list, " + 'e.g. "239.0.0.69,224.0.0.69".' + ) + # IPv4Address accepts only canonical dotted-decimal (no leading + # zeros), so the raw entry is already the normalised spelling. + if entry not in groups: + groups.append(entry) + return tuple(groups) or DEFAULT_MESH_UDP_GROUPS + + +MESH_UDP_GROUPS = _parse_mesh_udp_groups(os.environ.get("MESH_UDP_GROUP", "")) +"""IPv4 multicast groups joined in UDP transport mode, in configured order. + +Parsed from the comma-separated :envvar:`MESH_UDP_GROUP` by +:func:`_parse_mesh_udp_groups`; the UDP provider opens one group-bound socket +per entry. Defaults to :data:`DEFAULT_MESH_UDP_GROUPS`; a single address +restricts the listener to that group.""" + +MESH_UDP_GROUP = ",".join(MESH_UDP_GROUPS) +"""Normalised comma-joined form of :data:`MESH_UDP_GROUPS`. + +Kept for callers that read the single-string setting; the UDP provider reads +:data:`MESH_UDP_GROUPS`.""" MESH_UDP_PORT = int(os.environ.get("MESH_UDP_PORT", "4403").strip() or "4403") """UDP port for the Mesh-via-UDP multicast group. @@ -691,6 +757,7 @@ __all__ = [ "PRIMARY_CHANNEL_KEY", "PRIMARY_CHANNEL_NAME", "MESH_UDP_GROUP", + "MESH_UDP_GROUPS", "MESH_UDP_PORT", "INGESTOR_NODE_ID", "RETICULUM_CONFIG_DIR", diff --git a/data/mesh_ingestor/protocols/meshtastic_udp.py b/data/mesh_ingestor/protocols/meshtastic_udp.py index 56d93ef9..991b881d 100644 --- a/data/mesh_ingestor/protocols/meshtastic_udp.py +++ b/data/mesh_ingestor/protocols/meshtastic_udp.py @@ -22,8 +22,8 @@ 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 +background thread reads datagrams off one multicast socket per joined group +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 @@ -47,6 +47,7 @@ transport's packet shape. from __future__ import annotations +import select import socket import threading @@ -60,6 +61,13 @@ from .meshtastic_udp_decode import ( ) from .meshtastic_udp_socket import open_multicast_socket +_RECV_POLL_SECS = 1.0 +"""Seconds the receive loop waits on its sockets before re-checking the stop flag. + +Matches the 1-second socket timeout set by +:func:`~data.mesh_ingestor.protocols.meshtastic_udp_socket.open_multicast_socket`, +which bounded the same re-check while the loop read a single socket.""" + class _UdpInterface: """Minimal interface object standing in for a Meshtastic library interface. @@ -76,23 +84,24 @@ class _UdpInterface: """Initialise an unconnected interface with no known nodes.""" self.nodes: dict = {} self.isConnected = threading.Event() - self._sock: socket.socket | None = None + self._socks: list[socket.socket] = [] self._thread: threading.Thread | None = None self._stop = threading.Event() def close(self) -> None: - """Stop the receive thread and release the socket. + """Stop the receive thread and release every group 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. + Signals :attr:`_stop` first so the receive loop's next wait timeout + (or the socket closes below, whichever comes first) causes it to + exit, then closes each socket (best-effort -- close errors are not + actionable here, and one failing close must not leave the others + open) and joins the thread with a bounded timeout so shutdown can + never hang indefinitely. """ self._stop.set() - if self._sock is not None: + for sock in self._socks: try: - self._sock.close() + sock.close() except OSError: pass if self._thread is not None: @@ -148,7 +157,14 @@ class MeshtasticUdpProvider: def connect( self, *, active_candidate: str | None ) -> tuple[object, str | None, str | None]: - """Open the multicast socket and start the background receive thread. + """Join every configured multicast group and start the receive thread. + + Opens one socket per entry of :data:`config.MESH_UDP_GROUPS`, all on + :data:`config.MESH_UDP_PORT`: each socket is bound to its group's + address, so it receives that group's datagrams only. Joining is + all-or-nothing -- if any group fails, the sockets already opened are + closed and the error propagates, so the daemon retries the whole + connect just as it does when a single group fails. Parameters: active_candidate: Ignored (there is no serial/BLE candidate @@ -159,11 +175,22 @@ class MeshtasticUdpProvider: 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. + live :class:`_UdpInterface`, a ``udp://group[,group...]:port`` + string naming the joined groups, and *active_candidate* unchanged. + + Raises: + OSError: When a group cannot be joined (for example, no + multicast-capable route on the host). """ iface = _UdpInterface() - iface._sock = open_multicast_socket(config.MESH_UDP_GROUP, config.MESH_UDP_PORT) + try: + for group in config.MESH_UDP_GROUPS: + iface._socks.append(open_multicast_socket(group, config.MESH_UDP_PORT)) + except Exception: + # Release the groups already joined before failing, so each + # reconnect attempt the daemon makes does not leak a socket. + iface.close() + raise # 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 @@ -188,46 +215,61 @@ class MeshtasticUdpProvider: target=self._recv_loop, args=(iface,), daemon=True ) iface._thread.start() - target = f"udp://{config.MESH_UDP_GROUP}:{config.MESH_UDP_PORT}" + target = f"udp://{','.join(config.MESH_UDP_GROUPS)}:{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. + """Wait on all of *iface*'s group sockets and handle datagrams until stopped. - Runs on the background thread started by :meth:`connect`. A - ``socket.timeout`` (the socket has a 1-second timeout, see + Runs on the background thread started by :meth:`connect`. One + ``select`` waits on every joined group's socket at once, so this one + thread serves all groups; its :data:`_RECV_POLL_SECS` timeout is what + re-checks the stop flag. A socket reported readable whose ``recvfrom`` + still times out (each keeps the 1-second timeout set by :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. + is skipped. Any other ``OSError`` -- or the ``ValueError`` ``select`` + raises for a socket that is already closed -- means + :meth:`_UdpInterface.close` closed a socket out from under this + thread, and 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 + iface: The interface whose sockets 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", - ) + ready, _w, _x = select.select(iface._socks, [], [], _RECV_POLL_SECS) + except (OSError, ValueError): + # A closed socket's fileno() is -1, which select rejects + # with ValueError rather than OSError. + return + for sock in ready: + try: + raw, _addr = sock.recvfrom(65535) + except socket.timeout: + # Readable yet empty by the time recvfrom ran (Linux + # can discard a datagram that fails its checksum after + # waking select); the other ready sockets still count. + continue + except OSError: + # Closed under the read; ``finally`` still runs. + return + 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 diff --git a/data/tools/README.md b/data/tools/README.md index deab1082..2f71fcf5 100644 --- a/data/tools/README.md +++ b/data/tools/README.md @@ -18,7 +18,8 @@ produce the real-traffic fixtures under # 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 +python data/tools/capture_udp_fixtures.py --out capture.jsonl --max 40 +# Node on firmware 2.8 or later: add --group 239.0.0.69 # Optional live decode summary of primary-channel packets: python data/tools/capture_udp_fixtures.py --out capture.jsonl --primary-only ``` @@ -34,8 +35,8 @@ gate. The runtime decode/crypto it exercises *is* fully covered by ## `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. +requires `network_mode: host` (multicast cannot reach a bridged container) and +reads the same `.env` as the standard deployment. ### `.env` keys the UDP deployment reads @@ -48,7 +49,7 @@ PRIMARY_CHANNEL_NAME=MediumFast # REQUIRED: name of channel 0 (or the preset # 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_GROUP=239.0.0.69,224.0.0.69 # both groups; set one address to restrict MESH_UDP_PORT=4403 # plus the standard API_TOKEN / INSTANCE_DOMAIN ``` diff --git a/data/tools/compose.udp.pi.yml b/data/tools/compose.udp.pi.yml index f978199a..a3313b72 100644 --- a/data/tools/compose.udp.pi.yml +++ b/data/tools/compose.udp.pi.yml @@ -15,8 +15,8 @@ # 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. +# `network_mode: host` is REQUIRED - multicast cannot reach a bridged +# container. services: ingestor: @@ -31,7 +31,7 @@ services: # 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_GROUP: ${MESH_UDP_GROUP:-239.0.0.69,224.0.0.69} MESH_UDP_PORT: ${MESH_UDP_PORT:-4403} INGESTOR_NODE_ID: ${INGESTOR_NODE_ID:-} API_TOKEN: ${API_TOKEN} diff --git a/flake.nix b/flake.nix index 643b151a..15d8a3ff 100644 --- a/flake.nix +++ b/flake.nix @@ -419,8 +419,8 @@ meshUdpGroup = lib.mkOption { type = lib.types.str; - default = "224.0.0.69"; - description = "Multicast group for Meshtastic \"Mesh via UDP\""; + default = "239.0.0.69,224.0.0.69"; + description = "Comma-separated multicast groups for Meshtastic \"Mesh via UDP\". The default listens on both; set one address to listen on that group only"; }; meshUdpPort = lib.mkOption { diff --git a/tests/test_config_udp_groups_unit.py b/tests/test_config_udp_groups_unit.py new file mode 100644 index 00000000..20c08d29 --- /dev/null +++ b/tests/test_config_udp_groups_unit.py @@ -0,0 +1,203 @@ +# 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 the ``MESH_UDP_GROUP`` multicast-group list (issue #903). + +Meshtastic firmware 2.8 sends "Mesh via UDP" to ``239.0.0.69`` and earlier +firmware to ``224.0.0.69``, with no fallback either way, so the UDP transport +joins both by default and :envvar:`MESH_UDP_GROUP` accepts a comma-separated +list, validated at import time like ``PRIMARY_CHANNEL_KEY`` (ACCEPTANCE UH-A1). + +The packaged deployment defaults are pinned here as well: the image ``ENV`` +and the Nix option override the code default, so a stale single-group default +there would keep every such deployment deaf to one firmware line. +""" + +from __future__ import annotations + +import importlib +import re +import sys +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)) + +import data.mesh_ingestor.config as config + +#: The shipped default, spelled out rather than read back from ``config`` so a +#: change to the default has to be made here deliberately. +EXPECTED_GROUPS = ("239.0.0.69", "224.0.0.69") + +#: :data:`EXPECTED_GROUPS` as an operator writes it in ``MESH_UDP_GROUP``. +EXPECTED_SETTING = ",".join(EXPECTED_GROUPS) + + +@pytest.fixture +def load_config(monkeypatch): + """Reload ``config`` under a given ``MESH_UDP_GROUP``; restore defaults after. + + Yields a callable taking the raw value (``None`` unsets the variable) that + reloads the module and returns it, so import-time parsing and validation + run exactly as they do at ingestor startup. + """ + + def _load(value: str | None): + """Reload ``config`` with *value* as ``MESH_UDP_GROUP`` (``None`` unsets it).""" + if value is None: + monkeypatch.delenv("MESH_UDP_GROUP", raising=False) + else: + monkeypatch.setenv("MESH_UDP_GROUP", value) + return importlib.reload(config) + + yield _load + monkeypatch.delenv("MESH_UDP_GROUP", raising=False) + importlib.reload(config) + + +class TestParseMeshUdpGroups: + """Tests for :func:`config._parse_mesh_udp_groups`.""" + + def test_single_group(self): + """One address yields a one-group tuple, restricting the listener.""" + assert config._parse_mesh_udp_groups("224.0.0.69") == ("224.0.0.69",) + + def test_list_keeps_configured_order(self): + """A comma-separated list is split in the order it was written.""" + assert config._parse_mesh_udp_groups("224.0.0.69,239.0.0.69") == ( + "224.0.0.69", + "239.0.0.69", + ) + + def test_whitespace_and_empty_segments_are_ignored(self): + """Padding is trimmed and empty segments are skipped, not rejected.""" + assert ( + config._parse_mesh_udp_groups(" 239.0.0.69 ,, 224.0.0.69 ,") + == EXPECTED_GROUPS + ) + + def test_duplicates_collapse_onto_first_position(self): + """A repeated group is joined once, where it first appeared.""" + assert config._parse_mesh_udp_groups("224.0.0.69,239.0.0.69, 224.0.0.69") == ( + "224.0.0.69", + "239.0.0.69", + ) + + @pytest.mark.parametrize("raw", ["", " ", ",", " , ", '""', "''"]) + def test_blank_falls_back_to_both_groups(self, raw): + """Blank, separator-only, or quote-only values mean the default. + + A quote-only value is what a Compose ``${VAR:-""}`` default delivers + (see ``test_config_unit.TestComposeDefaults``); it carries no address. + """ + assert config._parse_mesh_udp_groups(raw) == EXPECTED_GROUPS + + @pytest.mark.parametrize("group", ["224.0.0.0", "239.255.255.255"]) + def test_multicast_block_boundaries_are_accepted(self, group): + """Both ends of ``224.0.0.0/4`` are valid groups.""" + assert config._parse_mesh_udp_groups(group) == (group,) + + @pytest.mark.parametrize( + "entry", + [ + "not-an-address", + "10.0.0.1", # unicast + "223.255.255.255", # just below the multicast block + "240.0.0.1", # just above it (reserved) + "255.255.255.255", # limited broadcast + "239.0.0.256", # octet out of range + "239.0.0.69:4403", # port belongs in MESH_UDP_PORT + "ff02::1", # IPv6 multicast; the transport is IPv4-only + '"239.0.0.69"', # quoted address: quotes are content, not stripped + ], + ) + def test_non_multicast_entry_raises_naming_variable_and_entry(self, entry): + """Anything outside IPv4 ``224.0.0.0/4`` is rejected with a clear message.""" + with pytest.raises(ValueError) as excinfo: + config._parse_mesh_udp_groups(f"239.0.0.69,{entry}") + message = str(excinfo.value) + assert "MESH_UDP_GROUP" in message + assert repr(entry) in message + + +class TestMeshUdpGroupsAtImport: + """Import-time resolution of :data:`config.MESH_UDP_GROUPS`.""" + + def test_default_listens_on_both_groups(self, load_config): + """Unset, the ingestor joins the firmware 2.8+ group and the earlier one.""" + assert load_config(None).MESH_UDP_GROUPS == EXPECTED_GROUPS + + def test_list_is_normalised_into_both_names(self, load_config): + """The tuple and the comma-joined string agree after normalisation.""" + cfg = load_config(" 224.0.0.69 ,239.0.0.69,224.0.0.69") + assert cfg.MESH_UDP_GROUPS == ("224.0.0.69", "239.0.0.69") + assert cfg.MESH_UDP_GROUP == "224.0.0.69,239.0.0.69" + + def test_one_address_restricts_to_that_group(self, load_config): + """Setting a single address listens on that group only.""" + cfg = load_config("239.0.0.69") + assert cfg.MESH_UDP_GROUPS == ("239.0.0.69",) + assert cfg.MESH_UDP_GROUP == "239.0.0.69" + + def test_invalid_entry_fails_at_import(self, load_config): + """A bad entry stops startup instead of failing every reconnect. + + Without import-time validation the entry first fails inside + ``connect()``, which the daemon's generic handler retries forever + (the failure mode ACCEPTANCE UH-A1 closed for ``PRIMARY_CHANNEL_KEY``). + """ + with pytest.raises(ValueError, match="MESH_UDP_GROUP entry '10.0.0.1'"): + load_config("239.0.0.69,10.0.0.1") + + def test_both_names_are_exported(self): + """The parsed tuple joins the string setting on the public surface.""" + assert {"MESH_UDP_GROUP", "MESH_UDP_GROUPS"} <= set(config.__all__) + + +class TestMeshUdpGroupDeploymentSurface: + """Every packaged default delivers both groups, or defers to the code. + + A packaged default overrides :data:`config.DEFAULT_MESH_UDP_GROUPS`, so a + surface still pinning ``224.0.0.69`` alone would silently keep that whole + deployment path on one group. + """ + + def test_image_stages_default_to_both_groups(self): + """Both ``data/Dockerfile`` stages declare the full default list.""" + text = (REPO_ROOT / "data" / "Dockerfile").read_text(encoding="utf-8") + assert re.findall(r"^\s*MESH_UDP_GROUP=(\S+)", text, re.MULTILINE) == [ + EXPECTED_SETTING, + EXPECTED_SETTING, + ] + + def test_nix_option_defaults_to_both_groups(self): + """The NixOS ``meshUdpGroup`` option defaults to the full list.""" + text = (REPO_ROOT / "flake.nix").read_text(encoding="utf-8") + match = re.search( + r'meshUdpGroup = lib\.mkOption \{.*?default = "([^"]*)";', text, re.S + ) + assert match, "flake.nix no longer declares the meshUdpGroup option" + assert match.group(1) == EXPECTED_SETTING + + @pytest.mark.parametrize( + "path", ["docker-compose.yml", "data/tools/compose.udp.pi.yml"] + ) + def test_compose_default_is_bare_or_both_groups(self, path): + """A Compose default is either bare (code default) or the full list.""" + text = (REPO_ROOT / path).read_text(encoding="utf-8") + match = re.search(r"MESH_UDP_GROUP:\s*\$\{MESH_UDP_GROUP:-([^}]*)\}", text) + assert match, f"{path} no longer passes MESH_UDP_GROUP through" + assert match.group(1) in ("", EXPECTED_SETTING) diff --git a/tests/test_config_unit.py b/tests/test_config_unit.py index db4b6bd2..5af7bf19 100644 --- a/tests/test_config_unit.py +++ b/tests/test_config_unit.py @@ -712,7 +712,7 @@ class TestUdpTransportDefaults: 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_GROUP == "239.0.0.69,224.0.0.69" assert config.MESH_UDP_PORT == 4403 assert config.INGESTOR_NODE_ID is None @@ -746,7 +746,7 @@ class TestUdpTransportDefaults: monkeypatch.setenv("MESH_UDP_GROUP", " ") importlib.reload(config) - assert config.MESH_UDP_GROUP == "224.0.0.69" + assert config.MESH_UDP_GROUP == "239.0.0.69,224.0.0.69" def test_mesh_udp_port_custom_value(self, monkeypatch): """A custom MESH_UDP_PORT is parsed to an int.""" diff --git a/tests/test_meshtastic_udp_unit.py b/tests/test_meshtastic_udp_unit.py index fac28420..cc4f0308 100644 --- a/tests/test_meshtastic_udp_unit.py +++ b/tests/test_meshtastic_udp_unit.py @@ -19,21 +19,26 @@ Coverage strategy mirrors ``tests/test_meshtastic_udp_decode_unit.py``: ``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. + the no-``decoded`` drop path, the receive loop's wait/timeout/OSError/ + dispatch branches, and the lifecycle of :class:`_UdpInterface`) with fakes: + a scripted ``select`` drives the receive loop synchronously, with no thread. +3. **Connect tests** join every group over local ``AF_UNIX`` socketpairs, so + the real ``select`` waits on real file descriptors and one test runs the + real reader thread end-to-end. -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. +No test opens a real network socket or a wait of more than a few +milliseconds: the scripted ``select`` sets the interface's stop flag once its +script is spent, and the real-thread test shortens the loop's poll interval, +so a hung test is not possible. """ from __future__ import annotations import base64 +import importlib import json import os +import select import socket import sys import threading @@ -313,103 +318,190 @@ class TestHandleDatagramDropPaths: # --------------------------------------------------------------------------- +class _ScriptedSock: + """Fake group socket replaying scripted ``recvfrom`` results. + + Each positional entry is the bytes of one datagram to return or an + exception instance to raise, consumed in order. ``close()`` records the + call and raises *close_error* when one is given. + """ + + def __init__(self, *script, close_error: BaseException | None = None): + """Queue *script* for ``recvfrom`` and remember *close_error*.""" + self._script = list(script) + self._close_error = close_error + self.closed = False + + def recvfrom(self, bufsize): + """Return the next scripted datagram, or raise the next scripted error.""" + step = self._script.pop(0) + if isinstance(step, BaseException): + raise step + return step, ("192.0.2.1", 4403) + + def close(self): + """Record the close, raising the configured error if there is one.""" + self.closed = True + if self._close_error is not None: + raise self._close_error + + +class _ScriptedSelect: + """Stand-in for the :mod:`select` module replaying scripted readiness. + + Installed as ``udp_mod.select``, so only the provider sees it. Each wait + consumes one step: the list of sockets to report readable, or an + exception instance to raise. Once the script is spent it sets the + interface's stop flag and reports nothing ready, so a loop under test + always terminates. + """ + + def __init__(self, iface, *script): + """Queue *script* for the waits of the loop serving *iface*.""" + self._iface = iface + self._script = list(script) + self.waits: list[tuple[list, float]] = [] + + def select(self, rlist, wlist, xlist, timeout): + """Record the wait, then return (or raise) the next scripted step.""" + self.waits.append((list(rlist), timeout)) + if not self._script: + self._iface._stop.set() + return [], [], [] + step = self._script.pop(0) + if isinstance(step, BaseException): + raise step + return step, [], [] + + +@pytest.fixture +def run_loop(monkeypatch): + """Run ``_recv_loop`` synchronously over scripted sockets and readiness. + + Returns ``run(socks, *script, handle=None)``: it builds a connected + interface over *socks*, installs a :class:`_ScriptedSelect` replaying + *script*, replaces ``_handle_datagram`` with *handle* (by default, record + the raw bytes), runs the loop to completion, and returns + ``(iface, selector, handled)``. + """ + + def run(socks, *script, handle=None): + """Run the loop to completion over *socks*, replaying *script*.""" + iface = _UdpInterface() + iface._socks = list(socks) + iface.isConnected.set() + selector = _ScriptedSelect(iface, *script) + monkeypatch.setattr(udp_mod, "select", selector) + handled: list[bytes] = [] + provider = MeshtasticUdpProvider() + monkeypatch.setattr( + provider, + "_handle_datagram", + handle or (lambda raw, _iface: handled.append(raw)), + ) + provider._recv_loop(iface) + return iface, selector, handled + + return run + + 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) + def test_idle_wait_rechecks_stop_flag(self, run_loop): + """An empty wait (the poll timeout) loops back to the stop-flag check.""" + sock = _ScriptedSock() + iface, selector, handled = run_loop([sock], []) + assert selector.waits == [([sock], udp_mod._RECV_POLL_SECS)] * 2 + assert handled == [] + # Loop exit clears isConnected so a dead reader is detectable. 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] = [] + def test_dispatches_datagrams_from_either_group_socket(self, run_loop): + """One thread serves every group: whichever socket is ready is read. - 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) + The first wait reports only the second group's socket ready, the next + reports both, and every wait covers both sockets. + """ + new_group = _ScriptedSock(b"from-239") + old_group = _ScriptedSock(b"from-224", b"again-224") + _iface, selector, handled = run_loop( + [new_group, old_group], [old_group], [new_group, old_group] ) - provider._recv_loop(iface) - assert handled == [raw] + assert handled == [b"from-224", b"from-239", b"again-224"] + assert all(rlist == [new_group, old_group] for rlist, _ in selector.waits) - def test_handle_datagram_exception_is_swallowed_loop_survives(self, monkeypatch): + def test_ready_socket_that_times_out_is_skipped(self, run_loop): + """Readable-then-empty skips only that socket; the loop carries on.""" + quiet = _ScriptedSock(socket.timeout()) + busy = _ScriptedSock(b"datagram") + _iface, selector, handled = run_loop([quiet, busy], [quiet, busy]) + + assert handled == [b"datagram"] + assert len(selector.waits) == 2 + + @pytest.mark.parametrize("error", [OSError("bad fd"), ValueError("fd -1")]) + def test_wait_error_ends_loop_and_clears_connected(self, run_loop, error): + """A socket closed under the wait ends the loop and marks it dead.""" + iface, selector, _handled = run_loop([_ScriptedSock()], error) + + assert len(selector.waits) == 1 + assert not iface.isConnected.is_set() + + def test_recv_oserror_ends_loop_and_clears_connected(self, run_loop): + """An OSError from recvfrom (socket closed after the wait) ends the loop.""" + dead = _ScriptedSock(OSError("socket closed")) + iface, selector, handled = run_loop([dead], [dead]) + + assert len(selector.waits) == 1 + assert handled == [] + assert not iface.isConnected.is_set() + + def test_handle_datagram_exception_is_swallowed_loop_survives(self, run_loop): """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 + raises on the first datagram; the loop must go on to the second and exit cleanly on the stop flag rather than propagating. """ + seen: list[bytes] = [] + + def boom_once(raw, _iface): + """Record *raw*, raising for the ``b"bad"`` datagram only.""" + seen.append(raw) + if raw == b"bad": + raise ValueError("simulated bad datagram") + + sock = _ScriptedSock(b"bad", b"good") + iface, _selector, _handled = run_loop( + [sock], [sock], [sock], handle=boom_once + ) # must not raise + + assert seen == [b"bad", b"good"] + assert not iface.isConnected.is_set() + + def test_real_select_rejects_a_closed_socket_and_the_loop_ends(self): + """The real ``select`` raises ValueError on a closed socket; the loop ends. + + Pins the premise of the ValueError branch with no script: a closed + socket's ``fileno()`` is -1, which ``select`` rejects. That is the race + where :meth:`_UdpInterface.close` runs between the stop-flag check and + the wait. + """ + receiver, sender = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM) + sender.close() + receiver.close() + with pytest.raises(ValueError): + select.select([receiver], [], [], 0) + iface = _UdpInterface() + iface._socks = [receiver] iface.isConnected.set() - calls = {"n": 0} + MeshtasticUdpProvider()._recv_loop(iface) # must return, not raise - 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() @@ -422,33 +514,33 @@ 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.""" + """A fresh interface has no nodes, sockets or thread, and is not connected.""" iface = _UdpInterface() assert iface.nodes == {} assert isinstance(iface.isConnected, threading.Event) assert not iface.isConnected.is_set() - assert iface._sock is None + assert iface._socks == [] 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.""" + """close() must not raise when no socket was opened and no thread started.""" 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().""" + def test_close_closes_every_socket_and_swallows_oserror(self): + """close() closes each group socket, even after one close() raises.""" iface = _UdpInterface() + failing = _ScriptedSock(close_error=OSError("already closed")) + healthy = _ScriptedSock() + iface._socks = [failing, healthy] - class RaisingSock: - def close(self): - raise OSError("already closed") - - iface._sock = RaisingSock() iface.close() # must not raise + + assert failing.closed and healthy.closed assert iface._stop.is_set() def test_close_joins_thread(self): @@ -473,20 +565,106 @@ class TestUdpInterfaceLifecycle: # --------------------------------------------------------------------------- -# MeshtasticUdpProvider.connect (full lifecycle through a fake socket) +# MeshtasticUdpProvider.connect (group joins over local socketpairs) # --------------------------------------------------------------------------- -class TestConnectLifecycle: - """Exercises connect() end-to-end with a fake socket and a real thread.""" +class _GroupSockets: + """Stand-in for ``open_multicast_socket`` backed by local socketpairs. - def test_connect_returns_triple_and_receives_then_closes(self, monkeypatch): - """connect() starts the receive thread; close() stops it cleanly.""" + Each join records ``(group, port)`` and returns the receiving end of a + fresh ``AF_UNIX`` datagram socketpair: a real file descriptor the real + ``select`` can wait on, with no network socket involved. The sending end + stays here so a test can deliver a datagram to one group's socket. + """ + + def __init__(self) -> None: + """Start with no joins recorded and no group set to fail.""" + self.joined: list[tuple[str, int]] = [] + self.fail_on: str | None = None + self._receivers: dict[str, socket.socket] = {} + self._senders: dict[str, socket.socket] = {} + + def open(self, group: str, port: int) -> socket.socket: + """Join *group* on *port*, or raise ``OSError`` when it is :attr:`fail_on`.""" + if group == self.fail_on: + raise OSError(f"simulated join failure for {group}") + self.joined.append((group, port)) + receiver, sender = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM) + receiver.settimeout(1.0) # as open_multicast_socket does + self._receivers[group] = receiver + self._senders[group] = sender + return receiver + + def send(self, group: str, payload: bytes) -> None: + """Deliver *payload* to the provider's socket for *group*.""" + self._senders[group].send(payload) + + def is_closed(self, group: str) -> bool: + """Return whether the provider's socket for *group* has been closed.""" + return self._receivers[group].fileno() == -1 + + def close_all(self) -> None: + """Close both ends of every socketpair handed out.""" + for sock in (*self._receivers.values(), *self._senders.values()): + sock.close() + + +@pytest.fixture +def group_sockets(monkeypatch): + """Route the provider's group joins to a fresh :class:`_GroupSockets`.""" + sockets = _GroupSockets() + monkeypatch.setattr(udp_mod, "open_multicast_socket", sockets.open) + yield sockets + sockets.close_all() + + +@pytest.fixture +def idle_reader(monkeypatch): + """Make the receive thread a no-op, for tests that only inspect connect().""" + monkeypatch.setattr(MeshtasticUdpProvider, "_recv_loop", lambda self, iface: None) + + +@pytest.fixture +def shipped_config(monkeypatch): + """Reload ``config`` with the UDP group and port unset, as shipped. + + Reloads again on teardown. That reload runs before ``monkeypatch`` + restores the variables, so it yields the shipped defaults, not the + caller's environment; a later test that needs environment-derived values + must reload ``config`` itself. + """ + monkeypatch.delenv("MESH_UDP_GROUP", raising=False) + monkeypatch.delenv("MESH_UDP_PORT", raising=False) + yield importlib.reload(udp_mod.config) + importlib.reload(udp_mod.config) + + +class TestConnectLifecycle: + """Exercises connect() end-to-end over local sockets and a real thread.""" + + def test_connect_returns_triple_and_receives_then_closes( + self, monkeypatch, group_sockets + ): + """One reader thread serves every group; close() releases them all. + + The real captured primary datagram arrives on one group's socket and + a synthetic primary packet on the other's; both must reach on_receive. + """ primary_raw, _private_raw = _load_fixture_raw() + captured = mesh_pb2.MeshPacket() + captured.ParseFromString(primary_raw) + synthetic_raw = _encrypt_packet( + udp_decode.channel_hash("MediumFast", "AQ=="), packet_id=0x5151 + ) 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_GROUPS", ("239.0.0.69", "224.0.0.69") + ) monkeypatch.setattr(udp_mod.config, "MESH_UDP_PORT", 4403) + # A short poll keeps close() from idling a full second in select. + monkeypatch.setattr(udp_mod, "_RECV_POLL_SECS", 0.01) received: list[dict] = [] monkeypatch.setattr( @@ -495,45 +673,76 @@ class TestConnectLifecycle: 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 target == "udp://239.0.0.69,224.0.0.69:4403" assert next_candidate == "ignored" assert iface.isConnected.is_set() + group_sockets.send("224.0.0.69", primary_raw) + group_sockets.send("239.0.0.69", synthetic_raw) deadline = time.monotonic() + 2.0 - while not received and time.monotonic() < deadline: + while len(received) < 2 and time.monotonic() < deadline: time.sleep(0.01) - assert len(received) == 1 + assert sorted(packet["id"] for packet in received) == sorted( + [captured.id, 0x5151] + ) iface.close() assert not iface._thread.is_alive() assert not iface.isConnected.is_set() - assert fake_sock.closed + assert group_sockets.is_closed("239.0.0.69") + assert group_sockets.is_closed("224.0.0.69") + + +@pytest.mark.usefixtures("idle_reader") +class TestConnectGroups: + """connect() opens one group-bound socket per configured group.""" + + def test_shipped_default_joins_both_groups(self, shipped_config, group_sockets): + """Unconfigured, both Meshtastic groups are joined (issue #903). + + Firmware 2.8 and later sends to 239.0.0.69 and earlier firmware to + 224.0.0.69, so a listener on only one group hears nothing from the + other firmware line. + """ + iface, target, _next = MeshtasticUdpProvider().connect(active_candidate=None) + iface.close() + + assert group_sockets.joined == [("239.0.0.69", 4403), ("224.0.0.69", 4403)] + assert target == "udp://239.0.0.69,224.0.0.69:4403" + + def test_one_configured_group_opens_one_socket(self, monkeypatch, group_sockets): + """A single configured group listens on that group only.""" + monkeypatch.setattr(udp_mod.config, "MESH_UDP_GROUPS", ("224.0.0.69",)) + monkeypatch.setattr(udp_mod.config, "MESH_UDP_PORT", 4403) + + iface, target, _next = MeshtasticUdpProvider().connect(active_candidate=None) + iface.close() + + assert group_sockets.joined == [("224.0.0.69", 4403)] + assert target == "udp://224.0.0.69:4403" + + def test_failed_join_closes_joined_sockets_and_propagates( + self, monkeypatch, group_sockets + ): + """A group that cannot be joined fails connect() without leaking the others. + + The daemon retries a failed connect with backoff, so a socket left open + on each attempt would pile up for as long as the group stays unjoinable. + """ + monkeypatch.setattr( + udp_mod.config, "MESH_UDP_GROUPS", ("239.0.0.69", "224.0.0.69") + ) + group_sockets.fail_on = "224.0.0.69" + + with pytest.raises(OSError, match="224.0.0.69"): + MeshtasticUdpProvider().connect(active_candidate=None) + + assert [group for group, _port in group_sockets.joined] == ["239.0.0.69"] + assert group_sockets.is_closed("239.0.0.69") # --------------------------------------------------------------------------- @@ -636,27 +845,12 @@ class TestPrimaryChannelFilter: assert len(captured) == 1 # primary still accepted +@pytest.mark.usefixtures("group_sockets", "idle_reader") 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) @@ -676,7 +870,6 @@ class TestConnectLogsPrimaryFilter: 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] = []