""" Google authentication provider implementation for Mail Intake. This module provides a **Google OAuth–based authentication provider** used primarily for Gmail access. It encapsulates all Google-specific authentication concerns, including: - Credential loading and persistence - Token refresh handling - Interactive OAuth flow initiation No Google authentication details should leak outside this module. """ import os import pickle from typing import Sequence import google.auth.exceptions from google.auth.transport.requests import Request from google_auth_oauthlib.flow import InstalledAppFlow from mail_intake.auth.base import MailIntakeAuthProvider from mail_intake.exceptions import MailIntakeAuthError class MailIntakeGoogleAuth(MailIntakeAuthProvider): """ Google OAuth provider for Gmail access. This provider implements the `MailIntakeAuthProvider` interface using Google's OAuth 2.0 flow and credential management libraries. Responsibilities: - Load cached credentials from disk when available - Refresh expired credentials when possible - Initiate an interactive OAuth flow only when required - Persist refreshed or newly obtained credentials This class is synchronous by design and maintains a minimal internal state. """ def __init__( self, credentials_path: str, token_path: str, scopes: Sequence[str], ): """ Initialize the Google authentication provider. Args: credentials_path: Path to the Google OAuth client secrets file. token_path: Path where OAuth tokens will be cached. scopes: OAuth scopes required for access. """ self.credentials_path = credentials_path self.token_path = token_path self.scopes = list(scopes) def get_credentials(self): """ Retrieve valid Google OAuth credentials. This method attempts to: 1. Load cached credentials from disk 2. Refresh expired credentials when possible 3. Perform an interactive OAuth login as a fallback 4. Persist valid credentials for future use Returns: Google OAuth credentials object suitable for use with Google API clients. Raises: MailIntakeAuthError: If credentials cannot be loaded, refreshed, or obtained via interactive authentication. """ creds = None # Attempt to load cached credentials if os.path.exists(self.token_path): try: with open(self.token_path, "rb") as fh: creds = pickle.load(fh) except Exception: creds = None # Validate / refresh credentials if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: try: creds.refresh(Request()) except google.auth.exceptions.RefreshError: creds = None # Interactive login if refresh failed or creds missing if not creds: if not os.path.exists(self.credentials_path): raise MailIntakeAuthError( f"Google credentials file not found: {self.credentials_path}" ) try: flow = InstalledAppFlow.from_client_secrets_file( self.credentials_path, self.scopes, ) creds = flow.run_local_server(port=0) except Exception as exc: raise MailIntakeAuthError( "Failed to complete Google OAuth flow" ) from exc # Persist refreshed / new credentials try: with open(self.token_path, "wb") as fh: pickle.dump(creds, fh) except Exception as exc: raise MailIntakeAuthError( f"Failed to write token file: {self.token_path}" ) from exc return creds