diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index 949cc96..55ba7ae 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -67,6 +67,7 @@ from paskia.db.structs import ( DomainConfig, Org, Permission, + RemoteConfig, ResetToken, Role, Session, @@ -90,6 +91,7 @@ __all__ = [ "Org", "Permission", "DomainConfig", + "RemoteConfig", "ResetToken", "Role", "Session", diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 7647dc2..05662b2 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -13,7 +13,7 @@ from uuid import UUID import uuid7 -from paskia import oidc_notify +from paskia import oidc_notify, syncfeed from paskia.config import SESSION_LIFETIME from paskia.db.structs import ( DB, @@ -103,6 +103,7 @@ def update_permission( _db.permissions[uuid].scope = scope _db.permissions[uuid].display_name = display_name _db.permissions[uuid].domain = domain + syncfeed.emit("permissions", str(uuid), _db.permissions[uuid]) def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None: @@ -155,6 +156,7 @@ def update_org_name( raise ValueError(f"Organization {uuid} not found") with _transaction("admin:update_org_name", ctx): _db.orgs[uuid].display_name = display_name + syncfeed.emit("orgs", str(uuid), _db.orgs[uuid]) def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None: @@ -180,6 +182,7 @@ def add_permission_to_org( with _transaction("admin:add_permission_to_org", ctx): _db.permissions[permission_uuid].orgs[org_uuid] = True + syncfeed.emit("permissions", str(permission_uuid), _db.permissions[permission_uuid]) def remove_permission_from_org( @@ -197,6 +200,7 @@ def remove_permission_from_org( with _transaction("admin:remove_permission_from_org", ctx): _db.permissions[permission_uuid].orgs.pop(org_uuid, None) + syncfeed.emit("permissions", str(permission_uuid), _db.permissions[permission_uuid]) def create_role(role: Role, *, ctx: SessionContext | None = None) -> None: @@ -220,6 +224,7 @@ def update_role_name( raise ValueError(f"Role {uuid} not found") with _transaction("admin:update_role_name", ctx): _db.roles[uuid].display_name = display_name + syncfeed.emit("roles", str(uuid), _db.roles[uuid]) def add_permission_to_role( @@ -235,6 +240,7 @@ def add_permission_to_role( raise ValueError(f"Permission {permission_uuid} not found") with _transaction("admin:add_permission_to_role", ctx): _db.roles[role_uuid].permissions[permission_uuid] = True + syncfeed.emit("roles", str(role_uuid), _db.roles[role_uuid]) def remove_permission_from_role( @@ -248,6 +254,7 @@ def remove_permission_from_role( raise ValueError(f"Role {role_uuid} not found") with _transaction("admin:remove_permission_from_role", ctx): _db.roles[role_uuid].permissions.pop(permission_uuid, None) + syncfeed.emit("roles", str(role_uuid), _db.roles[role_uuid]) def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None: @@ -302,6 +309,7 @@ def update_user_display_name( slug = slugify_name(display_name) if slug and not is_username_taken(slug, exclude_uuid=uuid): user.preferred_username = slug + syncfeed.emit("users", str(uuid), user) def update_user_info( @@ -380,6 +388,7 @@ def update_user_info( user.preferred_username = preferred_username if telephone is not _UNSET: user.telephone = telephone + syncfeed.emit("users", str(uuid), user) def update_user_role( @@ -395,6 +404,7 @@ def update_user_role( raise ValueError(f"Role {role_uuid} not found") with _transaction("admin:update_user_role", ctx): _db.users[uuid].role_uuid = role_uuid + syncfeed.emit("users", str(uuid), _db.users[uuid]) def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None: @@ -429,6 +439,7 @@ def update_credential_sign_count( _db.credentials[uuid].sign_count = sign_count if last_used: _db.credentials[uuid].last_used = last_used + syncfeed.emit("credentials", str(uuid), _db.credentials[uuid]) def delete_credential( @@ -476,6 +487,7 @@ def update_session( s.validated = validated if issuer is not None: s.issuer = issuer + syncfeed.emit("sessions", key, s) def delete_session( @@ -598,6 +610,7 @@ def login( # Update credential _db.credentials[credential_uuid].sign_count = sign_count _db.credentials[credential_uuid].last_used = now + syncfeed.emit("credentials", str(credential_uuid), _db.credentials[credential_uuid]) return token @@ -625,6 +638,7 @@ def oidc_login( # Update credential _db.credentials[credential_uuid].sign_count = sign_count _db.credentials[credential_uuid].last_used = now + syncfeed.emit("credentials", str(credential_uuid), _db.credentials[credential_uuid]) def create_credential_session( diff --git a/paskia/db/structs.py b/paskia/db/structs.py index c7748c9..0c340b2 100644 --- a/paskia/db/structs.py +++ b/paskia/db/structs.py @@ -9,7 +9,7 @@ from uuid import UUID import msgspec import uuid7 -from paskia import db +from paskia import db, syncfeed from paskia.util import passphrase as passphrase_util from paskia.util.crypto import hash_secret @@ -51,6 +51,7 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True): def store(self) -> None: """Store this permission in the database. Must be called inside a transaction.""" db.data().permissions[self.uuid] = self + syncfeed.emit("permissions", str(self.uuid), self) def delete(self) -> None: """Delete this permission and remove it from all roles. @@ -59,8 +60,10 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True): """ _data = db.data() for role in _data.roles.values(): - role.permissions.pop(self.uuid, None) + if role.permissions.pop(self.uuid, None) is not None: + syncfeed.emit("roles", str(role.uuid), role) del _data.permissions[self.uuid] + syncfeed.emit("permissions", str(self.uuid), None) @classmethod def create( @@ -103,6 +106,7 @@ class Org(msgspec.Struct, dict=True): def store(self) -> None: """Store this organization in the database. Must be called inside a transaction.""" db.data().orgs[self.uuid] = self + syncfeed.emit("orgs", str(self.uuid), self) def delete(self) -> None: """Delete this org and cascade to roles, users. Remove from permissions. @@ -111,12 +115,16 @@ class Org(msgspec.Struct, dict=True): """ _data = db.data() for p in _data.permissions.values(): - p.orgs.pop(self.uuid, None) + if p.orgs.pop(self.uuid, None) is not None: + syncfeed.emit("permissions", str(p.uuid), p) for role in self.roles: for user in role.users: del _data.users[user.uuid] + syncfeed.emit("users", str(user.uuid), None) del _data.roles[role.uuid] + syncfeed.emit("roles", str(role.uuid), None) del _data.orgs[self.uuid] + syncfeed.emit("orgs", str(self.uuid), None) @classmethod def create(cls, display_name: str, created_at: datetime | None = None) -> Org: @@ -170,10 +178,12 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True): def store(self) -> None: """Store this role in the database. Must be called inside a transaction.""" db.data().roles[self.uuid] = self + syncfeed.emit("roles", str(self.uuid), self) def delete(self) -> None: """Delete this role from the database. Must be called inside a transaction.""" del db.data().roles[self.uuid] + syncfeed.emit("roles", str(self.uuid), None) @classmethod def create( @@ -254,6 +264,7 @@ class User(msgspec.Struct, dict=True, omit_defaults=True, kw_only=True): def store(self) -> None: """Store this user in the database. Must be called inside a transaction.""" db.data().users[self.uuid] = self + syncfeed.emit("users", str(self.uuid), self) def delete(self) -> None: """Delete this user and cascade to credentials, sessions, reset tokens. @@ -263,11 +274,14 @@ class User(msgspec.Struct, dict=True, omit_defaults=True, kw_only=True): _data = db.data() for cred in self.credentials: del _data.credentials[cred.uuid] + syncfeed.emit("credentials", str(cred.uuid), None) for sess in self.sessions: del _data.sessions[sess.key] + syncfeed.emit("sessions", sess.key, None) for token in self.reset_tokens: del _data.reset_tokens[token.key] del _data.users[self.uuid] + syncfeed.emit("users", str(self.uuid), None) @classmethod def create( @@ -331,6 +345,7 @@ class Credential(msgspec.Struct, dict=True): def store(self) -> None: """Store this credential in the database. Must be called inside a transaction.""" db.data().credentials[self.uuid] = self + syncfeed.emit("credentials", str(self.uuid), self) def delete(self) -> None: """Delete this credential and all its sessions. @@ -340,7 +355,9 @@ class Credential(msgspec.Struct, dict=True): _data = db.data() for sess in self.sessions: del _data.sessions[sess.key] + syncfeed.emit("sessions", sess.key, None) del _data.credentials[self.uuid] + syncfeed.emit("credentials", str(self.uuid), None) @classmethod def create( @@ -418,10 +435,13 @@ class Session(msgspec.Struct, dict=True, omit_defaults=True): _data.sessions[self.key] = self _data.users[self.user_uuid].last_seen = last_seen _data.users[self.user_uuid].visits += 1 + syncfeed.emit("sessions", self.key, self) + syncfeed.emit("users", str(self.user_uuid), _data.users[self.user_uuid]) def delete(self) -> None: """Delete this session from the database. Must be called inside a transaction.""" del db.data().sessions[self.key] + syncfeed.emit("sessions", self.key, None) @classmethod def create( @@ -622,6 +642,21 @@ class OriginEntry(msgspec.Struct, omit_defaults=True): auth_host: bool = False # This site hosts the account/admin interface +class RemoteConfig(msgspec.Struct, omit_defaults=True): + """Upstream paskia instance backing a remote (satellite-served) domain. + + The satellite keeps a RAM-only read replica of the remote's tables and + answers session-dependent reads locally; mutations are forwarded. The + token authenticates the sync channel (the remote reads accepted tokens + from its PASKIA_SYNC_TOKENS environment, never from its database). + """ + + url: str # e.g. "https://auth.example.com" + token: str = "" + cache_ttl: int = 60 # staleness bound (seconds) while the sync channel is down + refresh_interval: int = 300 # full re-sync cadence (seconds) + + class DomainConfig(msgspec.Struct, omit_defaults=True): """Configuration for one domain (one WebAuthn rp-id). @@ -641,6 +676,7 @@ class DomainConfig(msgspec.Struct, omit_defaults=True): rp_name: str | None = None origins: dict[str, bool | OriginEntry] = {} + remote: RemoteConfig | None = None class Config(msgspec.Struct, omit_defaults=True): diff --git a/paskia/domains.py b/paskia/domains.py index 3e8ef00..0d2c035 100644 --- a/paskia/domains.py +++ b/paskia/domains.py @@ -14,13 +14,15 @@ domain are in-domain, entries outside it are related. from __future__ import annotations +import asyncio import contextvars import logging import os from fastapi_vue.hostutil import parse_endpoints -from paskia.db.structs import Config, DomainConfig, OriginEntry +from paskia.db import operations +from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig from paskia.sansio import Passkey from paskia.util import hostutil from paskia.util.constants import DEFAULT_PORT @@ -107,9 +109,7 @@ class Domain: """ if self._store is not None: return self._store - from paskia import db - - return db.data() + return operations._db @store.setter def store(self, value) -> None: @@ -119,6 +119,11 @@ class Domain: def rp_name(self) -> str: return self.passkey.rp_name + @property + def remote(self) -> RemoteConfig | None: + """Upstream config when this domain is served as a satellite.""" + return self.config.remote + @property def own_auth_host(self) -> str | None: """This domain's own auth host as host[:port], if configured.""" @@ -225,6 +230,12 @@ def validate_config( for rp_id, domain in config.domains.items(): hostutil.validate_rp_id(rp_id) + if domain.remote is not None and not domain.remote.url.startswith( + ("https://", "http://") + ): + raise ValueError( + f"Domain '{rp_id}': remote URL must be an http(s) URL" + ) domain_auth_host: str | None = None related_count = 0 @@ -291,6 +302,11 @@ def validate_config( f"Domain '{rp_id}' has {related_count} related origins " f"(maximum {related_origin_cap})" ) + if domain.remote is not None and domain_auth_host is None: + raise ValueError( + f"Domain '{rp_id}' is remote — it must mark an auth host " + "(profile, admin and sign-in pages live there)" + ) rp_ids = set(config.domains) for hn, owner in auth_hosts.items(): @@ -381,6 +397,20 @@ def sanitize_config( auth_seen = True origins[key] = props + if domain.remote is not None: + if not domain.remote.url.startswith(("https://", "http://")): + warn(f"Domain '{rp_id}': invalid remote URL — remote dropped") + remote = None + else: + remote = domain.remote + if not auth_seen: + warn( + f"Domain '{rp_id}': remote domain without an auth host — " + "profile, admin and sign-in pages have nowhere to live" + ) + else: + remote = None + related = sorted(k for k in origins if is_related_key(rp_id, k)) if len(related) > related_origin_cap: warn( @@ -390,7 +420,7 @@ def sanitize_config( for key in related[related_origin_cap:]: del origins[key] - domains[rp_id] = DomainConfig(rp_name=domain.rp_name, origins=origins) + domains[rp_id] = DomainConfig(rp_name=domain.rp_name, origins=origins, remote=remote) if not domains: raise ValueError("No servable domain in the stored configuration") @@ -479,6 +509,17 @@ def _derive_site( _registry: DomainRegistry | None = None _listen: list[str] | None = None +_rebuild_listeners: list = [] + + +def add_rebuild_listener(fn) -> None: + """Register fn(registry), called after every init_registry rebuild.""" + _rebuild_listeners.append(fn) + + +def remove_rebuild_listener(fn) -> None: + if fn in _rebuild_listeners: + _rebuild_listeners.remove(fn) def configure(*, listen: list[str] | None = None) -> None: @@ -519,6 +560,12 @@ def init_registry(config: Config) -> DomainRegistry: """Build and install the global registry from a combined configuration.""" global _registry _registry = build(config) + for fn in _rebuild_listeners: + result = fn(_registry) + if asyncio.iscoroutine(result): + # init_registry runs within a running loop in every serving + # context (lifespan, tests, admin rebuild). + asyncio.get_running_loop().create_task(result) return _registry diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py index e4ab88e..aa8ea55 100644 --- a/paskia/fastapi/api.py +++ b/paskia/fastapi/api.py @@ -14,11 +14,11 @@ from fastapi import ( from fastapi.responses import JSONResponse from fastapi.security import HTTPBearer -from paskia import authcode, db +from paskia import authcode, db, satellite from paskia._version import __version__ from paskia.authsession import EXPIRES, get_reset, session_ctx from paskia.domains import current_domain -from paskia.fastapi import authz, session, user +from paskia.fastapi import authz, proxy, session, user from paskia.fastapi.response import MsgspecResponse from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME, get_client_ip from paskia.util import hostutil, htmlutil, passphrase, permutil, userinfo @@ -98,8 +98,17 @@ def _parse_perm(perm: list[str]) -> list[tuple[str, ...]]: def _store(request: Request): - """The dispatched domain's data store (local DB or read replica).""" - return request.state.domain.store + """The dispatched domain's data store (local DB or read replica). + + Remote domains fail closed once the sync channel has been silent for + longer than their cache_ttl. + """ + domain = request.state.domain + if domain.remote is not None: + replica = satellite.manager.replica_for(domain) + if replica is None or not replica.available(): + raise HTTPException(503, "Remote authentication service unavailable") + return domain.store @app.post("/validate") @@ -128,13 +137,22 @@ async def validate_token( if auth and renew: consumed = datetime.now(UTC) - ctx.session.validated if not timedelta(0) < consumed < _REFRESH_INTERVAL: - db.update_session( - ctx.session.key, - ip=get_client_ip(request), - user_agent=request.headers.get("user-agent"), - validated=datetime.now(UTC), - ctx=ctx, - ) + replica = satellite.manager.replica_for(request.state.domain) + if replica is not None: + replica.refresh_session( + ctx.session.key, + datetime.now(UTC), + get_client_ip(request), + request.headers.get("user-agent", ""), + ) + else: + db.update_session( + ctx.session.key, + ip=get_client_ip(request), + user_agent=request.headers.get("user-agent"), + validated=datetime.now(UTC), + ctx=ctx, + ) renewed = True _set_log_extra(request, ctx.session.key) resp = MsgspecResponse( @@ -359,8 +377,10 @@ async def api_user_info( @app.get("/token-info") -async def token_info(credentials=Depends(bearer_auth)): +async def token_info(request: Request, credentials=Depends(bearer_auth)): """Get reset/device-add token info. Pass token via Bearer header.""" + if request.state.domain.remote is not None: + return await proxy.proxy_to_remote(request, request.state.domain.remote) if not credentials or not credentials.credentials: raise HTTPException(401, "Bearer token required") token = credentials.credentials @@ -383,6 +403,13 @@ async def token_info(credentials=Depends(bearer_auth)): @app.post("/logout") async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE): + if request.state.domain.remote is not None: + proxied = await proxy.proxy_to_remote(request, request.state.domain.remote) + if auth and proxied.status_code == 200: + replica = satellite.manager.replica_for(request.state.domain) + if replica is not None: + replica.evict_session(auth) + return proxied if not auth: return {"message": "Already logged out"} host = request.headers.get("host") @@ -407,6 +434,11 @@ async def api_set_session( if not auth or not auth.credentials: raise HTTPException(400, "Bearer token required") + if request.state.domain.remote is not None: + # The exchange code lives in the remote's RAM; redeem it there. The + # session itself reaches the replica via the sync channel. + return await proxy.proxy_to_remote(request, request.state.domain.remote) + host = hostutil.normalize_host(request.headers.get("host", "")) if not host: raise HTTPException(400, "Host header required") diff --git a/paskia/fastapi/mainapp.py b/paskia/fastapi/mainapp.py index f2296ed..beb79b3 100644 --- a/paskia/fastapi/mainapp.py +++ b/paskia/fastapi/mainapp.py @@ -7,11 +7,11 @@ from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, RedirectResponse from fastapi_vue import env -from paskia import authcode, db, domains, remoteauth +from paskia import authcode, db, domains, remoteauth, satellite from paskia.bootstrap import bootstrap_if_needed from paskia.db.background import start_background, stop_background from paskia.db.lifecycle import kanta -from paskia.fastapi import admin, api, auth_host, oid, ws +from paskia.fastapi import admin, api, auth_host, oid, sync, ws from paskia.fastapi.admin.adminapp import adminapp from paskia.fastapi.dispatch import DispatchMiddleware @@ -50,6 +50,7 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path domains.init_registry(db.data().config) await remoteauth.init() await authcode.start() + await satellite.manager.start() except ValueError as e: logging.error(f"⚠️ {e}") # Re-raise to fail fast @@ -60,6 +61,7 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path await start_background() yield await stop_background() + await satellite.manager.stop() await authcode.stop() @@ -83,6 +85,7 @@ app.middleware("http")(auth_host.redirect_middleware) app.add_middleware(DispatchMiddleware) app.mount("/auth/api/admin/", admin.app) +app.mount("/auth/api/sync", sync.app) app.mount("/auth/api/", api.app) app.mount("/auth/ws/", ws.app) app.mount("/auth/oidc/", oid.app) diff --git a/paskia/fastapi/oid.py b/paskia/fastapi/oid.py index 62d8e49..1936865 100644 --- a/paskia/fastapi/oid.py +++ b/paskia/fastapi/oid.py @@ -22,6 +22,7 @@ from fastapi.security import HTTPBearer from paskia import authcode, db from paskia.db.structs import OIDC, Session +from paskia.fastapi import proxy from paskia.util import avatar, oidjwt from paskia.util.crypto import hash_secret @@ -30,6 +31,15 @@ _logger = logging.getLogger(__name__) app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) +@app.middleware("http") +async def proxy_remote_domain(request: Request, call_next): + """OIDC key material and sessions stay on the remote; proxy everything.""" + remote = request.state.domain.remote + if remote is not None: + return await proxy.proxy_to_remote(request, remote) + return await call_next(request) + + def _provider() -> OIDC: """Return the instance-global OIDC provider state.""" return db.data().oidc diff --git a/paskia/fastapi/proxy.py b/paskia/fastapi/proxy.py new file mode 100644 index 0000000..7e69a6b --- /dev/null +++ b/paskia/fastapi/proxy.py @@ -0,0 +1,62 @@ +"""HTTP forwarding for remote domains: mutations proxied to the remote. + +The original Host header is preserved so the remote dispatches the request +to the same domain (sessions are host-bound). User cookies authenticate the +forwarded call; no satellite credentials are involved. +""" + +import httpx +from fastapi import Request, Response + +from paskia.db.structs import RemoteConfig + +_TIMEOUT = httpx.Timeout(15.0, connect=5.0) + +_HOP_BY_HOP = { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "content-length", + "accept-encoding", +} + +_SKIP_RESPONSE_HEADERS = _HOP_BY_HOP | {"content-encoding"} + +_clients: dict[str, httpx.AsyncClient] = {} + + +def _client(base_url: str) -> httpx.AsyncClient: + client = _clients.get(base_url) + if client is None: + client = httpx.AsyncClient(base_url=base_url, timeout=_TIMEOUT) + _clients[base_url] = client + return client + + +async def proxy_to_remote(request: Request, remote: RemoteConfig) -> Response: + """Forward this request to the remote instance unchanged.""" + headers = { + k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP + } + upstream = await _client(remote.url).request( + request.method, + request.url.path, + params=request.url.query, + content=await request.body(), + headers=headers, + ) + response_headers = { + k: v + for k, v in upstream.headers.multi_items() + if k.lower() not in _SKIP_RESPONSE_HEADERS + } + return Response( + content=upstream.content, + status_code=upstream.status_code, + headers=response_headers, + ) diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py new file mode 100644 index 0000000..a210e16 --- /dev/null +++ b/paskia/fastapi/sync.py @@ -0,0 +1,114 @@ +"""Sync WebSocket endpoint: serves snapshots and live events to satellites. + +Token-gated via PASKIA_SYNC_TOKENS (env); closed when unset. All state is +RAM-only (syncfeed); the database schema is untouched. +""" + +import asyncio +import logging +from datetime import datetime + +import msgspec +from fastapi import FastAPI, WebSocket, WebSocketDisconnect + +from paskia import db, syncfeed +from paskia.fastapi.wsutil import websocket_error_handler + +_logger = logging.getLogger(__name__) + +app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) + +_SNAPSHOT_TABLES = ( + ("permissions", "permissions"), + ("orgs", "orgs"), + ("roles", "roles"), + ("users", "users"), + ("credentials", "credentials"), + ("sessions", "sessions"), +) + + +def _snapshot_messages() -> list[bytes]: + data = db.data() + messages = [] + for table, attr in _SNAPSHOT_TABLES: + items = [ + [str(key), msgspec.to_builtins(obj)] for key, obj in getattr(data, attr).items() + ] + messages.append(syncfeed.encode({"type": "snapshot", "table": table, "items": items})) + return messages + + +async def _send(ws: WebSocket, message: dict | bytes) -> None: + await ws.send_bytes(message if isinstance(message, bytes) else syncfeed.encode(message)) + + +async def _apply_client_message(message: dict) -> None: + """Satellite write-behind: session refresh (validated/ip/user-agent).""" + if message.get("type") != "session_refresh": + return + key = message.get("key") or "" + session = db.data().sessions.get(key) + if session is None: + return + try: + validated = msgspec.convert(message.get("validated"), datetime) + except msgspec.ValidationError: + return + db.update_session( + key, + ip=str(message.get("ip") or session.ip), + user_agent=str(message.get("user_agent") or session.user_agent), + validated=validated, + ) + + +@app.websocket("/ws") +@websocket_error_handler +async def sync_websocket(ws: WebSocket): + tokens = syncfeed.tokens_from_env() + auth = ws.headers.get("authorization", "") + token = auth.removeprefix("Bearer ").strip() + if not tokens or token not in tokens: + await ws.close(code=1008) + return + await ws.accept() + + feed = syncfeed.feed + await _send(ws, {"type": "hello", "generation": feed.generation, "seq": feed.seq}) + + # The client always speaks first: resume request (possibly null fields) + resume = msgspec.json.decode(await ws.receive_bytes()) + queue = feed.subscribe() + try: + replay = None + if ( + resume.get("type") == "resume" + and resume.get("generation") == feed.generation + and isinstance(resume.get("seq"), int) + ): + replay = feed.replay_since(resume["seq"]) + if replay is not None: + for event in replay: + await _send(ws, event) + else: + for chunk in _snapshot_messages(): + await _send(ws, chunk) + await _send(ws, {"type": "ready", "seq": feed.seq}) + + sender = asyncio.create_task(_pump(ws, queue)) + try: + while True: + await _apply_client_message(msgspec.json.decode(await ws.receive_bytes())) + finally: + sender.cancel() + finally: + feed.unsubscribe(queue) + + +async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: + try: + while True: + await _send(ws, await queue.get()) + except (WebSocketDisconnect, RuntimeError, asyncio.CancelledError): + pass diff --git a/paskia/satellite.py b/paskia/satellite.py new file mode 100644 index 0000000..c8093ae --- /dev/null +++ b/paskia/satellite.py @@ -0,0 +1,248 @@ +"""Satellite side of remote domains: RAM-only read replicas. + +For each domain configured with ``DomainConfig.remote`` a replica of the +remote's tables (another plain DB instance, never persisted) is attached to +the runtime Domain as its store, fed by a sync WebSocket to the remote and +refreshed by periodic full snapshots. Session refreshes from /validate are +written back over the same channel. + +While the sync channel has been silent for longer than the domain's +cache_ttl the replica is considered unavailable (fail-closed; set a large +cache_ttl for fail-open behavior bounded by session expiry). +""" + +import asyncio +import logging +import time +from datetime import UTC, datetime +from uuid import UUID + +import msgspec +import websockets + +from paskia import domains +from paskia.authsession import EXPIRES +from paskia.db.structs import ( + DB, + Credential, + Org, + Permission, + RemoteConfig, + Role, + Session, + User, +) +from paskia.util.crypto import hash_secret + +_logger = logging.getLogger(__name__) + +_TABLES = { + "permissions": (Permission, True), + "orgs": (Org, True), + "roles": (Role, True), + "users": (User, True), + "credentials": (Credential, True), + "sessions": (Session, False), +} + +_RECONNECT_DELAY = 5 +_SWEEP_INTERVAL = 60 + + +def _apply(replica: DB, table: str, key: str, op: str, fields: dict | None) -> None: + cls, uuid_key = _TABLES[table] + store = getattr(replica, table) + store_key = UUID(key) if uuid_key else key + if op == "delete": + store.pop(store_key, None) + return + obj = msgspec.convert(fields, cls) + if uuid_key: + obj.uuid = store_key + else: + obj.key = key + store[store_key] = obj + + +class RemoteReplica: + """One remote instance's replica, its sync client and write-behind queue.""" + + def __init__(self, remote: RemoteConfig): + self.remote = remote + self.db = DB() + self.generation: str | None = None + self.seq = 0 + self.last_contact = 0.0 # monotonic time of last snapshot/event + self._pending_refresh: dict[str, dict] = {} + self._refresh_signal = asyncio.Event() + self._task: asyncio.Task | None = None + self._sweeper: asyncio.Task | None = None + self._stopped = True + + def available(self) -> bool: + return ( + self.last_contact > 0 + and time.monotonic() - self.last_contact <= self.remote.cache_ttl + ) + + def refresh_session(self, key: str, validated, ip: str, user_agent: str) -> None: + """Apply a /validate refresh locally and queue it for the remote.""" + session = self.db.sessions.get(key) + if session is not None: + session.validated = validated + session.ip = ip + session.user_agent = user_agent + self._pending_refresh[key] = { + "type": "session_refresh", + "key": key, + "validated": msgspec.to_builtins(validated), + "ip": ip, + "user_agent": user_agent, + } + self._refresh_signal.set() + + def evict_session(self, secret: str) -> None: + self.db.sessions.pop(hash_secret("cookie", secret), None) + + async def start(self) -> None: + self._stopped = False + self._task = asyncio.create_task(self._run()) + self._sweeper = asyncio.create_task(self._sweep()) + + async def stop(self) -> None: + self._stopped = True + for task in (self._task, self._sweeper): + if task: + task.cancel() + with asyncio.suppress(asyncio.CancelledError): + await task + + async def _sweep(self) -> None: + while True: + await asyncio.sleep(_SWEEP_INTERVAL) + limit = datetime.now(UTC) - EXPIRES + for key in [k for k, s in self.db.sessions.items() if s.validated < limit]: + del self.db.sessions[key] + + async def _run(self) -> None: + while not self._stopped: + try: + await self._connect() + except asyncio.CancelledError: + raise + except Exception as e: + _logger.info("Sync to %s failed: %s", self.remote.url, e) + if not self._stopped: + await asyncio.sleep(_RECONNECT_DELAY) + + async def _connect(self) -> None: + ws_url = self.remote.url.replace("http", "ws", 1) + "/auth/api/sync/ws" + # Periodic full snapshots reconcile any drift; resume is cheaper. + full_resync = self.generation is None or ( + time.monotonic() - self.last_contact > self.remote.refresh_interval + ) + resume = {} if full_resync else {"generation": self.generation, "seq": self.seq} + async with websockets.connect( + ws_url, additional_headers={"Authorization": f"Bearer {self.remote.token}"} + ) as ws: + hello = msgspec.json.decode(await ws.recv()) + if hello.get("type") != "hello": + raise ValueError("sync: expected hello") + await ws.send(msgspec.json.encode({"type": "resume", **resume})) + sender = asyncio.create_task(self._send_loop(ws)) + staging: DB | None = None + try: + while True: + message = msgspec.json.decode(await ws.recv()) + self.last_contact = time.monotonic() + mtype = message.get("type") + if mtype == "snapshot": + if staging is None: + staging = DB() + for key, fields in message["items"]: + _apply(staging, message["table"], key, "upsert", fields) + elif mtype == "event": + if staging is not None or ( + self.generation is not None + and message["seq"] != self.seq + 1 + ): + raise ValueError("sync: event out of order") + self.seq = message["seq"] + _apply( + self.db, + message["table"], + message["key"], + message["op"], + message.get("fields"), + ) + elif mtype == "ready": + if staging is not None: + self.db = staging + staging = None + attach_stores() + self.generation = hello["generation"] + self.seq = message["seq"] + finally: + sender.cancel() + with asyncio.suppress(asyncio.CancelledError): + await sender + + async def _send_loop(self, ws) -> None: + while True: + self._refresh_signal.clear() + while self._pending_refresh: + _, message = self._pending_refresh.popitem() + await ws.send(msgspec.json.encode(message)) + await self._refresh_signal.wait() + + +class SatelliteManager: + """Replicas keyed by remote URL; domains sharing a remote share one.""" + + def __init__(self): + self.replicas: dict[str, RemoteReplica] = {} + + def replica_for(self, domain: domains.Domain) -> RemoteReplica | None: + if domain.remote is None: + return None + return self.replicas.get(domain.remote.url) + + async def start(self) -> None: + domains.add_rebuild_listener(self.reconcile) + await self.reconcile(domains.registry()) + + async def stop(self) -> None: + domains.remove_rebuild_listener(self.reconcile) + for replica in self.replicas.values(): + await replica.stop() + self.replicas.clear() + + async def reconcile(self, registry: domains.DomainRegistry) -> None: + """Attach stores and start/stop replicas to match the config.""" + wanted = {} + for domain in registry.domains: + if domain.remote is not None: + wanted.setdefault(domain.remote.url, domain.remote) + for url in list(self.replicas): + if url not in wanted: + await self.replicas.pop(url).stop() + for url, remote in wanted.items(): + replica = self.replicas.get(url) + if replica is None or replica.remote != remote: + if replica is not None: + await replica.stop() + replica = RemoteReplica(remote) + self.replicas[url] = replica + await replica.start() + attach_stores() + + +manager = SatelliteManager() + + +def attach_stores() -> None: + """Attach each remote domain's store to its replica.""" + for domain in domains.registry().domains: + replica = manager.replica_for(domain) + if replica is not None: + domain.store = replica.db diff --git a/paskia/syncfeed.py b/paskia/syncfeed.py new file mode 100644 index 0000000..ca8ac4a --- /dev/null +++ b/paskia/syncfeed.py @@ -0,0 +1,88 @@ +"""RAM-only change feed letting satellite instances mirror this server. + +Nothing here touches the database file: events are held in a bounded ring +buffer and pushed to connected satellites over the sync WebSocket +(fastapi/sync.py). Satellites authenticate with a token from the +PASKIA_SYNC_TOKENS environment variable (comma-separated); with the +variable unset the sync endpoint stays closed. +""" + +import asyncio +import itertools +import logging +import os +import secrets +from collections import deque + +import msgspec + +_logger = logging.getLogger(__name__) + +# Tables mirrored by satellites (reset tokens, OIDC data and domain config +# are instance-local and never replicated). +TABLES = ("permissions", "orgs", "roles", "users", "credentials", "sessions") + +_RING_SIZE = 2000 + + +class SyncFeed: + """Sequenced change events with replay for reconnecting satellites.""" + + def __init__(self): + self.generation = secrets.token_hex(8) + self._seq = itertools.count(1) + self.seq = 0 + self.events: deque[dict] = deque(maxlen=_RING_SIZE) + self.subscribers: set[asyncio.Queue] = set() + + def emit(self, table: str, key: str, obj) -> None: + """Publish an upsert (obj given) or delete (obj None).""" + self.seq = next(self._seq) + event = { + "type": "event", + "seq": self.seq, + "table": table, + "key": key, + "op": "upsert" if obj is not None else "delete", + "fields": msgspec.to_builtins(obj) if obj is not None else None, + } + self.events.append(event) + for queue in self.subscribers: + try: + queue.put_nowait(event) + except asyncio.QueueFull: + # Slow consumer: drop it; the client reconnects and resyncs. + self.subscribers.discard(queue) + + def replay_since(self, seq: int) -> list[dict] | None: + """Events after seq, or None when the ring no longer reaches back.""" + if not self.events: + return [] if seq == self.seq else None + oldest = self.events[0]["seq"] + if seq < oldest - 1: + return None + return [e for e in self.events if e["seq"] > seq] + + def subscribe(self) -> asyncio.Queue: + queue: asyncio.Queue = asyncio.Queue(maxsize=1000) + self.subscribers.add(queue) + return queue + + def unsubscribe(self, queue: asyncio.Queue) -> None: + self.subscribers.discard(queue) + + +feed = SyncFeed() + + +def emit(table: str, key: str, obj) -> None: + feed.emit(table, key, obj) + + +def tokens_from_env() -> set[str]: + """Accepted sync tokens (PASKIA_SYNC_TOKENS, comma-separated).""" + return {t.strip() for t in os.environ.get("PASKIA_SYNC_TOKENS", "").split(",") if t.strip()} + + +def encode(message: dict) -> bytes: + return msgspec.json.encode(message)