Files
paskia/tests/test_domains.py
T
LeoVasanko 7ff8869e1d CLI: positional rp-id/rp-name; init adds domains to an existing database
- 'paskia init [rp-id] [rp-name]' and 'paskia migrate [rp-id]' are now
  positional; comma separation and the --rp-id/--rp-name flags are gone.
- With an existing paskia.kantadb, init adds the rp-id as a new domain
  (seeding its OIDC provider) or updates an existing domain's rp-name.
- Origin allow-list semantics clarified: the bare '*' entry allows
  anything within the rp-id domain on any scheme and port (also the
  empty-list default and its display in the admin UI, replacing the
  synthetic '*.rp-id' row); '*.x' wildcards are https-only; exact entries
  match scheme, host and port. Legacy '*.rp-id' wildcards migrate to '*'
  to preserve their any-scheme meaning.
2026-09-07 03:07:38 +00:00

771 lines
29 KiB
Python

"""Tests for the multi-domain machinery: registry resolution, config
validation, ASGI dispatch, domain 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, domains
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, DomainConfig, OriginEntry
from paskia.fastapi.dispatch import DispatchMiddleware
from paskia.sansio import Passkey
# -------------------------------------------------------------------------
# Registry construction helpers
# -------------------------------------------------------------------------
def build_registry(configs: dict[str, DomainConfig]) -> domains.DomainRegistry:
"""Build and install a registry from domain configs (listen unset)."""
domains.configure(listen=None)
return domains.init_registry(Config(domains=configs))
ROR_CONFIG = Config(
domains={
"company.com": DomainConfig(
rp_name="Company",
origins={"auth.company.com": OriginEntry(auth_host=True)},
related={"app.com": True},
),
"pro.com": DomainConfig(rp_name="Pro"),
}
)
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.domains)
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.domains)
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(
{"example.com": DomainConfig(), "sub.example.com": DomainConfig()}
)
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.domains)
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.domains)
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.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
# -------------------------------------------------------------------------
# Cross-domain configuration validation
# -------------------------------------------------------------------------
class TestValidateConfig:
def test_valid(self):
domains.validate_config(ROR_CONFIG)
def test_related_origin_cap(self):
domains.validate_config(
Config(
domains={
"company.com": DomainConfig(
related={f"app{i}.com": True for i in range(5)}
)
}
)
)
with pytest.raises(ValueError, match="related origins"):
domains.validate_config(
Config(
domains={
"company.com": DomainConfig(
related={f"app{i}.com": True for i in range(6)}
)
}
)
)
def test_origin_outside_rp_id_rejected(self):
"""In-domain origins are an allow-list; cross-domain needs related."""
with pytest.raises(ValueError, match="outside the rp-id domain"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"elsewhere.com": True})})
)
def test_related_origin_inside_own_domain_rejected(self):
"""Subdomains of the rp-id are covered already; listing is an error."""
with pytest.raises(ValueError, match="within the rp-id domain"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(related={"app.a.com": True})})
)
def test_wildcard_related_origin_rejected(self):
"""ROR entries are always individual origins; wildcards are meaningless."""
with pytest.raises(ValueError, match="wildcard"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(related={"*.b.com": True})})
)
def test_wildcard_origin_in_domain_accepted(self):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"*.a.com": True})})
)
with pytest.raises(ValueError, match="outside the rp-id domain"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"*.b.com": True})})
)
def test_wildcard_auth_host_rejected(self):
with pytest.raises(ValueError, match="cannot be the auth host"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"*.a.com": OriginEntry(auth_host=True)}
)
}
)
)
def test_star_origin_accepted_not_auth_host(self):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"*": True})})
)
with pytest.raises(ValueError, match="cannot be the auth host"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"*": OriginEntry(auth_host=True)}
)
}
)
)
def test_auth_host_collision(self):
with pytest.raises(ValueError, match="collides with a related origin"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.a.com": OriginEntry(auth_host=True)}
),
"b.com": DomainConfig(related={"auth.a.com": True}),
}
)
)
def test_related_origin_inside_other_domain(self):
with pytest.raises(ValueError, match="falls inside domain"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(related={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
)
def test_auth_host_must_not_collide_with_rp_id(self):
with pytest.raises(ValueError, match="collides with an rp-id"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"b.a.com": OriginEntry(auth_host=True)}
),
"b.a.com": DomainConfig(),
}
)
)
# -------------------------------------------------------------------------
# Best-effort serving: stored config sanitization
# -------------------------------------------------------------------------
class TestSanitizeConfig:
"""Serving never fails on stored config problems; it degrades + warns."""
def test_cross_domain_origin_moved_to_related(self):
config, warnings = domains.sanitize_config(
Config(domains={"localhost": DomainConfig(origins={"example.com": True})})
)
domain = config.domains["localhost"]
assert domain.origins == {}
assert domain.related == {"example.com": True}
assert any("related origin" in w for w in warnings)
domains.validate_config(config) # sanitized config is strict-clean
def test_malformed_origin_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(origins={"https://": True})})
)
assert config.domains["a.com"].origins == {}
assert warnings
def test_invalid_rp_id_domain_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"not a domain!": DomainConfig(), "ok.com": DomainConfig()})
)
assert list(config.domains) == ["ok.com"]
assert any("dropped" in w for w in warnings)
def test_related_inside_own_domain_dropped(self):
config, _ = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(related={"app.a.com": True})})
)
assert config.domains["a.com"].related == {}
def test_wildcard_related_origin_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(related={"*.b.com": True})})
)
assert config.domains["a.com"].related == {}
assert any("wildcard" in w for w in warnings)
domains.validate_config(config) # sanitized config is strict-clean
def test_cap_exceeded_truncated(self):
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
related={f"app{i}.com": True for i in range(6)}
)
}
)
)
assert len(config.domains["a.com"].related) == 5
assert any("maximum" in w for w in warnings)
def test_auth_host_outside_domain_becomes_related(self):
"""An auth-marked origin outside the rp-id degrades to a related origin."""
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.b.com": OriginEntry(auth_host=True)}
)
}
)
)
domain = config.domains["a.com"]
assert domain.origins == {}
assert domain.related == {"auth.b.com": True}
assert any("related origin" in w for w in warnings)
def test_auth_host_colliding_with_rp_id_cleared(self):
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.a.com": OriginEntry(auth_host=True)}
),
"auth.a.com": DomainConfig(),
}
)
)
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):
config, _ = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(related={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
)
assert config.domains["a.com"].related == {}
def test_no_domains_is_fatal(self):
with pytest.raises(ValueError, match="No servable domain"):
domains.sanitize_config(Config(domains={}))
with pytest.raises(ValueError, match="No servable domain"):
domains.sanitize_config(Config(domains={"not a domain!": DomainConfig()}))
def test_build_tolerates_and_serves(self):
# Cross-domain entry stored in origins: served as a related origin
reg = build_registry({"localhost": DomainConfig(origins={"example.com": True})})
assert reg.warnings
domain = reg.get("localhost")
assert domain.related_origins == ["https://example.com"]
domain.passkey.validate_origin("https://example.com")
# -------------------------------------------------------------------------
# Origin validation semantics (Passkey)
# -------------------------------------------------------------------------
class TestOriginValidation:
"""In-domain allow-list and related origins are separate concerns."""
def test_default_allows_whole_subtree(self):
p = Passkey(rp_id="example.com")
assert p.validate_origin("https://example.com") == "https://example.com"
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_allow_list_restricts_subtree(self):
p = Passkey(rp_id="example.com", origins=["https://app.example.com"])
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com")
def test_related_origins_are_additive(self):
p = Passkey(rp_id="example.com", related_origins=["https://app2.com"])
assert p.validate_origin("https://app.example.com") # subtree stays open
assert p.validate_origin("https://app2.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_wildcard_entry_matches_subtree(self):
p = Passkey(rp_id="example.com", origins=["*.example.com"])
assert p.validate_origin("https://example.com")
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_wildcard_is_https_only(self):
"""A '*.example.com' entry does not fall back to other schemes."""
p = Passkey(rp_id="example.com", origins=["*.example.com"])
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://example.com")
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."""
p = Passkey(rp_id="example.com", origins=["*"])
assert p.validate_origin("https://example.com")
assert p.validate_origin("http://app.example.com:8080")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
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")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://localhost:4403")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://localhost:4404")
def test_sub_wildcard_matches_only_its_subtree(self):
p = Passkey(rp_id="example.com", origins=["*.app.example.com"])
assert p.validate_origin("https://app.example.com")
assert p.validate_origin("https://www.app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.example.com")
def test_wildcard_related_origin_rejected(self):
with pytest.raises(ValueError, match="wildcard"):
Passkey(rp_id="example.com", related_origins=["*.other.com"])
def test_related_origins_combined_with_allow_list(self):
p = Passkey(
rp_id="example.com",
origins=["https://app.example.com"],
related_origins=["https://app2.com"],
)
assert p.validate_origin("https://app.example.com")
assert p.validate_origin("https://app2.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.example.com")
def test_constructor_rejects_mixed_up_fields(self):
with pytest.raises(ValueError, match="related origin"):
Passkey(rp_id="example.com", origins=["https://app2.com"])
with pytest.raises(ValueError, match="within the rp-id domain"):
Passkey(rp_id="example.com", related_origins=["https://app.example.com"])
def test_domain_wires_both_lists(self):
reg = build_registry(ROR_CONFIG.domains)
p = reg.get("company.com").passkey
assert p.validate_origin("https://app.com") # related origin
assert p.validate_origin("https://auth.company.com") # allow-listed
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.company.com") # not allow-listed
# -------------------------------------------------------------------------
# ASGI dispatch
# -------------------------------------------------------------------------
class TestDispatchMiddleware:
@pytest.mark.asyncio
async def test_http_unknown_host_421(self):
build_registry(ROR_CONFIG.domains)
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_domain(self):
build_registry(ROR_CONFIG.domains)
stub, _sent = await drive_http(
DispatchMiddleware(StubApp()), [(b"host", b"app.com.")]
)
assert stub.scope is not None
assert stub.scope["state"]["domain"].rp_id == "company.com"
@pytest.mark.asyncio
async def test_http_current_domain_set_inside_request(self):
build_registry(ROR_CONFIG.domains)
seen = {}
async def app(scope, receive, send):
seen["domain"] = domains.current_domain()
await drive_http(DispatchMiddleware(app), [(b"host", b"pro.com")])
assert seen["domain"].rp_id == "pro.com"
# Contextvar is reset after the request; with several domains there
# is no implicit current domain outside a request context.
with pytest.raises(RuntimeError, match="request context"):
domains.current_domain()
@pytest.mark.asyncio
async def test_ws_unknown_host_closed(self):
build_registry(ROR_CONFIG.domains)
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_domain_origin(self):
build_registry(ROR_CONFIG.domains)
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"]["domain"].rp_id == "company.com"
@pytest.mark.asyncio
async def test_ws_cross_domain_requires_effective_auth_host(self):
build_registry(ROR_CONFIG.domains)
# pro.com page connecting to the shared auth host: allowed, pro domain
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"
# 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_domain(self):
build_registry(ROR_CONFIG.domains)
# Missing origin
stub, _ = await drive_ws(DispatchMiddleware(StubApp()), [(b"host", b"pro.com")])
assert stub.scope["state"]["domain"].rp_id == "pro.com"
# Unknown origin: host domain 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"]["domain"].rp_id == "pro.com"
# -------------------------------------------------------------------------
# Domain binding of auth codes
# -------------------------------------------------------------------------
class TestAuthCodeDomainBinding:
@pytest.mark.asyncio
async def test_cookie_code_rejected_on_other_domain(
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_domain(
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 "domain" 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_domain_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", "*.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")
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}
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.domains") is None
# -------------------------------------------------------------------------
# Bootstrap caveat: admin credential is checked on the configured domains
# -------------------------------------------------------------------------
class TestBootstrapCaveat:
@pytest.mark.asyncio
async def test_admin_without_credentials_gets_link(
self, test_db: DB, domain_registry
):
assert await check_admin_credentials() is True
@pytest.mark.asyncio
async def test_admin_with_domain_credential_ok(
self, test_db: DB, domain_registry, test_user, test_credential
):
assert await check_admin_credentials() is False
@pytest.mark.asyncio
async def test_admin_with_only_unconfigured_domain_credential_gets_link(
self, test_db: DB, domain_registry, test_user
):
"""A passkey under an rp-id outside the config 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