Spaces:
Running
Running
File size: 2,464 Bytes
ccec886 16e8620 ccec886 16e8620 ccec886 197023a ccec886 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
import os
import httpx
from authlib.integrations.starlette_client import OAuth
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
OAUTH_CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID")
OAUTH_CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET")
OAUTH_SCOPES = os.environ.get("OAUTH_SCOPES")
OAUTH_SCOPES = "profile" # TODO: remove when openid is fixed (honor nonce)
OPENID_PROVIDER_URL = os.environ.get("OPENID_PROVIDER_URL")
for value in (OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET, OAUTH_SCOPES, OPENID_PROVIDER_URL):
if value is None:
raise ValueError("Missing environment variable")
USER_INFO_URL = OPENID_PROVIDER_URL + "/oauth/userinfo"
METADATA_URL = OPENID_PROVIDER_URL + "/.well-known/openid-configuration"
oauth = OAuth()
oauth.register(
name="huggingface",
client_id=OAUTH_CLIENT_ID,
client_secret=OAUTH_CLIENT_SECRET,
client_kwargs={"scope": OAUTH_SCOPES},
server_metadata_url=METADATA_URL,
)
async def landing(request: Request):
if request.session.get("user"):
return RedirectResponse("/gradio")
else:
return RedirectResponse(request.url_for("oauth_login"))
async def oauth_login(request: Request):
redirect_uri = request.url_for("oauth_redirect_callback")
return await oauth.huggingface.authorize_redirect(request, redirect_uri)
async def oauth_redirect_callback(request: Request):
token = await oauth.huggingface.authorize_access_token(request)
async with httpx.AsyncClient() as client:
resp = await client.get(USER_INFO_URL, headers={"Authorization": f"Bearer {token['access_token']}"})
user_info = resp.json()
request.session["user"] = user_info # TODO: we should store token instead
return RedirectResponse(request.url_for("landing"))
async def check_oauth(request: Request, call_next):
if request.url.path.startswith("/gradio") and not request.session.get("user"): # protected route but not authenticated
return RedirectResponse("/")
return await call_next(request)
def get_app() -> FastAPI:
app = FastAPI()
app.middleware("http")(check_oauth)
app.add_middleware(SessionMiddleware, secret_key="session-secret-key") # TODO: make this is secret key
app.get("/")(landing)
app.get("/auth/huggingface")(oauth_login)
app.get("/auth/callback")(oauth_redirect_callback)
return app
|