Add secure Rokid gateway

README.md

@@ -1,13 +1,34 @@
 ---
 title: ASQ Rokid Gateway
-emoji: 💻
+emoji: 🥽
 colorFrom: green
 colorTo: gray
 sdk: gradio
-sdk_version: 6.28.0
-python_version: '3.13'
+sdk_version: 5.49.1
 app_file: app.py
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# ASQ Rokid Gateway
+
+Secure gateway between the Rokid AIUI application and the private English ASQ Space.
+
+Required Hugging Face Space secrets:
+
+- `ACCESS_CODE`: exactly six digits.
+- `SESSION_SECRET`: a long random value used to sign short-lived sessions.
+- `HF_TOKEN`: a read token that can access the private target Space.
+
+Optional variables:
+
+- `TARGET_SPACE_ID`: defaults to `ASQ2/ASQ`.
+- `TARGET_API_NAME`: defaults to `/chat_flow`.
+- `SESSION_TTL_SECONDS`: defaults to 14,400 seconds.
+
+Public endpoints:
+
+- `GET /health`
+- `POST /auth` with `{"code":"123456"}`
+- `POST /suggest` with `{"transcript":"..."}` and `Authorization: Bearer <token>`
+
+No Hugging Face token or access code is embedded in the Rokid application or committed to this repository.

app.py

@@ -0,0 +1,171 @@
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import time
+from typing import Any
+
+import gradio as gr
+import uvicorn
+from fastapi import FastAPI, Header, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from gradio_client import Client
+from pydantic import BaseModel, Field
+
+
+ACCESS_CODE = os.getenv("ACCESS_CODE", "")
+SESSION_SECRET = os.getenv("SESSION_SECRET", "")
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+TARGET_SPACE_ID = os.getenv("TARGET_SPACE_ID", "ASQ2/ASQ")
+TARGET_API_NAME = os.getenv("TARGET_API_NAME", "/chat_flow")
+SESSION_TTL_SECONDS = int(os.getenv("SESSION_TTL_SECONDS", "14400"))
+
+
+class AuthRequest(BaseModel):
+    code: str = Field(min_length=6, max_length=6, pattern=r"^\d{6}$")
+
+
+class SuggestRequest(BaseModel):
+    transcript: str = Field(min_length=1, max_length=4000)
+
+
+sessions: dict[str, dict[str, Any]] = {}
+
+
+def _b64encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
+
+
+def _b64decode(data: str) -> bytes:
+    return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))
+
+
+def _sign(payload: str) -> str:
+    return _b64encode(
+        hmac.new(SESSION_SECRET.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).digest()
+    )
+
+
+def create_session_token() -> tuple[str, int, str]:
+    session_id = secrets.token_urlsafe(18)
+    expires_at = int(time.time()) + SESSION_TTL_SECONDS
+    payload = _b64encode(
+        json.dumps({"sid": session_id, "exp": expires_at}, separators=(",", ":")).encode("utf-8")
+    )
+    return f"{payload}.{_sign(payload)}", expires_at, session_id
+
+
+def verify_session_token(token: str) -> dict[str, Any]:
+    if not SESSION_SECRET:
+        raise HTTPException(status_code=503, detail="Gateway is not configured")
+    try:
+        payload, signature = token.split(".", 1)
+        if not hmac.compare_digest(signature, _sign(payload)):
+            raise ValueError("bad signature")
+        data = json.loads(_b64decode(payload))
+        if int(data["exp"]) <= int(time.time()):
+            raise ValueError("expired")
+        session = sessions.get(data["sid"])
+        if not session:
+            raise ValueError("unknown session")
+        return session
+    except (ValueError, KeyError, json.JSONDecodeError):
+        raise HTTPException(status_code=401, detail="Invalid or expired session")
+
+
+def extract_question(result: Any) -> tuple[str, list[dict[str, Any]], int]:
+    if not isinstance(result, (list, tuple)) or len(result) < 4:
+        raise RuntimeError("Unexpected response from the ASQ Space")
+    history = result[0]
+    turns = int(result[3])
+    if not isinstance(history, list) or not history:
+        raise RuntimeError("ASQ returned no conversation history")
+    last = history[-1]
+    if isinstance(last, dict):
+        question = str(last.get("content", "")).strip()
+    elif isinstance(last, (list, tuple)) and len(last) >= 2:
+        question = str(last[1]).strip()
+    else:
+        question = ""
+    if not question:
+        raise RuntimeError("ASQ returned an empty question")
+    return question, history, turns
+
+
+def call_asq(transcript: str, session: dict[str, Any]) -> tuple[str, list[dict[str, Any]], int]:
+    if not HF_TOKEN:
+        raise RuntimeError("HF_TOKEN is not configured")
+    client = Client(TARGET_SPACE_ID, token=HF_TOKEN, verbose=False)
+    result = client.predict(
+        transcript,
+        session["history"],
+        "",
+        "",
+        session["turns"],
+        api_name=TARGET_API_NAME,
+    )
+    return extract_question(result)
+
+
+api = FastAPI(title="ASQ Rokid Gateway", version="0.1.0")
+api.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["GET", "POST", "OPTIONS"],
+    allow_headers=["Authorization", "Content-Type"],
+)
+
+
+@api.get("/health")
+def health() -> dict[str, Any]:
+    return {
+        "ok": True,
+        "configured": bool(ACCESS_CODE and SESSION_SECRET and HF_TOKEN),
+        "target": TARGET_SPACE_ID,
+    }
+
+
+@api.post("/auth")
+def authenticate(request: AuthRequest) -> dict[str, Any]:
+    if not ACCESS_CODE or not SESSION_SECRET:
+        raise HTTPException(status_code=503, detail="Gateway is not configured")
+    if not hmac.compare_digest(request.code, ACCESS_CODE):
+        raise HTTPException(status_code=401, detail="Incorrect code")
+    token, expires_at, session_id = create_session_token()
+    sessions[session_id] = {"history": [], "turns": 0, "expires_at": expires_at}
+    return {"token": token, "expiresIn": SESSION_TTL_SECONDS}
+
+
+@api.post("/suggest")
+async def suggest(
+    request: SuggestRequest,
+    authorization: str | None = Header(default=None),
+) -> dict[str, str]:
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing session token")
+    session = verify_session_token(authorization[7:])
+    try:
+        question, history, turns = await asyncio.to_thread(call_asq, request.transcript.strip(), session)
+    except Exception as error:
+        raise HTTPException(status_code=502, detail=f"ASQ inference failed: {error}")
+    session["history"] = history
+    session["turns"] = turns
+    return {"question": question}
+
+
+with gr.Blocks(title="ASQ Rokid Gateway") as status_page:
+    gr.Markdown(
+        "# ASQ Rokid Gateway\n"
+        "This service provides the authenticated API used by the Rokid glasses application.\n\n"
+        "Use `GET /health` to check configuration. The `/auth` and `/suggest` endpoints require the six-digit access flow."
+    )
+
+
+api = gr.mount_gradio_app(api, status_page, path="/ui")
+
+
+if __name__ == "__main__":
+    uvicorn.run(api, host="0.0.0.0", port=7860)