diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 122d379..b4ed78c 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -265,12 +265,30 @@ class PacketRouter: return False def _companion_bridges_for_packet(self, packet, metadata: dict) -> dict: - """Return companion bridges unless policy drop pre-check blocks delivery.""" + """Return companion bridges unless policy drop or origin exclusion blocks delivery. + + This is the single choke point every companion fan-out reads, so the + "a node never hears its own transmission" rule is applied here once and + covers ADVERT, ACK, PATH, RESPONSE, GRP_TXT, GRP_DATA and the dest-hash + candidate paths — the same rule ``inject_packet`` already applies to the + 0x88 raw-RX echo. + """ companion_bridges = getattr(self.daemon, "companion_bridges", {}) if not companion_bridges: return {} if self._policy_blocks_companion(packet, metadata): return {} + # Both isinstance checks are load-bearing, not defensive noise: bridges are + # keyed by the int hash byte (main.py:880) while origin_hash is the "0xHH" + # text form the frame servers compare against (main.py:787, :1167), and a + # non-int key would raise inside the f-string. Do not simplify either away. + origin_hash = getattr(packet, "_injected_origin_hash", None) + if isinstance(origin_hash, str): + companion_bridges = { + key: bridge + for key, bridge in companion_bridges.items() + if not (isinstance(key, int) and f"0x{key:02x}" == origin_hash) + } return companion_bridges async def _fan_out_to_bridges(self, packet, bridges, *, context: str) -> tuple: @@ -385,6 +403,14 @@ class PacketRouter: "timestamp": getattr(packet, "timestamp", 0), } + # Tag the originating companion before anything reaches the air: enqueue() + # is a bare queue.put, so the packet itself is the only carrier that + # survives to _route_packet, where the fan-out withholds it from that + # bridge — the same "a node never hears its own transmission" rule the + # 0x88 echo below applies. Set ahead of the TX so a core without the + # matching Packet slot fails before transmitting rather than after. + packet._injected_origin_hash = origin_hash + # Serialize injects so one local TX completes before the next runs # (avoids duty-cycle or dispatcher races where a later packet goes out first) async with self._inject_lock: @@ -421,7 +447,10 @@ class PacketRouter: except Exception as e: logger.debug("Failed to echo injected TX to companions: %s", e) - # Enqueue so router can deliver to companion(s): TXT_MSG -> dest bridge, ACK -> all bridges (sender sees ACK) + # Enqueue so router can deliver to the *other* companion(s): TXT_MSG -> dest + # bridge, ACK/ADVERT/GRP_TXT/GRP_DATA -> every bridge except the origin. + # The originating companion never gets its own packet back: that exclusion + # is enforced once in _companion_bridges_for_packet via the tag set above. await self.enqueue(packet) if wait_for_ack: @@ -572,8 +601,13 @@ class PacketRouter: await deliver(snr, rssi, path_len, path_bytes, payload_bytes) elif payload_type == AdvertHandler.payload_type(): - # Process advertisement packet for neighbor tracking - if self.daemon.advert_helper: + # Process advertisement packet for neighbor tracking. A locally injected + # advert is our own TX, never something heard off the air, so feeding it + # to the helper would record a phantom neighbour from the zeroed rssi/snr + # and burn a rate-limit token. AdvertHelper's own self-skip + # (handler_helpers/advert.py) only knows the repeater identity, not the + # co-hosted companions. Same _injected_for_tx idiom as the TRACE branches. + if self.daemon.advert_helper and not getattr(packet, "_injected_for_tx", False): rssi = getattr(packet, "rssi", 0) snr = getattr(packet, "snr", 0.0) await self.daemon.advert_helper.process_advert_packet(packet, rssi, snr) diff --git a/tests/test_companion_injector_wiring.py b/tests/test_companion_injector_wiring.py new file mode 100644 index 0000000..22e0bd6 --- /dev/null +++ b/tests/test_companion_injector_wiring.py @@ -0,0 +1,79 @@ +"""Every companion bridge must inject with its own hash as origin_hash. + +The origin exclusion in PacketRouter (the 0x88 raw-RX echo and the companion +fan-out) is only as good as this binding: an injector built without +``origin_hash``, or with the wrong companion's hash, silently sends a +companion its own transmission back. Nothing else asserts the binding, so it +is pinned here for both the config-load path and the hot-add path. +""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from openhop_core import LocalIdentity +from repeater.companion.frame_server import CompanionFrameServer +from repeater.identity_manager import IdentityManager +from repeater.main import RepeaterDaemon +from repeater.packet_router import PacketRouter + +# Distinct seeds whose derived public keys do not collide on the first byte, +# so the two companions occupy different companion_bridges slots. +LOADED_KEY = "11" * 32 # public key starts 0xd0 +HOT_ADDED_KEY = "22" * 32 # public key starts 0xa0 + + +def _config(companions=()): + return { + "repeater": {"node_name": "n", "identity_key": b"\x10" * 32}, + "logging": {}, + "radio": {}, + "identities": {"companions": list(companions), "room_servers": []}, + } + + +def _daemon(companions=()): + daemon = RepeaterDaemon(_config(companions), radio=object()) + daemon.identity_manager = IdentityManager({}) + daemon.router = PacketRouter(daemon) + return daemon + + +def _expected_hash(identity_key_hex: str) -> str: + return f"0x{LocalIdentity(seed=bytes.fromhex(identity_key_hex)).get_public_key()[0]:02x}" + + +def _assert_injector_bound(daemon, expected_hash: str) -> None: + """The bridge injects as expected_hash, and its frame server agrees.""" + bridge = daemon.companion_bridges[int(expected_hash, 16)] + injector = bridge._packet_injector + + assert injector.func == daemon.router.inject_packet + assert injector.keywords["origin_hash"] == expected_hash + # The frame server compares its own companion_hash against the exclude_hash + # inject_packet passes down, so the two must be the same string. + frame_server = next(fs for fs in daemon.companion_frame_servers if fs.bridge is bridge) + assert frame_server.companion_hash == expected_hash + # Same string again as the bridge's SQLite namespace key. + assert bridge._companion_hash == expected_hash + + +@pytest.mark.asyncio +async def test_companion_bridge_injector_is_bound_to_its_own_origin_hash(): + daemon = _daemon(companions=({"name": "loaded", "identity_key": LOADED_KEY},)) + + with patch.object(CompanionFrameServer, "start", AsyncMock()): + await daemon._load_companion_identities() + + assert len(daemon.companion_bridges) == 1 + _assert_injector_bound(daemon, _expected_hash(LOADED_KEY)) + + # Hot-add path: a second companion added at runtime must be bound to its + # own hash, not to the one already loaded. + await daemon.add_companion_from_config( + {"name": "hot", "identity_key": HOT_ADDED_KEY, "settings": {"tcp_port": 5001}} + ) + + assert len(daemon.companion_bridges) == 2 + _assert_injector_bound(daemon, _expected_hash(LOADED_KEY)) + _assert_injector_bound(daemon, _expected_hash(HOT_ADDED_KEY)) diff --git a/tests/test_packet_router.py b/tests/test_packet_router.py index a1e46f5..dadcb81 100644 --- a/tests/test_packet_router.py +++ b/tests/test_packet_router.py @@ -15,6 +15,7 @@ or: """ import asyncio +import functools import hashlib import time import unittest @@ -36,12 +37,14 @@ from openhop_core.node.handlers.text import TextMessageHandler from openhop_core.node.handlers.trace import TraceHandler from openhop_core.protocol.constants import ( PAYLOAD_TYPE_GRP_DATA, + PAYLOAD_TYPE_GRP_TXT, ROUTE_TYPE_DIRECT, ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_DIRECT, ) from openhop_core.protocol import LocalIdentity, Packet, PacketBuilder +from repeater.companion.bridge import RepeaterCompanionBridge from repeater.packet_router import ( PacketRouter, _companion_dedup_key, @@ -83,6 +86,7 @@ def _make_packet(payload_type: int = 0xFF): pkt.snr = 5.0 pkt.timestamp = time.time() pkt._injected_for_tx = False + pkt._injected_origin_hash = None pkt.path = bytearray() pkt.calculate_packet_hash.return_value = b"\x01" * 32 pkt.mark_do_not_retransmit = MagicMock() @@ -1323,6 +1327,119 @@ class TestInjectedTxRawEcho(unittest.IsolatedAsyncioTestCase): self.assertTrue(ok) +class TestInjectedOriginExclusion(unittest.IsolatedAsyncioTestCase): + """A companion never hears its own transmission back through the router. + + inject_packet tags the packet with the originating companion's "0xHH" hash + and _companion_bridges_for_packet withholds that bridge from every fan-out, + the same rule the 0x88 raw-RX echo already applies to the frame servers. + """ + + @staticmethod + def _daemon_with_two_bridges(): + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = None + origin = _make_bridge() + other = _make_bridge() + daemon.companion_bridges = {0x1A: origin, 0x2B: other} + return daemon, origin, other + + @staticmethod + async def _inject_and_drain(router, pkt, origin_hash): + await router.start() + try: + injected = await router.inject_packet(pkt, origin_hash=origin_hash) + await asyncio.sleep(0.05) # let queue drain into _route_packet + finally: + await router.stop() + return injected + + async def test_injected_advert_is_not_redelivered_to_the_originating_bridge(self): + """A companion's own advert reaches the other bridge only.""" + daemon, origin, other = self._daemon_with_two_bridges() + router = PacketRouter(daemon) + pkt = _make_packet(AdvertHandler.payload_type()) + + self.assertTrue(await self._inject_and_drain(router, pkt, "0x1a")) + + origin.process_received_packet.assert_not_awaited() + other.process_received_packet.assert_awaited_once() + + async def test_inject_without_origin_hash_still_fans_out_to_all_bridges(self): + """Repeater-originated injects (trace, discovery, login, text) have no + origin companion, so every bridge must still be offered the packet.""" + daemon, origin, other = self._daemon_with_two_bridges() + router = PacketRouter(daemon) + pkt = _make_packet(AdvertHandler.payload_type()) + + self.assertTrue(await self._inject_and_drain(router, pkt, None)) + + origin.process_received_packet.assert_awaited_once() + other.process_received_packet.assert_awaited_once() + + async def test_ota_advert_still_reaches_every_bridge(self): + """Anti-over-blocking control: an advert heard off the air is untagged, + so the origin filter must leave both bridges in place.""" + daemon, origin, other = self._daemon_with_two_bridges() + router = PacketRouter(daemon) + + await router._route_packet(_make_packet(AdvertHandler.payload_type())) + + origin.process_received_packet.assert_awaited_once() + other.process_received_packet.assert_awaited_once() + + async def test_injected_group_message_is_not_returned_to_its_origin(self): + """GRP_TXT fans out to every companion, so it needs the same exclusion.""" + daemon, origin, other = self._daemon_with_two_bridges() + router = PacketRouter(daemon) + pkt = _make_packet(PAYLOAD_TYPE_GRP_TXT) + + self.assertTrue(await self._inject_and_drain(router, pkt, "0x1a")) + + origin.process_received_packet.assert_not_awaited() + other.process_received_packet.assert_awaited_once() + + async def test_injected_advert_skips_advert_helper_neighbour_tracking(self): + """Our own TX is not a neighbour observation: feeding it to the helper + would record a phantom neighbour from the zeroed rssi/snr and burn a + rate-limit token. An advert heard off the air must still reach it.""" + daemon, _origin, _other = self._daemon_with_two_bridges() + daemon.advert_helper = MagicMock() + daemon.advert_helper.process_advert_packet = AsyncMock() + router = PacketRouter(daemon) + + injected = _make_packet(AdvertHandler.payload_type()) + self.assertTrue(await self._inject_and_drain(router, injected, "0x1a")) + daemon.advert_helper.process_advert_packet.assert_not_awaited() + + await router._route_packet(_make_packet(AdvertHandler.payload_type())) + daemon.advert_helper.process_advert_packet.assert_awaited_once() + + async def test_self_advert_never_returns_to_its_own_companion_end_to_end(self): + """A real bridge advertising through the real injector must not end up + holding itself as a contact once the router queue has drained.""" + identity = LocalIdentity() + companion_hash = identity.get_public_key()[0] + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = None + router = PacketRouter(daemon) + bridge = RepeaterCompanionBridge( + identity, + functools.partial(router.inject_packet, origin_hash=f"0x{companion_hash:02x}"), + companion_hash=f"0x{companion_hash:02x}", + ) + daemon.companion_bridges = {companion_hash: bridge} + + await router.start() + try: + self.assertTrue(await bridge.advertise()) + await asyncio.sleep(0.05) # let queue drain into _route_packet + finally: + await router.stop() + + self.assertEqual(bridge.get_contact_count(), 0) + + class TestCompanionDeliveryFailureHandling(unittest.IsolatedAsyncioTestCase): """Dedupe marking and candidate-loop behaviour when companion bridges raise."""