scripts/gmail_backend.py — Approval-Gated Mail

#!/usr/bin/env python3
# Gmail backend for Approval-Gated Mail 0.2.1.
# By Zero Width Lab (zerowidthlab.com)
from __future__ import annotations

import base64
import hashlib
import hmac
import http.client
import json
import os
import re
import secrets
import socket
import stat
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from email import policy
from email.errors import MessageError
from email.message import EmailMessage
from email.utils import parsedate_to_datetime
from email.parser import BytesParser
from email.utils import getaddresses
from email.utils import make_msgid
from email.header import decode_header, make_header
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

__author__ = "Zero Width Lab"
SERVICE = "approval-gated-mail-gmail"
SCOPES = ("https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.compose")
API = "https://gmail.googleapis.com/gmail/v1/users/me"
TOKEN = "https://oauth2.googleapis.com/token"
REVOKE = "https://oauth2.googleapis.com/revoke"
HOSTS = {"gmail.googleapis.com", "oauth2.googleapis.com"}
GUIDANCE = "run auth login again; External apps in Testing expire every 7 days"


class RetryableRead(Exception):
  def __init__(self, status: int, delay: float) -> None:
    self.status = status
    self.delay = delay


def b64decode(value: str) -> bytes:
  return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def pkce_pair() -> tuple[str, str]:
  verifier = secrets.token_urlsafe(48)
  challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode("ascii")).digest()).rstrip(b"=").decode("ascii")
  return verifier, challenge


def quoted_query(value: str) -> str:
  if '"' in value or "\\" in value:
    raise ValueError('Gmail search text cannot contain a quote or backslash')
  return '"' + value + '"'


def checked_url(url: str, error_type: type[Exception]) -> None:
  parsed = urllib.parse.urlsplit(url)
  if parsed.scheme != "https" or parsed.hostname not in HOSTS or parsed.username is not None or parsed.password is not None:
    raise error_type("refusing network access outside approved Gmail hosts")


def retry_after_seconds(value: str | None, fallback: float) -> float:
  if not value:
    return fallback
  return min(int(value), 30) if value.isascii() and value.isdecimal() else fallback


def decoded_header(value: str) -> str:
  try:
    return str(make_header(decode_header(value)))
  except (UnicodeError, ValueError, LookupError, MessageError, RecursionError):
    return value


def validate_mailbox(value: str, label: str) -> str:
  rule = f"invalid {label} mailbox: encoded words, comments, group syntax, '@' in the display name and non-ASCII addresses are not accepted"
  if any(ord(char) < 32 or 0x7f <= ord(char) <= 0x9f for char in value) or any(char in value for char in (":", ";", "(", ")", "\\", "=?")):
    raise ValueError(rule)
  parsed = getaddresses([value])
  if len(parsed) != 1 or "@" in parsed[0][0] or not parsed[0][1].isascii() or parsed[0][1].count("@") != 1 or not all(parsed[0][1].split("@")) or any(char.isspace() for char in parsed[0][1]):
    raise ValueError(rule)
  return value


def validate_header(value: str, label: str) -> str:
  if any(ord(char) < 32 or 0x7f <= ord(char) <= 0x9f for char in value):
    raise ValueError(f"invalid {label} header")
  return value


def validate_message_id(value: str, strict: bool = False) -> str:
  pattern = r"<[A-Za-z0-9!#$%&'*+/=?^_`{|}~.-]+@[A-Za-z0-9.-]+>" if strict else r"<[^<>\s\x00-\x1f\x7f-\x9f@]+@[^<>\s\x00-\x1f\x7f-\x9f@]+>"
  if not re.fullmatch(pattern, value):
    raise ValueError("invalid --message-id; expected <local@domain> without whitespace")
  return value


class RedirectGuard(urllib.request.HTTPRedirectHandler):
  def __init__(self, error_type: type[Exception]) -> None:
    self.error_type = error_type

  def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> Any:
    checked_url(newurl, self.error_type)
    if urllib.parse.urlsplit(req.full_url).hostname != urllib.parse.urlsplit(newurl).hostname:
      raise self.error_type("refusing a cross-host redirect")
    return super().redirect_request(req, fp, code, msg, headers, newurl)


