fix(repeater): validate local identity collisions

repeater/identity_manager.py

@@ -11,16 +11,46 @@ 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_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)

repeater/main.py

@@ -44,6 +44,10 @@ logger = logging.getLogger("RepeaterDaemon")
 _COMPANION_LOAD_RETRY_DELAY_SEC = 0.5
 
 
+class IdentityConfigurationError(RuntimeError):
+    """A configured local identity cannot be represented safely."""
+
+
 async def _load_companion_rows_verified(
     loader, counter, kind: str, companion_hash_str: str, name: str, **loader_kwargs
 ):
@@ -120,6 +124,116 @@ class RepeaterDaemon:
         _log_buffer.setLevel(getattr(logging, log_level))
         root_logger.addHandler(_log_buffer)
 
+    def _configured_identity_specs(self, identity_type: str) -> list[tuple]:
+        """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((name, identity, identity_config, identity_type))
+
+        return specs
+
+    def _validate_identity_specs(
+        self, specs: list[tuple], *, include_registered: bool = True
+    ) -> None:
+        """Reject duplicate local names or one-byte public-key prefixes."""
+        hashes: dict[int, str] = {}
+        names: dict[str, str] = {}
+
+        if include_registered and self.identity_manager:
+            for hash_byte, (_, _, _) in getattr(self.identity_manager, "identities", {}).items():
+                registered_name = getattr(self.identity_manager, "registered_hashes", {}).get(
+                    hash_byte, "unknown"
+                )
+                hashes[hash_byte] = registered_name
+                if ":" in registered_name:
+                    names[registered_name.split(":", 1)[1]] = registered_name
+
+            for registered_name, (_, _, identity_type) in getattr(
+                self.identity_manager, "named_identities", {}
+            ).items():
+                names.setdefault(registered_name, identity_type)
+
+        for name, identity, _, identity_type in specs:
+            label = f"{identity_type}:{name}"
+            hash_byte = identity.get_public_key()[0]
+            existing_name = names.get(name)
+            if existing_name:
+                raise IdentityConfigurationError(
+                    f"Local identity name '{name}' conflicts with existing identity "
+                    f"'{existing_name}'"
+                )
+            existing_hash = hashes.get(hash_byte)
+            if existing_hash:
+                raise IdentityConfigurationError(
+                    f"Local identity '{label}' (hash=0x{hash_byte:02X}) conflicts "
+                    f"with '{existing_hash}'; local identities must have unique "
+                    "one-byte public-key prefixes"
+                )
+            names[name] = label
+            hashes[hash_byte] = label
+
+    def _preflight_configured_local_identities(self, local_identity) -> None:
+        """Validate every configured local identity before stateful setup begins."""
+        specs = [
+            ("repeater", local_identity, self.config, "repeater"),
+            *self._configured_identity_specs("room_server"),
+            *self._configured_identity_specs("companion"),
+        ]
+        self._validate_identity_specs(specs, include_registered=False)
+
     async def initialize(self):
 
         logger.info(f"Initializing repeater: {self.config['repeater']['node_name']}")
@@ -242,11 +356,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 +370,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,44 +626,11 @@ class RepeaterDaemon:
             raise
 
     async def _load_additional_identities(self):
-        from openhop_core import LocalIdentity
+        room_specs = self._configured_identity_specs("room_server")
+        self._validate_identity_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 name, room_identity, room_config, _ in room_specs:
             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,
@@ -551,7 +645,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 +730,20 @@ 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._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._validate_identity_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 +754,9 @@ class RepeaterDaemon:
             else self.config.get("radio", {})
         )
 
-        for comp_config in companions_config:
+        for name, identity, comp_config, _ in companion_specs:
             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 +857,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 +881,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 +1001,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 +1085,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(

tests/test_companion_state_load.py

@@ -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"
@@ -187,7 +188,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,

tests/test_identity_collision_preflight.py

@@ -0,0 +1,90 @@
+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_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 == {}

tests/test_identity_manager.py

@@ -26,6 +26,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)