Seth
Claude Sonnet 5
Unify Google OAuth callback: Drive-connect reuses the login redirect_uri
c7a1227 | """Shared helpers for the Google OAuth flow, used by both login (auth_routes.py) and the | |
| Drive-connect flow (chat_agent_drive.py). Both flows funnel through the SAME registered | |
| /api/auth/google/callback redirect_uri (distinguished by which state is pending in the | |
| session) so a Google Cloud Console project only ever needs one authorized redirect URI. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from typing import Any, Dict | |
| import httpx | |
| from fastapi import HTTPException, Request | |
| GOOGLE_TOKEN_URI = "https://oauth2.googleapis.com/token" | |
| def normalize_https_public_origin(origin: str) -> str: | |
| """Spaces like Hugging Face are always HTTPS; fix http:// saved by mistake or bad defaults.""" | |
| o = origin.strip().rstrip("/") | |
| if o.startswith("http://") and ".hf.space" in o: | |
| return "https://" + o[7:] | |
| return o | |
| def public_request_base(request: Request) -> str: | |
| """ | |
| Browser-facing origin (scheme + host, no path). Uses proxy headers so OAuth redirect_uri | |
| is https://... behind Hugging Face / nginx (request.base_url alone is often http:// internally). | |
| """ | |
| fwd_proto = request.headers.get("x-forwarded-proto") | |
| if fwd_proto: | |
| proto = fwd_proto.split(",")[0].strip().lower() | |
| else: | |
| proto = (request.url.scheme or "http").lower() | |
| fwd_host = request.headers.get("x-forwarded-host") | |
| if fwd_host: | |
| host = fwd_host.split(",")[0].strip() | |
| else: | |
| host = request.headers.get("host") or request.url.netloc or "" | |
| if host: | |
| base = f"{proto}://{host}" if proto else f"https://{host}" | |
| base = normalize_https_public_origin(base) | |
| return base.rstrip("/") | |
| base = str(request.base_url).rstrip("/") | |
| return normalize_https_public_origin(base) | |
| def google_client_configured() -> bool: | |
| return bool( | |
| os.environ.get("GOOGLE_CLIENT_ID", "").strip() and os.environ.get("GOOGLE_CLIENT_SECRET", "").strip() | |
| ) | |
| def google_callback_redirect_uri(request: Request) -> str: | |
| """The single redirect_uri registered in Google Cloud Console, shared by login and | |
| Drive-connect — set GOOGLE_REDIRECT_URI or FRONTEND_ORIGIN to pin it explicitly.""" | |
| explicit = os.environ.get("GOOGLE_REDIRECT_URI", "").strip() | |
| if explicit: | |
| return normalize_https_public_origin(explicit) | |
| fe = os.environ.get("FRONTEND_ORIGIN", "").strip() | |
| if fe: | |
| return normalize_https_public_origin(fe) + "/api/auth/google/callback" | |
| return public_request_base(request) + "/api/auth/google/callback" | |
| async def exchange_code_for_tokens(code: str, redirect_uri: str) -> Dict[str, Any]: | |
| """Exchange an OAuth authorization code for tokens (shared by login and Drive-connect).""" | |
| client_id = os.environ["GOOGLE_CLIENT_ID"].strip() | |
| client_secret = os.environ["GOOGLE_CLIENT_SECRET"].strip() | |
| async with httpx.AsyncClient() as client: | |
| token_res = await client.post( | |
| GOOGLE_TOKEN_URI, | |
| data={ | |
| "code": code, | |
| "client_id": client_id, | |
| "client_secret": client_secret, | |
| "redirect_uri": redirect_uri, | |
| "grant_type": "authorization_code", | |
| }, | |
| headers={"Content-Type": "application/x-www-form-urlencoded"}, | |
| ) | |
| if token_res.status_code != 200: | |
| raise HTTPException(status_code=502, detail="Token exchange failed") | |
| return token_res.json() | |