#!/usr/bin/env python3 """Authenticated room bootstrap. All identity/routing is resolved server-side.""" import base64 import hashlib import hmac import json import os import time import uuid import httpx from fastapi import FastAPI, Request from fastapi.responses import JSONResponse try: from web.voice_catalog import catalog except ModuleNotFoundError: # Direct execution: python /app/web/token_server.py from voice_catalog import catalog API_KEY = os.getenv("LIVEKIT_API_KEY", "") API_SECRET = os.getenv("LIVEKIT_API_SECRET", "") AGENT_NAME = os.getenv("VOICE_AGENT_NAME", "voice-assistant") CONTROL_URL = os.getenv("CONTROL_URL", "http://127.0.0.1:8092").rstrip("/") CONTROL_SECRET = os.getenv("CONTROL_SERVICE_SECRET", "") COOKIE_NAME = "hope_session" app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) def b64url(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode() def make_token(room_name: str, identity: str) -> str: if not API_KEY or not API_SECRET: raise RuntimeError("LiveKit credentials are not configured") now = int(time.time()) payload = {"iss": API_KEY, "sub": identity, "nbf": now - 5, "exp": now + 300, "jti": str(uuid.uuid4()), "video": {"roomJoin": True, "room": room_name, "canPublish": True, "canSubscribe": True, "canPublishData": True, "canPublishSources": ["microphone"]}, "roomConfig": {"agents": [{"agentName": AGENT_NAME, "jobType": "JT_ROOM"}]}} head = b64url(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()) body = b64url(json.dumps(payload).encode()) sig = hmac.new(API_SECRET.encode(), f"{head}.{body}".encode(), hashlib.sha256).digest() return f"{head}.{body}.{b64url(sig)}" @app.post("/token") async def token(request: Request): headers = {"Cache-Control": "no-store"} if not CONTROL_SECRET or not API_KEY or not API_SECRET: return JSONResponse({"error": "Service configuration unavailable"}, 503, headers=headers) auth = request.headers.get("authorization", "") bearer = auth[7:] if auth.startswith("Bearer ") else "" credential = bearer or request.cookies.get(COOKIE_NAME, "") if not credential or len(credential) > 512: return JSONResponse({"error": "Sign in required"}, 401, headers=headers) if not bearer: origin = request.headers.get("origin", "") allowed = set(filter(None, os.getenv("CONTROL_ALLOWED_ORIGINS", "").split(","))) if origin not in allowed: return JSONResponse({"error": "Invalid request origin"}, 403, headers=headers) # The conversation identifier is only a request to bind an existing chat; # the trusted account service performs ownership checks before granting it. body_bytes = bytearray() async for chunk in request.stream(): body_bytes.extend(chunk) if len(body_bytes) > 4096: return JSONResponse({"error": "Request body too large"}, 413, headers=headers) try: body = json.loads(body_bytes) if body_bytes else {} except (ValueError, UnicodeDecodeError): return JSONResponse({"error": "Invalid request"}, 400, headers=headers) if not isinstance(body, dict): return JSONResponse({"error": "Invalid request"}, 400, headers=headers) conversation_id = body.get("conversation_id") if conversation_id is not None and (not isinstance(conversation_id, str) or not 1 <= len(conversation_id) <= 128): return JSONResponse({"error": "Invalid conversation"}, 400, headers=headers) try: async with httpx.AsyncClient(timeout=10) as client: control_headers = {"X-Control-Secret": CONTROL_SECRET} response = await client.post(f"{CONTROL_URL}/internal/resolve-session", headers=control_headers, json={"token": credential}) if response.status_code != 200: status = response.status_code if response.status_code in (401,403,503) else 503 return JSONResponse({"error": "Sign in required" if status==401 else "Your private agent is not ready"}, status, headers=headers) account = response.json() user = account["user"] room_name = f"voice-{user['id']}-{uuid.uuid4().hex}" grant = await client.post(f"{CONTROL_URL}/internal/rooms", headers=control_headers, json={"user_id": user["id"], "room_name": room_name, "ttl": 86400, "session_token": credential, "conversation_id": conversation_id}) if grant.status_code != 201: status = grant.status_code if grant.status_code in (401, 403, 404, 409, 503) else 503 return JSONResponse({"error": "Conversation unavailable"}, status, headers=headers) granted = grant.json() return JSONResponse({"token": make_token(room_name, f"user:{user['id']}"), "room": room_name, "conversation_id": granted.get("conversation_id"), "user": {"id": user["id"], "display_name": user["display_name"]}}, headers=headers) except (httpx.HTTPError, ValueError, KeyError): return JSONResponse({"error": "Account service unavailable"}, 503, headers=headers) @app.get("/voices") async def voices(request: Request): """Resolve the caller at the trusted control boundary before reading the cache.""" headers = {"Cache-Control": "no-store"} if not CONTROL_SECRET: return JSONResponse({"error": "Service configuration unavailable"}, 503, headers=headers) auth = request.headers.get("authorization", "") bearer = auth[7:] if auth.startswith("Bearer ") else "" credential = bearer or request.cookies.get(COOKIE_NAME, "") if not credential or len(credential) > 512: return JSONResponse({"error": "Sign in required"}, 401, headers=headers) # Safe GET requests normally have no Origin. If supplied with cookie auth, # enforce the same configured origin allowlist as the bootstrap boundary. origin = request.headers.get("origin", "") if not bearer and origin: allowed = set(filter(None, os.getenv("CONTROL_ALLOWED_ORIGINS", "").split(","))) if origin not in allowed: return JSONResponse({"error": "Invalid request origin"}, 403, headers=headers) try: async with httpx.AsyncClient(timeout=10) as client: response = await client.post(f"{CONTROL_URL}/internal/resolve-session", headers={"X-Control-Secret": CONTROL_SECRET}, json={"token": credential}) if response.status_code != 200: status = response.status_code if response.status_code in (401, 403, 503) else 503 return JSONResponse({"error": "Sign in required" if status == 401 else "Your private agent is not ready"}, status, headers=headers) # Verify the expected control response without exposing private metadata. account = response.json() if not isinstance(account, dict) or not isinstance(account.get("user"), dict) or not account["user"].get("id"): return JSONResponse({"error": "Account service unavailable"}, 503, headers=headers) except (httpx.HTTPError, ValueError, TypeError): return JSONResponse({"error": "Account service unavailable"}, 503, headers=headers) return JSONResponse(await catalog.get(), headers=headers) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8091, access_log=False)