diff --git a/repeater/handler_helpers/acl.py b/repeater/handler_helpers/acl.py index 4fb1437..f51f5ad 100644 --- a/repeater/handler_helpers/acl.py +++ b/repeater/handler_helpers/acl.py @@ -5,12 +5,46 @@ from typing import Dict, Optional from openhop_core.protocol import Identity from openhop_core.protocol.constants import PUB_KEY_SIZE +# ACL roles come from openhop_core, which mirrors firmware +# ``src/helpers/ClientACL.h``: the role is the LOW TWO BITS of the permissions +# byte and ADMIN is 3 — it is not "the 0x02 bit". +# +# This import is deliberately fail-closed. A core without these symbols still +# builds the login reply's is_admin byte from ``permissions & 0x02``, which +# also matches READ_WRITE (2); pairing it with this module would silently +# announce a room server's read-write clients as admins. Refusing to start is +# the safe failure. +try: + from openhop_core.protocol.constants import ( + PERM_ACL_ADMIN, + PERM_ACL_GUEST, + PERM_ACL_READ_ONLY, + PERM_ACL_READ_WRITE, + PERM_ACL_ROLE_MASK, + ) + from openhop_core.protocol.constants import acl_is_admin as is_admin_permissions + from openhop_core.protocol.constants import acl_role as role_of +except ImportError as exc: # pragma: no cover - exercised by the install, not tests + raise ImportError( + "openhop_core is too old: it does not export PERM_ACL_* / acl_is_admin. " + "Install openhop_core with the ACL role fix (fix/login-perms or later) — " + "an older core encodes admin as the 0x02 bit and would announce " + "read-write clients as admins." + ) from exc + logger = logging.getLogger("ACL") -PERM_ACL_GUEST = 0x01 -PERM_ACL_ADMIN = 0x02 -PERM_ACL_READ_WRITE = 0x01 -PERM_ACL_ROLE_MASK = 0x03 +_ROLE_NAMES = { + PERM_ACL_GUEST: "guest", + PERM_ACL_READ_ONLY: "read_only", + PERM_ACL_READ_WRITE: "read_write", + PERM_ACL_ADMIN: "admin", +} + + +def role_name(permissions: int) -> str: + """Human-readable role name for logs and the web API.""" + return _ROLE_NAMES[role_of(permissions)] class ClientInfo: @@ -28,10 +62,14 @@ class ClientInfo: self.sync_since = 0 # For room servers - timestamp of last synced message def is_admin(self) -> bool: - return (self.permissions & PERM_ACL_ROLE_MASK) == PERM_ACL_ADMIN + return is_admin_permissions(self.permissions) def is_guest(self) -> bool: - return (self.permissions & PERM_ACL_ROLE_MASK) == PERM_ACL_GUEST + return role_of(self.permissions) == PERM_ACL_GUEST + + def role_name(self) -> str: + """Role name ("guest"/"read_only"/"read_write"/"admin") for logs and the API.""" + return role_name(self.permissions) class ACL: @@ -148,8 +186,8 @@ class ACL: if self._is_replay(client, timestamp): return False, 0 self._touch_client_session(client, shared_secret, timestamp, sync_since=sync_since) - if (client.permissions & PERM_ACL_ROLE_MASK) == 0: - client.permissions |= PERM_ACL_GUEST + # No role normalisation needed: PERM_ACL_GUEST *is* role 0, so a + # client stored with no role bits already reads back as a guest. return True, client.permissions permissions = 0 @@ -161,8 +199,14 @@ class ACL: permissions = PERM_ACL_ADMIN logger.info(f"Admin password validated for '{target_identity_name or 'unknown'}'") elif guest_pwd and password == guest_pwd: - permissions = PERM_ACL_READ_WRITE - logger.info(f"Guest password validated for '{target_identity_name or 'unknown'}'") + # Firmware splits the guest password by server type. simple_repeater + # grants GUEST (may fetch base telemetry, may not change settings); + # simple_room_server grants READ_WRITE (may post and read messages). + permissions = PERM_ACL_READ_WRITE if is_room_server else PERM_ACL_GUEST + logger.info( + f"Guest password validated for '{target_identity_name or 'unknown'}' " + f"(role={role_name(permissions)})" + ) else: logger.info(f"Invalid password for '{target_identity_name or 'unknown'}'") return False, 0 @@ -183,7 +227,7 @@ class ACL: client.permissions &= ~PERM_ACL_ROLE_MASK client.permissions |= permissions - logger.info(f"Login success! Permissions: {'ADMIN' if client.is_admin() else 'GUEST'}") + logger.info(f"Login success! Role: {client.role_name()}") return True, client.permissions def get_client(self, pub_key: bytes) -> Optional[ClientInfo]: diff --git a/repeater/handler_helpers/protocol_request.py b/repeater/handler_helpers/protocol_request.py index 8111104..b537987 100644 --- a/repeater/handler_helpers/protocol_request.py +++ b/repeater/handler_helpers/protocol_request.py @@ -419,7 +419,10 @@ class ProtocolRequestHelper: result = bytearray() for ci in identity_acl.get_all_clients(): if ci.permissions == 0: - continue # skip deleted entries + # Skip deleted entries. Firmware does the same check verbatim, + # so plain guests (role 0, no extra bits) are deliberately + # absent from the access list on both implementations. + continue pubkey = ci.id.get_public_key() result.extend(pubkey[:6]) # 6-byte pub_key prefix result.append(ci.permissions & 0xFF) diff --git a/repeater/handler_helpers/text.py b/repeater/handler_helpers/text.py index ed0fe5d..f21a1f2 100644 --- a/repeater/handler_helpers/text.py +++ b/repeater/handler_helpers/text.py @@ -13,6 +13,7 @@ import time from openhop_core.node.handlers.text import TextMessageHandler from openhop_core.protocol import CryptoUtils, Identity +from .acl import is_admin_permissions from .mesh_cli import MeshCLI from .room_server import RoomServer @@ -521,17 +522,16 @@ class TextHelper: pubkey = client_info.id.get_public_key() if pubkey[0] == src_hash: permissions = getattr(client_info, "permissions", 0) - PERM_ACL_ADMIN = 0x02 - return (permissions & 0x02) == PERM_ACL_ADMIN + return is_admin_permissions(permissions) return False sender_pubkey = sender_client.id.get_public_key() for client_info in identity_acl.get_all_clients(): if client_info.id.get_public_key() == sender_pubkey: - # Check admin bit (0x02 = PERM_ACL_ADMIN) + # Role is the low two bits with ADMIN == 3; a 0x02 bit test + # would also let READ_WRITE clients run admin CLI commands. permissions = getattr(client_info, "permissions", 0) - PERM_ACL_ADMIN = 0x02 - return (permissions & 0x02) == PERM_ACL_ADMIN + return is_admin_permissions(permissions) return False diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index fa6b480..7cf14d7 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -26,6 +26,7 @@ from repeater.companion.utils import ( validate_companion_config_capacity, ) from repeater.config import resolve_storage_dir +from repeater.handler_helpers.acl import role_name as acl_role_name from repeater.policy_engine import PolicyEngine from repeater.service_utils import get_buildroot_image_info from repeater.utils_packet import create_scoped_advert_packet @@ -6885,7 +6886,13 @@ class APIEndpoints: "public_key": pub_key[:8].hex() + "..." + pub_key[-4:].hex(), "public_key_full": pub_key.hex(), "address": address_bytes.hex(), - "permissions": "admin" if client.is_admin() else "guest", + # Accurate role name: with the guest password + # split (repeater=guest, room server=read_write) + # an "admin or guest" label would hide the + # difference between the non-admin roles. + # Read the byte rather than calling a method: + # ACL clients are duck-typed in several places. + "permissions": acl_role_name(getattr(client, "permissions", 0)), "last_activity": client.last_activity, "last_login_success": client.last_login_success, "last_timestamp": client.last_timestamp, diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index ca8912d..03eb921 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -5243,12 +5243,14 @@ components: example: "e1" permissions: type: string - enum: [admin, guest, read_only] + enum: [admin, read_write, read_only, guest] description: | - Client permission level: - - admin: Full access - - guest: Limited access + Client ACL role (low two bits of the permissions byte, matching + MeshCore ClientACL.h): + - admin: Full access, including settings and CLI + - read_write: May send and receive messages (room server guests) - read_only: Read-only access + - guest: Base telemetry only (repeater guests, read-only logins) example: "admin" last_activity: type: integer diff --git a/tests/test_acl_role_numbering.py b/tests/test_acl_role_numbering.py new file mode 100644 index 0000000..6fb40a9 --- /dev/null +++ b/tests/test_acl_role_numbering.py @@ -0,0 +1,280 @@ +"""ACL role numbering must match upstream MeshCore firmware. + +Firmware ``src/helpers/ClientACL.h`` puts the role in the LOW TWO BITS of the +permissions byte (GUEST=0, READ_ONLY=1, READ_WRITE=2, ADMIN=3) and derives +admin with ``(permissions & 3) == 3``. Every stock MeshCore client decodes the +login reply's ACL byte that way, so any divergence here shows up as an OpenHop +admin being rendered read-write/guest in third-party apps (openhop_repeater +issue #388). +""" + +from pathlib import Path + +import pytest +import yaml + +from openhop_core.protocol.acl_conformance import ROLE_MASK, ROLE_VALUES + +from repeater.handler_helpers.acl import ( + ACL, + PERM_ACL_ADMIN, + PERM_ACL_GUEST, + PERM_ACL_READ_ONLY, + PERM_ACL_READ_WRITE, + PERM_ACL_ROLE_MASK, + ClientInfo, + is_admin_permissions, + role_name, + role_of, +) + +ROOM_CFG = { + "type": "room_server", + "settings": {"admin_password": "room-admin", "guest_password": "room-guest"}, +} + + +class _Id: + def __init__(self, pubkey: bytes): + self._pubkey = pubkey + + def get_public_key(self): + return self._pubkey + + +def test_role_constants_match_firmware_clientacl_h(): + """Literal pin against the firmware header, via the shared vector table.""" + assert PERM_ACL_ROLE_MASK == ROLE_MASK == 0x03 + assert PERM_ACL_GUEST == ROLE_VALUES["guest"] == 0x00 + assert PERM_ACL_READ_ONLY == ROLE_VALUES["read_only"] == 0x01 + assert PERM_ACL_READ_WRITE == ROLE_VALUES["read_write"] == 0x02 + assert PERM_ACL_ADMIN == ROLE_VALUES["admin"] == 0x03 + # Every role must be distinct; GUEST and READ_WRITE used to collide on 0x01. + roles = {PERM_ACL_GUEST, PERM_ACL_READ_ONLY, PERM_ACL_READ_WRITE, PERM_ACL_ADMIN} + assert len(roles) == 4 + + +def test_roles_come_from_openhop_core_and_never_diverge(): + """We re-export core's values rather than keeping a second copy. + + This must not skip on an old core. A core without these symbols encodes + the login reply's is_admin byte as ``permissions & 0x02``, which also + matches READ_WRITE — pairing it with our numbering would announce a room + server's read-write clients as admins. acl.py fails closed at import for + exactly that reason, and this asserts the contract it depends on. + """ + from openhop_core.protocol import constants + + assert constants.PERM_ACL_ROLE_MASK is PERM_ACL_ROLE_MASK + assert constants.PERM_ACL_GUEST is PERM_ACL_GUEST + assert constants.PERM_ACL_READ_ONLY is PERM_ACL_READ_ONLY + assert constants.PERM_ACL_READ_WRITE is PERM_ACL_READ_WRITE + assert constants.PERM_ACL_ADMIN is PERM_ACL_ADMIN + assert is_admin_permissions is constants.acl_is_admin + assert role_of is constants.acl_role + + +@pytest.mark.parametrize( + "permissions, expect_admin", + [ + (PERM_ACL_GUEST, False), + (PERM_ACL_READ_ONLY, False), + (PERM_ACL_READ_WRITE, False), # regression: 0x02 bit test said True here + (PERM_ACL_ADMIN, True), + ], +) +def test_is_admin_is_an_equality_test_not_a_bit_test(permissions, expect_admin): + assert is_admin_permissions(permissions) is expect_admin + assert ClientInfo(_Id(b"P" * 32), permissions).is_admin() is expect_admin + + +def test_reserved_upper_bits_do_not_change_the_role(): + """Bits above the role mask are reserved flags, not part of the role.""" + assert is_admin_permissions(PERM_ACL_ADMIN | 0xFC) is True + assert is_admin_permissions(PERM_ACL_READ_WRITE | 0xFC) is False + assert role_of(PERM_ACL_READ_ONLY | 0xF0) == PERM_ACL_READ_ONLY + assert role_name(PERM_ACL_ADMIN | 0x80) == "admin" + + +def test_role_names(): + assert role_name(PERM_ACL_GUEST) == "guest" + assert role_name(PERM_ACL_READ_ONLY) == "read_only" + assert role_name(PERM_ACL_READ_WRITE) == "read_write" + assert role_name(PERM_ACL_ADMIN) == "admin" + + +def test_admin_password_grants_admin_on_repeater_and_room_server(): + repeater_acl = ACL(admin_password="rpt-admin", guest_password="rpt-guest") + ok, perms = repeater_acl.authenticate_client( + client_identity=_Id(b"A" * 32), + shared_secret=b"s" * 32, + password="rpt-admin", + timestamp=100, + ) + assert (ok, perms) == (True, PERM_ACL_ADMIN) + + room_acl = ACL() + ok, perms = room_acl.authenticate_client( + client_identity=_Id(b"B" * 32), + shared_secret=b"s" * 32, + password="room-admin", + timestamp=100, + target_identity_name="room-a", + target_identity_config=ROOM_CFG, + ) + assert (ok, perms) == (True, PERM_ACL_ADMIN) + + +def test_guest_password_is_guest_on_a_repeater(): + """Repeater guests may fetch base telemetry but must not change settings. + + Matches firmware simple_repeater handleLoginReq, which assigns + PERM_ACL_GUEST for the guest password. + """ + acl = ACL(admin_password="rpt-admin", guest_password="rpt-guest") + ok, perms = acl.authenticate_client( + client_identity=_Id(b"C" * 32), + shared_secret=b"s" * 32, + password="rpt-guest", + timestamp=100, + ) + assert ok is True + assert perms == PERM_ACL_GUEST + client = acl.get_client(b"C" * 32) + assert client.is_admin() is False + assert client.is_guest() is True + + +def test_guest_password_is_read_write_on_a_room_server(): + """Room-server guests may post and read messages. + + Matches firmware simple_room_server, which assigns PERM_ACL_READ_WRITE for + the room password and reserves PERM_ACL_GUEST for allow_read_only. + """ + acl = ACL() + ok, perms = acl.authenticate_client( + client_identity=_Id(b"D" * 32), + shared_secret=b"s" * 32, + password="room-guest", + timestamp=100, + target_identity_name="room-a", + target_identity_config=ROOM_CFG, + ) + assert ok is True + assert perms == PERM_ACL_READ_WRITE + client = acl.get_client(b"D" * 32) + assert client.is_admin() is False + assert client.is_guest() is False # read-write, so posting is allowed + + +def test_blank_password_read_only_access_is_guest(): + """allow_read_only mirrors firmware's allow_read_only branch → GUEST.""" + acl = ACL(allow_read_only=True) + ok, perms = acl.authenticate_client( + client_identity=_Id(b"E" * 32), + shared_secret=b"s" * 32, + password="", + timestamp=100, + ) + assert (ok, perms) == (True, PERM_ACL_GUEST) + assert acl.get_client(b"E" * 32).is_guest() is True + + +def test_role_change_replaces_the_role_without_touching_reserved_bits(): + acl = ACL(admin_password="rpt-admin", guest_password="rpt-guest") + identity = _Id(b"F" * 32) + + ok, perms = acl.authenticate_client( + client_identity=identity, + shared_secret=b"s" * 32, + password="rpt-admin", + timestamp=100, + ) + assert (ok, perms) == (True, PERM_ACL_ADMIN) + + client = acl.get_client(b"F" * 32) + client.permissions |= 0x40 # a reserved flag set elsewhere + + ok, perms = acl.authenticate_client( + client_identity=identity, + shared_secret=b"s" * 32, + password="rpt-guest", + timestamp=101, + ) + assert ok is True + assert role_of(perms) == PERM_ACL_GUEST + assert perms & 0x40 # reserved flag preserved + assert client.is_admin() is False + + +def test_every_role_name_is_allowed_by_the_openapi_schema(): + """The ACL client list advertises an enum; every role we emit must be in it. + + The role split added "read_write", which the schema did not previously + allow — generated clients and validators reject undeclared values. + """ + spec = yaml.safe_load( + (Path(__file__).resolve().parents[1] / "repeater/web/openapi.yaml").read_text() + ) + enum = spec["components"]["schemas"]["ACLClient"]["properties"]["permissions"]["enum"] + + emitted = { + role_name(PERM_ACL_GUEST), + role_name(PERM_ACL_READ_ONLY), + role_name(PERM_ACL_READ_WRITE), + role_name(PERM_ACL_ADMIN), + } + assert emitted <= set(enum), f"roles missing from the OpenAPI enum: {emitted - set(enum)}" + + +def test_acl_refuses_to_import_against_a_core_without_the_role_constants(monkeypatch): + """The fail-closed guard must actually fire, with a message that helps. + + A core lacking these symbols still builds the login reply's admin byte + from ``permissions & 0x02``, which would announce a room server's + READ_WRITE clients as admins. Refusing to import is the safe failure, and + it is worth a test because nothing else exercises that path. + """ + import importlib + import sys + import types + + import openhop_core.protocol.constants as real_constants + + stub = types.ModuleType("openhop_core.protocol.constants") + for name in dir(real_constants): + if not name.startswith("__"): + setattr(stub, name, getattr(real_constants, name)) + for missing in ( + "PERM_ACL_ROLE_MASK", + "PERM_ACL_GUEST", + "PERM_ACL_READ_ONLY", + "PERM_ACL_READ_WRITE", + "PERM_ACL_ADMIN", + "acl_is_admin", + "acl_role", + ): + delattr(stub, missing) + + monkeypatch.setitem(sys.modules, "openhop_core.protocol.constants", stub) + monkeypatch.delitem(sys.modules, "repeater.handler_helpers.acl") + + with pytest.raises(ImportError) as exc: + importlib.import_module("repeater.handler_helpers.acl") + + message = str(exc.value) + assert "openhop_core is too old" in message + assert "fix/login-perms" in message + + +def test_acl_imports_normally_against_the_installed_core(): + """Companion to the guard test: the happy path must still work. + + Guards that always fire are worse than no guard, and the test above + manipulates sys.modules. + """ + import importlib + + module = importlib.import_module("repeater.handler_helpers.acl") + assert module.PERM_ACL_ADMIN == 0x03 + assert module.is_admin_permissions(0x03) is True diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index 9aed860..9673bd3 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/tests/test_api_endpoints_core_coverage.py @@ -7,6 +7,11 @@ from unittest.mock import AsyncMock, MagicMock, mock_open, patch import cherrypy import pytest +from repeater.handler_helpers.acl import ( + PERM_ACL_ADMIN, + PERM_ACL_GUEST, + PERM_ACL_READ_WRITE, +) from repeater.web.api_endpoints import APIEndpoints @@ -2316,15 +2321,15 @@ class _FakeIdentityObj: class _FakeClient: - def __init__(self, key_hex: str, admin: bool): + def __init__(self, key_hex: str, permissions: int): self.id = SimpleNamespace(get_public_key=lambda: bytes.fromhex(key_hex)) - self._admin = admin + self.permissions = permissions self.last_activity = 1.0 self.last_login_success = 2.0 self.last_timestamp = 3.0 def is_admin(self): - return self._admin + return (self.permissions & 0x03) == PERM_ACL_ADMIN class _FakeACL: @@ -2462,7 +2467,11 @@ def test_acl_endpoints_paths(cherrypy_ctx): request.method = "GET" assert api.acl_info()["success"] is False - clients = [_FakeClient("aa" * 32, True), _FakeClient("bb" * 32, False)] + clients = [ + _FakeClient("aa" * 32, PERM_ACL_ADMIN), + _FakeClient("bb" * 32, PERM_ACL_READ_WRITE), + _FakeClient("cc" * 32, PERM_ACL_GUEST), + ] acl = _FakeACL(clients) login_helper = SimpleNamespace(get_acl_dict=lambda: {0x42: acl, 0x51: _FakeACL([])}) id_mgr = SimpleNamespace( @@ -2492,6 +2501,12 @@ def test_acl_endpoints_paths(cherrypy_ctx): all_clients = api.acl_clients() assert all_clients["success"] is True assert all_clients["data"]["count"] >= 1 + # The label is the real role. Collapsing it to "admin or guest" would hide + # a room server's read-write clients behind the guest label. + labels = {c["public_key_full"][:2]: c["permissions"] for c in all_clients["data"]["clients"]} + assert labels["aa"] == "admin" + assert labels["bb"] == "read_write" + assert labels["cc"] == "guest" assert api.acl_clients(identity_hash="bad")["success"] is False assert api.acl_clients(identity_name="missing")["success"] is False diff --git a/tests/test_handler_helpers_path_protocol_text.py b/tests/test_handler_helpers_path_protocol_text.py index af9ba41..27bc271 100644 --- a/tests/test_handler_helpers_path_protocol_text.py +++ b/tests/test_handler_helpers_path_protocol_text.py @@ -8,7 +8,12 @@ from openhop_core.node.handlers.result import HandlerResult from openhop_core.protocol import Identity, LocalIdentity from openhop_core.protocol.packet_builder import PacketBuilder -from repeater.handler_helpers.acl import ACL, PERM_ACL_ADMIN, ClientInfo +from repeater.handler_helpers.acl import ( + ACL, + PERM_ACL_ADMIN, + PERM_ACL_READ_WRITE, + ClientInfo, +) from repeater.handler_helpers.path import PathHelper from repeater.handler_helpers.protocol_request import ProtocolRequestHelper from repeater.handler_helpers.text import TextHelper @@ -424,13 +429,20 @@ def test_protocol_request_owner_info_fallback_version(): def test_text_helper_cli_prefix_and_admin_permission_checks(): + # 0x21 is ADMIN (role 3). 0x22 is READ_WRITE (role 2) — the case that must + # NOT pass the admin gate: the old `permissions & 0x02` test let a + # read-write client run admin CLI commands. acl = _FakeACL( [ _FakeClient( - pubkey=bytes([0x21]) + b"x" * 31, shared_secret=b"k" * 32, permissions=0x02 + pubkey=bytes([0x21]) + b"x" * 31, + shared_secret=b"k" * 32, + permissions=PERM_ACL_ADMIN, ), _FakeClient( - pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32, permissions=0x01 + pubkey=bytes([0x22]) + b"x" * 31, + shared_secret=b"k" * 32, + permissions=PERM_ACL_READ_WRITE, ), ] ) diff --git a/tests/test_login_reply_acl_wire.py b/tests/test_login_reply_acl_wire.py new file mode 100644 index 0000000..e061475 --- /dev/null +++ b/tests/test_login_reply_acl_wire.py @@ -0,0 +1,164 @@ +"""End-to-end: repeater ACL role -> the bytes a MeshCore client actually reads. + +The login reply is 13 bytes; byte 6 is the legacy is_admin flag and byte 7 is +the ACL permissions byte. A stock companion forwards both verbatim into +PUSH_CODE_LOGIN_SUCCESS, and modern clients derive the displayed role from +byte 7 with ``(perms & 3) == 3``. openhop_repeater issue #388 was exactly this +pair disagreeing: byte 6 said admin while byte 7 decoded as read-write. + +These tests drive the real openhop_core LoginServerHandler with the real +repeater ACL as its authenticate callback, then decrypt the reply and assert +the wire bytes. +""" + +import struct +import time + +import pytest +from openhop_core.node.handlers.login_server import LoginServerHandler +from openhop_core.protocol import CryptoUtils, Identity, LocalIdentity, Packet +from openhop_core.protocol.constants import ( + PAYLOAD_TYPE_ANON_REQ, + PAYLOAD_TYPE_RESPONSE, + ROUTE_TYPE_FLOOD, +) + +from openhop_core.protocol.acl_conformance import OUTBOUND + +from repeater.handler_helpers.acl import ACL + +ROOM_CFG = { + "type": "room_server", + "settings": {"admin_password": "room-admin", "guest_password": "room-guest"}, +} + +# How each conformance vector's (server_type, credential) is produced here. +# The ACL is built with these passwords; "" exercises allow_read_only. +CREDENTIALS = { + ("repeater", "admin_password"): ("rpt-admin", None), + ("repeater", "guest_password"): ("rpt-guest", None), + ("repeater", "blank_read_only"): ("", None), + ("room_server", "admin_password"): ("room-admin", ROOM_CFG), + ("room_server", "guest_password"): ("room-guest", ROOM_CFG), +} + + +def _build_login_packet( + client: LocalIdentity, server: LocalIdentity, password: str, *, room: bool = False +) -> Packet: + """Build an ANON_REQ login packet the way a client does. + + Repeater format is ``timestamp(4) + password + NUL``; a room server inserts + a 4-byte ``sync_since`` before the password (firmware reads it at data[8]). + """ + server_id = Identity(server.get_public_key()) + shared_secret = server_id.calc_shared_secret(client.get_private_key()) + plaintext = struct.pack(" bytes: + """Run a real login through the ACL + handler and return the 13 reply bytes.""" + server = LocalIdentity() + client = LocalIdentity() + + def authenticate(client_identity, shared_secret, password_, timestamp, sync_since=None): + return acl.authenticate_client( + client_identity=client_identity, + shared_secret=shared_secret, + password=password_, + timestamp=timestamp, + sync_since=sync_since, + target_identity_name="node", + target_identity_config=room_config or {}, + ) + + sent = [] + handler = LoginServerHandler( + local_identity=server, + log_fn=lambda _msg: None, + authenticate_callback=authenticate, + is_room_server=bool(room_config), + ) + handler.set_send_packet_callback(lambda pkt, *a, **kw: sent.append(pkt)) + + await handler(_build_login_packet(client, server, password, room=bool(room_config))) + assert sent, "handler sent no login response" + + shared_secret = Identity(client.get_public_key()).calc_shared_secret(server.get_private_key()) + plaintext = CryptoUtils.mac_then_decrypt( + shared_secret[:16], shared_secret, bytes(sent[0].payload[2:]) + ) + # PATH return envelope: path_len(1) + path(0) + extra_type(1) + reply(13) + assert plaintext[0] == 0 + assert plaintext[1] & 0x0F == PAYLOAD_TYPE_RESPONSE + return plaintext[2:15] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "server_type, credential, admin_code, permissions", + OUTBOUND, + ids=[f"{srv}-{cred}" for srv, cred, _, _ in OUTBOUND], +) +async def test_outbound_conformance_vectors(server_type, credential, admin_code, permissions): + """The real ACL plus the real core handler must emit the literal bytes. + + This is the only place both halves of the fix meet, and the expectations + are literals from openhop_core.protocol.acl_conformance rather than + PERM_ACL_* — a symbolic assertion would follow the constants if they ever + drift away from the mesh again, which is exactly how #388 stayed hidden. + """ + password, room_config = CREDENTIALS[(server_type, credential)] + acl = ACL( + admin_password="rpt-admin", + guest_password="rpt-guest", + allow_read_only=True, + ) + reply = await _login_reply(acl, password, room_config=room_config) + + assert reply[6] == admin_code + assert reply[7] == permissions + + +@pytest.mark.asyncio +async def test_admin_is_role_three_not_the_0x02_bit(): + """The single assertion that would have caught #388. + + A stock client computes (perms & 3) == 3. Our admin used to be 0x02, which + that expression reads as READ_WRITE. + """ + acl = ACL(admin_password="rpt-admin", guest_password="rpt-guest") + reply = await _login_reply(acl, "rpt-admin") + + assert reply[7] & 0x03 == 3 + assert reply[7] != 0x02 + + +@pytest.mark.asyncio +async def test_repeater_and_room_guests_get_different_roles(): + """The split is observable on the wire, not just in our own vocabulary. + + GUEST and READ_WRITE were both 0x01 before, so these two logins were + indistinguishable to every client. + """ + rpt = await _login_reply( + ACL(admin_password="rpt-admin", guest_password="rpt-guest"), "rpt-guest" + ) + room = await _login_reply(ACL(), "room-guest", room_config=ROOM_CFG) + + assert rpt[7] == 0x00 + assert room[7] == 0x02 + assert rpt[7] != room[7]