""" MeshBay Node — local control API (loopback, default port 18000). A JSON-only FastAPI app: node status, groups and roots, roster and denylist, node settings, connected peers, and the audit log. It is the single control plane for the node — the `meshbay-node` CLI and the desktop client's Node page are both clients of it. (Chat is served to browsers over MNP/WebRTC, not here.) Served only on 127.0.0.1 — never network-exposed — and every request is gated by a per-run session token (11.5.3) written to `/ui-token`. There is no server-rendered UI: the Node page ships in the desktop client (see `docs/refactor-node-ui.md`). """ import logging from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse from meshbay_common.background import spawn from meshbay_node import __version__, ops from meshbay_node.indexer.indexer import DirectoryIndexer log = logging.getLogger(__name__) def _op(coro): """ Run an operation and translate its refusal into a JSON response. The operations live in `meshbay_node.ops` and know nothing about HTTP. This is the whole of the HTTP adapter: without it each handler would carry its own status codes, and the MNP handler in Stage B3 would carry a second set that slowly stopped agreeing. """ async def run(): try: return await coro() except ops.OpError as e: return JSONResponse(e.as_dict(), e.status) return run() async def _display_names(state: dict) -> tuple[dict[str, str], dict[str, str]]: """(user_id -> username, group_id -> name) for rendering ids a human reads. Usernames come from the roster (the node's own record); group names from node.toml. Both are best-effort — a missing entry just leaves the caller with the raw id to shorten. """ users: dict[str, str] = {} roster = state.get("roster") if roster: try: for ident in await roster.list_identities(): if ident.get("username"): users[ident["user_id"]] = ident["username"] except Exception: pass groups: dict[str, str] = {} config = state.get("config") if config: for g in config.groups: if getattr(g, "id", None): groups[g.id] = g.name return users, groups def create_ui_app(state: dict) -> FastAPI: app = FastAPI( title="MeshBay Node Admin", version=__version__, docs_url=None, redoc_url=None, ) @app.middleware("http") async def _require_session_token(request, call_next): """ Gate the control API behind a per-run token (11.5.3). "localhost only" is weaker than it sounds: any process on the machine can reach it, and a page in the operator's browser can reach it too via DNS rebinding. Since this API can re-initialise a group's GEK and read the audit log, an unauthenticated loopback service is a privilege boundary waiting to be crossed. The token is written to `/ui-token` at startup and accepted as ?t= or the X-MeshBay-Token header. """ from fastapi.responses import PlainTextResponse token = state.get("ui_token") if token: supplied = (request.query_params.get("t") or request.headers.get("X-MeshBay-Token")) if supplied != token: return PlainTextResponse("Forbidden", status_code=403) return await call_next(request) @app.middleware("http") async def _security_headers(request, call_next): """ Belt-and-braces for a loopback API that returns only JSON. Since the server no longer renders any HTML (the dashboard was removed 2026-09-01), the response has nothing an injected script could live in — but a DNS-rebound page or a content-sniffing client that manages to treat a body as a document still gets `default-src 'none'`, which forbids every fetch, script, style and frame. `nosniff` stops the sniffing in the first place. """ response = await call_next(request) response.headers["Content-Security-Policy"] = ( "default-src 'none'; frame-ancestors 'none'; base-uri 'none'" ) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "no-referrer" return response # ── JSON API ───────────────────────────────────────────────────────────── @app.get("/api/status") async def api_status(): indexes = state.get("indexes", {}) total_files = sum(idx.count for idx in indexes.values()) groups_ctx = state.get("groups_ctx", {}) webrtc = state.get("webrtc") status = state.get("status", "starting") needs = [] if status == "waiting_for_node_key": needs.append("node_key_link") if status == "running" and not groups_ctx: needs.append("group_add") if status == "running": roster = state.get("roster") if roster: members = await roster.list_members() operators = [m for m in members if m["role"] == "operator" and m["status"] == "active"] if not operators: needs.append("operator_pair") for gid, gctx in groups_ctx.items(): if not gctx.get("gek"): name = gctx.get("name", gid[:8]) needs.append(f"gek_init:{name}") return { "version": __version__, "status": status, "needs": needs, "hub_url": state.get("hub_url", ""), "username": state.get("username", ""), "quic_port": state.get("quic_port", 0), "endpoint_hint": state.get("endpoint_hint"), "group_count": len(groups_ctx), "total_files": total_files, "webrtc_peers": webrtc.active_peers if webrtc else 0, "pk_node_ed25519": state.get("pk_node_ed25519", ""), } @app.delete("/api/unlink") async def api_unlink(): hub = state.get("hub") if not hub: raise HTTPException(status_code=503, detail="Hub not connected") await hub.unlink_node_key() return {"status": "unlinked"} @app.get("/api/groups") async def api_groups(): return await _op(lambda: ops.list_groups(state)) @app.post("/api/groups/attach") async def attach_group(payload: dict): result = await _op(lambda: ops.attach_group( state, (payload.get("name") or "").strip(), (payload.get("shared_dir") or "").strip(), writable=bool(payload.get("writable", True)), )) reload_fn = state.get("reload_fn") if reload_fn: spawn(reload_fn()) return result @app.post("/api/groups/detach") async def detach_group(payload: dict): result = await _op(lambda: ops.detach_group( state, (payload.get("name") or payload.get("group_id") or "").strip(), )) reload_fn = state.get("reload_fn") if reload_fn: spawn(reload_fn()) return result @app.delete("/api/groups/{group_id}/files/{file_id}") async def delete_file(group_id: str, file_id: str): """Milestone 14.11 — the last operator action that needed a browser.""" return await _op(lambda: ops.delete_file(state, group_id, file_id)) @app.get("/api/denylist") async def api_denylist(): return await _op(lambda: ops.read_denylist(state)) @app.post("/api/denylist/clear") async def api_denylist_clear(subject: str = ""): return await _op(lambda: ops.clear_denylist(state, subject=subject)) @app.get("/api/index-cache") async def api_index_cache_stats(): return await _op(lambda: ops.index_cache_stats(state)) @app.post("/api/index-cache/prune") async def api_index_cache_prune(): return await _op(lambda: ops.prune_index_cache(state)) @app.post("/api/groups/{group_id}/video/rematch") async def api_video_rematch(group_id: str): return await _op(lambda: ops.rematch_video(state, group_id)) @app.get("/api/groups/{group_id}/files") async def api_group_files(group_id: str): groups_ctx = state.get("groups_ctx", {}) ctx = groups_ctx.get(group_id) if not ctx: return {"files": []} idx = ctx.get("index") if not idx: return {"files": []} return { "files": [ { "id": e.id, "name": e.name, "path": e.path, "size": e.size, "type": e.type, "added_at": e.added_at, } for e in idx.entries ] } @app.get("/api/peers") async def api_peers(): webrtc = state.get("webrtc") if not webrtc: return {"peers": []} users, groups = await _display_names(state) peers = [] for pid, session in list(webrtc._sessions.items()): from meshbay_node.transport.webrtc_server import _get_remote_ip uid = session._user_id or "" gid = session._group_id or "" peers.append({ "peer_id": pid, "user_id": uid, "username": session._username or users.get(uid, ""), "group_id": gid, "group_name": groups.get(gid, ""), "remote_ip": session._remote_ip or _get_remote_ip(session._pc), "state": session._pc.connectionState, }) return {"peers": peers} @app.get("/api/audit") async def api_audit( since: float = 0, limit: int = 200, offset: int = 0, user_id: str | None = Query(default=None), event: str | None = Query(default=None), ): audit = state.get("audit_store") if not audit: return {"entries": [], "offset": 0, "limit": limit, "has_more": False} limit = max(1, min(limit, 1000)) offset = max(0, offset) # Fetch one extra row to know whether a next page exists without a count. rows = await audit.get_entries( since=since, limit=limit + 1, offset=offset, user_id=user_id, event=event) has_more = len(rows) > limit entries = rows[:limit] # Legacy rows and pre-handshake events store user_id only; group_id is # never a name. Resolve both for display — no migration, the roster and # node.toml are the node's own records. names, groups = await _display_names(state) return { "offset": offset, "limit": limit, "has_more": has_more, "entries": [ { "id": e.id, "timestamp": e.timestamp, "user_id": e.user_id, "username": e.username or names.get(e.user_id, ""), "ip": e.ip, "event": e.event, "group_id": e.group_id, "group_name": groups.get(e.group_id, ""), "detail": e.detail, } for e in entries ], } # ── Operator pairing (localhost only) ────────────────────────────────── @app.post("/api/operator/pair") async def operator_pair(): return await _op(lambda: ops.pair_operator(state)) @app.get("/api/roster") async def api_roster(group_id: str = ""): return await _op(lambda: ops.read_roster(state, group_id)) @app.post("/api/groups/{group_id}/invites") async def create_invite(group_id: str, username: str): return await _op(lambda: ops.create_invite(state, group_id, username)) @app.get("/api/resolve") async def resolve_user(username: str): return await _op(lambda: ops.resolve_user(state, username)) @app.post("/api/members/{user_id}/revoke") async def revoke_member(user_id: str, group_id: str): return await _op(lambda: ops.revoke_member(state, user_id, group_id)) @app.post("/api/members/{user_id}/unpin") async def unpin_member(user_id: str): return await _op(lambda: ops.unpin_member(state, user_id)) # ── Chat encryption (operator only, localhost) ───────────────────────── @app.get("/api/groups/{group_id}/chat") async def chat_status(group_id: str): return await _op(lambda: ops.chat_status(state, group_id)) @app.post("/api/groups/{group_id}/chat/epoch") async def rotate_chat_epoch(group_id: str): return await _op(lambda: ops.open_chat_epoch(state, group_id)) @app.post("/api/groups/{group_id}/chat/encrypt-history") async def encrypt_chat_history(group_id: str): return await _op(lambda: ops.encrypt_chat_history(state, group_id)) @app.post("/api/groups/{group_id}/chat/prune") async def prune_chat(group_id: str, max_age_days: int): return await _op(lambda: ops.prune_chat(state, group_id, max_age_days)) # ── GEK initialization (operator only, localhost) ────────────────────── @app.post("/api/groups/{group_id}/gek") 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(), writable=bool(payload.get("writable", payload.get("upload", False))), removable=bool(payload.get("removable", False)), )) reload_fn = state.get("reload_fn") if reload_fn: spawn(reload_fn()) return result @app.patch("/api/groups/{group_id}/roots/{root_name}") async def update_root(group_id: str, root_name: str, payload: dict): result = await _op(lambda: ops.update_root( state, group_id, root_name, writable=payload.get("writable"), removable=payload.get("removable"), )) reload_fn = state.get("reload_fn") if reload_fn: spawn(reload_fn()) return result @app.put("/api/groups/{group_id}/roots/{root_name}/eject") async def eject_root(group_id: str, root_name: str): return await _op(lambda: ops.eject_root(state, group_id, root_name)) @app.put("/api/groups/{group_id}/roots/{root_name}/plug") async def plug_root(group_id: str, root_name: str): return await _op(lambda: ops.plug_root(state, group_id, root_name)) @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: spawn(reload_fn()) return result # `spawn` above schedules the reload (and whatever initial scan it # triggers) on the daemon's own event loop — it has no link to # this HTTP request or to any browser tab. Closing the client that made # this call does not cancel it: the scan is the node's own background # work, not something borrowed from the request that started it. @app.get("/api/groups/{group_id}/index-status") async def index_status(group_id: str): """ Polled by the Create Group wizard and by "add a directory" in Settings — the same source either way, since both just start a scan on this group's indexer. `current_dir` is a basename only, and is never sent over MNP (see IndexProgress in indexer.py) — this route is loopback-only, for the operator's own screen. Reads state["indexers"] rather than groups_ctx: a brand-new group is registered there before its (possibly long) initial scan runs, but is only added to groups_ctx once that scan finishes (it is not yet authorized for member connections either way — see _reload_config) — this is precisely the window the wizard needs to watch. """ indexer = state.get("indexers", {}).get(group_id) progress = indexer.progress if indexer else None if progress is None: return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, "current_dir": ""} return { "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "current_dir": progress.current_dir, } @app.get("/api/index-status") async def index_status_all(): """ Every group's indexing at once, for the client's progress band — which is on screen whatever page the operator is on, so it cannot ask per group. Names roots, like `current_dir` above: loopback only, the operator's own screen. Reads state["indexers"] for the same reason. """ config = state.get("config") names = {g.id: g.name for g in config.groups} if config else {} groups = [] for gid, indexer in list(state.get("indexers", {}).items()): p = indexer.progress groups.append({ "group_id": gid, "group_name": names.get(gid, gid[:8]), "scanning": p.scanning, "kind": p.kind, "root": p.root, "current_dir": p.current_dir, "scanned_bytes": p.scanned_bytes, "total_bytes": p.total_bytes, "files_done": p.files_done, "files_total": p.files_total, "queued": list(p.queued), }) return {"groups": groups} # ── Enabled apps (operator only, localhost) ──────────────────────────── # # Same loopback shape as member-upload: the Create Group wizard sets this # once, right after creating the group and before the (potentially long) # initial scan, so an operator narrowing this down to just Files+Videos # never briefly has Chat live for other members to notice. @app.put("/api/groups/{group_id}/apps") async def set_enabled_apps(group_id: str, payload: dict): apps = payload.get("apps") if not isinstance(apps, list) or not apps: raise HTTPException(400, "apps must be a non-empty list") return await _op(lambda: ops.set_enabled_apps(state, group_id, apps)) # ── App directories (operator only, localhost) ──────────────────────── # # The loopback twin of the `app_directories` MNP op. One endpoint for every # application, keyed by the app's own name, so adding one needs no route # here — the same reason the op is generic. `ALLOWED_APPS` is checked on # the MNP path; here the caller is already on localhost holding the run # token, and `ops` refuses a directory outside the group's roots either # way, so an unknown key writes one unread settings row and nothing else. @app.put("/api/groups/{group_id}/app-directories/{app_key}") async def set_app_directories(group_id: str, app_key: str, payload: dict): dirs = payload.get("directories") if not isinstance(dirs, list): raise HTTPException(400, "directories must be a list") return await _op(lambda: ops.set_app_directories( state, group_id, app_key, [str(d) for d in dirs])) @app.put("/api/groups/{group_id}/chat-directory") async def set_chat_directory(group_id: str, payload: dict): return await _op(lambda: ops.set_chat_directory( state, group_id, str(payload.get("path") or ""))) @app.put("/api/groups/{group_id}/chat-link-preview") async def set_chat_link_preview(group_id: str, payload: dict): return await _op(lambda: ops.set_chat_link_preview( state, group_id, bool(payload.get("enabled", True)))) @app.put("/api/groups/{group_id}/search-listed") async def set_search_listed(group_id: str, payload: dict): return await _op(lambda: ops.set_search_listed( state, group_id, bool(payload.get("listed", True)))) # ── Scan settings (operator only, localhost) ────────────────────────── @app.put("/api/groups/{group_id}/scan-settings") async def set_scan_settings(group_id: str, payload: dict): return await _op(lambda: ops.set_scan_settings( state, group_id, float(payload.get("reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS)), float(payload.get("debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS)), )) # ── Reload config ──────────────────────────────────────────────────── @app.post("/api/reload") async def reload_config(): # start_reload, not reload_config: this must return before a # brand-new group's synchronous initial scan finishes (minutes, not # seconds, on a real library) — see ops.start_reload for why. return await _op(lambda: ops.start_reload(state)) # ── Node settings (operator only, localhost) ─────────────────────────── @app.get("/api/node-settings") async def get_node_settings(): return await _op(lambda: ops.get_node_settings(state)) @app.put("/api/node-settings") async def update_node_settings(payload: dict): return await _op(lambda: ops.set_node_settings(state, payload)) # ── Transfers (operator only, localhost) ─────────────────────────────── @app.get("/api/transfers") async def get_transfers(): return await _op(lambda: ops.list_transfers(state)) @app.put("/api/groups/{group_id}/transfer-limits") async def set_transfer_limits(group_id: str, payload: dict): # The same `ops.set_transfer_limits` the signed MNP handler calls. The # op existed with only that one door, and nothing anywhere opened it — # so the per-member cap sat at its default of 2 with no way to change # it, which from outside is indistinguishable from a hardcoded 2. return await _op(lambda: ops.set_transfer_limits( state, group_id, int(payload.get("downloads", 0)), int(payload.get("uploads", 0)))) return app