From 28fa0ff61f12f0e3120568991721cef634d7fde4 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 1 Oct 2026 06:30:32 +0000 Subject: [PATCH] Tests for the link model + lint/format pass test_remote.py: link URL parsing, RAM-only offer mint/spend/expiry, pushed-domain application (accept/idempotence/removal/local conflict/ auth-host collision), admin link lifecycle (offer, scope, revoke) and uplink list/unlink, remote-write rejection on the domains API, live feed scope mutation. test_satellite_e2e.py: real registration handshake over TCP (key material identical on both sides, duplicate and bad-URL rejections), established link reconnect, revocation unlinking the satellite, live scope shrink stopping the feed, domain conflicts reported upstream. Also: ruff format drift fixes in startupbox/test_cli. --- paskia/__main__.py | 11 +- paskia/db/operations.py | 1 - paskia/fastapi/sync.py | 8 +- paskia/satellite.py | 17 +- paskia/util/startupbox.py | 6 +- tests/test_cli.py | 4 +- tests/test_remote.py | 531 ++++++++++++++++++++++-------------- tests/test_satellite_e2e.py | 263 +++++++++++++----- 8 files changed, 547 insertions(+), 294 deletions(-) diff --git a/paskia/__main__.py b/paskia/__main__.py index f84b829..375904f 100644 --- a/paskia/__main__.py +++ b/paskia/__main__.py @@ -201,8 +201,8 @@ def cmd_satellite(args: argparse.Namespace) -> None: and the pushed domains are stored. Afterwards 'paskia' serves the linked domains. """ - from paskia import satellite - from paskia.db.lifecycle import kanta + from paskia import satellite # noqa: PLC0415 (heavy deps: fastapi, websockets) + from paskia.db.lifecycle import kanta # noqa: PLC0415 (binds the db path) db_path = db_file_path() fresh = not db_path.exists() @@ -231,7 +231,10 @@ def cmd_satellite(args: argparse.Namespace) -> None: except (ValueError, RuntimeError) as e: raise SystemExit(f"Registration failed: {e}") from e - domains_list = ", ".join(info["domains"]) or "(none yet — the upstream can scope the link later)" + domains_list = ( + ", ".join(info["domains"]) + or "(none yet — the upstream can scope the link later)" + ) print(f"🛰 Linked to {info['url']} — serving: {domains_list}") for rp_id, reason in (info.get("rejected") or {}).items(): print(f"⚠️ Domain {rp_id} rejected: {reason}") @@ -310,7 +313,7 @@ def main(): "--save", action="store_true", help="Save --listen to the database for future runs. " - "Use --listen \"\" to clear the stored endpoints.", + 'Use --listen "" to clear the stored endpoints.', ) init_parser = argparse.ArgumentParser( diff --git a/paskia/db/operations.py b/paskia/db/operations.py index ae3c3a0..3d88ffd 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -33,7 +33,6 @@ from paskia.db.structs import ( ) from paskia.util.crypto import hash_secret from paskia.util.nameutil import slugify_name -from paskia.util.passphrase import generate as generate_passphrase _logger = logging.getLogger(__name__) diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py index 8fb1564..4e28401 100644 --- a/paskia/fastapi/sync.py +++ b/paskia/fastapi/sync.py @@ -162,9 +162,7 @@ async def drop_domain_from_links(rp_id: str) -> None: """A domain was deleted upstream: shrink every link's scope and push.""" for link_id, link in db.data().satellite_links.items(): if rp_id in link.rp_ids: - db.set_satellite_link_scope( - link_id, [r for r in link.rp_ids if r != rp_id] - ) + db.set_satellite_link_scope(link_id, [r for r in link.rp_ids if r != rp_id]) await push_link_domains(link_id) @@ -300,7 +298,7 @@ async def sync_websocket(ws: WebSocket): ) finally: sender.cancel() - except (WebSocketDisconnect, RuntimeError): + except WebSocketDisconnect, RuntimeError: pass except Exception: _logger.exception("Sync WebSocket failed") @@ -314,7 +312,7 @@ async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: try: while True: await _send(ws, await queue.get()) - except (WebSocketDisconnect, RuntimeError, asyncio.CancelledError): + except WebSocketDisconnect, RuntimeError, asyncio.CancelledError: pass except Exception: _logger.debug("Sync event pump ended", exc_info=True) diff --git a/paskia/satellite.py b/paskia/satellite.py index 0d533a8..a854ac9 100644 --- a/paskia/satellite.py +++ b/paskia/satellite.py @@ -43,7 +43,7 @@ import msgspec import websockets from fastapi import HTTPException, Request, Response -from paskia import db, domains +from paskia import db, domains, syncfeed from paskia.config import SESSION_LIFETIME from paskia.db.structs import ( DB, @@ -60,7 +60,6 @@ from paskia.db.structs import ( Uplink, User, ) -from paskia import syncfeed from paskia.util.apistructs import ApiSessionContext from paskia.util.crypto import hash_secret @@ -75,7 +74,7 @@ _RECONNECT_DELAY = 5 _SWEEP_INTERVAL = 60 -class LinkRevoked(Exception): +class LinkRevokedError(Exception): """The upstream closed the handshake with a policy violation: the link is revoked (or the key desynced) — the satellite unlinks its side.""" @@ -370,7 +369,7 @@ class UplinkReplica: for task in (self._task, self._sweeper): if task: task.cancel() - with contextlib.suppress(asyncio.CancelledError, LinkRevoked): + with contextlib.suppress(asyncio.CancelledError, LinkRevokedError): await task async def _sweep(self) -> None: @@ -387,7 +386,7 @@ class UplinkReplica: await self._connect() except asyncio.CancelledError: raise - except LinkRevoked: + except LinkRevokedError: raise # the manager unlinks via the done-callback except Exception as e: _logger.info("Sync to %s failed: %s", self.uplink.url, e) @@ -404,7 +403,7 @@ class UplinkReplica: return msgspec.json.decode(await ws.recv()) except websockets.exceptions.ConnectionClosedError as e: if e.code == 1008: - raise LinkRevoked(self.uplink.url) from e + raise LinkRevokedError(self.uplink.url) from e raise async def _connect(self) -> None: @@ -476,7 +475,9 @@ class UplinkReplica: replica_db=self.db, ) await ws.send( - syncfeed.encode({"type": "domains_ack", "rejected": rejected}) + syncfeed.encode( + {"type": "domains_ack", "rejected": rejected} + ) ) elif mtype == "ready": if staging is not None: @@ -504,7 +505,7 @@ def _replica_done(replica: UplinkReplica, task: asyncio.Task) -> None: if task.cancelled(): return exc = task.exception() - if not isinstance(exc, LinkRevoked): + if not isinstance(exc, LinkRevokedError): return _logger.info("Uplink %s revoked the link — unlinking", replica.uplink.url) try: diff --git a/paskia/util/startupbox.py b/paskia/util/startupbox.py index 6e4a2c7..a9675a7 100644 --- a/paskia/util/startupbox.py +++ b/paskia/util/startupbox.py @@ -70,9 +70,9 @@ def _covered_by_wildcard(key: str, pattern: str) -> bool: matched = hostutil.is_subdomain(hostname, base) else: # '*.base' covers exactly one subdomain level - matched = hostname.endswith(f".{base}") and "." not in hostname[ - : -len(base) - 1 - ] + matched = ( + hostname.endswith(f".{base}") and "." not in hostname[: -len(base) - 1] + ) if not matched: return False if hostutil.is_subdomain(base, "localhost"): diff --git a/tests/test_cli.py b/tests/test_cli.py index 3b55000..a78efee 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -324,7 +324,9 @@ def test_migrate_merges_kantadb_into_existing_database(run_cli, tmp_path): def test_migrate_merge_label_combines_rp_ids(run_cli, tmp_path): """A multi-domain source merges in one transaction, rp-ids slash-joined.""" run_cli("init", "company.com") - src = write_kantadb(tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb") + src = write_kantadb( + tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb" + ) run_cli("migrate", str(src)) diff --git a/tests/test_remote.py b/tests/test_remote.py index bcd8b7e..18c21c4 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -1,4 +1,4 @@ -"""Tests for remote (satellite) domains: config, replica application, feed.""" +"""Tests for remote (satellite) domains: config, links, replica, feed.""" import asyncio import secrets @@ -21,8 +21,10 @@ from paskia.db.structs import ( OriginEntry, RemoteConfig, Session, + Uplink, User, ) +from paskia.fastapi import sync from paskia.fastapi.mainapp import app from paskia.fastapi.session import AUTH_COOKIE_NAME from paskia.util.apistructs import ( @@ -36,10 +38,11 @@ from paskia.util.crypto import hash_secret from .conftest import TEST_RP_ID, create_test_session REMOTE_URL = "http://remote.test" +UPLINK_ID = "up1" def _api_ctx(display_name="Remote Admin", permissions=("auth:admin",)): - """A master-pushed per-session context, as embedded in session events.""" + """An upstream-pushed per-session context, as embedded in session events.""" return ApiSessionContext( user=ApiUserContext(uuid=UUID(int=104), display_name=display_name), org=ApiOrgContext(uuid=UUID(int=101), display_name="Org"), @@ -48,6 +51,12 @@ def _api_ctx(display_name="Remote Admin", permissions=("auth:admin",)): ) +def _uplink(**kw) -> Uplink: + uplink = Uplink(url=REMOTE_URL, created=datetime.now(UTC), key=b"k", **kw) + uplink.uplink_id = UPLINK_ID + return uplink + + def _remote_domain_config(**kw) -> Config: return Config( domains={ @@ -57,7 +66,7 @@ def _remote_domain_config(**kw) -> Config: "**.example.com": True, "auth.example.com": OriginEntry(auth_host=True), }, - remote=RemoteConfig(url=REMOTE_URL, token="t", **kw), + remote=RemoteConfig(link=UPLINK_ID, **kw), ), } ) @@ -74,17 +83,21 @@ def test_remote_domain_requires_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 + assert config.domains["example.com"].remote.link == UPLINK_ID + + +def test_sanitize_drops_remote_without_link(): + """A remote config pointing nowhere (e.g. a v1 leftover) degrades to a + plain local domain instead of failing to serve.""" + config = _remote_domain_config() + config.domains["example.com"].remote = RemoteConfig() + sanitized, warnings = domains.sanitize_config(config) + assert any("remote dropped" in w for w in warnings) + assert sanitized.domains["example.com"].remote is None + domains.validate_config(sanitized) def test_apply_upsert_and_delete(): @@ -209,9 +222,7 @@ async def test_operations_repush_session_contexts(test_db, test_user, test_crede @pytest.mark.asyncio async def test_replica_refresh_and_evict(): - replica = satellite.RemoteReplica( - RemoteConfig(url=REMOTE_URL, token="t"), "example.com" - ) + replica = satellite.UplinkReplica(_uplink()) token = secrets.token_urlsafe(12) session = Session.create( user=UUID(int=1), @@ -231,24 +242,120 @@ async def test_replica_refresh_and_evict(): assert queued["type"] == "session_refresh" assert queued["ip"] == "2.2.2.2" - # Host-keyed dispatch eviction (the replica's domain is resolved by host) + # Host-keyed dispatch eviction (the replica's uplink is resolved by host) domains.configure(listen=["localhost:4401"]) domains.init_registry(_remote_domain_config()) - satellite.manager.replicas["example.com"] = replica + satellite.manager.replicas[UPLINK_ID] = replica satellite.evict_session(token, "app2.example.com") assert not replica.db.sessions - satellite.manager.replicas.pop("example.com") + satellite.manager.replicas.pop(UPLINK_ID) def test_availability_gate(): - replica = satellite.RemoteReplica( - RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60), "example.com" - ) + replica = satellite.UplinkReplica(_uplink(cache_ttl=60)) assert not replica.available() # never synced replica.last_contact = time.monotonic() assert replica.available() +# ------------------------------------------------------------------------- +# Link URLs and registration offers +# ------------------------------------------------------------------------- + + +def test_parse_link_url(): + base, key = satellite.parse_link_url( + "wss://alpha.bravo.charlie.delta.echo@auth.example.com" + ) + assert base == "https://auth.example.com" + assert key == "alpha.bravo.charlie.delta.echo" + base, key = satellite.parse_link_url("ws://k3y@127.0.0.1:4401") + assert base == "http://127.0.0.1:4401" + assert key == "k3y" + with pytest.raises(ValueError, match="wss"): + satellite.parse_link_url("https://alpha.bravo@auth.example.com") + with pytest.raises(ValueError, match="key"): + satellite.parse_link_url("wss://auth.example.com") + + +def test_offer_mint_and_spend(): + """Offers are wss URLs with a RAM-only, single-use, expiring key.""" + url = sync.create_offer(["example.com"], "https://auth.example.com") + assert url.startswith("wss://") + assert url.endswith("@auth.example.com") + key = url.split("://", 1)[1].split("@", 1)[0] + offer = sync._pop_offer(key) + assert offer is not None and offer.rp_ids == ["example.com"] + assert sync._pop_offer(key) is None # single-use + assert sync._pop_offer("unknown.key.here") is None + + expired = sync.create_offer([], "http://127.0.0.1:4401") + assert expired.startswith("ws://") + key = expired.split("://", 1)[1].split("@", 1)[0] + sync._offers[key].created -= sync.OFFER_TTL + 1 + assert sync._pop_offer(key) is None # expired + + +# ------------------------------------------------------------------------- +# Pushed-domain application (satellite side) +# ------------------------------------------------------------------------- + +_PUSHED = { + "rp_id": "example.com", + "rp_name": "Example", + "origins": {"**.example.com": True, "auth.example.com": {"auth_host": True}}, +} + + +@pytest.mark.asyncio +async def test_apply_domains_push_accept_and_remove(test_db, domain_registry): + """A push persists the accepted set (marked remote via the link); a + later push without the domain removes it.""" + rejected = satellite.apply_domains_push(UPLINK_ID, [_PUSHED]) + assert not rejected + domain = test_db.config.domains["example.com"] + assert domain.remote.link == UPLINK_ID + assert domain.rp_name == "Example" + assert domains.registry().get("example.com").remote is not None + + # Idempotent: the same push again changes nothing + assert not satellite.apply_domains_push(UPLINK_ID, [_PUSHED]) + + # The uplink's scope shrank: the domain leaves the local config + rejected = satellite.apply_domains_push(UPLINK_ID, []) + assert not rejected + assert "example.com" not in test_db.config.domains + assert domains.registry().get("example.com") is None + + +@pytest.mark.asyncio +async def test_apply_domains_push_rejects_local_conflict(test_db, domain_registry): + """A pushed rp-id that this instance serves locally is rejected.""" + rejected = satellite.apply_domains_push( + UPLINK_ID, [{**_PUSHED, "rp_id": TEST_RP_ID}] + ) + assert TEST_RP_ID in rejected + assert test_db.config.domains[TEST_RP_ID].remote is None + + +@pytest.mark.asyncio +async def test_apply_domains_push_rejects_auth_host_collision(test_db, domain_registry): + """A pushed auth host that collides with a local rp-id is rejected + (the merged config must stay servable).""" + rejected = satellite.apply_domains_push( + UPLINK_ID, + [ + { + "rp_id": "example.com", + "rp_name": "Example", + "origins": {"**.example.com": True, TEST_RP_ID: {"auth_host": True}}, + } + ], + ) + assert "example.com" in rejected + assert "example.com" not in test_db.config.domains + + # ------------------------------------------------------------------------- # API-level: endpoints served from an injected replica # ------------------------------------------------------------------------- @@ -289,20 +396,19 @@ async def remote_client(test_db): config = _remote_domain_config() domains.configure(listen=["localhost:4401"]) domains.init_registry(config) + ops_db.create_uplink(UPLINK_ID, _uplink(), {}) replica_db, secret = _replica_db() - replica = satellite.RemoteReplica( - RemoteConfig(url=REMOTE_URL, token="t"), "example.com" - ) + replica = satellite.UplinkReplica(_uplink()) replica.db = replica_db replica.last_contact = time.monotonic() replica.connected = True - satellite.manager.replicas["example.com"] = replica + satellite.manager.replicas[UPLINK_ID] = replica 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("example.com", None) + satellite.manager.replicas.pop(UPLINK_ID, None) @pytest.mark.asyncio @@ -373,178 +479,58 @@ async def test_logout_proxied_and_evicted(remote_client, monkeypatch): @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 - +async def test_admin_rejects_remote_writes(client, session_token, test_db): + """Remote domains are uplink-managed: the domains API refuses to + create, edit or delete them.""" headers = { "Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", } - origins = {"**.example.com": True, "auth.example.com": {"auth_host": True}} - - # PATCH without the remote key preserves it (and its token) - r = await client.patch( - "/auth/api/admin/domains/example.com", - json={"rp_name": "Ex", "origins": origins}, - headers=headers, - ) - assert r.status_code == 200, r.text - assert stored.remote.url == "http://remote.test" - assert stored.remote.token == "sekret" - - # PATCH with a new URL but no token keeps the stored token - r = await client.patch( - "/auth/api/admin/domains/example.com", - json={ - "rp_name": "Ex", - "origins": origins, - "remote": {"url": "http://other.test", "cache_ttl": 30}, - }, - headers=headers, - ) - assert r.status_code == 200, r.text - assert stored.remote.url == "http://other.test" - assert stored.remote.token == "sekret" - - # PATCH with remote: null clears it - r = await client.patch( - "/auth/api/admin/domains/example.com", - json={"rp_name": "Ex", "origins": origins, "remote": None}, - headers=headers, - ) - assert r.status_code == 200, r.text - assert stored.remote is None - - -@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}", + "origins": { + "**.example.com": True, + "auth.example.com": {"auth_host": True}, + }, + "remote": {"link": "x"}, }, + headers=headers, ) assert r.status_code == 400 - assert "auth host" in r.json()["detail"] + assert "uplink" in r.json()["detail"] - -@pytest.mark.asyncio -async def test_admin_remote_values_clamped(client, session_token, test_db): - """Nonsensical TTL/interval values are clamped to the supported floor.""" - r = await client.post( - "/auth/api/admin/domains/", - json={ - "rp_id": "example.com", - "origins": { - "**.example.com": True, - "auth.example.com": {"auth_host": True}, - }, - "remote": { - "url": "http://remote.test", - "token": "t", - "cache_ttl": -5, - "refresh_interval": 1, - }, - }, - headers={ - "Host": "localhost:4401", - "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + # An existing remote domain (as an uplink push would have stored it) + ops_db.apply_uplink_domains( + UPLINK_ID, + { + "example.com": DomainConfig( + rp_name="Example", + origins={ + "**.example.com": True, + "auth.example.com": OriginEntry(auth_host=True), + }, + remote=RemoteConfig(link=UPLINK_ID), + ) }, ) - assert r.status_code == 200, r.text - remote = test_db.config.domains["example.com"].remote - assert remote.cache_ttl == 1 - assert remote.refresh_interval == 30 - - -@pytest.mark.asyncio -async def test_admin_remote_url_userinfo_token(client, session_token, test_db): - """The sync URL minted on an upstream carries the token as userinfo; - it is split off before storing (and never echoed back).""" - r = await client.post( - "/auth/api/admin/domains/", - json={ - "rp_id": "example.com", - "origins": { - "**.example.com": True, - "auth.example.com": {"auth_host": True}, - }, - "remote": {"url": "https://alpha.bravo.charlie.delta.echo@remote.test"}, - }, - headers={ - "Host": "localhost:4401", - "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", - }, - ) - assert r.status_code == 200, r.text - remote = test_db.config.domains["example.com"].remote - assert remote.url == "https://remote.test" - assert remote.token == "alpha.bravo.charlie.delta.echo" - - # Saving without userinfo keeps the stored token r = await client.patch( "/auth/api/admin/domains/example.com", - json={ - "rp_name": "Ex", - "origins": { - "**.example.com": True, - "auth.example.com": {"auth_host": True}, - }, - "remote": {"url": "https://remote.test"}, - }, - headers={ - "Host": "localhost:4401", - "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", - }, - ) - assert r.status_code == 200, r.text - assert ( - test_db.config.domains["example.com"].remote.token - == "alpha.bravo.charlie.delta.echo" + json={"rp_name": "Ex", "origins": {"**.example.com": True}}, + headers=headers, ) + assert r.status_code == 400 + assert "uplink" in r.json()["detail"] + r = await client.delete("/auth/api/admin/domains/example.com", headers=headers) + assert r.status_code == 400 + assert "unlink" in r.json()["detail"] def test_feed_scoped_subscription_filters(): - """A scoped subscriber gets only its rp-id's sessions/credentials + """A scoped subscriber gets only its rp-ids' sessions/credentials (upserts filtered, deletes passed); the user graph is never published.""" - scoped = syncfeed.subscribe("example.com") + scoped = syncfeed.subscribe({"example.com"}) try: in_scope = Session.create( user=UUID(int=1), @@ -575,6 +561,31 @@ def test_feed_scoped_subscription_filters(): syncfeed.unsubscribe(scoped) +def test_feed_scope_updates_live(): + """The subscriber scope is mutable in place: a link's scope adjustment + takes effect without a reconnect.""" + scope = {"example.com"} + queue = syncfeed.subscribe(scope) + try: + session = Session.create( + user=UUID(int=1), + credential=UUID(int=2), + key="k1", + host="localhost", + ip="1.1.1.1", + user_agent="ua", + validated=datetime.now(UTC), + rp_id="localhost", + ) + syncfeed.emit("sessions", "k1", session) + assert not _drain(queue) + scope.add("localhost") # scope widened live + syncfeed.emit("sessions", "k1", session) + assert [e["key"] for e in _drain(queue)] == ["k1"] + finally: + syncfeed.unsubscribe(queue) + + def _drain(queue): items = [] while not queue.empty(): @@ -658,18 +669,17 @@ async def test_user_api_proxied_on_remote_auth_host(remote_client, monkeypatch): # ------------------------------------------------------------------------- -# Sync tokens (this instance as the upstream) +# Satellite links and uplinks (admin API) # ------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_sync_token_lifecycle(client, session_token, test_db): - """Mint (auth host required), list, revoke — the raw token appears once.""" +async def test_admin_satellite_link_lifecycle(client, session_token, test_db): + """Offer minting (auth host required), scope adjustment, revocation.""" headers = { "Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", } - r = await client.post( "/auth/api/admin/domains/", json={ @@ -681,14 +691,13 @@ async def test_sync_token_lifecycle(client, session_token, test_db): ) assert r.status_code == 200, r.text - # No auth host yet — refused + # No auth host yet — the domain cannot be scoped into a link r = await client.post( - "/auth/api/admin/domains/example.com/sync-token", headers=headers + "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers ) assert r.status_code == 400 assert "auth host" in r.json()["detail"] - # Mark an auth host, then mint r = await client.patch( "/auth/api/admin/domains/example.com", json={ @@ -701,55 +710,155 @@ async def test_sync_token_lifecycle(client, session_token, test_db): headers=headers, ) assert r.status_code == 200, r.text + + # Mint an offer: a wss URL carrying the RAM-only initial key r = await client.post( - "/auth/api/admin/domains/example.com/sync-token", headers=headers + "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers ) assert r.status_code == 200, r.text body = r.json() - assert body["url"].startswith("https://") + assert body["url"].startswith("wss://") assert "@auth.example.com" in body["url"] - raw_token = body["url"].split("://", 1)[1].split("@", 1)[0] - assert raw_token.count(".") == 4 # five-word passphrase + key = body["url"].split("://", 1)[1].split("@", 1)[0] + assert key.count(".") == 4 # five-word passphrase + assert body["expires_in"] > 0 - r = await client.get( - "/auth/api/admin/domains/example.com/sync-tokens", headers=headers + # Unknown domains and nested scopes are refused + r = await client.post( + "/auth/api/admin/satellites/", + json={"rp_ids": ["nope.example"]}, + headers=headers, ) - assert [t["id"] for t in r.json()] == [body["id"]] - assert "alpha" not in r.text and raw_token not in r.text # hashes only + assert r.status_code == 400 - r = await client.delete( - f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers + # Register a link directly (the handshake itself is covered live), + # then list, adjust scope, revoke + ops_db.create_satellite_link("ln1", ["example.com"], b"key", label="sat-1") + r = await client.get("/auth/api/admin/satellites/", headers=headers) + assert r.status_code == 200 + (entry,) = r.json() + assert entry["id"] == "ln1" + assert entry["label"] == "sat-1" + assert entry["rp_ids"] == ["example.com"] + assert entry["connected"] is False + + r = await client.patch( + "/auth/api/admin/satellites/ln1", json={"rp_ids": []}, headers=headers ) assert r.status_code == 200 - assert not test_db.sync_tokens - r = await client.delete( - f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers - ) + assert test_db.satellite_links["ln1"].rp_ids == [] + + r = await client.delete("/auth/api/admin/satellites/ln1", headers=headers) + assert r.status_code == 200 + assert not test_db.satellite_links + r = await client.delete("/auth/api/admin/satellites/ln1", headers=headers) assert r.status_code == 400 @pytest.mark.asyncio -async def test_sync_token_refused_for_remote_domain(client, session_token): - """A satellite-served domain cannot be an upstream: no nesting.""" +async def test_admin_offer_scope_rejects_remote_domain(client, session_token, test_db): + """A satellite-served domain cannot be scoped into a link: no nesting.""" headers = { "Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", } - r = await client.post( - "/auth/api/admin/domains/", - json={ - "rp_id": "example.com", - "origins": { - "**.example.com": True, - "auth.example.com": {"auth_host": True}, - }, - "remote": {"url": "https://alpha.bravo.charlie.delta.echo@remote.test"}, + ops_db.apply_uplink_domains( + UPLINK_ID, + { + "example.com": DomainConfig( + origins={ + "**.example.com": True, + "auth.example.com": OriginEntry(auth_host=True), + }, + remote=RemoteConfig(link=UPLINK_ID), + ) }, - headers=headers, ) - assert r.status_code == 200, r.text r = await client.post( - "/auth/api/admin/domains/example.com/sync-token", headers=headers + "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers ) assert r.status_code == 400 assert "nested" in r.json()["detail"] + + +@pytest.mark.asyncio +async def test_admin_uplinks_list_and_unlink(client, session_token, test_db): + """Uplinks list shows served domains and replica state; unlink removes + the uplink and its domains.""" + headers = { + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + } + ops_db.create_uplink(UPLINK_ID, _uplink(), {}) + ops_db.apply_uplink_domains( + UPLINK_ID, + { + "example.com": DomainConfig( + origins={ + "**.example.com": True, + "auth.example.com": OriginEntry(auth_host=True), + }, + remote=RemoteConfig(link=UPLINK_ID), + ) + }, + ) + r = await client.get("/auth/api/admin/uplinks/", headers=headers) + assert r.status_code == 200 + (entry,) = r.json() + assert entry["id"] == UPLINK_ID + assert entry["url"] == REMOTE_URL + assert entry["domains"] == ["example.com"] + assert entry["connected"] is False + + r = await client.delete(f"/auth/api/admin/uplinks/{UPLINK_ID}", headers=headers) + assert r.status_code == 200 + assert not test_db.uplinks + assert "example.com" not in test_db.config.domains + r = await client.get("/auth/api/admin/uplinks/", headers=headers) + assert r.json() == [] + + +@pytest.mark.asyncio +async def test_admin_register_uplink_endpoint(client, session_token, monkeypatch): + """The registration endpoint runs the handshake synchronously (the + handshake itself is covered by the live tests) and reports failures.""" + headers = { + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + } + + async def fake_register(url): + return { + "link": "ln1", + "url": "https://remote.test", + "domains": ["example.com"], + "rejected": {}, + } + + monkeypatch.setattr(satellite, "register_uplink", fake_register) + r = await client.post( + "/auth/api/admin/uplinks/", + json={"url": "wss://alpha.bravo.charlie.delta.echo@remote.test"}, + headers=headers, + ) + assert r.status_code == 200 + assert r.json()["domains"] == ["example.com"] + + async def failing_register(url): + raise RuntimeError("the upstream rejected the link URL") + + monkeypatch.setattr(satellite, "register_uplink", failing_register) + r = await client.post( + "/auth/api/admin/uplinks/", json={"url": "wss://x@remote.test"}, headers=headers + ) + assert r.status_code == 400 + assert "rejected" in r.json()["detail"] + + async def malformed(url): + raise ValueError("A link URL is a wss:// URL") + + monkeypatch.setattr(satellite, "register_uplink", malformed) + r = await client.post( + "/auth/api/admin/uplinks/", json={"url": "nope"}, headers=headers + ) + assert r.status_code == 400 diff --git a/tests/test_satellite_e2e.py b/tests/test_satellite_e2e.py index c05a69f..1391a2e 100644 --- a/tests/test_satellite_e2e.py +++ b/tests/test_satellite_e2e.py @@ -1,16 +1,20 @@ """Live end-to-end tests for satellite domains over a real sync channel. -A real uvicorn server plays the remote (its sync WebSocket fed from the +A real uvicorn server plays the upstream (its sync WebSocket fed from the process-global database), while the satellite side runs through the ASGI -app with a real RemoteReplica started by the real SatelliteManager -reconcile. Unlike test_remote.py (which injects a warm replica by hand), -nothing here is stubbed: the snapshot, live events, write-behind, dead-peer -TTL handling and reconnect reconciliation all cross a TCP socket. +app with a real UplinkReplica started by the real SatelliteManager. Unlike +test_remote.py (which injects a warm replica by hand), nothing here is +stubbed: the 3-way handshake, pushed domains, snapshot, live events, +write-behind, dead-peer TTL handling, reconnect reconciliation, live scope +changes and revocation all cross a TCP socket. One in-process distortion remains: both sides share the module-level -database and registry, so endpoints the satellite *forwards* to the remote -(logout, set-session, OIDC) would re-forward and loop here — those paths -are covered in test_remote.py with a mocked forward_request. +database and registry, so the satellite's pushed-domain application is +arranged to be a no-op (the link's domains are absent from the shared +config — the satellite registry is synthetic instead), and endpoints the +satellite *forwards* to the upstream (logout, set-session, OIDC) would +re-forward and loop here — those paths are covered in test_remote.py with +a mocked forward_request. """ import asyncio @@ -22,6 +26,7 @@ from types import SimpleNamespace from uuid import UUID import httpx +import msgspec.structs import pytest import pytest_asyncio import uvicorn @@ -34,7 +39,9 @@ from paskia.db.structs import ( DomainConfig, OriginEntry, RemoteConfig, + Uplink, ) +from paskia.fastapi import sync from paskia.fastapi.mainapp import app from paskia.fastapi.session import AUTH_COOKIE_NAME @@ -42,6 +49,8 @@ from .conftest import TEST_RP_ID, create_test_session REMOTE_DOMAIN = "example.com" APP_HOST = "app2.example.com" +LINK_ID = "live-link" +LINK_KEY = b"e2e-link-key" def _free_port() -> int: @@ -60,15 +69,12 @@ async def wait_for(desc: str, condition, timeout: float = 10) -> None: @pytest_asyncio.fixture -async def live_remote(test_db): - """A uvicorn-served 'remote' and the satellite manager wired against it. +async def live_server(test_db): + """A uvicorn-served upstream over a real TCP socket. - The manager is started for real: reconcile builds the replica from the - domain config and its sync client connects over TCP. Yielded namespace: - client (ASGI, satellite side), replica, db (the remote's database), - url/port of the remote, start_server (for restarts), servers. + Yields a namespace: db, url, port, start_server (for restarts), + servers. """ - _, live_token = ops_db.create_sync_token(REMOTE_DOMAIN) servers: list[tuple[uvicorn.Server, asyncio.Task]] = [] async def start_server(port: int) -> uvicorn.Server: @@ -84,7 +90,34 @@ async def live_remote(test_db): port = _free_port() url = f"http://127.0.0.1:{port}" + domains.configure(listen=["localhost:4401"]) + domains.init_registry(test_db.config) await start_server(port) + try: + yield SimpleNamespace( + db=test_db, url=url, port=port, start_server=start_server, servers=servers + ) + finally: + for server, task in servers: + server.should_exit = True + await task + + +@pytest_asyncio.fixture +async def live_remote(live_server): + """An established link against the live upstream, and the satellite + manager started for real: the replica connects over TCP and runs the + established-link handshake (hello {link, key} → welcome → ready). + + Both sides' link rows are stored directly (registration itself is + covered by the dedicated tests); the satellite's remote domain lives + only in the registry, not the shared config, so pushed-domain + application stays a no-op. + """ + live = live_server + ops_db.create_satellite_link(LINK_ID, [REMOTE_DOMAIN], LINK_KEY, label="e2e") + uplink = Uplink(url=live.url, created=datetime.now(UTC), key=LINK_KEY) + ops_db.create_uplink(LINK_ID, uplink, {}) domains.configure(listen=["localhost:4401"]) domains.init_registry( @@ -96,13 +129,13 @@ async def live_remote(test_db): f"**.{REMOTE_DOMAIN}": True, f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True), }, - remote=RemoteConfig(url=url, token=live_token), + remote=RemoteConfig(link=LINK_ID), ), } ) ) await satellite.manager.start() - replica = satellite.manager.replicas[REMOTE_DOMAIN] + replica = satellite.manager.replicas[LINK_ID] await wait_for("initial snapshot", lambda: replica.available()) transport = httpx.ASGITransport(app=app) @@ -110,21 +143,11 @@ async def live_remote(test_db): transport=transport, base_url="http://localhost:4401" ) as client: try: - yield SimpleNamespace( - client=client, - replica=replica, - db=test_db, - url=url, - port=port, - token=live_token, - start_server=start_server, - servers=servers, - ) + live.client = client + live.replica = replica + yield live finally: await satellite.manager.stop() - for server, task in servers: - server.should_exit = True - await task def _app_headers(secret: str) -> dict[str, str]: @@ -134,7 +157,7 @@ def _app_headers(secret: str) -> dict[str, str]: @pytest_asyncio.fixture async def remote_credential(test_db, test_user): """A credential registered on the remote domain. The feed is rp-id - scoped: only the domain's own credentials replicate to its satellites.""" + scoped: only the link's own domains' credentials replicate.""" credential = Credential.create( credential_id=os.urandom(32), user=test_user.uuid, @@ -147,11 +170,72 @@ async def remote_credential(test_db, test_user): return credential +# ------------------------------------------------------------------------- +# Registration (the real 3-way handshake, crossing TCP) +# ------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_live_registration(live_server): + """A RAM-only offer is spent by the handshake; both sides store the + link with identical key material, and the established-link handshake + works right after.""" + live = live_server + offer_url = sync.create_offer([], live.url) + assert offer_url.startswith("ws://") + + result = await satellite.register_uplink(offer_url) + link_id = result["link"] + assert result["domains"] == [] # empty scope: register first, scope later + + uplink = live.db.uplinks[link_id] + assert uplink.url == live.url + link = live.db.satellite_links[link_id] + assert uplink.key == link.key and len(link.key) == 64 # both parts + assert link.label # the satellite's self-reported name + + # Re-registering to the same upstream is refused (the offer itself is + # single-use — covered in test_remote.py's offer test) + with pytest.raises(ValueError, match="Already linked"): + await satellite.register_uplink(offer_url) + + # The established-link handshake: the replica connects and syncs + await satellite.manager.start() + try: + replica = satellite.manager.replicas[link_id] + await wait_for("established link sync", lambda: replica.available()) + finally: + await satellite.manager.stop() + + +@pytest.mark.asyncio +async def test_live_registration_rejections(live_server): + """Malformed URLs, unknown keys and unreachable hosts all fail the + registration without storing anything.""" + with pytest.raises(ValueError, match="wss"): + await satellite.register_uplink("https://key@auth.example.com") + with pytest.raises(ValueError, match="key"): + await satellite.register_uplink("wss://auth.example.com") + with pytest.raises(RuntimeError, match="rejected"): + await satellite.register_uplink( + f"ws://unknown.key@127.0.0.1:{live_server.port}" + ) + with pytest.raises(RuntimeError, match="unreachable"): + await satellite.register_uplink("ws://some.key@127.0.0.1:1") + assert not live_server.db.uplinks + assert not live_server.db.satellite_links + + +# ------------------------------------------------------------------------- +# Established link: sync behavior +# ------------------------------------------------------------------------- + + @pytest.mark.asyncio async def test_live_snapshot_and_context_repush( live_remote, test_user, remote_credential ): - """Sessions arrive with their master-computed context; user-graph + """Sessions arrive with their upstream-computed context; user-graph changes eagerly re-push the affected sessions' contexts.""" live = live_remote key, _ = create_test_session( @@ -207,8 +291,8 @@ async def test_live_permission_revocation_propagates( @pytest.mark.asyncio async def test_live_forward_and_writebehind(live_remote, test_user, remote_credential): - """A session created on the remote authorizes locally; the /validate - refresh travels back over the wire into the remote's database.""" + """A session created on the upstream authorizes locally; the /validate + refresh travels back over the wire into the upstream's database.""" live = live_remote key, secret = create_test_session( user_uuid=test_user.uuid, @@ -224,7 +308,7 @@ async def test_live_forward_and_writebehind(live_remote, test_user, remote_crede assert r.status_code == 204 assert r.headers["remote-name"] == "Test Admin" - # Age the session on the remote (propagates), then validate on the satellite + # Age the session on the upstream (propagates), then validate on the satellite stale = datetime.now(UTC) - timedelta(minutes=10) ops_db.update_session(key, validated=stale) await wait_for( @@ -235,7 +319,7 @@ async def test_live_forward_and_writebehind(live_remote, test_user, remote_crede assert r.status_code == 200 assert r.json()["renewed"] is True await wait_for( - "write-behind to land on the remote", + "write-behind to land on the upstream", lambda: live.db.sessions[key].validated > stale, ) @@ -262,10 +346,11 @@ async def test_live_session_delete_propagates( @pytest.mark.asyncio -async def test_live_wrong_token_rejected(live_remote): - bad = satellite.RemoteReplica( - RemoteConfig(url=live_remote.url, token="wrong"), REMOTE_DOMAIN - ) +async def test_live_wrong_key_rejected(live_remote): + """An uplink whose key does not match the stored link's never syncs.""" + uplink = Uplink(url=live_remote.url, created=datetime.now(UTC), key=b"not-the-key") + uplink.uplink_id = "bad-link" + bad = satellite.UplinkReplica(uplink) await bad.start() try: await asyncio.sleep(1.5) @@ -299,7 +384,10 @@ async def test_live_disconnect_ttl_and_reconnect( r = await live.client.get("/auth/api/forward", headers=_app_headers(secret)) assert r.status_code == 204 # still authoritative within cache_ttl - live.replica.remote.cache_ttl = 0 # expire the trust window + # Expire the trust window (a copy: the db-owned struct is not mutated) + uplink_copy = msgspec.structs.replace(live.replica.uplink, cache_ttl=0) + uplink_copy.uplink_id = LINK_ID + live.replica.uplink = uplink_copy r = await live.client.get("/auth/api/forward", headers=_app_headers(secret)) assert r.status_code == 503 @@ -315,21 +403,32 @@ async def test_live_disconnect_ttl_and_reconnect( @pytest.mark.asyncio -async def test_reconcile_stops_replica_when_remote_removed(live_remote): - """A registry rebuild without the remote config stops the replica.""" - domains.init_registry( - Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})}) - ) +async def test_unlink_stops_replica(live_remote): + """Unlinking drops the uplink, its domains and the replica.""" + live = live_remote + await satellite.manager.unlink(LINK_ID) + assert LINK_ID not in satellite.manager.replicas + assert LINK_ID not in live.db.uplinks + + +@pytest.mark.asyncio +async def test_live_revocation_unlinks_satellite(live_remote): + """Revoking the link upstream drops the connection; the satellite's + handshake retry gets a policy close and unlinks its side.""" + live = live_remote + ops_db.delete_satellite_link(LINK_ID) + await sync.close_link(LINK_ID) await wait_for( - "replica shutdown", lambda: REMOTE_DOMAIN not in satellite.manager.replicas + "satellite to unlink", lambda: LINK_ID not in live.db.uplinks, timeout=25 ) + assert LINK_ID not in satellite.manager.replicas @pytest.mark.asyncio async def test_live_out_of_scope_session_not_replicated( live_remote, test_user, test_credential, remote_credential ): - """The feed is scoped to the token's rp-id: sessions of other domains + """The feed is scoped to the link's rp-ids: sessions of other domains (here the satellite's own) never cross the wire.""" live = live_remote local_key, _ = create_test_session( @@ -356,8 +455,8 @@ async def test_live_out_of_scope_session_not_replicated( async def test_live_out_of_scope_writebehind_ignored( live_remote, test_user, test_credential, remote_credential ): - """A session_refresh for a session outside the token's scope is ignored - by the remote (fail-closed), while an in-scope one lands.""" + """A session_refresh for a session outside the link's scope is ignored + by the upstream (fail-closed), while an in-scope one lands.""" live = live_remote stale = datetime.now(UTC) - timedelta(minutes=10) local_key, _ = create_test_session( @@ -386,15 +485,57 @@ async def test_live_out_of_scope_writebehind_ignored( @pytest.mark.asyncio -async def test_live_scope_mismatch_rejected(live_remote): - """A valid token scoped to another domain is useless: the ready message - names its scope and the replica refuses to go live on the mismatch.""" - bad = satellite.RemoteReplica( - RemoteConfig(url=live_remote.url, token=live_remote.token), "other.org" +async def test_live_scope_shrink_stops_feed(live_remote, test_user, remote_credential): + """Narrowing the link's scope on the upstream takes effect on the live + connection: sessions of removed domains stop flowing.""" + live = live_remote + ops_db.set_satellite_link_scope(LINK_ID, []) + await sync.push_link_domains(LINK_ID) + + removed_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, ) - await bad.start() - try: - await asyncio.sleep(1.5) - assert not bad.available() - finally: - await bad.stop() + # Positive control: widen the scope again; a session created now does + # arrive, so the earlier one demonstrably passed a closed pipe. + ops_db.set_satellite_link_scope(LINK_ID, [REMOTE_DOMAIN]) + await sync.push_link_domains(LINK_ID) + remote_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, + ) + await wait_for( + "re-scoped session sync", lambda: remote_key in live.replica.db.sessions + ) + assert removed_key not in live.replica.db.sessions + + +@pytest.mark.asyncio +async def test_live_domain_conflict_reported_upstream(live_remote, test_db): + """A pushed domain the satellite serves locally is rejected, and the + rejection is reported into the upstream's link state.""" + # The shared config now holds the domain authoritatively — from the + # satellite's view this is the local-conflict case. + ops_db.create_domain( + REMOTE_DOMAIN, + DomainConfig( + rp_name="Example", + origins={ + f"**.{REMOTE_DOMAIN}": True, + f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True), + }, + ), + ) + await sync.push_link_domains(LINK_ID) + await wait_for( + "rejection to arrive", + lambda: ( + REMOTE_DOMAIN in sync.links_status().get(LINK_ID, {}).get("rejected", {}) + ), + ) + # The satellite kept its (synthetic) registry — no local overwrite + assert test_db.config.domains[REMOTE_DOMAIN].remote is None