From 0a59e26e45936b3949d46892e09a95dcaafd7e08 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 1 Oct 2026 06:40:07 +0000 Subject: [PATCH] Bound handshake reads; audit ctx for domain-delete scope shrinking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An accepted-but-silent peer no longer holds a sync connection open forever: each dialogue frame read has a 15 s bound, closed with 1011 — never 1008, which the satellite reads as revocation. Domain deletion passes the admin's session context to link scope shrinks for the audit log. --- paskia/fastapi/admin/domains.py | 2 +- paskia/fastapi/sync.py | 27 ++++++++++++++++++++++----- paskia/satellite.py | 17 +++++++++++++---- 3 files changed, 36 insertions(+), 10 deletions(-) diff --git a/paskia/fastapi/admin/domains.py b/paskia/fastapi/admin/domains.py index 7305e54..111d588 100644 --- a/paskia/fastapi/admin/domains.py +++ b/paskia/fastapi/admin/domains.py @@ -219,5 +219,5 @@ async def admin_delete_domain( ) db.delete_domain(rp_id, ctx=ctx) _rebuild_registry() - await sync.drop_domain_from_links(rp_id) + await sync.drop_domain_from_links(rp_id, ctx=ctx) return {"status": "ok"} diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py index 4f83a0c..775f9ba 100644 --- a/paskia/fastapi/sync.py +++ b/paskia/fastapi/sync.py @@ -49,6 +49,7 @@ _logger = logging.getLogger(__name__) app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) OFFER_TTL = 600 # registration offers live 10 minutes, RAM-only +_HANDSHAKE_TIMEOUT = 15 # seconds per dialogue frame before a plain close @dataclass @@ -158,11 +159,13 @@ async def refresh_links_serving(rp_id: str) -> None: await push_link_domains(link_id) -async def drop_domain_from_links(rp_id: str) -> None: +async def drop_domain_from_links(rp_id: str, *, ctx=None) -> 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], ctx=ctx + ) await push_link_domains(link_id) @@ -188,9 +191,17 @@ async def _handshake(ws: WebSocket) -> _Connection | None: """The 3-way link dialogue: hello → welcome → ready. Returns the connection state for an accepted link, or None after - closing on rejection (bad offer, unknown link or key). + closing on rejection (bad offer, unknown link or key). Reads are + bounded: a peer that stalls mid-dialogue gets a plain server-error + close — never 1008, which the satellite reads as revocation. """ - hello = msgspec.json.decode(await ws.receive_bytes()) + try: + hello = msgspec.json.decode( + await asyncio.wait_for(ws.receive_bytes(), _HANDSHAKE_TIMEOUT) + ) + except TimeoutError: + await ws.close(code=1011) + return None if hello.get("type") != "hello": await ws.close(code=1008) return None @@ -225,7 +236,13 @@ async def _handshake(ws: WebSocket) -> _Connection | None: welcome["key"] = server_part.hex() await _send(ws, welcome) - ready = msgspec.json.decode(await ws.receive_bytes()) + try: + ready = msgspec.json.decode( + await asyncio.wait_for(ws.receive_bytes(), _HANDSHAKE_TIMEOUT) + ) + except TimeoutError: + await ws.close(code=1011) + return None if ready.get("type") != "ready": await ws.close(code=1008) return None diff --git a/paskia/satellite.py b/paskia/satellite.py index a854ac9..c405c30 100644 --- a/paskia/satellite.py +++ b/paskia/satellite.py @@ -72,6 +72,7 @@ _TABLES = { _RECONNECT_DELAY = 5 _SWEEP_INTERVAL = 60 +_HANDSHAKE_TIMEOUT = 15 # seconds the upstream may take to answer a hello class LinkRevokedError(Exception): @@ -274,11 +275,15 @@ async def register_uplink(url: str) -> dict: ) ) try: - welcome = msgspec.json.decode(await ws.recv()) + welcome = msgspec.json.decode( + await asyncio.wait_for(ws.recv(), _HANDSHAKE_TIMEOUT) + ) except websockets.exceptions.ConnectionClosed as e: raise RuntimeError( "the upstream rejected the link URL (unknown or expired offer)" ) from e + except TimeoutError as e: + raise RuntimeError("the upstream did not answer in time") from e if welcome.get("type") != "welcome": raise RuntimeError("the upstream did not complete the registration") link_id = welcome.get("link") or "" @@ -398,9 +403,13 @@ class UplinkReplica: if not self._stopped: await asyncio.sleep(_RECONNECT_DELAY) - async def _recv(self, ws) -> dict: + async def _recv(self, ws, timeout: float | None = None) -> dict: try: - return msgspec.json.decode(await ws.recv()) + if timeout is None: + message = await ws.recv() + else: + message = await asyncio.wait_for(ws.recv(), timeout) + return msgspec.json.decode(message) except websockets.exceptions.ConnectionClosedError as e: if e.code == 1008: raise LinkRevokedError(self.uplink.url) from e @@ -423,7 +432,7 @@ class UplinkReplica: } ) ) - welcome = await self._recv(ws) + welcome = await self._recv(ws, timeout=_HANDSHAKE_TIMEOUT) if welcome.get("type") != "welcome": raise ValueError("sync: protocol error (expected welcome)") # The upstream re-pushes the link's domain set on every connect