LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Best-effort serve for bad stored config + admin self-lockout guards

Browse Source 
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.
This commit is contained in:
LeoVasanko
2026-09-06 22:52:00 +00:00
parent b9e6f4bc27
commit f5bd0a469c
6 changed files with 438 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/MultiSite.md
+31 -16
View File
@@ -172,9 +172,9 @@ class Config(msgspec.Struct, omit_defaults=True):
outside it, are capped (default 5), and must not collide with another
realm's rp-id/auth-host/related origins nor fall inside another
realm's domain. Misfiled entries (cross-domain in `origins`, in-domain
in `related_origins`) are rejected. These rules are enforced both at
startup and at admin write time. Origins are never _implicitly_
cross-domain.
in `related_origins`) are rejected. These rules are enforced at admin
write time; at startup the stored config is sanitized best-effort
instead (§3.2). Origins are never _implicitly_ cross-domain.
### 3.2 CLI: bootstrap (`paskia init`) vs. serve (`paskia`)
@@ -201,12 +201,18 @@ instance:
selects `<rp-id>.paskiadb` by name; the others are left in place.
- **`paskia`** — serve. Takes **no realm options**; only `--listen`
(per-run override of stored `Config.listen`, never persisted). Startup:
open `paskia.kantadb` → validate the stored realm set cross-realm
(rp-ids distinct; auth hosts distinct from each other and from every
rp-id; related origins capped and collision-free) → build the realm
registry → serve. The serve command never converts databases: with no
`paskia.kantadb`, the startup error points at `paskia init`, or at
`paskia migrate` when legacy `*.paskiadb` candidates are present.
open `paskia.kantadb` → sanitize the stored realm set best-effort →
build the realm registry → serve. Sanitization never refuses to start:
misfiled origin entries are reclassified (a cross-domain `origins`
entry is served as a related origin) or dropped, colliding
auth-hosts/related origins resolve first-come-wins, over-cap related
lists truncate, and unsalvageable realms are skipped — each producing
a startup warning, because fixing the stored config is the admin
interface's job and it must stay reachable to do so. Only a config
with no servable realm at all is fatal. The serve command never
converts databases: with no `paskia.kantadb`, the startup error points
at `paskia init`, or at `paskia migrate` when legacy `*.paskiadb`
candidates are present.
Nested rp-ids are allowed (longest-suffix dispatch determinism). Adding a
child rp-id moves **no data** — users are global; only new ceremonies
@@ -375,9 +381,16 @@ effective_auth_host(realm) = realm.auth_host or first_configured_auth_host or No
Changing a realm's rp-id itself is **not supported** (it would orphan
every credential stamped with the old rp-id) — delete and recreate
instead.
- Delete: refused for the last remaining realm and while any credential
- Delete: refused for the last remaining realm, while any credential
carries the realm's rp-id (re-enroll or delete those credentials
first); cascades nothing else (users/orgs are global).
first), and for the realm the admin is currently using; cascades
nothing else (users/orgs are global).
- **Lockout guard**: an update that would make the admin's current
host unable to run passkey ceremonies for the realm they are on is
refused (unless an auth host takes over ceremonies — it is always
allowed). Fixing a broken stored config is always possible: serve
sanitizes it best-effort (§3.2) so the admin interface stays
reachable on a working realm.
- The admin UI has a Realms section with a table (rp-id, name,
effective auth host, sign-in site and related domain counts),
per-row edit/delete and an add-realm dialog. The dialog edits the
@@ -431,9 +444,9 @@ effective_auth_host(realm) = realm.auth_host or first_configured_auth_host or No
background cleanup task (DB is global).
- The kanta bootstrap hook only ever fires for a database created by
`paskia init` or `paskia migrate`; the serve command never bootstraps.
- The registry is built from the stored `Config` after open; per-realm
`Passkey` instances constructed (each realm's origins validated at
startup — fail-fast, including related-origin cap checks).
- The registry is built from the stored `Config` after open, sanitized
best-effort (§3.2) so startup never fails on config content; per-realm
`Passkey` instances are constructed from the sanitized config.
- The admin-credential check runs at serve startup and reprints a usable
registration link when the admin lacks a credential under the default
realm (§9).
@@ -476,8 +489,10 @@ effective_auth_host(realm) = realm.auth_host or first_configured_auth_host or No
origins sharing an rp-id share one security boundary — do not mix trust
levels within a realm.
- **Realm administration**: realm create/update/delete is gated on
`auth:admin` — deployment-wide by design; validation runs on every
write, not just at startup.
`auth:admin` — deployment-wide by design. Writes are strictly
validated (cross-realm rules plus self-lockout guards); startup never
refuses a stored config — it sanitizes with warnings so the admin
interface stays reachable to fix problems.
- **Passkeys**: rp-id binding browser-enforced and server-recorded;
ceremonies, credential scans, exclude/allow lists all scoped to the
origin realm's rp-id. No cross-realm oracle in the scan.
paskia/__main__.py
+11 -5
View File
@@ -168,14 +168,20 @@ def cmd_serve(args: argparse.Namespace) -> None:
raise SystemExit(f"Database {db_path} not found — run 'paskia init' first.")
config = _load_stored_config(db_path)
try:
validate_config(config)
except ValueError as e:
raise SystemExit(f"Invalid stored configuration: {e}") from e
listen = _split_multi(args.listen) or config.listen
configure_realms(listen=listen)
registry = build_registry(config)
try:
registry = build_registry(config)
except ValueError as e:
raise SystemExit(f"Invalid stored configuration: {e}") from e
for warning in registry.warnings:
# Serving is best-effort; fixing the stored config is the admin's
# job via the admin interface on any working realm.
print(
f"⚠️ Config problem (fix via the admin interface): {warning}",
file=sys.stderr,
)
# Pass process-global serve parameters to the server process(es)
os.environ["PASKIA_CONFIG"] = msgspec.json.encode(
paskia/fastapi/admin/realms.py
+43
View File
@@ -16,6 +16,7 @@ from paskia.fastapi import authz
from paskia.fastapi.admin.errors import install_error_handlers
from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE
from paskia.sansio import Passkey
from paskia.util import hostutil, oidjwt
from paskia.util.apistructs import ApiRealm
@@ -65,6 +66,41 @@ def _rebuild_registry() -> None:
realms.init_registry(db.data().config)
def _check_not_locking_self_out(
request: Request,
realm: RealmConfig,
) -> None:
"""Refuse realm changes that lock the admin out of their current host.
Applies when the admin edits the realm they are currently using and the
new config has no auth host (with an auth host, ceremonies move there
and it is always allowed). The admin's current host must remain able to
run passkey ceremonies under the new config.
"""
current: realms.Realm = request.state.realm
if realm.rp_id != current.rp_id or realm.auth_host:
return
raw_host = (request.headers.get("host") or "").rstrip(".")
if not raw_host:
return
probe = Passkey(
rp_id=realm.rp_id,
origins=realm.origins,
related_origins=realm.related_origins,
)
for scheme in ("https", "http"):
try:
probe.validate_origin(f"{scheme}://{raw_host}")
return # Current host still works — no lockout
except ValueError:
pass
raise ValueError(
f"This change would lock you out: '{raw_host}' could no longer "
f"run passkey ceremonies for realm '{realm.rp_id}'. Add it to the "
"allowed sign-in sites (or configure an auth host) before saving."
)
@app.get("/")
async def admin_list_realms(request: Request, auth=AUTH_COOKIE):
"""List all realms with derived URLs (master admin only)."""
@@ -152,6 +188,7 @@ async def admin_update_realm(
listen=config.listen,
)
realms.validate_config(would_be)
_check_not_locking_self_out(request, updated)
db.update_realm(
rp_id,
@@ -175,6 +212,12 @@ async def admin_delete_realm(
ctx = await authz.verify(
auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m"
)
current: realms.Realm = request.state.realm
if rp_id == current.rp_id:
raise ValueError(
"Cannot delete the realm you are currently using — authenticate "
"on another realm first"
)
db.delete_realm(rp_id, ctx=ctx)
_rebuild_registry()
oidjwt.clear_key(rp_id)
paskia/realms.py
+171 -3
View File
@@ -11,6 +11,7 @@ contextvar set by the dispatch middleware.
from __future__ import annotations
import contextvars
import logging
import os
from fastapi_vue.hostutil import parse_endpoints
@@ -19,6 +20,8 @@ from paskia.db.structs import Config, RealmConfig
from paskia.util import hostutil
from paskia.util.constants import DEFAULT_PORT
logger = logging.getLogger(__name__)
# Maximum number of related (non-subdomain) origins per realm. WebAuthn
# Related Origin Requests require browsers to support at least 5 labels.
DEFAULT_RELATED_ORIGIN_CAP = 5
@@ -94,6 +97,7 @@ class RealmRegistry:
self._by_rp_id = {r.rp_id: r for r in realms}
self._auth_hosts: dict[str, Realm] = {}
self._related_hosts: dict[str, Realm] = {}
self.warnings: list[str] = []
for realm in realms:
if own := realm.own_auth_host:
self._auth_hosts[hostutil.normalize_host(own) or own] = realm
@@ -232,6 +236,160 @@ def validate_config(
)
def sanitize_config(
config: Config, *, related_origin_cap: int = DEFAULT_RELATED_ORIGIN_CAP
) -> tuple[Config, list[str]]:
"""Best-effort repair of a stored configuration for serving.
Serving must never fail because of stored realm config: fixing it is
the admin's job via the admin UI, which is reachable only on a running
server. Returns a sanitized copy (the stored config is left untouched)
plus a warning for every degradation made. The result always passes
``validate_config``.
"""
warnings: list[str] = []
if not config.realms:
raise ValueError("At least one realm (rp-id) is required")
def warn(msg: str) -> None:
warnings.append(msg)
realms: list[RealmConfig] = []
seen_rp_ids: set[str] = set()
for realm in config.realms:
rp_id = realm.rp_id
try:
hostutil.validate_rp_id(rp_id)
except ValueError as e:
warn(f"Realm dropped: {e}")
continue
if rp_id in seen_rp_ids:
warn(f"Duplicate realm '{rp_id}' dropped (first entry kept)")
continue
seen_rp_ids.add(rp_id)
auth_host = realm.auth_host
if auth_host:
try:
hostutil.validate_auth_host(auth_host, rp_id)
except ValueError as e:
warn(f"Realm '{rp_id}': {e} — auth host ignored")
auth_host = None
origins = []
related = list(realm.related_origins or [])
for origin in realm.origins or []:
hn = hostutil.origin_hostname(origin)
if not hn:
warn(f"Realm '{rp_id}': invalid origin '{origin}' dropped")
continue
if hostutil.is_subdomain(hn, rp_id):
origins.append(origin)
else:
warn(
f"Realm '{rp_id}': origin '{origin}' is outside the rp-id "
"domain — treating it as a related origin; fix the lists "
"in the admin interface"
)
related.append(origin)
related_ok = []
for origin in related:
hn = hostutil.origin_hostname(origin)
if not hn:
warn(f"Realm '{rp_id}': invalid related origin '{origin}' dropped")
continue
if hostutil.is_subdomain(hn, rp_id):
warn(
f"Realm '{rp_id}': related origin '{origin}' is within the "
"rp-id domain — dropped (subdomains need no related entry)"
)
continue
related_ok.append(origin)
related_ok = list(dict.fromkeys(related_ok))
if len(related_ok) > related_origin_cap:
warn(
f"Realm '{rp_id}': {len(related_ok)} related origins exceed the "
f"maximum of {related_origin_cap} — extras dropped"
)
del related_ok[related_origin_cap:]
realms.append(
RealmConfig(
rp_id=rp_id,
rp_name=realm.rp_name,
auth_host=auth_host,
origins=list(dict.fromkeys(origins)) or None,
related_origins=related_ok or None,
)
)
if not realms:
raise ValueError("No servable realm in the stored configuration")
# Cross-realm collisions: keep the first configured claimant, drop the
# rest with a warning so dispatch stays deterministic.
rp_ids = {r.rp_id for r in realms}
seen_auth_hosts: dict[str, str] = {}
for realm in realms:
if not realm.auth_host:
continue
hn = hostutil.normalize_host(hostutil.auth_host_netloc(realm.auth_host))
if hn in rp_ids:
warn(
f"Realm '{realm.rp_id}': auth host '{hn}' collides with an "
"rp-id — auth host ignored"
)
realm.auth_host = None
elif hn in seen_auth_hosts:
warn(
f"Realm '{realm.rp_id}': auth host '{hn}' is also used by "
f"'{seen_auth_hosts[hn]}' — auth host ignored"
)
realm.auth_host = None
else:
seen_auth_hosts[hn] = realm.rp_id
seen_related: dict[str, str] = {}
for realm in realms:
keep = []
for origin in realm.related_origins or []:
hn = hostutil.origin_hostname(origin)
if hn in rp_ids:
warn(
f"Realm '{realm.rp_id}': related origin '{origin}' "
"collides with an rp-id — dropped"
)
elif other := next(
(
o
for o in rp_ids
if o != realm.rp_id and hostutil.is_subdomain(hn, o)
),
None,
):
warn(
f"Realm '{realm.rp_id}': related origin '{origin}' falls "
f"inside realm '{other}' — dropped"
)
elif hn in seen_auth_hosts:
warn(
f"Realm '{realm.rp_id}': related origin '{origin}' is the "
f"auth host of '{seen_auth_hosts[hn]}' — dropped"
)
elif hn in seen_related:
warn(
f"Realm '{realm.rp_id}': related origin '{origin}' is also "
f"used by '{seen_related[hn]}' — dropped (first realm wins)"
)
else:
seen_related[hn] = realm.rp_id
keep.append(origin)
realm.related_origins = keep or None
return Config(realms=realms, listen=config.listen), warnings
def _derive_site(
realm: RealmConfig, *, listen_port: int | None, vite_url: str | None
) -> tuple[str, str]:
@@ -263,8 +421,14 @@ def configure(*, listen: list[str] | None = None) -> None:
def build(config: Config) -> RealmRegistry:
"""Validate and build a registry from a combined configuration."""
validate_config(config)
"""Build a registry from a stored configuration.
The config is sanitized best-effort (serving must not fail on stored
config problems — the admin UI fixes them on a running server);
warnings are logged and exposed on the registry.
"""
config, warnings = sanitize_config(config)
validate_config(config) # sanitize guarantees this; a raise means a bug
endpoint = next(iter(parse_endpoints(_listen, DEFAULT_PORT)), {})
vite_url = os.environ.get("PASKIA_VITE_URL")
realms = [
@@ -274,7 +438,11 @@ def build(config: Config) -> RealmRegistry:
)
for rc in config.realms
]
return RealmRegistry(realms)
registry = RealmRegistry(realms)
registry.warnings = warnings
for warning in warnings:
logger.warning("Config: %s", warning)
return registry
def init_registry(config: Config) -> RealmRegistry:
tests/test_admin.py
+63
View File
@@ -2048,6 +2048,69 @@ class TestRealms:
r = await client.delete("/auth/api/admin/realms/example.com", headers=headers)
assert r.status_code == 400
@pytest.mark.asyncio
async def test_update_realm_refuses_self_lockout(
self, client: httpx.AsyncClient, session_token: str
):
"""An allow-list excluding the admin's current host is refused."""
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
# Allow-list without the current host and no auth host → lockout
r = await client.patch(
"/auth/api/admin/realms/localhost",
json={
"rp_name": "",
"auth_host": "",
"origins": ["https://auth.localhost"],
},
headers=headers,
)
assert r.status_code == 400
assert "lock you out" in r.text
# Allow-list including the current host is fine
r = await client.patch(
"/auth/api/admin/realms/localhost",
json={
"rp_name": "",
"auth_host": "",
"origins": ["https://localhost:4401"],
},
headers=headers,
)
assert r.status_code == 200, r.text
# An allow-list without the current host is also fine when an auth
# host is set: ceremonies move there (and it is always allowed).
# Done last: with an auth host set, the API here routes differently.
r = await client.patch(
"/auth/api/admin/realms/localhost",
json={
"rp_name": "",
"auth_host": "auth.localhost",
"origins": ["https://auth.localhost"],
},
headers=headers,
)
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_delete_current_realm_refused(
self, client: httpx.AsyncClient, session_token: str
):
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
r = await client.post(
"/auth/api/admin/realms/", json={"rp_id": "example.com"}, headers=headers
)
assert r.status_code == 200
# Deleting the realm in use is refused even if it has no credentials
r = await client.delete("/auth/api/admin/realms/localhost", headers=headers)
assert r.status_code == 400
assert "currently using" in r.text
# Deleting another realm while authenticated here is fine
r = await client.delete("/auth/api/admin/realms/example.com", headers=headers)
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_effective_auth_host_fallback(
self,
tests/test_realms.py
+119
View File
@@ -251,6 +251,125 @@ class TestValidateConfig:
)
# -------------------------------------------------------------------------
# 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)
# -------------------------------------------------------------------------