class GmailTokenStore:
  def __init__(self, error_type: type[Exception], runner: Any = None) -> None:
    self.error_type = error_type
    if runner is not None:
      if os.environ.get("GMAIL_FAKE_SECURITY_RUNNER") != "1" or os.environ.get("GMAIL_ALLOW_FAKE_SECURITY") != "1":
        raise error_type("fake security runner requires both GMAIL_FAKE_SECURITY_RUNNER=1 and GMAIL_ALLOW_FAKE_SECURITY=1")
      print("warning: fake Gmail security runner active; real Keychain is not used", file=sys.stderr)
    self.runner = runner or subprocess.run
    self.file_mode = os.environ.get("GMAIL_TOKEN_STORE") == "file" or sys.platform != "darwin"
    self.path = Path(os.environ.get("GMAIL_TOKEN_FILE", "~/.config/approval-gated-mail/gmail-token.json")).expanduser()

  def _security(self, args: list[str], input_data: bytes | None = None, *, missing_ok: bool = False) -> str | None:
    try:
      result = self.runner(["/usr/bin/security", *args], input=input_data, capture_output=True, timeout=10, check=False)
    except (OSError, subprocess.TimeoutExpired) as exc:
      raise self.error_type("Gmail Keychain operation failed") from exc
    if result.returncode == 44 and missing_ok:
      return None
    if result.returncode:
      raise self.error_type("Gmail Keychain item is unavailable")
    return result.stdout.decode("utf-8").strip()

  def _identity(self, account: str) -> None:
    if account.startswith("-") or not re.fullmatch(r"[A-Za-z0-9@._+-]{1,128}", account) or not re.fullmatch(r"[A-Za-z0-9@._+-]{1,128}", SERVICE):
      raise self.error_type("invalid Gmail Keychain account or service")

  def _check_file(self) -> None:
    try:
      details = self.path.lstat()
    except FileNotFoundError:
      return
    if not stat.S_ISREG(details.st_mode) or (os.name != "nt" and details.st_mode & 0o077):
      raise self.error_type("Gmail token file must be a regular 0600 file, not a symlink")

  def read(self, account: str = "default") -> dict[str, Any] | None:
    if self.file_mode:
      self._check_file()
      try:
        flags = os.O_RDONLY | (os.O_NOFOLLOW if os.name == "posix" else 0)
        descriptor = os.open(self.path, flags)
      except FileNotFoundError:
        return None
      except OSError as exc:
        raise self.error_type("cannot read Gmail token file") from exc
      opened = os.fstat(descriptor)
      if not stat.S_ISREG(opened.st_mode) or (os.name != "nt" and opened.st_mode & 0o077):
        os.close(descriptor)
        raise self.error_type("Gmail token file must remain a regular 0600 file")
      with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
        try:
          data = json.load(handle)
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
          raise self.error_type("invalid Gmail token file JSON") from exc
    else:
      self._identity(account)
      secret = self._security(["find-generic-password", "-a", account, "-s", SERVICE, "-w"], missing_ok=True)
      if secret is None:
        return None
      try:
        data = json.loads(secret)
      except (UnicodeError, ValueError) as exc:
        raise self.error_type("invalid Gmail token store JSON") from exc
    if not isinstance(data, dict):
      raise self.error_type("invalid Gmail token store")
    return data

  def write(self, data: Mapping[str, Any], account: str = "default") -> None:
    secret = json.dumps(data, separators=(",", ":"), ensure_ascii=True)
    if not self.file_mode:
      self._identity(account)
      command = f"add-generic-password -a {account} -s {SERVICE} -U -X {secret.encode('utf-8').hex()}\n"
      self._security(["-i"], command.encode("ascii"))
      try:
        stored = self._security(["find-generic-password", "-a", account, "-s", SERVICE, "-w"])
      except self.error_type as exc:
        raise self.error_type("token was not stored in the Keychain") from exc
      if stored != secret:
        raise self.error_type("token was not stored in the Keychain")
      return
    self._check_file()
    self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    if not self.path.exists():
      flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | (os.O_NOFOLLOW if os.name == "posix" else 0)
      try:
        descriptor = os.open(self.path, flags, 0o600)
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
          handle.write(secret)
        return
      except FileExistsError:
        self._check_file()
    descriptor, temporary = tempfile.mkstemp(prefix=".gmail-token-", dir=self.path.parent)
    try:
      if os.name != "nt":
        os.fchmod(descriptor, 0o600)
      with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        handle.write(secret)
      self._check_file()
      os.replace(temporary, self.path)
    finally:
      if os.path.exists(temporary):
        os.unlink(temporary)

  def delete(self, account: str = "default") -> None:
    if self.file_mode:
      self._check_file()
      try:
        self.path.unlink()
      except FileNotFoundError:
        pass
    else:
      self._identity(account)
      self._security(["delete-generic-password", "-a", account, "-s", SERVICE])


