From 86c7f52a696d8c7fb69e2cc5bc959304f0d27af7 Mon Sep 17 00:00:00 2001 From: Afri <220195275+l5yth@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:41:08 +0200 Subject: [PATCH] data: keep CodeQL off a blank UDP group and a test channel name (#1006) --- data/mesh_ingestor/config.py | 18 +++++++++++------- tests/test_channel_scope_unit.py | 28 +++++++++++++++------------- tests/test_config_udp_groups_unit.py | 16 ++++++++++++++++ 3 files changed, 42 insertions(+), 20 deletions(-) diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index 8b14ed53..5069ae17 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -397,19 +397,23 @@ def _parse_mesh_udp_groups(raw: str) -> tuple[str, ...]: if not entry: continue try: - is_multicast = ipaddress.IPv4Address(entry).is_multicast + address = ipaddress.IPv4Address(entry) except ValueError: # not an IPv4 address at all - is_multicast = False - if not is_multicast: + address = None + if address is None or not address.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) + # Keep the parsed address rather than the raw fragment. IPv4Address + # accepts only canonical dotted-decimal (no leading zeros), so the text + # is the same, but only a validated group can reach ``bind``: CodeQL + # (py/bind-socket-all-network-interfaces) traced the blank that + # ``_clean_env_fragment`` returns through ``entry`` (SPEC CQ1). + group = str(address) + if group not in groups: + groups.append(group) return tuple(groups) or DEFAULT_MESH_UDP_GROUPS diff --git a/tests/test_channel_scope_unit.py b/tests/test_channel_scope_unit.py index a4030074..d6b8324a 100644 --- a/tests/test_channel_scope_unit.py +++ b/tests/test_channel_scope_unit.py @@ -73,7 +73,7 @@ SENDER_ID = "!a1b2c3d4" PRIMARY = 0 """Index of the probe radio's primary channel, ``LongFast``.""" -SECRET = 1 +SECONDARY = 1 """Index of the probe radio's secondary channel, ``Secret``.""" MODES = { @@ -178,7 +178,7 @@ class _ChannelTable: settings=channel_pb2.ChannelSettings(name="LongFast"), ), channel_pb2.Channel( - index=SECRET, + index=SECONDARY, role=role.SECONDARY, settings=channel_pb2.ChannelSettings(name="Secret"), ), @@ -304,12 +304,12 @@ class TestIngestFilterReason: def test_excluded_channel_names_the_reason(self, scope, mode): """Each mode drops ``Secret`` with its own reason and admits ``LongFast``.""" scope.apply(mode) - assert channels.ingest_filter_reason(SECRET) == MODE_REASONS[mode] + assert channels.ingest_filter_reason(SECONDARY) == MODE_REASONS[mode] assert channels.ingest_filter_reason(PRIMARY) is None def test_no_filter_admits_every_channel(self, scope): """With nothing configured every channel, named or not, is admitted.""" - assert channels.ingest_filter_reason(SECRET) is None + assert channels.ingest_filter_reason(SECONDARY) is None assert channels.ingest_filter_reason(7) is None def test_unnamed_channel_is_dropped_by_an_allowlist(self, scope): @@ -330,7 +330,9 @@ class TestIngestFilterReason: assert channels.ingest_filter_reason(None, via_mqtt=True) == "via_mqtt" assert channels.ingest_filter_reason(PRIMARY) is None scope.apply("hidden") - assert channels.ingest_filter_reason(SECRET, via_mqtt=True) == "hidden-channel" + assert ( + channels.ingest_filter_reason(SECONDARY, via_mqtt=True) == "hidden-channel" + ) @pytest.mark.parametrize( "modes, expected", @@ -345,7 +347,7 @@ class TestIngestFilterReason: ``PRIMARY_CHANNEL_ONLY``, ``ALLOWED_CHANNELS``, ``HIDDEN_CHANNELS`` names the reason.""" scope.apply(*modes) - assert channels.ingest_filter_reason(SECRET) == expected + assert channels.ingest_filter_reason(SECONDARY) == expected def test_unnamed_primary_channel_matches_primary_channel_name(self, scope): """With no names captured (the passive UDP transport), channel 0 is @@ -355,7 +357,7 @@ class TestIngestFilterReason: assert channels.ingest_filter_reason(PRIMARY) == "disallowed-channel" scope.apply(PRIMARY_CHANNEL_NAME="MediumFast") assert channels.ingest_filter_reason(PRIMARY) is None - assert channels.ingest_filter_reason(SECRET) == "disallowed-channel" + assert channels.ingest_filter_reason(SECONDARY) == "disallowed-channel" scope.apply(ALLOWED_CHANNELS=(), HIDDEN_CHANNELS=("MediumFast",)) assert channels.ingest_filter_reason(PRIMARY) == "hidden-channel" @@ -378,7 +380,7 @@ class TestStorePacketDictChannelScope: def test_filtered_channel_packet_not_forwarded(self, radio, mode, port): """A packet heard on ``Secret`` queues nothing, whatever its port.""" radio.apply(mode) - radio.hear(port, channel=SECRET) + radio.hear(port, channel=SECONDARY) assert ( radio.posts == [] ), f"{mode}: {port} heard on 'Secret' was queued to {_routes(radio.posts)}" @@ -408,7 +410,7 @@ class TestStorePacketDictChannelScope: """DEBUG keeps each filter's log line and fields, now for every port.""" radio.apply(mode, DEBUG=True) capsys.readouterr() - radio.hear("POSITION_APP", channel=SECRET) + radio.hear("POSITION_APP", channel=SECONDARY) out = capsys.readouterr().out assert line in out and field in out @@ -453,7 +455,7 @@ class TestStorePacketDictChannelScope: """A handler-stamped ``decoded.channel`` is the attribution used.""" scope.apply("hidden") packet = _position_packet(channel=PRIMARY) - packet["decoded"]["channel"] = SECRET + packet["decoded"]["channel"] = SECONDARY handlers.store_packet_dict(packet) assert scope.posts == [] and scope.reasons == ["hidden-channel"] @@ -526,7 +528,7 @@ class TestSnapshotChannelScope: def node_db(self, radio): """Load a primary, a ``Secret`` and a via_mqtt node into the library nodeDB.""" radio.iface._handleFromRadio(_node_info_frame(0x0A0A0A0A)) - radio.iface._handleFromRadio(_node_info_frame(0x0B0B0B0B, channel=SECRET)) + radio.iface._handleFromRadio(_node_info_frame(0x0B0B0B0B, channel=SECONDARY)) radio.iface._handleFromRadio(_node_info_frame(0x0C0C0C0C, via_mqtt=True)) return radio @@ -589,7 +591,7 @@ class TestSnapshotChannelScope: (here a text the live path dropped) stays out of the upsert body.""" radio.apply("hidden") radio.iface._handleFromRadio(_node_info_frame(SENDER)) - radio.hear("TEXT_MESSAGE_APP", channel=SECRET) + radio.hear("TEXT_MESSAGE_APP", channel=SECONDARY) assert radio.reasons == ["hidden-channel"] assert "lastReceived" in radio.iface.nodes[SENDER_ID] upserts = _snapshot(radio) @@ -620,7 +622,7 @@ class TestMeshtasticSnapshotHooks: """``channel`` (absent = 0) and ``viaMqtt`` drive the shared policy.""" scope.apply("hidden", DROP_VIA_MQTT=True) provider = MeshtasticProvider() - assert provider.snapshot_filter_reason("!1", {"channel": SECRET}) == ( + assert provider.snapshot_filter_reason("!1", {"channel": SECONDARY}) == ( "hidden-channel" ) assert provider.snapshot_filter_reason("!2", {"viaMqtt": True}) == "via_mqtt" diff --git a/tests/test_config_udp_groups_unit.py b/tests/test_config_udp_groups_unit.py index 259b93d2..84866016 100644 --- a/tests/test_config_udp_groups_unit.py +++ b/tests/test_config_udp_groups_unit.py @@ -25,6 +25,7 @@ there would keep every such deployment deaf to one firmware line. from __future__ import annotations +import ipaddress import re import sys from pathlib import Path @@ -82,6 +83,21 @@ class TestParseMeshUdpGroups: """ assert config._parse_mesh_udp_groups(raw) == EXPECTED_GROUPS + def test_every_group_is_the_parsed_multicast_address(self): + """Each group is the canonical text of a validated address (SPEC CQ1). + + Quote-only and blank fragments around the addresses yield no group. + """ + groups = config._parse_mesh_udp_groups(" '' ,239.0.0.69,\"\", 224.0.0.69 ") + assert groups == EXPECTED_GROUPS + for group in groups: + address = ipaddress.IPv4Address(group) + assert (type(group), str(address), address.is_multicast) == ( + str, + group, + True, + ) + @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."""