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") == "<hidden>"
+        assert format_log_uuid("secret", "oidc.example.com.key") == "<hidden>"
+
+    def test_oidc_key_path_component_visible(self):
+        # The path component itself must stay visible ("oidc.<rp-id>.key = <hidden>")
+        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