Merge pull request #357 from agessaman/fix/consume-on-decrypt

fix(router): consume packets only on authenticated local handling (#353)

repeater/handler_helpers/login.py

@@ -253,7 +253,17 @@ class LoginHelper:
             handler = self.handlers.get(dest_hash)
             if handler:
                 logger.debug(f"Routing login to identity: hash=0x{dest_hash:02X}")
-                await handler(packet)
+                # The handler authenticates only when the request decrypted for
+                # this identity. Otherwise the dest hash collided with ours but the
+                # ANON_REQ is not really for us — do not consume it, so the engine
+                # can still forward/re-flood it (#353).
+                result = await handler(packet)
+                if not result.authenticated:
+                    logger.debug(
+                        f"ANON_REQ dest 0x{dest_hash:02X} did not decrypt for a local "
+                        f"identity (hash collision), allowing forward"
+                    )
+                    return False
                 packet.mark_do_not_retransmit()
                 return True
             else:

repeater/handler_helpers/protocol_request.py

@@ -124,13 +124,22 @@ class ProtocolRequestHelper:
             if not handler_info:
                 return False
 
-            # Let core handler build response
-            response_packet = await handler_info["handler"](packet)
+            # Let the core handler decrypt and build a response. It returns a
+            # HandlerResult: authenticated is False when the one-byte dest hash
+            # collided with ours but the REQ did not decrypt for a local client —
+            # do not consume it, so the engine can still forward/re-flood it.
+            result = await handler_info["handler"](packet)
+            if not result.authenticated:
+                logger.debug(
+                    f"REQ dest 0x{dest_hash:02X} did not decrypt for a local identity "
+                    f"(hash collision), allowing forward"
+                )
+                return False
 
             # Send response after delay
-            if response_packet and self.packet_injector:
+            if result.response and self.packet_injector:
                 await asyncio.sleep(SERVER_RESPONSE_DELAY_MS / 1000.0)
-                await self.packet_injector(response_packet, wait_for_ack=False)
+                await self.packet_injector(result.response, wait_for_ack=False)
 
             packet.mark_do_not_retransmit()
             return True

repeater/handler_helpers/text.py

@@ -235,8 +235,18 @@ class TextHelper:
                     f"dest=0x{dest_hash:02X}, src=0x{src_hash:02X}"
                 )
 
