fix(router): consume PATH and RESPONSE only after MAC authentication

Extend the authenticated-ownership model to the PATH and RESPONSE
routing branches so a packet is consumed (do-not-retransmit) only when
a local identity MAC-verifies it. Prefix-only collisions and forged
traffic stay eligible for the forwarding engine.

- PathHelper marks do-not-retransmit and reports authenticated only
  after a successful MAC decrypt with a valid path envelope; invalid or
  truncated envelopes remain forwardable.
- PacketRouter aggregates authenticated results across the path helper
  and companion bridges for PATH and RESPONSE, skipping the engine only
  on authenticated ownership while preserving empty-path DIRECT release
  hygiene.

repeater/engine.py

@@ -1080,8 +1080,7 @@ class RepeaterHandler(BaseHandler):
         random_mult = secrets.randbelow(5001) / 1000.0
         delay_s = (base_delay_ms * random_mult) / 1000.0
 
-        # Apply score-based delay adjustment ONLY if delay >= 50ms threshold
-        # (matching C++ reactive behavior in Dispatcher::calcRxDelay)
+        # OpenHop's optional score gate is applied only after the 50 ms threshold.
         if delay_s >= 0.05 and self.use_score_for_tx:
             score = self.calculate_packet_score(snr, packet_len)
             # Higher score = shorter delay: max(0.2, 1.0 - score)

repeater/handler_helpers/path.py

@@ -32,6 +32,7 @@ class PathHelper:
         from openhop_core.protocol.crypto import CryptoUtils
         from openhop_core.protocol.packet_utils import PathUtils
 
+        authenticated = False
         try:
             if len(packet.payload) < 2:
                 return False
@@ -83,28 +84,38 @@ class PathHelper:
                 logger.debug("Decrypted PATH data too short")
                 return False
             path_len_byte = decrypted[0]
-            if PathUtils.is_valid_path_len(path_len_byte):
-                path_byte_len = PathUtils.get_path_byte_len(path_len_byte)
-                path_hops = PathUtils.get_path_hash_count(path_len_byte)
-            else:
-                # Legacy fallback for malformed/old packets: treat first byte as raw path bytes.
-                path_byte_len = path_len_byte
-                path_hops = path_byte_len
+            if not PathUtils.is_valid_path_len(path_len_byte):
+                logger.debug(f"Invalid PATH length encoding: 0x{path_len_byte:02X}")
+                return False
+            path_byte_len = PathUtils.get_path_byte_len(path_len_byte)
+            path_hops = PathUtils.get_path_hash_count(path_len_byte)
 
-            if len(decrypted) < 1 + path_byte_len:
+            # Firmware (Mesh.cpp) reads the extra_type byte unconditionally after
+            # the path and consumes (markDoNotRetransmit) even when it is absent.
+            # We are stricter: a MAC-verified PATH with a valid path_len but no
+            # extra_type byte is treated as not-for-us and left forwardable, since
+            # it can only be a malformed packet. This is the safer divergence.
+            extra_start = 1 + path_byte_len
+            if len(decrypted) < extra_start + 1:
                 logger.debug(
-                    f"PATH data truncated: need {1 + path_byte_len} bytes, got {len(decrypted)}"
+                    f"PATH data truncated: need {extra_start + 1} bytes, got {len(decrypted)}"
                 )
                 return False
 
             path_data = decrypted[1 : 1 + path_byte_len]
 
+            # The destination hash selected a local identity and the MAC
+            # verified with one of its clients. From here on, consume the
+            # packet even if a bookkeeping or notification side effect fails.
+            authenticated = True
+            packet.mark_do_not_retransmit()
+
             # Update client's out_path (same as C++ memcpy); out_path_len keeps
             # the encoded byte so direct sends put it on the wire as-is.
             client.out_path = bytearray(path_data)
