Restricted

"""OAuth-gated dashboard Space. Access model (real, server-side enforcement — not client-side): 1. The Space ships NO data. The dashboard JSONs live in the PRIVATE HF dataset repo. 2. A visitor must "Sign in with Hugging Face" (built-in Space OAuth, identity-only scope). 3. At callback we verify — with the *user's own* token — that they can read the private dataset repo. If HF says no (403/not found), they are not authorized: no data, no UI. 4. Only authorized sessions get the dashboard HTML and the /data/.json files, which the backend streams from the private repo using the Space owner token. So "approved" == "has read access to the private dataset on HF", which the owner controls by adding collaborators (or via an org). The public landing page reveals nothing. """ import os, secrets, base64, json import httpx from fastapi import FastAPI, Request, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse, FileResponse, PlainTextResponse from starlette.middleware.sessions import SessionMiddleware from huggingface_hub import HfApi, hf_hub_download from huggingface_hub.utils import HfHubHTTPError, RepositoryNotFoundError, GatedRepoError DATASET = "hunarbatra/lie_auditors_sd_eval_v4" # PRIVATE; access to THIS gates the dashboard DASH_REPO = "hunarbatra/sd_dashboard_data" # PRIVATE; holds the dashboard JSONs (moved out of DATASET so its viewer works) PROVIDER = os.environ.get("OPENID_PROVIDER_URL", "https://huggingface.co").rstrip("/") CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID", "") CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET", "") SPACE_HOST = os.environ.get("SPACE_HOST", "") # e.g. hunarbatra-0c7e4d4e.hf.space OWNER_TOKEN = os.environ.get("HF_TOKEN", "") # Space secret: owner token to read its own private data # explicit allowlist of approved HF usernames (Space secret APPROVED_USERS="alice,bob"); owner always allowed APPROVED = {u.strip().lower() for u in os.environ.get("APPROVED_USERS", "hunarbatra").split(",") if u.strip()} REDIRECT_URI = f"https://{SPACE_HOST}/login/callback" if SPACE_HOST else "http://localhost:7860/login/callback" app = FastAPI() app.add_middleware(SessionMiddleware, secret_key=os.environ.get("SESSION_SECRET", secrets.token_hex(32)), same_site="none", https_only=True) # none+Secure so the session cookie also works inside the HF Space iframe LOGIN_PAGE = """ Restricted

Restricted access

Authorized accounts only.

%%ERR%% """ DENIED_PAGE = """Not authorized

This account is not authorized.

Signed in as %%USER%%. Sign out

""" def authed(request): return bool(request.session.get("ok")) @app.get("/", response_class=HTMLResponse) def root(request: Request): if authed(request): return FileResponse("index.html") return HTMLResponse(LOGIN_PAGE.replace("%%ERR%%", "")) @app.get("/login") def login(request: Request): state = secrets.token_urlsafe(16) request.session["state"] = state scope = "openid profile read-repos" # read-repos so we can verify the user's access to the private dataset url = (f"{PROVIDER}/oauth/authorize?client_id={CLIENT_ID}" f"&redirect_uri={REDIRECT_URI}&response_type=code&scope={scope.replace(' ', '%20')}&state={state}") return RedirectResponse(url) @app.get("/logout") def logout(request: Request): request.session.clear() return RedirectResponse("/") @app.get("/login/callback", response_class=HTMLResponse) def callback(request: Request, code: str = "", state: str = ""): if not code or state != request.session.get("state"): return HTMLResponse(LOGIN_PAGE.replace("%%ERR%%", "

"), status_code=400) # exchange code -> user access token (Basic auth with client id/secret) basic = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode() r = httpx.post(f"{PROVIDER}/oauth/token", data={"grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI}, headers={"Authorization": f"Basic {basic}"}, timeout=30) if r.status_code != 200: return HTMLResponse(LOGIN_PAGE.replace("%%ERR%%", "

Token exchange failed.

"), status_code=400) user_token = r.json().get("access_token", "") # identify the user, then verify — with THEIR token — that they can read the private dataset. # This is the gate: access == read access to hunarbatra/lie_auditors_sd_eval_v2 (HF enforces it). try: who = HfApi(token=user_token).whoami() user = who.get("name") or who.get("fullname") or "user" except Exception: user = "user" try: HfApi(token=user_token).repo_info(DATASET, repo_type="dataset") # raises if no access except (RepositoryNotFoundError, GatedRepoError, HfHubHTTPError): request.session.clear() return HTMLResponse(DENIED_PAGE.replace("%%USER%%", user), status_code=403) request.session["ok"] = True request.session["user"] = user return RedirectResponse("/", status_code=302) @app.get("/whoami") def whoami(request: Request): if not authed(request): raise HTTPException(401) return {"user": request.session.get("user")} @app.get("/data/{name}.json") def data(name: str, request: Request): if not authed(request): raise HTTPException(status_code=401, detail="not signed in") if not name.replace("_", "").replace("-", "").isalnum(): raise HTTPException(status_code=400, detail="bad name") # session already proved access to DATASET at login; stream the dashboard JSON from DASH_REPO via owner token try: path = hf_hub_download(DASH_REPO, f"{name}.json", repo_type="dataset", token=OWNER_TOKEN) except Exception: raise HTTPException(status_code=404, detail="not found") return FileResponse(path, media_type="application/json") @app.get("/healthz") def healthz(): return PlainTextResponse("ok")