""" MeshBay Node — local administration web UI (localhost:18000). FastAPI app providing: - Dashboard: node status, connected peers, group overview - Groups: file listing, shared directory info - Peers: connected WebRTC/QUIC clients - Audit log: IP + action log for legal compliance - API endpoints for all data (JSON) Served only on 127.0.0.1 — not exposed to the network. No authentication required (localhost only). """ import base64 import json import logging import time from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ from meshbay_common.crypto import generate_gek, wrap_gek_aes log = logging.getLogger(__name__) def create_ui_app(state: dict) -> FastAPI: app = FastAPI( title="MeshBay Node Admin", version=__version__, docs_url=None, redoc_url=None, ) # ── 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") return { "version": __version__, "status": state.get("status", "starting"), "hub_url": state.get("hub_url", ""), "username": state.get("username", ""), "node_port": state.get("node_port", 0), "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.get("/api/groups") async def api_groups(): groups_ctx = state.get("groups_ctx", {}) config = state.get("config") result = [] for gid, ctx in groups_ctx.items(): cfg = None if config: cfg = next((g for g in config.groups if g.id == gid), None) idx = ctx.get("index") result.append({ "id": gid, "name": cfg.name if cfg else gid[:8], "shared_dir": str(ctx.get("shared_root", "")), "visibility": cfg.visibility if cfg else "private", "file_count": idx.count if idx else 0, "index_version": idx.version if idx else 0, }) return {"groups": result} @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": []} peers = [] for pid, session in list(webrtc._sessions.items()): from meshbay_node.transport.webrtc_server import _get_remote_ip peers.append({ "peer_id": pid, "user_id": session._user_id or "", "username": session._username or "", "group_id": session._group_id or "", "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, user_id: str | None = Query(default=None), event: str | None = Query(default=None), ): audit = state.get("audit_store") if not audit: return {"entries": []} entries = await audit.get_entries( since=since, limit=limit, user_id=user_id, event=event) return { "entries": [ { "id": e.id, "timestamp": e.timestamp, "user_id": e.user_id, "username": e.username, "ip": e.ip, "event": e.event, "group_id": e.group_id, "detail": e.detail, } for e in entries ] } @app.get("/api/config") async def api_config(): config = state.get("config") if not config: return {} return { "hub_url": config.hub.url, "username": config.hub.username, "node_port": config.node.port, "quic_port": config.node.quic_port, "http_port": config.node.http_port, "ui_port": config.node.ui_port, "data_dir": str(config.data_dir), "groups": [ { "id": g.id, "name": g.name, "shared_dir": g.shared_dir, "visibility": g.visibility, } for g in config.groups ], } # ── GEK initialization (operator only, localhost) ────────────────────── @app.post("/api/groups/{group_id}/gek") async def init_gek(group_id: str): """Generate GEK, wrap for all group members, store, and activate.""" groups_ctx = state.get("groups_ctx", {}) if group_id not in groups_ctx: return JSONResponse({"error": "Group not hosted on this node"}, 404) hub = state.get("hub") if not hub or not hub._session: return JSONResponse({"error": "Hub not connected"}, 503) bundle_store = state.get("bundle_store") if not bundle_store: return JSONResponse({"error": "Bundle store not available"}, 503) await hub.ensure_fresh_token() session = hub._session members_resp = await hub._http.get( f"/v1/groups/{group_id}/members", headers=session.auth_headers, ) if not members_resp.is_success: return JSONResponse( {"error": f"Failed to fetch members: {members_resp.status_code}"}, 502) members = members_resp.json().get("members", []) if not members: return JSONResponse({"error": "No members in group"}, 400) existing_gek = groups_ctx[group_id].get("gek") gek = existing_gek or generate_gek() wrapped_count = 0 errors = [] for member in members: username = member["username"] user_id = member["user_id"] try: pk_data = await hub.get_user_pubkeys(username) pk_x_raw = base64.b64decode(pk_data["pk_x25519"]) bundle = wrap_gek_aes(gek, pk_x_raw) await bundle_store.store( group_id, user_id, bundle["pk_eph_b64"], bundle["nonce_b64"], bundle["wrapped_b64"], ) wrapped_count += 1 log.info("GEK wrapped for %s (%s)", username, user_id[:8]) except Exception as e: errors.append(f"{username}: {e}") log.warning("Failed to wrap GEK for %s: %s", username, e) if wrapped_count == 0: return JSONResponse( {"error": "Failed to wrap GEK for any member", "details": errors}, 500) # Also store a copy wrapped for the node keystore X25519 key # so the daemon can reload GEK on restart without the operator's browser keys config = state.get("config") node_user_id = hub._session.user_id if hub._session else None pk_x_node_raw = state.get("pk_x25519_raw") if pk_x_node_raw and node_user_id: try: node_bundle = wrap_gek_aes(gek, pk_x_node_raw) await bundle_store.store( group_id, f"_node_{node_user_id}", node_bundle["pk_eph_b64"], node_bundle["nonce_b64"], node_bundle["wrapped_b64"], ) log.info("GEK also wrapped for node keystore (daemon reload)") except Exception as e: log.warning("Failed to wrap GEK for node keystore: %s", e) groups_ctx[group_id]["gek"] = gek log.info("GEK initialized for group %s — wrapped for %d/%d members", group_id[:8], wrapped_count, len(members)) webrtc = state.get("webrtc") if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]: webrtc._ctx["groups"][group_id]["gek"] = gek return { "status": "ok", "group_id": group_id, "wrapped_count": wrapped_count, "total_members": len(members), "errors": errors, } # ── Chat endpoints ─────────────────────────────────────────────────────── _chat_subscribers: list[WebSocket] = [] @app.get("/api/chat/history") async def chat_history(since: float = 0, limit: int = 100): chat_store = state.get("chat_store") if not chat_store: return {"messages": []} msgs = await chat_store.get_messages(since=since, limit=limit) return { "messages": [ { "id": m.id, "sender_id": m.sender_id, "iteration": m.iteration, "timestamp": m.timestamp, "thread_id": m.thread_id, } for m in msgs ] } @app.websocket("/ws/chat") async def chat_websocket(ws: WebSocket): await ws.accept() _chat_subscribers.append(ws) try: while True: await ws.receive_text() except WebSocketDisconnect: pass finally: _chat_subscribers.remove(ws) async def broadcast_chat_to_ui(msg: dict) -> None: payload = json.dumps(msg) dead = [] for ws in _chat_subscribers: try: await ws.send_text(payload) except Exception: dead.append(ws) for ws in dead: _chat_subscribers.remove(ws) app.broadcast_chat = broadcast_chat_to_ui # ── HTML UI ────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def root(): return _render_page(state) @app.get("/audit", response_class=HTMLResponse) async def audit_page(): return _render_audit_page() return app def _fmt_size(n: int) -> str: if n < 1024: return f"{n} B" if n < 1024 * 1024: return f"{n / 1024:.1f} KB" if n < 1024 * 1024 * 1024: return f"{n / (1024 * 1024):.1f} MB" return f"{n / (1024 * 1024 * 1024):.2f} GB" def _render_page(state: dict) -> str: status = state.get("status", "starting") indexes = state.get("indexes", {}) groups_ctx = state.get("groups_ctx", {}) config = state.get("config") webrtc = state.get("webrtc") total_files = sum(idx.count for idx in indexes.values()) peer_count = webrtc.active_peers if webrtc else 0 status_color = { "running": "#22c55e", "error": "#ef4444", "waiting_for_node_key": "#f97316", }.get(status, "#f59e0b") # Groups section groups_html = "" for gid, ctx in groups_ctx.items(): cfg = None if config: cfg = next((g for g in config.groups if g.id == gid), None) idx = ctx.get("index") name = cfg.name if cfg else gid[:8] shared = ctx.get("shared_root", "") vis = cfg.visibility if cfg else "private" fcount = idx.count if idx else 0 total_size = sum(e.size for e in idx.entries) if idx else 0 file_rows = "" if idx: for e in sorted(idx.entries, key=lambda x: x.name): file_rows += ( f"
Directory: {shared}
Files: {fcount} — Total: {_fmt_size(total_size)}
{gek_action}ID: {gid}
| Name | Type | Size | Path |
|---|
| User | IP | Group | State |
|---|
No groups configured
'}Hub: {state.get("hub_url", "—")}
QUIC port: {state.get("quic_port", "—")} — TCP port: {state.get("node_port", "—")}
Node ID: {state.get("endpoint_hint") or "—"}
To connect to your group from a browser, link this node to your hub account. Copy the key below and paste it in Settings > Link Node on the hub.
{state.get("pk_node_ed25519", "—")}
This is the node's Ed25519 public key. It's safe to share — it identifies this node but cannot be used to impersonate it.
| Time | Event | User | IP | Group | Detail |
|---|