From 2039c47e46125460a854ce327a315362d20a7abc Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Mon, 7 Sep 2026 06:33:54 +0000
Subject: [PATCH] Tests: align with instance-global OIDC and per-domain auth
 hosts

- '*' origin is wildcard shorthand (https-only outside localhost)
- related origins may fall inside another domain's rp-id; the listing
  wins dispatch, an exact rp-id always wins
- shared auth hosts resolve best-suffix; no cross-domain fallback
- OIDC codes are host-independent; log censoring path is oidc.key
- legacy wildcards convert as-is; legacy OIDC carries over as-is
---
 paskia/db/operations.py              |   8 +-
 paskia/domains.py                    |   4 +-
 paskia/fastapi/admin/oidc_clients.py |   4 +-
 paskia/fastapi/dispatch.py           |   4 +-
 paskia/fastapi/ws.py                 |   4 +-
 tests/test_admin.py                  |  20 +--
 tests/test_api.py                    |   3 +-
 tests/test_cli.py                    |  13 +-
 tests/test_domains.py                | 194 +++++++++++++++++++++------
 9 files changed, 182 insertions(+), 72 deletions(-)

diff --git a/paskia/db/operations.py b/paskia/db/operations.py
index 10bd63d..904dff7 100644
--- a/paskia/db/operations.py
+++ b/paskia/db/operations.py
@@ -771,9 +771,7 @@ def delete_domain(rp_id: str, *, ctx: SessionContext | None = None) -> None:
 # -------------------------------------------------------------------------
 
 
-def create_oid_client(
-    client: Client, *, ctx: SessionContext | None = None
-) -> None:
+def create_oid_client(client: Client, *, ctx: SessionContext | None = None) -> None:
     """Create a new OIDC client."""
     if client.uuid in _db.oidc.clients:
         raise ValueError(f"OIDC client {client.uuid} already exists")
