""" 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 - Coordination with a credential persistence layer No Google authentication details should leak outside this module. """ import os from typing import Sequence import google.auth.exceptions from google.auth.transport.requests import Request from google_auth_oauthlib.flow import InstalledAppFlow from google.oauth2.credentials import Credentials from mail_intake.auth.base import MailIntakeAuthProvider from mail_intake.credentials.store import CredentialStore 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 a credential store when available - Refresh expired credentials when possible - Initiate an interactive OAuth flow only when required - Persist refreshed or newly obtained credentials via the store This class is synchronous by design and maintains a minimal internal state. """ def __init__( self, credentials_path: str, store: CredentialStore[Credentials], scopes: Sequence[str], ): """ Initialize the Google authentication provider. Args: credentials_path: Path to the Google OAuth client secrets file used to initiate the OAuth 2.0 flow. store: Credential store responsible for persisting and retrieving Google OAuth credentials. scopes: OAuth scopes required for Gmail access. """ self.credentials_path = credentials_path self.store = store self.scopes = list(scopes) def get_credentials(self) -> Credentials: """ Retrieve valid Google OAuth credentials. This method attempts to: 1. Load cached credentials from the configured credential store 2. Refresh expired credentials when possible 3. Perform an interactive OAuth login as a fallback 4. Persist valid credentials for future use Returns: A ``google.oauth2.credentials.Credentials`` instance suitable for use with Google API clients. Raises: MailIntakeAuthError: If credentials cannot be loaded, refreshed, or obtained via interactive authentication. """ creds = self.store.load() # 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: self.store.clear() 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 or newly obtained credentials try: self.store.save(creds) except Exception as exc: raise MailIntakeAuthError( "Failed to persist Google OAuth credentials" ) from exc return creds