diff --git a/paskia/bootstrap.py b/paskia/bootstrap.py index c000318..b7962b2 100644 --- a/paskia/bootstrap.py +++ b/paskia/bootstrap.py @@ -67,7 +67,10 @@ async def check_admin_credentials() -> bool: # Check first admin user for credentials on any configured domain admin_user = admin_users[0] reg = domains.registry() - configured = sorted(d.rp_id for d in reg.domains) + # Remote domains hold their credentials on the remote instance + configured = sorted(d.rp_id for d in reg.domains if d.remote is None) + if not configured: + return False if not any(admin_user.credential_ids_for(rp_id) for rp_id in configured): # Admin exists but has no credential on any domain diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py index aa8ea55..5091144 100644 --- a/paskia/fastapi/api.py +++ b/paskia/fastapi/api.py @@ -189,13 +189,13 @@ async def check_user( data = _store(request) try: u = data.users[user_uuid] - role = u.role - org = role.org + role = data.roles[u.role_uuid] + org = data.orgs[role.org_uuid] except KeyError: raise HTTPException(status_code=404, detail="User not found") host = hostutil.normalize_host(request.headers.get("host")) - org_perm_uuids = {p.uuid for p in org.permissions} + org_perm_uuids = {p.uuid for p in data.permissions.values() if org.uuid in p.orgs} effective_perms = [] for perm_uuid in role.permission_set: @@ -236,7 +236,7 @@ def _remote_headers(ctx) -> dict[str, str]: "Remote-Session-Expires": ( (ctx.session.validated + EXPIRES).isoformat().replace("+00:00", "Z") ), - "Remote-Credential": str(ctx.session.credential), + "Remote-Credential": str(ctx.credential.uuid), } diff --git a/paskia/fastapi/dispatch.py b/paskia/fastapi/dispatch.py index 4b82300..32fc7bd 100644 --- a/paskia/fastapi/dispatch.py +++ b/paskia/fastapi/dispatch.py @@ -63,6 +63,11 @@ class DispatchMiddleware: host = _header(scope, "host") host_domain = registry.resolve(host) if host_domain is None: + # The sync endpoint is server-to-server and token-gated: the + # satellite may reach us via an address outside our domains. + if scope.get("path") == "/auth/api/sync/ws" and registry.domains: + await self._dispatch(scope, receive, send, registry.domains[0]) + return await send({"type": "websocket.close", "code": _WS_CLOSE_POLICY_VIOLATION}) return diff --git a/paskia/fastapi/proxy.py b/paskia/fastapi/proxy.py index 7e69a6b..9a12a10 100644 --- a/paskia/fastapi/proxy.py +++ b/paskia/fastapi/proxy.py @@ -50,13 +50,11 @@ async def proxy_to_remote(request: Request, remote: RemoteConfig) -> Response: 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, - ) + response = Response(content=upstream.content, status_code=upstream.status_code) + # Raw headers to preserve repeated Set-Cookie + response.raw_headers = [ + (k, v) + for k, v in upstream.headers.raw + if k.decode().lower() not in _SKIP_RESPONSE_HEADERS + ] + return response diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py index a210e16..19a6de8 100644 --- a/paskia/fastapi/sync.py +++ b/paskia/fastapi/sync.py @@ -12,7 +12,6 @@ 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__) @@ -64,7 +63,6 @@ async def _apply_client_message(message: dict) -> None: @app.websocket("/ws") -@websocket_error_handler async def sync_websocket(ws: WebSocket): tokens = syncfeed.tokens_from_env() auth = ws.headers.get("authorization", "") @@ -77,33 +75,40 @@ async def sync_websocket(ws: WebSocket): 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)) + # The client always speaks first: resume request (possibly null fields) + resume = msgspec.json.decode(await ws.receive_bytes()) + queue = feed.subscribe() try: - while True: - await _apply_client_message(msgspec.json.decode(await ws.receive_bytes())) + 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: - sender.cancel() - finally: - feed.unsubscribe(queue) + feed.unsubscribe(queue) + except WebSocketDisconnect: + pass + except Exception: + _logger.exception("Sync WebSocket failed") async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: diff --git a/paskia/satellite.py b/paskia/satellite.py index c8093ae..4309cd8 100644 --- a/paskia/satellite.py +++ b/paskia/satellite.py @@ -12,6 +12,7 @@ cache_ttl for fail-open behavior bounded by session expiry). """ import asyncio +import contextlib import logging import time from datetime import UTC, datetime @@ -73,6 +74,7 @@ class RemoteReplica: self.generation: str | None = None self.seq = 0 self.last_contact = 0.0 # monotonic time of last snapshot/event + self.connected = False self._pending_refresh: dict[str, dict] = {} self._refresh_signal = asyncio.Event() self._task: asyncio.Task | None = None @@ -80,9 +82,16 @@ class RemoteReplica: self._stopped = True def available(self) -> bool: - return ( - self.last_contact > 0 - and time.monotonic() - self.last_contact <= self.remote.cache_ttl + """Synced and either connected now or within cache_ttl of silence. + + The websockets keepalive drops a wedged connection, so a live + connection means events arrive within one round trip; after losing + it the replica remains trusted for cache_ttl. + """ + if not self.last_contact: + return False + return self.connected or ( + time.monotonic() - self.last_contact <= self.remote.cache_ttl ) def refresh_session(self, key: str, validated, ip: str, user_agent: str) -> None: @@ -114,7 +123,7 @@ class RemoteReplica: for task in (self._task, self._sweeper): if task: task.cancel() - with asyncio.suppress(asyncio.CancelledError): + with contextlib.suppress(asyncio.CancelledError): await task async def _sweep(self) -> None: @@ -132,6 +141,11 @@ class RemoteReplica: raise except Exception as e: _logger.info("Sync to %s failed: %s", self.remote.url, e) + if self.connected: + # The TTL clock starts when the feed goes down, not at the + # last message — an idle connection is healthy. + self.connected = False + self.last_contact = time.monotonic() if not self._stopped: await asyncio.sleep(_RECONNECT_DELAY) @@ -143,7 +157,11 @@ class RemoteReplica: ) 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}"} + ws_url, + additional_headers={"Authorization": f"Bearer {self.remote.token}"}, + # Prompt dead-peer detection: availability semantics count on it + ping_interval=5, + ping_timeout=5, ) as ws: hello = msgspec.json.decode(await ws.recv()) if hello.get("type") != "hello": @@ -182,9 +200,10 @@ class RemoteReplica: attach_stores() self.generation = hello["generation"] self.seq = message["seq"] + self.connected = True finally: sender.cancel() - with asyncio.suppress(asyncio.CancelledError): + with contextlib.suppress(asyncio.CancelledError): await sender async def _send_loop(self, ws) -> None: diff --git a/tests/test_remote.py b/tests/test_remote.py index 0c27fd3..3470122 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -1,25 +1,34 @@ """Tests for remote (satellite) domains: config, replica application, feed.""" -import asyncio -import os +import collections import secrets +import time from datetime import UTC, datetime from uuid import UUID +import httpx +import msgspec import pytest +import pytest_asyncio +from fastapi import Response import paskia.db.operations as ops_db from paskia import domains, satellite, syncfeed from paskia.db.structs import ( + DB, Config, Credential, DomainConfig, + Org, OriginEntry, + Permission, RemoteConfig, + Role, Session, User, ) -from paskia.db.structs import DB +from paskia.fastapi.mainapp import app +from paskia.fastapi.session import AUTH_COOKIE_NAME from paskia.util.crypto import hash_secret from .conftest import TEST_RP_ID @@ -79,8 +88,6 @@ def test_apply_upsert_and_delete(): def _builtins(obj): - import msgspec - return msgspec.to_builtins(obj) @@ -108,17 +115,15 @@ 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), + credential_id=secrets.token_bytes(32), user=UUID(int=1), aaguid=UUID(int=0), - public_key=os.urandom(64), + public_key=secrets.token_bytes(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] @@ -141,7 +146,7 @@ def test_feed_emit_and_replay(): def test_feed_ring_overflow_replay_none(): feed = syncfeed.SyncFeed() - feed.events = __import__("collections").deque(maxlen=3) + feed.events = 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 @@ -194,5 +199,180 @@ def test_availability_gate(): RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60) ) assert not replica.available() # never synced - replica.last_contact = __import__("time").monotonic() + replica.last_contact = time.monotonic() assert replica.available() + + +# ------------------------------------------------------------------------- +# API-level: endpoints served from an injected replica +# ------------------------------------------------------------------------- + + +def _replica_db() -> tuple[DB, str]: + """A replica DB holding one org/role/perm/user/credential/session.""" + replica = DB() + org = Org.create(display_name="Org") + org.uuid = UUID(int=101) + replica.orgs[org.uuid] = org + perm = Permission.create(scope="auth:admin", display_name="Admin") + perm.uuid = UUID(int=102) + perm.orgs[org.uuid] = True + replica.permissions[perm.uuid] = perm + role = Role.create(org=org.uuid, display_name="Admins", permissions={perm.uuid}) + role.uuid = UUID(int=103) + replica.roles[role.uuid] = role + user = User.create(display_name="Remote Admin", role=role.uuid) + user.uuid = UUID(int=104) + replica.users[user.uuid] = user + cred = Credential.create( + credential_id=b"cid", + user=user.uuid, + aaguid=UUID(int=0), + public_key=b"pk", + sign_count=0, + rp_id="example.com", + ) + cred.uuid = UUID(int=105) + replica.credentials[cred.uuid] = cred + secret = secrets.token_urlsafe(12) + session = Session.create( + user=user.uuid, + credential=cred.uuid, + key=hash_secret("cookie", secret), + host="app2.example.com", + ip="127.0.0.1", + user_agent="pytest", + validated=datetime.now(UTC), + rp_id="example.com", + ) + replica.sessions[session.key] = session + return replica, secret + + +@pytest_asyncio.fixture +async def remote_client(test_db): + """ASGI client with example.com as a remote domain on a warm replica.""" + config = _remote_domain_config() + domains.configure(listen=["localhost:4401"]) + domains.init_registry(config) + replica_db, secret = _replica_db() + replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t")) + replica.db = replica_db + replica.last_contact = time.monotonic() + replica.connected = True + satellite.manager.replicas[REMOTE_URL] = replica + satellite.attach_stores() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, base_url="http://localhost:4401" + ) as client: + yield client, secret, replica + satellite.manager.replicas.pop(REMOTE_URL, None) + + +@pytest.mark.asyncio +async def test_forward_served_from_replica(remote_client): + client, secret, _ = remote_client + r = await client.get( + "/auth/api/forward?perm=auth:admin", + headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}, + ) + assert r.status_code == 204 + assert r.headers["remote-name"] == "Remote Admin" + assert r.headers["remote-groups"] == "auth:admin" + + +@pytest.mark.asyncio +async def test_forward_replica_denies_missing_perm(remote_client): + client, secret, _ = remote_client + r = await client.get( + "/auth/api/forward?perm=other:scope", + headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}, + ) + assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_validate_renews_locally_and_queues_writebehind(remote_client): + client, secret, replica = remote_client + session = next(iter(replica.db.sessions.values())) + session.validated = datetime(2020, 1, 1, tzinfo=UTC) # force refresh threshold + r = await client.post( + "/auth/api/validate", + headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}, + ) + assert r.status_code == 200 + assert r.json()["renewed"] is True + assert session.validated.year > 2020 # applied to the replica + queued = replica._pending_refresh[session.key] + assert queued["type"] == "session_refresh" + + +@pytest.mark.asyncio +async def test_remote_domain_503_when_replica_stale(remote_client): + client, secret, replica = remote_client + replica.connected = False + replica.last_contact = 0 + r = await client.get( + "/auth/api/forward", + headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}, + ) + assert r.status_code == 503 + + +@pytest.mark.asyncio +async def test_logout_proxied_and_evicted(remote_client, monkeypatch): + client, secret, replica = remote_client + + async def fake_proxy(request, remote): + return Response(status_code=200, content=b'{"message": "Logged out"}') + + monkeypatch.setattr("paskia.fastapi.proxy.proxy_to_remote", fake_proxy) + r = await client.post( + "/auth/api/logout", + headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}, + ) + assert r.status_code == 200 + assert not replica.db.sessions # evicted optimistically + + +@pytest.mark.asyncio +async def test_admin_configures_remote_domain(client, session_token, test_db): + """The admin domains API stores remote config and masks the token.""" + r = await client.post( + "/auth/api/admin/domains/", + json={ + "rp_id": "example.com", + "rp_name": "Example", + "origins": {"**.example.com": True, "auth.example.com": {"auth_host": True}}, + "remote": {"url": "http://remote.test", "token": "sekret", "cache_ttl": 30}, + }, + headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"}, + ) + assert r.status_code == 200, r.text + stored = test_db.config.domains["example.com"] + assert stored.remote.url == "http://remote.test" + assert stored.remote.token == "sekret" + + r = await client.get( + "/auth/api/admin/domains/", + headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"}, + ) + entry = next(d for d in r.json() if d["rp_id"] == "example.com") + assert entry["remote"]["url"] == "http://remote.test" + assert "token" not in entry["remote"] # write-only + + +@pytest.mark.asyncio +async def test_admin_remote_domain_requires_auth_host(client, session_token): + r = await client.post( + "/auth/api/admin/domains/", + json={ + "rp_id": "example.com", + "origins": {"**.example.com": True}, + "remote": {"url": "http://remote.test"}, + }, + headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"}, + ) + assert r.status_code == 400 + assert "auth host" in r.json()["detail"]