From eaf28e28b72be939d9f31001c6f47633ae482ba7 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 14:32:57 -0700 Subject: [PATCH] feat(mqtt): periodic neighbours publication with serialized scope discovery Python port of the firmware's WITH_MQTT_NEIGHBORS feature (MeshCore PR #35 plus the aba571ed pacing rework). Each cycle refreshes the zero-hop neighbour table with a node-discovery broadcast, asks every neighbour for its region scopes, and publishes the assembled table to the MQTT neighbors topic. - NeighborScopeHelper: client side of the anon-regions request (the server side already existed in openhop_core). Queries are strictly serialized -- firing them as a burst makes the responses collide, which is what the firmware fix addressed. router.inject_packet resolves at the firmware's logTx/logTxFail boundary, so the response deadline is armed only once the request is on air and the firmware's QUEUED/PENDING state machine collapses into an await. The response window is sized from radio parameters, mirroring neighborDiscoverQueryTimeoutMs(). - NeighborsPublisher: owns the schedule, the immutable per-cycle snapshot, and the payload. The snapshot merges live discovery responses over the stored table because get_neighbors() serves a 60s cache that store_advert() does not invalidate. - packet_router: offer PAYLOAD_TYPE_RESPONSE to the scope helper before the companion fan-out; it consumes only on an authenticated tag match. - Per-broker `neighbors: true` opt-in (default off) plus a mqtt_brokers.neighbors.enabled master switch; interval is 12-336h as in firmware, rejected rather than clamped. - `discover.scopes` mesh CLI command triggers one cycle immediately. The whole zero-hop table is published including neighbours that did not answer (timeout / send_failed), matching the firmware payload. No 10 KB cap: that existed for a fixed PSRAM buffer. --- config.yaml.example | 21 + repeater/data_acquisition/mqtt_handler.py | 50 ++ repeater/handler_helpers/__init__.py | 2 + repeater/handler_helpers/mesh_cli.py | 53 ++ repeater/handler_helpers/neighbor_scopes.py | 403 +++++++++++++ repeater/main.py | 40 ++ repeater/neighbors_publisher.py | 559 +++++++++++++++++ repeater/packet_router.py | 23 +- repeater/web/api_endpoints.py | 99 ++++ tests/test_mqtt_neighbors.py | 626 ++++++++++++++++++++ 10 files changed, 1875 insertions(+), 1 deletion(-) create mode 100644 repeater/handler_helpers/neighbor_scopes.py create mode 100644 repeater/neighbors_publisher.py create mode 100644 tests/test_mqtt_neighbors.py diff --git a/config.yaml.example b/config.yaml.example index d6c0927..3d3816c 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -545,6 +545,26 @@ mqtt_brokers: email: "" brokers: [] + # Periodic neighbours publication (the "neighbors" topic). + # + # Each cycle runs a zero-hop node-discovery broadcast to refresh the neighbour + # table, then asks each zero-hop repeater for its region scopes with an + # anonymous regions request, then publishes the assembled table. Scope queries + # are issued ONE AT A TIME - firing them as a burst makes the responses collide. + # + # Off unless a broker opts in below with `neighbors: true`; `enabled` here is a + # master kill switch for the whole feature. + # neighbors: + # enabled: true # master on/off (default true) + # interval_hours: 24 # 12-336, how often a cycle runs + # discovery_timeout_seconds: 60 # node-discovery collection window + # scope_response_timeout_seconds: 0 # 0 = size the window from radio settings + # max_neighbors: 32 # cap on neighbours queried per cycle + # max_neighbor_age_seconds: 86400 # ignore zero-hop rows older than this + # max_sweep_seconds: 900 # give up on a cycle after this long + # + # `discover.scopes` over the mesh CLI runs one cycle immediately. + # Below is the broker object schema: # enabled: true|false # Enable this specific mqtt broker # name: "" # Internal name for this broker @@ -560,6 +580,7 @@ mqtt_brokers: # letsmesh, waev - MC2MQTT family flavors (same topic structure, network-specific identity) # mqtt - legacy openhop-repeater local-broker convention (custom topic, singular 'packet') # retain_status: true|false # Sets MQTT "retain" on status messages so they remain on the broker when disconnected. Also enforces a QOS of 1 (guaranteed delivery) + # neighbors: true|false # Publish the periodic neighbours table to this broker's "neighbors" topic (default false). Only enable for brokers that expect the topic - some reject unknown topics and close the connection. # tls: # enabled: true|false # Enable TLS. If the endpoint's certificate is self-signed, the Root CA should be added to the OS's certificate store. # insecure: true|false # Validate TLS certificates diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 6bc8b35..6c417a3 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/repeater/data_acquisition/mqtt_handler.py @@ -239,6 +239,12 @@ class _BrokerConnection: self.enabled = broker.get("enabled", False) self.retain_status = broker.get("retain_status", False) + # Opt-in per broker, default off. The neighbors topic is not part of every + # MC2MQTT deployment's contract, and a broker that rejects an unexpected + # topic closes the connection (rc=16) - see the format auto-correction + # below for the same hazard. This is the openhop equivalent of the + # firmware's packets-only MeshRank exclusion. + self.neighbors_enabled = bool(broker.get("neighbors", False)) self._tls_verified = False @@ -1016,6 +1022,50 @@ class MeshCoreToMqttPusher: return results + def has_neighbors_brokers(self) -> bool: + """True when at least one enabled broker opted into the neighbors topic.""" + return any(conn.enabled and conn.neighbors_enabled for conn in self.connections) + + def has_connected_neighbors_brokers(self) -> bool: + """True when an opted-in broker is connected and could receive a publish.""" + return any( + conn.enabled and conn.neighbors_enabled and conn.is_connected() + for conn in self.connections + ) + + def publish_neighbors(self, payload: dict): + """Publish the neighbours table to every broker that opted in. + + QoS 1 and non-retained, matching the firmware's ``publishNeighbors()``: + the table is a periodic snapshot, so a stale retained copy would outlive + its usefulness, but it is infrequent enough to be worth delivering once. + """ + message = json.dumps(payload) + neighbor_count = len(payload.get("neighbors", [])) + logger.debug( + f"Publishing topic='neighbors', neighbors={neighbor_count}, " + f"bytes={len(message.encode('utf-8'))}" + ) + + results = [] + with self._lock: + for conn in self.connections: + if not (conn.enabled and conn.neighbors_enabled): + continue + if not conn.is_connected(): + logger.warning( + f"Cannot publish neighbors to {conn.broker['name']} - not connected" + ) + continue + result = conn.publish("neighbors", message, retain=False, qos=1) + results.append((conn.broker["name"], result)) + _trace(f"Published to {conn.broker['name']} -- neighbors") + + if not results: + logger.warning("No connected broker opted into the neighbors topic") + + return results + def publish_mqtt(self, payload: dict, subtopic: str, retain: bool = False, qos: int = 0): """Publish message to brokers using the legacy custom-MQTT format only. diff --git a/repeater/handler_helpers/__init__.py b/repeater/handler_helpers/__init__.py index 444658b..31f24f8 100644 --- a/repeater/handler_helpers/__init__.py +++ b/repeater/handler_helpers/__init__.py @@ -3,6 +3,7 @@ from .advert import AdvertHelper from .discovery import DiscoveryHelper from .login import LoginHelper +from .neighbor_scopes import NeighborScopeHelper from .path import PathHelper from .protocol_request import ProtocolRequestHelper from .text import TextHelper @@ -13,6 +14,7 @@ __all__ = [ "DiscoveryHelper", "AdvertHelper", "LoginHelper", + "NeighborScopeHelper", "TextHelper", "PathHelper", "ProtocolRequestHelper", diff --git a/repeater/handler_helpers/mesh_cli.py b/repeater/handler_helpers/mesh_cli.py index 40b7ee2..1ecf87e 100644 --- a/repeater/handler_helpers/mesh_cli.py +++ b/repeater/handler_helpers/mesh_cli.py @@ -247,6 +247,8 @@ class MeshCLI: return self._cmd_neighbors() elif command.startswith("neighbor.remove "): return self._cmd_neighbor_remove(command) + elif command.startswith("discover.scopes"): + return self._cmd_discover_scopes(command) elif command.startswith("discover.neighbors"): return self._cmd_discover_neighbors(command) @@ -326,6 +328,7 @@ class MeshCLI: " neighbors List neighbors", " neighbor.remove Remove neighbor by pubkey", " discover.neighbors Send zero-hop neighbor discovery", + " discover.scopes Discover neighbor scopes, publish to MQTT", " tempradio ", " setperm Set ACL permissions", " log start|stop|erase Logging control", @@ -380,6 +383,12 @@ class MeshCLI: ), "neighbors": "List known neighbor nodes from the routing table.", "discover.neighbors": "Send a neighbor discovery request.", + "discover.scopes": ( + "discover.scopes\n" + " Refresh the zero-hop neighbor table, query each neighbor for its\n" + " region scopes, then publish the table to the MQTT neighbors topic.\n" + " Requires a broker configured with neighbors: true." + ), "setperm": "setperm \u2014 Set ACL permissions for a node.", "log": "log start|stop|erase \u2014 Control logging.", } @@ -1295,6 +1304,50 @@ class MeshCLI: logger.error(f"discover.neighbors failed: {e}", exc_info=True) return f"Error: {e}" + def _cmd_discover_scopes(self, command: str) -> str: + """Refresh the neighbour table, query each neighbour's scopes, publish once. + + Manual trigger for the periodic MQTT neighbors cycle (firmware + ``discover.scopes``). The cycle takes minutes -- a node-discovery window + plus one serialized scope query per neighbour -- so this schedules it and + returns immediately rather than holding the CLI reply open. + """ + sub = command[15:] + if sub.strip(): + return "Err - discover.scopes has no options" + + daemon_instance = getattr(self.config_manager, "daemon", None) + publisher = getattr(daemon_instance, "neighbors_publisher", None) + if not publisher: + return "Error: Neighbors publisher not available" + + if not publisher.enabled(): + return "Err - neighbors publishing is disabled (no broker opted in)" + + if publisher.status().get("phase") == "active": + return "Err - neighbors cycle already active" + + import asyncio + + loop = self._event_loop + if loop is None: + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop is None or not loop.is_running(): + return "Error: Event loop not available" + + try: + loop.call_soon_threadsafe( + lambda: asyncio.create_task(publisher.run_cycle(trigger="manual")) + ) + return "OK - neighbor scope discovery started" + except Exception as e: + logger.error(f"discover.scopes failed: {e}", exc_info=True) + return f"Error: {e}" + # ==================== Temporary Radio Commands ==================== def _cmd_tempradio(self, command: str) -> str: diff --git a/repeater/handler_helpers/neighbor_scopes.py b/repeater/handler_helpers/neighbor_scopes.py new file mode 100644 index 0000000..cab3510 --- /dev/null +++ b/repeater/handler_helpers/neighbor_scopes.py @@ -0,0 +1,403 @@ +"""Neighbour scope-query helper for openHop Repeater. + +Client side of the anonymous *regions* request. The server side already lives in +:class:`~openhop_core.node.handlers.anon_request.AnonRequestHandler` (wired up by +:mod:`repeater.handler_helpers.login`); this module asks the question instead of +answering it, so the MQTT ``neighbors`` topic can report which region scopes each +zero-hop neighbour serves. + +Wire format mirrors firmware ``MyMesh::sendAnonRegionsReq``: a PAYLOAD_TYPE_ANON_REQ +whose plaintext is ``tag(4) + ANON_REQ_TYPE_REGIONS + 0x00``, where the trailing +``0x00`` asks for a zero-hop reply path. The reply is a PAYLOAD_TYPE_RESPONSE +datagram (``dest_hash(1) + src_hash(1) + cipher``) whose plaintext is +``tag(4) + clock(4) + comma_separated_scope_names``. + +Two firmware behaviours are load-bearing and are reproduced here: + +* The request must be **route-direct** — ``AnonRequestHandler`` (and the firmware + regions handler it mirrors) ignores flooded discovery sub-types outright. +* Queries are issued **strictly one at a time**. Firing them as a burst makes every + responder answer into the same window and the replies collide; the firmware fixed + this by walking the snapshot one entry at a time and only arming the response + deadline once the request has actually transmitted (MeshCore ``aba571ed``). + ``router.inject_packet`` already resolves at that exact point — it awaits + ``dispatcher.send_packet`` under the engine TX lock — so the firmware's + QUEUED/PENDING state machine collapses into a sequential ``await`` here. +""" + +import asyncio +import logging +import time +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional + +from openhop_core.protocol import CryptoUtils, Identity +from openhop_core.protocol.constants import ( + ANON_REQ_TYPE_REGIONS, + MAX_PACKET_PAYLOAD, + PAYLOAD_TYPE_RESPONSE, +) + +logger = logging.getLogger("NeighborScopes") + +# Per-neighbour outcome, published verbatim in the MQTT neighbors payload. +STATUS_RESPONDED = "responded" +STATUS_TIMEOUT = "timeout" +STATUS_SEND_FAILED = "send_failed" + +# Responder-side reply delay to budget for. Firmware uses SERVER_RESPONSE_DELAY +# (500 ms); openhop_core's AnonRequestHandler uses 300 ms. Budget the larger so a +# firmware neighbour is not written off early. +SERVER_RESPONSE_DELAY_MS = 500.0 + +# Slack for scheduler jitter, matching the firmware's own 360 ms constant. +RESPONSE_TIMEOUT_SLACK_MS = 360.0 + +# Bounds for the auto-sized response timeout. The floor keeps a fast radio config +# from producing an unreachably tight window; the ceiling keeps a slow one (SF12, +# narrow bandwidth) from stalling a whole sweep on one dead neighbour. +MIN_RESPONSE_TIMEOUT_SEC = 5.0 +MAX_RESPONSE_TIMEOUT_SEC = 120.0 + +# Used when no AirtimeManager is available to size the window from radio params. +FALLBACK_RESPONSE_TIMEOUT_SEC = 30.0 + +# Whole-sweep budget, and how long a duty-cycle backlog may be before the sweep +# gives up rather than queueing behind forwarding traffic for minutes. +DEFAULT_MAX_SWEEP_SECONDS = 900.0 +DEFAULT_DUTY_CYCLE_ABORT_SECONDS = 30.0 + + +@dataclass(frozen=True) +class NeighborSnapshot: + """One neighbour, frozen at sweep start. + + Firmware originally indexed into the live ``neighbours[]`` table and had to + stop doing that (``aba571ed``) because an advert arriving mid-pass could + reshuffle it. Same reasoning here: the sqlite neighbour table keeps changing + while a multi-minute sweep runs, so what gets published is decided up front. + """ + + pubkey: str + last_seen: float = 0.0 + snr: float = 0.0 + + @property + def pubkey_bytes(self) -> bytes: + return bytes.fromhex(self.pubkey) + + +@dataclass +class ScopeResult: + status: str + scopes: str = "" + + +@dataclass +class _PendingQuery: + pubkey: str + tag: int + src_hash: int + future: "asyncio.Future[str]" = field(repr=False, default=None) + + +class _ScopeTarget: + """Minimal contact stand-in for ``PacketBuilder.create_anon_request``. + + ``out_path_len = 0`` (rather than -1) selects direct routing with an empty + path, i.e. the zero-hop request the responder's route-direct gate requires. + """ + + def __init__(self, pubkey_hex: str): + self.public_key = pubkey_hex + self.out_path_len = 0 + self.out_path = b"" + + +class NeighborScopeHelper: + """Issues anon-regions queries and matches their responses.""" + + def __init__( + self, + local_identity, + packet_injector: Optional[Callable] = None, + airtime_manager=None, + config: Optional[dict] = None, + ): + self.local_identity = local_identity + self.packet_injector = packet_injector + self.airtime_manager = airtime_manager + + self._pending: Optional[_PendingQuery] = None + self._sweep_lock = asyncio.Lock() + + self._response_timeout_override = 0.0 + self._max_sweep_seconds = DEFAULT_MAX_SWEEP_SECONDS + self._duty_cycle_abort_seconds = DEFAULT_DUTY_CYCLE_ABORT_SECONDS + self._direct_tx_delay_factor = 0.5 + self.refresh_config(config or {}) + + # ------------------------------------------------------------------ + # Configuration + # ------------------------------------------------------------------ + def refresh_config(self, config: dict) -> None: + """Re-read the tunables so a live config update takes effect next sweep.""" + neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {} + + self._response_timeout_override = _as_float( + neighbors_cfg.get("scope_response_timeout_seconds", 0), 0.0 + ) + self._max_sweep_seconds = max( + 1.0, _as_float(neighbors_cfg.get("max_sweep_seconds"), DEFAULT_MAX_SWEEP_SECONDS) + ) + self._duty_cycle_abort_seconds = max( + 0.0, + _as_float( + neighbors_cfg.get("duty_cycle_abort_seconds"), DEFAULT_DUTY_CYCLE_ABORT_SECONDS + ), + ) + self._direct_tx_delay_factor = _as_float( + (config.get("delays", {}) or {}).get("direct_tx_delay_factor"), 0.5 + ) + + def response_timeout(self) -> float: + """How long to wait for a reply once the request is on air. + + Mirrors firmware ``neighborDiscoverQueryTimeoutMs()``: the responder's + fixed reply delay, plus however long it may defer the transmission, plus + airtime for one packet ahead of the response and the response itself. + The firmware term there is ``getCADFailMaxDuration()``; openhop's + equivalent deferral is the engine's random direct-TX window, + ``[0, 5 * airtime * direct_tx_delay_factor]``. + """ + if self._response_timeout_override > 0: + return self._response_timeout_override + + airtime_ms = self._estimate_response_airtime_ms() + if airtime_ms <= 0: + return FALLBACK_RESPONSE_TIMEOUT_SEC + + deferral_ms = 5.0 * airtime_ms * max(0.0, self._direct_tx_delay_factor) + total_ms = ( + SERVER_RESPONSE_DELAY_MS + deferral_ms + (2.0 * airtime_ms) + RESPONSE_TIMEOUT_SLACK_MS + ) + return min(MAX_RESPONSE_TIMEOUT_SEC, max(MIN_RESPONSE_TIMEOUT_SEC, total_ms / 1000.0)) + + def _estimate_response_airtime_ms(self) -> float: + if not self.airtime_manager: + return 0.0 + try: + # Worst-case response size, matching the firmware's + # getEstAirtimeFor(MAX_PACKET_PAYLOAD + 2). + return float(self.airtime_manager.calculate_airtime(MAX_PACKET_PAYLOAD + 2)) + except Exception as e: + logger.debug(f"Could not estimate response airtime: {e}") + return 0.0 + + # ------------------------------------------------------------------ + # Sweep + # ------------------------------------------------------------------ + @property + def active(self) -> bool: + return self._sweep_lock.locked() + + async def sweep(self, targets: List[NeighborSnapshot]) -> Dict[str, ScopeResult]: + """Query every target's scopes, one request in flight at a time. + + Returns a result per target keyed by lowercase pubkey hex. Targets that + are never reached (sweep budget or duty-cycle backlog) come back as + ``timeout``, matching the firmware's treatment of unsent entries. + """ + results: Dict[str, ScopeResult] = {} + if not targets: + return results + + if self._sweep_lock.locked(): + raise RuntimeError("neighbor scope sweep already active") + + async with self._sweep_lock: + deadline = time.monotonic() + self._max_sweep_seconds + timeout = self.response_timeout() + logger.info( + "Scope sweep starting: %d neighbour(s), %.1fs response window, %.0fs sweep budget", + len(targets), + timeout, + self._max_sweep_seconds, + ) + + abandoned = False + for target in targets: + if abandoned or time.monotonic() >= deadline: + if not abandoned: + logger.warning( + "Scope sweep budget exhausted; %s and later marked timeout", + target.pubkey[:8], + ) + abandoned = True + results[target.pubkey] = ScopeResult(STATUS_TIMEOUT) + continue + + result = await self._query_one(target, timeout) + results[target.pubkey] = result + + if result.status == STATUS_SEND_FAILED and self._duty_cycle_backlogged(): + # The airtime budget is gone; the rest of the sweep would sit + # behind forwarding traffic. Publish what resolved instead. + logger.warning("Scope sweep abandoned: duty-cycle budget exhausted") + abandoned = True + + responded = sum(1 for r in results.values() if r.status == STATUS_RESPONDED) + logger.info("Scope sweep complete: %d/%d responded", responded, len(results)) + return results + + async def _query_one(self, target: NeighborSnapshot, timeout: float) -> ScopeResult: + if not self.packet_injector: + logger.warning("No packet injector available - cannot query neighbour scopes") + return ScopeResult(STATUS_SEND_FAILED) + + try: + packet, tag = self._build_request(target) + except Exception as e: + logger.warning(f"Could not build scope request for {target.pubkey[:8]}: {e}") + return ScopeResult(STATUS_SEND_FAILED) + + if not self._duty_cycle_allows(packet): + logger.warning( + "Skipping scope query for %s: duty-cycle budget exhausted", target.pubkey[:8] + ) + return ScopeResult(STATUS_SEND_FAILED) + + loop = asyncio.get_running_loop() + pending = _PendingQuery( + pubkey=target.pubkey, + tag=tag, + src_hash=target.pubkey_bytes[0], + future=loop.create_future(), + ) + self._pending = pending + + try: + # Resolves only once the packet is actually on air (or has failed): + # the engine awaits dispatcher.send_packet under its TX lock and + # defers local TX until the duty cycle allows. This is the firmware's + # logTx / logTxFail boundary, which is where the response deadline is + # armed -- hence the wait_for below, and not a moment earlier. + sent = await self.packet_injector(packet, wait_for_ack=False) + except Exception as e: + self._pending = None + logger.warning(f"Scope request send failed for {target.pubkey[:8]}: {e}") + return ScopeResult(STATUS_SEND_FAILED) + + if not sent: + self._pending = None + logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}") + return ScopeResult(STATUS_SEND_FAILED) + + try: + scopes = await asyncio.wait_for(pending.future, timeout) + logger.debug(f"Scope response from {target.pubkey[:8]}: '{scopes}'") + return ScopeResult(STATUS_RESPONDED, scopes) + except asyncio.TimeoutError: + logger.debug(f"Scope query timed out for {target.pubkey[:8]}") + return ScopeResult(STATUS_TIMEOUT) + finally: + self._pending = None + + def _build_request(self, target: NeighborSnapshot): + from openhop_core.protocol.packet_builder import PacketBuilder + + # tag(4) is prepended by create_anon_request as the request timestamp; + # ANON_REQ_TYPE_REGIONS selects the scopes reply and 0x00 asks for a + # zero-hop reply path (firmware inner[4], inner[5]). + return PacketBuilder.create_anon_request( + _ScopeTarget(target.pubkey), + self.local_identity, + req_data=bytes([ANON_REQ_TYPE_REGIONS, 0x00]), + ) + + # ------------------------------------------------------------------ + # Duty cycle + # ------------------------------------------------------------------ + def _duty_cycle_allows(self, packet) -> bool: + """Pre-flight the airtime budget. + + Stands in for the firmware's ``getFreeCount() >= 5`` packet-pool guard: + different scarce resource, same job of refusing to enqueue a request the + transmit path cannot honour. A short backlog is fine — the engine defers + local TX on its own — so only a long one is treated as a failure. + """ + if not self.airtime_manager: + return True + try: + airtime_ms = self.airtime_manager.calculate_airtime(packet.get_raw_length()) + can_tx, wait_time = self.airtime_manager.can_transmit(airtime_ms) + except Exception as e: + logger.debug(f"Duty-cycle pre-flight failed, allowing send: {e}") + return True + return can_tx or wait_time <= self._duty_cycle_abort_seconds + + def _duty_cycle_backlogged(self) -> bool: + if not self.airtime_manager: + return False + try: + airtime_ms = self.airtime_manager.calculate_airtime(MAX_PACKET_PAYLOAD) + can_tx, wait_time = self.airtime_manager.can_transmit(airtime_ms) + except Exception: + return False + return not can_tx and wait_time > self._duty_cycle_abort_seconds + + # ------------------------------------------------------------------ + # Response matching + # ------------------------------------------------------------------ + async def process_response_packet(self, packet) -> bool: + """Consume a PAYLOAD_TYPE_RESPONSE that answers the in-flight query. + + Returns True only when the packet decrypts under the pending neighbour's + shared secret *and* echoes its tag, so an unrelated response still falls + through to the companion bridges. Like firmware + ``handleNeighborDiscoverResponse``, a reply that arrives after its entry + has timed out is not accepted. + """ + pending = self._pending + if pending is None or pending.future is None or pending.future.done(): + return False + + try: + if packet.get_payload_type() != PAYLOAD_TYPE_RESPONSE: + return False + + payload = bytes(getattr(packet, "payload", b"") or b"") + if len(payload) < 3: + return False + + our_hash = self.local_identity.get_public_key()[0] + if payload[0] != our_hash or payload[1] != pending.src_hash: + return False + + peer = Identity(bytes.fromhex(pending.pubkey)) + shared_secret = peer.calc_shared_secret(self.local_identity.get_private_key()) + plaintext = CryptoUtils.mac_then_decrypt(shared_secret[:16], shared_secret, payload[2:]) + if not plaintext or len(plaintext) < 8: + return False + + # plaintext: tag(4) + responder clock(4) + comma-separated scope names + if int.from_bytes(plaintext[:4], "little") != pending.tag: + return False + + scopes = bytes(plaintext[8:]).decode("utf-8", errors="replace").rstrip("\x00").strip() + if not pending.future.done(): + pending.future.set_result(scopes) + return True + + except Exception as e: + logger.debug(f"Error matching scope response: {e}") + return False + + +def _as_float(value, default: float) -> float: + try: + if value is None: + return default + return float(value) + except (TypeError, ValueError): + return default diff --git a/repeater/main.py b/repeater/main.py index 53102ac..5c284b6 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -31,6 +31,7 @@ from repeater.handler_helpers import ( AdvertHelper, DiscoveryHelper, LoginHelper, + NeighborScopeHelper, PathHelper, ProtocolRequestHelper, TextHelper, @@ -38,6 +39,7 @@ from repeater.handler_helpers import ( ) from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec from repeater.logging_utils import normalize_log_level +from repeater.neighbors_publisher import NeighborsPublisher from repeater.packet_router import PacketRouter from repeater.region_map_builder import build_region_map from repeater.sensors import SensorManager @@ -99,6 +101,8 @@ class RepeaterDaemon: self.trace_helper = None self.advert_helper = None self.discovery_helper = None + self.neighbor_scope_helper = None + self.neighbors_publisher = None self.login_helper = None self.text_helper = None self.path_helper = None @@ -654,6 +658,35 @@ class RepeaterDaemon: # 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 + # Neighbour scope discovery + periodic MQTT neighbors publication. + # Created unconditionally and self-gating: the publisher idles unless + # the master switch is on and some broker opted in with neighbors:true. + self.neighbor_scope_helper = NeighborScopeHelper( + local_identity=self.local_identity, + packet_injector=self.router.inject_packet, + airtime_manager=( + getattr(self.repeater_handler, "airtime_mgr", None) + if self.repeater_handler + else None + ), + config=self.config, + ) + self.neighbors_publisher = NeighborsPublisher( + config=self.config, + local_identity=self.local_identity, + discovery_helper=self.discovery_helper, + scope_helper=self.neighbor_scope_helper, + mqtt_handler_provider=lambda: getattr( + getattr(self.repeater_handler, "storage", None), "mqtt_handler", None + ), + storage_provider=lambda: getattr(self.repeater_handler, "storage", None), + self_scopes_fn=( + self.login_helper._format_region_names if self.login_helper else None + ), + ) + self.neighbors_publisher.start() + logger.info("Neighbors publisher initialized") + # Optional pyMC_Glass integration loop (inform/control plane) self.glass_handler = GlassHandler( config=self.config, @@ -1632,6 +1665,13 @@ class RepeaterDaemon: "http server", asyncio.to_thread(self.http_server.stop), timeout=3 ) + # Stop the neighbours publication loop. + if self.neighbors_publisher: + try: + await self.neighbors_publisher.stop() + except Exception as e: + logger.warning(f"Error stopping neighbors publisher: {e}") + # Stop Glass inform loop if self.glass_handler: await self._shutdown_step("glass handler", self.glass_handler.stop()) diff --git a/repeater/neighbors_publisher.py b/repeater/neighbors_publisher.py new file mode 100644 index 0000000..dc38d40 --- /dev/null +++ b/repeater/neighbors_publisher.py @@ -0,0 +1,559 @@ +"""Periodic neighbours publication for the MQTT ``neighbors`` topic. + +Python port of the firmware's two-stage neighbours cycle +(MeshCore ``simple_repeater/MyMesh.cpp``, ``WITH_MQTT_NEIGHBORS``): + +* **Stage 1** — a zero-hop node-discovery broadcast refreshes the neighbour table. + Firmware calls ``sendNodeDiscoverReq()`` and waits out its collection window; + here that is a :class:`~repeater.handler_helpers.discovery.DiscoveryHelper` + session, whose results are persisted by the same enricher the ``discover.neighbors`` + CLI command uses. +* **Stage 2** — one anon-regions scope query per neighbour, issued serially by + :class:`~repeater.handler_helpers.neighbor_scopes.NeighborScopeHelper`. +* **Publish** — the assembled table goes to every enabled broker that opted in + with ``neighbors: true``. + +The published table is the whole zero-hop neighbour snapshot, including entries +that did not answer the scope query (``timeout`` / ``send_failed``), matching the +firmware payload. Unlike the firmware there is no 10 KB cap: the ESP32 needed a +fixed PSRAM buffer, a Linux host does not. +""" + +import asyncio +import logging +import time +from datetime import datetime, timezone +from typing import Any, Callable, Dict, List, Optional + +from repeater.handler_helpers.neighbor_scopes import ( + STATUS_RESPONDED, + STATUS_TIMEOUT, + NeighborSnapshot, + ScopeResult, +) + +logger = logging.getLogger("NeighborsPublisher") + +# Firmware parity: mqtt.neighbors.interval accepts 12-336 hours, default 24, and +# rejects out-of-range values rather than clamping them. +MIN_INTERVAL_HOURS = 12 +MAX_INTERVAL_HOURS = 336 +DEFAULT_INTERVAL_HOURS = 24 + +# Firmware waits out the 60 s node-discover collection window before querying scopes. +DEFAULT_DISCOVERY_TIMEOUT_SECONDS = 60.0 + +# Firmware bounds the table by MAX_NEIGHBOURS; this is the openhop equivalent so a +# large advert history cannot turn into an unbounded sweep. +DEFAULT_MAX_NEIGHBORS = 32 + +# Zero-hop rows older than this are treated as gone and left out of the pass. +DEFAULT_MAX_NEIGHBOR_AGE_SECONDS = 86400.0 + +# How often the loop wakes to re-evaluate its schedule. +_TICK_SECONDS = 30.0 + +# Phases reported by status(), mirroring the firmware's NeighborsPhase. +PHASE_DISABLED = "disabled" +PHASE_SCHEDULED = "scheduled" +PHASE_ACTIVE = "active" +PHASE_DUE = "due" + + +def build_neighbors_payload( + *, + origin: str, + origin_id: str, + self_scopes: str, + entries: List[dict], + timestamp: Optional[str] = None, +) -> dict: + """Assemble the ``neighbors`` topic payload. + + ``entries`` are ordered most- to least-useful (most recently heard first, then + stronger SNR, then pubkey) exactly as the firmware orders them. The firmware + needs that order so it can drop the tail when its fixed buffer fills; we keep + it because it is the documented shape of the topic and it puts the useful rows + first for consumers. + """ + ordered = sorted( + entries, + key=lambda e: (e.get("heard_secs_ago", 0), -float(e.get("snr", 0.0)), e.get("pubkey", "")), + ) + return { + "timestamp": timestamp or datetime.now(timezone.utc).isoformat(), + "origin": origin, + "origin_id": origin_id, + "self": {"scopes": self_scopes or ""}, + "neighbors": ordered, + } + + +class NeighborsPublisher: + """Owns the periodic neighbours cycle and its manual trigger.""" + + def __init__( + self, + config: dict, + *, + local_identity=None, + discovery_helper=None, + scope_helper=None, + mqtt_handler_provider: Optional[Callable[[], Any]] = None, + storage_provider: Optional[Callable[[], Any]] = None, + self_scopes_fn: Optional[Callable[[], str]] = None, + ): + self.config = config + self.local_identity = local_identity + self.discovery_helper = discovery_helper + self.scope_helper = scope_helper + self._mqtt_handler_provider = mqtt_handler_provider + self._storage_provider = storage_provider + self._self_scopes_fn = self_scopes_fn + + self._task: Optional[asyncio.Task] = None + self._running = False + self._active = False + # Discovery responses collected during the current cycle, keyed by pubkey. + self._discovery_seen: Dict[str, dict] = {} + self._next_publish_at: Optional[float] = None + self._last_result: Optional[str] = None + self._last_publish_at: Optional[float] = None + + # ------------------------------------------------------------------ + # Config accessors (re-read every cycle so live edits take effect) + # ------------------------------------------------------------------ + @property + def _neighbors_config(self) -> dict: + return (self.config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {} + + @property + def master_enabled(self) -> bool: + """Feature kill switch. Defaults on; the per-broker flags are the control.""" + return bool(self._neighbors_config.get("enabled", True)) + + @property + def interval_seconds(self) -> float: + hours = self._neighbors_config.get("interval_hours", DEFAULT_INTERVAL_HOURS) + return normalize_interval_hours(hours) * 3600.0 + + @property + def discovery_timeout(self) -> float: + try: + value = float( + self._neighbors_config.get( + "discovery_timeout_seconds", DEFAULT_DISCOVERY_TIMEOUT_SECONDS + ) + ) + except (TypeError, ValueError): + return DEFAULT_DISCOVERY_TIMEOUT_SECONDS + return max(1.0, value) + + @property + def max_neighbors(self) -> int: + try: + value = int(self._neighbors_config.get("max_neighbors", DEFAULT_MAX_NEIGHBORS)) + except (TypeError, ValueError): + return DEFAULT_MAX_NEIGHBORS + return max(1, value) + + @property + def max_neighbor_age(self) -> float: + try: + value = float( + self._neighbors_config.get( + "max_neighbor_age_seconds", DEFAULT_MAX_NEIGHBOR_AGE_SECONDS + ) + ) + except (TypeError, ValueError): + return DEFAULT_MAX_NEIGHBOR_AGE_SECONDS + return max(1.0, value) + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + def start(self) -> None: + if self._task is not None and not self._task.done(): + return + self._running = True + self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher") + logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0) + + async def stop(self) -> None: + self._running = False + task = self._task + self._task = None + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception as e: + logger.debug(f"Neighbors publisher shutdown error ignored: {e}") + logger.info("Neighbors publisher stopped") + + def status(self) -> dict: + if not self.enabled(): + phase = PHASE_DISABLED + elif self._active: + phase = PHASE_ACTIVE + elif self._next_publish_at is None or time.monotonic() >= self._next_publish_at: + phase = PHASE_DUE + else: + phase = PHASE_SCHEDULED + + secs_until_next = None + if phase == PHASE_SCHEDULED and self._next_publish_at is not None: + secs_until_next = max(0, int(self._next_publish_at - time.monotonic())) + + return { + "phase": phase, + "secs_until_next": secs_until_next, + "last_result": self._last_result, + "last_publish_at": self._last_publish_at, + "interval_hours": self.interval_seconds / 3600.0, + } + + def enabled(self) -> bool: + """True when the master switch is on and some broker opted in.""" + if not self.master_enabled: + return False + handler = self._mqtt_handler() + return bool(handler and handler.has_neighbors_brokers()) + + def _mqtt_handler(self): + if not self._mqtt_handler_provider: + return None + try: + return self._mqtt_handler_provider() + except Exception as e: + logger.debug(f"MQTT handler unavailable: {e}") + return None + + def _storage(self): + if not self._storage_provider: + return None + try: + return self._storage_provider() + except Exception as e: + logger.debug(f"Storage unavailable: {e}") + return None + + def _local_pubkey_hex(self) -> str: + if not self.local_identity: + return "" + try: + return self.local_identity.get_public_key().hex().lower() + except Exception: + return "" + + # ------------------------------------------------------------------ + # Loop + # ------------------------------------------------------------------ + async def _run_loop(self) -> None: + try: + while self._running: + try: + await self._tick() + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"Neighbors publisher cycle failed: {e}", exc_info=True) + self._last_result = f"error: {e}" + self._reschedule() + await asyncio.sleep(_TICK_SECONDS) + except asyncio.CancelledError: + logger.debug("Neighbors publisher loop cancelled") + raise + + async def _tick(self) -> None: + if not self.enabled(): + # Drop the schedule so re-enabling runs a pass promptly, matching the + # firmware's next_neighbors_publish = 0 reset. + self._next_publish_at = None + return + + handler = self._mqtt_handler() + if not handler or not handler.has_connected_neighbors_brokers(): + logger.debug("Neighbors publish deferred: no connected opted-in broker") + return + + now = time.monotonic() + if self._next_publish_at is not None and now < self._next_publish_at: + return + + await self.run_cycle(trigger="periodic") + + def _reschedule(self) -> None: + self._next_publish_at = time.monotonic() + self.interval_seconds + + # ------------------------------------------------------------------ + # Cycle + # ------------------------------------------------------------------ + async def run_cycle(self, trigger: str = "manual") -> dict: + """Run one full discovery + scopes + publish cycle.""" + if self._active: + return {"success": False, "error": "neighbors cycle already active"} + + self._active = True + started = time.monotonic() + self._discovery_seen = {} + try: + await self._refresh_neighbor_table() + targets = self._snapshot_neighbors() + + scope_results: Dict[str, ScopeResult] = {} + if targets and self.scope_helper: + scope_results = await self.scope_helper.sweep(targets) + elif targets: + logger.warning("No scope helper available; publishing without scopes") + + payload = self._build_payload(targets, scope_results) + published = self._publish(payload) + + responded = sum(1 for r in scope_results.values() if r.status == STATUS_RESPONDED) + self._last_result = ( + f"ok ({len(targets)} neighbours, {responded} with scopes)" + if published + else "publish failed (no connected broker)" + ) + self._last_publish_at = time.time() + logger.info( + "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes", + trigger, + time.monotonic() - started, + len(targets), + responded, + ) + return { + "success": True, + "neighbors": len(targets), + "responded": responded, + "published": published, + } + finally: + self._active = False + self._reschedule() + + async def _refresh_neighbor_table(self) -> None: + """Stage 1: zero-hop node discovery, awaited to completion. + + ``prefix_only=False`` is required, not cosmetic: the scope query needs the + neighbour's full 32-byte public key to derive a shared secret. + """ + if not self.discovery_helper: + logger.debug("No discovery helper; using the stored neighbour table as-is") + return + + try: + self.discovery_helper.cleanup_sessions() + session = self.discovery_helper.create_session( + timeout=self.discovery_timeout, + filter_mask=(1 << 2), # repeaters + since=0, + prefix_only=False, + result_enricher=self._enrich_discovery_result, + ) + await self.discovery_helper.execute_session(session["session_id"]) + except Exception as e: + logger.warning(f"Neighbour table refresh failed, using stored table: {e}") + + def _enrich_discovery_result(self, result: dict) -> dict: + """Record each discovery response for this cycle, and persist it. + + This is what makes stage 1 matter: without it the snapshot would report + whatever ``last_seen``/``snr`` the advert history happened to hold, rather + than what answered just now. Mirrors firmware ``putNeighbour()`` on a + discovery response. + + The in-memory copy is not redundant with the write. ``get_neighbors()`` + serves a 60 s cache that ``store_advert()`` does not invalidate, so a + neighbour that answered seconds ago can be missing from, or stale in, the + table read that follows. The snapshot merges this dict over that read. + """ + pubkey = str(result.get("pub_key") or "").strip().lower() + if not pubkey or len(pubkey) != 64: + return result + + local_pubkey = self._local_pubkey_hex() + if local_pubkey and pubkey == local_pubkey: + return result + + node_type_raw = int(result.get("node_type", 0) or 0) + snr_raw = result.get("response_snr", result.get("snr")) + if node_type_raw == 2: + self._discovery_seen[pubkey] = { + "last_seen": time.time(), + "snr": float(snr_raw) if snr_raw is not None else 0.0, + } + + storage = self._storage() + record_advert = getattr(storage, "record_advert", None) if storage else None + if not callable(record_advert): + return result + + node_type = int(result.get("node_type", 0) or 0) + rssi = result.get("rssi") + snr = result.get("response_snr", result.get("snr")) + try: + record_advert( + { + "timestamp": time.time(), + "pubkey": pubkey, + "node_name": result.get("node_name"), + "is_repeater": node_type == 2, + "route_type": 2, + "contact_type": {1: "Chat Node", 2: "Repeater", 3: "Room Server"}.get( + node_type, "Unknown" + ), + "latitude": None, + "longitude": None, + "rssi": int(rssi) if rssi is not None else None, + "snr": float(snr) if snr is not None else None, + "is_new_neighbor": True, + "zero_hop": True, + } + ) + except Exception as e: + logger.debug(f"Could not persist discovery result for {pubkey[:8]}: {e}") + + return result + + def _snapshot_neighbors(self) -> List[NeighborSnapshot]: + """Freeze the zero-hop repeater table for this pass. + + Taken once, up front: the sweep can run for minutes and the table keeps + changing underneath it (firmware hit the same problem and stopped indexing + into its live ``neighbours[]`` in ``aba571ed``). + + Sources are the stored zero-hop repeater table (which also carries + neighbours heard via advert but silent during discovery, as the firmware's + table does) merged with this cycle's discovery responses, which win on + ``last_seen``/``snr`` because they are first-hand and the table read is + served from a cache the advert write does not invalidate. + """ + storage = self._storage() + neighbors = {} + if storage: + try: + neighbors = storage.get_neighbors() or {} + except Exception as e: + logger.warning(f"Could not read neighbour table: {e}") + + local_pubkey = self._local_pubkey_hex() + now = time.time() + max_age = self.max_neighbor_age + + merged: Dict[str, dict] = {} + for pubkey, info in neighbors.items(): + if not info.get("is_repeater") or not info.get("zero_hop"): + continue + key = str(pubkey or "").lower() + if len(key) != 64: + # Scope queries need the full key for ECDH; a prefix cannot be used. + continue + last_seen = float(info.get("last_seen") or 0.0) + if last_seen <= 0 or (now - last_seen) > max_age: + continue + merged[key] = {"last_seen": last_seen, "snr": float(info.get("snr") or 0.0)} + + for key, seen in (self._discovery_seen or {}).items(): + merged[key] = dict(seen) + + snapshots = [ + NeighborSnapshot(pubkey=key, last_seen=info["last_seen"], snr=info["snr"]) + for key, info in merged.items() + if not (local_pubkey and key == local_pubkey) + ] + + # Freshest first, then strongest, then pubkey for a deterministic order -- + # the same ordering the firmware applies before it starts querying. + snapshots.sort(key=lambda s: (-s.last_seen, -s.snr, s.pubkey)) + if len(snapshots) > self.max_neighbors: + logger.info( + "Neighbour table has %d entries; querying the freshest %d", + len(snapshots), + self.max_neighbors, + ) + snapshots = snapshots[: self.max_neighbors] + return snapshots + + def _build_payload( + self, targets: List[NeighborSnapshot], scope_results: Dict[str, ScopeResult] + ) -> dict: + now = time.time() + entries = [] + for target in targets: + result = scope_results.get(target.pubkey) or ScopeResult(STATUS_TIMEOUT) + heard_secs_ago = int(max(0.0, now - target.last_seen)) if target.last_seen else 0 + entries.append( + { + "pubkey": target.pubkey, + "snr": round(target.snr, 2), + "heard_secs_ago": heard_secs_ago, + "scopes": result.scopes or "", + "status": result.status, + } + ) + + handler = self._mqtt_handler() + origin = getattr(handler, "node_name", "") if handler else "" + origin_id = getattr(handler, "public_key", "") if handler else "" + if not origin: + origin = self.config.get("repeater", {}).get("node_name", "openHop-Repeater") + if not origin_id and self.local_identity: + try: + origin_id = self.local_identity.get_public_key().hex().upper() + except Exception: + origin_id = "" + + return build_neighbors_payload( + origin=origin, + origin_id=origin_id, + self_scopes=self._self_scopes(), + entries=entries, + ) + + def _self_scopes(self) -> str: + if not self._self_scopes_fn: + return "" + try: + return self._self_scopes_fn() or "" + except Exception as e: + logger.debug(f"Could not read local scopes: {e}") + return "" + + def _publish(self, payload: dict) -> bool: + handler = self._mqtt_handler() + if not handler: + return False + try: + results = handler.publish_neighbors(payload) + except Exception as e: + logger.error(f"Neighbors publish failed: {e}") + return False + return bool(results) + + +def normalize_interval_hours(value) -> float: + """Validate an interval in hours, falling back to the default when invalid. + + Firmware rejects out-of-range values instead of clamping them; the API layer + surfaces that rejection to the user, and this fallback keeps a hand-edited + config.yaml from producing a nonsense schedule. + """ + try: + hours = float(value) + except (TypeError, ValueError): + return float(DEFAULT_INTERVAL_HOURS) + if hours < MIN_INTERVAL_HOURS or hours > MAX_INTERVAL_HOURS: + logger.warning( + "mqtt_brokers.neighbors.interval_hours=%s outside %d-%d; using %d", + value, + MIN_INTERVAL_HOURS, + MAX_INTERVAL_HOURS, + DEFAULT_INTERVAL_HOURS, + ) + return float(DEFAULT_INTERVAL_HOURS) + return hours diff --git a/repeater/packet_router.py b/repeater/packet_router.py index b4ed78c..a6ecc54 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -713,7 +713,28 @@ class PacketRouter: # to first hop instead of original requester). consumed = False dest_hash = packet.payload[0] if packet.payload and len(packet.payload) >= 1 else None - companion_bridges = self._companion_bridges_for_packet(packet, metadata) + + # A neighbour's answer to our own anon-regions scope query is addressed + # to this repeater's identity, and no companion bridge owns it. Offer it + # to the scope helper first; it consumes the packet only when the + # ciphertext authenticates under the pending neighbour's shared secret + # AND echoes that query's tag, so an unrelated RESPONSE still falls + # through to the companion paths below. + scope_helper = getattr(self.daemon, "neighbor_scope_helper", None) + scope_consumed = False + if scope_helper is not None: + try: + scope_consumed = await scope_helper.process_response_packet(packet) + except Exception as e: + logger.debug(f"Neighbor scope response matching failed: {e}") + + if scope_consumed: + packet.mark_do_not_retransmit() + processed_by_injection = True + self._record_for_ui(packet, metadata) + companion_bridges = {} + else: + 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: _, consumed = await self._fan_out_to_bridges( diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 8deba85..c1dbf6f 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -2258,13 +2258,25 @@ class APIEndpoints: "reconnecting": conn.has_pending_reconnect(), }, "format": conn.format, + "neighbors": getattr(conn, "neighbors_enabled", False), } ) + # Schedule summary for the neighbors topic, the openhop equivalent of + # the firmware's trailing "nbr: /" in `get mqtt.status`. + neighbors_status = None + publisher = getattr(self.daemon_instance, "neighbors_publisher", None) + if publisher: + try: + neighbors_status = publisher.status() + except Exception as exc: + logger.debug(f"mqtt_status could not read neighbors publisher: {exc}") + return self._success( { "handler_active": handler is not None, "brokers": connected_brokers, + "neighbors": neighbors_status, } ) except Exception as e: @@ -2325,6 +2337,79 @@ class APIEndpoints: logger.error(f"Error listing broker presets: {e}") return self._error(str(e)) + @staticmethod + def _validate_neighbors_settings(raw): + """Validate the ``mqtt_brokers.neighbors`` block. + + Returns ``(settings, error)``; ``error`` is None on success. The interval + is rejected rather than clamped when out of range, matching the firmware's + ``set mqtt.neighbors.interval`` behavior. + """ + from repeater.neighbors_publisher import ( + DEFAULT_INTERVAL_HOURS, + MAX_INTERVAL_HOURS, + MIN_INTERVAL_HOURS, + ) + + if not isinstance(raw, dict): + return None, "neighbors must be an object" + + settings = {} + + if "enabled" in raw: + settings["enabled"] = bool(raw["enabled"]) + + if "interval_hours" in raw: + try: + interval = int(raw["interval_hours"]) + except (TypeError, ValueError): + return None, "neighbors.interval_hours must be a number" + if interval < MIN_INTERVAL_HOURS or interval > MAX_INTERVAL_HOURS: + return ( + None, + f"neighbors.interval_hours must be between {MIN_INTERVAL_HOURS} " + f"and {MAX_INTERVAL_HOURS} (default {DEFAULT_INTERVAL_HOURS})", + ) + settings["interval_hours"] = interval + + if "discovery_timeout_seconds" in raw: + try: + timeout = float(raw["discovery_timeout_seconds"]) + except (TypeError, ValueError): + return None, "neighbors.discovery_timeout_seconds must be a number" + if timeout < 5 or timeout > 300: + return None, "neighbors.discovery_timeout_seconds must be between 5 and 300" + settings["discovery_timeout_seconds"] = timeout + + if "scope_response_timeout_seconds" in raw: + try: + scope_timeout = float(raw["scope_response_timeout_seconds"]) + except (TypeError, ValueError): + return None, "neighbors.scope_response_timeout_seconds must be a number" + if scope_timeout < 0 or scope_timeout > 300: + return None, "neighbors.scope_response_timeout_seconds must be between 0 and 300" + settings["scope_response_timeout_seconds"] = scope_timeout + + if "max_neighbors" in raw: + try: + max_neighbors = int(raw["max_neighbors"]) + except (TypeError, ValueError): + return None, "neighbors.max_neighbors must be a number" + if max_neighbors < 1 or max_neighbors > 255: + return None, "neighbors.max_neighbors must be between 1 and 255" + settings["max_neighbors"] = max_neighbors + + if "max_neighbor_age_seconds" in raw: + try: + max_age = float(raw["max_neighbor_age_seconds"]) + except (TypeError, ValueError): + return None, "neighbors.max_neighbor_age_seconds must be a number" + if max_age < 60: + return None, "neighbors.max_neighbor_age_seconds must be at least 60" + settings["max_neighbor_age_seconds"] = max_age + + return settings, None + @cherrypy.expose @cherrypy.tools.json_out() @cherrypy.tools.json_in() @@ -2365,6 +2450,17 @@ class APIEndpoints: mqtt_updates["owner"] = str(data["owner"]).strip() if "email" in data: mqtt_updates["email"] = str(data["email"]).strip() + if "neighbors" in data: + neighbors_settings, error = self._validate_neighbors_settings(data["neighbors"]) + if error: + return self._error(error) + # update_and_save replaces a section's key outright, so merge onto + # the stored block instead of letting a partial POST drop the + # settings it did not mention. + existing_neighbors = (self.config.get("mqtt_brokers", {}) or {}).get( + "neighbors", {} + ) or {} + mqtt_updates["neighbors"] = {**existing_neighbors, **neighbors_settings} # if "disallowed_packet_types" in data: # mqtt_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"]) if "brokers" in data: @@ -2402,6 +2498,9 @@ class APIEndpoints: "format": str(b["format"]).strip(), "disallowed_packet_types": list(b.get("disallowed_packet_types", [])), "retain_status": bool(b.get("retain_status", False)), + # Opt-in per broker; brokers that do not expect the + # neighbors topic can reject it and drop the connection. + "neighbors": bool(b.get("neighbors", False)), "tls": { "enabled": bool( b.get("tls", {}).get("enabled", True if port == 443 else False) diff --git a/tests/test_mqtt_neighbors.py b/tests/test_mqtt_neighbors.py new file mode 100644 index 0000000..af34fe1 --- /dev/null +++ b/tests/test_mqtt_neighbors.py @@ -0,0 +1,626 @@ +"""Tests for the MQTT neighbours feature. + +Covers the three pieces that carry real risk: + +* the serialized scope sweep (one query in flight, deadline armed only after the + request transmits) — the pacing the firmware had to be fixed to get right; +* response matching, which must consume a reply only when it authenticates AND + echoes the pending tag, so unrelated RESPONSE traffic still reaches companions; +* the publish gate, which must reach opted-in brokers only. +""" + +import asyncio +import json +import time +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from openhop_core.protocol import CryptoUtils, Identity, LocalIdentity +from openhop_core.protocol.constants import PAYLOAD_TYPE_RESPONSE +from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher +from repeater.handler_helpers.neighbor_scopes import ( + STATUS_RESPONDED, + STATUS_SEND_FAILED, + STATUS_TIMEOUT, + NeighborScopeHelper, + NeighborSnapshot, +) +from repeater.neighbors_publisher import ( + MAX_INTERVAL_HOURS, + MIN_INTERVAL_HOURS, + NeighborsPublisher, + build_neighbors_payload, + normalize_interval_hours, +) + + +# ==================================================================== +# Scaffolding +# ==================================================================== +class _FakePacket: + """Minimal Packet stand-in for the response-matching path.""" + + def __init__(self, payload: bytes): + self.payload = bytearray(payload) + + def get_payload_type(self): + return PAYLOAD_TYPE_RESPONSE + + def get_raw_length(self): + return len(self.payload) + 2 + + +def _make_response_packet( + responder: LocalIdentity, requester: LocalIdentity, tag: int, scopes: str +) -> _FakePacket: + """Build the reply a neighbour sends: dest_hash + src_hash + encrypted body.""" + requester_identity = Identity(requester.get_public_key()) + shared_secret = requester_identity.calc_shared_secret(responder.get_private_key()) + plaintext = tag.to_bytes(4, "little") + int(time.time()).to_bytes(4, "little") + scopes.encode() + cipher = CryptoUtils.encrypt_then_mac(shared_secret[:16], shared_secret, plaintext) + payload = bytes([requester.get_public_key()[0], responder.get_public_key()[0]]) + cipher + return _FakePacket(payload) + + +def _helper_with_injector(local_identity, injector, config=None): + return NeighborScopeHelper( + local_identity=local_identity, + packet_injector=injector, + airtime_manager=None, + config=config or {}, + ) + + +# ==================================================================== +# Payload +# ==================================================================== +def test_payload_orders_most_useful_first(): + payload = build_neighbors_payload( + origin="node", + origin_id="AA" * 32, + self_scopes="DEN,APRS", + entries=[ + {"pubkey": "cc", "snr": 3.0, "heard_secs_ago": 900, "scopes": "", "status": "timeout"}, + { + "pubkey": "aa", + "snr": 1.0, + "heard_secs_ago": 10, + "scopes": "DEN", + "status": "responded", + }, + {"pubkey": "bb", "snr": 9.0, "heard_secs_ago": 10, "scopes": "", "status": "timeout"}, + ], + ) + + assert [e["pubkey"] for e in payload["neighbors"]] == ["bb", "aa", "cc"] + assert payload["self"] == {"scopes": "DEN,APRS"} + assert payload["origin_id"] == "AA" * 32 + assert payload["timestamp"] + + +def test_payload_keeps_every_entry_regardless_of_size(): + """Unlike the firmware there is no fixed publish buffer, so nothing is dropped.""" + entries = [ + { + "pubkey": f"{i:064x}", + "snr": 1.0, + "heard_secs_ago": i, + "scopes": "SCOPE" * 20, + "status": "responded", + } + for i in range(200) + ] + payload = build_neighbors_payload( + origin="node", origin_id="AA" * 32, self_scopes="", entries=entries + ) + + assert len(payload["neighbors"]) == 200 + assert len(json.dumps(payload)) > 10240 + + +@pytest.mark.parametrize( + "value,expected", + [ + (24, 24.0), + (MIN_INTERVAL_HOURS, float(MIN_INTERVAL_HOURS)), + (MAX_INTERVAL_HOURS, float(MAX_INTERVAL_HOURS)), + (MIN_INTERVAL_HOURS - 1, 24.0), # out of range -> default, never clamped + (MAX_INTERVAL_HOURS + 1, 24.0), + ("nonsense", 24.0), + (None, 24.0), + ], +) +def test_interval_validation(value, expected): + assert normalize_interval_hours(value) == expected + + +# ==================================================================== +# Scope sweep pacing +# ==================================================================== +@pytest.mark.asyncio +async def test_sweep_keeps_exactly_one_query_in_flight(): + """The pacing guarantee: query N+1 is not sent until N has resolved.""" + local = LocalIdentity() + peers = [LocalIdentity() for _ in range(3)] + targets = [ + NeighborSnapshot(pubkey=p.get_public_key().hex(), last_seen=time.time(), snr=5.0) + for p in peers + ] + + peers_by_hex = {p.get_public_key().hex(): p for p in peers} + in_flight = 0 + max_in_flight = 0 + sent_to = [] + helper = None + + async def injector(packet, wait_for_ack=False): + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + pending = helper._pending + sent_to.append(pending.pubkey) + await asyncio.sleep(0) # yield, so an overlapping send would be observable + # The neighbour answers once the request has "transmitted". + response = _make_response_packet(peers_by_hex[pending.pubkey], local, pending.tag, "DEN") + asyncio.get_running_loop().call_soon( + lambda: asyncio.ensure_future(helper.process_response_packet(response)) + ) + in_flight -= 1 + return True + + helper = _helper_with_injector(local, injector) + results = await helper.sweep(targets) + + assert max_in_flight == 1 + assert sent_to == [t.pubkey for t in targets] + assert all(r.status == STATUS_RESPONDED for r in results.values()) + assert all(r.scopes == "DEN" for r in results.values()) + + +@pytest.mark.asyncio +async def test_sweep_marks_timeout_without_a_response(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + return True + + helper = _helper_with_injector( + local, + injector, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_TIMEOUT + assert helper._pending is None + + +@pytest.mark.asyncio +async def test_sweep_marks_send_failed_when_transmit_fails(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + return False + + helper = _helper_with_injector(local, injector) + started = time.monotonic() + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_SEND_FAILED + # No response window is opened for a packet that never transmitted. + assert time.monotonic() - started < 1.0 + assert helper._pending is None + + +@pytest.mark.asyncio +async def test_sweep_budget_marks_remaining_targets_timeout(): + local = LocalIdentity() + peers = [LocalIdentity() for _ in range(3)] + targets = [ + NeighborSnapshot(pubkey=p.get_public_key().hex(), last_seen=time.time()) for p in peers + ] + + async def injector(packet, wait_for_ack=False): + await asyncio.sleep(0.05) + return True + + helper = _helper_with_injector( + local, + injector, + config={ + "mqtt_brokers": { + "neighbors": { + "scope_response_timeout_seconds": 0.05, + "max_sweep_seconds": 0.12, + } + } + }, + ) + results = await helper.sweep(targets) + + assert len(results) == 3 + assert results[targets[-1].pubkey].status == STATUS_TIMEOUT + + +@pytest.mark.asyncio +async def test_response_timeout_scales_with_radio_settings(): + local = LocalIdentity() + + class _Airtime: + def __init__(self, airtime_ms): + self._airtime_ms = airtime_ms + + def calculate_airtime(self, payload_len): + return self._airtime_ms + + fast = NeighborScopeHelper(local, airtime_manager=_Airtime(50), config={}) + slow = NeighborScopeHelper(local, airtime_manager=_Airtime(2000), config={}) + + assert slow.response_timeout() > fast.response_timeout() + + override = NeighborScopeHelper( + local, + airtime_manager=_Airtime(2000), + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 7}}}, + ) + assert override.response_timeout() == 7 + + +# ==================================================================== +# Response matching +# ==================================================================== +@pytest.mark.asyncio +async def test_response_with_matching_tag_is_consumed(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + captured = {} + + async def injector(packet, wait_for_ack=False): + captured["tag"] = helper._pending.tag + response = _make_response_packet(peer, local, captured["tag"], "DEN,APRS") + asyncio.get_running_loop().call_soon( + lambda: asyncio.ensure_future(helper.process_response_packet(response)) + ) + return True + + helper = _helper_with_injector(local, injector) + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_RESPONDED + assert results[target.pubkey].scopes == "DEN,APRS" + + +@pytest.mark.asyncio +async def test_response_with_wrong_tag_is_not_consumed(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + bogus = _make_response_packet(peer, local, helper._pending.tag ^ 0xFFFF, "DEN") + assert await helper.process_response_packet(bogus) is False + return True + + helper = _helper_with_injector( + local, + injector, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_TIMEOUT + + +@pytest.mark.asyncio +async def test_response_from_unrelated_node_is_not_consumed(): + local = LocalIdentity() + peer = LocalIdentity() + stranger = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + # Correct tag, wrong sender: must not resolve the pending query. + foreign = _make_response_packet(stranger, local, helper._pending.tag, "DEN") + assert await helper.process_response_packet(foreign) is False + return True + + helper = _helper_with_injector( + local, + injector, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_TIMEOUT + + +@pytest.mark.asyncio +async def test_response_with_no_query_pending_is_ignored(): + local = LocalIdentity() + peer = LocalIdentity() + helper = _helper_with_injector(local, None) + + packet = _make_response_packet(peer, local, 1234, "DEN") + assert await helper.process_response_packet(packet) is False + + +# ==================================================================== +# Publish gating +# ==================================================================== +def _pusher_with_brokers(brokers): + config = { + "repeater": {"node_name": "test-node"}, + "radio": { + "spreading_factor": 8, + "bandwidth": 62500, + "coding_rate": 8, + "preamble_length": 17, + "frequency": 869618000, + }, + "mqtt_brokers": { + "iata_code": "LAX", + "status_interval": 0, + "owner": "", + "email": "", + "brokers": brokers, + }, + } + identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("AB" * 32)) + return MeshCoreToMqttPusher(local_identity=identity, config=config) + + +def _broker(name, *, neighbors=False, enabled=True, fmt="letsmesh"): + return { + "name": name, + "enabled": enabled, + "host": f"{name}.example", + "port": 1883, + "transport": "tcp", + "format": fmt, + "use_jwt_auth": False, + "neighbors": neighbors, + "tls": {"enabled": False, "insecure": False}, + } + + +def _capture(conn): + captured = [] + + def _fake_publish(topic, payload, retain=False, qos=0): + captured.append({"topic": topic, "payload": payload, "retain": retain, "qos": qos}) + return None + + conn.client = MagicMock() + conn.client.publish = _fake_publish + conn._running = True + return captured + + +def test_publish_neighbors_reaches_only_opted_in_brokers(): + pusher = _pusher_with_brokers([_broker("opted-in", neighbors=True), _broker("plain")]) + opted_in, plain = pusher.connections + opted_capture = _capture(opted_in) + plain_capture = _capture(plain) + + pusher.publish_neighbors({"neighbors": [], "self": {"scopes": ""}}) + + assert len(opted_capture) == 1 + assert plain_capture == [] + assert opted_capture[0]["topic"] == "meshcore/LAX/" + "AB" * 32 + "/neighbors" + assert opted_capture[0]["qos"] == 1 + assert opted_capture[0]["retain"] is False + + +def test_publish_neighbors_skips_disabled_broker(): + pusher = _pusher_with_brokers([_broker("off", neighbors=True, enabled=False)]) + captured = _capture(pusher.connections[0]) + + results = pusher.publish_neighbors({"neighbors": []}) + + assert captured == [] + assert results == [] + + +def test_publish_neighbors_uses_custom_base_topic_for_legacy_format(): + pusher = _pusher_with_brokers([_broker("lan", neighbors=True, fmt="mqtt")]) + captured = _capture(pusher.connections[0]) + + pusher.publish_neighbors({"neighbors": []}) + + assert captured[0]["topic"] == "meshcore/repeater/test-node/neighbors" + + +def test_broker_opt_in_flags(): + pusher = _pusher_with_brokers([_broker("plain")]) + assert pusher.has_neighbors_brokers() is False + + pusher = _pusher_with_brokers([_broker("opted-in", neighbors=True)]) + assert pusher.has_neighbors_brokers() is True + assert pusher.has_connected_neighbors_brokers() is False + + pusher.connections[0]._running = True + assert pusher.has_connected_neighbors_brokers() is True + + +# ==================================================================== +# Publisher gating and snapshotting +# ==================================================================== +def _publisher(config, *, handler=None, storage=None, **kwargs): + return NeighborsPublisher( + config=config, + mqtt_handler_provider=lambda: handler, + storage_provider=lambda: storage, + **kwargs, + ) + + +def test_publisher_disabled_without_opted_in_broker(): + handler = SimpleNamespace(has_neighbors_brokers=lambda: False) + publisher = _publisher({"mqtt_brokers": {}}, handler=handler) + + assert publisher.enabled() is False + assert publisher.status()["phase"] == "disabled" + + +def test_master_switch_overrides_broker_opt_in(): + handler = SimpleNamespace(has_neighbors_brokers=lambda: True) + config = {"mqtt_brokers": {"neighbors": {"enabled": False}}} + publisher = _publisher(config, handler=handler) + + assert publisher.master_enabled is False + assert publisher.enabled() is False + + config["mqtt_brokers"]["neighbors"]["enabled"] = True + assert publisher.enabled() is True + + +def test_snapshot_filters_to_fresh_zero_hop_repeaters(): + now = time.time() + local = LocalIdentity() + local_key = local.get_public_key().hex() + storage = SimpleNamespace( + get_neighbors=lambda: { + "aa" * 32: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 4.0}, + "bb" * 32: {"is_repeater": True, "zero_hop": False, "last_seen": now, "snr": 9.0}, + "cc" * 32: {"is_repeater": False, "zero_hop": True, "last_seen": now, "snr": 9.0}, + "dd" * 32: { + "is_repeater": True, + "zero_hop": True, + "last_seen": now - 100000, + "snr": 9.0, + }, + "ee": {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 9.0}, + local_key: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 9.0}, + } + ) + publisher = _publisher({"mqtt_brokers": {}}, storage=storage, local_identity=local) + + snapshot = publisher._snapshot_neighbors() + + # Multi-hop, non-repeater, stale, short-key and self rows are all excluded. + assert [s.pubkey for s in snapshot] == ["aa" * 32] + + +def test_snapshot_merges_this_cycle_discovery_over_the_cached_table(): + """get_neighbors() serves a 60s cache that store_advert() does not invalidate. + + A neighbour that answered discovery seconds ago must still be queried, with + its live SNR, even when the cached table has not caught up. + """ + now = time.time() + storage = SimpleNamespace( + get_neighbors=lambda: { + "aa" * 32: { + "is_repeater": True, + "zero_hop": True, + "last_seen": now - 500, + "snr": -3.0, + } + } + ) + publisher = _publisher({"mqtt_brokers": {}}, storage=storage) + + # Responses seen during this cycle: one refresh, one brand-new neighbour. + publisher._discovery_seen = { + "aa" * 32: {"last_seen": now, "snr": 8.0}, + "bb" * 32: {"last_seen": now, "snr": 6.0}, + } + + snapshot = {s.pubkey: s for s in publisher._snapshot_neighbors()} + + assert set(snapshot) == {"aa" * 32, "bb" * 32} + assert snapshot["aa" * 32].snr == 8.0 # live value wins over the cached row + assert snapshot["aa" * 32].last_seen == now + + +def test_snapshot_is_capped_and_ordered_freshest_first(): + now = time.time() + storage = SimpleNamespace( + get_neighbors=lambda: { + f"{i:064x}": { + "is_repeater": True, + "zero_hop": True, + "last_seen": now - i, + "snr": 1.0, + } + for i in range(10) + } + ) + publisher = _publisher({"mqtt_brokers": {"neighbors": {"max_neighbors": 3}}}, storage=storage) + + snapshot = publisher._snapshot_neighbors() + + assert [s.pubkey for s in snapshot] == [f"{i:064x}" for i in range(3)] + + +@pytest.mark.asyncio +async def test_cycle_publishes_table_with_unanswered_neighbors(): + """Firmware parity: the whole zero-hop table is published, timeouts included.""" + now = time.time() + published = [] + handler = SimpleNamespace( + has_neighbors_brokers=lambda: True, + has_connected_neighbors_brokers=lambda: True, + publish_neighbors=lambda payload: published.append(payload) or [("b", None)], + node_name="test-node", + public_key="AB" * 32, + ) + storage = SimpleNamespace( + get_neighbors=lambda: { + "aa" * 32: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 4.0}, + "bb" * 32: { + "is_repeater": True, + "zero_hop": True, + "last_seen": now - 30, + "snr": 2.0, + }, + } + ) + + class _Sweeper: + async def sweep(self, targets): + from repeater.handler_helpers.neighbor_scopes import ScopeResult + + return { + targets[0].pubkey: ScopeResult(STATUS_RESPONDED, "DEN"), + targets[1].pubkey: ScopeResult(STATUS_TIMEOUT), + } + + publisher = _publisher( + {"mqtt_brokers": {}}, + handler=handler, + storage=storage, + scope_helper=_Sweeper(), + self_scopes_fn=lambda: "DEN,APRS", + ) + + result = await publisher.run_cycle(trigger="test") + + assert result["success"] is True + assert result["neighbors"] == 2 + assert result["responded"] == 1 + + payload = published[0] + assert payload["self"] == {"scopes": "DEN,APRS"} + statuses = {e["pubkey"]: e["status"] for e in payload["neighbors"]} + assert statuses == {"aa" * 32: STATUS_RESPONDED, "bb" * 32: STATUS_TIMEOUT} + # A cycle always arms the next one, so a failure cannot wedge the schedule. + assert publisher.status()["phase"] == "scheduled" + + +@pytest.mark.asyncio +async def test_cycle_rejects_reentry_while_active(): + publisher = _publisher({"mqtt_brokers": {}}) + publisher._active = True + + result = await publisher.run_cycle() + + assert result["success"] is False