"""HF Space: FastAPI lobbies/WS + full Dust2 game client (www/)."""
from __future__ import annotations
import mimetypes
import os
import re
import secrets
from pathlib import Path
from typing import Optional
import uvicorn
import httpx
from fastapi import Cookie, FastAPI, Header, Request, Response, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from lobbies import board
from relay import handle_match_ws
WWW = Path(__file__).resolve().parent / "www"
_GUEST_RE = re.compile(r"^[A-Za-z0-9_-]{2,32}$")
mimetypes.add_type("model/gltf-binary", ".glb")
mimetypes.add_type("model/gltf+json", ".gltf")
class ClaimBody(BaseModel):
username: str = Field(..., min_length=1)
seat: str = Field(..., min_length=1)
avatarUrl: Optional[str] = None
class LeaveBody(BaseModel):
username: str = Field(..., min_length=1)
class HeartbeatBody(BaseModel):
username: str = Field(..., min_length=1)
BOARD_HTML = """
RL-PVP · Lobby board
Dust2 · RL-PVP
Public lobby board + spectate. Playing requires the local client with HF login (for identity / analytics).
Spectate (no HF account): click Spectate on a live lobby.
In spectate: ← → switch player · Esc back. No movement, no weapons, no play.
Play: local Dust2 only (bash start.sh + HF_TOKEN).
Red = side X · Blue = side Y (labels are seat numbers only).
Loading lobbies…
"""
def _ensure_guest(response: Response, dust2_user: str | None) -> str:
name = (dust2_user or "").strip()
if not name or not _GUEST_RE.fullmatch(name):
name = f"guest-{secrets.token_hex(3)}"
response.set_cookie(
key="dust2_user",
value=name,
max_age=60 * 60 * 24 * 30,
httponly=False,
samesite="lax",
)
return name
def _hf_username_from_auth(authorization: str | None) -> str | None:
if not authorization:
return None
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
return None
token = parts[1].strip()
if not token:
return None
try:
resp = httpx.get(
"https://huggingface.co/api/whoami-v2",
headers={"Authorization": f"Bearer {token}"},
timeout=15.0,
)
if resp.status_code != 200:
return None
data = resp.json()
name = data.get("name") or data.get("fullname")
return str(name) if name else None
except Exception:
return None
def create_app() -> FastAPI:
api = FastAPI()
api.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@api.get("/api/health")
def health():
return {"ok": True, "host": "space", "hasClient": WWW.is_dir(), "playAllowed": False}
@api.get("/api/config")
def config(response: Response, dust2_user: str | None = Cookie(default=None)):
# Guest id is for spectate only — Space never allows play.
username = _ensure_guest(response, dust2_user)
return {
"ok": True,
"spaceUrl": "",
"username": username,
"avatarUrl": None,
"authError": None,
"hasToken": False,
"hasSpaceUrl": False,
"lobbyHost": "space",
"host": "space",
"playAllowed": False,
}
@api.get("/api/lobbies")
def get_lobbies():
return board.snapshot()
@api.post("/api/lobbies/{mode}/{lobby_id}/claim")
def claim_seat(
mode: str,
lobby_id: str,
body: ClaimBody,
authorization: str | None = Header(default=None),
):
# Play only via local client proxy with a real HF token (analytics identity).
hf_user = _hf_username_from_auth(authorization)
if not hf_user:
return JSONResponse(
{
"ok": False,
"error": "Play disabled on Space. Use local Dust2 with HF_TOKEN.",
},
status_code=403,
)
result = board.claim(
mode, lobby_id, hf_user, body.seat, avatar_url=body.avatarUrl
)
return JSONResponse(result, status_code=200 if result.get("ok") else 400)
@api.post("/api/lobbies/leave")
def leave_lobby(body: LeaveBody, authorization: str | None = Header(default=None)):
hf_user = _hf_username_from_auth(authorization)
# Allow leave for HF-auth players; ignore anonymous spoof leave of others
if hf_user:
return board.leave(hf_user)
if body.username and not str(body.username).startswith("guest-"):
# Unauthenticated leave of a real seat — reject
return JSONResponse({"ok": False, "error": "HF auth required"}, status_code=403)
return {"ok": True}
@api.post("/api/lobbies/heartbeat")
def heartbeat(body: HeartbeatBody, authorization: str | None = Header(default=None)):
hf_user = _hf_username_from_auth(authorization)
if hf_user:
board.heartbeat(hf_user)
elif body.username and not str(body.username).startswith("guest-"):
return JSONResponse({"ok": False, "error": "HF auth required"}, status_code=403)
return {"ok": True}
@api.websocket("/ws/match/{mode}/{lobby_id}")
async def match_ws(
ws: WebSocket,
mode: str,
lobby_id: str,
user: str = "",
role: str = "play",
):
spectate = (role or "").lower() in ("spectate", "spec", "watch")
# Browser on Space: spectate only. Play sockets are for seated HF players
# (local client claims with HF_TOKEN, then connects play WS).
if not spectate:
# Allow play only if this user already holds a seat (HF-authenticated claim).
seated = board.by_user.get((user or "").strip())
if not seated or seated[0] != lobby_id:
await ws.accept()
await ws.send_text(
'{"type":"error","error":"Spectate only in browser. Play via local Dust2 + HF_TOKEN."}'
)
await ws.close()
return
await handle_match_ws(ws, mode, lobby_id, user, role=role)
@api.get("/board", response_class=HTMLResponse)
def lobby_board():
return BOARD_HTML
@api.get("/")
def game_index(request: Request):
# Game client on Space is spectate-entry only.
q = request.query_params
if q.get("spectate") == "1" and q.get("lobby"):
index = WWW / "index.html"
if not index.is_file():
return HTMLResponse(
"
Game client missing
Rebuild Space with www/.
",
status_code=503,
)
return FileResponse(index)
return RedirectResponse(url="/board", status_code=302)
if WWW.is_dir():
for mount, folder in (("assets", "assets"), ("js", "js"), ("css", "css")):
path = WWW / folder
if path.is_dir():
api.mount(f"/{mount}", StaticFiles(directory=str(path)), name=mount)
return api
app = create_app()
def main() -> None:
port = int(os.environ.get("PORT") or "7860")
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
if __name__ == "__main__":
main()