Merge pull request #377 from agessaman/feat/origin-exclusion

fix(repeater): implement origin exclusion for injected packets

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)

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))

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."""