Spaces:
Sleeping
Sleeping
File size: 3,723 Bytes
d705087 5b40291 d705087 5b40291 d705087 5b40291 8adbc1e 5b40291 8adbc1e 5b40291 8adbc1e 5b40291 8adbc1e 5b40291 d705087 5b40291 d705087 5b40291 d705087 8adbc1e d705087 5b40291 d705087 5b40291 d705087 5b40291 8adbc1e d705087 | 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 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | # auth.py
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse
from google_auth_oauthlib.flow import Flow
from dotenv import load_dotenv
import os
load_dotenv()
router = APIRouter()
CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID")
CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET")
ENV_REDIRECT_URI = os.getenv("REDIRECT_URI") # optional
SCOPES = ["https://www.googleapis.com/auth/calendar"]
def build_redirect_uri(request: Request) -> str:
if ENV_REDIRECT_URI:
return ENV_REDIRECT_URI.strip().rstrip("/")
proto = request.headers.get("x-forwarded-proto") or request.url.scheme
host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.hostname
return f"{proto}://{host}/auth/callback".rstrip("/")
def build_flow(redirect_uri: str) -> Flow:
return Flow.from_client_config(
{
"web": {
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
"redirect_uris": [redirect_uri],
}
},
scopes=SCOPES,
)
@router.get("/auth/login")
def login(request: Request):
redirect_uri = build_redirect_uri(request)
flow = build_flow(redirect_uri)
flow.redirect_uri = redirect_uri
auth_url, _ = flow.authorization_url(
prompt="consent",
include_granted_scopes="true",
access_type="offline",
)
return RedirectResponse(auth_url)
# accept GET/POST and trailing slash
@router.api_route("/auth/callback", methods=["GET", "POST"])
@router.api_route("/auth/callback/", methods=["GET", "POST"])
async def auth_callback(request: Request):
# try query first
code = request.query_params.get("code")
# also accept POST (some proxies can end up posting back)
if not code and request.method == "POST":
try:
form = await request.form()
code = form.get("code")
except Exception:
code = None
redirect_uri = build_redirect_uri(request)
flow = build_flow(redirect_uri)
flow.redirect_uri = redirect_uri
if not code:
# Return diagnostics so we can see what arrived
return JSONResponse(
status_code=400,
content={
"detail": "Missing code",
"method": request.method,
"url": str(request.url),
"query": dict(request.query_params),
"headers": {
"x-forwarded-proto": request.headers.get("x-forwarded-proto"),
"x-forwarded-host": request.headers.get("x-forwarded-host"),
"host": request.headers.get("host"),
},
"derived_redirect_uri": redirect_uri,
},
)
try:
flow.fetch_token(code=code)
cred = flow.credentials
return {
"access_token": cred.token,
"refresh_token": cred.refresh_token,
"expiry": cred.expiry.isoformat() if cred.expiry else None,
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch token: {e}")
@router.get("/auth/debug")
def auth_debug(request: Request):
return {
"derived_redirect_uri": build_redirect_uri(request),
"request_url": str(request.url),
"headers": {
"x-forwarded-proto": request.headers.get("x-forwarded-proto"),
"x-forwarded-host": request.headers.get("x-forwarded-host"),
"host": request.headers.get("host"),
},
}
|