# 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"), }, }