class GmailBackend:
  is_gmail = True
  def __init__(self, shared: Any, config: Mapping[str, Any]) -> None:
    self.shared = shared
    self.config = config
    self.fake_path = os.environ.get("GMAIL_FAKE_TRANSPORT")
    self.fake_responses: list[dict[str, Any]] = []
    if self.fake_path:
      if os.environ.get("GMAIL_ALLOW_FAKE") != "1":
        raise shared.CliError("GMAIL_FAKE_TRANSPORT requires GMAIL_ALLOW_FAKE=1")
      print("warning: Gmail fake transport active; no real network requests will be made", file=sys.stderr)
      self.fake_responses = shared.GraphClient._load_fake_responses(Path(self.fake_path))
    self.opener = urllib.request.build_opener(RedirectGuard(shared.CliError))
    self.store = GmailTokenStore(shared.CliError)
    self.access_token: str | None = None
    self.approved_account: str | None = None
    self.skipped: list[str] = []

  def _client(self) -> dict[str, str]:
    path = os.environ.get("GMAIL_CLIENT_FILE")
    if not path:
      raise self.shared.CliError("GMAIL_CLIENT_FILE must point to your Desktop OAuth client JSON")
    try:
      installed = json.loads(Path(path).read_text(encoding="utf-8"))["installed"]
      client_id = installed["client_id"]
    except (OSError, ValueError, KeyError, TypeError) as exc:
      raise self.shared.CliError("invalid GMAIL_CLIENT_FILE Desktop OAuth client JSON") from exc
    if not isinstance(client_id, str) or not client_id:
      raise self.shared.CliError("Desktop OAuth client ID is missing")
    result = {"client_id": client_id}
    if isinstance(installed.get("client_secret"), str):
      result["client_secret"] = installed["client_secret"]
    return result

  def request(self, method: str, url: str, *, json_body: Any = None, form: Mapping[str, str] | None = None, authenticated: bool = True, _bearer: str | None = None) -> Any:
    checked_url(url, self.shared.CliError)
    method = method.upper()
    token_endpoint = url == TOKEN
    bearer = _bearer
    if authenticated and not self.fake_path:
      bearer = bearer or self._token()
    if authenticated and method != "GET" and url.startswith(API + "/"):
      if not self.approved_account or self.account_address(_bearer=bearer) != self.approved_account:
        raise self.shared.CliError("Gmail account changed or approval missing; no write made")
    attempts = 3 if method == "GET" or token_endpoint else 1
    if self.fake_path:
      for attempt in range(attempts):
        try:
          return self._fake(method, url, json_body)
        except RetryableRead as exc:
          if attempt == attempts - 1:
            raise self.shared.CliError(f"Gmail HTTP {exc.status}; read retries exhausted") from exc
          time.sleep(exc.delay)
    headers = {"Accept": "application/json"}
    if authenticated:
      headers["Authorization"] = "Bearer " + bearer
    payload = None
    if json_body is not None:
      payload = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
      headers["Content-Type"] = "application/json"
    elif form is not None:
      payload = urllib.parse.urlencode(form).encode("utf-8")
      headers["Content-Type"] = "application/x-www-form-urlencoded"
    request = urllib.request.Request(url, data=payload, headers=headers, method=method)
    for attempt in range(attempts):
      try:
        with self.opener.open(request, timeout=30) as response:
          raw = response.read()
        return json.loads(raw) if raw else None
      except urllib.error.HTTPError as exc:
        try:
          body = exc.read()
        except (OSError, http.client.HTTPException, ValueError, RecursionError) as read_error:
          if token_endpoint and attempt < attempts - 1:
            time.sleep(attempt + 1)
            continue
          raise self._connection_error(method, url) from read_error
        finally:
          exc.close()
        try:
          details = json.loads(body)
        except (ValueError, UnicodeError):
          details = {}
        error = details.get("error", {}) if isinstance(details, dict) else {}
        reason = error.get("message") or error.get("status") or error.get("code") or "" if isinstance(error, dict) else str(error)
        reasons = error.get("errors", []) if isinstance(error, dict) else []
        quota = isinstance(reasons, list) and any(isinstance(item, dict) and item.get("reason") in ("rateLimitExceeded", "userRateLimitExceeded") for item in reasons)
        if token_endpoint and reason in ("invalid_grant", "invalid_client", "invalid_request"):
          raise self.shared.CliError(GUIDANCE if reason == "invalid_grant" else "OAuth token request " + reason) from exc
        if authenticated and exc.code == 401:
          raise self.shared.CliError("Gmail authorization rejected; run auth login") from exc
        retryable = exc.code == 429 or 500 <= exc.code <= 599 or (exc.code == 403 and quota)
        if (method == "GET" or token_endpoint) and retryable and attempt < attempts - 1:
          retry_after = exc.headers.get("Retry-After") if exc.headers else None
          delay = retry_after_seconds(retry_after, attempt + 1)
          time.sleep(delay)
          continue
        if token_endpoint:
          raise self.shared.CliError(f"OAuth token HTTP {exc.code}: {self.shared.display_text(reason)}") from exc
        if url == REVOKE:
          raise self.shared.CliError(f"token revoke failed: HTTP {exc.code}: {self.shared.display_text(reason)}") from exc
        if method != "GET":
          if 500 <= exc.code <= 599:
            raise self.shared.CliError("outcome unknown; run verify-sent with the Message-ID" + (" and check the Drafts folder" if "/drafts" in url else "")) from exc
          raise self.shared.CliError(f"refused, not sent (HTTP {exc.code}): {self.shared.display_text(reason)}") from exc
        raise self.shared.CliError(f"Gmail HTTP {exc.code}: {self.shared.display_text(reason)}") from exc
      except (urllib.error.URLError, OSError, http.client.HTTPException, ValueError, RecursionError) as exc:
        if token_endpoint and attempt < attempts - 1:
          time.sleep(attempt + 1)
          continue
        raise self._connection_error(method, url) from exc
    raise self.shared.CliError("Gmail read retries exhausted")

  def _connection_error(self, method: str, url: str) -> Exception:
    if url == REVOKE:
      return self.shared.CliError("token revoke failed: request failed")
    if method != "GET" and url != TOKEN:
      return self.shared.CliError("outcome unknown; run verify-sent with the Message-ID" + (" and check the Drafts folder" if "/drafts" in url else ""))
    return self.shared.CliError("Gmail read request failed")

  def _fake(self, method: str, url: str, json_body: Any) -> Any:
    for index, response in enumerate(self.fake_responses):
      if str(response.get("method", method)).upper() != method:
        continue
      if "url" in response and response["url"] != url:
        continue
      if "url_contains" in response and response["url_contains"] not in url:
        continue
      if "path" in response and response["path"] != urllib.parse.urlsplit(url).path:
        continue
      if "request_json" in response and response["request_json"] != json_body:
        continue
      item = self.fake_responses.pop(index)
      if int(item.get("status", 200)) >= 400:
        status = int(item["status"])
        details = item.get("json", item.get("body", ""))
        reason = details.get("error", "") if isinstance(details, dict) else str(details)
        if url == REVOKE:
          raise self.shared.CliError(f"token revoke failed: HTTP {status}: {self.shared.display_text(reason)}")
        if "invalid_grant" in str(reason):
          raise self.shared.CliError(GUIDANCE)
        if url == TOKEN and str(reason) in ("invalid_client", "invalid_request"):
          raise self.shared.CliError("OAuth token request " + str(reason))
        if status == 401 and url != TOKEN:
          raise self.shared.CliError("Gmail authorization rejected; run auth login")
        if method == "GET" or url == TOKEN:
          quota = isinstance(reason, dict) and any(isinstance(error, dict) and error.get("reason") in ("rateLimitExceeded", "userRateLimitExceeded") for error in reason.get("errors", []))
          if status not in (429, 500, 502, 503, 504) and not (status == 403 and quota):
            raise self.shared.CliError(f"Gmail HTTP {status}: {self.shared.display_text(reason)}")
          headers = item.get("headers") or {}
          delay = str(headers.get("Retry-After", ""))
          raise RetryableRead(status, retry_after_seconds(delay, 0))
        if status >= 500:
          raise self.shared.CliError("outcome unknown; run verify-sent with the Message-ID" + (" and check the Drafts folder" if "/drafts" in url else ""))
        raise self.shared.CliError(f"refused, not sent (HTTP {status}): {self.shared.display_text(reason)}")
      return item.get("json", item.get("body"))
    raise self.shared.CliError(f"fake transport has no matching response for {method} {url}")

  def _token(self) -> str:
    if self.access_token:
      return self.access_token
    data = self.store.read()
    if not data:
      raise self.shared.CliError("Gmail authorization missing; run auth login")
    if data.get("expires_at", 0) > time.time() + 60 and data.get("access_token"):
      self.access_token = data["access_token"]
      return self.access_token
    refresh = data.get("refresh_token")
    if not refresh:
      raise self.shared.CliError(GUIDANCE)
    response = self.request("POST", TOKEN, form={**self._client(), "refresh_token": refresh, "grant_type": "refresh_token"}, authenticated=False)
    if not isinstance(response, dict) or not response.get("access_token"):
      raise self.shared.CliError(GUIDANCE)
    data.update(response)
    data["expires_at"] = time.time() + int(response.get("expires_in", 3600))
    self.store.write(data)
    self.access_token = data["access_token"]
    return self.access_token

  def account_address(self, *, _bearer: str | None = None) -> str:
    result = self.request("GET", API + "/profile", _bearer=_bearer)
    address = result.get("emailAddress") if isinstance(result, dict) else None
    if not isinstance(address, str):
      raise self.shared.CliError("cannot read Gmail account identity; no write made")
    address = address.strip().casefold()
    try:
      validate_mailbox(address, "authenticated account")
    except ValueError as exc:
      raise self.shared.CliError("invalid Gmail account identity; no write made") from exc
    if getaddresses([address])[0][1] != address:
      raise self.shared.CliError("invalid Gmail account identity; no write made")
    return address

  def auth_status(self) -> dict[str, Any]:
    data = self.store.read()
    return {"authorized": bool(data and data.get("refresh_token")), "account": str(data.get("account") or "default") if data else "default", "expired": bool(data and data.get("expires_at", 0) < time.time())}

  def auth_revoke(self) -> str | None:
    data = self.store.read()
    if not data:
      return "no local Gmail token"
    token = data.get("refresh_token") or data.get("access_token")
    remote_error = None
    try:
      if token:
        self.request("POST", REVOKE, form={"token": token}, authenticated=False)
    except self.shared.CliError as exc:
      remote_error = str(exc)
    finally:
      self.store.delete()
    return remote_error

  def auth_login(self, timeout: int = 180) -> None:
    client = self._client()
    verifier, challenge = pkce_pair()
    state = secrets.token_urlsafe(32)
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
      listener.bind(("127.0.0.1", 0))
      listener.listen(1)
      deadline = time.monotonic() + timeout
      redirect = f"http://127.0.0.1:{listener.getsockname()[1]}/callback"
      params = {"client_id": client["client_id"], "redirect_uri": redirect, "response_type": "code", "scope": " ".join(SCOPES), "state": state, "code_challenge": challenge, "code_challenge_method": "S256", "access_type": "offline", "prompt": "consent"}
      consent = "https://accounts.google.com/o/oauth2/v2/auth?" + urllib.parse.urlencode(params)
      print("Open this consent URL in your browser: " + consent)
      while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
          raise self.shared.CliError("Gmail authorization timed out")
        listener.settimeout(remaining)
        try:
          connection, peer = listener.accept()
        except socket.timeout as exc:
          raise self.shared.CliError("Gmail authorization timed out") from exc
        except OSError:
          time.sleep(min(0.05, remaining))
          continue
        valid = False
        with connection:
          try:
            connection.settimeout(min(10, remaining))
            request_line = connection.recv(8192).split(b"\r\n", 1)[0].decode("ascii", "replace")
            pieces = request_line.split(" ")
            target = pieces[1] if len(pieces) >= 2 else ""
            parsed = urllib.parse.urlsplit(target)
            query = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
            valid = peer[0] == "127.0.0.1" and parsed.path == "/callback" and hmac.compare_digest(query.get("state", [""])[0].encode(), state.encode())
            body = b"Return to Approval-Gated Mail to continue." if valid else b"Not found."
            status = b"200 OK" if valid else b"404 Not Found"
            connection.sendall(b"HTTP/1.1 " + status + b"\r\nContent-Type: text/plain\r\nContent-Length: " + str(len(body)).encode() + b"\r\nConnection: close\r\n\r\n" + body)
          except (socket.timeout, OSError, ValueError, UnicodeError):
            pass
        if valid:
          break
      if query.get("error"):
        raise self.shared.CliError("authorization denied: " + self.shared.display_text(query["error"][0]))
      granted = set(query.get("scope", [""])[0].split())
      missing = set(SCOPES) - granted
      if missing:
        raise self.shared.CliError("authorization missing scope: " + ", ".join(sorted(missing)))
      code = query.get("code", [""])[0]
      if not code:
        raise self.shared.CliError("OAuth authorization code missing")
    response = self.request("POST", TOKEN, form={**client, "code": code, "code_verifier": verifier, "redirect_uri": redirect, "grant_type": "authorization_code"}, authenticated=False)
    if not isinstance(response, dict) or not response.get("access_token") or not response.get("refresh_token"):
      raise self.shared.CliError("OAuth token response lacked access or refresh token")
    response["expires_at"] = time.time() + int(response.get("expires_in", 3600))
    self.store.write(response)

  def _collection(self, path: str, count: int, params: Mapping[str, str] | None = None) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    token: str | None = None
    while len(records) < count:
      query = {"maxResults": str(min(100, count - len(records))), **(params or {})}
      if token:
        query["pageToken"] = token
      result = self.request("GET", API + path + "?" + urllib.parse.urlencode(query))
      if not isinstance(result, dict):
        raise self.shared.CliError("invalid Gmail list response")
      messages = result.get("messages", [])
      if not isinstance(messages, list):
        raise self.shared.CliError("invalid Gmail message list")
      for item in messages:
        if isinstance(item, dict):
          records.append(item)
        else:
          self.skipped.append("invalid Gmail message list item")
      token = result.get("nextPageToken")
      if not token or not isinstance(token, str) or not messages:
        break
    return records[:count]

  def list_messages(self, mailbox: str, count: int, search: str | None = None, sender: str | None = None) -> list[dict[str, Any]]:
    del mailbox
    if (search is not None and ('"' in search or "\\" in search)) or (sender is not None and ('"' in sender or "\\" in sender)):
      raise self.shared.CliError("Gmail search text cannot contain a quote or backslash")
    q = "from:" + quoted_query(sender) if sender is not None else quoted_query(search) if search is not None else None
    items = self._collection("/messages", count, {"q": q} if q else None)
    records: list[dict[str, Any]] = []
    for item in items:
      message_id = item.get("id") if isinstance(item, dict) else None
      if not isinstance(message_id, str):
        self.skipped.append("message without ID")
        continue
      try:
        records.append(self.get_message("", message_id, include_body=False))
      except (self.shared.CliError, MessageError, RecursionError, ValueError, TypeError, KeyError, AttributeError) as exc:
        self.skipped.append("message " + self.shared.display_text(message_id) + ": " + self.shared.display_text(exc))
    return records

  def _header(self, payload: Mapping[str, Any], name: str) -> str:
    headers = payload.get("headers", [])
    if not isinstance(headers, list):
      raise self.shared.CliError("invalid Gmail headers")
    if any(not isinstance(item, dict) for item in headers):
      raise self.shared.CliError("invalid Gmail header entry")
    return next((str(item.get("value") or "") for item in headers if isinstance(item, dict) and str(item.get("name") or "").casefold() == name.casefold()), "")

  def _parts(self, payload: Mapping[str, Any], *, body_only: bool = False) -> list[Mapping[str, Any]]:
    stack = [(payload, 0)]
    result = []
    while stack:
      part, depth = stack.pop()
      if depth > 50:
        raise self.shared.CliError("Gmail MIME depth exceeds 50")
      if body_only and self._attachment_part(part):
        continue
      result.append(part)
      children = part.get("parts", []) or []
      if not isinstance(children, list):
        raise self.shared.CliError("invalid Gmail MIME parts")
      if any(not isinstance(child, dict) for child in children):
        raise self.shared.CliError("invalid Gmail MIME child")
      stack.extend((child, depth + 1) for child in reversed(children))
    return result

  def _attachment_part(self, part: Mapping[str, Any]) -> bool:
    body = part.get("body") or {}
    return bool(part.get("filename") or (isinstance(body, dict) and body.get("attachmentId")) or
          self._header(part, "Content-Disposition").split(";", 1)[0].strip().casefold() == "attachment")

  def _body(self, payload: Mapping[str, Any]) -> str:
    candidates: list[tuple[str, str]] = []
    for part in self._parts(payload, body_only=True):
      mime = part.get("mimeType") or ""
      if not isinstance(mime, str):
        raise self.shared.CliError("invalid Gmail MIME type")
      body = part.get("body") or {}
      if not isinstance(body, dict):
        raise self.shared.CliError("invalid Gmail MIME body")
      if isinstance(body, dict) and isinstance(body.get("data"), str):
        try:
          raw = b64decode(body["data"])
        except (ValueError, base64.binascii.Error) as exc:
          raise self.shared.CliError("invalid Gmail MIME data") from exc
        content_type = self._header(part, "Content-Type") if "headers" in part else mime
        match = re.search(r"charset\s*=\s*['\"]?([^;'\"\s]+)", content_type, re.I)
        charset = match.group(1) if match else "utf-8"
        try:
          decoded = raw.decode(charset, "replace")
        except LookupError:
          decoded = raw.decode("utf-8", "replace")
        candidates.append((mime, decoded))
    for mime in ("text/plain", "text/html"):
      for kind, value in candidates:
        if kind.casefold() == mime:
          return self.shared.html_to_text(value) if mime == "text/html" else value.replace("\r\n", "\n").replace("\r", "\n")
    return ""

  def _has_attachments(self, payload: Mapping[str, Any]) -> bool:
    return any(self._attachment_part(part) for part in self._parts(payload))

  def _record(self, value: Mapping[str, Any], include_body: bool) -> dict[str, Any]:
    payload = value.get("payload") or {}
    if not isinstance(payload, dict):
      raise self.shared.CliError("invalid Gmail payload")
    def address(name: str) -> list[dict[str, str]]:
      try:
        return [{"name": decoded_header(display), "address": addr} for display, addr in getaddresses([self._header(payload, name)]) if addr]
      except RecursionError as exc:
        raise self.shared.CliError("invalid Gmail address header") from exc
    labels = value.get("labelIds") or []
    if not isinstance(labels, list):
      raise self.shared.CliError("invalid Gmail labels")
    timestamp = value.get("internalDate")
    try:
      date = datetime.fromtimestamp(int(timestamp) / 1000, timezone.utc).isoformat().replace("+00:00", "Z") if timestamp else ""
    except (ValueError, TypeError, OverflowError) as exc:
      raise self.shared.CliError("invalid Gmail date") from exc
    record = {"id": str(value.get("id") or ""), "subject": decoded_header(self._header(payload, "Subject")), "from": (address("From") or [{}])[0], "to": address("To"), "cc": address("Cc"), "bcc": address("Bcc"), "receivedDateTime": date, "sentDateTime": date if "SENT" in labels else "", "isRead": "UNREAD" not in labels, "hasAttachments": self._has_attachments(payload), "conversationId": str(value.get("threadId") or ""), "rfc822msgid": self._header(payload, "Message-ID"), "labelIds": labels}
    if include_body:
      record["body"] = self._body(payload)
    return record

  def get_message(self, mailbox: str, message_id: str, include_body: bool = True) -> dict[str, Any]:
    del mailbox
    url = API + "/messages/" + urllib.parse.quote(message_id, safe="") + "?format=" + ("full" if include_body else "metadata")
    value = self.request("GET", url)
    if not isinstance(value, dict):
      raise self.shared.CliError("invalid Gmail message response")
    return self._record(value, include_body)

  def thread_messages(self, mailbox: str, thread_id: str) -> list[dict[str, Any]]:
    del mailbox
    result = self.request("GET", API + "/threads/" + urllib.parse.quote(thread_id, safe="") + "?format=full")
    if not isinstance(result, dict):
      raise self.shared.CliError("invalid Gmail thread response")
    messages = result.get("messages", [])
    if not isinstance(messages, list):
      raise self.shared.CliError("invalid Gmail thread messages")
    records = []
    for item in messages:
      message_id = item.get("id") if isinstance(item, dict) else None
      if not isinstance(message_id, str):
        self.skipped.append("invalid Gmail thread message without ID")
        continue
      try:
        records.append(self._record(item, True))
      except (self.shared.CliError, MessageError, RecursionError, ValueError, TypeError, KeyError, AttributeError) as exc:
        self.skipped.append("message " + self.shared.display_text(message_id) + ": " + self.shared.display_text(exc))
    return sorted(records, key=lambda item: item["receivedDateTime"])

  def attachments(self, message_id: str) -> list[dict[str, Any]]:
    fields = "partId,filename,mimeType,body(size,attachmentId)"
    for _ in range(50):
      fields = "partId,filename,mimeType,body(size,attachmentId),parts(" + fields + ")"
    query = urllib.parse.urlencode({"format": "full", "fields": "payload(" + fields + ")"})
    result = self.request("GET", API + "/messages/" + urllib.parse.quote(message_id, safe="") + "?" + query)
    records: list[dict[str, Any]] = []
    payload = result.get("payload") if isinstance(result, dict) else None
    if not isinstance(payload, dict):
      raise self.shared.CliError("invalid Gmail attachment payload")
    for part in self._parts(payload):
      if part.get("filename"):
        body = part.get("body") or {}
        if not isinstance(body, dict):
          raise self.shared.CliError("invalid Gmail attachment body")
        try:
          size = int(body.get("size") or 0)
        except (ValueError, TypeError, OverflowError) as exc:
          raise self.shared.CliError("invalid Gmail attachment size") from exc
        records.append({"id": str(body.get("attachmentId") or ""), "name": str(part.get("filename") or ""), "size": size, "contentType": str(part.get("mimeType") or "")})
    return records

  def sent_items(self, count: int) -> list[dict[str, Any]]:
    items = self._collection("/messages", count, {"labelIds": "SENT"})
    if any(not isinstance(item.get("id"), str) for item in items):
      raise self.shared.CliError("invalid Gmail Sent message ID")
    return [self.get_message("", item["id"], include_body=False) for item in items]

  def refuse_replay(self, message_id: str) -> None:
    query = urllib.parse.urlencode({"q": "rfc822msgid:" + message_id, "includeSpamTrash": "true"})
    result = self.request("GET", API + "/messages?" + query)
    if not isinstance(result, dict) or not isinstance(result.get("messages", []), list):
      raise self.shared.CliError("invalid Gmail replay guard response")
    if result.get("messages"):
      raise self.shared.CliError("a message with this Message-ID already exists; run verify-sent")

  def reply_metadata(self, message_id: str) -> dict[str, str]:
    result = self.request("GET", API + "/messages/" + urllib.parse.quote(message_id, safe="") + "?format=metadata&metadataHeaders=Message-ID&metadataHeaders=References&metadataHeaders=Subject")
    if not isinstance(result, dict):
      raise self.shared.CliError("invalid reply target response")
    payload = result.get("payload") or {}
    if not isinstance(payload, dict):
      raise self.shared.CliError("invalid reply target payload")
    return {"message_id": self._header(payload, "Message-ID"), "references": self._header(payload, "References"), "subject": self._header(payload, "Subject"), "thread_id": str(result.get("threadId") or "")}

  def build_request(self, kind: str, sender: str, to_values: Sequence[str], cc_values: Sequence[str], bcc_values: Sequence[str], subject: str, body: str, reply: Mapping[str, str] | None = None, message_id: str | None = None) -> dict[str, Any]:
    validate_mailbox(sender, "sender")
    for label, values in (("to", to_values), ("cc", cc_values), ("bcc", bcc_values)):
      for value in values:
        validate_mailbox(value, label)
    validate_header(subject, "subject")
    if message_id is None:
      domain = getaddresses([sender])[0][1].rsplit("@", 1)[1]
      message_id = make_msgid(domain=domain)
    validate_message_id(message_id, strict=True)
    message = EmailMessage(policy=policy.SMTP)
    message["From"] = sender
    message["To"] = ", ".join(to_values)
    if cc_values:
      message["Cc"] = ", ".join(cc_values)
    if bcc_values:
      message["Bcc"] = ", ".join(bcc_values)
    message["Subject"] = subject
    message["Message-ID"] = message_id
    if reply:
      if not reply.get("message_id") or not reply.get("thread_id"):
        raise self.shared.CliError("reply target lacks Message-ID or threadId")
      for key in ("message_id", "references", "subject", "thread_id"):
        validate_header(reply.get(key, ""), "reply " + key)
      validate_message_id(reply["message_id"])
      message["In-Reply-To"] = reply["message_id"]
      message["References"] = (reply.get("references", "") + " " + reply["message_id"]).strip()
    message.set_content(body)
    raw_bytes = message.as_bytes()
    parsed = BytesParser(policy=policy.default).parsebytes(raw_bytes)
    for name, values in (("From", [sender]), ("To", to_values), ("Cc", cc_values), ("Bcc", bcc_values)):
      addresses = [item.addr_spec for header in parsed.get_all(name, []) for item in header.addresses]
      expected = [getaddresses([value])[0][1] for value in values]
      if addresses != expected:
        raise ValueError(f"invalid {name.lower()} mailbox after encoding")
    for name, value in parsed.items():
      validate_header(str(value), name.lower())
    raw = base64.urlsafe_b64encode(raw_bytes).rstrip(b"=").decode("ascii")
    json_body = {"message": {"raw": raw}} if kind == "draft" else {"raw": raw}
    if reply:
      (json_body["message"] if kind == "draft" else json_body)["threadId"] = reply["thread_id"]
    return {"method": "POST", "url": API + ("/drafts" if kind == "draft" else "/messages/send"), "jsonBody": json_body, "replyTarget": dict(reply or {}), "messageId": message_id}

  def raw_message_id(self, raw: str) -> str:
    return str(BytesParser(policy=policy.default).parsebytes(b64decode(raw)).get("Message-ID") or "")

License: MIT (read it).

Back to the listing