Scopes: contribute-repos write-discussions email.
Checking session…""" @app.get("/oauth/login") def oauth_login(request: Request) -> RedirectResponse: if not (OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET): raise HTTPException(503, "OAuth is not configured") _prune() state = secrets.token_urlsafe(32) _states[state] = time.time() params = urlencode( { "response_type": "code", "client_id": OAUTH_CLIENT_ID, "redirect_uri": _redirect_uri(request), "scope": SCOPES, "state": state, "orgIds": ORG_ID, } ) return RedirectResponse(f"{HUB}/oauth/authorize?{params}") @app.get("/oauth/callback", name="oauth_callback") async def oauth_callback(request: Request) -> RedirectResponse: code = request.query_params.get("code") state = request.query_params.get("state") if not code or not state or _states.pop(state, None) is None: raise HTTPException(400, "Invalid OAuth state") async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client: response = await client.post( f"{HUB}/oauth/token", auth=(OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET), data={ "grant_type": "authorization_code", "code": code, "redirect_uri": _redirect_uri(request), "client_id": OAUTH_CLIENT_ID, }, headers={"Accept": "application/json"}, ) response.raise_for_status() token_data = response.json() access_token = token_data.get("access_token") if not access_token: raise HTTPException(502, "Provider returned no access token") userinfo_response = await client.get( f"{HUB}/oauth/userinfo", headers={"Authorization": f"Bearer {access_token}"}, ) userinfo_response.raise_for_status() sid = secrets.token_urlsafe(32) _sessions[sid] = { "created_at": time.time(), "token_data": token_data, "userinfo": userinfo_response.json(), } redirect = RedirectResponse("/") redirect.set_cookie( "probe_sid", sid, max_age=3600, httponly=True, secure=True, samesite="none", ) return redirect def _session(probe_sid: str | None) -> dict[str, Any]: if not probe_sid or probe_sid not in _sessions: raise HTTPException(401, "Not signed in") return _sessions[probe_sid] @app.get("/probe/status") def probe_status(probe_sid: str | None = Cookie(default=None)) -> dict[str, Any]: session = _sessions.get(probe_sid or "") if not session: return {"authenticated": False} userinfo = session["userinfo"] return { "authenticated": True, "username": userinfo.get("preferred_username") or userinfo.get("name") or "unknown", } async def _refresh(token_data: dict[str, Any]) -> tuple[str, dict[str, Any]]: refresh_token = token_data.get("refresh_token") report = { "present": bool(refresh_token), "succeeded": False, "access_token_rotated": None, "refresh_token_rotated": None, "expires_in": None, } if not refresh_token: return str(token_data["access_token"]), report async with httpx.AsyncClient(timeout=30) as client: response = await client.post( f"{HUB}/oauth/token", auth=(OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET), data={ "grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": OAUTH_CLIENT_ID, }, headers={"Accept": "application/json"}, ) if not response.is_success: report["error"] = f"HTTP {response.status_code}" return str(token_data["access_token"]), report refreshed = response.json() new_access = str(refreshed.get("access_token") or token_data["access_token"]) new_refresh = refreshed.get("refresh_token") report.update( { "succeeded": True, "access_token_rotated": new_access != token_data["access_token"], "refresh_token_rotated": bool(new_refresh and new_refresh != refresh_token), "expires_in": refreshed.get("expires_in"), } ) return new_access, report def _run_lifecycle(token: str, userinfo: dict[str, Any]) -> dict[str, Any]: result: dict[str, Any] = { "whoami": False, "create_bucket": False, "write_bucket": False, "open_pr": False, "comment_pr": False, "close_pr": False, "personal_repo_write_denied": False, "cleanup_bucket": False, } suffix = secrets.token_hex(4) agent_id = f"test-oauth-{suffix}" bucket = f"{ORG}/rl-{agent_id}" api = HfApi(token=token) pr_num: int | None = None try: identity = whoami(token=token) username = str(identity.get("name") or "") if not username: raise RuntimeError("whoami returned no username") result["whoami"] = True result["username_matches_userinfo"] = username == ( userinfo.get("preferred_username") or userinfo.get("name") ) result["token_auth_type"] = (identity.get("auth") or {}).get("type") create_bucket(bucket, token=token) result["create_bucket"] = True batch_bucket_files( bucket, add=[(b"hello from the OAuth scope probe\n", "probe.txt")], token=token, ) result["write_bucket"] = True commit = api.create_commit( repo_id=f"{ORG}/knowledge-base", repo_type="dataset", operations=[ CommitOperationAdd( f"sources/probe-oauth-{suffix}.md", b"# OAuth scope validation probe\n", ) ], commit_message=f"source: OAuth scope validation {suffix}", commit_description=f"agent: {agent_id}\n\nTemporary OAuth scope validation probe.", create_pr=True, ) pr_num = commit.pr_num if pr_num is None: raise RuntimeError("create_commit returned no PR number") result["open_pr"] = True result["pr_num"] = pr_num api.comment_discussion( f"{ORG}/knowledge-base", pr_num, comment=f"/comment\n\nagent: {agent_id}\n\nOAuth permission probe.", repo_type="dataset", ) result["comment_pr"] = True api.change_discussion_status( f"{ORG}/knowledge-base", pr_num, new_status="closed", repo_type="dataset", ) result["close_pr"] = True try: api.create_commit( repo_id=PERSONAL_REPO, repo_type="model", operations=[CommitOperationAdd("oauth-should-fail.txt", b"denied\n")], commit_message="OAuth scope probe: this write must fail", create_pr=False, ) result["personal_repo_write_error"] = "UNEXPECTED SUCCESS" except Exception as error: result["personal_repo_write_denied"] = True result["personal_repo_denial_type"] = type(error).__name__ except Exception as error: result["error_step"] = next( (key for key, value in result.items() if value is False), "unknown" ) result["error_type"] = type(error).__name__ result["error"] = str(error)[:500] finally: if pr_num is not None and not result["close_pr"]: try: api.change_discussion_status( f"{ORG}/knowledge-base", pr_num, new_status="closed", repo_type="dataset", ) result["close_pr_cleanup"] = True except Exception as error: result["close_pr_cleanup_error"] = type(error).__name__ try: delete_bucket(bucket, missing_ok=True, token=token) result["cleanup_bucket"] = True except Exception as error: result["cleanup_bucket_error"] = type(error).__name__ required = ( "whoami", "create_bucket", "write_bucket", "open_pr", "comment_pr", "close_pr", "personal_repo_write_denied", "cleanup_bucket", ) result["passed"] = all(result[key] is True for key in required) return result @app.post("/probe/run") async def run_probe(probe_sid: str | None = Cookie(default=None)) -> JSONResponse: session = _session(probe_sid) token, refresh_report = await _refresh(session["token_data"]) result = await asyncio.to_thread(_run_lifecycle, token, session["userinfo"]) userinfo = session["userinfo"] result["userinfo"] = { "sub_present": bool(userinfo.get("sub")), "email_present": bool(userinfo.get("email")), "organization_fields": sorted( (userinfo.get("organizations") or userinfo.get("orgs") or [{}])[0].keys() ) if (userinfo.get("organizations") or userinfo.get("orgs")) else [], } result["refresh"] = refresh_report return JSONResponse(result, headers={"Cache-Control": "no-store"})