From 65987fc772ccc0eab9877851b40dea8be98b70d9 Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:54:38 +0200 Subject: [PATCH] fix(email): test saved OAuth accounts through shared transports (#5653) * fix(email): use XOAUTH2 in test-connection for Google OAuth accounts The test-connection endpoint was password-only and had no awareness of OAuth accounts. For Google-connected accounts this caused two failures: - IMAP: "Need IMAP host, username, and password" because imap_pass is empty (no password is stored for OAuth accounts) - SMTP: 535 BadCredentials because smtp.login() was called with an empty password instead of an XOAUTH2 token Fix: include oauth_provider and token fields in saved_body when hydrating from the DB, then use conn.authenticate("XOAUTH2") / smtp.auth("XOAUTH2") for Google accounts in both the IMAP and SMTP test paths, mirroring what _send_smtp_message already does for real sends. Co-Authored-By: Claude Sonnet 4.6 * test(email): add OAuth2 tests for test-connection endpoint Covers the XOAUTH2 changes made to routes/email_routes.py: - Google OAuth accounts must not be rejected with 'Need IMAP host, username, and password' (no stored password for OAuth accounts) - IMAP and SMTP test paths must use conn.authenticate('XOAUTH2') for Google accounts - Password accounts must still use conn.login() / smtp.login() Co-Authored-By: Claude Sonnet 4.6 * fix(email): bind OAuth account tests to Google transport --------- Co-authored-by: TNTBA Co-authored-by: Claude Sonnet 4.6 --- routes/email_helpers.py | 17 +- routes/email_routes.py | 132 ++++++- tests/test_email_test_connection_oauth.py | 436 ++++++++++++++++++++++ 3 files changed, 570 insertions(+), 15 deletions(-) create mode 100644 tests/test_email_test_connection_oauth.py diff --git a/routes/email_helpers.py b/routes/email_helpers.py index b04be6066..4bbf0563f 100644 --- a/routes/email_helpers.py +++ b/routes/email_helpers.py @@ -1035,13 +1035,23 @@ def _coerce_imap_timeout_seconds(raw: str | None) -> int: _IMAP_TIMEOUT_SECONDS = _coerce_imap_timeout_seconds(os.environ.get("ODYSSEUS_IMAP_TIMEOUT_SECONDS")) -def _open_imap_connection(host: str, port: int, *, starttls: bool, timeout: int = _IMAP_TIMEOUT_SECONDS): +def _open_imap_connection( + host: str, + port: int, + *, + starttls: bool, + timeout: int = _IMAP_TIMEOUT_SECONDS, + ssl_context=None, +): """Open an IMAP connection using the configured security mode.""" port = int(port or 993) if starttls: conn = imaplib.IMAP4(host, port, timeout=timeout) try: - conn.starttls() + if ssl_context: + conn.starttls(ssl_context=ssl_context) + else: + conn.starttls() except Exception: # Don't leak the open plain socket if the STARTTLS upgrade is # rejected; close it before propagating. (#3174) @@ -1051,7 +1061,8 @@ def _open_imap_connection(host: str, port: int, *, starttls: bool, timeout: int pass raise elif port == 993: - conn = imaplib.IMAP4_SSL(host, port, timeout=timeout) + kwargs = {"ssl_context": ssl_context} if ssl_context else {} + conn = imaplib.IMAP4_SSL(host, port, timeout=timeout, **kwargs) else: conn = imaplib.IMAP4(host, port, timeout=timeout) try: diff --git a/routes/email_routes.py b/routes/email_routes.py index 26a44e2d6..211b26fb6 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -20,6 +20,7 @@ import email as email_mod import email.header import email.utils import smtplib +import ssl import json import re import html @@ -47,6 +48,7 @@ from routes.email_helpers import ( _load_settings, _save_settings, _get_email_config, _send_smtp_message, _smtp_security_mode, _IMAP_TIMEOUT_SECONDS, _open_imap_connection, + _get_valid_google_token, _xoauth2_bytes, _xoauth2_raw, make_oauth_state, verify_oauth_state, EmailNotConfiguredError, _imap_connect, _imap, _decode_header, _detect_sent_folder, _detect_drafts_folder, @@ -65,6 +67,27 @@ logger = logging.getLogger(__name__) ODYSSEUS_MAIL_ORIGIN = "odysseus-ui" EMAIL_READ_ATTACHMENT_VERSION = 2 +_GOOGLE_OAUTH_IMAP_HOST = "imap.gmail.com" +_GOOGLE_OAUTH_SMTP_HOST = "smtp.gmail.com" +_SERVER_OWNED_OAUTH_FIELDS = { + "oauth_provider", + "oauth_access_token", + "oauth_refresh_token", + "oauth_token_expiry", +} + + +def _normalized_mail_host(value) -> str: + """Normalize a mail hostname for exact provider-bound comparisons.""" + return str(value or "").strip().lower().rstrip(".") + + +def _google_oauth_imap_transport_allowed(port: int, starttls: bool) -> bool: + return (port == 993 and not starttls) or (port == 143 and starttls) + + +def _google_oauth_smtp_transport_allowed(port: int, security: str) -> bool: + return (port == 465 and security == "ssl") or (port == 587 and security == "starttls") def _safe_attachment_zip_name(name: str, fallback: str) -> str: @@ -5013,15 +5036,24 @@ def setup_email_routes(): "smtp_security": _smtp_security_mode({"smtp_security": getattr(row, "smtp_security", ""), "smtp_port": row.smtp_port}), "smtp_user": row.smtp_user or "", "smtp_password": _decrypt(row.smtp_password or ""), + "oauth_provider": row.oauth_provider or "", + "oauth_access_token": row.oauth_access_token or "", + "oauth_refresh_token": row.oauth_refresh_token or "", + "oauth_token_expiry": row.oauth_token_expiry or "", + "account_id": acc_id, } for key, value in body.items(): - if key == "account_id": + if key == "account_id" or key in _SERVER_OWNED_OAUTH_FIELDS: continue if value not in (None, ""): saved_body[key] = value body = saved_body finally: db.close() + else: + # OAuth state belongs to a saved, owner-checked account. Do not let + # inline test payloads select the OAuth branch or supply token data. + body = {key: value for key, value in body.items() if key not in _SERVER_OWNED_OAUTH_FIELDS} imap_result = {"ok": False} smtp_result = None @@ -5031,11 +5063,33 @@ def setup_email_routes(): imap_user = (body.get("imap_user") or "").strip() imap_pass = body.get("imap_password") or "" imap_starttls = bool(body.get("imap_starttls")) + oauth_provider = body.get("oauth_provider") or "" + + google_token = None + google_token_loaded = False + google_ssl_context = ( + ssl.create_default_context() + if oauth_provider == "google" + else None + ) + + def _google_token(): + nonlocal google_token, google_token_loaded + if not google_token_loaded: + google_token = _get_valid_google_token(body.get("account_id"), body) + google_token_loaded = True + if not google_token: + raise RuntimeError("Google OAuth token unavailable — reconnect the account") + return google_token if imap_port_err: imap_result = {"ok": False, "error": imap_port_err} - elif not (imap_host and imap_user and imap_pass): + elif not (imap_host and imap_user and (imap_pass or oauth_provider == "google")): imap_result = {"ok": False, "error": "Need IMAP host, username, and password"} + elif oauth_provider == "google" and _normalized_mail_host(imap_host) != _GOOGLE_OAUTH_IMAP_HOST: + imap_result = {"ok": False, "error": "Google OAuth IMAP requires imap.gmail.com"} + elif oauth_provider == "google" and not _google_oauth_imap_transport_allowed(imap_port, imap_starttls): + imap_result = {"ok": False, "error": "Google OAuth IMAP requires TLS on port 993 or STARTTLS on port 143"} else: # Connection mode resolution: # STARTTLS on → plain IMAP4 + .starttls() (upgrade) @@ -5045,14 +5099,23 @@ def setup_email_routes(): # port (Dovecot on 31143, etc.) would always fail the SSL # handshake because they're not actually wrapped in TLS. try: + imap_kwargs = { + "starttls": imap_starttls, + "timeout": _IMAP_TIMEOUT_SECONDS, + } + if google_ssl_context: + imap_kwargs["ssl_context"] = google_ssl_context conn = _open_imap_connection( imap_host, imap_port, - starttls=imap_starttls, - timeout=_IMAP_TIMEOUT_SECONDS, + **imap_kwargs, ) try: - conn.login(imap_user, imap_pass) + if oauth_provider == "google": + token = _google_token() + conn.authenticate("XOAUTH2", lambda x: _xoauth2_bytes(imap_user, token)) + else: + conn.login(imap_user, imap_pass) imap_result = {"ok": True} finally: try: conn.logout() @@ -5064,25 +5127,70 @@ def setup_email_routes(): smtp_port, smtp_port_err = _coerce_port(body.get("smtp_port"), 465) if smtp_host and smtp_port_err: smtp_result = {"ok": False, "error": smtp_port_err} + elif oauth_provider == "google" and smtp_host and _normalized_mail_host(smtp_host) != _GOOGLE_OAUTH_SMTP_HOST: + smtp_result = {"ok": False, "error": "Google OAuth SMTP requires smtp.gmail.com"} + elif ( + oauth_provider == "google" + and smtp_host + and not _google_oauth_smtp_transport_allowed( + smtp_port, + _smtp_security_mode({"smtp_security": body.get("smtp_security"), "smtp_port": smtp_port}), + ) + ): + smtp_result = {"ok": False, "error": "Google OAuth SMTP requires TLS on port 465 or STARTTLS on port 587"} elif smtp_host: smtp_security = _smtp_security_mode({"smtp_security": body.get("smtp_security"), "smtp_port": smtp_port}) smtp_user = (body.get("smtp_user") or imap_user).strip() smtp_pass = body.get("smtp_password") or imap_pass + smtp = None try: if smtp_security == "ssl": - smtp = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=10) + smtp_kwargs = ( + {"context": google_ssl_context} + if google_ssl_context + else {} + ) + smtp = smtplib.SMTP_SSL( + smtp_host, + smtp_port, + timeout=10, + **smtp_kwargs, + ) else: smtp = smtplib.SMTP(smtp_host, smtp_port, timeout=10) if smtp_security == "starttls": - smtp.starttls() - try: + try: + if google_ssl_context: + smtp.starttls(context=google_ssl_context) + else: + smtp.starttls() + except Exception: + # STARTTLS failed before the auth cleanup block. + # Close the still-open plaintext socket explicitly. + try: + smtp.close() + except Exception: + pass + smtp = None + raise + if oauth_provider == "google": + token = _google_token() + smtp.ehlo() + smtp.auth("XOAUTH2", lambda challenge=None: _xoauth2_raw(smtp_user, token), initial_response_ok=True) + else: smtp.login(smtp_user, smtp_pass) - smtp_result = {"ok": True} - finally: - try: smtp.quit() - except Exception: pass + smtp_result = {"ok": True} except Exception as e: smtp_result = {"ok": False, "error": _friendly_email_auth_error("SMTP", smtp_host, e)} + finally: + if smtp is not None: + try: + smtp.quit() + except Exception: + try: + smtp.close() + except Exception: + pass return { "ok": imap_result["ok"] and (smtp_result is None or smtp_result["ok"]), diff --git a/tests/test_email_test_connection_oauth.py b/tests/test_email_test_connection_oauth.py new file mode 100644 index 000000000..cc524d958 --- /dev/null +++ b/tests/test_email_test_connection_oauth.py @@ -0,0 +1,436 @@ +"""Tests for Google OAuth2 support in the /api/email/accounts/test endpoint. + +Covers the changes made to routes/email_routes.py: + +- test_account_config: OAuth accounts must not require a stored password. +- IMAP and SMTP test paths must use XOAUTH2 for Google accounts. +- Password accounts must still use conn.login() / smtp.login(). + +These tests use only in-memory SQLite (via SQLAlchemy) and mock network +objects — no live email server or real OAuth credentials are needed. +""" + +import time +import unittest.mock as mock + +import pytest + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def _make_orm_db(): + """Return (Session, SessionFactory) backed by an isolated in-memory SQLite DB.""" + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from core.database import Base + engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine) + Factory = sessionmaker(bind=engine) + return Factory(), Factory + + +def _make_orm_account(session, account_id="acct-1", owner="alice", **kwargs): + from core.database import EmailAccount + row = EmailAccount( + id=account_id, + owner=owner, + name=kwargs.get("name", "Test"), + from_address=kwargs.get("from_address", "me@nia.law"), + imap_host=kwargs.get("imap_host", "imap.gmail.com"), + imap_port=kwargs.get("imap_port", 993), + imap_user=kwargs.get("imap_user", "me@nia.law"), + imap_starttls=kwargs.get("imap_starttls", False), + smtp_host=kwargs.get("smtp_host", "smtp.gmail.com"), + smtp_port=kwargs.get("smtp_port", 587), + smtp_security=kwargs.get("smtp_security", "starttls"), + smtp_user=kwargs.get("smtp_user", "me@nia.law"), + ) + for k, v in kwargs.items(): + if hasattr(row, k): + setattr(row, k, v) + session.add(row) + session.commit() + return row + + +# ── test_connection route: OAuth awareness ──────────────────────────────────── + +@pytest.mark.asyncio +async def test_test_connection_oauth_account_uses_xoauth2_for_imap_and_smtp(): + """The saved-account test must use XOAUTH2 for both mail protocols.""" + from src.secret_storage import encrypt as _enc + from routes.email_routes import setup_email_routes + + future_expiry = str(int(time.time()) + 7200) + db, Factory = _make_orm_db() + _make_orm_account( + db, account_id="acct-oauth", owner="alice", + oauth_provider="google", + oauth_access_token=_enc("ya29.live"), + oauth_refresh_token=_enc("1//refresh"), + oauth_token_expiry=future_expiry, + ) + db.close() + + router = setup_email_routes() + test_conn = None + for route in router.routes: + if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()): + test_conn = route.endpoint + break + assert test_conn is not None, "test-connection route not found" + + mock_imap_conn = mock.MagicMock() + mock_smtp_conn = mock.MagicMock() + + class _FakeReq: + async def json(self): + return {"account_id": "acct-oauth"} + + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \ + mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \ + mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn), \ + mock.patch("routes.email_routes._get_valid_google_token", return_value="ya29.live") as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is True + assert result["imap"].get("ok") is True, \ + f"OAuth IMAP test must succeed, got: {result['imap']}" + assert result["smtp"].get("ok") is True, \ + f"OAuth SMTP test must succeed, got: {result['smtp']}" + mock_imap_conn.authenticate.assert_called_once() + assert mock_imap_conn.authenticate.call_args[0][0] == "XOAUTH2" + mock_imap_conn.login.assert_not_called() + mock_smtp_conn.auth.assert_called_once() + assert mock_smtp_conn.auth.call_args[0][0] == "XOAUTH2" + mock_smtp_conn.login.assert_not_called() + token_getter.assert_called_once() + + +@pytest.mark.asyncio +async def test_test_connection_password_account_still_uses_login(): + """Existing password accounts must still go through the login() path.""" + from src.secret_storage import encrypt as _enc + from routes.email_routes import setup_email_routes + + db, Factory = _make_orm_db() + _make_orm_account( + db, account_id="acct-pw", owner="alice", + imap_host="imap.example.com", + imap_user="me@example.com", + smtp_host="smtp.example.com", + smtp_user="me@example.com", + imap_password=_enc("hunter2"), + ) + db.close() + + router = setup_email_routes() + test_conn = None + for route in router.routes: + if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()): + test_conn = route.endpoint + break + + mock_imap_conn = mock.MagicMock() + mock_smtp_conn = mock.MagicMock() + + class _FakeReq: + async def json(self): + return {"account_id": "acct-pw"} + + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \ + mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \ + mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn): + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is True + mock_imap_conn.login.assert_called_once_with("me@example.com", "hunter2") + mock_imap_conn.authenticate.assert_not_called() + mock_smtp_conn.login.assert_called_once_with("me@example.com", "hunter2") + mock_smtp_conn.auth.assert_not_called() + + +@pytest.mark.asyncio +async def test_test_connection_rejects_non_google_hosts_before_oauth_auth(): + """Saved Google tokens must never be routed to edited custom hosts.""" + from src.secret_storage import encrypt as _enc + from routes.email_routes import setup_email_routes + + db, Factory = _make_orm_db() + _make_orm_account( + db, + account_id="acct-oauth", + owner="alice", + oauth_provider="google", + oauth_access_token=_enc("ya29.live"), + oauth_refresh_token=_enc("1//refresh"), + oauth_token_expiry=str(int(time.time()) + 7200), + ) + db.close() + + router = setup_email_routes() + test_conn = next( + route.endpoint + for route in router.routes + if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()) + ) + + class _FakeReq: + async def json(self): + return { + "account_id": "acct-oauth", + "imap_host": "collector.invalid", + "smtp_host": "collector.invalid", + } + + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch("routes.email_routes._open_imap_connection") as open_imap, \ + mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \ + mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \ + mock.patch("routes.email_routes._get_valid_google_token") as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is False + assert "imap.gmail.com" in result["imap"]["error"] + assert "smtp.gmail.com" in result["smtp"]["error"] + open_imap.assert_not_called() + open_smtp.assert_not_called() + open_smtp_ssl.assert_not_called() + token_getter.assert_not_called() + + +@pytest.mark.asyncio +async def test_test_connection_rejects_insecure_oauth_transports_before_auth(): + """Google OAuth credentials must not be tested over plaintext transports.""" + from src.secret_storage import encrypt as _enc + from routes.email_routes import setup_email_routes + + db, Factory = _make_orm_db() + _make_orm_account( + db, + account_id="acct-oauth", + owner="alice", + oauth_provider="google", + oauth_access_token=_enc("ya29.live"), + oauth_refresh_token=_enc("1//refresh"), + oauth_token_expiry=str(int(time.time()) + 7200), + ) + db.close() + + router = setup_email_routes() + test_conn = next( + route.endpoint + for route in router.routes + if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()) + ) + + class _FakeReq: + async def json(self): + return { + "account_id": "acct-oauth", + "imap_port": 143, + "imap_starttls": False, + "smtp_port": 587, + "smtp_security": "none", + } + + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch("routes.email_routes._open_imap_connection") as open_imap, \ + mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \ + mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \ + mock.patch("routes.email_routes._get_valid_google_token") as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is False + assert "TLS" in result["imap"]["error"] + assert "TLS" in result["smtp"]["error"] + open_imap.assert_not_called() + open_smtp.assert_not_called() + open_smtp_ssl.assert_not_called() + token_getter.assert_not_called() + + +@pytest.mark.asyncio +async def test_test_connection_does_not_accept_inline_oauth_state(): + """Only an owner-checked saved account may select the OAuth branch.""" + from routes.email_routes import setup_email_routes + + router = setup_email_routes() + test_conn = next( + route.endpoint + for route in router.routes + if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()) + ) + + class _FakeReq: + async def json(self): + return { + "imap_host": "imap.gmail.com", + "imap_user": "me@example.com", + "oauth_provider": "google", + "oauth_access_token": "client-supplied-token", + } + + with mock.patch("routes.email_routes._open_imap_connection") as open_imap, \ + mock.patch("routes.email_routes._get_valid_google_token") as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is False + assert result["imap"]["error"] == "Need IMAP host, username, and password" + open_imap.assert_not_called() + token_getter.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("imap_starttls", "imap_port"), + [(False, 993), (True, 143)], +) +async def test_test_connection_verifies_imap_tls_before_loading_oauth_token( + imap_starttls, + imap_port, +): + """Both Google IMAP TLS modes receive a certificate-verifying context; + certificate rejection happens before the bearer token is loaded.""" + import ssl + + from routes.email_routes import setup_email_routes + from src.secret_storage import encrypt as _enc + + db, Factory = _make_orm_db() + _make_orm_account( + db, + account_id="acct-imap-tls", + owner="alice", + imap_port=imap_port, + imap_starttls=imap_starttls, + smtp_host="", + oauth_provider="google", + oauth_access_token=_enc("ya29.live"), + oauth_refresh_token=_enc("1//refresh"), + oauth_token_expiry=str(int(time.time()) + 7200), + ) + db.close() + + router = setup_email_routes() + test_conn = next( + route.endpoint + for route in router.routes + if route.path == "/api/email/accounts/test" + and "POST" in getattr(route, "methods", set()) + ) + + class _FakeReq: + async def json(self): + return {"account_id": "acct-imap-tls"} + + context = ssl.create_default_context() + starttls_conn = mock.MagicMock() + starttls_conn.starttls.side_effect = ssl.SSLCertVerificationError( + "untrusted certificate" + ) + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch( + "routes.email_routes.ssl.create_default_context", + return_value=context, + ), mock.patch( + "routes.email_helpers.imaplib.IMAP4", + return_value=starttls_conn, + ) as imap_cls, mock.patch( + "routes.email_helpers.imaplib.IMAP4_SSL", + side_effect=ssl.SSLCertVerificationError("untrusted certificate"), + ) as imap_ssl_cls, mock.patch( + "routes.email_routes._get_valid_google_token" + ) as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is False + assert result["imap"]["ok"] is False + assert context.check_hostname is True + assert context.verify_mode == ssl.CERT_REQUIRED + token_getter.assert_not_called() + if imap_starttls: + assert starttls_conn.starttls.call_args.kwargs["ssl_context"] is context + imap_ssl_cls.assert_not_called() + else: + assert imap_ssl_cls.call_args.kwargs["ssl_context"] is context + imap_cls.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("smtp_security", "smtp_port"), + [("ssl", 465), ("starttls", 587)], +) +async def test_test_connection_verifies_smtp_tls_before_loading_oauth_token( + smtp_security, + smtp_port, +): + """Both Google SMTP TLS modes reject an invalid certificate before any + XOAUTH2 credential is obtained or sent.""" + import ssl + + from routes.email_routes import setup_email_routes + from src.secret_storage import encrypt as _enc + + db, Factory = _make_orm_db() + _make_orm_account( + db, + account_id="acct-smtp-tls", + owner="alice", + imap_host="", + smtp_port=smtp_port, + smtp_security=smtp_security, + oauth_provider="google", + oauth_access_token=_enc("ya29.live"), + oauth_refresh_token=_enc("1//refresh"), + oauth_token_expiry=str(int(time.time()) + 7200), + ) + db.close() + + router = setup_email_routes() + test_conn = next( + route.endpoint + for route in router.routes + if route.path == "/api/email/accounts/test" + and "POST" in getattr(route, "methods", set()) + ) + + class _FakeReq: + async def json(self): + return {"account_id": "acct-smtp-tls"} + + context = ssl.create_default_context() + starttls_smtp = mock.MagicMock() + starttls_smtp.starttls.side_effect = ssl.SSLCertVerificationError( + "untrusted certificate" + ) + with mock.patch("core.database.SessionLocal", Factory), \ + mock.patch( + "routes.email_routes.ssl.create_default_context", + return_value=context, + ), mock.patch( + "routes.email_routes.smtplib.SMTP", + return_value=starttls_smtp, + ) as smtp_cls, mock.patch( + "routes.email_routes.smtplib.SMTP_SSL", + side_effect=ssl.SSLCertVerificationError("untrusted certificate"), + ) as smtp_ssl_cls, mock.patch( + "routes.email_routes._get_valid_google_token" + ) as token_getter: + result = await test_conn(req=_FakeReq(), owner="alice") + + assert result["ok"] is False + assert result["smtp"]["ok"] is False + assert context.check_hostname is True + assert context.verify_mode == ssl.CERT_REQUIRED + token_getter.assert_not_called() + if smtp_security == "starttls": + assert starttls_smtp.starttls.call_args.kwargs["context"] is context + assert starttls_smtp.close.called + smtp_ssl_cls.assert_not_called() + else: + assert smtp_ssl_cls.call_args.kwargs["context"] is context + smtp_cls.assert_not_called()