Merge pull request #512 from agessaman/fix/acl-target-multibyte-hash

fix(api): match ACL identity_hash on the full multi-byte hash

repeater/web/api_endpoints.py

@@ -389,6 +389,11 @@ class APIEndpoints:
 
         return None
 
+    def _hash_byte_count(self) -> int:
+        """Bytes in a node hash under the configured path_hash_mode."""
+        mode = self.config.get("mesh", {}).get("path_hash_mode", 0)
+        return {0: 1, 1: 2, 2: 3}.get(mode, 1)
+
     def _fmt_hash(self, pubkey: bytes) -> str:
         """Format a node hash as a hex string respecting the configured path_hash_mode.
 
@@ -396,8 +401,7 @@ class APIEndpoints:
         path_hash_mode 1           → 2-byte  "0x1927"
         path_hash_mode 2           → 3-byte  "0x192722"
         """
-        mode = self.config.get("mesh", {}).get("path_hash_mode", 0)
-        byte_count = {0: 1, 1: 2, 2: 3}.get(mode, 1)
+        byte_count = self._hash_byte_count()
         hex_chars = byte_count * 2
         value = int.from_bytes(bytes(pubkey[:byte_count]), "big")
         return f"0x{value:0{hex_chars}X}"
@@ -7574,14 +7578,10 @@ class APIEndpoints:
             if not self.daemon_instance or not hasattr(self.daemon_instance, "login_helper"):
                 return self._error("Login helper not available")
 
-            target_hash = None
+            target_prefix = None
             if identity_hash:
                 try:
-                    target_hash = (
-                        int(identity_hash, 16)
-                        if identity_hash.startswith("0x")
-                        else int(identity_hash)
-                    )
+                    target_prefix = self._parse_hash_prefix(identity_hash)
                 except ValueError:
                     return self._error(f"Invalid identity_hash format: {identity_hash}")
 
@@ -7602,7 +7602,7 @@ class APIEndpoints:
                 identity_pubkey = identity.get_public_key()
                 if identity_name and name != identity_name:
                     continue
-                if target_hash is not None and identity_pubkey[0] != target_hash:
+                if target_prefix is not None and not identity_pubkey.startswith(target_prefix):
                     continue
 
                 if acl.load_error:
@@ -7664,13 +7664,29 @@ class APIEndpoints:
         if identity_name:
             return [o for o in owners if o[0] == identity_name]
         if identity_hash is not None and identity_hash != "":  # 0 is a hash byte
-            identity_hash = str(identity_hash)  # JSON may send a number
-            target = (
-                int(identity_hash, 16) if identity_hash.startswith("0x") else int(identity_hash)
-            )
-            return [o for o in owners if o[2].get_public_key()[0] == target]
+            prefix = self._parse_hash_prefix(identity_hash)
+            return [o for o in owners if o[2].get_public_key().startswith(prefix)]
         return owners
 
