diff --git a/repeater/handler_helpers/acl.py b/repeater/handler_helpers/acl.py index e2ad745..6d38bce 100644 --- a/repeater/handler_helpers/acl.py +++ b/repeater/handler_helpers/acl.py @@ -48,6 +48,30 @@ class ACL: self.allow_read_only = allow_read_only self.clients: Dict[bytes, ClientInfo] = {} + def _is_replay(self, client: ClientInfo, timestamp: int) -> bool: + if timestamp <= client.last_timestamp: + logger.warning( + f"Possible replay attack! timestamp={timestamp}, last={client.last_timestamp}" + ) + return True + return False + + def _touch_client_session( + self, + client: ClientInfo, + shared_secret: bytes, + timestamp: int, + sync_since: int = None, + ) -> None: + now = int(time.time()) + client.last_timestamp = timestamp + client.last_activity = now + client.last_login_success = now + client.shared_secret = shared_secret + if sync_since is not None: + client.sync_since = sync_since + logger.debug(f"Stored sync_since={sync_since} for client") + def authenticate_client( self, client_identity: Identity, @@ -106,13 +130,23 @@ class ACL: if not password: client = self.clients.get(pub_key) if client is None: - if self.allow_read_only: - logger.info("Blank password, allowing read-only guest access") - return True, PERM_ACL_GUEST - else: + if not self.allow_read_only: logger.info("Blank password, sender not in ACL and read-only disabled") return False, 0 - logger.info(f"ACL-based login for {pub_key[:6].hex()}...") + if len(self.clients) >= self.max_clients: + logger.warning("ACL full, cannot add client") + return False, 0 + client = ClientInfo(client_identity, PERM_ACL_GUEST) + self.clients[pub_key] = client + logger.info("Blank password, allowing read-only guest access") + else: + logger.info(f"ACL-based login for {pub_key[:6].hex()}...") + + if self._is_replay(client, timestamp): + return False, 0 + self._touch_client_session(client, shared_secret, timestamp, sync_since=sync_since) + if (client.permissions & PERM_ACL_ROLE_MASK) == 0: + client.permissions |= PERM_ACL_GUEST return True, client.permissions permissions = 0 @@ -140,23 +174,11 @@ class ACL: self.clients[pub_key] = client logger.info(f"Added new client {pub_key[:6].hex()}...") - if timestamp <= client.last_timestamp: - logger.warning( - f"Possible replay attack! timestamp={timestamp}, last={client.last_timestamp}" - ) + if self._is_replay(client, timestamp): return False, 0 - - client.last_timestamp = timestamp - client.last_activity = int(time.time()) - client.last_login_success = int(time.time()) + self._touch_client_session(client, shared_secret, timestamp, sync_since=sync_since) client.permissions &= ~PERM_ACL_ROLE_MASK client.permissions |= permissions - client.shared_secret = shared_secret - - # Store sync_since for room server clients - if sync_since is not None: - client.sync_since = sync_since - logger.debug(f"Stored sync_since={sync_since} for client") logger.info(f"Login success! Permissions: {'ADMIN' if client.is_admin() else 'GUEST'}") return True, client.permissions diff --git a/repeater/handler_helpers/login.py b/repeater/handler_helpers/login.py index 216e6db..43189ac 100644 --- a/repeater/handler_helpers/login.py +++ b/repeater/handler_helpers/login.py @@ -122,7 +122,7 @@ class LoginHelper: def auth_callback_with_context( client_identity, shared_secret, password, timestamp, sync_since=None ): - return identity_acl.authenticate_client( + success, permissions = identity_acl.authenticate_client( client_identity=client_identity, shared_secret=shared_secret, password=password, @@ -132,6 +132,26 @@ class LoginHelper: target_identity_name=name, target_identity_config=config, ) + if success and identity_type == "room_server" and self.sqlite_handler is not None: + try: + sync_kwargs = {} + if sync_since is not None: + sync_kwargs["sync_since"] = sync_since + self.sqlite_handler.upsert_client_sync( + room_hash=f"0x{hash_byte:02X}", + client_pubkey=client_identity.get_public_key().hex(), + pending_ack_crc=0, + push_post_timestamp=0, + ack_timeout_time=0, + push_failures=0, + last_activity=time.time(), + **sync_kwargs, + ) + except Exception as e: + logger.warning( + f"Failed to reset room sync guard state after login for hash=0x{hash_byte:02X}: {e}" + ) + return success, permissions handler = LoginServerHandler( local_identity=identity, diff --git a/repeater/handler_helpers/mesh_cli.py b/repeater/handler_helpers/mesh_cli.py index aa379f7..4d2631e 100644 --- a/repeater/handler_helpers/mesh_cli.py +++ b/repeater/handler_helpers/mesh_cli.py @@ -2,7 +2,6 @@ import logging from pathlib import Path from typing import Any, Callable, Dict, Optional - logger = logging.getLogger(__name__) @@ -329,8 +328,8 @@ class MeshCLI: def _cmd_version(self) -> str: """Get version information.""" role = "room_server" if self.identity_type == "room_server" else "repeater" - version = self.config.get("version", "1.0.0") - return f"pyMC_{role} v{version}" + version = self.config.get("version", "13") + return f"openHop_{role} v{version}" # ==================== Get Commands ==================== diff --git a/repeater/handler_helpers/path.py b/repeater/handler_helpers/path.py index 98e8e1b..da187e5 100644 --- a/repeater/handler_helpers/path.py +++ b/repeater/handler_helpers/path.py @@ -1,3 +1,4 @@ +import asyncio import logging import time @@ -5,14 +6,31 @@ logger = logging.getLogger("PathHelper") class PathHelper: - def __init__(self, acl_dict=None, log_fn=None): + def __init__(self, acl_dict=None, log_fn=None, ack_received_callback=None): self.acl_dict = acl_dict or {} self.log_fn = log_fn or logger.info + self.ack_received_callback = ack_received_callback + + async def _register_ack_crc(self, ack_crc: int) -> None: + """Propagate an ACK CRC to the configured callback.""" + if ack_crc is None: + return + callback = self.ack_received_callback + if callback is None: + return + try: + result = callback(ack_crc) + if asyncio.iscoroutine(result): + await result + except Exception as e: + logger.debug(f"ACK callback failed for CRC {ack_crc:08X}: {e}") async def process_path_packet(self, packet): + from openhop_core.protocol.constants import PAYLOAD_TYPE_ACK from openhop_core.protocol.crypto import CryptoUtils + from openhop_core.protocol.packet_utils import PathUtils try: if len(packet.payload) < 2: @@ -60,30 +78,54 @@ class PathHelper: return False # Parse decrypted PATH data - # Format: path_len(1) + path[path_len] + extra_type(1) + extra[...] + # Format: path_len(1) + path[path_byte_len] + extra_type(1) + extra[...] if len(decrypted) < 1: 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 - path_len = decrypted[0] - if len(decrypted) < 1 + path_len: + if len(decrypted) < 1 + path_byte_len: logger.debug( - f"PATH data truncated: need {1 + path_len} bytes, got {len(decrypted)}" + f"PATH data truncated: need {1 + path_byte_len} bytes, got {len(decrypted)}" ) return False - path_data = decrypted[1 : 1 + path_len] + path_data = decrypted[1 : 1 + path_byte_len] # Update client's out_path (same as C++ memcpy) client.out_path = bytearray(path_data) - client.out_path_len = path_len + client.out_path_len = ( + path_len_byte if PathUtils.is_valid_path_len(path_len_byte) else path_byte_len + ) client.last_activity = int(time.time()) logger.info( f"Updated out_path for client 0x{src_hash:02X} -> 0x{dest_hash:02X}: " - f"path_len={path_len}, path={[hex(b) for b in path_data]}" + f"path_len_byte=0x{path_len_byte:02X}, hops={path_hops}, " + f"path={[hex(b) for b in path_data]}" ) + # 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 :] + if extra_type == PAYLOAD_TYPE_ACK and len(extra_payload) >= 4: + ack_crc = int.from_bytes(extra_payload[:4], "little") + logger.info( + f"PATH bundled ACK extracted for client 0x{src_hash:02X}: CRC={ack_crc:08X}" + ) + + 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 diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 7b8d45d..80bed78 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -6,6 +6,7 @@ from typing import Dict from openhop_core.protocol import CryptoUtils, PacketBuilder from openhop_core.protocol.constants import PAYLOAD_TYPE_TXT_MSG +from openhop_core.protocol.packet_utils import PathUtils logger = logging.getLogger("RoomServer") @@ -328,6 +329,12 @@ class RoomServer: if sync_state: failures = sync_state.get("push_failures", 0) + if failures >= MAX_PUSH_FAILURES: + logger.debug( + f"Room '{self.room_name}': Client 0x{client_info.id.get_public_key()[0]:02X} " + f"at max failures ({failures}), skipping push" + ) + return False if failures > 0: # Apply exponential backoff backoff_idx = min(failures, len(RETRY_BACKOFF_SCHEDULE) - 1) @@ -356,11 +363,9 @@ class RoomServer: plaintext = ( timestamp.to_bytes(4, "little") + bytes([flags]) + author_prefix + message_bytes ) - - # Calculate expected ACK (same algorithm as openhop_core) - attempt = 0 - pack_data = PacketBuilder._pack_timestamp_data(timestamp, attempt, message_bytes) - ack_hash = CryptoUtils.sha256(pack_data + client_info.id.get_public_key())[:4] + # Calculate expected ACK (MeshCore signed text): + # sha256(timestamp + flags + author_prefix + text + recipient_pubkey)[:4] + ack_hash = CryptoUtils.sha256(plaintext + client_info.id.get_public_key())[:4] expected_ack_crc = int.from_bytes(ack_hash, "little") # Determine routing based on stored out_path @@ -378,29 +383,52 @@ class RoomServer: # Add stored path for direct routing if route_type == "direct" and len(client_info.out_path) > 0: - packet.path = bytearray(client_info.out_path[: client_info.out_path_len]) - packet.path_len = client_info.out_path_len + if PathUtils.is_valid_path_len(client_info.out_path_len): + path_byte_len = PathUtils.get_path_byte_len(client_info.out_path_len) + packet.path = bytearray(client_info.out_path[:path_byte_len]) + packet.path_len = client_info.out_path_len + else: + # Legacy fallback: treat stored path as 1-byte-hop path and clamp to + # valid encoded range (0-63 hops). + legacy_hops = min(len(client_info.out_path), 63) + packet.path = bytearray(client_info.out_path[:legacy_hops]) + packet.path_len = legacy_hops # Calculate ACK timeout if route_type == "flood": ack_timeout = PUSH_ACK_TIMEOUT_FLOOD_MS / 1000.0 else: - path_len = client_info.out_path_len if client_info.out_path_len >= 0 else 0 + if PathUtils.is_valid_path_len(client_info.out_path_len): + path_len = PathUtils.get_path_hash_count(client_info.out_path_len) + else: + path_len = min(len(client_info.out_path), 63) ack_timeout = ( PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1) ) / 1000.0 # Update client sync state with pending ACK + current_sync_since = ( + sync_state.get("sync_since", 0) + if sync_state + else getattr(client_info, "sync_since", 0) + ) self.db.upsert_client_sync( room_hash=f"0x{self.room_hash:02X}", client_pubkey=client_info.id.get_public_key().hex(), + sync_since=current_sync_since, pending_ack_crc=expected_ack_crc, push_post_timestamp=post["post_timestamp"], ack_timeout_time=time.time() + ack_timeout, + last_activity=time.time(), ) # Send packet (dispatcher will track ACK automatically) # This blocks for the entire transmission duration (0.5-9 seconds) - success = await self.packet_injector(packet, wait_for_ack=True) + success = await self.packet_injector( + packet, + wait_for_ack=True, + expected_crc=expected_ack_crc, + ack_timeout_s=ack_timeout, + ) # SAFETY: Release transmission lock AFTER send completes self.global_limiter.release() @@ -460,7 +488,7 @@ class RoomServer: pending_ack_crc=0, ) - if failures >= 3: + if failures >= MAX_PUSH_FAILURES: logger.warning( f"Room '{self.room_name}': Client 0x{client_pubkey[0]:02X} " f"has {failures} consecutive failures" @@ -625,7 +653,7 @@ class RoomServer: ) continue - if push_failures >= 3: + if push_failures >= MAX_PUSH_FAILURES: logger.debug( f"Skipping client 0x{client.id.get_public_key()[0]:02X} (max failures)" ) diff --git a/repeater/main.py b/repeater/main.py index 9c95d2e..65621ee 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -178,8 +178,12 @@ class RepeaterDaemon: from openhop_core import LocalIdentity from openhop_core.node.dispatcher import Dispatcher - self.dispatcher = Dispatcher(self.radio) + dedupe_enabled = bool( + self.config.get("repeater", {}).get("dispatcher_dedupe_enabled", False) + ) + self.dispatcher = Dispatcher(self.radio, dedupe_enabled=dedupe_enabled) logger.info("Dispatcher initialized") + logger.info("Dispatcher dedupe enabled: %s", dedupe_enabled) # Initialize Identity Manager for additional identities (e.g., room servers) self.identity_manager = IdentityManager(self.config) @@ -373,6 +377,11 @@ class RepeaterDaemon: self.path_helper = PathHelper( acl_dict=self.login_helper.get_acl_dict(), # Per-identity ACLs log_fn=logger.info, + ack_received_callback=( + self.dispatcher._register_ack_received + if self.dispatcher and hasattr(self.dispatcher, "_register_ack_received") + else None + ), ) logger.info("PATH packet processing helper initialized") @@ -403,9 +412,7 @@ class RepeaterDaemon: n, ) - # Subscribe to parsed packets (pre-dedup) so duplicate path variants - # still appear in the web UI even though the Dispatcher blocks them. - self.dispatcher.add_raw_packet_subscriber(self._on_raw_packet_for_dedup_logging) + self._register_duplicate_logging_hook(dedupe_enabled) # When trace reaches final node, push PUSH_CODE_TRACE_DATA (0x89) to companion clients (firmware onTraceRecv) self.trace_helper.on_trace_complete = self._on_trace_complete_for_companions @@ -909,6 +916,14 @@ class RepeaterDaemon: except Exception as e: logger.debug("Push RX raw to companion: %s", e) + def _register_duplicate_logging_hook(self, dedupe_enabled: bool) -> None: + """Register pre-dedup duplicate logging only when dispatcher dedupe is active.""" + if not self.dispatcher or not dedupe_enabled: + return + # When dispatcher dedupe is disabled, duplicates still flow through + # router -> repeater_handler and are already recorded there. + self.dispatcher.add_raw_packet_subscriber(self._on_raw_packet_for_dedup_logging) + def _on_raw_packet_for_dedup_logging(self, pkt, data: bytes, analysis: dict) -> None: """Record duplicate packets for UI visibility. diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 9b43e66..503607a 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -8,6 +8,7 @@ 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 from openhop_core.node.handlers.path import PathHandler from openhop_core.node.handlers.protocol_request import ProtocolRequestHandler from openhop_core.node.handlers.protocol_response import ProtocolResponseHandler @@ -261,6 +262,16 @@ class PacketRouter: except Exception as e: logger.debug("Record for UI failed: %s", e) + async def _register_ack_with_dispatcher(self, ack_crc: int, context: str) -> None: + """Best-effort ACK CRC registration with the dispatcher waiter path.""" + dispatcher = getattr(self.daemon, "dispatcher", None) + if dispatcher is None or not hasattr(dispatcher, "_register_ack_received"): + return + try: + await dispatcher._register_ack_received(ack_crc) + except Exception as e: + logger.debug("Dispatcher %s registration error: %s", context, e) + async def enqueue(self, packet): """Add packet to router queue.""" if self.queue.full(): @@ -271,7 +282,14 @@ class PacketRouter: pass await self.queue.put(packet) - async def inject_packet(self, packet, wait_for_ack: bool = False, origin_hash=None): + async def inject_packet( + self, + packet, + wait_for_ack: bool = False, + expected_crc=None, + origin_hash=None, + ack_timeout_s: float = 5.0, + ): try: metadata = { "rssi": getattr(packet, "rssi", 0), @@ -327,11 +345,20 @@ class PacketRouter: dispatcher = getattr(self.daemon, "dispatcher", None) if dispatcher and hasattr(dispatcher, "wait_for_ack"): try: - expected_crc = packet.get_crc() - ack_ok = await dispatcher.wait_for_ack(expected_crc, timeout=5.0) + wait_crc = ( + expected_crc if expected_crc is not None else packet.get_crc() + ) + wait_timeout = ( + float(ack_timeout_s) + if isinstance(ack_timeout_s, (int, float)) and ack_timeout_s > 0 + else 5.0 + ) + ack_ok = await dispatcher.wait_for_ack(wait_crc, timeout=wait_timeout) if not ack_ok: logger.warning( - "Injected packet ACK timeout (crc=%08X)", expected_crc + "Injected packet ACK timeout (crc=%08X, timeout=%.1fs)", + wait_crc, + wait_timeout, ) return False except Exception as e: @@ -463,6 +490,10 @@ class PacketRouter: self._record_for_ui(packet, metadata) elif payload_type == AckHandler.payload_type(): + # Ensure ACK CRC reaches dispatcher waiter path even when only router fallback is active. + if len(getattr(packet, "payload", b"")) >= 4: + ack_crc = int.from_bytes(packet.payload[:4], "little") + await self._register_ack_with_dispatcher(ack_crc, "ACK") # 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) @@ -472,6 +503,15 @@ class PacketRouter: except Exception as e: logger.debug(f"Companion bridge ACK error: {e}") + elif payload_type == MultipartAckHandler.payload_type(): + # MULTIPART ACK wrapper: low nibble of first byte is embedded payload type. + if ( + len(getattr(packet, "payload", b"")) >= 5 + and (packet.payload[0] & 0x0F) == AckHandler.payload_type() + ): + ack_crc = int.from_bytes(packet.payload[1:5], "little") + await self._register_ack_with_dispatcher(ack_crc, "multi-ACK") + elif payload_type == TextMessageHandler.payload_type(): dest_hash = packet.payload[0] if packet.payload else None companion_bridges = self._companion_bridges_for_packet(packet, metadata) @@ -486,6 +526,13 @@ class PacketRouter: self._record_for_ui(packet, metadata) 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. + if self.daemon.path_helper: + try: + await self.daemon.path_helper.process_path_packet(packet) + except Exception as e: + logger.debug(f"Path helper processing error: {e}") 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: @@ -506,8 +553,6 @@ class PacketRouter: len(companion_bridges), ) # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. - elif self.daemon.path_helper: - await self.daemon.path_helper.process_path_packet(packet) elif payload_type == LoginResponseHandler.payload_type(): # PAYLOAD_TYPE_RESPONSE (0x01): payload is dest_hash(1)+src_hash(1)+encrypted. diff --git a/tests/test_handler_helpers_acl_advert.py b/tests/test_handler_helpers_acl_advert.py index fecc80f..3babf58 100644 --- a/tests/test_handler_helpers_acl_advert.py +++ b/tests/test_handler_helpers_acl_advert.py @@ -40,6 +40,21 @@ def test_acl_blank_password_guest_rules_and_room_server_password_requirements(): ) assert ok is True assert perms == PERM_ACL_GUEST + guest_client = acl.get_client(identity.get_public_key()) + assert guest_client is not None + assert guest_client.permissions == PERM_ACL_GUEST + assert guest_client.shared_secret == b"secret" + assert guest_client.last_timestamp == 10 + + # Blank-password logins now track replay/timestamp just like password logins. + replay_ok, replay_perms = acl.authenticate_client( + client_identity=identity, + shared_secret=b"secret", + password="", + timestamp=10, + ) + assert replay_ok is False + assert replay_perms == 0 acl_ro_disabled = ACL(allow_read_only=False) ok2, perms2 = acl_ro_disabled.authenticate_client( diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index cd1d9c3..009a9c1 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -149,7 +149,12 @@ async def test_room_server_push_post_to_client_success_direct_route_sets_path_an assert ok is True assert bytes(packet.path) == b"\xaa\xbb" assert packet.path_len == 2 - injector.assert_awaited_once_with(packet, wait_for_ack=True) + injector.assert_awaited_once_with( + packet, + wait_for_ack=True, + expected_crc=67305985, + ack_timeout_s=10.0, + ) rs._handle_ack_received.assert_awaited_once_with( client.id.get_public_key(), post["post_timestamp"] ) diff --git a/tests/test_main_py_coverage.py b/tests/test_main_py_coverage.py index 1d9e24d..d049ec3 100644 --- a/tests/test_main_py_coverage.py +++ b/tests/test_main_py_coverage.py @@ -76,6 +76,20 @@ def test_get_stats_includes_public_key_gps_sensors_and_radio_state(): assert stats["radio_error"] == "missing device" +def test_register_duplicate_logging_hook_only_when_dispatcher_dedup_enabled(): + daemon = RepeaterDaemon(_base_config(), radio=object()) + dispatcher = SimpleNamespace(add_raw_packet_subscriber=MagicMock()) + daemon.dispatcher = dispatcher + + daemon._register_duplicate_logging_hook(False) + dispatcher.add_raw_packet_subscriber.assert_not_called() + + daemon._register_duplicate_logging_hook(True) + dispatcher.add_raw_packet_subscriber.assert_called_once_with( + daemon._on_raw_packet_for_dedup_logging + ) + + def test_detect_container_from_proc_env_and_fallback_path(): with patch("builtins.open", MagicMock()) as open_mock: open_mock.return_value.__enter__.return_value.read.return_value = b"container=docker"