mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-09 18:33:14 +02:00
Merge pull request #370 from agessaman/fix/more-things
Region-scoped flood replies, relaxed identity guard, cleaner config errors
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -5,6 +5,23 @@ from typing import Any, Dict, Iterable, Optional, Tuple
|
||||
logger = logging.getLogger("IdentityManager")
|
||||
|
||||
|
||||
def _namespace_for(identity_type: str) -> str:
|
||||
"""Routing/persistence namespace a local identity occupies.
|
||||
|
||||
A one-byte dest-hash collision is only unrepresentable when both
|
||||
identities share a namespace. Companions key their runtime bridge
|
||||
(``daemon.companion_bridges[hash]``) and their SQLite state
|
||||
(``companion_prefs`` PRIMARY KEY, plus the contacts/channels/messages
|
||||
tables) by the hash byte. The repeater and every room server instead
|
||||
share the login/text/protocol helper ``handlers[hash]`` slot and the
|
||||
``room_*`` tables, so they form one "server" namespace. A companion and
|
||||
a server-side identity live in physically separate stores, and the packet
|
||||
router (``_consume_via_local_candidates``) offers a colliding packet to
|
||||
both and lets HMAC pick the owner, so they may safely share a prefix.
|
||||
"""
|
||||
return "companion" if identity_type == "companion" else "server"
|
||||
|
||||
|
||||
class IdentityConfigurationError(RuntimeError):
|
||||
"""A configured local identity cannot be represented safely."""
|
||||
|
||||
@@ -30,25 +47,30 @@ class IdentitySpec:
|
||||
class IdentityManager:
|
||||
def __init__(self, config: dict):
|
||||
self.config = config
|
||||
self.identities: Dict[int, Tuple[Any, dict, str]] = {}
|
||||
self.identities: Dict[Tuple[int, str], Tuple[Any, dict, str]] = {}
|
||||
self.named_identities: Dict[str, Tuple[Any, dict, str]] = {}
|
||||
self.registered_hashes: Dict[int, str] = {}
|
||||
self.registered_hashes: Dict[Tuple[int, str], str] = {}
|
||||
|
||||
def registration_error(self, name: str, identity) -> Optional[str]:
|
||||
def registration_error(self, name: str, identity, identity_type: str) -> Optional[str]:
|
||||
"""Return a reason this identity cannot be registered, or ``None``.
|
||||
|
||||
Local protocol routing and companion persistence are keyed by the first
|
||||
byte of the public key. A collision therefore cannot be represented
|
||||
safely, even though the full public keys differ. Names must also be
|
||||
unique because callers use them to locate the configured service.
|
||||
Local protocol routing and per-identity persistence are keyed by the
|
||||
first public-key byte *within a namespace* (see :func:`_namespace_for`).
|
||||
Two identities that share both the hash byte and the namespace cannot
|
||||
be represented safely even though their full keys differ; a companion
|
||||
and a server-side identity may share a prefix because they live in
|
||||
separate stores that the packet router disambiguates by HMAC. Names
|
||||
must also be unique because callers use them to locate the service.
|
||||
"""
|
||||
hash_byte = identity.get_public_key()[0]
|
||||
key = (hash_byte, _namespace_for(identity_type))
|
||||
|
||||
if hash_byte in self.identities:
|
||||
existing_name = self.registered_hashes.get(hash_byte, "unknown")
|
||||
if key in self.identities:
|
||||
existing_name = self.registered_hashes.get(key, "unknown")
|
||||
return (
|
||||
f"Identity '{name}' (hash=0x{hash_byte:02X}) conflicts with "
|
||||
f"existing identity '{existing_name}'"
|
||||
f"existing identity '{existing_name}'; identities of the same "
|
||||
f"class must have unique one-byte public-key prefixes"
|
||||
)
|
||||
|
||||
if name in self.named_identities:
|
||||
@@ -66,17 +88,17 @@ class IdentityManager:
|
||||
|
||||
Each spec is checked against the currently registered identities (via
|
||||
:meth:`registration_error`) and against the other specs in the batch,
|
||||
without mutating any state. Local protocol routing and companion
|
||||
persistence are keyed by the first public-key byte, so a one-byte
|
||||
prefix collision cannot be represented even though the full keys
|
||||
differ, and names must be unique because callers use them to locate
|
||||
the configured service.
|
||||
without mutating any state. A one-byte prefix collision is only
|
||||
rejected when both identities share a namespace (see
|
||||
:func:`_namespace_for`); a companion may share a prefix with a
|
||||
server-side identity. Names must be unique because callers use them to
|
||||
locate the configured service.
|
||||
"""
|
||||
batch_hashes: Dict[int, IdentitySpec] = {}
|
||||
batch_keys: Dict[Tuple[int, str], IdentitySpec] = {}
|
||||
batch_names: Dict[str, IdentitySpec] = {}
|
||||
|
||||
for spec in specs:
|
||||
error = self.registration_error(spec.name, spec.identity)
|
||||
error = self.registration_error(spec.name, spec.identity, spec.identity_type)
|
||||
if error:
|
||||
raise IdentityConfigurationError(error)
|
||||
existing = batch_names.get(spec.name)
|
||||
@@ -85,52 +107,69 @@ class IdentityManager:
|
||||
f"Local identity name '{spec.name}' conflicts with existing "
|
||||
f"identity '{existing.label}'"
|
||||
)
|
||||
existing = batch_hashes.get(spec.hash_byte)
|
||||
key = (spec.hash_byte, _namespace_for(spec.identity_type))
|
||||
existing = batch_keys.get(key)
|
||||
if existing is not None:
|
||||
raise IdentityConfigurationError(
|
||||
f"Local identity '{spec.label}' (hash=0x{spec.hash_byte:02X}) "
|
||||
f"conflicts with '{existing.label}'; local identities must "
|
||||
"have unique one-byte public-key prefixes"
|
||||
f"conflicts with '{existing.label}'; identities of the same "
|
||||
"class must have unique one-byte public-key prefixes"
|
||||
)
|
||||
batch_names[spec.name] = spec
|
||||
batch_hashes[spec.hash_byte] = spec
|
||||
batch_keys[key] = spec
|
||||
|
||||
def validate_identity(self, name: str, identity) -> bool:
|
||||
def validate_identity(self, name: str, identity, identity_type: str) -> bool:
|
||||
"""Log and report whether an identity can be registered without mutation."""
|
||||
error = self.registration_error(name, identity)
|
||||
error = self.registration_error(name, identity, identity_type)
|
||||
if error:
|
||||
logger.error("Identity registration rejected: %s", error)
|
||||
return False
|
||||
return True
|
||||
|
||||
def register_identity(self, name: str, identity, config: dict, identity_type: str):
|
||||
if not self.validate_identity(name, identity):
|
||||
if not self.validate_identity(name, identity, identity_type):
|
||||
return False
|
||||
|
||||
hash_byte = identity.get_public_key()[0]
|
||||
key = (hash_byte, _namespace_for(identity_type))
|
||||
|
||||
self.identities[hash_byte] = (identity, config, identity_type)
|
||||
self.identities[key] = (identity, config, identity_type)
|
||||
self.named_identities[name] = (identity, config, identity_type)
|
||||
self.registered_hashes[hash_byte] = f"{identity_type}:{name}"
|
||||
self.registered_hashes[key] = f"{identity_type}:{name}"
|
||||
|
||||
logger.info(
|
||||
f"Identity registered: name={name}, hash=0x{hash_byte:02X}, type={identity_type}"
|
||||
)
|
||||
return True
|
||||
|
||||
def get_identity_by_hash(self, hash_byte: int) -> Optional[Tuple[Any, dict, str]]:
|
||||
return self.identities.get(hash_byte)
|
||||
def get_identity_by_hash(
|
||||
self, hash_byte: int, namespace: Optional[str] = None
|
||||
) -> Optional[Tuple[Any, dict, str]]:
|
||||
"""Return the identity registered at ``hash_byte``.
|
||||
|
||||
With ``namespace`` given, returns that namespace's identity; otherwise
|
||||
returns the first identity at the hash (a companion and a server-side
|
||||
identity may both be registered there since they no longer collide).
|
||||
"""
|
||||
if namespace is not None:
|
||||
return self.identities.get((hash_byte, namespace))
|
||||
for (registered_hash, _ns), value in self.identities.items():
|
||||
if registered_hash == hash_byte:
|
||||
return value
|
||||
return None
|
||||
|
||||
def get_identity_by_name(self, name: str) -> Optional[Tuple[Any, dict, str]]:
|
||||
return self.named_identities.get(name)
|
||||
|
||||
def has_identity(self, hash_byte: int) -> bool:
|
||||
return hash_byte in self.identities
|
||||
def has_identity(self, hash_byte: int, namespace: Optional[str] = None) -> bool:
|
||||
if namespace is not None:
|
||||
return (hash_byte, namespace) in self.identities
|
||||
return any(registered_hash == hash_byte for registered_hash, _ns in self.identities)
|
||||
|
||||
def list_identities(self) -> list:
|
||||
identities = []
|
||||
for hash_byte, (identity, config, id_type) in self.identities.items():
|
||||
name = self.registered_hashes.get(hash_byte, "unknown")
|
||||
for (hash_byte, namespace), (identity, config, id_type) in self.identities.items():
|
||||
name = self.registered_hashes.get((hash_byte, namespace), "unknown")
|
||||
identities.append(
|
||||
{
|
||||
"hash": f"0x{hash_byte:02X}",
|
||||
|
||||
+75
-1
@@ -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.
|
||||
@@ -962,7 +1027,7 @@ class RepeaterDaemon:
|
||||
|
||||
if self.identity_manager is None:
|
||||
raise RuntimeError("Identity manager must be initialized before adding a companion")
|
||||
registration_error = self.identity_manager.registration_error(name, identity)
|
||||
registration_error = self.identity_manager.registration_error(name, identity, "companion")
|
||||
if registration_error:
|
||||
raise ValueError(f"Cannot add companion: {registration_error}")
|
||||
|
||||
@@ -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:
|
||||
@@ -1704,6 +1773,11 @@ def main():
|
||||
asyncio.run(daemon.run())
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Repeater stopped")
|
||||
except IdentityConfigurationError as e:
|
||||
# A misconfigured local identity is an actionable config problem, not a
|
||||
# crash: report just the message so the fix is obvious, no stack trace.
|
||||
logger.error("Identity configuration error: %s", e)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
logger.error(f"Fatal error: {e}", exc_info=True)
|
||||
sys.exit(1)
|
||||
|
||||
@@ -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
|
||||
@@ -21,21 +21,41 @@ class _SeedFirstByteIdentity:
|
||||
return self._public_key[:3]
|
||||
|
||||
|
||||
def _config(*, companions=()):
|
||||
def _config(*, companions=(), room_servers=()):
|
||||
return {
|
||||
"repeater": {"node_name": "n", "identity_key": b"\x10" * 32},
|
||||
"logging": {},
|
||||
"identities": {"companions": list(companions)},
|
||||
"identities": {
|
||||
"companions": list(companions),
|
||||
"room_servers": list(room_servers),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_startup_preflight_rejects_default_repeater_hash_collision():
|
||||
def test_startup_preflight_allows_companion_sharing_repeater_prefix():
|
||||
"""A companion may share the repeater's one-byte prefix: they live in
|
||||
separate runtime stores (companion_bridges vs helper.handlers) and DB
|
||||
tables, and the packet router disambiguates them on-air by HMAC."""
|
||||
daemon = RepeaterDaemon(
|
||||
_config(companions=({"name": "comp", "identity_key": "10" * 32},)),
|
||||
radio=object(),
|
||||
)
|
||||
local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
|
||||
|
||||
with patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity):
|
||||
# Must not raise: the cross-namespace collision is representable.
|
||||
daemon._preflight_configured_local_identities(local_identity)
|
||||
|
||||
|
||||
def test_startup_preflight_rejects_repeater_room_server_prefix_collision():
|
||||
"""The repeater and a room server share the server namespace (the login/
|
||||
text helper handlers[hash] slot), so a prefix collision is still rejected."""
|
||||
daemon = RepeaterDaemon(
|
||||
_config(room_servers=({"name": "room", "identity_key": "10" * 32},)),
|
||||
radio=object(),
|
||||
)
|
||||
local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
|
||||
|
||||
with patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity):
|
||||
with pytest.raises(IdentityConfigurationError, match="one-byte public-key prefixes"):
|
||||
daemon._preflight_configured_local_identities(local_identity)
|
||||
@@ -93,10 +113,12 @@ async def test_invalid_config_entry_logs_once_across_preflight_and_load(caplog):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hot_added_companion_collision_is_rejected_before_stateful_setup():
|
||||
# A second companion sharing an already-registered companion's prefix must
|
||||
# be rejected (same companion namespace: companion_bridges + companion_* DB).
|
||||
daemon = RepeaterDaemon(_config(), radio=object())
|
||||
daemon.identity_manager = IdentityManager({})
|
||||
daemon.identity_manager.register_identity(
|
||||
"repeater", _SeedFirstByteIdentity(b"\x33" * 32), {}, "repeater"
|
||||
"existing", _SeedFirstByteIdentity(b"\x33" * 32), {}, "companion"
|
||||
)
|
||||
|
||||
comp_config = {"name": "comp", "identity_key": "33" * 32, "settings": {}}
|
||||
|
||||
@@ -34,8 +34,8 @@ def test_identity_manager_rejects_duplicate_names_without_mutating_state():
|
||||
id_b = _FakeIdentity(bytes([0x22]) + b"B" * 31)
|
||||
|
||||
assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
|
||||
assert "already registered" in mgr.registration_error("alpha", id_b)
|
||||
assert mgr.validate_identity("alpha", id_b) is False
|
||||
assert "already registered" in mgr.registration_error("alpha", id_b, "companion")
|
||||
assert mgr.validate_identity("alpha", id_b, "companion") is False
|
||||
assert mgr.get_identity_by_hash(0x22) is None
|
||||
|
||||
|
||||
@@ -63,15 +63,17 @@ def test_identity_manager_list_and_type_filtering():
|
||||
|
||||
def test_identity_manager_list_handles_none_identity_fields():
|
||||
mgr = IdentityManager(config={})
|
||||
mgr.identities[0x44] = (None, {}, "repeater")
|
||||
mgr.registered_hashes[0x44] = "repeater:ghost"
|
||||
mgr.identities[(0x44, "server")] = (None, {}, "repeater")
|
||||
mgr.registered_hashes[(0x44, "server")] = "repeater:ghost"
|
||||
|
||||
listed = mgr.list_identities()
|
||||
assert listed[0]["address"] == "N/A"
|
||||
assert listed[0]["public_key"] is None
|
||||
|
||||
|
||||
def test_validate_specs_rejects_intra_batch_hash_collision():
|
||||
def test_validate_specs_rejects_intra_batch_same_namespace_hash_collision():
|
||||
"""Two server-side identities (repeater + room server) share the login/
|
||||
text helper handlers[hash] slot, so a prefix collision is unrepresentable."""
|
||||
mgr = IdentityManager(config={})
|
||||
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
|
||||
id_b = _FakeIdentity(bytes([0x11]) + b"B" * 31)
|
||||
@@ -80,11 +82,63 @@ def test_validate_specs_rejects_intra_batch_hash_collision():
|
||||
mgr.validate_specs(
|
||||
[
|
||||
IdentitySpec("alpha", id_a, {}, "repeater"),
|
||||
IdentitySpec("beta", id_b, {}, "room_server"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def test_validate_specs_rejects_two_companions_same_prefix():
|
||||
"""Two companions share companion_bridges[hash] and the companion_* DB
|
||||
keying, so their prefix must be unique."""
|
||||
mgr = IdentityManager(config={})
|
||||
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
|
||||
id_b = _FakeIdentity(bytes([0x11]) + b"B" * 31)
|
||||
|
||||
with pytest.raises(IdentityConfigurationError, match="one-byte public-key prefixes"):
|
||||
mgr.validate_specs(
|
||||
[
|
||||
IdentitySpec("alpha", id_a, {}, "companion"),
|
||||
IdentitySpec("beta", id_b, {}, "companion"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def test_validate_specs_allows_companion_sharing_prefix_with_server_identity():
|
||||
"""A companion and a server-side identity live in separate runtime stores
|
||||
and DB tables (the router disambiguates by HMAC), so they may share a
|
||||
one-byte prefix. Verified for both the repeater and a room server."""
|
||||
mgr = IdentityManager(config={})
|
||||
repeater = _FakeIdentity(bytes([0x11]) + b"R" * 31)
|
||||
room = _FakeIdentity(bytes([0x22]) + b"S" * 31)
|
||||
comp_vs_repeater = _FakeIdentity(bytes([0x11]) + b"C" * 31)
|
||||
comp_vs_room = _FakeIdentity(bytes([0x22]) + b"D" * 31)
|
||||
|
||||
mgr.validate_specs(
|
||||
[
|
||||
IdentitySpec("rep", repeater, {}, "repeater"),
|
||||
IdentitySpec("room", room, {}, "room_server"),
|
||||
IdentitySpec("comp-a", comp_vs_repeater, {}, "companion"),
|
||||
IdentitySpec("comp-b", comp_vs_room, {}, "companion"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def test_register_companion_sharing_repeater_prefix_keeps_both():
|
||||
"""Cross-namespace registration keeps both entries addressable."""
|
||||
mgr = IdentityManager(config={})
|
||||
repeater = _FakeIdentity(bytes([0x11]) + b"R" * 31)
|
||||
companion = _FakeIdentity(bytes([0x11]) + b"C" * 31)
|
||||
|
||||
assert mgr.register_identity("rep", repeater, {}, "repeater") is True
|
||||
assert mgr.register_identity("comp", companion, {}, "companion") is True
|
||||
|
||||
assert mgr.get_identity_by_hash(0x11, "server")[0] is repeater
|
||||
assert mgr.get_identity_by_hash(0x11, "companion")[0] is companion
|
||||
assert mgr.has_identity(0x11, "server") is True
|
||||
assert mgr.has_identity(0x11, "companion") is True
|
||||
assert len(mgr.list_identities()) == 2
|
||||
|
||||
|
||||
def test_validate_specs_rejects_intra_batch_duplicate_name():
|
||||
mgr = IdentityManager(config={})
|
||||
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
|
||||
@@ -107,8 +161,9 @@ def test_validate_specs_rejects_registered_collisions_without_mutation():
|
||||
|
||||
assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
|
||||
|
||||
# A room server (server namespace) collides with the registered repeater.
|
||||
with pytest.raises(IdentityConfigurationError, match="conflicts"):
|
||||
mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "companion")])
|
||||
mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "room_server")])
|
||||
with pytest.raises(IdentityConfigurationError, match="already registered"):
|
||||
mgr.validate_specs([IdentitySpec("alpha", id_name_collision, {}, "companion")])
|
||||
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import logging
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from repeater.companion.constants import STATS_TYPE_CORE, STATS_TYPE_PACKETS, STATS_TYPE_RADIO
|
||||
from repeater.identity_manager import IdentityConfigurationError
|
||||
from repeater.main import RepeaterDaemon
|
||||
from repeater.main import main as repeater_main
|
||||
from openhop_core.node.dispatcher import Dispatcher
|
||||
@@ -533,3 +535,35 @@ def test_main_entrypoint_success_and_fatal_paths(monkeypatch):
|
||||
repeater_main()
|
||||
|
||||
exit_mock.assert_called_once_with(1)
|
||||
|
||||
|
||||
def test_main_identity_config_error_exits_cleanly_without_traceback(caplog):
|
||||
"""A configured-identity collision exits 1 with a clean message and no
|
||||
stack-trace dump (regression: the fatal handler used to log exc_info=True
|
||||
for every exception)."""
|
||||
|
||||
class _Args:
|
||||
config = "/tmp/test.yaml"
|
||||
log_level = None
|
||||
|
||||
fake_daemon = SimpleNamespace(run=MagicMock(return_value=object()))
|
||||
err = IdentityConfigurationError(
|
||||
"Local identity 'companion:B' (hash=0x77) conflicts with 'companion:A'"
|
||||
)
|
||||
|
||||
with (
|
||||
patch("argparse.ArgumentParser.parse_args", return_value=_Args()),
|
||||
patch("repeater.main.load_config", return_value=_base_config()),
|
||||
patch("repeater.main.RepeaterDaemon", return_value=fake_daemon),
|
||||
patch("asyncio.run", side_effect=err),
|
||||
patch("sys.exit", side_effect=SystemExit(1)) as exit_mock,
|
||||
caplog.at_level(logging.ERROR),
|
||||
):
|
||||
with pytest.raises(SystemExit):
|
||||
repeater_main()
|
||||
|
||||
exit_mock.assert_called_once_with(1)
|
||||
config_errors = [r for r in caplog.records if "Identity configuration error" in r.getMessage()]
|
||||
assert len(config_errors) == 1
|
||||
assert config_errors[0].exc_info is None # no traceback attached
|
||||
assert not any("Fatal error" in r.getMessage() for r in caplog.records)
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user