diff --git a/config.yaml.example b/config.yaml.example index d8e17f2..f404dd7 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -35,6 +35,11 @@ repeater: # Enable quality-based packet filtering and adaptive delays use_score_for_tx: false + # Multi-ack redundancy (0 = off, 1 = on; matches MeshCore's multi.acks pref). + # When on, a relayed routed ACK is preceded by a MULTIPART-wrapped copy so + # ACKs survive lossy links. Also settable via the mesh CLI: set multi.acks + multi_acks: 0 + # Score threshold for quality monitoring (future use) # Currently reserved for potential future features like dashboard alerts, # proactive statistics collection, or advanced filtering strategies. diff --git a/repeater/engine.py b/repeater/engine.py index 265be9e..a7eb704 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -4,6 +4,7 @@ import logging import secrets import time from collections import OrderedDict, deque +from enum import Enum from typing import Optional, Tuple from openhop_core.node.handlers.base import BaseHandler @@ -23,6 +24,7 @@ from openhop_core.protocol.constants import ( ROUTE_TYPE_TRANSPORT_FLOOD, ) from openhop_core.protocol.packet_utils import ( + PacketHashingUtils, PacketHeaderUtils, PathUtils, flood_rx_metrics, @@ -52,6 +54,57 @@ LOOP_DETECT_MAX_COUNTERS = { } +class DropReason(str, Enum): + """Canonical, non-alarming reasons a packet was intentionally not retransmitted. + + Single source of truth shared with the packet router, replacing the string + prefixes that used to be duplicated there. Members subclass ``str`` so they + compare equal to their text, JSON-serialize, and persist to SQLite as plain + strings unchanged. Where the engine still needs a detail suffix (hop counts, + loop mode, multipart sub-case) it formats the member into a larger string; + the router accepts both the bare member and any string that begins with one. + """ + + DUPLICATE = "Duplicate" + MAX_FLOOD_HOPS = "Max flood hops limit reached" + PATH_HOP_COUNT_MAX = "Path hop count at maximum" + PATH_EXCEEDS_MAX_SIZE = "Path would exceed MAX_PATH_SIZE" + DIRECT_NO_PATH = "Direct: no path" + DIRECT_NOT_FOR_US = "Direct: not for us" + UNSCOPED_FLOOD_DISABLED = "Unscoped flood policy disabled" + TRANSPORT_CODE_NOT_ALLOWED = "Transport code not allowed to flood" + FLOOD_LOOP_DETECTED = "FLOOD loop detected" + MARKED_DO_NOT_RETRANSMIT = "Marked do not retransmit" + REPEAT_DISABLED = "Repeat disabled" + NO_TX_MODE = "No TX mode" + DUTY_CYCLE_LIMIT = "Duty cycle limit" + EMPTY_PAYLOAD = "Empty payload" + PATH_TOO_LONG = "Path too long" + INVALID_ADVERT = "Invalid advert packet" + MULTIPART = "Multipart" + + # Python 3.11+ formats a (str, Enum) member as "DropReason.X" under str()/%s; + # return the value so log lines and stored records keep the plain reason text. + def __str__(self) -> str: + return self.value + + +class ForwardResult(tuple): + """A forwarding decision that unpacks as a plain ``(packet, delay_seconds)`` pair. + + ``extras`` carries additional ``(packet, delay_seconds)`` transmissions that + accompany the primary packet: MeshCore's routeDirectRecvAcks queues its + optional multi-ack redundancy copies ahead of the plain ACK at the same + scheduled time, so the caller must create the extra TX tasks before the + primary one. + """ + + def __new__(cls, packet: Packet, delay_s: float, extras=()): + result = super().__new__(cls, (packet, delay_s)) + result.extras = tuple(extras) + return result + + class RepeaterHandler(BaseHandler): @staticmethod def payload_type() -> int: @@ -93,6 +146,7 @@ class RepeaterHandler(BaseHandler): ) self.use_score_for_tx = config.get("repeater", {}).get("use_score_for_tx", False) self.score_threshold = config.get("repeater", {}).get("score_threshold", 0.3) + self.multi_acks = self._normalize_multi_acks(config) self.max_flood_hops = config.get("repeater", {}).get("max_flood_hops", 64) self.send_advert_interval_hours = config.get("repeater", {}).get( "send_advert_interval_hours", 10 @@ -314,6 +368,16 @@ class RepeaterHandler(BaseHandler): # Capture the forwarded path (after modification) forwarded_path_hashes = fwd_pkt.get_path_hashes_hex() + # MeshCore queues multi-ack redundancy copies ahead of the primary + # ACK at the same scheduled time, so create their TX tasks first. + # Each task re-checks the duty-cycle gate before transmitting. + extra_tx_tasks = [] + for extra_pkt, extra_delay in getattr(result, "extras", ()): + extra_airtime_ms = self.airtime_mgr.calculate_airtime(extra_pkt.get_raw_length()) + extra_tx_tasks.append( + await self.schedule_retransmit(extra_pkt, extra_delay, extra_airtime_ms) + ) + # Check duty-cycle before scheduling TX airtime_ms = self.airtime_mgr.calculate_airtime(fwd_pkt.get_raw_length()) @@ -368,7 +432,7 @@ class RepeaterHandler(BaseHandler): f"wait={wait_time:.1f}s before retry" ) self.dropped_count += 1 - drop_reason = "Duty cycle limit" + drop_reason = DropReason.DUTY_CYCLE_LIMIT else: tx_task = await self.schedule_retransmit( fwd_pkt, delay, airtime_ms, local_transmission=local_transmission @@ -399,13 +463,22 @@ class RepeaterHandler(BaseHandler): f"LBT: {lbt_attempts} attempts, {total_lbt_delay:.0f}ms delay, " f"backoffs={lbt_backoff_delays_ms}" ) + + # Redundancy copies ride alongside the primary result: collect them + # without letting a failed extra change the primary outcome. + for extra_task in extra_tx_tasks: + try: + if await extra_task: + self.forwarded_count += 1 + except Exception as e: + logger.warning(f"Multi-ack redundancy TX failed: {e}") else: self.dropped_count += 1 # Determine drop reason if local_transmission and not allow_local_tx: - drop_reason = policy_reason or "No TX mode" + drop_reason = policy_reason or DropReason.NO_TX_MODE elif not allow_forward: - drop_reason = policy_reason or "Repeat disabled" + drop_reason = policy_reason or DropReason.REPEAT_DISABLED else: # Check if packet has a specific drop reason set by handlers drop_reason = processed_packet.drop_reason or self._get_drop_reason( @@ -433,7 +506,7 @@ class RepeaterHandler(BaseHandler): # Set drop reason for duplicates and count flood vs direct dups if is_dupe and drop_reason is None: - drop_reason = "Duplicate" + drop_reason = DropReason.DUPLICATE if is_dupe: if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.flood_dup_count += 1 @@ -482,7 +555,11 @@ class RepeaterHandler(BaseHandler): if self.storage: try: # Only skip mqtt for actual invalid/bad packets - invalid_reasons = ["Invalid advert packet", "Empty payload", "Path too long"] + invalid_reasons = ( + DropReason.INVALID_ADVERT, + DropReason.EMPTY_PAYLOAD, + DropReason.PATH_TOO_LONG, + ) skip_mqtt = drop_reason in invalid_reasons if drop_reason else False self.storage.record_packet(packet_record, skip_mqtt_if_invalid=skip_mqtt) except Exception as e: @@ -631,7 +708,7 @@ class RepeaterHandler(BaseHandler): src_hash, dst_hash, transmitted=False, - drop_reason="Duplicate", + drop_reason=DropReason.DUPLICATE, is_duplicate=True, packet_hash=pkt_hash_full, ) @@ -795,13 +872,13 @@ class RepeaterHandler(BaseHandler): def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str: if self.is_duplicate(packet, packet_hash=packet_hash): - return "Duplicate" + return DropReason.DUPLICATE if not packet or not packet.payload: - return "Empty payload" + return DropReason.EMPTY_PAYLOAD if len(packet.path or []) > MAX_PATH_SIZE: - return "Path too long" + return DropReason.PATH_TOO_LONG route_type = packet.header & PH_ROUTE_MASK @@ -811,15 +888,15 @@ class RepeaterHandler(BaseHandler): "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True) ) if not unscoped_flood_allow: - return "Unscoped flood policy disabled" + return DropReason.UNSCOPED_FLOOD_DISABLED if route_type == ROUTE_TYPE_DIRECT: hash_size = packet.get_path_hash_size() if not packet.path or len(packet.path) < hash_size: - return "Direct: no path" + return DropReason.DIRECT_NO_PATH next_hop = bytes(packet.path[:hash_size]) if next_hop != self.local_hash_bytes[:hash_size]: - return "Direct: not for us" + return DropReason.DIRECT_NOT_FOR_US # Default reason return "Unknown" @@ -850,7 +927,7 @@ class RepeaterHandler(BaseHandler): def validate_packet(self, packet: Packet) -> Tuple[bool, str]: if not packet or not packet.payload: - return False, "Empty payload" + return False, DropReason.EMPTY_PAYLOAD if packet.get_path_hash_size() > 3: return False, "Reserved path hash size is invalid" @@ -863,6 +940,16 @@ class RepeaterHandler(BaseHandler): return True, "" + @staticmethod + def _normalize_multi_acks(config: dict) -> int: + """Read ``repeater.multi_acks``, constrained to 0..1 like MeshCore's + CommonCLI prefs load.""" + try: + value = int(config.get("repeater", {}).get("multi_acks", 0)) + except (TypeError, ValueError): + return 0 + return max(0, min(1, value)) + def _normalize_loop_detect_mode(self, mode) -> str: if isinstance(mode, str): normalized = mode.strip().lower() @@ -1006,7 +1093,7 @@ class RepeaterHandler(BaseHandler): if packet.is_marked_do_not_retransmit(): # Check if packet has custom drop reason if not packet.drop_reason: - packet.drop_reason = "Marked do not retransmit" + packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None # Check unscoped flood policy @@ -1016,24 +1103,24 @@ class RepeaterHandler(BaseHandler): route_type = packet.header & PH_ROUTE_MASK if route_type == ROUTE_TYPE_FLOOD: if not unscoped_flood_allow: - packet.drop_reason = "Unscoped flood policy disabled" + packet.drop_reason = DropReason.UNSCOPED_FLOOD_DISABLED return None # Check transport scopes flood policy if route_type == ROUTE_TYPE_TRANSPORT_FLOOD: allowed, check_reason = self._check_transport_codes(packet) if not allowed: - packet.drop_reason = "Transport code not allowed to flood" + packet.drop_reason = DropReason.TRANSPORT_CODE_NOT_ALLOWED return None mode = self._get_loop_detect_mode() if self._is_flood_looped(packet, mode): - packet.drop_reason = f"FLOOD loop detected ({mode})" + packet.drop_reason = f"{DropReason.FLOOD_LOOP_DETECTED} ({mode})" return None # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256. if self.is_duplicate(packet, packet_hash=packet_hash): - packet.drop_reason = "Duplicate" + packet.drop_reason = DropReason.DUPLICATE return None if packet.path is None: @@ -1045,17 +1132,17 @@ class RepeaterHandler(BaseHandler): hop_count = packet.get_path_hash_count() if self.max_flood_hops > 0 and hop_count >= self.max_flood_hops: - packet.drop_reason = f"Max flood hops limit reached ({hop_count}/{self.max_flood_hops})" + packet.drop_reason = f"{DropReason.MAX_FLOOD_HOPS} ({hop_count}/{self.max_flood_hops})" return None # path_len encodes hop count in 6 bits (0-63); adding ourselves must not exceed 63 if hop_count >= 63: - packet.drop_reason = "Path hop count at maximum (63), cannot append" + packet.drop_reason = f"{DropReason.PATH_HOP_COUNT_MAX} (63), cannot append" return None # Check path won't exceed MAX_PATH_SIZE after append if (hop_count + 1) * hash_size > MAX_PATH_SIZE: - packet.drop_reason = "Path would exceed MAX_PATH_SIZE" + packet.drop_reason = DropReason.PATH_EXCEEDS_MAX_SIZE return None self.mark_seen(packet, packet_hash=packet_hash) @@ -1082,7 +1169,7 @@ class RepeaterHandler(BaseHandler): # Check if packet is marked do-not-retransmit if packet.is_marked_do_not_retransmit(): if not packet.drop_reason: - packet.drop_reason = "Marked do not retransmit" + packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None hash_size = packet.get_path_hash_size() @@ -1090,17 +1177,17 @@ class RepeaterHandler(BaseHandler): # Check if we're the next hop if not packet.path or len(packet.path) < hash_size: - packet.drop_reason = "Direct: no path" + packet.drop_reason = DropReason.DIRECT_NO_PATH return None next_hop = bytes(packet.path[:hash_size]) if next_hop != self.local_hash_bytes[:hash_size]: - packet.drop_reason = "Direct: not for us" + packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256. if self.is_duplicate(packet, packet_hash=packet_hash): - packet.drop_reason = "Duplicate" + packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet, packet_hash=packet_hash) @@ -1116,8 +1203,8 @@ class RepeaterHandler(BaseHandler): MULTIPART_ACK_SPACING_MS = 300 def forward_multipart_direct( - self, packet: Packet, packet_hash: Optional[str] = None - ) -> Optional[Tuple[Packet, float]]: + self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None + ) -> Optional["ForwardResult"]: """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does. MeshCore never relays the multipart wrapper as an ordinary directed @@ -1125,7 +1212,8 @@ class RepeaterHandler(BaseHandler): 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 + type — is dropped. Returns a ``ForwardResult`` — ``(ack_packet, + delay_seconds)`` plus any multi-ack redundancy ``extras`` — when a regenerated ACK should be sent, otherwise ``None``. INVARIANT: purely synchronous, mirroring flood_forward / direct_forward — @@ -1134,7 +1222,7 @@ class RepeaterHandler(BaseHandler): # 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" + packet.drop_reason = f"{DropReason.MULTIPART}: not direct-routed" return None valid, reason = self.validate_packet(packet) @@ -1143,7 +1231,7 @@ class RepeaterHandler(BaseHandler): return None if packet.is_marked_do_not_retransmit(): - packet.drop_reason = packet.drop_reason or "Marked do not retransmit" + packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT return None payload = packet.payload or b"" @@ -1152,7 +1240,7 @@ class RepeaterHandler(BaseHandler): # 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" + packet.drop_reason = f"{DropReason.MULTIPART}: unsupported embedded type" return None hash_size = packet.get_path_hash_size() @@ -1161,33 +1249,127 @@ class RepeaterHandler(BaseHandler): # 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" + packet.drop_reason = DropReason.DIRECT_NO_PATH return None if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: - packet.drop_reason = "Direct: not for us" + packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None + # Dedupe before regenerating, on MeshCore's exact seen key: hasSeen() + # runs on a copy that keeps the MULTIPART payload type but carries the + # unwrapped ACK payload (the `remaining` count is stripped). Keeping + # that key distinct from the plain ACK's is what lets a multi-ack + # redundancy pair — MULTIPART copy plus plain ACK — survive every hop + # instead of the second half being swallowed as a duplicate. The + # pre-computed hash belongs to the wrapper as received, so recompute. + seen_key = ( + PacketHashingUtils.calculate_packet_hash( + PAYLOAD_TYPE_MULTIPART, packet.path_len, bytes(payload[1:]) + ) + .hex() + .upper() + ) + if self.is_duplicate(packet, packet_hash=seen_key): + packet.drop_reason = DropReason.DUPLICATE + return None + self.mark_seen(packet, packet_hash=seen_key) + # 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. + # drop the multipart header byte and force the ACK payload/route type, then + # remove this node from the path (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.transport_codes = [0, 0] 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 + extras, delay_ms = self._multi_ack_extras( + packet, float((remaining + 1) * self.MULTIPART_ACK_SPACING_MS), snr + ) + return ForwardResult(packet, delay_ms / 1000.0, extras) + + def _multi_ack_extras( + self, ack_packet: Packet, base_delay_ms: float, snr: float + ) -> Tuple[list, float]: + """Build MeshCore's optional multi-ack redundancy copies for a relayed ACK. + + With ``multi_acks`` enabled, routeDirectRecvAcks precedes the plain ACK + with a MULTIPART-wrapped copy (``remaining<<4 | ACK`` prefix byte) on the + already self-removed path, pushing the accumulated delay out by a direct + retransmit delay + 300 ms per copy; the plain ACK itself slides to the + final accumulated delay. Returns ``(extras, plain_ack_delay_ms)``. + """ + extra = self.multi_acks + extras = [] + delay_ms = base_delay_ms + while extra > 0: + delay_ms += ( + self._calculate_tx_delay(ack_packet, snr) * 1000.0 + self.MULTIPART_ACK_SPACING_MS + ) + wrapped = Packet() + wrapped.header = (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT + wrapped.payload = bytearray( + bytes([(extra << 4) | PAYLOAD_TYPE_ACK]) + bytes(ack_packet.payload) + ) + wrapped.payload_len = len(wrapped.payload) + wrapped.path = bytearray(ack_packet.path) + wrapped.path_len = ack_packet.path_len + extras.append((wrapped, delay_ms / 1000.0)) + extra -= 1 + return extras, delay_ms + + def forward_routed_ack( + self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None + ) -> Optional["ForwardResult"]: + """Relay a routed ACK the way MeshCore's routeDirectRecvAcks does. + + An ACK at an intermediate direct hop is not repeated verbatim: MeshCore + regenerates it (createAck) as a plain DIRECT ACK — transport codes + dropped, every other header bit cleared — removes this node from the + path, and transmits it with zero retransmit delay rather than the + generic direct forwarding delay. With ``multi_acks`` enabled, the plain + ACK is preceded by a MULTIPART-wrapped redundancy copy and both slide + out by a direct retransmit delay + 300 ms. + + INVARIANT: purely synchronous, mirroring direct_forward — the + is_duplicate + mark_seen pair must stay atomic within the event loop. + """ + 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 DropReason.MARKED_DO_NOT_RETRANSMIT + return None + + hash_size = packet.get_path_hash_size() + hop_count = packet.get_path_hash_count() + + # A final-hop ACK (no remaining path) is consumed locally, never relayed. + if not packet.path or len(packet.path) < hash_size: + packet.drop_reason = DropReason.DIRECT_NO_PATH + return None + if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: + packet.drop_reason = DropReason.DIRECT_NOT_FOR_US + return None + + # The packet hash covers only the payload type and payload, so the + # received form and the regenerated plain ACK share one key — matching + # MeshCore's hasSeen() on the incoming ACK before removeSelfFromPath(). + if self.is_duplicate(packet, packet_hash=packet_hash): + packet.drop_reason = DropReason.DUPLICATE + return None + self.mark_seen(packet, packet_hash=packet_hash) + + packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT + packet.transport_codes = [0, 0] + packet.path = bytearray(packet.path[hash_size:]) + packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) + + extras, delay_ms = self._multi_ack_extras(packet, 0.0, snr) + return ForwardResult(packet, delay_ms / 1000.0, extras) @staticmethod def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float: @@ -1257,8 +1439,18 @@ class RepeaterHandler(BaseHandler): # 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) + payload_type = packet.get_payload_type() + if payload_type == PAYLOAD_TYPE_MULTIPART: + return self.forward_multipart_direct(packet, snr, packet_hash=packet_hash) + + # Routed ACKs also get their own MeshCore branch: an intermediate direct + # hop re-emits the ACK immediately (routeDirectRecvAcks) instead of the + # delayed generic direct forward. Flood ACKs stay on the generic path. + if payload_type == PAYLOAD_TYPE_ACK and route_type in ( + ROUTE_TYPE_DIRECT, + ROUTE_TYPE_TRANSPORT_DIRECT, + ): + return self.forward_routed_ack(packet, snr, 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) @@ -1600,6 +1792,7 @@ class RepeaterHandler(BaseHandler): repeater_config = self.config.get("repeater", {}) self.use_score_for_tx = repeater_config.get("use_score_for_tx", False) self.score_threshold = repeater_config.get("score_threshold", 0.3) + self.multi_acks = self._normalize_multi_acks(self.config) self.send_advert_interval_hours = repeater_config.get("send_advert_interval_hours", 10) self.cache_ttl = repeater_config.get("cache_ttl", 60) self.max_flood_hops = repeater_config.get("max_flood_hops", 64) diff --git a/repeater/handler_helpers/advert.py b/repeater/handler_helpers/advert.py index 263d31f..36bca19 100644 --- a/repeater/handler_helpers/advert.py +++ b/repeater/handler_helpers/advert.py @@ -15,6 +15,8 @@ from typing import Dict, Optional, Tuple from openhop_core.node.handlers.advert import AdvertHandler +from repeater.engine import DropReason + logger = logging.getLogger("AdvertHelper") @@ -551,7 +553,7 @@ class AdvertHelper: if not advert_data or not advert_data.get("valid"): logger.warning("Invalid advert packet received, dropping.") packet.mark_do_not_retransmit() - packet.drop_reason = "Invalid advert packet" + packet.drop_reason = DropReason.INVALID_ADVERT return # Extract data from parsed advert diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 8e3029f..d08b9eb 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -5,7 +5,6 @@ import time from openhop_core.node.handlers.ack import AckHandler from openhop_core.node.handlers.advert import AdvertHandler from openhop_core.node.handlers.control import ControlHandler -from openhop_core.node.handlers.group_text import GroupTextHandler from openhop_core.node.handlers.login_response import LoginResponseHandler from openhop_core.node.handlers.login_server import LoginServerHandler from openhop_core.node.handlers.multipart import MultipartAckHandler @@ -16,11 +15,13 @@ 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, PH_ROUTE_MASK, ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT, ) +from repeater.engine import DropReason from repeater.policy_engine import PolicyDecision, PolicyEngine logger = logging.getLogger("PacketRouter") @@ -29,27 +30,11 @@ logger = logging.getLogger("PacketRouter") # so the client is not spammed with duplicate telemetry when the mesh delivers multiple copies. _COMPANION_DEDUPE_TTL_SEC = 60.0 -# Drop reasons that are normal policy outcomes and should not be warning-level. -# TODO: create Enum in engine for drop reasons and use it here and in engine instead of string matching. -_EXPECTED_DROP_REASON_PREFIXES = ( - "Duplicate", - "Max flood hops limit reached", - "Path hop count at maximum", - "Path would exceed MAX_PATH_SIZE", - "Direct: no path", - "Direct: not for us", - "Unscoped flood policy disabled", - "Transport code not allowed to flood", - "FLOOD loop detected", - "Marked do not retransmit", - "Repeat disabled", - "No TX mode", - "Duty cycle limit", - "Empty payload", - "Path too long", - "Invalid advert packet", - "Multipart", -) +# Normal policy/forwarding drop outcomes (logged at debug, not warning). The engine +# is the single source of truth via DropReason; string-form acceptance is retained +# because records and the recent_packets fallback still carry detailed text variants +# (e.g. "Max flood hops limit reached (5/5)"). +_EXPECTED_DROP_REASONS = frozenset(DropReason) def _companion_dedup_key(packet) -> str | None: @@ -77,10 +62,14 @@ def _is_direct_intermediate_hop(packet) -> bool: ) -def _is_expected_drop_reason(reason: str | None) -> bool: +def _is_expected_drop_reason(reason) -> bool: + # Membership is the fast path for a DropReason member or its exact text; the + # startswith scan covers detailed variants that embed a suffix. + if reason in _EXPECTED_DROP_REASONS: + return True if not isinstance(reason, str) or not reason: return False - return any(reason.startswith(prefix) for prefix in _EXPECTED_DROP_REASON_PREFIXES) + return any(reason.startswith(expected) for expected in _EXPECTED_DROP_REASONS) def _drop_reason_from_recent_packets(handler, packet) -> str | None: @@ -181,11 +170,28 @@ class PacketRouter: if exc is not None: logger.error("_route_packet raised: %s", exc, exc_info=exc) - def _should_deliver_path_to_companions(self, packet) -> bool: - """Return True if this PATH/protocol-response should be delivered to companions (first of duplicates).""" + def _was_delivered_to_companions(self, packet) -> bool: + """Pure check: True if this PATH/protocol-response was already delivered (unexpired). + + Does not mutate delivery state, so callers can deliver first and only record + the packet as delivered once a bridge has actually received it. + """ key = _companion_dedup_key(packet) if not key: - return True + return False + expiry = self._companion_delivered.get(key) + return expiry is not None and expiry > time.time() + + def _mark_delivered_to_companions(self, packet) -> None: + """Record that this PATH/protocol-response reached companions. + + Called only after a bridge has received the packet, so a delivery that + raised for every bridge is retried on the next copy instead of being + suppressed for the full dedupe TTL. + """ + key = _companion_dedup_key(packet) + if not key: + return now = time.time() # Prune expired entries only when the dict grows large, avoiding a full # dict comprehension on every packet. 200 entries × 60 s TTL means a @@ -196,9 +202,13 @@ class PacketRouter: self._companion_delivered = { k: v for k, v in self._companion_delivered.items() if v > now } - if key in self._companion_delivered: - return False self._companion_delivered[key] = now + _COMPANION_DEDUPE_TTL_SEC + + def _should_deliver_path_to_companions(self, packet) -> bool: + """Atomic check-and-mark: True on the first of duplicate PATH/protocol-responses.""" + if self._was_delivered_to_companions(packet): + return False + self._mark_delivered_to_companions(packet) return True def _policy_companion_decision(self, packet, metadata: dict) -> PolicyDecision | None: @@ -263,6 +273,25 @@ class PacketRouter: return {} return companion_bridges + async def _fan_out_to_bridges(self, packet, bridges, *, context: str) -> bool: + """Offer packet to each bridge; True if any bridge authenticated it. + + Accepts a dict of bridges — pass a single-entry dict for targeted delivery + to the bridge that owns ``dest_hash``. A bridge that raises is logged and + skipped; ``result.authenticated`` is read directly (every bridge returns a + HandlerResult) so a broken contract surfaces instead of being hidden. + """ + authenticated = False + for bridge in bridges.values(): + try: + result = await bridge.process_received_packet(packet) + except Exception as e: + logger.debug("Companion bridge %s error: %s", context, e) + continue + if result.authenticated is True: + authenticated = True + return authenticated + async def _consume_via_local_candidates( self, packet, metadata: dict, dest_hash, helper, process_method_name: str ) -> bool: @@ -537,11 +566,7 @@ class PacketRouter: # Also feed adverts to companion bridges (for contact/path updates), # but keep policy drop final just like the other companion paths. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge advert error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="advert") elif payload_type == LoginServerHandler.payload_type(): # Route ANON_REQ/login to the local identity that owns it. The on-air @@ -566,11 +591,7 @@ class PacketRouter: # ACK has no dest in payload (4-byte CRC only); deliver to all bridges so sender sees send_confirmed. # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge ACK error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="ACK") elif payload_type == MultipartAckHandler.payload_type(): # MULTIPART ACK wrapper: low nibble of first byte is embedded payload type. @@ -610,18 +631,22 @@ class PacketRouter: 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): - 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): + if not self._was_delivered_to_companions(packet): + consumed = ( + await self._fan_out_to_bridges( + packet, {dest_hash: companion_bridges[dest_hash]}, context="PATH" + ) + or consumed + ) + self._mark_delivered_to_companions(packet) + elif companion_bridges and not self._was_delivered_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: - 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}") + consumed = ( + await self._fan_out_to_bridges(packet, companion_bridges, context="PATH") + or consumed + ) + self._mark_delivered_to_companions(packet) logger.debug( "PATH dest=0x%02x (anon) delivered to %d bridge(s) for matching", dest_hash or 0, @@ -643,23 +668,15 @@ class PacketRouter: 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: - 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, - ) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") + consumed = await self._fan_out_to_bridges( + packet, {dest_hash: companion_bridges[dest_hash]}, context="RESPONSE" + ) + logger.info("RESPONSE dest=0x%02x delivered to companion bridge", dest_hash) elif dest_hash == local_hash and companion_bridges: # Response addressed to this repeater (e.g. path-based reply to first hop) - for bridge in companion_bridges.values(): - try: - 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}") + consumed = await self._fan_out_to_bridges( + packet, companion_bridges, context="RESPONSE" + ) logger.info( "RESPONSE dest=0x%02x (local) delivered to %d companion bridge(s)", dest_hash, @@ -669,12 +686,9 @@ class PacketRouter: # Dest not in bridges and not local: likely ANON_REQ response (dest = ephemeral # sender hash). Deliver to all bridges; each will try to decrypt and ignore if # not relevant (firmware-like behavior, works with multiple companion bridges). - for bridge in companion_bridges.values(): - try: - 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}") + consumed = await self._fan_out_to_bridges( + packet, companion_bridges, context="RESPONSE" + ) logger.debug( "RESPONSE dest=0x%02x (anon) delivered to %d bridge(s) for matching", dest_hash or 0, @@ -691,23 +705,16 @@ class PacketRouter: elif payload_type == ProtocolResponseHandler.payload_type(): # PAYLOAD_TYPE_PATH (0x08): protocol responses (telemetry, binary, etc.). - # Deliver at most once per logical packet so the client is not spammed with duplicates. - # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. + # Deliver at most once per logical packet so the client is not spammed with duplicates, + # but always deliver at a final hop (we are the destination). Do not set + # processed_by_injection for a middle hop so the packet still reaches engine forwarding. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges and self._should_deliver_path_to_companions(packet): - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") - if companion_bridges and _is_direct_final_hop(packet): - # DIRECT with empty path: we're the final hop; ensure delivery to all bridges (anon) - if not self._should_deliver_path_to_companions(packet): - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE (final hop) error: {e}") + final_hop = _is_direct_final_hop(packet) + if companion_bridges and (final_hop or not self._was_delivered_to_companions(packet)): + await self._fan_out_to_bridges(packet, companion_bridges, context="RESPONSE") + self._mark_delivered_to_companions(packet) + if companion_bridges and final_hop: + # DIRECT with empty path: we're the final hop, so consume after delivery. processed_by_injection = True self._record_for_ui(packet, metadata) @@ -732,36 +739,22 @@ class PacketRouter: if companion_bridges and _is_direct_final_hop(packet): # 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) - except Exception as e: - logger.debug(f"Companion bridge REQ (final hop) error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="REQ") processed_by_injection = True self._record_for_ui(packet, metadata) - elif payload_type == GroupTextHandler.payload_type(): + elif payload_type == PAYLOAD_TYPE_GRP_TXT: # GRP_TXT: pass to all companions (they filter by channel); still forward. # Policy drop is final and blocks companion delivery. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges: - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge GRP_TXT error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_TXT") elif payload_type == PAYLOAD_TYPE_GRP_DATA: # MeshCore forwards direct packets with remaining hops before payload # handling. Otherwise, companions authenticate and filter channels. if not _is_direct_intermediate_hop(packet): companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges: - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge GRP_DATA error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_DATA") # Only pass to repeater engine if not already processed by injection # Skip engine for packets we injected for TX (already sent; avoid double-send/double-count) diff --git a/tests/test_engine.py b/tests/test_engine.py index 069cfb4..0ea4315 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -38,6 +38,13 @@ LOCAL_HASH = 0xAB # repeater's own 1-byte path hash # (forward_multipart_direct), covered by TestForwardMultipartDirect. _GENERIC_FORWARD_PAYLOAD_TYPES = [pt for pt in range(16) if pt != PAYLOAD_TYPE_MULTIPART] +# On direct routes, ACK also leaves the generic path — routed ACKs are +# regenerated by forward_routed_ack (TestForwardRoutedAck); flood ACKs stay +# generic. +_GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES = [ + pt for pt in _GENERIC_FORWARD_PAYLOAD_TYPES if pt != PAYLOAD_TYPE_ACK +] + def _make_config(**overrides) -> dict: """Return a minimal valid config dict for RepeaterHandler.""" @@ -477,21 +484,47 @@ class TestForwardMultipartDirect: 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() + def test_dedup_key_keeps_multipart_type_with_unwrapped_payload(self, handler): + """The seen key matches MeshCore's hasSeen() copy: MULTIPART payload + type over the unwrapped ACK payload, checked before removeSelfFromPath. + The `remaining` count is stripped, so re-wraps of the same ACK dedupe.""" + from openhop_core.protocol.packet_utils import PacketHashingUtils + + pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC])) + key = ( + PacketHashingUtils.calculate_packet_hash( + PAYLOAD_TYPE_MULTIPART, pkt.path_len, b"\x4d\xab\xaf\x95" + ) + .hex() + .upper() + ) handler.forward_multipart_direct(pkt) assert key in handler.seen_packets + # A re-wrap of the same ACK with a different remaining count is a duplicate. + rewrap = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC])) + assert handler.forward_multipart_direct(rewrap) is None + assert rewrap.drop_reason == "Duplicate" + + def test_multipart_relay_does_not_block_plain_ack_relay(self, handler): + """A multi-ack redundancy pair (MULTIPART copy + plain ACK) must survive + the hop: MeshCore's seen keys differ by payload type, so relaying the + MULTIPART copy first must not swallow the plain ACK as a duplicate.""" + wrapper = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC])) + assert handler.forward_multipart_direct(wrapper) is not None + + plain = _make_direct_packet( + payload=b"\x4d\xab\xaf\x95", + path=bytes([LOCAL_HASH, 0xCC]), + payload_type=PAYLOAD_TYPE_ACK, + ) + result = handler.forward_routed_ack(plain) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + assert bytes(ack.payload) == b"\x4d\xab\xaf\x95" + def test_marked_do_not_retransmit_multipart_dropped(self, handler): pkt = _make_multipart_ack_packet() pkt.mark_do_not_retransmit() @@ -554,6 +587,259 @@ def test_forward_multipart_matches_firmware_wire_vectors(local_hash_bytes, wire_ assert delay == pytest.approx(0.9) +# =================================================================== +# 2c. forward_routed_ack — MeshCore routeDirectRecvAcks parity +# =================================================================== + + +def _make_direct_ack_packet( + crc: bytes = b"\x4d\xab\xaf\x95", + path: bytes = None, + route: int = ROUTE_TYPE_DIRECT, + transport_codes=None, +) -> Packet: + """Build a routed ACK: 4-byte CRC payload on a direct route.""" + if path is None: + path = bytes([LOCAL_HASH, 0xCC]) + pkt = Packet() + pkt.header = route | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) + pkt.payload = bytearray(crc) + pkt.payload_len = len(pkt.payload) + pkt.path = bytearray(path) + pkt.path_len = len(path) + if transport_codes is not None: + pkt.transport_codes = list(transport_codes) + return pkt + + +class TestForwardRoutedAck: + """forward_routed_ack mirrors MeshCore: regenerate the ACK as a plain DIRECT + packet and relay it immediately, never through the generic delayed forward.""" + + def test_intermediate_ack_relays_with_zero_delay(self, handler): + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + 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) + + def test_transport_direct_ack_regenerated_as_plain_direct(self, handler): + pkt = _make_direct_ack_packet( + route=ROUTE_TYPE_TRANSPORT_DIRECT, transport_codes=(0x1234, 0x5678) + ) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + # createAck rebuilds the header: plain DIRECT, transport codes dropped. + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + assert not ack.has_transport_codes() + assert ack.transport_codes == [0, 0] + + def test_final_hop_ack_not_relayed(self, handler): + pkt = _make_direct_ack_packet(path=b"") + pkt.path_len = 0 + result = handler.forward_routed_ack(pkt) + assert result is None + assert "no path" in pkt.drop_reason + + def test_wrong_next_hop_ack_dropped(self, handler): + pkt = _make_direct_ack_packet(path=bytes([0xFE, 0xCC])) + result = handler.forward_routed_ack(pkt) + assert result is None + assert "not for us" in pkt.drop_reason + + def test_duplicate_ack_dropped(self, handler): + handler.forward_routed_ack(_make_direct_ack_packet()) + pkt2 = _make_direct_ack_packet() + result = handler.forward_routed_ack(pkt2) + assert result is None + assert pkt2.drop_reason == "Duplicate" + + def test_marked_do_not_retransmit_ack_dropped(self, handler): + pkt = _make_direct_ack_packet() + pkt.mark_do_not_retransmit() + result = handler.forward_routed_ack(pkt) + assert result is None + + def test_process_packet_dispatches_direct_ack(self, handler): + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + result = handler.process_packet(pkt, snr=5.0) + assert result is not None + ack, delay = result + # The routed-ACK branch relays immediately; the generic direct forward + # would have applied a randomized direct TX delay instead. + assert delay == pytest.approx(0.0) + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + + def test_process_packet_keeps_flood_ack_on_generic_path(self, handler): + pkt = _make_flood_packet( + payload=b"\x4d\xab\xaf\x95", path=b"\x11", payload_type=PAYLOAD_TYPE_ACK + ) + result = handler.process_packet(pkt, snr=5.0) + assert result is not None + fwd, _ = result + # Generic flood forwarding appends this node's hash. + assert fwd.get_route_type() == ROUTE_TYPE_FLOOD + assert list(fwd.path) == [0x11, LOCAL_HASH] + + +# Literal MeshCore-format routed-ACK frames (Packet::writeTo layout). MeshCore's +# routeDirectRecvAcks re-emits through createAck: the header is rebuilt from +# scratch (payload type ACK, route DIRECT, version bits cleared, transport codes +# dropped) and this node is removed from the path. Independent wire fixtures — +# do NOT rebuild them through Packet.write_to()/PathUtils. +FIRMWARE_ROUTED_ACK_VECTORS = ( + # header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x02 = 1-byte/2-hop; + # path AB CC (AB = us); CRC 95AFAB4D. out: path popped to CC. + pytest.param( + bytes([0xAB]), + bytes.fromhex("0E02ABCC4DABAF95"), + 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("0E42ABCD11224DABAF95"), + bytes.fromhex("0E4111224DABAF95"), + id="direct-2-byte-2-hop", + ), + # header 0x0F = ACK<<2 | ROUTE_TRANSPORT_DIRECT; transport codes 34127856; + # the regenerated ACK is forced to plain DIRECT (codes dropped). + pytest.param( + bytes([0xAB]), + bytes.fromhex("0F3412785602ABCC4DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="transport-direct-1-byte-2-hop", + ), + # header 0x4E = version bit set on a direct ACK; createAck clears it. + pytest.param( + bytes([0xAB]), + bytes.fromhex("4E02ABCC4DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="direct-version-bits-cleared", + ), +) + + +@pytest.mark.parametrize("local_hash_bytes, wire_in, wire_out", FIRMWARE_ROUTED_ACK_VECTORS) +def test_forward_routed_ack_matches_firmware_wire_vectors(local_hash_bytes, wire_in, wire_out): + """Parse a firmware-format routed ACK and assert the exact relayed bytes.""" + handler = _make_handler_with_hash(local_hash_bytes) + pkt = Packet() + pkt.read_from(wire_in) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert ack.write_to() == wire_out + assert delay == pytest.approx(0.0) + + +class TestMultiAckRedundancy: + """multi_acks=1 adds MeshCore's MULTIPART-wrapped redundancy copy ahead of + the plain ACK; both slide out by the direct retransmit delay + 300 ms.""" + + def test_default_off_produces_no_extras(self, handler): + result = handler.forward_routed_ack(_make_direct_ack_packet()) + assert result.extras == () + + def test_extra_precedes_plain_ack_at_same_delay(self, handler): + handler.multi_acks = 1 + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + with patch.object(handler, "_calculate_tx_delay", return_value=0.25): + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + # delay = direct retransmit delay (0.25s) + 300ms, shared by both copies. + assert delay == pytest.approx(0.55) + assert len(result.extras) == 1 + wrapped, wrapped_delay = result.extras[0] + assert wrapped_delay == pytest.approx(0.55) + # MULTIPART wrapper: (remaining=1)<<4 | ACK prefix byte + the CRC, on the + # already self-removed path, plain DIRECT route. + assert wrapped.get_payload_type() == PAYLOAD_TYPE_MULTIPART + assert wrapped.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(wrapped.payload) == b"\x13\x4d\xab\xaf\x95" + assert bytes(wrapped.path) == bytes(ack.path) + assert wrapped.path_len == ack.path_len + # Exact wire bytes: header 0x2A = MULTIPART<<2 | DIRECT. + assert wrapped.write_to() == bytes.fromhex("2A01CC134DABAF95") + + def test_multipart_relay_also_gets_extra(self, handler): + handler.multi_acks = 1 + pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC])) + with patch.object(handler, "_calculate_tx_delay", return_value=0.25): + result = handler.forward_multipart_direct(pkt) + assert result is not None + ack, delay = result + # base (remaining+1)*300ms = 0.9s, plus 0.25s + 300ms for the extra. + assert delay == pytest.approx(1.45) + assert len(result.extras) == 1 + _, wrapped_delay = result.extras[0] + assert wrapped_delay == pytest.approx(1.45) + + def test_multi_acks_clamped_like_firmware(self): + # MeshCore CommonCLI constrains multi_acks to 0..1 on prefs load. + handler = _make_handler_with_hash(bytes([LOCAL_HASH])) + handler.config.setdefault("repeater", {})["multi_acks"] = 5 + handler.reload_runtime_config() + assert handler.multi_acks == 1 + + handler.config["repeater"]["multi_acks"] = "bogus" + handler.reload_runtime_config() + assert handler.multi_acks == 0 + + handler.config["repeater"]["multi_acks"] = -3 + handler.reload_runtime_config() + assert handler.multi_acks == 0 + + def test_init_reads_repeater_multi_acks(self): + with ( + patch("repeater.engine.StorageCollector"), + patch("repeater.engine.RepeaterHandler._start_background_tasks"), + ): + from repeater.engine import RepeaterHandler + + handler = RepeaterHandler( + _make_config(repeater={"multi_acks": 1}), + _make_dispatcher(), + LOCAL_HASH, + local_hash_bytes=bytes([LOCAL_HASH]), + ) + assert handler.multi_acks == 1 + + @pytest.mark.asyncio + async def test_call_transmits_wrapped_copy_before_plain_ack(self, handler): + handler.multi_acks = 1 + handler.airtime_mgr.calculate_airtime = MagicMock(return_value=20.0) + handler.airtime_mgr.can_transmit = MagicMock(return_value=(True, 0.0)) + handler.airtime_mgr.record_tx = MagicMock() + handler.airtime_mgr.record_rx = MagicMock() + handler.dispatcher.send_packet = AsyncMock(return_value=True) + + pkt = _inject_from_wire(_make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))) + with ( + patch.object(handler, "_calculate_tx_delay", return_value=0.0), + patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock), + ): + await handler(pkt, {"snr": 3.0, "rssi": -80}, local_transmission=False) + + assert handler.dispatcher.send_packet.call_count == 2 + first = handler.dispatcher.send_packet.call_args_list[0].args[0] + second = handler.dispatcher.send_packet.call_args_list[1].args[0] + # MeshCore queues the MULTIPART redundancy copy ahead of the plain ACK. + assert first.write_to() == bytes.fromhex("2A01CC134DABAF95") + assert second.write_to() == bytes.fromhex("0E01CC4DABAF95") + assert handler.forwarded_count == 2 + + # =================================================================== # 3. process_packet — route dispatch # =================================================================== @@ -2163,7 +2449,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.path[-1] == LOCAL_HASH - @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) + @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_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( @@ -2211,7 +2497,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.transport_codes == [0x1111, 0x2222] - @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) + @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_transport_direct_injection_forwards( self, handler, payload_type ): diff --git a/tests/test_identities_endpoint_payload.py b/tests/test_identities_endpoint_payload.py new file mode 100644 index 0000000..78a3aaf --- /dev/null +++ b/tests/test_identities_endpoint_payload.py @@ -0,0 +1,127 @@ +"""Regression: /api/identities payload shape with the registered repeater identity. + +Since the daemon registers its own default identity in the IdentityManager +(so companion/room-server collisions against the repeater's hash byte are +caught), the endpoint's raw ``registered`` list carries a +``repeater:repeater`` entry and ``total_registered`` counts it. The web UI +is unaffected — it renders only the per-entry ``registered`` boolean on +configured room servers/companions, which match on ``room_server:``/ +``companion:``-prefixed names — but external API consumers see the new +entry, so this test pins the intended payload. +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import cherrypy +import pytest + +from openhop_core.protocol import LocalIdentity + +from repeater.identity_manager import IdentityManager +from repeater.web.api_endpoints import APIEndpoints + + +@pytest.fixture +def cherrypy_ctx(monkeypatch): + request = SimpleNamespace(method="GET", params={}, json={}) + response = SimpleNamespace(headers={}, status=200) + monkeypatch.setattr(cherrypy, "request", request, raising=False) + monkeypatch.setattr(cherrypy, "response", response, raising=False) + return request, response + + +def _distinct_identities(count): + """LocalIdentities with pairwise-distinct first hash bytes.""" + picked = [] + seen = set() + while len(picked) < count: + identity = LocalIdentity() + hash_byte = identity.get_public_key()[0] + if hash_byte not in seen: + seen.add(hash_byte) + picked.append(identity) + return picked + + +def _make_api(config, identity_manager): + api = APIEndpoints.__new__(APIEndpoints) + api.config = config + api.daemon_instance = SimpleNamespace(identity_manager=identity_manager) + api.send_advert_func = None + api.event_loop = None + api.stats_getter = None + api._config_path = "/tmp/test-config.yaml" + api.config_manager = MagicMock() + return api + + +def test_registered_list_includes_repeater_identity(cherrypy_ctx): + repeater_id, companion_id = _distinct_identities(2) + + config = { + "identities": { + "companions": [ + { + "name": "phone", + "identity_key": companion_id.get_private_key().hex(), + "settings": {}, + } + ], + "room_servers": [], + } + } + manager = IdentityManager(config) + assert manager.register_identity("repeater", repeater_id, config, "repeater") + assert manager.register_identity("phone", companion_id, {}, "companion") + + api = _make_api(config, manager) + payload = api.identities() + + assert payload["success"] is True + data = payload["data"] + registered = data["registered"] + assert data["total_registered"] == len(registered) == 2 + + by_name = {entry["name"]: entry for entry in registered} + repeater_entry = by_name["repeater:repeater"] + assert repeater_entry["type"] == "repeater" + assert repeater_entry["hash"] == f"0x{repeater_id.get_public_key()[0]:02X}" + assert repeater_entry["public_key"] == repeater_id.get_public_key().hex() + + companion_entry = by_name["companion:phone"] + assert companion_entry["type"] == "companion" + + # The configured-companion view (what the web UI renders) matches the + # companion by its prefixed name and is not disturbed by the repeater + # entry: it still reports the companion as registered. + assert data["total_configured_companions"] == 1 + ui_entry = data["configured_companions"][0] + assert ui_entry["name"] == "phone" + assert ui_entry["registered"] is True + + +def test_name_repeater_is_reserved_by_the_default_identity(cherrypy_ctx): + """Registering the default repeater identity reserves the bare name + "repeater": a room server or companion configured with that name is + rejected by the collision rules, and the endpoint reports it + unregistered rather than silently matching the repeater's entry.""" + repeater_id, room_id = _distinct_identities(2) + + config = { + "identities": { + "room_servers": [{"name": "repeater", "identity_key": "aa" * 32, "settings": {}}], + "companions": [], + } + } + manager = IdentityManager(config) + assert manager.register_identity("repeater", repeater_id, config, "repeater") + assert not manager.register_identity("repeater", room_id, {}, "room_server") + + api = _make_api(config, manager) + data = api.identities()["data"] + + assert data["total_registered"] == 1 + room_entry = data["configured"][0] + assert room_entry["registered"] is False + assert room_entry["hash"] is None