From 9394c381799efe11aef8a221721fd07f1d15de4f Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sun, 20 Sep 2026 23:26:00 +0000 Subject: [PATCH] Remote domain configuration via admin domains API ApiDomain carries the remote block (sync token write-only, never echoed); create/patch accept it, validated with the combined config (auth host mandatory for remote domains). db.update_domain replaces remote wholesale like the other domain fields. --- paskia/db/operations.py | 5 +- paskia/fastapi/admin/domains.py | 38 +++++- paskia/syncfeed.py | 5 +- paskia/util/apistructs.py | 11 +- tests/test_remote.py | 198 ++++++++++++++++++++++++++++++++ 5 files changed, 253 insertions(+), 4 deletions(-) create mode 100644 tests/test_remote.py diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 05662b2..97fabfc 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -23,6 +23,7 @@ from paskia.db.structs import ( Org, OriginEntry, Permission, + RemoteConfig, ResetToken, Role, Session, @@ -728,9 +729,10 @@ def update_domain( *, rp_name: str | None, origins: dict[str, bool | OriginEntry], + remote: RemoteConfig | None = None, ctx: SessionContext | None = None, ) -> None: - """Replace a domain's rp_name and origins table (wholesale). + """Replace a domain's rp_name, origins table and remote (wholesale). The rp-id itself is immutable: credentials are stamped with it, so changing it would orphan them — delete and recreate the domain instead. @@ -742,6 +744,7 @@ def update_domain( with _transaction("admin:update_domain", ctx): domain.rp_name = rp_name domain.origins = origins + domain.remote = remote def delete_domain(rp_id: str, *, ctx: SessionContext | None = None) -> None: diff --git a/paskia/fastapi/admin/domains.py b/paskia/fastapi/admin/domains.py index 94b5fa6..854532c 100644 --- a/paskia/fastapi/admin/domains.py +++ b/paskia/fastapi/admin/domains.py @@ -12,7 +12,7 @@ immediately. from fastapi import Body, FastAPI, Request from paskia import db, domains -from paskia.db.structs import Config, DomainConfig, OriginEntry +from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig from paskia.fastapi import authz from paskia.fastapi.admin.errors import install_error_handlers from paskia.fastapi.response import MsgspecResponse @@ -27,6 +27,14 @@ install_error_handlers(app) def _domain_to_api(domain: domains.Domain) -> ApiDomain: + remote = domain.config.remote + if remote is not None: + # The sync token is a bearer secret: never echoed back + remote = RemoteConfig( + url=remote.url, + cache_ttl=remote.cache_ttl, + refresh_interval=remote.refresh_interval, + ) return ApiDomain( rp_id=domain.rp_id, rp_name=domain.rp_name, @@ -34,6 +42,26 @@ def _domain_to_api(domain: domains.Domain) -> ApiDomain: site_url=domain.site_url, auth_site_url=domain.auth_site_url, auth_host=domain.own_auth_host, + remote=remote, + ) + + +def _normalize_remote(value, existing: RemoteConfig | None = None) -> RemoteConfig | None: + """Parse a remote object from the admin UI (raises on malformed). + + An absent/empty token keeps the previously stored one — the token is + write-only over the API. + """ + if value is None: + return None + if not isinstance(value, dict) or not isinstance(value.get("url"), str): + raise ValueError("remote must be an object with a url") + token = str(value.get("token") or "") or (existing.token if existing else "") + return RemoteConfig( + url=value["url"].rstrip("/"), + token=token, + cache_ttl=int(value.get("cache_ttl") or 60), + refresh_interval=int(value.get("refresh_interval") or 300), ) @@ -123,6 +151,7 @@ async def admin_create_domain( new = DomainConfig( rp_name=(payload.get("rp_name") or "").strip() or None, origins=_normalize_origins_map(payload.get("origins")), + remote=_normalize_remote(payload.get("remote")), ) config = db.data().config @@ -156,9 +185,15 @@ async def admin_update_domain( if rp_id not in config.domains: raise ValueError(f"Domain {rp_id} not found") + current_remote = config.domains[rp_id].remote updated = DomainConfig( rp_name=(payload.get("rp_name") or "").strip() or None, origins=_normalize_origins_map(payload.get("origins")), + remote=( + _normalize_remote(payload["remote"], existing=current_remote) + if "remote" in payload + else current_remote + ), ) would_be = Config( domains={k: updated if k == rp_id else v for k, v in config.domains.items()}, @@ -171,6 +206,7 @@ async def admin_update_domain( rp_id, rp_name=updated.rp_name, origins=updated.origins, + remote=updated.remote, ctx=ctx, ) _rebuild_registry() diff --git a/paskia/syncfeed.py b/paskia/syncfeed.py index ca8ac4a..690cd23 100644 --- a/paskia/syncfeed.py +++ b/paskia/syncfeed.py @@ -55,7 +55,10 @@ class SyncFeed: 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.""" + """Events after seq, or None when the ring no longer reaches back + (or the claimed seq is ahead of us, which cannot be reconciled).""" + if seq > self.seq: + return None if not self.events: return [] if seq == self.seq else None oldest = self.events[0]["seq"] diff --git a/paskia/util/apistructs.py b/paskia/util/apistructs.py index 8455929..376cb35 100644 --- a/paskia/util/apistructs.py +++ b/paskia/util/apistructs.py @@ -14,7 +14,15 @@ import msgspec from uarite import uaparse from paskia import db -from paskia.db.structs import Credential, Org, OriginEntry, Permission, Role, User +from paskia.db.structs import ( + Credential, + Org, + OriginEntry, + Permission, + RemoteConfig, + Role, + User, +) # ------------------------------------------------------------------------- # API structs - inherit from db structs, add uuid for serialization @@ -194,6 +202,7 @@ class ApiDomain(msgspec.Struct): site_url: str auth_site_url: str auth_host: str | None + remote: RemoteConfig | None = None class ApiTokenInfo(msgspec.Struct, omit_defaults=True): diff --git a/tests/test_remote.py b/tests/test_remote.py new file mode 100644 index 0000000..0c27fd3 --- /dev/null +++ b/tests/test_remote.py @@ -0,0 +1,198 @@ +"""Tests for remote (satellite) domains: config, replica application, feed.""" + +import asyncio +import os +import secrets +from datetime import UTC, datetime +from uuid import UUID + +import pytest + +import paskia.db.operations as ops_db +from paskia import domains, satellite, syncfeed +from paskia.db.structs import ( + Config, + Credential, + DomainConfig, + OriginEntry, + RemoteConfig, + Session, + User, +) +from paskia.db.structs import DB +from paskia.util.crypto import hash_secret + +from .conftest import TEST_RP_ID + +REMOTE_URL = "http://remote.test" + + +def _remote_domain_config(**kw) -> Config: + return Config( + domains={ + TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}), + "example.com": DomainConfig( + origins={ + "**.example.com": True, + "auth.example.com": OriginEntry(auth_host=True), + }, + remote=RemoteConfig(url=REMOTE_URL, token="t", **kw), + ), + } + ) + + +def test_remote_domain_valid(): + domains.validate_config(_remote_domain_config()) + + +def test_remote_domain_requires_auth_host(): + config = _remote_domain_config() + config.domains["example.com"].origins = {"**.example.com": True} + with pytest.raises(ValueError, match="auth host"): + domains.validate_config(config) + + +def test_remote_domain_requires_http_url(): + config = _remote_domain_config() + config.domains["example.com"].remote.url = "ftp://x" + with pytest.raises(ValueError, match="http"): + domains.validate_config(config) + + +def test_sanitize_preserves_remote(): + config, warnings = domains.sanitize_config(_remote_domain_config()) + assert not warnings + assert config.domains["example.com"].remote.url == REMOTE_URL + + +def test_apply_upsert_and_delete(): + replica = DB() + user = User.create(display_name="U", role=UUID(int=1)) + user.uuid = UUID(int=2) + satellite._apply( + replica, "users", str(user.uuid), "upsert", _builtins(user) + ) + assert replica.users[user.uuid].display_name == "U" + satellite._apply(replica, "users", str(user.uuid), "delete", None) + assert not replica.users + + +def _builtins(obj): + import msgspec + + return msgspec.to_builtins(obj) + + +def test_apply_session_roundtrip(): + """Sessions keep their string key and datetime/UUID fields.""" + replica = DB() + session = Session.create( + user=UUID(int=1), + credential=UUID(int=2), + key=hash_secret("cookie", "sekret"), + host="app2.example.com", + ip="127.0.0.1", + user_agent="ua", + validated=datetime.now(UTC), + rp_id="example.com", + ) + satellite._apply(replica, "sessions", session.key, "upsert", _builtins(session)) + stored = replica.sessions[session.key] + assert stored.host == "app2.example.com" + assert stored.validated == session.validated + assert stored.user_uuid == UUID(int=1) + + +def test_apply_credential_bytes_roundtrip(): + """credential_id/public_key are bytes over the wire (base64 in JSON).""" + replica = DB() + cred = Credential.create( + credential_id=os.urandom(32), + user=UUID(int=1), + aaguid=UUID(int=0), + public_key=os.urandom(64), + sign_count=3, + rp_id="example.com", + ) + cred.uuid = UUID(int=9) + # Simulate the full wire path: builtins -> JSON -> builtins + import msgspec + + wire = msgspec.json.decode(msgspec.json.encode(_builtins(cred))) + satellite._apply(replica, "credentials", str(cred.uuid), "upsert", wire) + stored = replica.credentials[cred.uuid] + assert stored.credential_id == cred.credential_id + assert stored.public_key == cred.public_key + assert stored.sign_count == 3 + + +def test_feed_emit_and_replay(): + feed = syncfeed.SyncFeed() + user = User.create(display_name="A", role=UUID(int=1)) + feed.emit("users", "k1", user) + feed.emit("users", "k1", None) + assert feed.seq == 2 + assert feed.replay_since(0)[0]["op"] == "upsert" + assert feed.replay_since(1)[0]["op"] == "delete" + assert feed.replay_since(2) == [] + assert feed.replay_since(99) is None + + +def test_feed_ring_overflow_replay_none(): + feed = syncfeed.SyncFeed() + feed.events = __import__("collections").deque(maxlen=3) + for i in range(5): + feed.emit("users", f"k{i}", None) + assert feed.replay_since(0) is None # fell off the ring + assert [e["seq"] for e in feed.replay_since(4)] == [5] + assert feed.replay_since(5) == [] + + +@pytest.mark.asyncio +async def test_operations_emit_events(test_db): + """Writes through db.operations land on the sync feed.""" + syncfeed.feed.events.clear() + syncfeed.feed.seq = 0 + user = next(iter(test_db.users.values())) + ops_db.update_user_display_name(user.uuid, "Renamed") + tables = {e["table"] for e in syncfeed.feed.events} + assert "users" in tables + key = syncfeed.feed.events[-1]["key"] + assert syncfeed.feed.events[-1]["fields"]["display_name"] == "Renamed" + assert key == str(user.uuid) + + +@pytest.mark.asyncio +async def test_replica_refresh_and_evict(): + replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t")) + token = secrets.token_urlsafe(12) + session = Session.create( + user=UUID(int=1), + credential=UUID(int=2), + key=hash_secret("cookie", token), + host="app2.example.com", + ip="1.1.1.1", + user_agent="ua", + validated=datetime(2020, 1, 1, tzinfo=UTC), + ) + replica.db.sessions[session.key] = session + + now = datetime.now(UTC) + replica.refresh_session(session.key, now, "2.2.2.2", "new-ua") + assert replica.db.sessions[session.key].validated == now + queued = replica._pending_refresh[session.key] + assert queued["type"] == "session_refresh" + assert queued["ip"] == "2.2.2.2" + + replica.evict_session(token) + assert not replica.db.sessions + + +def test_availability_gate(): + replica = satellite.RemoteReplica( + RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60) + ) + assert not replica.available() # never synced + replica.last_contact = __import__("time").monotonic() + assert replica.available()