""" # Summary 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 collections.abc import Sequence import google.auth.exceptions from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow 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. Notes: **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. **Guarantees:** - 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 (str): Path to the Google OAuth client secrets file used to initiate the OAuth 2.0 flow. store (CredentialStore[Credentials]): Credential store responsible for persisting and retrieving Google OAuth credentials. scopes (Sequence[str]): 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. Returns: Credentials: 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. Notes: **Lifecycle:** - Load cached credentials from the configured credential store. - Refresh expired credentials when possible. - Perform an interactive OAuth login as a fallback. - Persist valid credentials for future use. """ 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