fix(repeater): relax identity prefix guard to same-namespace collisions

Local identities occupy two routing/persistence namespaces: companions
(companion_bridges[hash] plus the companion_* tables keyed by the hash byte)
and server-side identities (the repeater and every room server, which share
the login/text/protocol helper handlers[hash] slot and the room_* tables). A
one-byte prefix collision is only unrepresentable when both identities share a
namespace; a companion and a server-side identity live in physically separate
stores, and the packet router (_consume_via_local_candidates) already offers a
colliding packet to both and lets HMAC pick the owner.

Key IdentityManager state by (hash_byte, namespace) instead of the bare hash so
the guard rejects only same-namespace collisions. This keeps blocking the pairs
that actually break -- companion<->companion (bridge overwrite plus
companion_prefs PRIMARY KEY corruption) and server<->server, i.e.
repeater<->room-server and room-server<->room-server (handlers[hash] overwrite
plus room_* corruption) -- while allowing a companion to share a prefix with the
repeater or a room server, which was previously rejected despite being only a
cosmetic label clash.

repeater/identity_manager.py

@@ -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}",

repeater/main.py

@@ -1027,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}")
 

tests/test_identity_collision_preflight.py

@@ -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": {}}

tests/test_identity_manager.py

@@ -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")])