From cdabc5d9e668b9ee73d354a6d655bbbf34be3627 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sun, 6 Sep 2026 04:33:08 +0000 Subject: [PATCH] Realm machinery tests: resolution, validation, dispatch, code binding, conversion - Registry resolve order (exact rp-id, auth host, related origin, longest suffix), port/trailing-dot normalization, unknown hosts. - Cross-realm validate_config: related-origin cap, auth-host/related collisions, related-inside-other-realm, auth-host vs rp-id. - ASGI dispatch: HTTP 421, realm in scope state, contextvar scoping and reset, WS pre-accept close and cross-realm effective-auth-host rule. - Auth codes bound to issuing realm (set-session and OIDC token). - Legacy conversion stamps credentials/sessions/OIDC with the rp-id. - OIDC key censoring in transaction logs; bootstrap default-realm caveat. --- tests/test_realms.py | 494 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 494 insertions(+) create mode 100644 tests/test_realms.py diff --git a/tests/test_realms.py b/tests/test_realms.py new file mode 100644 index 0000000..bdd971b --- /dev/null +++ b/tests/test_realms.py @@ -0,0 +1,494 @@ +"""Tests for the multi-realm machinery: registry resolution, config +validation, ASGI dispatch, realm binding of auth codes, legacy database +conversion, log censoring and bootstrap caveats. +""" + +from __future__ import annotations + +import asyncio +import os +from datetime import UTC, datetime +from uuid import UUID + +import httpx +import pytest +from kanta import Kanta + +from paskia import authcode, realms +from paskia.bootstrap import check_admin_credentials +from paskia.db import create_credential +from paskia.db.legacy import ( + LegacyConfig, + LegacyCredential, + LegacyDB, + LegacySession, + convert_legacy_database, +) +from paskia.db.lifecycle import format_log_uuid +from paskia.db.operations import DB +from paskia.db.structs import Client, Config, Credential, RealmConfig +from paskia.fastapi.dispatch import DispatchMiddleware + +# ------------------------------------------------------------------------- +# Registry construction helpers +# ------------------------------------------------------------------------- + + +def build_registry(*realm_configs: RealmConfig) -> realms.RealmRegistry: + """Build and install a registry from realm configs (listen unset).""" + realms.configure(listen=None) + return realms.init_registry(Config(realms=list(realm_configs))) + + +ROR_CONFIG = Config( + realms=[ + RealmConfig( + rp_id="company.com", + auth_host="https://auth.company.com", + origins=["https://auth.company.com", "https://app.com"], + ), + RealmConfig(rp_id="pro.com"), + ] +) + + +class StubApp: + """ASGI app recording the scope it was called with.""" + + def __init__(self): + self.scope = None + + async def __call__(self, scope, receive, send): + self.scope = scope + + +async def drive_ws(middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]]): + """Run a websocket scope through the middleware, capturing sent messages.""" + + async def receive(): + return {"type": "websocket.connect"} + + sent = [] + + async def send(message): + sent.append(message) + + stub = middleware.app + await middleware( + {"type": "websocket", "headers": headers, "path": "/"}, receive, send + ) + return stub, sent + + +async def drive_http( + middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]] +): + """Run an http scope through the middleware, capturing sent messages.""" + + async def receive(): + return {"type": "http.request", "body": b""} + + sent = [] + + async def send(message): + sent.append(message) + + stub = middleware.app + await middleware( + { + "type": "http", + "headers": headers, + "method": "GET", + "path": "/", + "query_string": b"", + }, + receive, + send, + ) + return stub, sent + + +# ------------------------------------------------------------------------- +# Host resolution +# ------------------------------------------------------------------------- + + +class TestResolve: + def test_exact_rp_id(self): + reg = build_registry(*ROR_CONFIG.realms) + assert reg.resolve("pro.com").rp_id == "pro.com" + assert reg.resolve("company.com").rp_id == "company.com" + + def test_auth_host_and_related_origin(self): + reg = build_registry(*ROR_CONFIG.realms) + assert reg.resolve("auth.company.com").rp_id == "company.com" + assert reg.resolve("app.com").rp_id == "company.com" + + def test_subdomain_suffix_longest_match(self): + reg = build_registry( + RealmConfig(rp_id="example.com"), RealmConfig(rp_id="sub.example.com") + ) + assert reg.resolve("www.example.com").rp_id == "example.com" + assert reg.resolve("api.sub.example.com").rp_id == "sub.example.com" + + def test_port_and_trailing_dot_normalized(self): + reg = build_registry(*ROR_CONFIG.realms) + assert reg.resolve("pro.com:8443").rp_id == "pro.com" + assert reg.resolve("app.com.").rp_id == "company.com" + + def test_unknown_host(self): + reg = build_registry(*ROR_CONFIG.realms) + assert reg.resolve("evil.com") is None + assert reg.resolve("") is None + assert reg.resolve(None) is None + + def test_effective_auth_host_fallback(self): + reg = build_registry(*ROR_CONFIG.realms) + company = reg.get("company.com") + pro = reg.get("pro.com") + assert reg.effective_auth_host(company) == "auth.company.com" + # pro.com has no own auth host: falls back to the first configured one + assert reg.effective_auth_host(pro) == "auth.company.com" + # No auth hosts at all: None + reg2 = build_registry(RealmConfig(rp_id="a.com"), RealmConfig(rp_id="b.com")) + assert reg2.effective_auth_host(reg2.get("a.com")) is None + + +# ------------------------------------------------------------------------- +# Cross-realm configuration validation +# ------------------------------------------------------------------------- + + +class TestValidateConfig: + def test_valid(self): + realms.validate_config(ROR_CONFIG) + + def test_related_origin_cap(self): + realms.validate_config( + Config( + realms=[ + RealmConfig( + rp_id="company.com", + origins=[f"https://app{i}.com" for i in range(5)], + ) + ] + ) + ) + with pytest.raises(ValueError, match="related origins"): + realms.validate_config( + Config( + realms=[ + RealmConfig( + rp_id="company.com", + origins=[f"https://app{i}.com" for i in range(6)], + ) + ] + ) + ) + + def test_auth_host_collision(self): + with pytest.raises(ValueError, match="collides with a related origin"): + realms.validate_config( + Config( + realms=[ + RealmConfig(rp_id="a.com", auth_host="https://auth.a.com"), + RealmConfig( + rp_id="b.com", + origins=["https://auth.a.com"], + ), + ] + ) + ) + + def test_related_origin_inside_other_realm(self): + with pytest.raises(ValueError, match="falls inside realm"): + realms.validate_config( + Config( + realms=[ + RealmConfig(rp_id="a.com", origins=["https://app.b.com"]), + RealmConfig(rp_id="b.com"), + ] + ) + ) + + def test_auth_host_must_not_collide_with_rp_id(self): + with pytest.raises(ValueError, match="collides with an rp-id"): + realms.validate_config( + Config( + realms=[ + RealmConfig(rp_id="a.com", auth_host="https://b.a.com"), + RealmConfig(rp_id="b.a.com"), + ] + ) + ) + + +# ------------------------------------------------------------------------- +# ASGI dispatch +# ------------------------------------------------------------------------- + + +class TestDispatchMiddleware: + @pytest.mark.asyncio + async def test_http_unknown_host_421(self): + build_registry(*ROR_CONFIG.realms) + stub, sent = await drive_http( + DispatchMiddleware(StubApp()), [(b"host", b"evil.com")] + ) + assert stub.scope is None # Inner app not called + assert sent[0]["type"] == "http.response.start" + assert sent[0]["status"] == 421 + + @pytest.mark.asyncio + async def test_http_dispatches_realm(self): + build_registry(*ROR_CONFIG.realms) + stub, _sent = await drive_http( + DispatchMiddleware(StubApp()), [(b"host", b"app.com.")] + ) + assert stub.scope is not None + assert stub.scope["state"]["realm"].rp_id == "company.com" + + @pytest.mark.asyncio + async def test_http_current_realm_set_inside_request(self): + reg = build_registry(*ROR_CONFIG.realms) + seen = {} + + async def app(scope, receive, send): + seen["realm"] = realms.current_realm() + + await drive_http(DispatchMiddleware(app), [(b"host", b"pro.com")]) + assert seen["realm"].rp_id == "pro.com" + # Contextvar is reset after the request + assert realms.current_realm() is reg.default + + @pytest.mark.asyncio + async def test_ws_unknown_host_closed(self): + build_registry(*ROR_CONFIG.realms) + stub, sent = await drive_ws( + DispatchMiddleware(StubApp()), [(b"host", b"evil.com")] + ) + assert stub.scope is None + assert sent == [{"type": "websocket.close", "code": 1008}] + + @pytest.mark.asyncio + async def test_ws_same_realm_origin(self): + build_registry(*ROR_CONFIG.realms) + stub, sent = await drive_ws( + DispatchMiddleware(StubApp()), + [(b"host", b"auth.company.com"), (b"origin", b"https://app.com")], + ) + assert sent == [] + assert stub.scope["state"]["realm"].rp_id == "company.com" + + @pytest.mark.asyncio + async def test_ws_cross_realm_requires_effective_auth_host(self): + build_registry(*ROR_CONFIG.realms) + # pro.com page connecting to the shared auth host: allowed, pro realm + stub, sent = await drive_ws( + DispatchMiddleware(StubApp()), + [(b"host", b"auth.company.com"), (b"origin", b"https://pro.com")], + ) + assert sent == [] + assert stub.scope["state"]["realm"].rp_id == "pro.com" + + # pro.com page connecting to some other host: closed pre-accept + stub, sent = await drive_ws( + DispatchMiddleware(StubApp()), + [(b"host", b"company.com"), (b"origin", b"https://pro.com")], + ) + assert stub.scope is None + assert sent == [{"type": "websocket.close", "code": 1008}] + + @pytest.mark.asyncio + async def test_ws_unknown_origin_uses_host_realm(self): + build_registry(*ROR_CONFIG.realms) + # Missing origin + stub, _ = await drive_ws(DispatchMiddleware(StubApp()), [(b"host", b"pro.com")]) + assert stub.scope["state"]["realm"].rp_id == "pro.com" + # Unknown origin: host realm applies (endpoint-side validation decides) + stub, _ = await drive_ws( + DispatchMiddleware(StubApp()), + [(b"host", b"pro.com"), (b"origin", b"https://evil.com")], + ) + assert stub.scope["state"]["realm"].rp_id == "pro.com" + + +# ------------------------------------------------------------------------- +# Realm binding of auth codes +# ------------------------------------------------------------------------- + + +class TestAuthCodeRealmBinding: + @pytest.mark.asyncio + async def test_cookie_code_rejected_on_other_realm( + self, client: httpx.AsyncClient, session_token: str + ): + code = authcode.store_cookie( + authcode.CookieCode( + session_key=session_token, + created=datetime.now(UTC), + rp_id="other.com", + ) + ) + response = await client.post( + "/auth/api/set-session", + headers={ + "Authorization": f"Bearer {code}", + "Host": "localhost:4401", + }, + ) + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_oidc_code_rejected_on_other_realm( + self, client: httpx.AsyncClient, test_db: DB + ): + oidc_client, secret = Client.create( + name="Test Client", + redirect_uris=["https://client.example/callback"], + client_secret="topsecret", + ) + store = test_db._store + with store.transaction("create_test_oidc_client"): + test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client + + code = authcode.store_oidc( + authcode.OIDCCode( + session_key="doesnotmatter1234", + created=datetime.now(UTC), + redirect_uri="https://client.example/callback", + scope="openid", + rp_id="other.com", + ) + ) + response = await client.post( + "/auth/oidc/token", + data={ + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "https://client.example/callback", + "client_id": str(oidc_client.uuid), + "client_secret": secret, + }, + headers={"Host": "localhost:4401"}, + ) + assert response.status_code == 400 + assert response.json()["error"] == "invalid_grant" + assert "realm" in response.json()["error_description"] + + +# ------------------------------------------------------------------------- +# Legacy database conversion +# ------------------------------------------------------------------------- + + +def _read_db(path) -> DB: + async def _read() -> DB: + new_db = DB() + kanta = Kanta(str(path), new_db) + await kanta.open(readonly=True) + return kanta.data + + return asyncio.run(_read()) + + +class TestLegacyConversion: + def test_convert_stamps_realm_everywhere(self, tmp_path): + src = tmp_path / "example.com.paskiadb" + src.mkdir() + src_file = src / "main.db" + + cred_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c5") + user_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c6") + + async def _write() -> None: + kanta = Kanta(str(src_file), LegacyDB()) + await kanta.open() + with kanta.transaction("test:seed"): + kanta.data.config = LegacyConfig( + rp_id="example.com", + rp_name="Example", + origins=["https://app.example.com"], + ) + kanta.data.credentials[cred_uuid] = LegacyCredential( + credential_id=b"credential-id", + user_uuid=user_uuid, + aaguid=UUID(int=0), + public_key=b"public-key", + sign_count=3, + created_at=datetime.now(UTC), + ) + kanta.data.sessions["session-key"] = LegacySession( + user_uuid=user_uuid, + credential_uuid=cred_uuid, + host="example.com", + ip="127.0.0.1", + user_agent="pytest", + validated=datetime.now(UTC), + ) + await kanta.close() + + asyncio.run(_write()) + + config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb") + assert config.default_realm.rp_id == "example.com" + assert config.default_realm.rp_name == "Example" + + converted = _read_db(tmp_path / "paskia.kantadb") + assert converted.credentials[cred_uuid].rp_id == "example.com" + assert converted.sessions["session-key"].rp_id == "example.com" + assert set(converted.oidc.keys()) == {"example.com"} + + +# ------------------------------------------------------------------------- +# Transaction log censoring +# ------------------------------------------------------------------------- + + +class TestLogCensoring: + def test_oidc_key_values_hidden(self): + assert format_log_uuid(b"raw-key-material", "oidc.localhost.key") == "" + assert format_log_uuid("secret", "oidc.example.com.key") == "" + + def test_oidc_key_path_component_visible(self): + # The path component itself must stay visible ("oidc..key = ") + assert format_log_uuid("key", "oidc.localhost.key") is None + + def test_other_paths_unaffected(self): + assert format_log_uuid("not-a-uuid", "oidc.localhost.clients") is None + assert format_log_uuid("not-a-uuid", "config.realms") is None + + +# ------------------------------------------------------------------------- +# Bootstrap caveat: admin credential is checked on the default realm +# ------------------------------------------------------------------------- + + +class TestBootstrapCaveat: + @pytest.mark.asyncio + async def test_admin_without_credentials_gets_link( + self, test_db: DB, realm_registry + ): + assert await check_admin_credentials() is True + + @pytest.mark.asyncio + async def test_admin_with_default_realm_credential_ok( + self, test_db: DB, realm_registry, test_user, test_credential + ): + assert await check_admin_credentials() is False + + @pytest.mark.asyncio + async def test_admin_with_only_other_realm_credential_gets_link( + self, test_db: DB, realm_registry, test_user + ): + """A passkey under a non-default realm does not satisfy the check.""" + cred = Credential.create( + credential_id=os.urandom(32), + user=test_user.uuid, + aaguid=UUID(int=0), + public_key=os.urandom(64), + sign_count=0, + rp_id="example.com", + ) + create_credential(cred) + assert await check_admin_credentials() is True