""" Google OAuth 2.0 client wrapper for authentication. Handles Google login flow and token management. """ import os from flask import session, url_for, current_app from google.oauth2 import credentials as Credentials from google_auth_oauthlib.flow import Flow from googleapiclient.discovery import build import json import requests class GoogleOAuth: """Google OAuth client for user authentication.""" def __init__(self): self.client_id = os.getenv("GOOGLE_CLIENT_ID") self.client_secret = os.getenv("GOOGLE_CLIENT_SECRET") # Build redirect URI manually instead of using url_for # This avoids requiring Flask context self.redirect_uri = os.getenv( "GOOGLE_REDIRECT_URI", "https://jadwaai.com/auth/google/callback" ) def get_authorization_url(self): """Generate Google OAuth authorization URL.""" flow = Flow.from_client_config( client_config={ "web": { "client_id": self.client_id, "client_secret": self.client_secret, "auth_uri": "https://accounts.google.com/o/oauth2/v2/auth", "token_uri": "https://oauth2.googleapis.com/token", } }, scopes=[ "openid", "email", "profile", ] ) flow.redirect_uri = self.redirect_uri # Generate authorization URL url, state = flow.authorization_url( access_type="offline", include_granted_scopes="true", prompt="consent" ) # Store the flow in session for later use in fetch_token # We need to serialize the flow's verifier for PKCE session["oauth_state"] = state session["oauth_code_verifier"] = flow.code_verifier return url def fetch_token(self, authorization_response): """Exchange authorization code for tokens using PKCE.""" current_app.logger.info(f"fetch_token called with: {authorization_response[:200]}") # Parse the authorization response to get the code from urllib.parse import urlparse, parse_qs parsed = urlparse(authorization_response) params = parse_qs(parsed.query) code = params.get('code', [None])[0] if not code: current_app.logger.error("No authorization code in response") raise ValueError("No authorization code in response") # Get the code verifier from session (required for PKCE) code_verifier = session.pop("oauth_code_verifier", None) if not code_verifier: current_app.logger.error("No code verifier in session - PKCE flow incomplete") try: # Use requests directly to exchange code for tokens import requests token_url = "https://oauth2.googleapis.com/token" # Log the request details (without sensitive data) current_app.logger.info(f"Requesting token from {token_url}") current_app.logger.info(f"Client ID (first 20 chars): {self.client_id[:20] if self.client_id else 'MISSING'}...") current_app.logger.info(f"Redirect URI: {self.redirect_uri}") current_app.logger.info(f"Has code_verifier: {bool(code_verifier)}") data = { 'code': code, 'client_id': self.client_id, 'client_secret': self.client_secret, 'redirect_uri': self.redirect_uri, 'grant_type': 'authorization_code' } # Add code_verifier for PKCE if available if code_verifier: data['code_verifier'] = code_verifier response = requests.post(token_url, data=data) current_app.logger.info(f"Token response status: {response.status_code}") if response.status_code == 200: token_data = response.json() return { "token": token_data.get("access_token"), "refresh_token": token_data.get("refresh_token"), "id_token": token_data.get("id_token"), "client_id": self.client_id, "client_secret": self.client_secret, } else: current_app.logger.error(f"Token request failed: {response.status_code}") current_app.logger.error(f"Response body: {response.text}") # Try to parse error for better message try: error_json = response.json() error_desc = error_json.get('error_description', error_json.get('error', str(response.status_code))) raise ValueError(f"Token request failed: {error_desc}") except: raise ValueError(f"Token request failed: {response.status_code} - {response.text[:200]}") except Exception as e: current_app.logger.error(f"Error fetching token: {e}") import traceback current_app.logger.error(traceback.format_exc()) raise ValueError(f"Error fetching token: {e}") def get_user_info(self, credentials_dict): """Get user info from access token.""" try: # Use the access token to make authenticated requests access_token = credentials_dict.get("token") if not access_token: current_app.logger.error("No access token in credentials") raise ValueError("No access token available") current_app.logger.info(f"Fetching user info with access token (first 50 chars: {access_token[:50] if access_token else 'none'})") import requests headers = {"Authorization": f"Bearer {access_token}"} current_app.logger.info(f"Request headers: {headers}") user_info = requests.get( "https://www.googleapis.com/oauth2/v3/userinfo", headers=headers ) current_app.logger.info(f"User info response status: {user_info.status_code}") if user_info.status_code == 200: user_data = user_info.json() current_app.logger.info(f"User data parsed: email={user_data.get('email')}, name={user_data.get('name')}") return { "google_id": user_data.get("sub", ""), "email": user_data.get("email", ""), "name": user_data.get("name", ""), "verified_email": user_data.get("email_verified", False), "picture": user_data.get("picture", ""), } else: current_app.logger.error(f"Failed to get user info: {user_info.status_code} {user_info.text}") raise ValueError(f"Failed to get user info: {user_info.status_code} - {user_info.text}") except Exception as e: current_app.logger.error(f"Error fetching user info: {e}") import traceback current_app.logger.error(traceback.format_exc()) raise ValueError(f"Error fetching user info: {e}") def refresh_google_token(user): """Refresh Google OAuth token for a user.""" from google.oauth2 import credentials as Credentials import google_auth_httplib2 if not user.google_access_token or not user.google_refresh_token: return None try: credentials = Credentials( token=user.google_access_token, refresh_token=user.google_refresh_token, token_uri="https://oauth2.googleapis.com/token", client_id=current_app.config.get("GOOGLE_CLIENT_ID"), client_secret=current_app.config.get("GOOGLE_CLIENT_SECRET"), ) # Refresh if expired if credentials.expired: credentials.refresh(google_auth_httplib2.AuthorizedSession()) user.google_access_token = credentials.token # Update refresh token if a new one is provided if credentials.refresh_token: user.google_refresh_token = credentials.refresh_token current_app.extensions['db'].session.commit() return credentials.token except Exception as e: current_app.logger.error(f"Failed to refresh Google token: {e}") return None