-                # Let handler decrypt the message first
-                await handler_info["handler"](packet)
+                # Let the handler attempt decryption. It authenticates only when
+                # the message actually decrypted for this identity. Otherwise the
+                # dest hash collided with ours (common with a 1-byte hash) but the
+                # packet is not really for us — do not consume it, so the engine can
+                # still forward/re-flood it (#353).
+                result = await handler_info["handler"](packet)
+                if not result.authenticated:
+                    logger.debug(
+                        f"TXT_MSG dest 0x{dest_hash:02X} did not decrypt for "
+                        f"'{handler_info['name']}' (hash collision), allowing forward"
+                    )
+                    return False
 
                 # Call placeholder for custom processing
                 await self._on_message_received(

repeater/packet_router.py

@@ -253,6 +253,41 @@ class PacketRouter:
             return {}
         return companion_bridges
 
+    async def _consume_via_local_candidates(
+        self, packet, metadata: dict, dest_hash, helper, process_method_name: str
+    ) -> bool:
+        """Try every local candidate that shares ``dest_hash`` and report consumption.
+
+        The on-air destination hash is only one byte, so several local identities
+        can share it. The candidates are the companion bridge registered at
+        ``dest_hash`` and the room-server / repeater identity registered in
+        ``helper`` (login, text, or protocol-request) at the same hash. Both are
+        offered the packet; the one whose key MAC-verifies it consumes it, the
+        other fails HMAC and no-ops.
+
+        Returns True only when at least one candidate authenticated (decrypted)
+        the packet. Consuming solely on authenticated handling is what lets a
+        one-byte prefix collision with a remote node — or a forged packet — fall
+        through to the forwarding engine instead of being swallowed.
+        """
+        companion_bridges = self._companion_bridges_for_packet(packet, metadata)
+        helper_handlers = getattr(helper, "handlers", {}) if helper else {}
+        has_companion = dest_hash is not None and dest_hash in companion_bridges
+        has_local_identity = dest_hash is not None and dest_hash in helper_handlers
+
+        consumed = False
+        if has_companion:
+            bridge_result = await companion_bridges[dest_hash].process_received_packet(packet)
+            if bridge_result.authenticated:
+                consumed = True
+        # Offer to the room-server / repeater identity when it shares the hash
+        # (collision) or when no local companion claims it at all (normal
+        # server-owned + remote-forward handling).
+        if helper and (has_local_identity or not has_companion):
+            if await getattr(helper, process_method_name)(packet):
+                consumed = True
+        return consumed
+
     def _record_for_ui(self, packet, metadata: dict) -> None:
         """Record an injection-only packet for the web UI (storage + recent_packets)."""
         handler = getattr(self.daemon, "repeater_handler", None)
@@ -474,33 +509,18 @@ class PacketRouter:
                     logger.debug(f"Companion bridge advert error: {e}")
 
         elif payload_type == LoginServerHandler.payload_type():
-            # Route to the local identity that owns this login. The on-air dest
-            # hash is only one byte, so a companion and a room-server identity
-            # can share it; decryption is the only real disambiguator. When both
-            # are registered under the same hash, offer the packet to both — the
-            # owner whose key decrypts replies, the other fails HMAC and no-ops.
-            # When dest is remote (not handled), login_helper passes it to the
-            # engine so DIRECT/FLOOD ANON_REQ can be forwarded. Our own injected
-            # ANON_REQ is suppressed by the engine's duplicate (mark_seen) check.
+            # Route ANON_REQ/login to the local identity that owns it. The on-air
+            # dest hash is only one byte, so a companion and a room-server identity
+            # can share it; decryption is the only real disambiguator. Offer the
+            # packet to every local candidate and consume only when one decrypts —
+            # a remote (or forged) ANON_REQ that merely collides on the prefix is
+            # then forwarded by the engine. Our own injected ANON_REQ is suppressed
+            # by the engine's duplicate (mark_seen) check.
             dest_hash = packet.payload[0] if packet.payload else None
-            companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            login_helper = self.daemon.login_helper
-            login_handlers = getattr(login_helper, "handlers", {}) if login_helper else {}
-
-            has_companion = dest_hash is not None and dest_hash in companion_bridges
-            has_room_server = dest_hash is not None and dest_hash in login_handlers
-
-            if has_companion:
-                await companion_bridges[dest_hash].process_received_packet(packet)
+            if await self._consume_via_local_candidates(
+                packet, metadata, dest_hash, self.daemon.login_helper, "process_login_packet"
+            ):
                 processed_by_injection = True
-            # Offer to login_helper when a room-server identity shares this hash
-            # (collision) or when no local companion claims it at all (normal
-            # repeater/room-server login + remote-forward handling).
-            if login_helper and (has_room_server or not has_companion):
-                handled = await login_helper.process_login_packet(packet)
-                if handled:
-                    processed_by_injection = True
-            if processed_by_injection:
                 self._record_for_ui(packet, metadata)
 
         elif payload_type == AckHandler.payload_type():
@@ -529,24 +549,15 @@ class PacketRouter:
         elif payload_type == TextMessageHandler.payload_type():
             # Same one-byte dest-hash collision handling as the login path above:
             # a companion and a room-server text identity can share a hash, and
-            # only decryption tells them apart. Offer to both when both are
-            # registered so a companion never shadows a room-server message.
+            # only decryption tells them apart. Offer to every local candidate and
+            # consume only when one decrypts, so a companion never shadows a
+            # room-server message and a prefix collision with a remote node still
+            # forwards.
             dest_hash = packet.payload[0] if packet.payload else None
-            companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            text_helper = self.daemon.text_helper
-            text_handlers = getattr(text_helper, "handlers", {}) if text_helper else {}
-
-            has_companion = dest_hash is not None and dest_hash in companion_bridges
-            has_text_identity = dest_hash is not None and dest_hash in text_handlers
-
-            if has_companion:
-                await companion_bridges[dest_hash].process_received_packet(packet)
+            if await self._consume_via_local_candidates(
+                packet, metadata, dest_hash, self.daemon.text_helper, "process_text_packet"
+            ):
                 processed_by_injection = True
-            if text_helper and (has_text_identity or not has_companion):
-                handled = await text_helper.process_text_packet(packet)
-                if handled:
-                    processed_by_injection = True
-            if processed_by_injection:
                 self._record_for_ui(packet, metadata)
 
         elif payload_type == PathHandler.payload_type():
@@ -654,26 +665,33 @@ class PacketRouter:
                 self._record_for_ui(packet, metadata)
 
         elif payload_type == ProtocolRequestHandler.payload_type():
+            # Same one-byte dest-hash collision handling as the login/text paths:
+            # a companion and a server (ACL) identity can share a hash and only
+            # decryption tells them apart. Offer to every local candidate and
+            # consume only when one authenticates the REQ; otherwise leave it for
+            # the engine.
             dest_hash = packet.payload[0] if packet.payload else None
-            companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            if dest_hash is not None and dest_hash in companion_bridges:
-                await companion_bridges[dest_hash].process_received_packet(packet)
+            if await self._consume_via_local_candidates(
+                packet,
+                metadata,
+                dest_hash,
+                self.daemon.protocol_request_helper,
+                "process_request_packet",
+            ):
                 processed_by_injection = True
                 self._record_for_ui(packet, metadata)
-            elif self.daemon.protocol_request_helper:
-                handled = await self.daemon.protocol_request_helper.process_request_packet(packet)
-                if handled:
+            else:
+                companion_bridges = self._companion_bridges_for_packet(packet, metadata)
+                if companion_bridges and _is_direct_final_hop(packet):
+                    # DIRECT with empty path: we're the final hop and cannot forward,
+                    # so deliver to all bridges for anon matching and consume regardless.
+                    for bridge in companion_bridges.values():
+                        try:
+                            await bridge.process_received_packet(packet)
+                        except Exception as e:
+                            logger.debug(f"Companion bridge REQ (final hop) error: {e}")
                     processed_by_injection = True
                     self._record_for_ui(packet, metadata)
-            elif companion_bridges and _is_direct_final_hop(packet):
-                # DIRECT with empty path: we're the final hop; deliver to all bridges for anon matching
-                for bridge in companion_bridges.values():
-                    try:
-                        await bridge.process_received_packet(packet)
-                    except Exception as e:
-                        logger.debug(f"Companion bridge REQ (final hop) error: {e}")
-                processed_by_injection = True
-                self._record_for_ui(packet, metadata)
 
         elif payload_type == GroupTextHandler.payload_type():
             # GRP_TXT: pass to all companions (they filter by channel); still forward.

tests/test_handler_helpers_path_protocol_text.py

@@ -5,10 +5,73 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from openhop_core.node.handlers.result import HandlerResult
+from openhop_core.protocol import Identity, LocalIdentity
+from openhop_core.protocol.packet_builder import PacketBuilder
+
+from repeater.handler_helpers.acl import ACL, PERM_ACL_ADMIN, ClientInfo
 from repeater.handler_helpers.path import PathHelper
 from repeater.handler_helpers.protocol_request import ProtocolRequestHelper
 from repeater.handler_helpers.text import TextHelper
 
+# ---------------------------------------------------------------------------
+# Real-crypto collision scaffolding (BUG-053 "try all local candidates").
+#
+# A one-byte destination hash can collide: a packet addressed to a remote node
+# (or a forged packet) can share its first public-key byte with a local
+# identity. These helpers build genuinely-encrypted packets so we can prove the
+# repeater consumes only when a local identity actually decrypts, and otherwise
+# forwards.
+# ---------------------------------------------------------------------------
+
+TXT_TYPE_CLI_DATA = 1  # no delivery ACK -> deterministic (no background ACK task)
+
+
+def _distinct_identities(n=3):
+    """Return `n` identities whose public keys start with distinct bytes."""
+    ids, seen = [], set()
+    while len(ids) < n:
+        idn = LocalIdentity()
+        first = idn.get_public_key()[0]
+        if first in seen:
+            continue
+        seen.add(first)
+        ids.append(idn)
+    return ids
+
+
+class _SendDest:
+    """Minimal contact object accepted by PacketBuilder as a send destination."""
+
+    def __init__(self, pubkey: bytes):
+        self.public_key = pubkey.hex()
+        self.out_path = []
+        self.out_path_len = -1
+
+
+def _force_dest_hash(packet, hash_byte: int):
+    """Rewrite the on-air one-byte dest hash to simulate a prefix collision."""
+    packet.payload = bytearray(packet.payload)
+    packet.payload[0] = hash_byte
+    return packet
+
+
+def _acl_with_client(sender, local_identity=None):
+    """ACL holding `sender` as an admin client.
+
+    When `local_identity` is given, the real shared secret is precomputed —
+    the protocol-request handler reads `client.shared_secret` directly.
+    """
+    acl = ACL()
+    sender_pub = sender.get_public_key()
+    client = ClientInfo(Identity(sender_pub), permissions=PERM_ACL_ADMIN)
+    if local_identity is not None:
+        client.shared_secret = Identity(sender_pub).calc_shared_secret(
+            local_identity.get_private_key()
+        )
+    acl.clients[sender_pub] = client
+    return acl
+
 
 class _FakeId:
     def __init__(self, pubkey: bytes):
@@ -169,7 +232,7 @@ async def test_protocol_request_process_routes_and_marks_no_retransmit():
     response_packet = object()
 
     async def _core_handler(_packet):
-        return response_packet
+        return HandlerResult.consumed(response_packet)
 
     helper.handlers[dest] = {"handler": _core_handler}
     pkt = _ReqPacket(payload=bytes([dest, 0x01, 0x02]))
@@ -182,6 +245,50 @@ async def test_protocol_request_process_routes_and_marks_no_retransmit():
     injector.assert_awaited_once_with(response_packet, wait_for_ack=False)
 
 
+@pytest.mark.asyncio
+async def test_protocol_request_forwards_on_dest_hash_collision():
+    """A REQ whose dest prefix collides with ours but does not decrypt for a
+    local client must not be consumed, so the engine can still forward it."""
+    injector = AsyncMock(return_value=True)
+    helper = ProtocolRequestHelper(identity_manager=MagicMock(), packet_injector=injector)
+
+    dest = 0x42
+
+    async def _core_handler(_packet):
+        # MAC failed for every same-hash candidate -> not for us.
+        return HandlerResult.not_for_us()
+
+    helper.handlers[dest] = {"handler": _core_handler}
+    pkt = _ReqPacket(payload=bytes([dest, 0x01, 0x02]))
+
+    handled = await helper.process_request_packet(pkt)
+
+    assert handled is False
+    pkt.mark_do_not_retransmit.assert_not_called()
+    injector.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_protocol_request_consumes_authenticated_without_response():
+    """A REQ that authenticates for us but yields no reply is still consumed."""
+    injector = AsyncMock(return_value=True)
+    helper = ProtocolRequestHelper(identity_manager=MagicMock(), packet_injector=injector)
+
+    dest = 0x42
+
+    async def _core_handler(_packet):
+        return HandlerResult.consumed()
+
+    helper.handlers[dest] = {"handler": _core_handler}
+    pkt = _ReqPacket(payload=bytes([dest, 0x01, 0x02]))
+
+    handled = await helper.process_request_packet(pkt)
+
+    assert handled is True
+    pkt.mark_do_not_retransmit.assert_called_once()
+    injector.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_protocol_request_process_exception_returns_false():
     helper = ProtocolRequestHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
@@ -330,7 +437,8 @@ async def test_text_helper_process_text_packet_routes_or_forwards():
     pkt_unknown = SimpleNamespace(payload=bytearray([0x55, 0x66]))
     assert await helper.process_text_packet(pkt_unknown) is False
 
-    h = AsyncMock()
+    # Handler decrypts successfully: consume and stop forwarding.
+    h = AsyncMock(return_value=HandlerResult.consumed())
     helper.handlers[0x10] = {"handler": h, "name": "id-a", "type": "repeater"}
     helper._on_message_received = AsyncMock()
     pkt = SimpleNamespace(payload=bytearray([0x10, 0x66]), mark_do_not_retransmit=MagicMock())
@@ -343,6 +451,26 @@ async def test_text_helper_process_text_packet_routes_or_forwards():
     pkt.mark_do_not_retransmit.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_text_helper_process_text_packet_hash_collision_forwards():
+    """dest hash matches a local identity but decryption fails (collision): the
+    packet is not ours, so it must NOT be consumed and must be left to forward (#353)."""
+    helper = TextHelper(identity_manager=MagicMock(), acl_dict={})
+
+    # Handler could not decrypt for this identity.
+    h = AsyncMock(return_value=HandlerResult.not_for_us())
+    helper.handlers[0x10] = {"handler": h, "name": "id-a", "type": "repeater"}
+    helper._on_message_received = AsyncMock()
+    pkt = SimpleNamespace(payload=bytearray([0x10, 0x66]), mark_do_not_retransmit=MagicMock())
+
+    handled = await helper.process_text_packet(pkt)
+
+    assert handled is False
+    h.assert_awaited_once_with(pkt)
+    helper._on_message_received.assert_not_awaited()
+    pkt.mark_do_not_retransmit.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_text_helper_send_packet_success_and_failures():
     injector = AsyncMock(side_effect=[True, RuntimeError("fail")])
@@ -438,3 +566,88 @@ async def test_text_helper_send_cli_reply_uses_direct_path_from_client():
     assert bytes(reply_packet.path) == b"\xaa\xbb"
     assert reply_packet.path_len == 2
     helper._send_packet.assert_awaited_once_with(reply_packet, wait_for_ack=False)
+
+
+# ---------------------------------------------------------------------------
+# Real-crypto collision integration tests (BUG-053).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_text_helper_real_crypto_consume_vs_collision_forward():
+    """A real, correctly-encrypted DM to a local room-server identity is consumed;
+    a DM encrypted for a remote node that merely collides on the one-byte dest
+    hash fails to decrypt and is left for the forwarding engine."""
+    local, sender, remote = _distinct_identities()
+    local_hash = local.get_public_key()[0]
+
+    helper = TextHelper(
+        identity_manager=MagicMock(),
+        acl_dict={local_hash: _acl_with_client(sender)},
+        packet_injector=AsyncMock(),
+    )
+    helper.register_identity("room-a", local, identity_type="room_server")
+    helper._on_message_received = AsyncMock()
+
+    # Genuine CLI_DATA DM to the local identity from a known sender.
+    genuine, _ = PacketBuilder.create_text_message(
+        _SendDest(local.get_public_key()),
+        sender,
+        "status",
+        message_type="flood",
+        txt_type=TXT_TYPE_CLI_DATA,
+    )
+    assert await helper.process_text_packet(genuine) is True
+    assert genuine.is_marked_do_not_retransmit()
+    helper._on_message_received.assert_awaited_once()
+
+    # DM encrypted for a REMOTE node whose one-byte dest hash collides with ours.
+    helper._on_message_received.reset_mock()
+    collision, _ = PacketBuilder.create_text_message(
+        _SendDest(remote.get_public_key()),
+        sender,
+        "not yours",
+        message_type="flood",
+        txt_type=TXT_TYPE_CLI_DATA,
+    )
+    _force_dest_hash(collision, local_hash)
+    assert await helper.process_text_packet(collision) is False
+    assert not collision.is_marked_do_not_retransmit()
+    helper._on_message_received.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_protocol_request_real_crypto_consume_vs_collision_forward():
+    """A real, correctly-encrypted REQ to a local identity is consumed and
+    answered; a REQ encrypted for a colliding remote node is left to forward."""
+    from openhop_core.node.handlers.protocol_request import REQ_TYPE_GET_STATUS
+
+    local, sender, remote = _distinct_identities()
+    local_hash = local.get_public_key()[0]
+
+    injector = AsyncMock(return_value=True)
+    helper = ProtocolRequestHelper(
+        identity_manager=MagicMock(),
+        packet_injector=injector,
+        acl_dict={local_hash: _acl_with_client(sender, local_identity=local)},
+    )
+    helper.register_identity("rep", local, identity_type="repeater")
+
+    genuine, _ = PacketBuilder.create_protocol_request(
+        _SendDest(local.get_public_key()), sender, REQ_TYPE_GET_STATUS
+    )
+    with patch(
+        "repeater.handler_helpers.protocol_request.asyncio.sleep", new_callable=AsyncMock
+    ):
+        assert await helper.process_request_packet(genuine) is True
+    assert genuine.is_marked_do_not_retransmit()
+    injector.assert_awaited()  # a response was transmitted
+
+    injector.reset_mock()
+    collision, _ = PacketBuilder.create_protocol_request(
+        _SendDest(remote.get_public_key()), sender, REQ_TYPE_GET_STATUS
+    )
+    _force_dest_hash(collision, local_hash)
+    assert await helper.process_request_packet(collision) is False
+    assert not collision.is_marked_do_not_retransmit()
+    injector.assert_not_awaited()

tests/test_handler_helpers_trace_discovery_login.py

@@ -4,13 +4,46 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+
+from openhop_core.node.handlers.result import HandlerResult
+from openhop_core.protocol import LocalIdentity
 from openhop_core.protocol.constants import PAYLOAD_TYPE_ANON_REQ, ROUTE_TYPE_DIRECT
+from openhop_core.protocol.packet_builder import PacketBuilder
 
 from repeater.handler_helpers.discovery import DiscoveryHelper
 from repeater.handler_helpers.login import LoginHelper
 from repeater.handler_helpers.trace import TraceHelper
 
 
+def _distinct_identities(n=3):
+    """Return `n` identities whose public keys start with distinct bytes."""
+    ids, seen = [], set()
+    while len(ids) < n:
+        idn = LocalIdentity()
+        first = idn.get_public_key()[0]
+        if first in seen:
+            continue
+        seen.add(first)
+        ids.append(idn)
+    return ids
+
+
+class _SendDest:
+    """Minimal contact object accepted by PacketBuilder as a send destination."""
+
+    def __init__(self, pubkey: bytes):
+        self.public_key = pubkey.hex()
+        self.out_path = []
+        self.out_path_len = -1
+
+
+def _force_dest_hash(packet, hash_byte: int):
+    """Rewrite the on-air one-byte dest hash to simulate a prefix collision."""
+    packet.payload = bytearray(packet.payload)
+    packet.payload[0] = hash_byte
+    return packet
+
+
 class DummyPacket:
     def __init__(
         self, *, route=ROUTE_TYPE_DIRECT, path=b"", payload=b"\x01\x02", snr=2.5, rssi=-70
@@ -489,7 +522,8 @@ def test_owner_and_features_callbacks_from_config():
 @pytest.mark.asyncio
 async def test_login_process_packet_routes_to_registered_handler_and_marks_no_retransmit():
     helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
-    login_handler = AsyncMock()
+    # Handler decrypts successfully: consume and stop forwarding.
+    login_handler = AsyncMock(return_value=HandlerResult.consumed())
     helper.handlers[0x62] = login_handler
 
     packet = SimpleNamespace(
@@ -505,6 +539,28 @@ async def test_login_process_packet_routes_to_registered_handler_and_marks_no_re
     packet.mark_do_not_retransmit.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_login_process_packet_hash_collision_forwards():
+    """dest hash matches a local identity but decryption fails (collision): the
+    ANON_REQ is not ours, so it must NOT be consumed and must forward (#353)."""
+    helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
+    # Handler could not decrypt for this identity.
+    login_handler = AsyncMock(return_value=HandlerResult.not_for_us())
+    helper.handlers[0x62] = login_handler
+
+    packet = SimpleNamespace(
+        payload=bytearray([0x62, 0xAA]),
+        get_payload_type=lambda: 0x01,
+        mark_do_not_retransmit=MagicMock(),
+    )
+
+    handled = await helper.process_login_packet(packet)
+
+    assert handled is False
+    login_handler.assert_awaited_once_with(packet)
+    packet.mark_do_not_retransmit.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_login_process_packet_unknown_and_short_payload_are_not_handled():
     helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
@@ -545,3 +601,32 @@ def test_login_acl_access_and_client_listing():
 
     all_clients = helper.list_authenticated_clients()
     assert {c["id"] for c in all_clients} == {"a1", "b1", "b2"}
+
+
+@pytest.mark.asyncio
+async def test_login_helper_real_crypto_consume_vs_collision_forward():
+    """A real ANON_REQ login to a local room-server identity is consumed (a wrong
+    password still decrypts, so it is ours to reject); an ANON_REQ encrypted for a
+    remote node that collides on the one-byte dest hash is left for forwarding."""
+    local, sender, remote = _distinct_identities()
+    local_hash = local.get_public_key()[0]
+
+    helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
+    helper.register_identity(
+        "room-a",
+        local,
+        identity_type="room_server",
+        config={"settings": {"admin_password": "secret"}},
+    )
+    assert local_hash in helper.handlers  # registration succeeded
+
+    # Genuine login (wrong password still decrypts -> ours to reject, not forward).
+    genuine = PacketBuilder.create_login_packet(_SendDest(local.get_public_key()), sender, "nope")
+    assert await helper.process_login_packet(genuine) is True
+    assert genuine.is_marked_do_not_retransmit()
+
+    # Login encrypted for a remote node whose dest hash collides with ours.
+    collision = PacketBuilder.create_login_packet(_SendDest(remote.get_public_key()), sender, "nope")
+    _force_dest_hash(collision, local_hash)
+    assert await helper.process_login_packet(collision) is False
+    assert not collision.is_marked_do_not_retransmit()

tests/test_packet_router.py

@@ -28,6 +28,7 @@ from openhop_core.node.handlers.login_server import LoginServerHandler
 from openhop_core.node.handlers.path import PathHandler
 from openhop_core.node.handlers.protocol_request import ProtocolRequestHandler
 from openhop_core.node.handlers.protocol_response import ProtocolResponseHandler
+from openhop_core.node.handlers.result import HandlerResult
 from openhop_core.node.handlers.text import TextMessageHandler
 from openhop_core.node.handlers.trace import TraceHandler
 from openhop_core.protocol.constants import ROUTE_TYPE_DIRECT
@@ -833,6 +834,84 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         b1.process_received_packet.assert_awaited_once()
         daemon.repeater_handler.assert_not_awaited()
 
+    async def test_route_protocol_request_companion_collision_forwards(self):
+        """A REQ whose dest prefix matches a companion but does not decrypt for
+        it must be forwarded, not swallowed."""
+        daemon = _make_daemon()
+        b1 = _make_bridge()
+        # Collision: the packet is not really for this companion.
+        b1.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x01: b1}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(ProtocolRequestHandler.payload_type())
+        pkt.header = 0x00  # transport-flood: not a direct final hop
+        pkt.path = bytearray([0xAA])
+        pkt.payload = bytes([0x01, 0xBB])
+        await router._route_packet(pkt)
+        b1.process_received_packet.assert_awaited_once()
+        # Not consumed -> handed to the forwarding engine.
+        daemon.repeater_handler.assert_awaited_once()
+
+    async def test_route_text_companion_collision_forwards(self):
+        """A TXT_MSG whose dest prefix matches a companion but does not decrypt
+        for it must be forwarded, not swallowed."""
+        daemon = _make_daemon()
+        b1 = _make_bridge()
+        b1.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x01: b1}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        pkt.header = 0x00
+        pkt.path = bytearray([0xAA])
+        pkt.payload = bytes([0x01, 0xBB])
+        await router._route_packet(pkt)
+        b1.process_received_packet.assert_awaited_once()
+        daemon.repeater_handler.assert_awaited_once()
+
+    async def test_route_text_tries_all_candidates_room_server_wins(self):
+        """A companion and a room-server text identity share a hash. The router
+        must try both; when the companion fails to decrypt but the room server
+        authenticates, the packet is consumed (not forwarded)."""
+        daemon = _make_daemon()
+        companion = _make_bridge()
+        companion.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x42: companion}
+        daemon.text_helper = MagicMock()
+        daemon.text_helper.handlers = {0x42: {"name": "room-a"}}
+        daemon.text_helper.process_text_packet = AsyncMock(return_value=True)
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        pkt.header = 0x00
+        pkt.path = bytearray([0xAA])
+        pkt.payload = bytes([0x42, 0xBB])
+        await router._route_packet(pkt)
+        # Both local candidates were tried; the room server consumed it.
+        companion.process_received_packet.assert_awaited_once()
+        daemon.text_helper.process_text_packet.assert_awaited_once()
+        daemon.repeater_handler.assert_not_awaited()
+
+    async def test_route_text_all_local_candidates_fail_forwards(self):
+        """When neither the companion nor the room-server identity at the hash
+        authenticates, the packet is left for the forwarding engine."""
+        daemon = _make_daemon()
+        companion = _make_bridge()
+        companion.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x42: companion}
+        daemon.text_helper = MagicMock()
+        daemon.text_helper.handlers = {0x42: {"name": "room-a"}}
+        daemon.text_helper.process_text_packet = AsyncMock(return_value=False)
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        pkt.header = 0x00
+        pkt.path = bytearray([0xAA])
+        pkt.payload = bytes([0x42, 0xBB])
+        await router._route_packet(pkt)
+        companion.process_received_packet.assert_awaited_once()
+        daemon.text_helper.process_text_packet.assert_awaited_once()
+        daemon.repeater_handler.assert_awaited_once()
+
     async def test_route_group_text_delivers_and_forwards(self):
         daemon = _make_daemon()
         b1 = _make_bridge()