diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 32 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/hub_client.py | 19 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops.py | 94 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roots.py | 70 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 159 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ui/app.py | 38 |
6 files changed, 247 insertions, 165 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 7e3ebc1..385a1da 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -125,6 +125,7 @@ class NodeDaemon: self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None + self._reload_lock = asyncio.Lock() async def run(self) -> None: log.info("MeshBay Node starting up") @@ -225,14 +226,13 @@ class NodeDaemon: ", ".join(str(r.path) for r in roots)) gek = None - if group_cfg.visibility == "private": - gek = await self._load_gek( - group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) - if gek: - log.info("GEK loaded for group %s", group_cfg.id[:8]) - else: - log.info("No GEK yet for group %s — will accept first setup", - group_cfg.name) + gek = await self._load_gek( + group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) + if gek: + log.info("GEK loaded for group %s", group_cfg.id[:8]) + else: + log.info("No GEK yet for group %s — will accept first setup", + group_cfg.name) indexer = DirectoryIndexer( roots=roots, @@ -399,7 +399,7 @@ class NodeDaemon: on_incoming=on_incoming, on_revocation=on_revocation, on_webrtc_offer=on_webrtc_offer, - group_ids=list(groups_ctx.keys()), + group_ids=lambda: list((self._state.get("groups_ctx") or {}).keys()), )) self._tasks.append(ws_task) log.info("Hub WS task started") @@ -470,7 +470,15 @@ class NodeDaemon: Handles root changes on existing groups, hot-loads new groups, and tears down removed groups. Existing connections are untouched: a member watching a film keeps watching it. + + Serialised by _reload_lock: fire-and-forget reloads from config-mutating + endpoints can overlap with the wizard's explicit /api/reload call, + and two concurrent hot-loads of the same group corrupt the runtime state. """ + async with self._reload_lock: + await self._reload_config_inner() + + async def _reload_config_inner(self) -> None: log.info("Reloading config from %s", self._config_path) try: fresh = load_config(self._config_path) @@ -537,7 +545,7 @@ class NodeDaemon: roots.refresh_availability() gek = None - if group_cfg.visibility == "private" and sk_x_raw and pk_x_raw: + if sk_x_raw and pk_x_raw: gek = await self._load_gek( group_cfg.id, node_user_id, sk_x_raw, pk_x_raw) if gek: @@ -616,6 +624,10 @@ class NodeDaemon: log.info("Reload complete — %d re-rooted, %d added, %d removed", changed, len(added_names), len(removed_names)) + if (added_names or removed_names) and self._hub: + gids = list((self._state.get("groups_ctx") or {}).keys()) + await self._hub.update_ws_groups(gids) + async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" import httpx as _httpx diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index b345187..92c39db 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -229,12 +229,24 @@ class HubClient: except Exception: pass + async def update_ws_groups(self, group_ids: list[str]) -> None: + """Tell the hub about changed group list without dropping the connection.""" + ws = self._ws + if ws: + try: + await ws.send(json.dumps({ + "type": "update_groups", + "group_ids": group_ids, + })) + except Exception: + pass + async def maintain_ws( self, on_incoming: Any = None, on_revocation: Any = None, on_webrtc_offer: Any = None, - group_ids: list[str] | None = None, + group_ids: list[str] | None = None, # static list or callable returning one ) -> None: """ Maintain a persistent WebSocket connection to the hub. @@ -267,8 +279,9 @@ class HubClient: "token": self._session.access_token, "node_id": self._session.node_id, } - if group_ids: - auth_msg["group_ids"] = group_ids + gids = group_ids() if callable(group_ids) else group_ids + if gids: + auth_msg["group_ids"] = gids await ws.send(json.dumps(auth_msg)) # Bounded: a hub that accepts the socket and then says nothing # — which is what it does for a few seconds while restarting — diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 2a76290..daddf17 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -157,38 +157,44 @@ async def pair_operator(state: dict) -> dict: return {"code": code, "expires_at": expires, "user_id": user_id} -async def create_invite(state: dict, group_id: str, username: str) -> dict: +async def create_invite(state: dict, group_id: str, username: str, *, + user_id: str = "", + created_by: str = "local-cli") -> dict: """ Issue an invitation code. The hub is asked for the account id and nothing else — never for a key. A hub that answered with the wrong account would produce an invite whose code it never learns, since the code goes to a human out of band. + + When ``user_id`` is supplied directly (MNP path), the hub lookup is skipped. """ roster = _roster(state) _group_ctx(state, group_id) - hub = _hub(state) - try: - account = await hub.get_user_pubkeys(username) - except Exception as e: - raise OpError(f"Unknown user {username!r}: {e}", status=404) from e + if not user_id: + hub = _hub(state) + try: + account = await hub.get_user_pubkeys(username) + except Exception as e: + raise OpError(f"Unknown user {username!r}: {e}", status=404) from e + user_id = account["user_id"] config = state.get("config") ttl = (config.node.invite_ttl_hours if config else 168) * 3600 code = await roster.create_invite( group_id=group_id, - user_id=account["user_id"], + user_id=user_id, role=ROLE_MEMBER, - created_by="local-cli", + created_by=created_by, ttl=ttl, username=username, ) invites = await roster.list_invites() expires = next((i["expires_at"] for i in invites - if i["user_id"] == account["user_id"] + if i["user_id"] == user_id and i["group_id"] == group_id), "") return {"code": code, "expires_at": expires, - "username": username, "user_id": account["user_id"]} + "username": username, "user_id": user_id} async def revoke_member(state: dict, user_id: str, group_id: str) -> dict: @@ -236,6 +242,10 @@ async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict: ctx = _group_ctx(state, group_id) hub = _hub(state) + if rotate and ctx.get("visibility") == "public": + raise OpError( + "Key rotation is not available for public groups", status=400) + bundle_store = state.get("bundle_store") if not bundle_store: raise OpError("Bundle store not available", status=503) @@ -361,20 +371,25 @@ async def attach_group(state: dict, name: str, shared_dir: str, # Appended as text rather than re-serialised: node.toml is hand-written and # full of comments explaining decisions, and a round trip through a TOML # writer would throw all of that away. + join_policy = group.get("join_policy", "invite") block = (f'\n[[groups]]\n' f'id = "{group["id"]}"\n' f'name = "{group["name"]}"\n' - f'visibility = "{group.get("visibility", "private")}"\n') + f'visibility = "{group.get("visibility", "private")}"\n' + f'join_policy = "{join_policy}"\n') + separate_upload = False if upload_dir: - upload_path = Path(upload_dir).expanduser() - try: - upload_path.mkdir(parents=True, exist_ok=True) - except OSError as e: - raise OpError(f"Cannot create {upload_path}: {e}") from e - block += f'upload_dir = "{upload_path}"\n' + upload_path = Path(upload_dir).expanduser().resolve() + if upload_path != path.resolve(): + separate_upload = True + try: + upload_path.mkdir(parents=True, exist_ok=True) + except OSError as e: + raise OpError(f"Cannot create {upload_path}: {e}") from e + block += f'upload_dir = "{upload_path}"\n' block += (f'\n [[groups.roots]]\n' f' path = "{path}"\n') - if not upload_dir: + if not separate_upload: block += f' upload = true\n' try: with conf_path.open("a") as f: @@ -385,7 +400,7 @@ async def attach_group(state: dict, name: str, shared_dir: str, result = {"group_id": group["id"], "name": group["name"], "shared_dir": str(path), "config": str(conf_path), "note": "restart the node to pick it up"} - if upload_dir: + if separate_upload: result["upload_dir"] = str(upload_path) return result @@ -552,9 +567,6 @@ async def add_root(state: dict, group_id: str, path: str, *, cfg.roots.append(RootSpec( path=str(added.path), name=added.name, kind=added.kind, upload=added.upload, direct=added.direct)) - groups_ctx = state.get("groups_ctx", {}) - if group_id in groups_ctx: - groups_ctx[group_id]["roots"] = built log.info("Root added: %s → group %s", added.name, group_id[:8]) return {"status": "added", "name": added.name, "path": str(added.path), @@ -602,15 +614,10 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict: built = RootSet.build(remaining) except RootError: built = None - if built is not None: - groups_ctx = state.get("groups_ctx", {}) - if group_id in groups_ctx: - groups_ctx[group_id]["roots"] = built log.info("Root removed: %s from group %s", root_name, group_id[:8]) return {"status": "removed", "name": root_name, "group_id": group_id, - "roots": built.describe() if built else [], - "note": "restart recommended to update the file index"} + "roots": built.describe() if built else []} # ── Files ──────────────────────────────────────────────────────────────────── @@ -682,3 +689,34 @@ async def clear_denylist(state: dict, *, subject: str = "") -> dict: log.warning("Denylist cleared (%s): %d entr(y/ies) removed", subject or "all", removed) return {"status": "cleared", "removed": removed, "subject": subject or "all"} + + +# ── Upload policy ─────────────────────────────────────────────────────────── + +async def set_member_upload(state: dict, group_id: str, allowed: bool) -> dict: + """ + Turn uploading by ordinary members on or off. + + The setting lives on the node (roster.db), not on the hub and not in + node.toml — changing it must not rewrite the operator's config file, + and must not need a restart. + """ + roster = _roster(state) + ctx = _group_ctx(state, group_id) + await roster.set_member_upload(group_id, allowed, + set_by=state.get("node_user_id", "")) + ctx["member_upload"] = allowed + log.info("Upload policy: %s for group %s", "on" if allowed else "off", + group_id[:8]) + return {"allowed": allowed, "group_id": group_id} + + +# ── Reload ────────────────────────────────────────────────────────────────── + +async def reload_config(state: dict) -> dict: + """Hot-reload node.toml without dropping connections.""" + reload_fn = state.get("reload_fn") + if not reload_fn: + raise OpError("Reload not available", status=503) + await reload_fn() + return {"status": "reloaded"} diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index bc27bf7..74ea2f6 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -29,6 +29,7 @@ are one directory. from __future__ import annotations import logging +import re from dataclasses import dataclass, field from pathlib import Path @@ -38,6 +39,75 @@ log = logging.getLogger(__name__) VALID_KINDS = ("generic", "video", "audio", "photo") +# Filenames and subdirectory names sent by clients. A leading dot is a hidden +# file on every platform, a leading hyphen confuses CLI tools, a leading space +# cannot start one, and a trailing space or dot is refused because it makes two +# different files look identical in a list. +SAFE_UPLOAD_NAME = re.compile( + r"^[^\W_]" # letter or digit — never ‘.’, ‘-’ or space + r"[\w .\-()\[\]'\u2019,&+#@]{0,127}" # body: word chars plus mild punctuation + r"(?<![ .])$", # and never ending on a space or a dot + re.UNICODE) + + +def _free_name(directory: Path, filename: str) -> str: + """ + `filename`, or the first "name (n).ext" that is not taken. + + Never returns the name of a file that exists, so an upload cannot replace + one — the property the per-user quarantine used to provide (C5a). + """ + if not (directory / filename).exists(): + return filename + stem, dot, ext = filename.rpartition(".") + if not dot: + stem, ext = filename, "" + for n in range(2, 1000): + candidate = f"{stem} ({n}){dot}{ext}" + if not (directory / candidate).exists(): + return candidate + raise FileExistsError(filename) + + +def safe_subdir(roots: "RootSet", rel: str) -> Path | None: + """ + Resolve a client-supplied directory inside one of the group's roots, or refuse. + + The path arrives from the wire, so every part is checked: the first segment + must name a root that is readable right now, each later segment against the + same allowlist as filenames, and the resolved result against that root's + directory. `..`, absolute paths, symlinks pointing out, and anything with a + separator in a segment are all refused here rather than in the caller, so + there is one place to get it right. + + The virtual root itself — `""` — is deliberately **not** resolvable. It is + not a directory on anyone's disk: a file cannot be written there and a + directory cannot be created there, because it belongs to no volume. Callers + that used to receive the shared root for an empty path now receive None, + which is the honest answer. + + The quarantine was the fix for C5a; what actually mattered in it — no + overwrite, a name allowlist, and confinement — is kept by this plus the + caller's existing checks. + """ + found = roots.split(rel or "") + if found is None: + return None + root, tail = found + if not root.available: + return None + parts = [seg for seg in tail.split("/") if seg not in ("", ".")] + if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts): + return None + try: + target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() + base = root.path.resolve() + except OSError: + return None + if target != base and base not in target.parents: + return None + return target + class RootError(ValueError): """A root set that cannot be built. The message is shown to the operator.""" diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 1f1f2d2..947d1f9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -28,7 +28,6 @@ import hashlib import hmac import logging import os -import re import struct import time from pathlib import Path @@ -86,8 +85,9 @@ from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex from meshbay_node import ops -from meshbay_node.roots import RootSet, entry_abs_path -from meshbay_node.roster import DEFAULT_INVITE_TTL +from meshbay_node.roots import ( + RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, +) log = logging.getLogger(__name__) @@ -135,81 +135,6 @@ JOIN_FAILURE_WINDOW = 600 # seconds # into, back up or empty — rather than a hidden tree of per-user uuids that # nobody could read, or files scattered wherever someone happened to be looking. UPLOAD_DIR_NAME = "uploads" -# Conservative allowlist: also what keeps markup out of filenames, which the node admin -# UI used to render unescaped (finding H2). -# An allowlist, still — C5a and H2 depend on it — but one that does not assume -# the world writes in ASCII. `été.txt` and `rapport (1).pdf` were refused, and -# the second of those is a name _free_name generates itself, so the node was -# rejecting files it had named. `\w` is Unicode here, which admits letters and -# digits of any script while `<`, `>`, `"`, `;`, `/`, `\` and control characters -# stay out. The first character must be a letter or digit, so ".." and dotfiles -# cannot start one, and a trailing space or dot is refused because it makes two -# different files look identical in a list. -SAFE_UPLOAD_NAME = re.compile( - r"^[^\W_]" # letter or digit — never '.', '-' or space - r"[\w .\-()\[\]'\u2019,&+#@]{0,127}" # body: word chars plus mild punctuation - r"(?<![ .])$", # and never ending on a space or a dot - re.UNICODE) - - -def _free_name(directory: Path, filename: str) -> str: - """ - `filename`, or the first "name (n).ext" that is not taken. - - Never returns the name of a file that exists, so an upload cannot replace - one — the property the per-user quarantine used to provide (C5a). - """ - if not (directory / filename).exists(): - return filename - stem, dot, ext = filename.rpartition(".") - if not dot: - stem, ext = filename, "" - for n in range(2, 1000): - candidate = f"{stem} ({n}){dot}{ext}" - if not (directory / candidate).exists(): - return candidate - raise FileExistsError(filename) - - -def safe_subdir(roots: RootSet, rel: str) -> Path | None: - """ - Resolve a client-supplied directory inside one of the group's roots, or refuse. - - The path arrives from the wire, so every part is checked: the first segment - must name a root that is readable right now, each later segment against the - same allowlist as filenames, and the resolved result against that root's - directory. `..`, absolute paths, symlinks pointing out, and anything with a - separator in a segment are all refused here rather than in the caller, so - there is one place to get it right. - - The virtual root itself — `""` — is deliberately **not** resolvable. It is - not a directory on anyone's disk: a file cannot be written there and a - directory cannot be created there, because it belongs to no volume. Callers - that used to receive the shared root for an empty path now receive None, - which is the honest answer. - - The quarantine was the fix for C5a; what actually mattered in it — no - overwrite, a name allowlist, and confinement — is kept by this plus the - caller's existing checks. - """ - found = roots.split(rel or "") - if found is None: - return None - root, tail = found - if not root.available: - return None - parts = [seg for seg in tail.split("/") if seg not in ("", ".")] - if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts): - return None - try: - target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() - base = root.path.resolve() - except OSError: - return None - if target != base and base not in target.parents: - return None - return target - def _extract_dtls_fingerprint(sdp: str) -> bytes: @@ -985,6 +910,12 @@ class WebRTCPeerSession: # the client is told it has no role on a node it administers. member = (await roster.get_member(group_id, user_id) or await roster.get_member("", user_id)) + if not member and self._group_join_policy(session_group) == "open": + await roster.set_member( + group_id=session_group, user_id=user_id, role=ROLE_MEMBER, + status="active", approved_by="open-join", + ) + member = await roster.get_member(session_group, user_id) await self._join_ok( user_id, pk_x_raw, session_group, role=member["role"] if member else "", @@ -1635,16 +1566,12 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"member_upload:{pending['subject']}") return - roster = self._ctx.get("roster") - if roster is None: - self._send({"type": "error", "detail": "No roster on this node"}) + try: + await self._run_op( + ops.set_member_upload, self._group_id or "", allowed) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) return - await roster.set_member_upload(self._group_id or "", allowed, - set_by=self._user_id) - # Stored *and* applied. The upload path is synchronous and reads this - # dict; leaving it to the next restart would make the panel say one - # thing while the node did another. - self._group_ctx()["member_upload"] = allowed self._audit("member_upload", pending["subject"]) # Everyone already connected is told, rather than finding out by having @@ -1928,14 +1855,11 @@ class WebRTCPeerSession: self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}") return - roster = self._ctx.get("roster") - if roster is None: - self._send({"type": "error", "detail": "Roster not available"}) - return - - group_id = self._group_id or "" - if not await roster.set_status(group_id, user_id, "revoked"): - self._send({"type": "error", "detail": "Not a member of this group"}) + try: + result = await self._run_op( + ops.revoke_member, user_id, self._group_id or "") + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) return # Anyone connected right now keeps the key they already unwrapped; what @@ -1948,14 +1872,11 @@ class WebRTCPeerSession: except Exception: pass - log.info("Member revoked by %s: user=%s group=%s", - self._user_id[:8], user_id[:8], group_id[:8] or "-") self._audit("member_revoke", user_id) self._send({ "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION, "user_id": user_id, - "reminder": "they still hold the current group key — rotate it with " - "meshbay-node gek-init", + "reminder": result.get("reminder", ""), }) async def _do_keypair_bundle_delete(self) -> None: @@ -2694,37 +2615,27 @@ class WebRTCPeerSession: self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") return - roster = self._ctx.get("roster") - if roster is None: - self._send({"type": "error", "detail": "Roster not available"}) - return - payload = pending["payload"] - code = await roster.create_invite( - group_id=payload["group_id"], - user_id=payload["user_id"], - role=ROLE_MEMBER, - created_by=self._user_id or "", - ttl=self._ctx.get("invite_ttl", DEFAULT_INVITE_TTL), - username=payload.get("username", ""), - ) - invites = await roster.list_invites() - expires = next( - (i["expires_at"] for i in invites - if i["user_id"] == payload["user_id"] - and i["group_id"] == payload["group_id"]), "") + try: + result = await self._run_op( + ops.create_invite, + payload["group_id"], + payload.get("username", ""), + user_id=payload["user_id"], + created_by=self._user_id or "", + ) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return - log.info("Invite created: group=%s user=%s", - payload["group_id"][:8], payload["user_id"][:8]) self._audit("invite_create", f"target={payload['user_id'][:8]}") - # The code exists in the clear exactly here and in the operator's hands. self._send({ "type": MNP.INVITE_RESULT, "v": MNP_VERSION, - "code": code, - "expires_at": expires, - "user_id": payload["user_id"], - "username": payload.get("username", ""), + "code": result["code"], + "expires_at": result["expires_at"], + "user_id": result["user_id"], + "username": result.get("username", ""), }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index a068a8d..b505f25 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -295,6 +295,44 @@ def create_ui_app(state: dict) -> FastAPI: async def init_gek(group_id: str, rotate: bool = False): return await _op(lambda: ops.set_gek(state, group_id, rotate=rotate)) + # ── Roots management (operator only, localhost) ──────────────────────── + + @app.post("/api/groups/{group_id}/roots") + async def add_root(group_id: str, payload: dict): + result = await _op(lambda: ops.add_root( + state, group_id, + (payload.get("path") or "").strip(), + name=(payload.get("name") or "").strip(), + kind=(payload.get("kind") or "generic").strip(), + upload=bool(payload.get("upload", False)), + )) + reload_fn = state.get("reload_fn") + if reload_fn: + asyncio.ensure_future(reload_fn()) + return result + + @app.delete("/api/groups/{group_id}/roots/{root_name}") + async def remove_root(group_id: str, root_name: str): + result = await _op(lambda: ops.remove_root(state, group_id, root_name)) + reload_fn = state.get("reload_fn") + if reload_fn: + asyncio.ensure_future(reload_fn()) + return result + + # ── Upload toggle (operator only, localhost) ───────────────────────── + + @app.put("/api/groups/{group_id}/member-upload") + async def set_member_upload(group_id: str, payload: dict): + return await _op(lambda: ops.set_member_upload( + state, group_id, bool(payload.get("allowed", False)), + )) + + # ── Reload config ──────────────────────────────────────────────────── + + @app.post("/api/reload") + async def reload_config(): + return await _op(lambda: ops.reload_config(state)) + # ── Chat endpoints ─────────────────────────────────────────────────────── _chat_subscribers: list[WebSocket] = [] |