Spaces:
Paused
Paused
| """ | |
| Google OAuth Authentication | |
| """ | |
| from fastapi import Request, HTTPException | |
| from fastapi.responses import RedirectResponse | |
| from typing import Optional | |
| import httpx | |
| import json | |
| import secrets | |
| from datetime import datetime, timedelta | |
| class GoogleOAuth: | |
| """Google OAuth 2.0 authentication handler.""" | |
| def __init__(self, client_id: str, client_secret: str, redirect_uri: str): | |
| self.client_id = client_id | |
| self.client_secret = client_secret | |
| self.redirect_uri = redirect_uri | |
| self.auth_url = "https://accounts.google.com/o/oauth2/v2/auth" | |
| self.token_url = "https://oauth2.googleapis.com/token" | |
| self.userinfo_url = "https://www.googleapis.com/oauth2/v2/userinfo" | |
| self._state_store: dict = {} # state -> timestamp | |
| def get_auth_url(self, state: Optional[str] = None) -> str: | |
| """Generate Google OAuth authorization URL.""" | |
| if not state: | |
| state = secrets.token_urlsafe(32) | |
| self._state_store[state] = datetime.utcnow() | |
| params = { | |
| "client_id": self.client_id, | |
| "redirect_uri": self.redirect_uri, | |
| "response_type": "code", | |
| "scope": "openid email profile", | |
| "state": state, | |
| "access_type": "offline", | |
| "prompt": "consent" | |
| } | |
| return f"{self.auth_url}?{'&'.join(f'{k}={v}' for k, v in params.items())}" | |
| def verify_state(self, state: str) -> bool: | |
| """Verify OAuth state parameter to prevent CSRF.""" | |
| if state not in self._state_store: | |
| return False | |
| # State expires after 10 minutes | |
| if datetime.utcnow() - self._state_store[state] > timedelta(minutes=10): | |
| del self._state_store[state] | |
| return False | |
| del self._state_store[state] | |
| return True | |
| async def exchange_code_for_token(self, code: str) -> dict: | |
| """Exchange authorization code for access token.""" | |
| data = { | |
| "code": code, | |
| "client_id": self.client_id, | |
| "client_secret": self.client_secret, | |
| "redirect_uri": self.redirect_uri, | |
| "grant_type": "authorization_code" | |
| } | |
| async with httpx.AsyncClient() as client: | |
| response = await client.post(self.token_url, data=data) | |
| response.raise_for_status() | |
| return response.json() | |
| async def get_user_info(self, access_token: str) -> dict: | |
| """Get user information from Google.""" | |
| async with httpx.AsyncClient() as client: | |
| response = await client.get( | |
| self.userinfo_url, | |
| headers={"Authorization": f"Bearer {access_token}"} | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| # Global OAuth instance | |
| _google_oauth: Optional[GoogleOAuth] = None | |
| def get_google_oauth() -> GoogleOAuth: | |
| """Get global Google OAuth instance.""" | |
| global _google_oauth | |
| if _google_oauth is None: | |
| import os | |
| _google_oauth = GoogleOAuth( | |
| client_id=os.getenv("GOOGLE_CLIENT_ID", ""), | |
| client_secret=os.getenv("GOOGLE_CLIENT_SECRET", ""), | |
| redirect_uri=os.getenv("GOOGLE_REDIRECT_URI", "http://localhost:7860/auth/callback") | |
| ) | |
| return _google_oauth | |