diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index d41cabd..ef088bf 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -30,6 +30,10 @@ class SQLiteHandler: # every write, which would defeat the cache under load. self._cumulative_counts_cache = {"timestamp": 0.0, "value": None} self._cumulative_counts_ttl_sec = 3.0 + # Optional callback fired after any transport_keys (region) write, so the + # daemon can rebuild the flood-reply RegionMap. Fired after commit, in the + # writer's thread; see set_transport_keys_changed_callback. + self._transport_keys_changed_cb = None # Thread-local storage for persistent SQLite connections. # Opening a new connection on every DB call is expensive on SD-card # storage: each sqlite3.connect() call triggers file-system operations @@ -80,6 +84,26 @@ class SQLiteHandler: self._packet_stats_cache.clear() self._neighbors_cache = {"timestamp": 0.0, "value": None} + def set_transport_keys_changed_callback(self, callback) -> None: + """Register a callback fired after any transport_keys (region) write. + + The daemon uses this to rebuild the flood-reply RegionMap when a named + region is added, removed, or has its flood policy changed (CLI, web API, + or Glass sync all funnel through the create/update/delete/sync methods + below). The callback runs after the write commits, in the writer's + thread; pass ``None`` to clear it. + """ + self._transport_keys_changed_cb = callback + + def _notify_transport_keys_changed(self) -> None: + callback = self._transport_keys_changed_cb + if callback is None: + return + try: + callback() + except Exception as e: + logger.error(f"transport_keys change callback failed: {e}", exc_info=True) + def _init_database(self): try: with self._connect() as conn: @@ -2798,7 +2822,9 @@ class SQLiteHandler: current_time, ), ) - return cursor.lastrowid + new_id = cursor.lastrowid + self._notify_transport_keys_changed() + return new_id except Exception as e: logger.error(f"Failed to create transport key: {e}") return None @@ -2917,7 +2943,10 @@ class SQLiteHandler: """, params, ) - return cursor.rowcount > 0 + changed = cursor.rowcount > 0 + if changed: + self._notify_transport_keys_changed() + return changed except Exception as e: logger.error(f"Failed to update transport key: {e}") return False @@ -2926,7 +2955,10 @@ class SQLiteHandler: try: with self._connect() as conn: cursor = conn.execute("DELETE FROM transport_keys WHERE id = ?", (key_id,)) - return cursor.rowcount > 0 + changed = cursor.rowcount > 0 + if changed: + self._notify_transport_keys_changed() + return changed except Exception as e: logger.error(f"Failed to delete transport key: {e}") return False @@ -3043,6 +3075,7 @@ class SQLiteHandler: db_ids[node["node_id"]] = int(cursor.lastrowid) conn.commit() + self._notify_transport_keys_changed() return {"applied_nodes": len(ordered), "generated_keys": generated_keys} def delete_advert(self, advert_id: int) -> bool: diff --git a/repeater/main.py b/repeater/main.py index 468ddd4..2612221 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -37,6 +37,7 @@ from repeater.handler_helpers import ( from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec from repeater.logging_utils import normalize_log_level from repeater.packet_router import PacketRouter +from repeater.region_map_builder import build_region_map from repeater.sensors import SensorManager from repeater.utils_packet import create_scoped_advert_packet from repeater.web.http_server import HTTPStatsServer, _log_buffer @@ -107,6 +108,11 @@ class RepeaterDaemon: self.router = None self.companion_bridges: dict[int, object] = {} self.companion_frame_servers: list = [] + # Shared RegionMap describing the named regions this repeater serves. + # Wired into the dispatcher and every companion bridge so core can + # re-scope flood replies to the region their request arrived under + # (firmware sendFloodReply parity). Rebuilt on any transport_keys change. + self._region_map = None # Parsed once during the startup preflight; the identity loaders reuse # them so config parsing (and its warnings) does not run twice. self._room_server_specs: list[IdentitySpec] | None = None @@ -213,6 +219,56 @@ class RepeaterDaemon: manager = self.identity_manager or IdentityManager(self.config) manager.validate_specs(specs) + def _get_sqlite_handler(self): + """Return the shared SQLiteHandler, or None if storage is unavailable.""" + handler = self.repeater_handler + storage = getattr(handler, "storage", None) if handler else None + return getattr(storage, "sqlite_handler", None) if storage else None + + def _init_region_map(self) -> None: + """Build the shared RegionMap, wire it into the dispatcher, and hook rebuilds. + + Called once storage is available (right after ``repeater_handler`` is + created) and before any companion bridge is built, so the dispatcher and + every bridge share the same instance. Runtime region edits (transport_keys + CRUD from the CLI, web API, or Glass sync) fire the storage change hook, + which reruns ``refresh_region_map``. + """ + sqlite_handler = self._get_sqlite_handler() + self._region_map = build_region_map(self.config, sqlite_handler) + if self.dispatcher is not None: + self.dispatcher.region_map = self._region_map + if sqlite_handler is not None and hasattr( + sqlite_handler, "set_transport_keys_changed_callback" + ): + sqlite_handler.set_transport_keys_changed_callback(self.refresh_region_map) + logger.info( + "Region map initialized with %d served region(s)", + len(self._region_map.regions), + ) + + def refresh_region_map(self) -> None: + """Rebuild the RegionMap and reassign it to the dispatcher and all bridges. + + Fires from the storage transport_keys change hook whenever a named region + is added, removed, or has its flood policy changed. A fresh instance is + reassigned (rather than mutated in place) because this may run in a + cherrypy worker thread while ``find_match`` iterates the map on the RX + hot path in the event-loop thread: an attribute rebind is atomic under + the GIL, so an in-flight match keeps using the old, fully-built map. + New bridges pick up the current instance at creation time. + """ + new_map = build_region_map(self.config, self._get_sqlite_handler()) + self._region_map = new_map + if self.dispatcher is not None: + self.dispatcher.region_map = new_map + for bridge in list(self.companion_bridges.values()): + try: + bridge.region_map = new_map + except Exception: + logger.debug("Failed to update region map on a companion bridge", exc_info=True) + logger.info("Region map refreshed with %d served region(s)", len(new_map.regions)) + async def initialize(self): logger.info(f"Initializing repeater: {self.config['repeater']['node_name']}") @@ -386,6 +442,11 @@ class RepeaterDaemon: send_advert_func=self.send_advert, ) + # Storage now exists: build the served-region map and wire it into the + # dispatcher so flood replies are re-scoped to their request's region. + # Runs before any companion bridge is created so all share one instance. + self._init_region_map() + # Create router self.router = PacketRouter(self) await self.router.start() @@ -784,6 +845,10 @@ class RepeaterDaemon: **bridge_kwargs, ) + # Share the dispatcher's served-region map so this bridge re-scopes + # its own flood replies to the region the request arrived under. + bridge.region_map = self._region_map + # Restore persisted state (contacts/channels/messages) from SQLite. # Raises CompanionStateLoadError instead of continuing with an # empty store when persisted rows exist but cannot be loaded. @@ -1018,6 +1083,10 @@ class RepeaterDaemon: **bridge_kwargs, ) + # Share the current served-region map (hot-reload path) so this bridge + # re-scopes its flood replies to the region the request arrived under. + bridge.region_map = self._region_map + # Restore persisted state; raises CompanionStateLoadError when persisted # rows exist but cannot be loaded (hot-reload callers surface the error). if sqlite_handler: diff --git a/repeater/region_map_builder.py b/repeater/region_map_builder.py new file mode 100644 index 0000000..052d329 --- /dev/null +++ b/repeater/region_map_builder.py @@ -0,0 +1,124 @@ +"""Build a core :class:`RegionMap` from the repeater's served transport-key regions. + +Core's flood-reply scoping (``region_map.apply_reply_scope``) re-scopes a flood +reply to the region its request arrived under, mirroring firmware +``simple_repeater::sendFloodReply``. For that to engage, the dispatcher and every +companion bridge need a ``RegionMap`` describing the named regions this repeater +serves. This module builds that map from the ``transport_keys`` table — the same +source ``login.LoginHelper._format_region_names`` reads. + +Firmware-parity notes: + +- The ``*`` wildcard (unscoped flood) is deliberately **not** a region entry. A + plain FLOOD request replies plain, so ``find_match`` must return ``None`` for + it. Wildcard handling lives in ``capture_recv_region`` (route-type based), not + here, and ``mesh.unscoped_flood_allow`` never changes the map contents. +- A deny-flood region carries ``REGION_DENY_FLOOD`` so + ``find_match(mask=REGION_DENY_FLOOD)`` skips it => its request replies plain. +- The transport key is derived from the region name via ``get_auto_key_for`` — + the same derivation senders and the repeater's own outgoing floods use. An + explicit stored key is only carried when it is genuinely custom material the + name would not reproduce (a private ``$`` region, or imported key material via + Glass sync); a redundant key that disagreed with the name would silently + re-scope replies to the wrong code. +""" + +from __future__ import annotations + +import base64 +import binascii +import logging +from typing import Optional + +from openhop_core.protocol.region_map import REGION_DENY_FLOOD, RegionEntry, RegionMap +from openhop_core.protocol.transport_keys import get_auto_key_for + +logger = logging.getLogger("RepeaterRegionMap") + + +def _decode_stored_key(raw) -> Optional[bytes]: + """Decode a stored ``transport_key`` to 16 raw bytes, or ``None``. + + Keys are stored base64-encoded (see ``SQLiteHandler.generate_transport_key``); + tolerate raw bytes too. Anything that is not exactly 16 bytes is ignored, so a + corrupt or wrong-length key falls back to name hashing rather than breaking + matching. + """ + if not raw: + return None + if isinstance(raw, (bytes, bytearray)): + key = bytes(raw) + else: + try: + key = base64.b64decode(str(raw), validate=True) + except (binascii.Error, ValueError): + return None + return key if len(key) == 16 else None + + +def build_region_map(config, sqlite_handler) -> RegionMap: + """Return a :class:`RegionMap` of the named regions this repeater serves. + + ``config`` is currently unused (the ``*`` wildcard is not a map entry) but is + kept in the signature so a future config-driven region source stays a + drop-in change for every caller. + """ + region_map = RegionMap() + if sqlite_handler is None: + return region_map + + try: + records = sqlite_handler.get_transport_keys() + except Exception as exc: # defensive: never let a bad read break startup + logger.warning("Failed to read transport keys for region map: %s", exc) + return region_map + + for rec in records or []: + name = (rec.get("name") or "").strip() + # Skip empty names and the wildcard: a plain/unscoped flood replies plain, + # so find_match must not resolve it to a region. + if not name or name == "*": + continue + + flood_policy = (rec.get("flood_policy") or "deny").strip().lower() + flags = 0 if flood_policy == "allow" else REGION_DENY_FLOOD + + private_keys = None + key_bytes = _decode_stored_key(rec.get("transport_key")) + if key_bytes is not None: + if name.startswith("$"): + # Private region: core never name-hashes a "$" name, so the stored + # key is the only usable key. Without it the region matches nothing. + private_keys = [key_bytes] + else: + # Public region: rely on name hashing unless the stored key is + # genuinely custom material the name would not reproduce. + try: + derived = get_auto_key_for(name) + except ValueError: + derived = None + if derived != key_bytes: + private_keys = [key_bytes] + + try: + region_id = int(rec.get("id") or 0) + except (TypeError, ValueError): + region_id = 0 + + parent_raw = rec.get("parent_id") + try: + parent = int(parent_raw) if parent_raw is not None else 0 + except (TypeError, ValueError): + parent = 0 + + region_map.add_region( + RegionEntry( + id=region_id, + parent=parent, + flags=flags, + name=name, + private_keys=private_keys, + ) + ) + + return region_map diff --git a/tests/test_region_map_wiring.py b/tests/test_region_map_wiring.py new file mode 100644 index 0000000..6a9d1cd --- /dev/null +++ b/tests/test_region_map_wiring.py @@ -0,0 +1,292 @@ +"""Served-region map: build correctness + dispatcher/bridge wiring. + +Core re-scopes a flood reply to the region its request arrived under +(``region_map.apply_reply_scope``), but only when a ``RegionMap`` is wired onto +the dispatcher and companion bridges. This repeater builds that map from the +``transport_keys`` table (``build_region_map``), assigns one shared instance to +the dispatcher and every bridge, and rebuilds it whenever a region is added, +removed, or re-flooded via the storage change hook. + +These tests cover the repeater's contribution: the record -> RegionEntry mapping +and flood matching, the storage change hook, and the daemon wiring that keeps a +non-None map on the dispatcher and all live bridges after a runtime change. +""" + +import base64 +from types import SimpleNamespace +from unittest.mock import MagicMock + +from openhop_core.protocol.packet import Packet +from openhop_core.protocol.region_map import REGION_DENY_FLOOD +from openhop_core.protocol.transport_keys import get_auto_key_for, scope_packet + +from repeater.data_acquisition.sqlite_handler import SQLiteHandler +from repeater.main import RepeaterDaemon +from repeater.region_map_builder import build_region_map + + +class _FakeHandler: + def __init__(self, records): + self._records = records + + def get_transport_keys(self): + return self._records + + +def _b64key(name): + return base64.b64encode(get_auto_key_for(name)).decode("ascii") + + +def _scoped_flood(key, payload=b"reply-body"): + """A TRANSPORT_FLOOD packet whose transport code was hashed with ``key``.""" + pkt = Packet() + pkt.payload = bytearray(payload) + scope_packet(pkt, key) + return pkt + + +def _plain_flood(payload=b"reply-body"): + pkt = Packet() + pkt.payload = bytearray(payload) + pkt.header = 0x01 # ROUTE_TYPE_FLOOD, no transport codes + return pkt + + +# --------------------------------------------------------------------------- +# build_region_map: record -> RegionEntry mapping and flood matching +# --------------------------------------------------------------------------- +def test_allow_region_matches_scoped_flood(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 1, + "name": "#usa", + "flood_policy": "allow", + "transport_key": _b64key("#usa"), + "parent_id": None, + } + ] + ), + ) + match = rm.find_match(_scoped_flood(get_auto_key_for("#usa")), mask=REGION_DENY_FLOOD) + assert match is not None + assert match.name == "#usa" + assert match.flags == 0 + + +def test_deny_region_is_skipped_under_flood_mask(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 2, + "name": "#secret", + "flood_policy": "deny", + "transport_key": _b64key("#secret"), + "parent_id": None, + } + ] + ), + ) + # The entry exists and carries the deny flag ... + assert [r.flags for r in rm.regions] == [REGION_DENY_FLOOD] + # ... so a flood scoped to it still replies plain (find_match returns None). + assert rm.find_match(_scoped_flood(get_auto_key_for("#secret")), mask=REGION_DENY_FLOOD) is None + + +def test_wildcard_and_empty_names_are_not_entries(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 3, + "name": "*", + "flood_policy": "allow", + "transport_key": None, + "parent_id": None, + }, + { + "id": 4, + "name": "", + "flood_policy": "allow", + "transport_key": None, + "parent_id": None, + }, + { + "id": 5, + "name": " ", + "flood_policy": "allow", + "transport_key": None, + "parent_id": None, + }, + ] + ), + ) + assert rm.regions == [] + + +def test_plain_flood_never_matches(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 1, + "name": "#usa", + "flood_policy": "allow", + "transport_key": _b64key("#usa"), + "parent_id": None, + } + ] + ), + ) + assert rm.find_match(_plain_flood(), mask=REGION_DENY_FLOOD) is None + + +def test_private_region_uses_stored_key(): + custom = b"\x11" * 16 + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 7, + "name": "$vip", + "flood_policy": "allow", + "transport_key": base64.b64encode(custom).decode("ascii"), + "parent_id": None, + } + ] + ), + ) + entry = rm.regions[0] + assert entry.private_keys == [custom] + assert rm.find_match(_scoped_flood(custom), mask=REGION_DENY_FLOOD) is not None + + +def test_private_region_without_key_matches_nothing(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 8, + "name": "$vip", + "flood_policy": "allow", + "transport_key": None, + "parent_id": None, + } + ] + ), + ) + # No usable key for a "$" region -> never matches (core never name-hashes it). + assert rm.find_match(_scoped_flood(get_auto_key_for("$vip")), mask=REGION_DENY_FLOOD) is None + + +def test_public_region_with_custom_key_carries_private_key(): + custom = b"\x22" * 16 + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 9, + "name": "#usa", + "flood_policy": "allow", + "transport_key": base64.b64encode(custom).decode("ascii"), + "parent_id": None, + } + ] + ), + ) + entry = rm.regions[0] + assert entry.private_keys == [custom] # differs from name hash -> carried through + assert rm.find_match(_scoped_flood(custom), mask=REGION_DENY_FLOOD) is not None + + +def test_public_region_with_auto_key_relies_on_name_hash(): + rm = build_region_map( + {}, + _FakeHandler( + [ + { + "id": 1, + "name": "#usa", + "flood_policy": "allow", + "transport_key": _b64key("#usa"), + "parent_id": None, + } + ] + ), + ) + # Stored key equals the name hash -> not carried as an explicit private key. + assert rm.regions[0].private_keys is None + + +def test_missing_storage_yields_empty_map(): + assert build_region_map({}, None).regions == [] + + +# --------------------------------------------------------------------------- +# Storage change hook fires on transport_keys writes +# --------------------------------------------------------------------------- +def test_transport_keys_change_hook_fires_on_writes(tmp_path): + handler = SQLiteHandler(tmp_path) + cb = MagicMock() + handler.set_transport_keys_changed_callback(cb) + + key_id = handler.create_transport_key("#usa", "allow") + assert key_id is not None + assert cb.call_count == 1 + + assert handler.update_transport_key(key_id, flood_policy="deny") + assert cb.call_count == 2 + + # A no-op update (unknown id) must not fire the hook. + assert not handler.update_transport_key(999999, flood_policy="allow") + assert cb.call_count == 2 + + assert handler.delete_transport_key(key_id) + assert cb.call_count == 3 + + handler.sync_transport_keys( + [ + {"node_id": "n1", "name": "#eu", "flood_policy": "allow"}, + ] + ) + assert cb.call_count == 4 + + +# --------------------------------------------------------------------------- +# Daemon wiring: dispatcher + bridges get a shared, non-None map that a runtime +# region change refreshes for every holder. +# --------------------------------------------------------------------------- +def test_daemon_wires_and_refreshes_region_map(tmp_path): + handler = SQLiteHandler(tmp_path) + handler.create_transport_key("#usa", "allow") + + daemon = RepeaterDaemon({"logging": {}, "mesh": {}}) + daemon.repeater_handler = SimpleNamespace(storage=SimpleNamespace(sqlite_handler=handler)) + daemon.dispatcher = SimpleNamespace(region_map=None) + daemon.companion_bridges = {1: SimpleNamespace(region_map=None)} + + daemon._init_region_map() + + # Dispatcher has a non-None map with the served region. + assert daemon.dispatcher.region_map is not None + assert [r.name for r in daemon.dispatcher.region_map.regions] == ["#usa"] + + # A runtime add fires the storage hook -> refresh reaches dispatcher + bridges. + handler.create_transport_key("#eu", "allow") + assert sorted(r.name for r in daemon.dispatcher.region_map.regions) == ["#eu", "#usa"] + assert daemon.companion_bridges[1].region_map is daemon.dispatcher.region_map + + # A runtime delete also reaches every holder. + eu = next(r for r in handler.get_transport_keys() if r["name"] == "#eu") + handler.delete_transport_key(eu["id"]) + assert [r.name for r in daemon.dispatcher.region_map.regions] == ["#usa"] + assert daemon.companion_bridges[1].region_map is daemon.dispatcher.region_map