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.
481 lines
17 KiB
Python
481 lines
17 KiB
Python
"""Realm registry: per-rp-id runtime state and host resolution.
|
|
|
|
A **realm** is one rp-id with its associated hosts and origins. The
|
|
registry is built from the stored combined ``Config`` at startup and
|
|
rebuilt on admin realm changes; request dispatch resolves hosts to realms
|
|
through it. The database itself is global — only the *current realm*
|
|
(passkey, site URLs, OIDC view) varies per request, tracked via a
|
|
contextvar set by the dispatch middleware.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextvars
|
|
import logging
|
|
import os
|
|
|
|
from fastapi_vue.hostutil import parse_endpoints
|
|
|
|
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
|
|
|
|
|
|
class Realm:
|
|
"""Runtime view of one realm: stored config plus derived values."""
|
|
|
|
def __init__(self, config: RealmConfig, site_url: str, site_path: str):
|
|
# Lazy import: paskia.sansio depends on paskia.db, which (via
|
|
# paskia.db.operations → paskia.oidc_notify) depends on this module.
|
|
from paskia.sansio import Passkey # noqa: PLC0415
|
|
|
|
self.config = config
|
|
self.site_url = site_url
|
|
self.site_path = site_path
|
|
self.passkey = Passkey(
|
|
rp_id=config.rp_id,
|
|
rp_name=config.rp_name,
|
|
origins=config.origins,
|
|
related_origins=config.related_origins,
|
|
)
|
|
|
|
@property
|
|
def rp_id(self) -> str:
|
|
return self.config.rp_id
|
|
|
|
@property
|
|
def rp_name(self) -> str:
|
|
return self.passkey.rp_name
|
|
|
|
@property
|
|
def own_auth_host(self) -> str | None:
|
|
"""This realm's own auth host as host[:port], if configured."""
|
|
if not self.config.auth_host:
|
|
return None
|
|
return hostutil.auth_host_netloc(self.config.auth_host)
|
|
|
|
@property
|
|
def related_origins(self) -> list[str]:
|
|
"""Configured related (cross-domain) origins for ROR."""
|
|
return list(self.config.related_origins or [])
|
|
|
|
@property
|
|
def is_root_mode(self) -> bool:
|
|
"""Whether this realm's UI lives at the site root (own auth host)."""
|
|
return self.config.auth_host is not None
|
|
|
|
@property
|
|
def ui_base_path(self) -> str:
|
|
return "/" if self.is_root_mode else "/auth/"
|
|
|
|
@property
|
|
def auth_site_url(self) -> str:
|
|
"""Base URL of this realm's auth site UI."""
|
|
return self.site_url + self.site_path
|
|
|
|
def api_url(self, path: str = "") -> str:
|
|
"""Return an absolute URL under the canonical /auth/api/ prefix."""
|
|
if not path:
|
|
return f"{self.site_url}/auth/api/"
|
|
return f"{self.site_url}/auth/api/{path.lstrip('/')}"
|
|
|
|
def reset_link_url(self, token: str) -> str:
|
|
"""Generate a reset link URL for the given token on this realm."""
|
|
return f"{self.auth_site_url}{token}"
|
|
|
|
|
|
class RealmRegistry:
|
|
"""Resolved realms and host lookup tables."""
|
|
|
|
def __init__(self, realms: list[Realm]):
|
|
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
|
|
for origin in realm.related_origins:
|
|
if hostname := hostutil.origin_hostname(origin):
|
|
self._related_hosts[hostname] = realm
|
|
|
|
@property
|
|
def realms(self) -> list[Realm]:
|
|
"""All realms, in configuration order (first is the default)."""
|
|
return list(self._by_rp_id.values())
|
|
|
|
@property
|
|
def default(self) -> Realm:
|
|
"""The default realm (first in configuration order)."""
|
|
return next(iter(self._by_rp_id.values()))
|
|
|
|
def get(self, rp_id: str) -> Realm | None:
|
|
return self._by_rp_id.get(rp_id)
|
|
|
|
def effective_auth_host(self, realm: Realm) -> str | None:
|
|
"""Auth host serving WS/restricted APIs for a realm: its own, or the
|
|
first configured auth host (in realm order) as a shared fallback.
|
|
|
|
Returns host[:port] suitable for URL building, or None.
|
|
"""
|
|
if realm.own_auth_host:
|
|
return realm.own_auth_host
|
|
for candidate in self._by_rp_id.values():
|
|
if candidate.own_auth_host:
|
|
return candidate.own_auth_host
|
|
return None
|
|
|
|
def resolve(self, host: str | None) -> Realm | None:
|
|
"""Resolve a request Host header to a realm.
|
|
|
|
Order: exact rp-id → exact auth host → exact related-origin
|
|
hostname → longest-suffix rp-id. Unknown hosts return None.
|
|
"""
|
|
h = hostutil.normalize_host(host)
|
|
if not h:
|
|
return None
|
|
if realm := self._by_rp_id.get(h):
|
|
return realm
|
|
if realm := self._auth_hosts.get(h):
|
|
return realm
|
|
if realm := self._related_hosts.get(h):
|
|
return realm
|
|
best = None
|
|
for rp_id, realm in self._by_rp_id.items():
|
|
if h.endswith(f".{rp_id}") and (
|
|
best is None or len(rp_id) > len(best.rp_id)
|
|
):
|
|
best = realm
|
|
return best
|
|
|
|
|
|
def validate_config(
|
|
config: Config, *, related_origin_cap: int = DEFAULT_RELATED_ORIGIN_CAP
|
|
) -> None:
|
|
"""Validate a combined configuration cross-realm. Raises ValueError."""
|
|
if not config.realms:
|
|
raise ValueError("At least one realm (rp-id) is required")
|
|
|
|
rp_ids: set[str] = set()
|
|
auth_hosts: dict[str, str] = {} # normalized host -> owning rp_id
|
|
related_hosts: dict[str, str] = {} # hostname -> owning rp_id
|
|
|
|
for realm in config.realms:
|
|
hostutil.validate_rp_id(realm.rp_id)
|
|
if realm.rp_id in rp_ids:
|
|
raise ValueError(f"Duplicate rp-id '{realm.rp_id}'")
|
|
rp_ids.add(realm.rp_id)
|
|
|
|
if realm.auth_host:
|
|
hostutil.validate_auth_host(realm.auth_host, realm.rp_id)
|
|
hn = hostutil.normalize_host(
|
|
hostutil.auth_host_netloc(realm.auth_host) or ""
|
|
)
|
|
if hn:
|
|
if hn in auth_hosts:
|
|
raise ValueError(
|
|
f"auth-host '{hn}' is configured for both "
|
|
f"'{auth_hosts[hn]}' and '{realm.rp_id}'"
|
|
)
|
|
auth_hosts[hn] = realm.rp_id
|
|
|
|
for origin in realm.origins or []:
|
|
hn = hostutil.origin_hostname(origin)
|
|
if not hn:
|
|
raise ValueError(f"Invalid origin URL: '{origin}'")
|
|
if not hostutil.is_subdomain(hn, realm.rp_id):
|
|
raise ValueError(
|
|
f"Origin '{origin}' is outside the rp-id domain "
|
|
f"'{realm.rp_id}' — configure it as a related origin instead"
|
|
)
|
|
|
|
if len(realm.related_origins or []) > related_origin_cap:
|
|
raise ValueError(
|
|
f"Realm '{realm.rp_id}' has {len(realm.related_origins or [])} "
|
|
f"related origins (maximum {related_origin_cap})"
|
|
)
|
|
for origin in realm.related_origins or []:
|
|
hn = hostutil.origin_hostname(origin)
|
|
if not hn:
|
|
raise ValueError(f"Invalid related origin URL: '{origin}'")
|
|
if hostutil.is_subdomain(hn, realm.rp_id):
|
|
raise ValueError(
|
|
f"Related origin '{origin}' is within the rp-id domain "
|
|
f"'{realm.rp_id}' — subdomains need no related origin entry"
|
|
)
|
|
if hn in related_hosts:
|
|
raise ValueError(
|
|
f"Related origin host '{hn}' is configured for both "
|
|
f"'{related_hosts[hn]}' and '{realm.rp_id}'"
|
|
)
|
|
related_hosts[hn] = realm.rp_id
|
|
|
|
for hn, owner in auth_hosts.items():
|
|
if hn in rp_ids:
|
|
raise ValueError(f"auth-host '{hn}' collides with an rp-id")
|
|
if hn in related_hosts:
|
|
raise ValueError(
|
|
f"auth-host '{hn}' collides with a related origin of "
|
|
f"realm '{related_hosts[hn]}'"
|
|
)
|
|
|
|
for hn, owner in related_hosts.items():
|
|
if hn in rp_ids:
|
|
raise ValueError(f"Related origin host '{hn}' collides with an rp-id")
|
|
for other in rp_ids:
|
|
if other != owner and hostutil.is_subdomain(hn, other):
|
|
raise ValueError(
|
|
f"Related origin host '{hn}' of realm '{owner}' "
|
|
f"falls inside realm '{other}'"
|
|
)
|
|
|
|
|
|
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]:
|
|
"""Compute a realm's site_url and site_path.
|
|
|
|
Priority: auth_host > origins[0] > PASKIA_VITE_URL (localhost realm
|
|
only) > http://localhost:port (localhost realm) > https://rp-id.
|
|
"""
|
|
if realm.auth_host:
|
|
return realm.auth_host, "/"
|
|
if realm.origins:
|
|
return realm.origins[0], "/auth/"
|
|
if realm.rp_id == "localhost":
|
|
if vite_url:
|
|
return vite_url.rstrip("/"), "/auth/"
|
|
if listen_port:
|
|
return f"http://localhost:{listen_port}", "/auth/"
|
|
return f"https://{realm.rp_id}", "/auth/"
|
|
|
|
|
|
_registry: RealmRegistry | None = None
|
|
_listen: list[str] | None = None
|
|
|
|
|
|
def configure(*, listen: list[str] | None = None) -> None:
|
|
"""Record process-global serve parameters for site URL derivation."""
|
|
global _listen
|
|
_listen = listen
|
|
|
|
|
|
def build(config: Config) -> RealmRegistry:
|
|
"""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 = [
|
|
Realm(
|
|
rc,
|
|
*_derive_site(rc, listen_port=endpoint.get("port"), vite_url=vite_url),
|
|
)
|
|
for rc in config.realms
|
|
]
|
|
registry = RealmRegistry(realms)
|
|
registry.warnings = warnings
|
|
for warning in warnings:
|
|
logger.warning("Config: %s", warning)
|
|
return registry
|
|
|
|
|
|
def init_registry(config: Config) -> RealmRegistry:
|
|
"""Build and install the global registry from a combined configuration."""
|
|
global _registry
|
|
_registry = build(config)
|
|
return _registry
|
|
|
|
|
|
def registry() -> RealmRegistry:
|
|
"""Return the global registry (must be initialized)."""
|
|
if _registry is None:
|
|
raise RuntimeError("Realm registry is not initialized")
|
|
return _registry
|
|
|
|
|
|
_current_realm: contextvars.ContextVar[Realm | None] = contextvars.ContextVar(
|
|
"paskia_current_realm", default=None
|
|
)
|
|
|
|
|
|
def set_current_realm(realm: Realm | None) -> contextvars.Token:
|
|
return _current_realm.set(realm)
|
|
|
|
|
|
def reset_current_realm(token: contextvars.Token) -> None:
|
|
_current_realm.reset(token)
|
|
|
|
|
|
def current_realm() -> Realm:
|
|
"""Return the request's realm, or the default realm without request context."""
|
|
realm = _current_realm.get()
|
|
if realm is not None:
|
|
return realm
|
|
return registry().default
|