""" 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 json import logging import time from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query from fastapi.responses import HTMLResponse from meshbay_node import __version__ 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, } @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 ], } # ── 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"}.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"{e.name}{e.type}" f"{_fmt_size(e.size)}{e.path or '/'}" ) groups_html += f"""

{name} {vis}

Directory: {shared}

Files: {fcount} — Total: {_fmt_size(total_size)}

ID: {gid}

File list {file_rows}
NameTypeSizePath
""" # Peers section peers_html = "" if webrtc: for pid, session in list(webrtc._sessions.items()): from meshbay_node.transport.webrtc_server import _get_remote_ip ip = session._remote_ip or _get_remote_ip(session._pc) peers_html += ( f"{session._username or session._user_id or '—'}" f"{ip or '—'}" f"{session._group_id[:8] if session._group_id else '—'}" f"{session._pc.connectionState}" ) if not peers_html: peers_html = 'No connected peers' return f""" MeshBay Node Admin

MeshBay Node {status}

{len(groups_ctx)}
Groups
{total_files}
Files
{peer_count}
Connected Peers
{state.get("username", "—")}
User

Connected Peers

{peers_html}
UserIPGroupState

Groups

{groups_html or '

No groups configured

'}

Node Configuration

Hub: {state.get("hub_url", "—")}

QUIC port: {state.get("quic_port", "—")} — TCP port: {state.get("node_port", "—")}

Node ID: {state.get("endpoint_hint") or "—"}

MeshBay Node v{__version__} — localhost only — status · groups · peers · audit · config — auto-refresh 10s
""" def _render_audit_page() -> str: return """ MeshBay Node — Audit Log

Audit Log

TimeEventUserIPGroupDetail
"""