Vishesh 'ironeagle' Bangotra
371b74806f
Squash-merge lib-standardisation: adopt repo standards across packaging, tooling, docs, and CI. - standardize pyproject.toml (canonical packaging, lint config, extras, GSDFC pydoclint) - add py.typed marker, modernize type hints and Google-style docstrings - nest generated docs under docs/lib and docs/mcp; add hand-written wiki - adopt standard Type A PyPI CI pipeline (tag==version guard + quality gate) - promote generated CHANGELOG.md; bump version to 0.0.3
136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
"""
|
||
# 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
|