fix(regions): scope flood replies to the request's region

Build a core RegionMap from the node's served regions and wire it into
the dispatcher and every companion bridge, so a flood reply is re-scoped
to the region its request arrived under (or left plain for a wildcard /
direct request) -- matching firmware simple_repeater::sendFloodReply.
Previously replies went out plain, so a reply to a request in region B
was dropped by B-only repeaters.

The map is built once from the node-wide transport_keys table (each named
region -> RegionEntry, flags=REGION_DENY_FLOOD for deny-flood regions; the
'*' wildcard is deliberately not an entry so plain floods reply plain). A
single shared instance reaches the dispatcher and all bridges. Public
regions rely on name-hashing for their key; a stored key is carried only
when it is genuinely custom material the name would not reproduce, keeping
reply-matching aligned with the forwarding transport-code check.

Region edits at runtime (CLI, web API, Glass sync) all funnel through the
transport_keys CRUD methods, which now fire a post-commit change callback;
the daemon rebuilds the map and reassigns a fresh instance to the
dispatcher and every live bridge (atomic rebind, safe against an in-flight
find_match on the RX thread).

Requires openhop_core with Dispatcher.region_map / CompanionBridge.region_map.

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:

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:

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

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