"""Real Google OAuth 2.0 + Gmail API integration. - Authorization code flow with state verification (CSRF protection on the OAuth handshake). - Tokens stored encrypted (Fernet) via app.core.encryption. - Refresh handled transparently; never logs secrets. - Sending via Gmail API (users.messages.send) or Gmail SMTP with XOAUTH2 (no passwords ever). """ import base64 import json import logging import secrets from datetime import datetime, timedelta, timezone from email.mime.image import MIMEImage from email.utils import formatdate from urllib.parse import urlencode import google.auth.transport.requests import googleapiclient.discovery import googleapiclient.errors import requests from google.oauth2 import credentials as gcred from google_auth_oauthlib.flow import Flow from googleapiclient.discovery import build from googleapiclient.http import MediaIoBaseUpload from app.core import encryption from app.core.config import settings log = logging.getLogger("gmail") # Minimum scope required to send mail through the user's own account. GMAIL_SCOPES = [ "https://www.googleapis.com/auth/gmail.send", "https://www.googleapis.com/auth/gmail.compose", ] USERINFO_SCOPE = "openid https://www.googleapis.com/auth/userinfo.email" ALL_SCOPES = GMAIL_SCOPES + [USERINFO_SCOPE] GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth" GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" GOOGLE_REVOKE_URL = "https://oauth2.googleapis.com/revoke" GOOGLE_USERINFO_URL = "https://openidconnect.googleapis.com/v1/userinfo" class GmailOauthError(Exception): pass # --------------------------------------------------------------------------- # OAuth client configuration resolution # --------------------------------------------------------------------------- def resolve_client_config(config=None) -> dict: """Return {client_id, client_secret, redirect_uri} for OAuth flows. Priority: explicit config dict (per-user imported config) > DB row > env file path > error. """ if config and config.client_id: return { "client_id": config.client_id, "client_secret": config.client_secret, # decrypted property "redirect_uri": config.redirect_uri or f"{settings.BASE_URL}/auth/google/callback", } path = settings.GOOGLE_OAUTH_CLIENT_JSON_PATH if path and __import__("os").path.exists(path): with open(path) as f: data = json.load(f) inst = data.get("installed") or data.get("web") if not inst: raise GmailOauthError("Invalid Google OAuth client JSON: no 'installed' or 'web' key") return { "client_id": inst["client_id"], "client_secret": inst["client_secret"], "redirect_uri": f"{settings.BASE_URL}/auth/google/callback", } raise GmailOauthError( "No Google OAuth client configuration available. Import a client_secret.json in " "Gmail Accounts → Import Gmail Configuration, or set GOOGLE_OAUTH_CLIENT_JSON_PATH." ) def parse_client_json(content: bytes) -> dict: """Validate an uploaded client_secret JSON. Returns normalized dict; raises on invalid.""" try: data = json.loads(content.decode("utf-8")) except Exception as e: raise GmailOauthError(f"File is not valid JSON: {e}") inst = data.get("installed") or data.get("web") if not inst or not inst.get("client_id") or not inst.get("client_secret"): raise GmailOauthError("JSON must contain an 'installed' or 'web' object with client_id and client_secret") return { "project_id": inst.get("project", "") or inst.get("project_id", ""), "client_id": inst["client_id"], "client_secret": inst["client_secret"], "redirect_uris": inst.get("redirect_uris", []), "auth_uri": inst.get("auth_uri", ""), "token_uri": inst.get("token_uri", ""), } # --------------------------------------------------------------------------- # Authorization flow # --------------------------------------------------------------------------- def generate_state() -> str: return secrets.token_urlsafe(32) def build_authorization_url(client: dict, state: str) -> str: params = { "client_id": client["client_id"], "redirect_uri": client["redirect_uri"], "response_type": "code", "scope": " ".join(ALL_SCOPES), "access_type": "offline", # needed to obtain a refresh token "prompt": "consent", # ensure refresh token is granted "include_granted_scopes": "true", "state": state, } return f"{GOOGLE_AUTH_URL}?{urlencode(params)}" def exchange_code_for_tokens(client: dict, code: str) -> dict: """Exchange authorization code for tokens. Real HTTP call to Google.""" data = { "code": code, "client_id": client["client_id"], "client_secret": client["client_secret"], "redirect_uri": client["redirect_uri"], "grant_type": "authorization_code", } resp = requests.post(GOOGLE_TOKEN_URL, data=data, timeout=20) if resp.status_code != 200: raise GmailOauthError(f"Token exchange failed: {resp.status_code} {resp.text[:300]}") tok = resp.json() if "access_token" not in tok: raise GmailOauthError("Token response missing access_token") return tok def refresh_access_token(client: dict, refresh_token: str) -> dict: data = { "client_id": client["client_id"], "client_secret": client["client_secret"], "refresh_token": refresh_token, "grant_type": "refresh_token", } resp = requests.post(GOOGLE_TOKEN_URL, data=data, timeout=20) if resp.status_code != 200: raise GmailOauthError(f"Token refresh failed: {resp.status_code} {resp.text[:300]}") return resp.json() def revoke_token(token: str) -> bool: """Revoke a token at Google. Returns True if revoked successfully.""" if not token: return False try: resp = requests.post(GOOGLE_REVOKE_URL, params={"token": token}, timeout=15) return resp.status_code == 200 except Exception: return False def get_userinfo(access_token: str) -> dict: resp = requests.get(GOOGLE_USERINFO_URL, headers={"Authorization": f"Bearer {access_token}"}, timeout=15) if resp.status_code != 200: raise GmailOauthError(f"userinfo failed: {resp.status_code}") return resp.json() # --------------------------------------------------------------------------- # Gmail client construction # --------------------------------------------------------------------------- class GmailClient: """Authenticated Gmail client for one account (API or XOAUTH2 SMTP).""" def __init__(self, account_email: str, access_token: str, refresh_token: str, client_config: dict, sending_method: str = "api", token_expiry=None): self.account_email = account_email self.client_config = client_config self.method = sending_method if sending_method in ("api", "smtp") else "api" creds = gcred.Credentials( token=access_token, refresh_token=refresh_token or None, token_uri=GOOGLE_TOKEN_URL, client_id=client_config["client_id"], client_secret=client_config["client_secret"], scopes=GMAIL_SCOPES, ) self._creds = creds self._service = None self.token_expiry = token_expiry def _ensure_valid(self): request = google.auth.transport.requests.Request() if not self._creds.valid: self._creds.refresh(request) return self._creds.token def _api(self): if self._service is None: self._ensure_valid() self._service = build("gmail", "v1", credentials=self._creds, cache_discovery=False) else: self._ensure_valid() return self._service # -- API send ------------------------------------------------------------ def send_mime_api(self, mime_message) -> dict: """Send via Gmail API users.messages.send. Returns real provider response.""" service = self._api() raw = base64.urlsafe_b64encode(mime_message.as_bytes()).decode() body = {"raw": raw} sent = service.users().messages().send(userId="me", body=body).execute() return {"ok": True, "message_id": sent.get("id", ""), "thread_id": sent.get("threadId", ""), "label_ids": sent.get("labelIds", []), "raw_response": sent} # -- SMTP send (XOAUTH2, no password) ------------------------------------- def send_mime_smtp(self, mime_message) -> dict: import smtplib from email.utils import parseaddr self._ensure_valid() access_token = self._creds.token auth_string = f"user={self.account_email}\x01auth=Bearer {access_token}\x01\x01" with smtplib.SMTP("smtp.gmail.com", 587, timeout=30) as smtp: smtp.ehlo() smtp.starttls(context=__import__("ssl").create_default_context()) smtp.ehlo() code, response = smtp.docmd("AUTH", "XOAUTH2 " + base64.b64encode(auth_string.encode()).decode()) refused = smtp.sendmail(self.account_email, [parseaddr(mime_message["To"])[1]], mime_message.as_bytes()) if refused: return {"ok": False, "error": f"SMTP refused recipients: {refused}", "error_class": "permanent"} return {"ok": True, "message_id": "", "thread_id": "", "raw_response": f"250 SMTP accepted ({code} {response.decode(errors='replace')})"} # -- Health / profile ------------------------------------------------------ def get_profile(self) -> dict: service = self._api() prof = service.users().getProfile(userId="me").execute() return {"email_address": prof.get("emailAddress"), "history_id": prof.get("historyId"), "messages_total": prof.get("messagesTotal"), "threads_total": prof.get("threadsTotal")} def verify_connection(self) -> dict: try: prof = self.get_profile() return {"ok": True, "email_address": prof.get("email_address"), "detail": "Gmail API connection verified."} except Exception as e: return {"ok": False, "detail": str(e)[:400]} def build_gmail_client_from_account(account, oauth_token, client_config=None) -> GmailClient: """Build a GmailClient from DB rows (sync session context - worker uses sync ORM).""" cfg = client_config or resolve_client_config_for_account(account) return GmailClient( account_email=account.email, access_token=oauth_token.access_token, # decrypted refresh_token=oauth_token.refresh_token, # decrypted client_config=cfg, sending_method=account.sending_method, token_expiry=oauth_token.expires_at, ) def resolve_client_config_for_account(account) -> dict: """Resolve client config for an account: its linked config row, the owner's latest imported active config, or the server default (env file path).""" # Called from sync or async context; imports DB lazily. from app.db.session import SyncSession from app.models import OAuthClientConfig from sqlalchemy import select as _select if account.oauth_config_id: with SyncSession() as s: cfg = s.get(OAuthClientConfig, account.oauth_config_id) if cfg and cfg.is_active: return {"client_id": cfg.client_id, "client_secret": cfg.client_secret, "redirect_uri": cfg.redirect_uri or f"{settings.BASE_URL}/auth/google/callback"} # Fall back: latest active config imported by the account's owner # (covers accounts created before the config import, seeded rows, etc.) with SyncSession() as s: cfg = (s.execute( _select(OAuthClientConfig) .where(OAuthClientConfig.user_id == account.user_id, OAuthClientConfig.is_active == True) # noqa: E712 .order_by(OAuthClientConfig.id.desc()))).scalars().first() if cfg: return {"client_id": cfg.client_id, "client_secret": cfg.client_secret, "redirect_uri": cfg.redirect_uri or f"{settings.BASE_URL}/auth/google/callback"} return resolve_client_config(None)