path: root/packages/meshbay-node/src/meshbay_node/ui/app.py

"""
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 `<data_dir>/ui-token`. There is
no server-rendered UI: the Node page ships in the desktop client (see
`docs/refactor-node-ui.md`).
"""

import asyncio
import logging

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse

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 `<data_dir>/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(),
            upload_dir=(payload.get("upload_dir") or "").strip(),
        ))
        reload_fn = state.get("reload_fn")
        if reload_fn:
            asyncio.ensure_future(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:
            asyncio.ensure_future(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))

    # ── 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(),
            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

    # asyncio.ensure_future 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,
        }

    # ── 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)),
        ))

    # ── 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))

    # ── 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))

    return app