"""Companion bridge — /api/companion/*. A thin, additive layer so a LAN client (e.g. a phone) can discover what a server offers and pair to it, without duplicating any LLM logic. Auth is enforced globally by AuthMiddleware (app.py), so reaching a handler here means the caller is authenticated by either a cookie session or a Bearer `ody_` API token. The read endpoints (ping/info/models) accept either; the pairing endpoints are admin-cookie only. Pairing CSRF posture: minting happens ONLY on POST. The session cookie is SameSite=Lax (routes/auth_routes.py), which a browser does not send on a cross-site POST, so an admin's cookie can't be used by a malicious page to mint a token -- the same protection the existing POST /api/tokens relies on. Minting on a GET would be unsafe (Lax cookies ride top-level GET navigations), so GET /pair only renders a form. """ import html from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse from core.middleware import require_admin from src.auth_helpers import get_current_user from companion import pairing as _pairing def token_owner(request: Request) -> str | None: """The real owner to attribute a request to, for read-scoping. Cookie sessions resolve to the logged-in username via get_current_user. Bearer-token callers come through as the sandboxed pseudo-user "api"; their real owner is stamped on request.state.api_token_owner by the auth middleware. Returns None when no owner can be resolved. """ if getattr(request.state, "api_token", False): return getattr(request.state, "api_token_owner", None) return get_current_user(request) def owner_can_see(row_owner, owner) -> bool: """Owner-scope rule for read endpoints. A caller sees a row when it is their own, or when it is a legacy null-owner ("shared") row. A caller must NEVER see another owner's row. Mirrors the `owner_filter` rule used elsewhere, expressed as a pure predicate so it can be tested directly and used as a defensive in-Python check alongside the SQL filter. """ return row_owner is None or row_owner == owner def mint_pairing_token(owner: str, invalidate=None) -> tuple[str, str]: """Mint a pairing token AND invalidate the auth middleware's in-memory token cache, so the new token is accepted on the very next request without a server restart. Returns (token_id, raw_token); the raw token is shown once. `invalidate` is the app's request.app.state.invalidate_token_cache callable (passed in so this stays a pure, testable unit). """ token_id, raw_token = _pairing.mint_token(owner) if callable(invalidate): invalidate() return token_id, raw_token def setup_companion_routes() -> APIRouter: router = APIRouter(prefix="/api/companion", tags=["companion"]) @router.get("/ping") def ping(request: Request): """Cheap, auth-validated health check. A 200 with ok=true confirms the host/port and credential are valid; middleware returns 401 otherwise.""" from core.constants import APP_VERSION return { "ok": True, "name": "odysseus", "version": APP_VERSION, "auth": "token" if getattr(request.state, "api_token", False) else "session", } @router.get("/info") def info(request: Request): """Server identity + coarse capability flags. `owner` is the caller's own identity (the token's owner for bearer callers).""" from core.constants import APP_VERSION return { "name": "odysseus", "version": APP_VERSION, "owner": token_owner(request), "capabilities": {"chat": True, "streaming": True}, } @router.get("/models") def models(request: Request): """LLM model endpoints the CALLER can use. The stock /api/models route scopes to get_current_user, which for a bearer token is the sandboxed pseudo-user "api" (owns nothing). Here we scope to the token's real owner instead, plus legacy null-owner shared rows -- the same rule as owner_filter. Read-only; never returns api_key material. """ import json as _json from core.database import SessionLocal, ModelEndpoint from src.endpoint_resolver import build_chat_url owner = token_owner(request) out = [] db = SessionLocal() try: q = db.query(ModelEndpoint).filter( ModelEndpoint.is_enabled == True, # noqa: E712 (ModelEndpoint.model_type == "llm") | (ModelEndpoint.model_type == None), # noqa: E711 ) if owner: q = q.filter((ModelEndpoint.owner == owner) | (ModelEndpoint.owner == None)) # noqa: E711 for ep in q.all(): if not owner_can_see(ep.owner, owner): continue try: model_ids = _json.loads(ep.cached_models) if ep.cached_models else [] except (ValueError, TypeError): model_ids = [] try: hidden = set(_json.loads(ep.hidden_models)) if ep.hidden_models else set() except (ValueError, TypeError): hidden = set() model_ids = [m for m in model_ids if m not in hidden] try: chat_url = build_chat_url(ep.base_url) except Exception: chat_url = ep.base_url out.append({ "endpoint_id": ep.id, "name": ep.name, "endpoint_url": chat_url, "models": model_ids, "supports_tools": ep.supports_tools, }) finally: db.close() return {"endpoints": out} @router.get("/pair") def pair_page(request: Request): """Admin-only pairing page. Renders a form that POSTs to mint a code. A GET never mints a credential: SameSite=Lax session cookies ride top-level GET navigations, so minting on GET would be triggerable by a link or (CSRF). The actual mint is the POST handler below. """ require_admin(request) page = """ Pair a device

Pair a device

Generate a one-time pairing code (a chat-scoped API token) for a LAN client.

Admin only. Each code mints a new token, shown once. Manage or revoke under Settings → API tokens.

""" return HTMLResponse(page) @router.post("/pair") def pair_create(request: Request): """Mint a pairing code. Admin-cookie only; CSRF-safe because the SameSite=Lax session cookie is not sent on a cross-site POST (same protection as POST /api/tokens). Minting invalidates the token cache so the code works immediately, no restart. `?format=json` returns the payload for an in-app pairing screen.""" require_admin(request) owner = get_current_user(request) invalidate = getattr(request.app.state, "invalidate_token_cache", None) token_id, raw_token = mint_pairing_token(owner, invalidate) hosts = _pairing.lan_ip_candidates() host = hosts[0] if hosts else "127.0.0.1" port = request.url.port or _pairing.default_port() payload = _pairing.pairing_payload(host, port, raw_token) qr = _pairing.pairing_qr_png_data_uri(payload) qr_ok = bool(qr and qr.startswith("data:image/png;base64,")) if (request.query_params.get("format") or "").lower() == "json": return { "host": host, "port": port, "token": raw_token, "token_id": token_id, "hosts": hosts, "payload": payload, "qr": qr if qr_ok else None, } import json as _json payload_json = _json.dumps(payload, separators=(",", ":")) # Only ever emit a known PNG data-URI into the src; every other value is # html.escaped. qr_block = ( f'Pairing QR' if qr_ok else "

QR rendering unavailable -- enter the details manually.

" ) page = f""" Pairing code

Pairing code

{qr_block}
Host: {html.escape(host)}
Port: {html.escape(str(port))}
Token: {html.escape(raw_token)}
Payload: {html.escape(payload_json)}

Shown once. This grants chat access to your Odysseus; revoke it in Settings → API tokens (id {html.escape(token_id)}). The device must be on the same network, and the server must bind to your LAN.

""" return HTMLResponse(page) return router