@@ -855,9 +853,7 @@ def reset_oid_client_secret(
         _db.oidc.clients[client_uuid] = updated
 
 
-def delete_oid_client(
-    client_uuid: UUID, *, ctx: SessionContext | None = None
-) -> None:
+def delete_oid_client(client_uuid: UUID, *, ctx: SessionContext | None = None) -> None:
     """Delete an OIDC client."""
     if client_uuid not in _db.oidc.clients:
         raise ValueError(f"OIDC client {client_uuid} not found")
diff --git a/paskia/domains.py b/paskia/domains.py
index c728305..b9d16f3 100644
--- a/paskia/domains.py
+++ b/paskia/domains.py
@@ -436,9 +436,7 @@ def _derive_site(
     if rp_id in domain.origins:
         return origin_url(rp_id), "/auth/"
     concrete = sorted(
-        k
-        for k in domain.origins
-        if k != "*" and not hostutil.is_wildcard_pattern(k)
+        k for k in domain.origins if k != "*" and not hostutil.is_wildcard_pattern(k)
     )
     if concrete:
         return origin_url(concrete[0]), "/auth/"
diff --git a/paskia/fastapi/admin/oidc_clients.py b/paskia/fastapi/admin/oidc_clients.py
index f8ab18b..2061520 100644
--- a/paskia/fastapi/admin/oidc_clients.py
+++ b/paskia/fastapi/admin/oidc_clients.py
@@ -201,9 +201,7 @@ async def admin_reset_oidc_client_secret(
         raise ValueError("secret_hash must be a SHA-256 hash (32 bytes)")
 
     try:
-        db.reset_oid_client_secret(
-            client_uuid, secret_hash, ctx=ctx
-        )
+        db.reset_oid_client_secret(client_uuid, secret_hash, ctx=ctx)
     except ValueError as e:
         raise HTTPException(status_code=404, detail=str(e))
 
diff --git a/paskia/fastapi/dispatch.py b/paskia/fastapi/dispatch.py
index f6aa68a..ae015a7 100644
--- a/paskia/fastapi/dispatch.py
+++ b/paskia/fastapi/dispatch.py
@@ -74,9 +74,7 @@ class DispatchMiddleware:
         if origin_domain is not None and origin_domain is not host_domain:
             # Cross-domain connection: only via the origin domain's own auth host.
             own = origin_domain.own_auth_host
-            if not own or hostutil.normalize_host(
-                host
-            ) != hostutil.normalize_host(own):
+            if not own or hostutil.normalize_host(host) != hostutil.normalize_host(own):
                 await send(
                     {"type": "websocket.close", "code": _WS_CLOSE_POLICY_VIOLATION}
                 )
diff --git a/paskia/fastapi/ws.py b/paskia/fastapi/ws.py
index af83134..fb70537 100644
--- a/paskia/fastapi/ws.py
+++ b/paskia/fastapi/ws.py
@@ -148,9 +148,7 @@ async def websocket_authenticate(
                 await ws.send_json({"status": 400, "detail": "Invalid redirect_uri"})
                 return
             # Store as the only allowed redirect URI
-            db.update_oid_client(
-                client_uuid, redirect_uris=[redirect_uri]
-            )
+            db.update_oid_client(client_uuid, redirect_uris=[redirect_uri])
             # Reload client to get updated redirect_uris
             oidc_client = db.data().oidc.clients.get(client_uuid)
         elif redirect_uri not in oidc_client.redirect_uris:
diff --git a/tests/test_admin.py b/tests/test_admin.py
index 694f71e..e371485 100644
--- a/tests/test_admin.py
+++ b/tests/test_admin.py
@@ -1830,7 +1830,7 @@ class TestDomains:
         assert domain["rp_id"] == "localhost"
         assert domain["origins"] == {}
         assert domain["related"] == {}
-        assert domain["effective_auth_host"] is None
+        assert domain["auth_host"] is None
         assert domain["site_url"] == "http://localhost:4401"
 
     @pytest.mark.asyncio
@@ -1945,9 +1945,6 @@ class TestDomains:
         assert created["rp_name"] == "Example"
         assert created["related"] == {"unrelated-site.com": True}
 
-        # OIDC provider seeded for the new domain
-        assert db.data().oidc_for("example.com") is not None
-
         r = await client.delete("/auth/api/admin/domains/example.com", headers=headers)
         assert r.status_code == 200, r.text
         assert "example.com" not in db.data().config.domains
@@ -2102,14 +2099,15 @@ class TestDomains:
         assert r.status_code == 200, r.text
 
     @pytest.mark.asyncio
-    async def test_effective_auth_host_fallback(
+    async def test_no_cross_domain_auth_host_fallback(
         self,
         client: httpx.AsyncClient,
         session_token: str,
         test_user,
         test_credential,
     ):
-        """A domain without its own auth host uses the shared one in settings."""
+        """A domain without its own auth host reports none — there is no
+        cross-domain fallback to another domain's auth host."""
         headers = await self._set_auth_host(
             client, session_token, test_user, test_credential
         )
@@ -2118,9 +2116,15 @@ class TestDomains:
         )
         assert r.status_code == 200
 
-        # Settings on the example.com host report the shared effective auth host
+        # Settings on the example.com host report no auth host
         r = await client.get("/auth/api/settings", headers={"Host": "example.com"})
         assert r.status_code == 200
         assert r.json()["rp_id"] == "example.com"
-        assert r.json()["auth_host"] == "auth.localhost"
+        assert r.json()["auth_host"] is None
         assert r.json()["own_auth_host"] is None
+
+        # The localhost domain still reports its own auth host
+        r = await client.get("/auth/api/settings", headers={"Host": "auth.localhost"})
+        assert r.status_code == 200
+        assert r.json()["auth_host"] == "auth.localhost"
+        assert r.json()["own_auth_host"] == "auth.localhost"
diff --git a/tests/test_api.py b/tests/test_api.py
index 45dbabc..96b1892 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -731,10 +731,9 @@ class TestOidcUserInfoEndpoint:
         if store is None:
             raise RuntimeError("Test DB store is not initialized")
         with store.transaction("create_test_oidc_client"):
-            test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client
+            test_db.oidc.clients[oidc_client.uuid] = oidc_client
 
         access_token = oidjwt.create_access_token(
-            "localhost",
             issuer="http://localhost:4401",
             subject=test_user.uuid,
             audience=str(oidc_client.uuid),
diff --git a/tests/test_cli.py b/tests/test_cli.py
index dc29b68..61d9cde 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -110,17 +110,22 @@ def test_init_adds_domains_to_existing_database(run_cli, tmp_path):
     config = stored_config(tmp_path)
     assert list(config.domains) == ["company.com", "app.com", "pro.com"]
     assert config.domains["pro.com"].rp_name == "Pro Corp"
-    # OIDC providers seeded for the added domains
-    assert set(converted_oidc(tmp_path)) == {"company.com", "app.com", "pro.com"}
 
 
-def converted_oidc(tmp_path):
+def test_init_seeds_one_global_oidc_key(run_cli, tmp_path):
+    """OIDC is instance-global: init seeds a single signing key."""
+    run_cli("init", "company.com")
+    run_cli("init", "app.com")
+    assert converted_oidc_key(tmp_path) is not None
+
+
+def converted_oidc_key(tmp_path):
     async def _read():
         new_db = DB()
         kanta = Kanta(str(tmp_path / "paskia.kantadb"), new_db)
         await kanta.open(readonly=True)
         try:
-            return set(kanta.data.oidc)
+            return kanta.data.oidc.key
         finally:
             await kanta.close()
 
diff --git a/tests/test_domains.py b/tests/test_domains.py
index 6675164..11ee934 100644
--- a/tests/test_domains.py
+++ b/tests/test_domains.py
@@ -26,9 +26,18 @@ from paskia.db.legacy import (
 )
 from paskia.db.lifecycle import format_log_uuid
 from paskia.db.operations import DB
-from paskia.db.structs import Client, Config, Credential, DomainConfig, OriginEntry
+from paskia.db.structs import (
+    OIDC,
+    Client,
+    Config,
+    Credential,
+    DomainConfig,
+    OriginEntry,
+    Session,
+)
 from paskia.fastapi.dispatch import DispatchMiddleware
 from paskia.sansio import Passkey
+from paskia.util.crypto import hash_secret
 
 # -------------------------------------------------------------------------
 # Registry construction helpers
@@ -143,16 +152,26 @@ class TestResolve:
         assert reg.resolve("") is None
         assert reg.resolve(None) is None
 
-    def test_effective_auth_host_fallback(self):
+    def test_auth_host_is_per_domain_no_fallback(self):
         reg = build_registry(ROR_CONFIG.domains)
-        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({"a.com": DomainConfig(), "b.com": DomainConfig()})
-        assert reg2.effective_auth_host(reg2.get("a.com")) is None
+        assert reg.get("company.com").own_auth_host == "auth.company.com"
+        # pro.com has no own auth host and there is no cross-domain fallback
+        assert reg.get("pro.com").own_auth_host is None
+
+    def test_shared_auth_host_resolves_best_suffix(self):
+        """Domains may share an auth host (nested rp-ids); the longest
+        rp-id suffix match wins, first configured as tiebreak."""
+        reg = build_registry(
+            {
+                "com": DomainConfig(
+                    origins={"auth.company.com": OriginEntry(auth_host=True)}
+                ),
+                "company.com": DomainConfig(
+                    origins={"auth.company.com": OriginEntry(auth_host=True)}
+                ),
+            }
+        )
+        assert reg.resolve("auth.company.com").rp_id == "company.com"
 
 
 # -------------------------------------------------------------------------
@@ -255,13 +274,39 @@ class TestValidateConfig:
                 )
             )
 
-    def test_related_origin_inside_other_domain(self):
-        with pytest.raises(ValueError, match="falls inside domain"):
+    def test_related_origin_may_fall_inside_other_domain(self):
+        """A related origin at/inside another domain's rp-id is allowed.
+        The related listing wins dispatch over suffix matching (a host that
+        *is* a configured rp-id always serves its own domain)."""
+        config = Config(
+            domains={
+                "a.com": DomainConfig(related={"app.b.com": True}),
+                "b.com": DomainConfig(),
+            }
+        )
+        domains.validate_config(config)
+        reg = build_registry(config.domains)
+        assert reg.resolve("app.b.com").rp_id == "a.com"
+        assert reg.resolve("b.com").rp_id == "b.com"
+
+    def test_related_origin_shared_when_covered_by_rp_id(self):
+        """Two domains may list the same related host when it falls inside
+        a configured rp-id; otherwise the collision is rejected."""
+        domains.validate_config(
+            Config(
+                domains={
+                    "a.com": DomainConfig(related={"app.b.com": True}),
+                    "c.com": DomainConfig(related={"app.b.com": True}),
+                    "b.com": DomainConfig(),
+                }
+            )
+        )
+        with pytest.raises(ValueError, match="configured for both"):
             domains.validate_config(
                 Config(
                     domains={
-                        "a.com": DomainConfig(related={"app.b.com": True}),
-                        "b.com": DomainConfig(),
+                        "a.com": DomainConfig(related={"shared.com": True}),
+                        "c.com": DomainConfig(related={"shared.com": True}),
                     }
                 )
             )
@@ -369,7 +414,8 @@ class TestSanitizeConfig:
         assert config.domains["a.com"].origins == {"auth.a.com": True}
         assert any("collides" in w for w in warnings)
 
-    def test_related_colliding_with_other_domain_dropped(self):
+    def test_related_inside_other_domain_kept(self):
+        """A related origin falling inside another domain's rp-id is kept."""
         config, _ = domains.sanitize_config(
             Config(
                 domains={
@@ -378,7 +424,21 @@ class TestSanitizeConfig:
                 }
             )
         )
-        assert config.domains["a.com"].related == {}
+        assert config.domains["a.com"].related == {"app.b.com": True}
+
+    def test_related_claimed_twice_first_wins(self):
+        """Two non-owner domains claiming one related host: first wins."""
+        config, warnings = domains.sanitize_config(
+            Config(
+                domains={
+                    "a.com": DomainConfig(related={"shared.com": True}),
+                    "b.com": DomainConfig(related={"shared.com": True}),
+                }
+            )
+        )
+        assert config.domains["a.com"].related == {"shared.com": True}
+        assert config.domains["b.com"].related == {}
+        assert any("first domain wins" in w for w in warnings)
 
     def test_no_domains_is_fatal(self):
         with pytest.raises(ValueError, match="No servable domain"):
@@ -440,14 +500,23 @@ class TestOriginValidation:
         with pytest.raises(ValueError, match="not allowed"):
             p.validate_origin("http://app.example.com:8080")
 
-    def test_star_entry_matches_any_scheme_and_port(self):
-        """The bare '*' entry allows anything within the rp-id domain."""
+    def test_star_entry_is_wildcard_shorthand(self):
+        """The bare '*' entry is shorthand for '*.{rp-id}' — https only."""
         p = Passkey(rp_id="example.com", origins=["*"])
         assert p.validate_origin("https://example.com")
-        assert p.validate_origin("http://app.example.com:8080")
+        assert p.validate_origin("https://app.example.com")
+        with pytest.raises(ValueError, match="not allowed"):
+            p.validate_origin("http://app.example.com:8080")
         with pytest.raises(ValueError, match="not allowed"):
             p.validate_origin("https://other.com")
 
+    def test_localhost_wildcard_matches_any_scheme_and_port(self):
+        """Under localhost, wildcards match any scheme and any port."""
+        p = Passkey(rp_id="localhost", origins=["*"])
+        assert p.validate_origin("http://localhost:8080")
+        assert p.validate_origin("http://app.localhost:3000")
+        assert p.validate_origin("https://localhost")
+
     def test_exact_entry_matches_scheme_and_port(self):
         p = Passkey(rp_id="localhost", origins=["http://localhost:4403"])
         assert p.validate_origin("http://localhost:4403")
@@ -555,15 +624,16 @@ class TestDispatchMiddleware:
         assert stub.scope["state"]["domain"].rp_id == "company.com"
 
     @pytest.mark.asyncio
-    async def test_ws_cross_domain_requires_effective_auth_host(self):
+    async def test_ws_cross_domain_requires_origin_own_auth_host(self):
         build_registry(ROR_CONFIG.domains)
-        # pro.com page connecting to the shared auth host: allowed, pro domain
+        # pro.com has no own auth host: its page may only connect to pro.com
+        # hosts — the company.com auth host does not serve foreign domains
         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"]["domain"].rp_id == "pro.com"
+        assert stub.scope is None
+        assert sent == [{"type": "websocket.close", "code": 1008}]
 
         # pro.com page connecting to some other host: closed pre-accept
         stub, sent = await drive_ws(
@@ -573,6 +643,35 @@ class TestDispatchMiddleware:
         assert stub.scope is None
         assert sent == [{"type": "websocket.close", "code": 1008}]
 
+    @pytest.mark.asyncio
+    async def test_ws_cross_domain_via_own_auth_host(self):
+        """On a shared auth host (nested rp-ids), the WS Origin selects the
+        domain: plain HTTP resolves to the longest-suffix claimant, but a
+        WebSocket from another claimant's page is dispatched by Origin."""
+        build_registry(
+            {
+                "com": DomainConfig(
+                    origins={"auth.company.com": OriginEntry(auth_host=True)}
+                ),
+                "company.com": DomainConfig(
+                    origins={"auth.company.com": OriginEntry(auth_host=True)}
+                ),
+            }
+        )
+        # Host alone resolves to company.com (longest suffix)
+        stub, sent = await drive_ws(
+            DispatchMiddleware(StubApp()), [(b"host", b"auth.company.com")]
+        )
+        assert stub.scope["state"]["domain"].rp_id == "company.com"
+        # A page on com (the other claimant) is accepted: the Host is its
+        # own auth host, and the Origin selects its domain
+        stub, sent = await drive_ws(
+            DispatchMiddleware(StubApp()),
+            [(b"host", b"auth.company.com"), (b"origin", b"https://com")],
+        )
+        assert sent == []
+        assert stub.scope["state"]["domain"].rp_id == "com"
+
     @pytest.mark.asyncio
     async def test_ws_unknown_origin_uses_host_domain(self):
         build_registry(ROR_CONFIG.domains)
@@ -614,25 +713,40 @@ class TestAuthCodeDomainBinding:
         assert response.status_code == 401
 
     @pytest.mark.asyncio
-    async def test_oidc_code_rejected_on_other_domain(
-        self, client: httpx.AsyncClient, test_db: DB
+    async def test_oidc_code_is_host_independent(
+        self, client: httpx.AsyncClient, test_db: DB, test_user, test_credential
     ):
+        """OIDC codes carry no domain binding: the provider is
+        instance-global, so a code is redeemable at any host."""
         oidc_client, secret = Client.create(
             name="Test Client",
             redirect_uris=["https://client.example/callback"],
             client_secret="topsecret",
         )
+        token = "doesnotmatter1234"
+        session = Session.create(
+            user=test_user.uuid,
+            credential=test_credential.uuid,
+            key=hash_secret("oidc", token),
+            host="other.com",
+            ip="127.0.0.1",
+            user_agent="pytest",
+            validated=datetime.now(UTC),
+            client=oidc_client.uuid,
+            rp_id="other.com",
+            issuer="https://other.com",
+        )
         store = test_db._store
-        with store.transaction("create_test_oidc_client"):
-            test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client
+        with store.transaction("seed_oidc_session"):
+            test_db.oidc.clients[oidc_client.uuid] = oidc_client
+            test_db.sessions[session.key] = session
 
         code = authcode.store_oidc(
             authcode.OIDCCode(
-                session_key="doesnotmatter1234",
+                session_key=token,
                 created=datetime.now(UTC),
                 redirect_uri="https://client.example/callback",
                 scope="openid",
-                rp_id="other.com",
             )
         )
         response = await client.post(
@@ -646,9 +760,8 @@ class TestAuthCodeDomainBinding:
             },
             headers={"Host": "localhost:4401"},
         )
-        assert response.status_code == 400
-        assert response.json()["error"] == "invalid_grant"
-        assert "domain" in response.json()["error_description"]
+        assert response.status_code == 200
+        assert response.json()["access_token"]
 
 
 # -------------------------------------------------------------------------
@@ -700,6 +813,7 @@ class TestLegacyConversion:
                     user_agent="pytest",
                     validated=datetime.now(UTC),
                 )
+                kanta.data.oidc = OIDC(key=b"legacy-signing-key")
             await kanta.close()
 
         asyncio.run(_write())
@@ -707,13 +821,14 @@ class TestLegacyConversion:
         config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
         domain = config.domains["example.com"]
         assert domain.rp_name == "Example"
-        # A legacy wildcard over the rp-id itself becomes the bare '*'
-        assert domain.origins == {"app.example.com": True, "*": True}
+        # Legacy wildcard origins convert as-is (https-only outside localhost)
+        assert domain.origins == {"app.example.com": True, "*.example.com": True}
 
         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"}
+        # The legacy OIDC provider carries over as the instance-global one
+        assert converted.oidc.key == b"legacy-signing-key"
 
 
 # -------------------------------------------------------------------------
@@ -723,15 +838,14 @@ class TestLegacyConversion:
 
 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>"
+        assert format_log_uuid(b"raw-key-material", "oidc.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
+        # The path component itself must stay visible ("oidc.key = <hidden>")
+        assert format_log_uuid("key", "oidc.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", "oidc.clients") is None
         assert format_log_uuid("not-a-uuid", "config.domains") is None