diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 4d8c4e7..6127f7b 100644 --- a/repeater/web/api_endpoints.py +++ b/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() diff --git a/tests/test_acl_api.py b/tests/test_acl_api.py index 808db7b..c7218d3 100644 --- a/tests/test_acl_api.py +++ b/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", [