"""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. Two independent authentication modes are supported: - OAuth 2.0 (auth_provider="google"): API send (users.messages.send) or SMTP with XOAUTH2 bearer tokens. Requires a Google Cloud OAuth client configuration. - Plain SMTP app-password (auth_provider="smtp_password"): logs into smtp.gmail.com with AUTH LOGIN using the Gmail address + a 16-character Google "app password" the user generated. No OAuth client needed. This is the "Bulk Add via SMTP" account type. """ def __init__(self, account_email: str, access_token: str = None, refresh_token: str = None, client_config: dict = None, sending_method: str = "api", token_expiry=None, smtp_password: str = None): self.account_email = account_email self.client_config = client_config self.smtp_password = smtp_password # Plain app-password accounts can only ever send via SMTP (there is no # OAuth token to call the Gmail API with). if smtp_password: self.method = "smtp" else: self.method = sending_method if sending_method in ("api", "smtp") else "api" self._creds = None if not smtp_password: creds = gcred.Credentials( token=access_token, refresh_token=refresh_token or None, token_uri=GOOGLE_TOKEN_URL, client_id=(client_config or {}).get("client_id"), client_secret=(client_config or {}).get("client_secret"), scopes=GMAIL_SCOPES, ) self._creds = creds self._service = None self.token_expiry = token_expiry def _ensure_valid(self): if self._creds is None: return None # app-password mode; nothing to refresh request = google.auth.transport.requests.Request() if not self._creds.valid: self._creds.refresh(request) return self._creds.token def _api(self): if self.smtp_password: raise GmailOauthError( "This account uses a plain SMTP app-password and has no OAuth " "token, so it cannot use the Gmail API. Use SMTP sending instead." ) 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 ------------------------------------------------------------- # Two auth modes on the same wire protocol (smtp.gmail.com:587, STARTTLS): # 1) XOAUTH2 with an OAuth bearer token (accounts connected via Google # OAuth sign-in with sending_method="smtp"). # 2) Plain AUTH LOGIN/PLAIN with the Gmail address + a 16-character # Google "app password" (accounts added via "Bulk Add via SMTP"). # Both paths send real messages through Google's real SMTP servers - no # simulated or "fake success" delivery. def send_mime_smtp(self, mime_message) -> dict: import smtplib from email.utils import parseaddr try: with smtplib.SMTP("smtp.gmail.com", 587, timeout=30) as smtp: smtp.ehlo() smtp.starttls(context=__import__("ssl").create_default_context()) smtp.ehlo() if self.smtp_password: # Plain app-password login (AUTH LOGIN/PLAIN via smtplib.login) smtp.login(self.account_email, self.smtp_password) auth_note = "AUTH LOGIN (app password)" else: # OAuth XOAUTH2 bearer-token login self._ensure_valid() access_token = self._creds.token auth_string = f"user={self.account_email}\x01auth=Bearer {access_token}\x01\x01" code, response = smtp.docmd( "AUTH", "XOAUTH2 " + base64.b64encode(auth_string.encode()).decode()) if code >= 400: return {"ok": False, "error": f"SMTP XOAUTH2 auth failed: {code} {response.decode(errors='replace')}", "error_class": "permanent"} auth_note = f"XOAUTH2 ({code} {response.decode(errors='replace')})" refused = smtp.sendmail(self.account_email, [parseaddr(mime_message["To"])[1]], mime_message.as_bytes()) except smtplib.SMTPAuthenticationError as e: return {"ok": False, "error": f"SMTP authentication failed: {e}", "error_class": "permanent"} except smtplib.SMTPException as e: return {"ok": False, "error": f"SMTP error: {e}", "error_class": "temporary"} 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 via {auth_note}"} # -- 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_smtp_login(self) -> dict: """Real connectivity test for app-password accounts: perform the actual STARTTLS + AUTH LOGIN handshake against smtp.gmail.com without sending a message. Returns ok=True only if Google actually accepts the login.""" import smtplib try: with smtplib.SMTP("smtp.gmail.com", 587, timeout=20) as smtp: smtp.ehlo() smtp.starttls(context=__import__("ssl").create_default_context()) smtp.ehlo() smtp.login(self.account_email, self.smtp_password) return {"ok": True, "email_address": self.account_email, "detail": "SMTP login verified (smtp.gmail.com:587, STARTTLS + AUTH LOGIN)."} except smtplib.SMTPAuthenticationError as e: return {"ok": False, "detail": f"SMTP authentication failed: {e}"} except Exception as e: return {"ok": False, "detail": str(e)[:400]} def verify_connection(self) -> dict: if self.smtp_password: return self.verify_smtp_login() 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=None, client_config=None) -> GmailClient: """Build a GmailClient from DB rows (sync session context - worker uses sync ORM). Branches on account.auth_provider: - "smtp_password": no OAuth token needed; uses the encrypted app password. - "google" (default): requires an oauth_token (API or XOAUTH2 SMTP). """ if getattr(account, "auth_provider", "google") == "smtp_password": return GmailClient(account_email=account.email, smtp_password=account.smtp_password) if oauth_token is None: raise GmailOauthError(f"No OAuth token stored for account {account.email}") 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, ) # --------------------------------------------------------------------------- # Bulk SMTP (app-password) account helpers # --------------------------------------------------------------------------- _APP_PASSWORD_RE = __import__("re").compile(r"^[A-Za-z]{4}(\s?[A-Za-z]{4}){3}$") def normalize_app_password(raw: str) -> str: """Google app passwords are displayed as 'xxxx xxxx xxxx xxxx' (16 letters, optional spaces). Google's SMTP server accepts them with or without spaces, but we normalize (strip spaces, lowercase is NOT required) for consistent storage/comparison.""" return (raw or "").strip().replace(" ", "") def validate_smtp_credentials_syntax(email: str, app_password: str) -> str | None: """Cheap local syntax validation before attempting a real network login. Returns an error string, or None if syntax looks OK.""" if not email or "@" not in email: return f"Invalid email address: {email!r}" pw = normalize_app_password(app_password) if len(pw) != 16 or not pw.isalpha(): return ("App password must be 16 letters (Google app passwords look like " "'abcd efgh ijkl mnop'), got a value with " f"{len(pw)} character(s) after removing spaces") return None def verify_smtp_login(email: str, app_password: str) -> dict: """Stateless real SMTP login test (used during bulk-add, before persisting).""" client = GmailClient(account_email=email, smtp_password=normalize_app_password(app_password)) return client.verify_smtp_login() 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)