82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
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.
|
|
|
|
Responsibilities:
|
|
- Load cached credentials from disk
|
|
- Refresh expired tokens when possible
|
|
- Trigger interactive login only when strictly required
|
|
|
|
This class is synchronous and intentionally state-light.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
credentials_path: str,
|
|
token_path: str,
|
|
scopes: Sequence[str],
|
|
):
|
|
self.credentials_path = credentials_path
|
|
self.token_path = token_path
|
|
self.scopes = list(scopes)
|
|
|
|
def get_credentials(self):
|
|
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
|