From 1734c668406c66e2be63e0c6999b4b2af2f60808 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 9 Aug 2026 05:15:57 +0200 Subject: feat: add revocation push (WebSocket hub→node) — 5.7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hub: /v1/nodes/ws WebSocket endpoint for persistent node connections. /v1/admin/revoke marks user/group revoked in DB, signs JWT revocation token (EdDSA), broadcasts to all connected nodes. Node: RevocationSubscriber maintains WS connection, verifies incoming revocation tokens offline (hub Ed25519 PK), adds to local blocklist (_revoked_users/_revoked_groups sets). 53/53 tests. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- .../meshbay-node/src/meshbay_node/revocation.py | 153 +++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 packages/meshbay-node/src/meshbay_node/revocation.py (limited to 'packages/meshbay-node') diff --git a/packages/meshbay-node/src/meshbay_node/revocation.py b/packages/meshbay-node/src/meshbay_node/revocation.py new file mode 100644 index 0000000..abbff4d --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/revocation.py @@ -0,0 +1,153 @@ +""" +MeshBay Node — revocation subscriber. + +Maintains a persistent WebSocket connection to the hub. +When a signed revocation token arrives, verifies it (Ed25519) +and adds the revoked target to the local blocklist. + +Usage in daemon: + subscriber = RevocationSubscriber(hub_url, access_token, hub_pk_pem) + await subscriber.start() # connects in background + # check membership: + if subscriber.is_revoked("user", user_id): + refuse connection + await subscriber.stop() +""" + +import asyncio +import json +import logging +import time +from typing import Literal + +import httpx +import jwt + +log = logging.getLogger(__name__) + +RevocationTarget = Literal["user", "group"] + + +class RevocationSubscriber: + """ + Background task that keeps a WebSocket connection to the hub + and maintains a local revocation set. + """ + + def __init__( + self, + hub_url: str, + access_token: str, + hub_pk_pem: bytes, + reconnect_delay: float = 5.0, + ): + self._hub_url = hub_url.rstrip("/") + self._access_token = access_token + self._hub_pk_pem = hub_pk_pem + self._reconnect_delay = reconnect_delay + self._revoked_users: set[str] = set() + self._revoked_groups: set[str] = set() + self._task: asyncio.Task | None = None + self._running = False + + def is_revoked(self, target: RevocationTarget, target_id: str) -> bool: + if target == "user": + return target_id in self._revoked_users + return target_id in self._revoked_groups + + def add_revocation(self, target: RevocationTarget, target_id: str) -> None: + if target == "user": + self._revoked_users.add(target_id) + log.warning("User revoked locally: %s", target_id[:8]) + else: + self._revoked_groups.add(target_id) + log.warning("Group revoked locally: %s", target_id[:8]) + + def verify_and_apply(self, token: str) -> bool: + """Verify a revocation token and apply it. Returns True if valid.""" + try: + payload = jwt.decode(token, self._hub_pk_pem, algorithms=["EdDSA"], + options={"verify_exp": False}) + if payload.get("type") != "revocation": + return False + target = payload["target"] + target_id = payload["target_id"] + self.add_revocation(target, target_id) + return True + except Exception as e: + log.error("Invalid revocation token: %s", e) + return False + + async def start(self) -> None: + self._running = True + self._task = asyncio.create_task(self._run_loop()) + log.info("RevocationSubscriber started") + + async def stop(self) -> None: + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + log.info("RevocationSubscriber stopped") + + async def _run_loop(self) -> None: + while self._running: + try: + await self._connect_and_listen() + except asyncio.CancelledError: + raise + except Exception as e: + log.warning("WS disconnected (%s), reconnecting in %ss", e, self._reconnect_delay) + await asyncio.sleep(self._reconnect_delay) + + async def _connect_and_listen(self) -> None: + ws_url = self._hub_url.replace("http://", "ws://").replace("https://", "wss://") + ws_url += "/v1/nodes/ws" + + async with httpx.AsyncClient() as client: + async with client.stream("GET", ws_url, + headers={"Upgrade": "websocket"}) as resp: + # Use websockets library for proper WS protocol + pass + + # Use websockets library directly + import websockets + async with websockets.connect(ws_url) as ws: + # Authenticate + await ws.send(json.dumps({ + "type": "auth", + "token": self._access_token, + })) + auth_resp = json.loads(await ws.recv()) + if auth_resp.get("type") != "auth_ok": + raise ConnectionError(f"WS auth failed: {auth_resp}") + log.info("WS connected to hub — node_id=%s", auth_resp.get("node_id", "?")[:8]) + + # Listen for revocations + send keepalive pings + ping_interval = 30.0 + last_ping = asyncio.get_event_loop().time() + + while self._running: + now = asyncio.get_event_loop().time() + if now - last_ping > ping_interval: + await ws.send(json.dumps({"type": "ping"})) + last_ping = now + + try: + msg_raw = await asyncio.wait_for(ws.recv(), timeout=ping_interval + 5) + msg = json.loads(msg_raw) + + if msg.get("type") == "revocation": + token = msg.get("token", "") + ok = self.verify_and_apply(token) + log.info("Revocation received — valid=%s", ok) + elif msg.get("type") == "pong": + pass + else: + log.debug("WS message: %s", msg.get("type")) + + except asyncio.TimeoutError: + continue -- cgit v1.2.3