Vishesh 'ironeagle' Bangotra
f22af90e98
- docs(mail_intake/__init__.py): document module-based public API and usage patterns - docs(mail_intake/ingestion/reader.py): document high-level ingestion orchestration - docs(mail_intake/adapters/base.py): document adapter contract for mail providers - docs(mail_intake/adapters/gmail.py): document Gmail adapter implementation and constraints - docs(mail_intake/auth/base.py): document authentication provider contract - docs(mail_intake/auth/google.py): document Google OAuth authentication provider - docs(mail_intake/models/message.py): document canonical email message model - docs(mail_intake/models/thread.py): document canonical email thread model - docs(mail_intake/parsers/body.py): document message body extraction logic - docs(mail_intake/parsers/headers.py): document message header normalization utilities - docs(mail_intake/parsers/subject.py): document subject normalization utilities - docs(mail_intake/config.py): document global configuration model - docs(mail_intake/exceptions.py): document library exception hierarchy
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""
|
||
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
|