LeoVasanko
f5bd0a469c
Serving never refuses to start because of stored realm config: the registry build sanitizes best-effort and warns — misfiled origin entries are reclassified (a cross-domain origins entry is served as a related origin) or dropped, collisions resolve first-come-wins, over-cap related lists truncate, unsalvageable realms are skipped. Fixing the stored config stays the admin interface's job, and it stays reachable on any working realm. Only a config with no servable realm at all is fatal. Admin realm writes stay strict and gain self-lockout guards: an update that would leave the admin's current host unable to run ceremonies for the realm they are on is refused (unless an auth host takes over ceremonies), and deleting the realm currently in use is refused.
698 lines
25 KiB
Python
698 lines
25 KiB
Python
"""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
|
|
from paskia.sansio import Passkey
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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"],
|
|
related_origins=["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",
|
|
related_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",
|
|
related_origins=[f"https://app{i}.com" 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"):
|
|
realms.validate_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(rp_id="a.com", origins=["https://elsewhere.com"])
|
|
]
|
|
)
|
|
)
|
|
|
|
def test_related_origin_inside_own_realm_rejected(self):
|
|
"""Subdomains of the rp-id are covered already; listing is an error."""
|
|
with pytest.raises(ValueError, match="within the rp-id domain"):
|
|
realms.validate_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(
|
|
rp_id="a.com", related_origins=["https://app.a.com"]
|
|
)
|
|
]
|
|
)
|
|
)
|
|
|
|
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",
|
|
related_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", related_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"),
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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 = realms.sanitize_config(
|
|
Config(
|
|
realms=[RealmConfig(rp_id="localhost", origins=["https://example.com"])]
|
|
)
|
|
)
|
|
realm = config.realms[0]
|
|
assert realm.origins is None
|
|
assert realm.related_origins == ["https://example.com"]
|
|
assert any("related origin" in w for w in warnings)
|
|
realms.validate_config(config) # sanitized config is strict-clean
|
|
|
|
def test_malformed_origin_dropped(self):
|
|
config, warnings = realms.sanitize_config(
|
|
Config(realms=[RealmConfig(rp_id="a.com", origins=["not a url"])])
|
|
)
|
|
assert config.realms[0].origins is None
|
|
assert warnings
|
|
|
|
def test_invalid_rp_id_realm_dropped(self):
|
|
config, warnings = realms.sanitize_config(
|
|
Config(
|
|
realms=[RealmConfig(rp_id="not a domain!"), RealmConfig(rp_id="ok.com")]
|
|
)
|
|
)
|
|
assert [r.rp_id for r in config.realms] == ["ok.com"]
|
|
assert any("dropped" in w for w in warnings)
|
|
|
|
def test_duplicate_rp_id_first_wins(self):
|
|
config, _warnings = realms.sanitize_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(rp_id="a.com", rp_name="First"),
|
|
RealmConfig(rp_id="a.com"),
|
|
]
|
|
)
|
|
)
|
|
assert len(config.realms) == 1
|
|
assert config.realms[0].rp_name == "First"
|
|
|
|
def test_related_inside_own_domain_dropped(self):
|
|
config, _ = realms.sanitize_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(rp_id="a.com", related_origins=["https://app.a.com"])
|
|
]
|
|
)
|
|
)
|
|
assert config.realms[0].related_origins is None
|
|
|
|
def test_cap_exceeded_truncated(self):
|
|
config, warnings = realms.sanitize_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(
|
|
rp_id="a.com",
|
|
related_origins=[f"https://app{i}.com" for i in range(6)],
|
|
)
|
|
]
|
|
)
|
|
)
|
|
assert len(config.realms[0].related_origins) == 5
|
|
assert any("maximum" in w for w in warnings)
|
|
|
|
def test_auth_host_outside_domain_ignored(self):
|
|
config, warnings = realms.sanitize_config(
|
|
Config(realms=[RealmConfig(rp_id="a.com", auth_host="https://auth.b.com")])
|
|
)
|
|
assert config.realms[0].auth_host is None
|
|
assert any("auth host ignored" in w for w in warnings)
|
|
|
|
def test_auth_host_colliding_with_rp_id_ignored(self):
|
|
config, warnings = realms.sanitize_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(rp_id="a.com", auth_host="https://auth.a.com"),
|
|
RealmConfig(rp_id="auth.a.com"),
|
|
]
|
|
)
|
|
)
|
|
assert config.realms[0].auth_host is None
|
|
assert any("collides with an rp-id" in w for w in warnings)
|
|
|
|
def test_related_colliding_with_other_realm_dropped(self):
|
|
config, _ = realms.sanitize_config(
|
|
Config(
|
|
realms=[
|
|
RealmConfig(rp_id="a.com", related_origins=["https://app.b.com"]),
|
|
RealmConfig(rp_id="b.com"),
|
|
]
|
|
)
|
|
)
|
|
assert config.realms[0].related_origins is None
|
|
|
|
def test_no_realms_is_fatal(self):
|
|
with pytest.raises(ValueError, match="realm"):
|
|
realms.sanitize_config(Config(realms=[]))
|
|
with pytest.raises(ValueError, match="No servable realm"):
|
|
realms.sanitize_config(Config(realms=[RealmConfig(rp_id="not a domain!")]))
|
|
|
|
def test_build_tolerates_and_serves(self):
|
|
# Cross-domain entry stored in origins: served as a related origin
|
|
reg = build_registry(
|
|
RealmConfig(rp_id="localhost", origins=["https://example.com"])
|
|
)
|
|
assert reg.warnings
|
|
realm = reg.get("localhost")
|
|
assert realm.related_origins == ["https://example.com"]
|
|
realm.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_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_realm_wires_both_lists(self):
|
|
reg = build_registry(*ROR_CONFIG.realms)
|
|
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.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
|