-            client.out_path_len = (
-                path_len_byte if PathUtils.is_valid_path_len(path_len_byte) else path_byte_len
-            )
+            # path_len_byte is guaranteed valid here (invalid encodings returned
+            # above), so store it verbatim for direct sends.
+            client.out_path_len = path_len_byte
             client.last_activity = int(time.time())
 
             logger.info(
@@ -115,7 +126,6 @@ class PathHelper:
 
             # Handle bundled ACK in PATH extra section.
             ack_crc = None
-            extra_start = 1 + path_byte_len
             if len(decrypted) > extra_start:
                 extra_type = decrypted[extra_start] & 0x0F
                 extra_payload = decrypted[extra_start + 1 :]
@@ -127,9 +137,8 @@ class PathHelper:
 
             if ack_crc is not None:
                 await self._register_ack_crc(ack_crc)
-            # Don't mark as do_not_retransmit - let it forward normally
-            return False
+            return authenticated
 
         except Exception as e:
             logger.error(f"Error processing PATH packet: {e}", exc_info=True)
-            return False
+            return authenticated

repeater/packet_router.py

@@ -563,27 +563,30 @@ class PacketRouter:
         elif payload_type == PathHandler.payload_type():
             # Always let PathHelper inspect/decrypt PATH first so out_path and bundled ACK state
             # are updated even when companion routing fan-out also happens for this packet.
+            consumed = False
             if self.daemon.path_helper:
                 try:
-                    await self.daemon.path_helper.process_path_packet(packet)
+                    consumed = (
+                        await self.daemon.path_helper.process_path_packet(packet)
+                    ) is True
                 except Exception as e:
                     logger.debug(f"Path helper processing error: {e}")
-            # The unconditional call above already covers PATH addressed to a
-            # local server identity (room server/repeater), so its out_path and
-            # any embedded ACK are handled before bridge delivery — the
-            # all-bridges branch below no longer swallows path returns for them.
+            # The helper/bridge results decide ownership: a direct middle hop
+            # that cannot authenticate remains eligible for engine forwarding,
+            # while a local MAC-authenticated PATH is consumed below.
             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:
                 if self._should_deliver_path_to_companions(packet):
-                    await companion_bridges[dest_hash].process_received_packet(packet)
-                # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
+                    result = await companion_bridges[dest_hash].process_received_packet(packet)
+                    consumed = consumed or getattr(result, "authenticated", False) is True
             elif companion_bridges and self._should_deliver_path_to_companions(packet):
                 # Dest not in bridges: path-return with ephemeral dest (e.g. multi-hop login).
                 # Deliver to all bridges; each will try to decrypt and ignore if not relevant.
                 for bridge in companion_bridges.values():
                     try:
-                        await bridge.process_received_packet(packet)
+                        result = await bridge.process_received_packet(packet)
+                        consumed = consumed or getattr(result, "authenticated", False) is True
                     except Exception as e:
                         logger.debug(f"Companion bridge PATH error: {e}")
                 logger.debug(
@@ -591,20 +594,25 @@ class PacketRouter:
                     dest_hash or 0,
                     len(companion_bridges),
                 )
-                # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
+            if consumed:
+                # A local MAC-authenticated PATH belongs to this node. Do not
+                # let the forwarding engine retransmit it, but retain it for UI.
+                processed_by_injection = True
+                self._record_for_ui(packet, metadata)
 
         elif payload_type == LoginResponseHandler.payload_type():
             # PAYLOAD_TYPE_RESPONSE (0x01): payload is dest_hash(1)+src_hash(1)+encrypted.
             # Deliver to the bridge that is the destination, or to all bridges when the
             # response is addressed to this repeater (path-based reply: firmware sends
             # to first hop instead of original requester).
-            # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
+            consumed = False
             dest_hash = packet.payload[0] if packet.payload and len(packet.payload) >= 1 else None
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
             local_hash = getattr(self.daemon, "local_hash", None)
             if dest_hash is not None and dest_hash in companion_bridges:
                 try:
-                    await companion_bridges[dest_hash].process_received_packet(packet)
+                    result = await companion_bridges[dest_hash].process_received_packet(packet)
+                    consumed = consumed or getattr(result, "authenticated", False) is True
                     logger.info(
                         "RESPONSE dest=0x%02x delivered to companion bridge",
                         dest_hash,
@@ -615,7 +623,8 @@ class PacketRouter:
                 # Response addressed to this repeater (e.g. path-based reply to first hop)
                 for bridge in companion_bridges.values():
                     try:
-                        await bridge.process_received_packet(packet)
+                        result = await bridge.process_received_packet(packet)
+                        consumed = consumed or getattr(result, "authenticated", False) is True
                     except Exception as e:
                         logger.debug(f"Companion bridge RESPONSE error: {e}")
                 logger.info(
@@ -629,7 +638,8 @@ class PacketRouter:
                 # not relevant (firmware-like behavior, works with multiple companion bridges).
                 for bridge in companion_bridges.values():
                     try:
-                        await bridge.process_received_packet(packet)
+                        result = await bridge.process_received_packet(packet)
+                        consumed = consumed or getattr(result, "authenticated", False) is True
                     except Exception as e:
                         logger.debug(f"Companion bridge RESPONSE error: {e}")
                 logger.debug(
@@ -637,8 +647,12 @@ class PacketRouter:
                     dest_hash or 0,
                     len(companion_bridges),
                 )
-            if companion_bridges and _is_direct_final_hop(packet):
-                # DIRECT with empty path: we're the final hop; don't pass to engine (it would drop with "Direct: no path")
+            if consumed:
+                processed_by_injection = True
+                self._record_for_ui(packet, metadata)
+            elif companion_bridges and _is_direct_final_hop(packet):
+                # DIRECT with empty path is engine release hygiene: there is
+                # no next hop, even when no local identity authenticated it.
                 processed_by_injection = True
                 self._record_for_ui(packet, metadata)
 
@@ -683,8 +697,8 @@ class PacketRouter:
             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.
+                    # OpenHop release hygiene: an empty-path DIRECT has no next hop,
+                    # so consume after offering it to bridges even without MAC ownership.
                     for bridge in companion_bridges.values():
                         try:
                             await bridge.process_received_packet(packet)

tests/test_handler_helpers_path_protocol_text.py

@@ -101,6 +101,7 @@ class _FakeACL:
 class _PathPacket:
     def __init__(self, payload: bytes):
         self.payload = bytearray(payload)
+        self.mark_do_not_retransmit = MagicMock()
 
 
 class _ReqPacket:
@@ -124,7 +125,8 @@ async def test_path_helper_updates_client_out_path_on_valid_decrypt():
     ):
         handled = await helper.process_path_packet(packet)
 
-    assert handled is False
+    assert handled is True
+    packet.mark_do_not_retransmit.assert_called_once_with()
     assert client.out_path_len == 2
     assert bytes(client.out_path) == b"\x99\x88"
     assert isinstance(client.last_activity, int)
@@ -217,6 +219,24 @@ async def test_path_helper_returns_false_for_non_matching_or_invalid_inputs():
     with patch("openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt", return_value=None):
         assert await helper.process_path_packet(_PathPacket(payload=b"\x11\x22\xaa\xbb")) is False
 
+    # A valid MAC with an invalid or truncated PATH envelope is not local
+    # ownership; the forwarding engine must remain eligible to handle it.
+    with patch(
+        "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt",
+        return_value=b"\x7f\x99\x88\x01",
+    ):
+        invalid_packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc")
+        assert await helper.process_path_packet(invalid_packet) is False
+        invalid_packet.mark_do_not_retransmit.assert_not_called()
+
+    with patch(
+        "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt",
+        return_value=b"\x02\x99",
+    ):
+        truncated_packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc")
+        assert await helper.process_path_packet(truncated_packet) is False
+        truncated_packet.mark_do_not_retransmit.assert_not_called()
+
 
 @pytest.mark.asyncio
 async def test_protocol_request_process_routes_and_marks_no_retransmit():

tests/test_packet_router.py

@@ -49,6 +49,7 @@ def _make_daemon():
     """Minimal daemon that satisfies PacketRouter without touching hardware."""
     daemon = MagicMock()
     daemon.repeater_handler = AsyncMock(return_value=True)
+    daemon.repeater_handler.record_packet_only = MagicMock()
     daemon.trace_helper = None
     daemon.discovery_helper = None
     daemon.advert_helper = None
@@ -774,6 +775,37 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         await router._route_packet(pkt)
         daemon.path_helper.process_path_packet.assert_awaited_once_with(pkt)
 
+    async def test_authenticated_flood_path_skips_engine(self):
+        daemon = _make_daemon()
+        daemon.path_helper = MagicMock()
+        daemon.path_helper.process_path_packet = AsyncMock(return_value=True)
+        bridge = _make_bridge()
+        bridge.process_received_packet = AsyncMock(return_value=HandlerResult.consumed())
+        daemon.companion_bridges = {0x01: bridge}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(PathHandler.payload_type())
+        pkt.payload = bytes([0x01, 0x22])
+
+        await router._route_packet(pkt)
+
+        bridge.process_received_packet.assert_awaited_once_with(pkt)
+        daemon.repeater_handler.assert_not_awaited()
+
+    async def test_unauthenticated_flood_path_reaches_engine(self):
+        daemon = _make_daemon()
+        daemon.path_helper = MagicMock()
+        daemon.path_helper.process_path_packet = AsyncMock(return_value=False)
+        bridge = _make_bridge()
+        bridge.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x01: bridge}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(PathHandler.payload_type())
+        pkt.payload = bytes([0x01, 0x22])
+
+        await router._route_packet(pkt)
+
+        daemon.repeater_handler.assert_awaited_once()
+
     async def test_route_path_dedupes_companion_delivery(self):
         daemon = _make_daemon()
         bridge = _make_bridge()
@@ -802,6 +834,33 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         b1.process_received_packet.assert_awaited_once()
         daemon.repeater_handler.assert_not_awaited()
 
+    async def test_authenticated_flood_response_skips_engine(self):
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        bridge.process_received_packet = AsyncMock(return_value=HandlerResult.consumed())
+        daemon.companion_bridges = {0x01: bridge}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(LoginResponseHandler.payload_type())
+        pkt.payload = bytes([0x01, 0x22])
+
+        await router._route_packet(pkt)
+
+        bridge.process_received_packet.assert_awaited_once_with(pkt)
+        daemon.repeater_handler.assert_not_awaited()
+
+    async def test_unauthenticated_flood_response_reaches_engine(self):
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        bridge.process_received_packet = AsyncMock(return_value=HandlerResult.not_for_us())
+        daemon.companion_bridges = {0x01: bridge}
+        router = PacketRouter(daemon)
+        pkt = _make_packet(LoginResponseHandler.payload_type())
+        pkt.payload = bytes([0x01, 0x22])
+
+        await router._route_packet(pkt)
+
+        daemon.repeater_handler.assert_awaited_once()
+
     async def test_route_protocol_response_final_hop_skips_engine(self):
         daemon = _make_daemon()
         b1 = _make_bridge()