LeoVasanko
f2e6f5784e
Finish the realm→domain terminology removal across source, tests, e2e and docs. The stored config drops all lists: Config.domains is keyed by rp-id, DomainConfig.origins/related are objects keyed by host (https:// omitted), values True or OriginEntry(auth_host=True). The default/primary domain concept is gone; ordering is display-time. Tests and e2e updated to the new API shapes (not run). Database re-migrated from the legacy backup into the new format.
731 lines
27 KiB
Python
731 lines
27 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_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_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"],
|
|
)
|
|
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"
|
|
assert domain.origins == {"app.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"}
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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
|