+    def _parse_hash_prefix(self, identity_hash) -> bytes:
+        """The public-key prefix a hash names, as ``_fmt_hash`` formats it.
+
+        A "0x" string carries its width (2-byte mode lists "0x0042", not "0x42").
+        A number does not, so it takes the fewest bytes that hold it: 0-255 is
+        one hash byte, as it was before multi-byte modes. Raises ValueError for
+        anything else.
+        """
+        text = str(identity_hash).strip()  # JSON may send a number
+        if text[:2].lower() == "0x":
+            digits = text[2:]
+            if not digits:
+                raise ValueError("empty hash")
+            return bytes.fromhex(digits.zfill(len(digits) + len(digits) % 2))
+        value = int(text)
+        if value < 0:
+            raise ValueError("negative hash")
+        return value.to_bytes(max(1, (value.bit_length() + 7) // 8), "big")
+
     @cherrypy.expose
     @cherrypy.tools.json_out()
     @cherrypy.tools.json_in()

tests/test_acl_api.py

@@ -131,6 +131,78 @@ def test_remove_accepts_a_numeric_identity_hash(db, request_ctx):
     assert _post(request_ctx, api.acl_remove_client, body)["success"] is True
 
 
+@pytest.mark.parametrize("path_hash_mode", [1, 2])
+def test_remove_matches_the_multi_byte_hash_acl_clients_lists(db, request_ctx, path_hash_mode):
+    """acl_clients lists a 2- or 3-byte identity_hash; remove must accept it back."""
+    daemon = _Daemon(db, [])
+    api = _api(daemon, {"mesh": {"path_hash_mode": path_hash_mode}})
+    acl = daemon.login_helper.get_acl_by_name("repeater")
+    admin = LocalIdentity().get_public_key()
+    acl.apply_permissions(admin, 3)
+
+    request_ctx.method = "GET"
+    listed = api.acl_clients(identity_name="repeater")["data"]["clients"]
+    identity_hash = listed[0]["identity_hash"]
+    assert len(identity_hash) == 2 + 2 * (path_hash_mode + 1)
+
+    removed = _post(
+        request_ctx,
+        api.acl_remove_client,
+        {"identity_hash": identity_hash, "client_pubkey": admin.hex()},
+    )
+    assert removed["success"] is True
+    assert removed["data"]["removed_from"] == ["repeater"]
+    assert db.load_acl_entries(daemon.local_identity.get_public_key().hex()) == []
+
+
+def test_a_multi_byte_hash_does_not_select_an_identity_sharing_only_its_first_byte(db):
+    api = _api(_Daemon(db, []), {"mesh": {"path_hash_mode": 1}})
+    (owner,) = api._acl_owners()
+    key = owner[2].get_public_key()
+    other = bytes([key[0], key[1] ^ 0xFF])
+    assert api._acl_targets(None, f"0x{key[:2].hex().upper()}") == [owner]
+    assert api._acl_targets(None, f"0x{other.hex()}") == []
+    assert api._acl_targets(None, f"0X{key[:2].hex()}") == [owner]
+    # A number has no width of its own: 0-255 still names the first byte alone.
+    assert api._acl_targets(None, key[0]) == [owner]
+    assert api._acl_targets(None, str(key[0])) == [owner]
+    if key[0]:  # a zero first byte drops out of a number
+        assert api._acl_targets(None, int.from_bytes(key[:2], "big")) == [owner]
+
+
+def test_a_leading_zero_hash_byte_keeps_its_width(db):
+    api = _api(_Daemon(db, []), {"mesh": {"path_hash_mode": 1}})
+    (owner,) = api._acl_owners()
+    key = owner[2].get_public_key()
+    assert api._parse_hash_prefix("0x0042") == b"\x00\x42"
+    assert api._parse_hash_prefix("0x042") == b"\x00\x42"
+    assert api._parse_hash_prefix("0x42") == b"\x42"
+    assert api._acl_targets(None, f"0x00{key[1]:02x}") == ([owner] if key[0] == 0 else [])
+
+
+def test_acl_clients_filters_on_the_multi_byte_hash(db, request_ctx):
+    daemon = _Daemon(db, [])
+    api = _api(daemon, {"mesh": {"path_hash_mode": 1}})
+    daemon.login_helper.get_acl_by_name("repeater").apply_permissions(
+        LocalIdentity().get_public_key(), 3
+    )
+    key = daemon.local_identity.get_public_key()
+    other = bytes([key[0], key[1] ^ 0xFF])
+
+    result = api.acl_clients(identity_hash=f"0x{key[:2].hex()}")
+    assert len(result["data"]["clients"]) == 1
+    assert api.acl_clients(identity_hash=f"0x{other.hex()}")["data"]["clients"] == []
+
+
+@pytest.mark.parametrize("identity_hash", ["0x", "0xzz", "-1", "nope", 1.5, [1], True])
+def test_remove_rejects_a_malformed_identity_hash(db, request_ctx, identity_hash):
+    api = _api(_Daemon(db, []))
+    body = {"identity_hash": identity_hash, "client_pubkey": "aa" * 32}
+    result = _post(request_ctx, api.acl_remove_client, body)
+    assert result["success"] is False
+    assert "identity_hash" in result["error"]
+
+
 @pytest.mark.parametrize(
     "body, message",
     [