import os import logging from typing import Optional, Dict, Any from google.auth.transport import requests as google_requests from google.oauth2 import id_token from google.auth.exceptions import GoogleAuthError logger = logging.getLogger(__name__) # Google OAuth configuration GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID") GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET") if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET: logger.warning("Google OAuth credentials not configured") class GoogleOAuthService: @staticmethod async def verify_google_token(token: str) -> Optional[Dict[str, Any]]: """ Verify Google ID token and return user information """ if not GOOGLE_CLIENT_ID: logger.error("Google Client ID not configured") return None try: # Verify the token id_info = id_token.verify_oauth2_token( token, google_requests.Request(), GOOGLE_CLIENT_ID ) # Verify that the token is issued by Google if id_info['iss'] not in ['accounts.google.com', 'https://accounts.google.com']: logger.error("Invalid token issuer") return None # Extract user information user_info = { 'google_id': id_info.get('sub'), 'email': id_info.get('email'), 'email_verified': id_info.get('email_verified', False), 'first_name': id_info.get('given_name'), 'last_name': id_info.get('family_name'), 'full_name': id_info.get('name'), 'picture': id_info.get('picture'), 'locale': id_info.get('locale', 'en') } logger.info(f"Successfully verified Google token for user: {user_info['email']}") return user_info except ValueError as e: logger.error(f"Invalid Google token: {e}") return None except GoogleAuthError as e: logger.error(f"Google auth error: {e}") return None except Exception as e: logger.error(f"Unexpected error verifying Google token: {e}") return None @staticmethod def get_google_oauth_url() -> Optional[str]: """ Generate Google OAuth URL for frontend to redirect users """ if not GOOGLE_CLIENT_ID: return None redirect_uri = os.getenv("GOOGLE_REDIRECT_URI", "http://localhost:3000/auth/google/callback") oauth_url = ( f"https://accounts.google.com/o/oauth2/auth?" f"client_id={GOOGLE_CLIENT_ID}&" f"redirect_uri={redirect_uri}&" f"scope=openid email profile&" f"response_type=code&" f"access_type=offline&" f"prompt=consent" ) return oauth_url @staticmethod async def exchange_code_for_token(code: str, redirect_uri: str) -> Optional[str]: """ Exchange authorization code for ID token """ if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET: logger.error("Google OAuth credentials not configured") return None try: import requests token_url = "https://oauth2.googleapis.com/token" data = { 'client_id': GOOGLE_CLIENT_ID, 'client_secret': GOOGLE_CLIENT_SECRET, 'code': code, 'grant_type': 'authorization_code', 'redirect_uri': redirect_uri } response = requests.post(token_url, data=data) if response.status_code == 200: token_data = response.json() return token_data.get('id_token') else: logger.error(f"Failed to exchange code for token: {response.text}") return None except Exception as e: logger.error(f"Error exchanging code for token: {e}") return None