From a208fc3763778a09e0cf4b48e12a1da2730a2b4e Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 15 Jul 2026 16:29:20 -0700 Subject: [PATCH] fix(repeater): relay multipart ACKs as regenerated direct ACKs Match MeshCore's forwardMultipartDirect: a multipart packet is never relayed as an ordinary directed packet. At an intermediate direct hop a multipart ACK is re-emitted as a plain DIRECT ACK with the wrapper byte stripped and this node removed from the path, deduped on the regenerated ACK form and spaced by (remaining + 1) * 300ms. Flood routing, final hops, and non-ACK embedded types are dropped rather than forwarded. Coverage includes literal firmware-format wire vectors for one- and two-byte hash widths and transport-direct input, end-to-end receive-to- transmit relaying, and the drop cases. --- repeater/engine.py | 88 ++++++++++++++ repeater/packet_router.py | 1 + tests/test_engine.py | 235 ++++++++++++++++++++++++++++++++++++-- 3 files changed, 313 insertions(+), 11 deletions(-) diff --git a/repeater/engine.py b/repeater/engine.py index 3b96997..19c2e47 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -10,10 +10,13 @@ from openhop_core.node.handlers.base import BaseHandler from openhop_core.protocol import Packet from openhop_core.protocol.constants import ( MAX_PATH_SIZE, + PAYLOAD_TYPE_ACK, PAYLOAD_TYPE_ADVERT, PAYLOAD_TYPE_ANON_REQ, + PAYLOAD_TYPE_MULTIPART, PAYLOAD_TYPE_TRACE, PH_ROUTE_MASK, + PH_TYPE_SHIFT, ROUTE_TYPE_DIRECT, ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_DIRECT, @@ -1036,6 +1039,84 @@ class RepeaterHandler(BaseHandler): return packet + # Multipart ACKs are spaced ~300ms apart per remaining fragment, matching + # MeshCore's forwardMultipartDirect (`(remaining + 1) * 300`). + MULTIPART_ACK_SPACING_MS = 300 + + def forward_multipart_direct( + self, packet: Packet, packet_hash: Optional[str] = None + ) -> Optional[Tuple[Packet, float]]: + """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does. + + MeshCore never relays the multipart wrapper as an ordinary directed + packet. Only a multipart-ACK at an intermediate direct hop is relayed, + and it is relayed by regenerating the embedded ACK as a plain DIRECT ACK + (the wrapper byte stripped) rather than repeating the wrapper. Every + other multipart case — flood routing, a final hop, or a non-ACK embedded + type — is dropped. Returns ``(ack_packet, delay_seconds)`` when a + regenerated ACK should be sent, otherwise ``None``. + + INVARIANT: purely synchronous, mirroring flood_forward / direct_forward — + the is_duplicate + mark_seen pair must stay atomic within the event loop. + """ + # MeshCore only forwards multipart on the direct-route branch; the flood + # switch case updates ACK state locally but never re-routes it. + if not packet.is_route_direct(): + packet.drop_reason = "Multipart: not direct-routed" + return None + + valid, reason = self.validate_packet(packet) + if not valid: + packet.drop_reason = reason + return None + + if packet.is_marked_do_not_retransmit(): + packet.drop_reason = packet.drop_reason or "Marked do not retransmit" + return None + + payload = packet.payload or b"" + remaining = payload[0] >> 4 + embedded_type = payload[0] & 0x0F + # Only multipart-ACKs are relayed; MeshCore leaves other types for future + # use and does not forward them. + if embedded_type != PAYLOAD_TYPE_ACK or len(payload) < 5: + packet.drop_reason = "Multipart: unsupported embedded type" + return None + + hash_size = packet.get_path_hash_size() + hop_count = packet.get_path_hash_count() + + # We must be the next hop (a final-hop multipart has no remaining path and + # is handled locally, not forwarded). + if not packet.path or len(packet.path) < hash_size: + packet.drop_reason = "Direct: no path" + return None + if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: + packet.drop_reason = "Direct: not for us" + return None + + # Rebuild the received wrapper as the plain DIRECT ACK MeshCore would emit: + # drop the multipart header byte and force the ACK payload/route type. The + # path is left intact for the seen-check, then this node is removed from it, + # matching forwardMultipartDirect's hasSeen()/removeSelfFromPath() order. + packet.payload = bytearray(payload[1:]) + packet.payload_len = len(packet.payload) + packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT + + # Dedupe on the regenerated ACK form (full path) before removing ourselves, + # so a repeated multipart ACK — or an equivalent plain ACK — is not relayed + # twice. The pre-computed hash belongs to the original wrapper, so recompute. + if self.is_duplicate(packet): + packet.drop_reason = "Duplicate" + return None + self.mark_seen(packet) + + packet.path = bytearray(packet.path[hash_size:]) + packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) + + delay_s = ((remaining + 1) * self.MULTIPART_ACK_SPACING_MS) / 1000.0 + return packet, delay_s + @staticmethod def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float: @@ -1120,6 +1201,13 @@ class RepeaterHandler(BaseHandler): """ route_type = packet.header & PH_ROUTE_MASK + # MeshCore routes multipart traffic through its own branch, never the + # generic flood/direct path — it regenerates embedded ACKs and drops + # everything else. Its delay is fixed by the fragment count, so return + # the (packet, delay) pair directly instead of _calculate_tx_delay. + if packet.get_payload_type() == PAYLOAD_TYPE_MULTIPART: + return self.forward_multipart_direct(packet, packet_hash=packet_hash) + if route_type == ROUTE_TYPE_FLOOD or route_type == ROUTE_TYPE_TRANSPORT_FLOOD: fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash) if fwd_pkt is None: diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 92b717d..8e3029f 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -48,6 +48,7 @@ _EXPECTED_DROP_REASON_PREFIXES = ( "Empty payload", "Path too long", "Invalid advert packet", + "Multipart", ) diff --git a/tests/test_engine.py b/tests/test_engine.py index 9743170..84f0ac0 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -15,6 +15,8 @@ import pytest from openhop_core.protocol import Packet, PacketBuilder from openhop_core.protocol.constants import ( MAX_PATH_SIZE, + PAYLOAD_TYPE_ACK, + PAYLOAD_TYPE_MULTIPART, PH_ROUTE_MASK, PH_TYPE_SHIFT, ROUTE_TYPE_DIRECT, @@ -30,6 +32,10 @@ from openhop_core.protocol.packet_utils import PathUtils LOCAL_HASH = 0xAB # repeater's own 1-byte path hash +# MULTIPART is not forwarded generically — it has its own MeshCore branch +# (forward_multipart_direct), covered by TestForwardMultipartDirect. +_GENERIC_FORWARD_PAYLOAD_TYPES = [pt for pt in range(16) if pt != PAYLOAD_TYPE_MULTIPART] + def _make_config(**overrides) -> dict: """Return a minimal valid config dict for RepeaterHandler.""" @@ -91,19 +97,26 @@ def _make_dispatcher(radio=None): return dispatcher -@pytest.fixture() -def handler(): - """Create a RepeaterHandler with mocked external dependencies.""" - config = _make_config() - dispatcher = _make_dispatcher() +def _make_handler_with_hash(local_hash_bytes: bytes): + """Create a RepeaterHandler whose local path hash is exactly local_hash_bytes.""" with ( patch("repeater.engine.StorageCollector"), patch("repeater.engine.RepeaterHandler._start_background_tasks"), ): from repeater.engine import RepeaterHandler - h = RepeaterHandler(config, dispatcher, LOCAL_HASH) - return h + return RepeaterHandler( + _make_config(), + _make_dispatcher(), + local_hash_bytes[0], + local_hash_bytes=local_hash_bytes, + ) + + +@pytest.fixture() +def handler(): + """Create a RepeaterHandler with mocked external dependencies.""" + return _make_handler_with_hash(bytes([LOCAL_HASH])) def _make_flood_packet( @@ -347,6 +360,176 @@ class TestDirectForward: assert pkt.path_len == len(pkt.path) == 2 +# =================================================================== +# 2b. forward_multipart_direct — MeshCore forwardMultipartDirect parity +# =================================================================== + + +def _make_multipart_ack_packet( + crc: bytes = b"\x4d\xab\xaf\x95", + remaining: int = 2, + path: bytes = None, + route: int = ROUTE_TYPE_DIRECT, + embedded_type: int = PAYLOAD_TYPE_ACK, +) -> Packet: + """Build a MULTIPART wrapper: [remaining<<4 | embedded_type] followed by a CRC.""" + if path is None: + path = bytes([LOCAL_HASH, 0xCC]) + pkt = Packet() + pkt.header = route | (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT) + pkt.payload = bytearray(bytes([(remaining << 4) | embedded_type]) + crc) + pkt.payload_len = len(pkt.payload) + pkt.path = bytearray(path) + pkt.path_len = len(path) + return pkt + + +class TestForwardMultipartDirect: + """forward_multipart_direct mirrors MeshCore: regenerate the embedded ACK, + drop every other multipart case.""" + + def test_intermediate_multipart_ack_regenerates_plain_direct_ack(self, handler): + pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC])) + result = handler.forward_multipart_direct(pkt) + assert result is not None + ack, delay = result + # Wrapper byte stripped, forced to a plain DIRECT ACK. + assert ack.get_payload_type() == PAYLOAD_TYPE_ACK + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(ack.payload) == b"\x4d\xab\xaf\x95" + # This node removed from the path. + assert list(ack.path) == [0xCC] + assert ack.path_len == PathUtils.encode_path_len(1, 1) + # (remaining + 1) * 300 ms fragment spacing. + assert delay == pytest.approx(0.9) + + def test_delay_scales_with_remaining_fragments(self, handler): + pkt = _make_multipart_ack_packet(remaining=0, path=bytes([LOCAL_HASH, 0xCC])) + _, delay = handler.forward_multipart_direct(pkt) + assert delay == pytest.approx(0.3) + + def test_transport_direct_multipart_ack_regenerates_plain_direct_ack(self, handler): + pkt = _make_multipart_ack_packet(crc=b"\x11\x22\x33\x44", route=ROUTE_TYPE_TRANSPORT_DIRECT) + ack, _ = handler.forward_multipart_direct(pkt) + # MeshCore forces ROUTE_TYPE_DIRECT on the regenerated ACK. + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(ack.payload) == b"\x11\x22\x33\x44" + + def test_flood_multipart_not_forwarded(self, handler): + pkt = _make_multipart_ack_packet(route=ROUTE_TYPE_FLOOD) + result = handler.forward_multipart_direct(pkt) + assert result is None + assert "not direct-routed" in pkt.drop_reason + + def test_final_hop_multipart_not_forwarded(self, handler): + pkt = _make_multipart_ack_packet(path=b"") + pkt.path_len = 0 + result = handler.forward_multipart_direct(pkt) + assert result is None + assert "no path" in pkt.drop_reason + + def test_wrong_next_hop_multipart_dropped(self, handler): + pkt = _make_multipart_ack_packet(path=bytes([0xFF, 0xCC])) + result = handler.forward_multipart_direct(pkt) + assert result is None + assert "not for us" in pkt.drop_reason + + def test_non_ack_embedded_type_not_forwarded(self, handler): + pkt = _make_multipart_ack_packet(embedded_type=0x02) # PAYLOAD_TYPE_TXT_MSG + result = handler.forward_multipart_direct(pkt) + assert result is None + assert "unsupported embedded type" in pkt.drop_reason + + def test_short_multipart_ack_not_forwarded(self, handler): + pkt = _make_multipart_ack_packet(crc=b"\x01\x02\x03") # only 4-byte payload + result = handler.forward_multipart_direct(pkt) + assert result is None + assert "unsupported embedded type" in pkt.drop_reason + + def test_duplicate_multipart_ack_dropped(self, handler): + handler.forward_multipart_direct(_make_multipart_ack_packet()) + pkt2 = _make_multipart_ack_packet() # same CRC + path → same regenerated ACK + result = handler.forward_multipart_direct(pkt2) + assert result is None + assert pkt2.drop_reason == "Duplicate" + + def test_dedup_key_is_regenerated_ack_before_pop(self, handler): + """The seen key is the plain-ACK form with the full path, matching + MeshCore's hasSeen()-before-removeSelfFromPath() order.""" + pkt = _make_multipart_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + expected = Packet() + expected.header = ROUTE_TYPE_DIRECT | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) + expected.payload = bytearray(b"\x4d\xab\xaf\x95") + expected.payload_len = 4 + expected.path = bytearray([LOCAL_HASH, 0xCC]) + expected.path_len = 2 + key = expected.calculate_packet_hash().hex().upper() + + handler.forward_multipart_direct(pkt) + assert key in handler.seen_packets + + def test_marked_do_not_retransmit_multipart_dropped(self, handler): + pkt = _make_multipart_ack_packet() + pkt.mark_do_not_retransmit() + result = handler.forward_multipart_direct(pkt) + assert result is None + + def test_process_packet_dispatches_multipart(self, handler): + pkt = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC])) + result = handler.process_packet(pkt, snr=5.0) + assert result is not None + ack, delay = result + assert ack.get_payload_type() == PAYLOAD_TYPE_ACK + assert delay == pytest.approx(0.6) + + +# Literal MeshCore-format frames (Packet::writeTo layout: header, optional 4-byte +# transport codes, encoded path_len, path bytes, payload). A multipart-ACK +# payload is `[remaining<<4 | PAYLOAD_TYPE_ACK]` + 4-byte CRC; forwardMultipartDirect +# strips that wrapper byte and re-emits a plain DIRECT ACK with this node removed +# from the path. These are independent wire fixtures — do NOT rebuild them through +# Packet.write_to()/PathUtils. remaining=2 in every case → (2+1)*300ms = 0.9s. +FIRMWARE_MULTIPART_ACK_VECTORS = ( + # header 0x2A = ROUTE_DIRECT | MULTIPART<<2; path_len 0x02 = 1-byte/2-hop; + # path AB CC (AB = us); payload 23=remaining2|ACK + CRC 95AFAB4D. + # out: header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x01; path CC; CRC. + pytest.param( + bytes([0xAB]), + bytes.fromhex("2A02ABCC234DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="direct-1-byte-2-hop", + ), + # 2-byte path width: path_len 0x42 = 2-byte/2-hop; path ABCD (us) + 1122. + pytest.param( + bytes([0xAB, 0xCD, 0xEF]), + bytes.fromhex("2A42ABCD1122234DABAF95"), + bytes.fromhex("0E4111224DABAF95"), + id="direct-2-byte-2-hop", + ), + # header 0x2B = ROUTE_TRANSPORT_DIRECT | MULTIPART<<2; transport codes + # 34127856; the regenerated ACK is forced to plain DIRECT (codes dropped). + pytest.param( + bytes([0xAB]), + bytes.fromhex("2B3412785602ABCC234DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="transport-direct-1-byte-2-hop", + ), +) + + +@pytest.mark.parametrize("local_hash_bytes, wire_in, wire_out", FIRMWARE_MULTIPART_ACK_VECTORS) +def test_forward_multipart_matches_firmware_wire_vectors(local_hash_bytes, wire_in, wire_out): + """Parse a firmware-format multipart frame and assert the exact relayed ACK bytes.""" + handler = _make_handler_with_hash(local_hash_bytes) + pkt = Packet() + pkt.read_from(wire_in) + result = handler.forward_multipart_direct(pkt) + assert result is not None + ack, delay = result + assert ack.write_to() == wire_out + assert delay == pytest.approx(0.9) + + # =================================================================== # 3. process_packet — route dispatch # =================================================================== @@ -1644,6 +1827,36 @@ class TestPacketInjectionRouting: sent_pkt = handler.dispatcher.send_packet.call_args.args[0] assert bytes(sent_pkt.path) == b"\x44\x55" + async def test_injected_multipart_ack_forwards_regenerated_plain_ack(self, handler): + """An intermediate hop relays a multipart ACK as a plain DIRECT ACK off the wire.""" + self._prepare_fast_tx(handler) + + pkt = _inject_from_wire( + _make_multipart_ack_packet( + crc=b"\x4d\xab\xaf\x95", remaining=2, path=bytes([LOCAL_HASH, 0x44]) + ) + ) + + with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock): + await handler(pkt, {"snr": 3.0, "rssi": -82}, local_transmission=False) + + assert handler.dispatcher.send_packet.call_count == 1 + sent_pkt = handler.dispatcher.send_packet.call_args.args[0] + assert sent_pkt.get_payload_type() == PAYLOAD_TYPE_ACK + assert sent_pkt.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(sent_pkt.payload) == b"\x4d\xab\xaf\x95" + assert bytes(sent_pkt.path) == b"\x44" + + async def test_injected_multipart_ack_flood_is_dropped(self, handler): + """MeshCore never flood-routes multipart; an intermediate hop drops it.""" + self._prepare_fast_tx(handler) + pkt = _inject_from_wire(_make_multipart_ack_packet(route=ROUTE_TYPE_FLOOD, path=b"\x11")) + + with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock): + await handler(pkt, {"snr": 3.0, "rssi": -82}, local_transmission=False) + + assert handler.dispatcher.send_packet.call_count == 0 + async def test_direct_for_other_node_is_dropped(self, handler): pkt = _inject_from_wire(_make_direct_packet(payload=b"\xaa\xbb", path=b"\xfe\x44")) metadata = {"snr": 2.0, "rssi": -90} @@ -1711,7 +1924,7 @@ class TestPacketInjectionRouting: assert len(original["duplicates"]) == 1 assert original["duplicates"][0]["drop_reason"] == "Duplicate" - @pytest.mark.parametrize("payload_type", range(16)) + @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_flood_injection_forwards(self, handler, payload_type): self._prepare_fast_tx(handler) pkt = _inject_from_wire( @@ -1733,7 +1946,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.path[-1] == LOCAL_HASH - @pytest.mark.parametrize("payload_type", range(16)) + @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_direct_injection_forwards(self, handler, payload_type): self._prepare_fast_tx(handler) pkt = _inject_from_wire( @@ -1755,7 +1968,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert bytes(sent_pkt.path) == b"\x44\x55" - @pytest.mark.parametrize("payload_type", range(16)) + @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_transport_flood_injection_forwards( self, handler, payload_type ): @@ -1781,7 +1994,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.transport_codes == [0x1111, 0x2222] - @pytest.mark.parametrize("payload_type", range(16)) + @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_transport_direct_injection_forwards( self, handler, payload_type ):