LeoVasanko
0a59e26e45
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.
341 lines
12 KiB
Python
341 lines
12 KiB
Python
"""Sync WebSocket endpoint: link handshake, snapshots and live events.
|
|
|
|
One endpoint serves both registration and established links — the first
|
|
binary frame decides. Registration is a 3-way dialogue:
|
|
|
|
satellite → upstream hello {link: null, key: <initial key>, label}
|
|
upstream → satellite welcome {link, key: <server part>, domains: [...]}
|
|
satellite → upstream ready {key: <client part>}
|
|
|
|
The initial key comes from a registration offer minted in the admin UI:
|
|
held in RAM only, short-lived and single-use. The permanent link
|
|
credential ``key`` is created during the handshake from parts both sides
|
|
contribute; both sides then store the link in their databases.
|
|
(Placeholder cryptography: the parts are exchanged as-is — the final
|
|
version derives the channel key from the handshake without sending key
|
|
material.)
|
|
|
|
An established link runs the same dialogue with hello {link, key} proving
|
|
the stored key; the welcome re-pushes the link's domain set (the scope may
|
|
have changed while offline), the ready is a bare ack. A link's scope is
|
|
adjustable at any time and pushed live as a ``domains`` frame — no new
|
|
cryptographic setup.
|
|
|
|
After the handshake the channel switches to plain JSON frames (to be
|
|
transparently encrypted in the final version): snapshot chunks per table,
|
|
`ready`, then live upsert/delete events within the link's scope; the
|
|
satellite sends session_refresh write-backs and domains_ack reports.
|
|
Reconnects always restart from a snapshot. Beyond the link table all
|
|
state is RAM-only (syncfeed).
|
|
"""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from urllib.parse import urlsplit
|
|
|
|
import msgspec
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
|
|
|
from paskia import db, syncfeed
|
|
from paskia.util.passphrase import generate as generate_passphrase
|
|
|
|
_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
|
|
class _Offer:
|
|
rp_ids: list[str]
|
|
created: float # monotonic
|
|
|
|
|
|
@dataclass
|
|
class _Connection:
|
|
"""One live link connection: its WebSocket, link id, live feed scope
|
|
(mutated in place on scope changes) and the satellite's latest
|
|
domains_ack rejections."""
|
|
|
|
ws: WebSocket
|
|
link_id: str
|
|
scope: set[str]
|
|
rejected: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
_offers: dict[str, _Offer] = {}
|
|
_connections: dict[asyncio.Queue, _Connection] = {}
|
|
|
|
|
|
def create_offer(rp_ids: list[str], base_url: str) -> str:
|
|
"""Mint a RAM-only registration offer; returns the wss link URL.
|
|
|
|
The URL is meant only for a satellite to connect to: a wss:// scheme
|
|
keeps browsers from opening it. The initial key travels as userinfo and
|
|
is never stored — losing the URL just means minting a new one.
|
|
"""
|
|
key = generate_passphrase()
|
|
now = time.monotonic()
|
|
for k, offer in list(_offers.items()): # lazy expiry sweep
|
|
if now - offer.created > OFFER_TTL:
|
|
_offers.pop(k, None)
|
|
_offers[key] = _Offer(rp_ids=list(rp_ids), created=now)
|
|
parts = urlsplit(base_url)
|
|
scheme = "wss" if parts.scheme == "https" else "ws"
|
|
return f"{scheme}://{key}@{parts.netloc}"
|
|
|
|
|
|
def _pop_offer(key: str) -> _Offer | None:
|
|
offer = _offers.pop(key, None) # single-use
|
|
if offer is None or time.monotonic() - offer.created > OFFER_TTL:
|
|
return None
|
|
return offer
|
|
|
|
|
|
def links_status() -> dict[str, dict]:
|
|
"""Live state per link id for the admin API: connected flag and the
|
|
satellite's latest domain rejections."""
|
|
status: dict[str, dict] = {}
|
|
for conn in _connections.values():
|
|
entry = status.setdefault(conn.link_id, {"connected": True, "rejected": {}})
|
|
entry["rejected"].update(conn.rejected)
|
|
return status
|
|
|
|
|
|
async def close_link(link_id: str) -> None:
|
|
"""Drop a link's live connections (revocation). The satellite treats
|
|
the policy close as terminal and unlinks its side."""
|
|
for conn in list(_connections.values()):
|
|
if conn.link_id == link_id:
|
|
with contextlib.suppress(Exception):
|
|
await conn.ws.close(code=1008)
|
|
|
|
|
|
def domain_infos(rp_ids) -> list[dict]:
|
|
"""The pushed configuration of each in-scope domain: rp-id, display
|
|
name and the origins table verbatim — the upstream is authoritative."""
|
|
infos = []
|
|
for rp_id in rp_ids:
|
|
domain = db.data().config.domains.get(rp_id)
|
|
if domain is None or domain.remote is not None:
|
|
continue # deleted, or itself satellite-served (no nesting)
|
|
infos.append(
|
|
{
|
|
"rp_id": rp_id,
|
|
"rp_name": domain.rp_name or "",
|
|
"origins": msgspec.to_builtins(domain.origins),
|
|
}
|
|
)
|
|
return infos
|
|
|
|
|
|
async def push_link_domains(link_id: str) -> None:
|
|
"""Send the link's current domain set to its live connections and
|
|
update their feed scope (after scope edits or origin changes)."""
|
|
link = db.data().satellite_links.get(link_id)
|
|
if link is None:
|
|
return
|
|
message = {"type": "domains", "domains": domain_infos(link.rp_ids)}
|
|
for conn in list(_connections.values()):
|
|
if conn.link_id != link_id:
|
|
continue
|
|
conn.scope.clear()
|
|
conn.scope.update(link.rp_ids)
|
|
with contextlib.suppress(Exception):
|
|
await _send(conn.ws, message)
|
|
|
|
|
|
async def refresh_links_serving(rp_id: str) -> None:
|
|
"""A domain changed upstream: push its updated info to links serving it."""
|
|
for link_id, link in db.data().satellite_links.items():
|
|
if rp_id in link.rp_ids:
|
|
await push_link_domains(link_id)
|
|
|
|
|
|
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], ctx=ctx
|
|
)
|
|
await push_link_domains(link_id)
|
|
|
|
|
|
async def _send(ws: WebSocket, message: dict) -> None:
|
|
await ws.send_bytes(syncfeed.encode(message))
|
|
|
|
|
|
def _snapshot_items(table: str, scope: set[str]) -> list:
|
|
"""Snapshot rows for a table, filtered to the link's rp-id scope;
|
|
sessions carry their ApiSessionContext as ``ctx``."""
|
|
items = []
|
|
for key, obj in getattr(db.data(), table).items():
|
|
if obj.rp_id not in scope:
|
|
continue
|
|
fields = msgspec.to_builtins(obj)
|
|
if table == "sessions":
|
|
fields["ctx"] = syncfeed.session_ctx_dict(obj)
|
|
items.append([str(key), fields])
|
|
return items
|
|
|
|
|
|
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). Reads are
|
|
bounded: a peer that stalls mid-dialogue gets a plain server-error
|
|
close — never 1008, which the satellite reads as revocation.
|
|
"""
|
|
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
|
|
link_id = hello.get("link") or ""
|
|
server_part: bytes | None = None
|
|
if link_id:
|
|
# Established link: prove the stored key.
|
|
link = db.data().satellite_links.get(link_id)
|
|
try:
|
|
proof = bytes.fromhex(hello.get("key") or "")
|
|
except ValueError:
|
|
proof = b""
|
|
if link is None or not proof or not secrets.compare_digest(link.key, proof):
|
|
await ws.close(code=1008)
|
|
return None
|
|
rp_ids = list(link.rp_ids)
|
|
else:
|
|
# Registration: spend the RAM-only offer, contribute a key part.
|
|
offer = _pop_offer(hello.get("key") or "")
|
|
if offer is None:
|
|
await ws.close(code=1008)
|
|
return None
|
|
link_id = secrets.token_urlsafe(12)
|
|
server_part = secrets.token_bytes(32)
|
|
rp_ids = offer.rp_ids
|
|
welcome: dict = {
|
|
"type": "welcome",
|
|
"link": link_id,
|
|
"domains": domain_infos(rp_ids),
|
|
}
|
|
if server_part is not None:
|
|
welcome["key"] = server_part.hex()
|
|
await _send(ws, welcome)
|
|
|
|
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
|
|
if server_part is not None:
|
|
# Registration completes: the permanent credential is both parts.
|
|
try:
|
|
client_part = bytes.fromhex(ready.get("key") or "")
|
|
except ValueError:
|
|
client_part = b""
|
|
if not client_part:
|
|
await ws.close(code=1008)
|
|
return None
|
|
db.create_satellite_link(
|
|
link_id, rp_ids, server_part + client_part, label=hello.get("label") or ""
|
|
)
|
|
return _Connection(
|
|
ws=ws,
|
|
link_id=link_id,
|
|
scope=set(rp_ids),
|
|
rejected=ready.get("rejected") or {},
|
|
)
|
|
|
|
|
|
async def _apply_client_message(message: dict, conn: _Connection) -> None:
|
|
"""Satellite → upstream frames: session_refresh write-backs (confined
|
|
to the link's scope) and domains_ack reports."""
|
|
mtype = message.get("type")
|
|
if mtype == "domains_ack":
|
|
conn.rejected = message.get("rejected") or {}
|
|
return
|
|
if mtype != "session_refresh":
|
|
return
|
|
key = message.get("key") or ""
|
|
session = db.data().sessions.get(key)
|
|
if session is None or session.rp_id not in conn.scope:
|
|
return
|
|
try:
|
|
validated = msgspec.convert(message.get("validated"), datetime)
|
|
except msgspec.ValidationError:
|
|
return
|
|
db.update_session(
|
|
key,
|
|
ip=message.get("ip") or None,
|
|
user_agent=message.get("user_agent") or None,
|
|
validated=validated,
|
|
)
|
|
|
|
|
|
@app.websocket("/ws")
|
|
async def sync_websocket(ws: WebSocket):
|
|
await ws.accept()
|
|
queue: asyncio.Queue | None = None
|
|
try:
|
|
conn = await _handshake(ws)
|
|
if conn is None:
|
|
return
|
|
queue = syncfeed.subscribe(conn.scope)
|
|
_connections[queue] = conn
|
|
for table in syncfeed.TABLES:
|
|
await _send(
|
|
ws,
|
|
{
|
|
"type": "snapshot",
|
|
"table": table,
|
|
"items": _snapshot_items(table, conn.scope),
|
|
},
|
|
)
|
|
await _send(ws, {"type": "ready", "rp_ids": sorted(conn.scope)})
|
|
|
|
sender = asyncio.create_task(_pump(ws, queue))
|
|
try:
|
|
while True:
|
|
await _apply_client_message(
|
|
msgspec.json.decode(await ws.receive_bytes()), conn
|
|
)
|
|
finally:
|
|
sender.cancel()
|
|
except WebSocketDisconnect, RuntimeError:
|
|
pass
|
|
except Exception:
|
|
_logger.exception("Sync WebSocket failed")
|
|
finally:
|
|
if queue is not None:
|
|
_connections.pop(queue, None)
|
|
syncfeed.unsubscribe(queue)
|
|
|
|
|
|
async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
|
|
try:
|
|
while True:
|
|
await _send(ws, await queue.get())
|
|
except WebSocketDisconnect, RuntimeError, asyncio.CancelledError:
|
|
pass
|
|
except Exception:
|
|
_logger.debug("Sync event pump ended", exc_info=True)
|