Bound handshake reads; audit ctx for domain-delete scope shrinking

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

@@ -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"}

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

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