Merge pull request #2 from agessaman/codex/local-identity-preflight-056

Consolidate identity collision validation in IdentityManager
This commit is contained in:
Adam Gessaman
2026-07-15 00:36:56 -07:00
committed by GitHub
6 changed files with 431 additions and 86 deletions
+92 -5
View File
@@ -1,9 +1,32 @@
import logging
from typing import Any, Dict, Optional, Tuple
from dataclasses import dataclass
from typing import Any, Dict, Iterable, Optional, Tuple
logger = logging.getLogger("IdentityManager")
class IdentityConfigurationError(RuntimeError):
"""A configured local identity cannot be represented safely."""
@dataclass(frozen=True)
class IdentitySpec:
"""A parsed-but-unregistered local identity from configuration."""
name: str
identity: Any # openhop_core LocalIdentity (or compatible)
config: dict
identity_type: str # "repeater" | "room_server" | "companion"
@property
def label(self) -> str:
return f"{self.identity_type}:{self.name}"
@property
def hash_byte(self) -> int:
return self.identity.get_public_key()[0]
class IdentityManager:
def __init__(self, config: dict):
self.config = config
@@ -11,16 +34,80 @@ class IdentityManager:
self.named_identities: Dict[str, Tuple[Any, dict, str]] = {}
self.registered_hashes: Dict[int, str] = {}
def register_identity(self, name: str, identity, config: dict, identity_type: str):
def registration_error(self, name: str, identity) -> 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.
"""
hash_byte = identity.get_public_key()[0]
if hash_byte in self.identities:
existing_name = self.registered_hashes.get(hash_byte, "unknown")
logger.error(
f"Hash collision! Identity '{name}' (hash=0x{hash_byte:02X}) "
f"conflicts with existing identity '{existing_name}'"
return (
f"Identity '{name}' (hash=0x{hash_byte:02X}) conflicts with "
f"existing identity '{existing_name}'"
)
if name in self.named_identities:
existing_identity, _, existing_type = self.named_identities[name]
existing_hash = existing_identity.get_public_key()[0]
return (
f"Identity name '{name}' is already registered for "
f"{existing_type} (hash=0x{existing_hash:02X})"
)
return None
def validate_specs(self, specs: Iterable[IdentitySpec]) -> None:
"""Raise ``IdentityConfigurationError`` on any name or hash collision.
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.
"""
batch_hashes: Dict[int, IdentitySpec] = {}
batch_names: Dict[str, IdentitySpec] = {}
for spec in specs:
error = self.registration_error(spec.name, spec.identity)
if error:
raise IdentityConfigurationError(error)
existing = batch_names.get(spec.name)
if existing is not None:
raise IdentityConfigurationError(
f"Local identity name '{spec.name}' conflicts with existing "
f"identity '{existing.label}'"
)
existing = batch_hashes.get(spec.hash_byte)
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"
)
batch_names[spec.name] = spec
batch_hashes[spec.hash_byte] = spec
def validate_identity(self, name: str, identity) -> bool:
"""Log and report whether an identity can be registered without mutation."""
error = self.registration_error(name, identity)
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):
return False
hash_byte = identity.get_public_key()[0]
self.identities[hash_byte] = (identity, config, identity_type)
self.named_identities[name] = (identity, config, identity_type)
+150 -78
View File
@@ -31,7 +31,7 @@ from repeater.handler_helpers import (
TextHelper,
TraceHelper,
)
from repeater.identity_manager import IdentityManager
from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
from repeater.packet_router import PacketRouter
from repeater.sensors import SensorManager
from repeater.utils_packet import create_scoped_advert_packet
@@ -105,6 +105,10 @@ class RepeaterDaemon:
self.router = None
self.companion_bridges: dict[int, object] = {}
self.companion_frame_servers: list = []
# 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
self._companion_specs: list[IdentitySpec] | None = None
self._shutdown_started = False
self._main_task = None
self.radio_status = "unknown"
@@ -120,6 +124,93 @@ class RepeaterDaemon:
_log_buffer.setLevel(getattr(logging, log_level))
root_logger.addHandler(_log_buffer)
def _configured_identity_specs(self, identity_type: str) -> list[IdentitySpec]:
"""Build valid configured local identities without registering them.
Invalid optional room-server or companion entries retain the existing
skip-and-log behavior. Valid entries are returned for collision
validation before they can create helper, database, or TCP state.
"""
from openhop_core import LocalIdentity
config_key = {
"room_server": "room_servers",
"companion": "companions",
}[identity_type]
configs = self.config.get("identities", {}).get(config_key) or []
specs = []
for identity_config in configs:
name = identity_config.get("name")
identity_key = identity_config.get("identity_key")
label = "Companion" if identity_type == "companion" else "Room server"
if not name or not identity_key:
logger.warning("Skipping %s config: missing name or identity_key", label.lower())
continue
try:
if isinstance(identity_key, str):
key_hex = (
normalize_companion_identity_key(identity_key)
if identity_type == "companion"
else identity_key
)
identity_key_bytes = bytes.fromhex(key_hex)
elif isinstance(identity_key, bytes):
identity_key_bytes = identity_key
else:
logger.error("%s '%s' identity_key has unknown type", label, name)
continue
except ValueError as error:
logger.error("%s '%s' identity_key invalid hex: %s", label, name, error)
continue
if len(identity_key_bytes) not in (32, 64):
logger.error(
"%s '%s' identity_key must be 32 bytes (hex) or 64 bytes "
"(MeshCore firmware key)",
label,
name,
)
continue
try:
identity = LocalIdentity(seed=identity_key_bytes)
except Exception as error:
logger.error("Failed to create %s identity '%s': %s", label.lower(), name, error)
continue
specs.append(
IdentitySpec(
name=name,
identity=identity,
config=identity_config,
identity_type=identity_type,
)
)
return specs
def _preflight_configured_local_identities(self, local_identity) -> None:
"""Validate every configured local identity before stateful setup begins.
The parsed room-server and companion specs are cached so the identity
loaders reuse them instead of re-parsing the config (and re-logging
every invalid entry). Collision rules live in
``IdentityManager.validate_specs``; at this point the manager holds no
registered identities, so this is a pure batch check.
"""
self._room_server_specs = self._configured_identity_specs("room_server")
self._companion_specs = self._configured_identity_specs("companion")
specs = [
IdentitySpec("repeater", local_identity, self.config, "repeater"),
*self._room_server_specs,
*self._companion_specs,
]
manager = self.identity_manager or IdentityManager(self.config)
manager.validate_specs(specs)
async def initialize(self):
logger.info(f"Initializing repeater: {self.config['repeater']['node_name']}")
@@ -242,11 +333,11 @@ class RepeaterDaemon:
logger.info("Dispatcher initialized")
logger.info("Dispatcher dedupe enabled: %s", dedupe_enabled)
# Initialize Identity Manager for additional identities (e.g., room servers)
# Track every local identity, including the default repeater.
self.identity_manager = IdentityManager(self.config)
logger.info("Identity manager initialized")
# Set up default repeater identity (not managed by identity manager)
# Set up the default repeater identity.
identity_key = self.config.get("repeater", {}).get("identity_key")
if not identity_key:
logger.error("No identity key found in configuration. Cannot init repeater.")
@@ -256,6 +347,19 @@ class RepeaterDaemon:
self.local_identity = local_identity
self.dispatcher.local_identity = local_identity
# A one-byte public-key prefix selects local routing, companion
# bridges, and companion SQLite namespaces. Reject all configured
# collisions before helpers, databases, or companion TCP servers
# have any state to overwrite.
self._preflight_configured_local_identities(local_identity)
if not self.identity_manager.register_identity(
name="repeater",
identity=local_identity,
config=self.config,
identity_type="repeater",
):
raise IdentityConfigurationError("Failed to register repeater identity")
pubkey = local_identity.get_public_key()
self.local_hash = pubkey[0]
self.local_hash_bytes = bytes(pubkey[:3])
@@ -499,49 +603,19 @@ class RepeaterDaemon:
raise
async def _load_additional_identities(self):
from openhop_core import LocalIdentity
room_specs = self._room_server_specs
if room_specs is None:
room_specs = self._configured_identity_specs("room_server")
self.identity_manager.validate_specs(room_specs)
identities_config = self.config.get("identities", {})
# Load room server identities
room_servers = identities_config.get("room_servers") or []
for room_config in room_servers:
for spec in room_specs:
name, room_identity = spec.name, spec.identity
try:
name = room_config.get("name")
identity_key = room_config.get("identity_key")
if not name or not identity_key:
logger.warning("Skipping room server config: missing name or identity_key")
continue
# Convert identity_key to bytes if it's a hex string
if isinstance(identity_key, bytes):
identity_key_bytes = identity_key
elif isinstance(identity_key, str):
try:
identity_key_bytes = bytes.fromhex(identity_key)
if len(identity_key_bytes) not in (32, 64):
logger.error(
f"Identity key for '{name}' is invalid length: {len(identity_key_bytes)} bytes (expected 32 or 64)"
)
continue
except ValueError as e:
logger.error(f"Identity key for '{name}' is not valid hex: {e}")
continue
else:
logger.error(
f"Identity key for '{name}' has unknown type: {type(identity_key)}"
)
continue
# Create the identity
room_identity = LocalIdentity(seed=identity_key_bytes)
# Register with the manager and all helpers
success = self._register_identity_everywhere(
name=name,
identity=room_identity,
config=room_config,
config=spec.config,
identity_type="room_server",
)
@@ -551,7 +625,13 @@ class RepeaterDaemon:
f"Loaded room server '{name}': hash=0x{room_hash:02x}, "
f"address={room_identity.get_address_bytes().hex()}"
)
else:
raise IdentityConfigurationError(
f"Failed to register room server identity '{name}'"
)
except IdentityConfigurationError:
raise
except Exception as e:
logger.error(f"Failed to load room server identity '{name}': {e}")
@@ -630,18 +710,22 @@ class RepeaterDaemon:
async def _load_companion_identities(self) -> None:
"""Load companion identities from config and create CompanionBridge + frame server for each."""
from openhop_core import LocalIdentity
from repeater.companion import CompanionFrameServer, RepeaterCompanionBridge
companions_config = self.config.get("identities", {}).get("companions") or []
if not companions_config:
companion_specs = self._companion_specs
if companion_specs is None:
companion_specs = self._configured_identity_specs("companion")
if not companion_specs:
return
# Validate the complete companion set before any bridge can restore or
# mutate a hash-keyed SQLite namespace, or any TCP server can bind.
self.identity_manager.validate_specs(companion_specs)
sqlite_handler = None
if self.repeater_handler and self.repeater_handler.storage:
sqlite_handler = self.repeater_handler.storage.sqlite_handler
if not sqlite_handler and companions_config:
if not sqlite_handler:
logger.warning(
"Companion persistence disabled: no storage (contacts/channels will not survive restart or disconnect)"
)
@@ -652,37 +736,10 @@ class RepeaterDaemon:
else self.config.get("radio", {})
)
for comp_config in companions_config:
for spec in companion_specs:
name, identity, comp_config = spec.name, spec.identity, spec.config
try:
name = comp_config.get("name")
identity_key = comp_config.get("identity_key")
settings = comp_config.get("settings") or {}
if not name or not identity_key:
logger.warning("Skipping companion config: missing name or identity_key")
continue
if isinstance(identity_key, str):
try:
identity_key_bytes = bytes.fromhex(
normalize_companion_identity_key(identity_key)
)
except ValueError as e:
logger.error(f"Companion '{name}' identity_key invalid hex: {e}")
continue
elif isinstance(identity_key, bytes):
identity_key_bytes = identity_key
else:
logger.error(f"Companion '{name}' identity_key has unknown type")
continue
if len(identity_key_bytes) not in (32, 64):
logger.error(
f"Companion '{name}' identity_key must be 32 bytes (hex) or 64 bytes (MeshCore firmware key)"
)
continue
identity = LocalIdentity(seed=identity_key_bytes)
pubkey = identity.get_public_key()
companion_hash = pubkey[0]
companion_hash_str = f"0x{companion_hash:02x}"
@@ -783,12 +840,18 @@ class RepeaterDaemon:
await frame_server.start()
self.companion_frame_servers.append(frame_server)
self.identity_manager.register_identity(
if not self.identity_manager.register_identity(
name=name,
identity=identity,
config=comp_config,
identity_type="companion",
)
):
# The complete set was prevalidated above. A failure here
# signals a concurrent/configuration error and must not be
# silently treated as a running companion.
raise IdentityConfigurationError(
f"Failed to register companion identity '{name}'"
)
limits = format_companion_bridge_limits(bridge_kwargs)
logger.info(
@@ -801,6 +864,8 @@ class RepeaterDaemon:
logger.error("%s", e)
except CompanionStateLoadError as e:
logger.error("Companion init aborted: %s", e)
except IdentityConfigurationError:
raise
except Exception as e:
logger.error(f"Failed to load companion '{name}': {e}", exc_info=True)
@@ -919,6 +984,12 @@ class RepeaterDaemon:
companion_hash = pubkey[0]
companion_hash_str = f"0x{companion_hash:02x}"
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)
if registration_error:
raise ValueError(f"Cannot add companion: {registration_error}")
if companion_hash in self.companion_bridges:
raise ValueError(f"Companion with hash 0x{companion_hash:02x} already loaded")
@@ -997,12 +1068,13 @@ class RepeaterDaemon:
await frame_server.start()
self.companion_frame_servers.append(frame_server)
self.identity_manager.register_identity(
if not self.identity_manager.register_identity(
name=name,
identity=identity,
config=comp_config,
identity_type="companion",
)
):
raise IdentityConfigurationError(f"Failed to register companion identity '{name}'")
limits = format_companion_bridge_limits(bridge_kwargs)
logger.info(
+3 -1
View File
@@ -14,6 +14,7 @@ import pytest
import repeater.main as main_module
from repeater.companion.utils import CompanionStateLoadError
from repeater.identity_manager import IdentityManager
from repeater.main import RepeaterDaemon, _load_companion_rows_verified
_HASH = "0xab"
@@ -146,6 +147,7 @@ class TestCompanionInitSurfacesLoadFailure:
},
}
daemon = RepeaterDaemon(config, radio=object())
daemon.identity_manager = IdentityManager({})
daemon.router = SimpleNamespace(inject_packet=AsyncMock())
daemon.repeater_handler = SimpleNamespace(
storage=SimpleNamespace(sqlite_handler=sqlite), radio_config={}
@@ -187,7 +189,7 @@ class TestCompanionInitSurfacesLoadFailure:
async def test_add_companion_from_config_raises(self):
sqlite = self._failing_sqlite()
daemon = self._daemon_with_companion(sqlite)
daemon.identity_manager = SimpleNamespace(named_identities={})
daemon.identity_manager = IdentityManager({})
comp_config = {"name": "hot-comp", "identity_key": "22" * 32, "settings": {}}
with (
patch("repeater.companion.RepeaterCompanionBridge") as bridge_cls,
+113
View File
@@ -0,0 +1,113 @@
import logging
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from repeater.identity_manager import IdentityManager
from repeater.main import IdentityConfigurationError, RepeaterDaemon
class _SeedFirstByteIdentity:
"""Deterministically expose a configured key's first byte as its hash."""
def __init__(self, seed: bytes):
self._public_key = bytes([seed[0]]) + b"P" * 31
def get_public_key(self):
return self._public_key
def get_address_bytes(self):
return self._public_key[:3]
def _config(*, companions=()):
return {
"repeater": {"node_name": "n", "identity_key": b"\x10" * 32},
"logging": {},
"identities": {"companions": list(companions)},
}
def test_startup_preflight_rejects_default_repeater_hash_collision():
daemon = RepeaterDaemon(
_config(companions=({"name": "comp", "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)
@pytest.mark.asyncio
async def test_companion_set_collision_is_rejected_before_bridge_or_server_creation():
daemon = RepeaterDaemon(
_config(
companions=(
{"name": "first", "identity_key": "21" * 32},
{"name": "second", "identity_key": "21" * 32},
)
),
radio=object(),
)
daemon.identity_manager = IdentityManager({})
daemon.repeater_handler = SimpleNamespace(storage=SimpleNamespace(sqlite_handler=object()))
with (
patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity),
patch("repeater.companion.RepeaterCompanionBridge") as bridge_cls,
patch("repeater.companion.CompanionFrameServer") as server_cls,
pytest.raises(IdentityConfigurationError, match="second"),
):
await daemon._load_companion_identities()
bridge_cls.assert_not_called()
server_cls.assert_not_called()
assert daemon.companion_bridges == {}
assert daemon.companion_frame_servers == []
@pytest.mark.asyncio
async def test_invalid_config_entry_logs_once_across_preflight_and_load(caplog):
"""Preflight parses the config once and the loaders reuse the cached
specs, so an invalid entry produces exactly one error per startup."""
daemon = RepeaterDaemon(
_config(companions=({"name": "bad", "identity_key": "not-hex"},)),
radio=object(),
)
daemon.identity_manager = IdentityManager({})
local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
with (
patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity),
caplog.at_level(logging.ERROR, logger="RepeaterDaemon"),
):
daemon._preflight_configured_local_identities(local_identity)
await daemon._load_companion_identities()
invalid_key_logs = [r for r in caplog.records if "invalid hex" in r.getMessage()]
assert len(invalid_key_logs) == 1
@pytest.mark.asyncio
async def test_hot_added_companion_collision_is_rejected_before_stateful_setup():
daemon = RepeaterDaemon(_config(), radio=object())
daemon.identity_manager = IdentityManager({})
daemon.identity_manager.register_identity(
"repeater", _SeedFirstByteIdentity(b"\x33" * 32), {}, "repeater"
)
comp_config = {"name": "comp", "identity_key": "33" * 32, "settings": {}}
with (
patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity),
patch("repeater.companion.RepeaterCompanionBridge") as bridge_cls,
patch("repeater.companion.CompanionFrameServer") as server_cls,
pytest.raises(ValueError, match="Cannot add companion"),
):
await daemon.add_companion_from_config(comp_config)
bridge_cls.assert_not_called()
server_cls.assert_not_called()
assert daemon.companion_bridges == {}
+71 -1
View File
@@ -1,4 +1,6 @@
from repeater.identity_manager import IdentityManager
import pytest
from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
class _FakeIdentity:
@@ -26,6 +28,17 @@ def test_identity_manager_register_lookup_and_collision_paths():
assert mgr.register_identity("beta", id_b_collision, {"k": 2}, "room_server") is False
def test_identity_manager_rejects_duplicate_names_without_mutating_state():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
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 mgr.get_identity_by_hash(0x22) is None
def test_identity_manager_list_and_type_filtering():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x22]) + b"A" * 31)
@@ -56,3 +69,60 @@ def test_identity_manager_list_handles_none_identity_fields():
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():
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, {}, "repeater"),
IdentitySpec("beta", id_b, {}, "companion"),
]
)
def test_validate_specs_rejects_intra_batch_duplicate_name():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
id_b = _FakeIdentity(bytes([0x22]) + b"B" * 31)
with pytest.raises(IdentityConfigurationError, match="repeater:alpha"):
mgr.validate_specs(
[
IdentitySpec("alpha", id_a, {}, "repeater"),
IdentitySpec("alpha", id_b, {}, "companion"),
]
)
def test_validate_specs_rejects_registered_collisions_without_mutation():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
id_hash_collision = _FakeIdentity(bytes([0x11]) + b"B" * 31)
id_name_collision = _FakeIdentity(bytes([0x22]) + b"C" * 31)
assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
with pytest.raises(IdentityConfigurationError, match="conflicts"):
mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "companion")])
with pytest.raises(IdentityConfigurationError, match="already registered"):
mgr.validate_specs([IdentitySpec("alpha", id_name_collision, {}, "companion")])
# Validation never registers anything.
assert mgr.get_identity_by_hash(0x22) is None
assert mgr.get_identity_by_name("beta") is None
def test_validate_specs_accepts_distinct_batch():
mgr = IdentityManager(config={})
mgr.validate_specs(
[
IdentitySpec("alpha", _FakeIdentity(bytes([0x11]) + b"A" * 31), {}, "repeater"),
IdentitySpec("beta", _FakeIdentity(bytes([0x22]) + b"B" * 31), {}, "room_server"),
IdentitySpec("gamma", _FakeIdentity(bytes([0x33]) + b"C" * 31), {}, "companion"),
]
)
+2 -1
View File
@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from repeater.identity_manager import IdentityManager
from repeater.main import RepeaterDaemon
@@ -43,7 +44,7 @@ async def test_load_additional_identities_valid_and_invalid_entries():
}
daemon = RepeaterDaemon(cfg, radio=object())
daemon.identity_manager = SimpleNamespace(list_identities=lambda: [1, 2])
daemon.identity_manager = IdentityManager({})
daemon._register_identity_everywhere = MagicMock(return_value=True)
with patch("openhop_core.LocalIdentity", _FakeLocalIdentity):