Merge current dev and refresh PR #6739 from the 7011 integration

routes/codex_routes.py

@@ -35,6 +35,7 @@ CALENDAR_READ_SCOPES = {"calendar:read", "calendar:write"}
 CALENDAR_WRITE_SCOPES = {"calendar:write"}
 DOCS_READ_SCOPES = {"documents:read", "documents:write"}
 DOCS_WRITE_SCOPES = {"documents:write"}
+CHAT_READ_SCOPES = {"chat"}
 WRITE_ACTIONS = {"add", "create", "new", "save", "remind", "update", "delete", "toggle_item", "remove", "remove_item"}
 
 
@@ -161,6 +162,7 @@ def setup_codex_routes(
     memory_router: APIRouter | None = None,
     calendar_router: APIRouter | None = None,
     document_router: APIRouter | None = None,
+    history_router: APIRouter | None = None,
 ) -> APIRouter:
     router = APIRouter(prefix="/api/codex", tags=["codex"])
     email_list_endpoint = _find_endpoint(email_router, "GET", "/api/email/list")
@@ -174,6 +176,7 @@ def setup_codex_routes(
     documents_library_endpoint = _find_endpoint(document_router, "GET", "/api/documents/library")
     documents_get_endpoint = _find_endpoint(document_router, "GET", "/api/document/{doc_id}")
     documents_create_endpoint = _find_endpoint(document_router, "POST", "/api/document")
+    chat_history_endpoint = _find_endpoint(history_router, "GET", "/api/history/{session_id}")
 
     @router.get("/capabilities")
     def capabilities(request: Request):
@@ -219,6 +222,11 @@ def setup_codex_routes(
                     "launch": scoped(COOKBOOK_LAUNCH_SCOPES),
                     "actions": ["tasks", "servers", "output", "serve", "stop"],
                 },
+                "chat": {
+                    "read": scoped(CHAT_READ_SCOPES),
+                    "actions": ["read"],
+                    "available": chat_history_endpoint is not None,
+                },
             },
             "safety": {
                 "email_send_requires_confirmation": True,
@@ -226,6 +234,33 @@ def setup_codex_routes(
             },
         }
 
+    @router.get("/chat/{session_id}")
+    async def read_chat(
+        request: Request,
+        session_id: str,
+        offset: int = 0,
+        limit: int = 100,
+    ):
+        _scope_owner(request, CHAT_READ_SCOPES)
+        if chat_history_endpoint is None:
+            raise HTTPException(503, "chat history unavailable")
+        offset, limit = _clamp_pagination(
+            offset,
+            limit,
+            default_limit=100,
+            max_limit=100,
+        )
+        result = await chat_history_endpoint(request, session_id, limit, offset)
+        return {
+            "session_id": session_id,
+            "name": result["name"],
+            "model": result["model"],
+            "messages": result["history"],
+            "total": result["total"],
+            "offset": result["offset"],
+            "limit": result["limit"],
+        }
+
     @router.get("/plugin.zip")
     def plugin_zip(request: Request):
         require_authenticated_request(request)

routes/editor_agent_routes.py

@@ -1732,6 +1732,7 @@ def setup_editor_agent_routes() -> APIRouter:
             from core.database import SessionLocal, ModelEndpoint
             from src.auth_helpers import owner_filter
             from src.endpoint_resolver import _endpoint_enabled_models
+            from src.image_model_ids import looks_like_image_generation_model
 
             db = SessionLocal()
             try:
@@ -1741,10 +1742,16 @@ def setup_editor_agent_routes() -> APIRouter:
                     owner,
                 )
                 for ep in query.all():
-                    if getattr(ep, "model_type", None) != "image":
-                        continue
+                    # Image-only endpoints (the local Qwen server) list their
+                    # models verbatim. Gateway/chat endpoints (OpenRouter, …)
+                    # mix chat and image models, so expose only ids that are
+                    # recognisably image-capable — the same test the request
+                    # router uses.
+                    image_endpoint = getattr(ep, "model_type", None) == "image"
                     provider = getattr(ep, "name", None) or getattr(ep, "base_url", "") or "image"
                     for name in _endpoint_enabled_models(ep):
+                        if not image_endpoint and not looks_like_image_generation_model(name):
+                            continue
                         key = (provider, name)
                         if key in seen:
                             continue

routes/email/email_helpers.py

@@ -17,6 +17,8 @@ import base64
 import time
 import imaplib
 import smtplib
+import socket
+import ssl
 import email as email_mod
 import email.header
 import email.utils
@@ -36,6 +38,7 @@ from typing import Optional, List
 
 from src.auth_helpers import _auth_disabled, get_current_user
 from src.secret_storage import decrypt as _decrypt
+from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
 
 logger = logging.getLogger(__name__)
 
@@ -160,6 +163,94 @@ def _smtp_security_mode(cfg: dict) -> str:
     return "ssl"
 
 
+# Raised when a mail host resolves into a denied range (see _mail_private_blocked).
+MailServerAddressBlocked = OutboundAddressBlocked
+
+
+def _mail_private_blocked(owner: str | None) -> bool:
+    """Whether private/loopback/shared mail destinations are denied for *owner*.
+
+    Link-local (cloud metadata), multicast, reserved and unspecified addresses
+    are always denied (src/url_safety.py). Private ranges are where a mail
+    tester or saved account becomes an internal-network probe, so they are
+    only allowed for an admin or in single-user mode (AUTH_ENABLED=false), the
+    same principals trusted with host-level tools. Operators of multi-user
+    deployments with a LAN mail server opt in with EMAIL_ALLOW_PRIVATE_IPS=true;
+    EMAIL_BLOCK_PRIVATE_IPS=true denies private ranges for everyone.
+    """
+    if os.getenv("EMAIL_BLOCK_PRIVATE_IPS", "").strip().lower() == "true":
+        return True
+    if os.getenv("EMAIL_ALLOW_PRIVATE_IPS", "").strip().lower() == "true":
+        return False
+    from src.tool_security import owner_is_admin_or_single_user
+
+    return not owner_is_admin_or_single_user((owner or "").strip() or None)
+
+
+def _mail_socket_timeout(timeout):
+    return socket._GLOBAL_DEFAULT_TIMEOUT if timeout is None else timeout
+
+
+class _PolicyIMAP4(imaplib.IMAP4):
+    """IMAP4 whose socket comes from connect_outbound_tcp.
+
+    The mail host is resolved once and connected to only at the addresses the
+    address policy approved, so DNS rebinding between check and connect is
+    impossible. STARTTLS still verifies against ``self.host``.
+    """
+
+    def __init__(self, host, port, *, block_private: bool, timeout=None):
+        self._block_private = block_private
+        super().__init__(host, port, timeout=timeout)
+
+    def _create_socket(self, timeout):
+        if timeout is not None and not timeout:
+            raise ValueError("Non-blocking socket (timeout=0) is not supported")
+        return connect_outbound_tcp(
+            self.host, self.port, timeout=_mail_socket_timeout(timeout), block_private=self._block_private,
+        )
+
+
+class _PolicyIMAP4_SSL(imaplib.IMAP4_SSL):
+    """IMAP4_SSL over a policy-checked socket; TLS SNI/verification use the hostname."""
+
+    def __init__(self, host, port, *, block_private: bool, timeout=None, ssl_context=None):
+        self._block_private = block_private
+        super().__init__(host, port, ssl_context=ssl_context, timeout=timeout)
+
+    def _create_socket(self, timeout):
+        sock = _PolicyIMAP4._create_socket(self, timeout)
+        return self.ssl_context.wrap_socket(sock, server_hostname=self.host)
+
+
+class _PolicySMTP(smtplib.SMTP):
+    """SMTP whose socket comes from connect_outbound_tcp (see _PolicyIMAP4)."""
+
+    def __init__(self, host="", port=0, *, block_private: bool, **kwargs):
+        self._block_private = block_private
+        super().__init__(host, port, **kwargs)
+
+    def _get_socket(self, host, port, timeout):
+        if timeout is not None and not timeout:
+            raise ValueError("Non-blocking socket (timeout=0) is not supported")
+        return connect_outbound_tcp(
+            host, port, timeout=timeout, block_private=self._block_private,
+            source_address=self.source_address,
+        )
+
+
+class _PolicySMTP_SSL(smtplib.SMTP_SSL):
+    """SMTP_SSL over a policy-checked socket; TLS SNI/verification use the hostname."""
+
+    def __init__(self, host="", port=0, *, block_private: bool, **kwargs):
+        self._block_private = block_private
+        super().__init__(host, port, **kwargs)
+
+    def _get_socket(self, host, port, timeout):
+        sock = _PolicySMTP._get_socket(self, host, port, timeout)
+        return self.context.wrap_socket(sock, server_hostname=self._host)
+
+
 def _send_smtp_message(cfg: dict, from_addr: str, recipients: list[str], message: str | bytes, timeout: int = 30) -> None:
     """Send through SMTP using the configured transport security mode."""
     host = cfg["smtp_host"]
@@ -178,14 +269,15 @@ def _send_smtp_message(cfg: dict, from_addr: str, recipients: list[str], message
             smtp.login(user, password)
 
     security = _smtp_security_mode(cfg)
+    block_private = _mail_private_blocked(cfg.get("owner"))
 
     if security == "ssl":
-        with smtplib.SMTP_SSL(host, port, timeout=timeout) as smtp:
+        with _PolicySMTP_SSL(host, port, timeout=timeout, block_private=block_private) as smtp:
             _auth_smtp(smtp)
             smtp.sendmail(from_addr, recipients, message)
         return
 
-    with smtplib.SMTP(host, port, timeout=timeout) as smtp:
+    with _PolicySMTP(host, port, timeout=timeout, block_private=block_private) as smtp:
         if security == "starttls":
             smtp.starttls()
         _auth_smtp(smtp)
@@ -224,6 +316,39 @@ def _friendly_email_auth_error(protocol: str, host: str, error: object) -> str:
     return raw[:200]
 
 
+# Bound at import: callers (and tests) may swap out imaplib.IMAP4 itself.
+_IMAP_ABORT = imaplib.IMAP4.abort
+
+
+def _mail_connection_test_error(protocol: str, host: str, error: BaseException) -> str:
+    """User-facing result for a failed IMAP/SMTP connection test.
+
+    Transport failures are reported by category, never with the peer's own
+    bytes: a non-mail service answering on the chosen host/port would
+    otherwise have its banner echoed back (an internal-service fingerprinting
+    oracle). Errors from a server that does speak the protocol (login/auth
+    rejections) keep their text via _friendly_email_auth_error, since that is
+    what the user needs to fix their settings.
+    """
+    if isinstance(error, MailServerAddressBlocked):
+        return f"{protocol} server address is not allowed by this server's network policy"
+    if isinstance(error, (_IMAP_ABORT, smtplib.SMTPConnectError, smtplib.SMTPServerDisconnected)):
+        return f"{protocol} server did not respond like an {protocol} server"
+    if isinstance(error, ssl.SSLCertVerificationError):
+        return f"{protocol} TLS certificate verification failed"
+    if isinstance(error, ssl.SSLError):
+        return f"{protocol} TLS handshake failed; check the port and security setting"
+    if isinstance(error, (socket.timeout, TimeoutError)):
+        return f"{protocol} connection timed out"
+    if isinstance(error, ConnectionRefusedError):
+        return f"{protocol} connection refused"
+    if isinstance(error, socket.gaierror):
+        return f"{protocol} server name could not be resolved"
+    if isinstance(error, OSError) and not isinstance(error, smtplib.SMTPException):
+        return f"{protocol} connection failed"
+    return _friendly_email_auth_error(protocol, host, error)
+
+
 def _strip_think(text: str) -> str:
     """Email-flavored think strip — thin wrapper over the central helper.
 
@@ -685,18 +810,21 @@ def _ensure_sender_signatures_table(conn):
         _lg.getLogger(__name__).warning(f"sender_signatures owner-migration skipped: {_mig_e}")
 
 
-def attachment_extract_dir(folder: str, uid: str) -> Path:
-    """Containment-safe extraction directory for an attachment.
+def attachment_extract_dir(folder: str, uid: str, *, owner: str, account_id: str | None) -> Path:
+    """Containment-safe extraction directory for one message's attachments.
 
-    `folder` and `uid` are user-controlled (query/path params). Flatten them to
-    a single safe path segment so a value like folder='../../tmp' can't escape
-    ATTACHMENTS_DIR, then assert containment as belt-and-suspenders."""
-    key = re.sub(r"[^A-Za-z0-9._-]", "_", f"{folder}_{uid}") or "_"
-    target = (ATTACHMENTS_DIR / key).resolve()
-    base = ATTACHMENTS_DIR.resolve()
-    if target != base and base not in target.parents:
+    IMAP UIDs are small per-mailbox counters, so `folder_uid` alone put every
+    user's and every account's "INBOX 42" in one directory, where extractions
+    overwrote each other by filename (a path handed to one user's agent could
+    then hold another user's attachment). See src.mail_attachment_paths: the
+    directory is keyed by (owner, account, folder, uid) and contained in
+    ATTACHMENTS_DIR whatever `folder`/`uid` hold."""
+    from src.mail_attachment_paths import attachment_scope_dir
+
+    try:
+        return attachment_scope_dir(ATTACHMENTS_DIR, folder, uid, owner=owner, account_id=account_id)
+    except ValueError:
         raise HTTPException(400, "Invalid attachment location")
-    return target
 
 
 def _init_scheduled_db():
@@ -1077,6 +1205,8 @@ def _get_email_config(account_id: str | None = None, owner: str = "") -> dict:
                 cfg = {
                     "account_id": row.id,
                     "account_name": row.name,
+                    # Principal for the outbound mail address policy.
+                    "owner": row.owner or owner or "",
                     "smtp_host": row.smtp_host or "",
                     "smtp_port": int(row.smtp_port or 465),
                     "smtp_security": _smtp_security_mode({"smtp_security": getattr(row, "smtp_security", ""), "smtp_port": row.smtp_port}),
@@ -1110,6 +1240,7 @@ def _get_email_config(account_id: str | None = None, owner: str = "") -> dict:
     cfg = {
         "account_id": resolved_id,
         "account_name": "legacy",
+        "owner": owner or "",
         "smtp_host": settings.get("smtp_host", os.environ.get("SMTP_HOST", "")),
         "smtp_port": int(settings.get("smtp_port", os.environ.get("SMTP_PORT", "465")) or 465),
         "smtp_security": _smtp_security_mode({
@@ -1173,11 +1304,16 @@ def _open_imap_connection(
     starttls: bool,
     timeout: int = _IMAP_TIMEOUT_SECONDS,
     ssl_context=None,
+    owner: str | None = None,
 ):
-    """Open an IMAP connection using the configured security mode."""
+    """Open an IMAP connection using the configured security mode.
+
+    `owner` is the principal the destination is judged for
+    (see _mail_private_blocked)."""
     port = int(port or 993)
+    block_private = _mail_private_blocked(owner)
     if starttls:
-        conn = imaplib.IMAP4(host, port, timeout=timeout)
+        conn = _PolicyIMAP4(host, port, timeout=timeout, block_private=block_private)
         try:
             if ssl_context:
                 conn.starttls(ssl_context=ssl_context)
@@ -1193,9 +1329,9 @@ def _open_imap_connection(
             raise
     elif port == 993:
         kwargs = {"ssl_context": ssl_context} if ssl_context else {}
-        conn = imaplib.IMAP4_SSL(host, port, timeout=timeout, **kwargs)
+        conn = _PolicyIMAP4_SSL(host, port, timeout=timeout, block_private=block_private, **kwargs)
     else:
-        conn = imaplib.IMAP4(host, port, timeout=timeout)
+        conn = _PolicyIMAP4(host, port, timeout=timeout, block_private=block_private)
     try:
         conn.sock.settimeout(timeout)
     except Exception:
@@ -1235,6 +1371,7 @@ def _imap_connect(account_id: str | None = None, owner: str = "",
         cfg["imap_port"],
         starttls=bool(cfg.get("imap_starttls")),
         timeout=timeout,
+        owner=cfg.get("owner"),
     )
     try:
         if cfg.get("oauth_provider") == "google":

routes/email/email_routes.py

@@ -25,17 +25,21 @@ import json
 import re
 import html
 import io
+import tempfile
 import zipfile
-from urllib.parse import parse_qs, unquote, urlparse
+from urllib.parse import parse_qs, quote, unquote, urlparse
 from html.parser import HTMLParser as _HTMLParser
 import logging
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
+from email.message import EmailMessage
+from email import policy as email_policy
 
 from email.mime.text import MIMEText
 from email.mime.multipart import MIMEMultipart
 
+from starlette.concurrency import run_in_threadpool
 from fastapi import APIRouter, Query, UploadFile, File, BackgroundTasks, HTTPException, Depends, Request
 from fastapi.responses import FileResponse, StreamingResponse
 from src.constants import DATA_DIR
@@ -61,6 +65,7 @@ from .email_helpers import (
     _fetch_sender_thread_context, _pre_retrieve_context,
     _EMAIL_REPLY_SYS_PROMPT_BASE, _POOL_HOOKS,
     _friendly_email_auth_error, _email_summary_failure_log_detail,
+    _mail_connection_test_error, _mail_private_blocked, _PolicySMTP, _PolicySMTP_SSL,
     _generate_email_summary, EMAIL_SUMMARY_ERROR_CODE, EMAIL_SUMMARY_ERROR_MESSAGE,
     SendEmailRequest, ExtractStyleRequest,
     ATTACHMENTS_DIR, COMPOSE_UPLOADS_DIR, SCHEDULED_DB,
@@ -83,6 +88,17 @@ _SERVER_OWNED_OAUTH_FIELDS = {
 _CACHE_HOOKS: dict = {"invalidate_list": None}
 
 
+def _check_mail_host(host, owner: str, label: str) -> str | None:
+    """Reject unsafe or unresolved mail hosts before saving or testing them."""
+    from src.url_safety import check_outbound_host
+    ok, reason = check_outbound_host(host, block_private=_mail_private_blocked(owner))
+    if ok:
+        return None
+    logger.warning("Rejected %s mail host for owner=%r: %s", label, owner, reason)
+    # Resolver and exception details belong in server logs, never the response.
+    return f"Rejected {label} host: server address is not allowed or does not resolve"
+
+
 def _normalized_mail_host(value) -> str:
     """Normalize a mail hostname for exact provider-bound comparisons."""
     return str(value or "").strip().lower().rstrip(".")
@@ -635,10 +651,15 @@ def _message_id_chain(*values: str) -> list[str]:
 
 
 def _uid_from_fetch_meta(meta_b: bytes) -> str:
-    m = re.search(rb"\bUID\s+(\d+)\b", meta_b)
+    m = re.search(rb"\bUID\s+(\d+)\b", meta_b, re.IGNORECASE)
     return m.group(1).decode() if m else ""
 
 
+def _size_from_fetch_meta(meta_b: bytes) -> int | None:
+    m = re.search(rb"\bRFC822\.SIZE\s+(\d+)\b", meta_b, re.IGNORECASE)
+    return int(m.group(1)) if m else None
+
+
 def _parse_list_unsubscribe_header(value: str | None) -> list[dict]:
     """Parse RFC List-Unsubscribe entries into safe reviewable actions.
 
@@ -1542,6 +1563,44 @@ def _envelope_recipients(*fields: str) -> list:
     return out
 
 
+_MD_LINK_URL_RE = re.compile(r"\((https?://)[^)\s]*")
+
+
+def _md_links_to_html(s: str) -> str:
+    """Linear ``re.sub(r"\\[([^\\]]+)\\]\\((https?://[^)\\s]+)\\)", '<a href="\\2">\\1</a>', s)``.
+
+    Every `[` before the next `]` shares that `]`, so only the first can match;
+    and when a URL runs into whitespace/end without `)`, every later `[` whose
+    `]` also lies before that stop fails the same way. Skipping those spans
+    keeps a `[[[` or `[a](http://x[a](http://x` flood O(n) instead of the
+    regex's O(n^2).
+    """
+    out: list[str] = []
+    emitted = scan = 0
+    while True:
+        open_at = s.find("[", scan)
+        if open_at < 0:
+            break
+        close_at = s.find("]", open_at + 1)
+        if close_at < 0:
+            break
+        scan = close_at + 1
+        if close_at == open_at + 1:
+            continue
+        url = _MD_LINK_URL_RE.match(s, close_at + 1)
+        if url is None:
+            continue
+        end = url.end()
+        if end < len(s) and s[end] == ")" and end > url.end(1):
+            out.append(s[emitted:open_at])
+            out.append(f'<a href="{s[url.start(1):end]}">{s[open_at + 1:close_at]}</a>')
+            emitted = scan = end + 1
+        elif end > url.end(1):
+            scan = max(scan, s.rfind("]", scan, end) + 1)
+    out.append(s[emitted:])
+    return "".join(out)
+
+
 def _md_to_email_html(text: str) -> str:
     """Render the compose markdown body to a SAFE HTML fragment for the email's
     text/html part. Everything is HTML-escaped FIRST (so a pasted <script> /
@@ -1557,7 +1616,7 @@ def _md_to_email_html(text: str) -> str:
         s = re.sub(r"~~([^~]+)~~", r"<del>\1</del>", s)
         s = re.sub(r"`([^`]+)`", r"<code>\1</code>", s)
         # links: text + http(s) url only (escape() already neutralised quotes)
-        s = re.sub(r"\[([^\]]+)\]\((https?://[^)\s]+)\)", r'<a href="\2">\1</a>', s)
+        s = _md_links_to_html(s)
         return s
 
     parts: list[str] = []
@@ -1776,11 +1835,6 @@ def setup_email_routes():
             return None
         return v[1]
 
-    # Stale-while-revalidate snapshot store: a last-good list response per key
-    # with NO TTL, so a list view can render instantly from a snapshot (or the
-    # durable index) and refresh in the background instead of blocking ~2-5 s on
-    # IMAP. Kept small and separate from the TTL cache so invalidation of the
-    # fresh cache never removes the fallback snapshot.
     _LIST_STALE = {}
 
     def _list_stale_get(key):
@@ -4012,7 +4066,7 @@ def setup_email_routes():
             _row, att = fixture_att
             filename = str(att.get("filename") or f"attachment-{index}.txt")
             safe_name = re.sub(r"[^\w\s\-.]", "_", filename).strip() or f"attachment-{index}.txt"
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             target_dir.mkdir(parents=True, exist_ok=True)
             filepath = target_dir / safe_name
             filepath.write_bytes(str(att.get("content") or "").encode("utf-8"))
@@ -4031,7 +4085,7 @@ def setup_email_routes():
             msg = email_mod.message_from_bytes(raw)
 
             # Extract to a per-email folder
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             filepath = _extract_attachment_to_disk(msg, index, target_dir)
             if not filepath:
                 return {"error": f"Attachment index {index} not found"}
@@ -4063,7 +4117,7 @@ def setup_email_routes():
             if not attachments:
                 raise HTTPException(status_code=404, detail="No downloadable attachments")
 
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             zip_buf = io.BytesIO()
             used_names: dict[str, int] = {}
             with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
@@ -4203,7 +4257,7 @@ def setup_email_routes():
                                     att for att in _list_attachments_from_msg(msg)
                                     if not _is_likely_signature_image_attachment(att)
                                 ]
-                                target_dir = attachment_extract_dir(folder, uid)
+                                target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=acct)
                                 for att in attachments:
                                     idx = att.get("index")
                                     if idx is None:
@@ -4233,6 +4287,414 @@ def setup_email_routes():
             headers={"Content-Disposition": f'attachment; filename="{zip_name}"'},
         )
 
+    @router.post("/messages-download-bulk")
+    async def download_bulk_messages(request: Request, owner: str = Depends(require_owner)):
+        """Export selected original email messages as .eml files in a ZIP."""
+        try:
+            payload = await request.json()
+        except Exception:
+            raise HTTPException(status_code=400, detail="Invalid JSON body")
+        messages = payload.get("messages") if isinstance(payload, dict) else None
+        if not isinstance(messages, list) or not messages:
+            raise HTTPException(status_code=400, detail="No emails selected")
+        if len(messages) > 250:
+            raise HTTPException(status_code=400, detail="Too many emails selected")
+
+        category = str(payload.get("category") or "selected").strip().lower()
+        category = re.sub(r"[^a-z0-9_-]+", "-", category).strip("-") or "selected"
+        selected_dates = sorted({
+            str(row.get("date") or "").strip()
+            for row in messages if isinstance(row, dict)
+            and re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(row.get("date") or "").strip())
+        })
+        date_span = ""
+        if selected_dates:
+            date_span = selected_dates[0] if len(selected_dates) == 1 else f"{selected_dates[0]} - {selected_dates[-1]}"
+        export_folder = f"emails-{category}" + (f" ({date_span})" if date_span else "")
+
+        # 512 MiB safety cap prevents unbounded memory growth when exporting
+        # many large messages.
+        max_total_bytes = 512 * 1024 * 1024
+        max_message_bytes = 64 * 1024 * 1024
+        spool_threshold_bytes = 32 * 1024 * 1024
+        used_names: set[str] = set()
+        exported = 0
+        skipped = 0
+        total_bytes = 0
+        validated_accounts: set[str] = set()
+
+        def _email_filename(row: dict, uid: str) -> str:
+            date = str(row.get("date") or "").strip()
+            date = date if re.fullmatch(r"\d{4}-\d{2}-\d{2}", date) else ""
+            subject = str(row.get("subject") or "email").strip()
+            subject = re.sub(r'[<>:"/\\|?*\x00-\x1f\x7f]+', "_", subject)
+            subject = re.sub(r"\s+", " ", subject).strip(" .")[:100] or "email"
+            safe_uid = re.sub(r"[^A-Za-z0-9_-]+", "_", uid).strip("_") or "message"
+            prefix = f"{date} - " if date else ""
+            return _safe_attachment_zip_name(f"{prefix}{subject} - {safe_uid}.eml", "email.eml")
+
+        def _fixture_message_bytes(fixture: dict, uid: str, folder: str) -> bytes:
+            msg = EmailMessage()
+            sender_name = str(fixture.get("from_name") or "").strip()
+            sender_address = str(fixture.get("from_address") or "").strip()
+            sender = email.utils.formataddr((sender_name, sender_address)) if sender_address else sender_name
+            for key, value in (
+                ("From", sender),
+                ("To", fixture.get("to") or owner),
+                ("Subject", fixture.get("subject") or "(no subject)"),
+                ("Message-ID", fixture.get("message_id") or f"<fixture-{uid}@mail.local>"),
+            ):
+                clean_value = re.sub(r"[\r\n]+", " ", str(value or "")).strip()
+                if clean_value:
+                    msg[key] = clean_value
+            raw_date = str(fixture.get("date") or "").strip()
+            if raw_date:
+                try:
+                    parsed_date = datetime.fromisoformat(raw_date.replace("Z", "+00:00"))
+                    if parsed_date.tzinfo is None:
+                        parsed_date = parsed_date.replace(tzinfo=timezone.utc)
+                except Exception:
+                    try:
+                        parsed_date = email.utils.parsedate_to_datetime(raw_date)
+                    except Exception:
+                        parsed_date = None
+                if parsed_date is not None:
+                    msg["Date"] = email.utils.format_datetime(parsed_date)
+            msg.set_content(str(fixture.get("body") or ""))
+            for attachment in fixture.get("attachments") or []:
+                try:
+                    index = int(attachment.get("index"))
+                    source = _fixture_attachment_source(uid, index, owner, folder)
+                    if not source:
+                        continue
+                    _row, raw_attachment = source
+                    content_type = str(raw_attachment.get("content_type") or "application/octet-stream")
+                    maintype, _, subtype = content_type.partition("/")
+                    if not subtype:
+                        maintype, subtype = "application", "octet-stream"
+                    msg.add_attachment(
+                        str(raw_attachment.get("content") or "").encode("utf-8"),
+                        maintype=maintype,
+                        subtype=subtype,
+                        filename=str(raw_attachment.get("filename") or attachment.get("filename") or f"attachment-{index}"),
+                    )
+                except Exception:
+                    logger.debug("Skipping malformed fixture email attachment uid=%s", uid, exc_info=True)
+            return msg.as_bytes(policy=email_policy.SMTP)
+
+        # Optimization: single message exports return raw .eml directly without ZIP compression.
+        if len(messages) == 1:
+            row = messages[0]
+            if not isinstance(row, dict):
+                raise HTTPException(status_code=404, detail="No selected emails could be exported")
+            uid = str(row.get("uid") or "").strip()
+            if not uid:
+                raise HTTPException(status_code=404, detail="No selected emails could be exported")
+            folder = str(row.get("folder") or "INBOX").strip() or "INBOX"
+            account_id = str(row.get("account_id") or row.get("account") or "").strip() or None
+            if account_id:
+                _assert_owns_account(account_id, owner)
+
+            filename = _email_filename(row, uid)
+            message_bytes: bytes | None = None
+
+            fixture = _fixture_email_read(uid, folder, owner)
+            if fixture is not None and not fixture.get("error"):
+                message_bytes = _fixture_message_bytes(fixture, uid, folder)
+            elif re.fullmatch(r"\d+", uid):
+                try:
+                    with _imap(account_id, owner=owner) as conn:
+                        conn.select(_q(folder), readonly=True)
+                        uid_b = uid.encode()
+                        st_sz, sz_data = None, None
+                        try:
+                            st_sz, sz_data = _imap_uid_fetch(conn, uid_b, "(UID RFC822.SIZE)")
+                        except Exception as sz_err:
+                            logger.warning("Single message size preflight query error for uid=%s: %s", uid, sz_err)
+
+                        if st_sz != "OK":
+                            logger.warning("Single message size preflight status=%s for uid=%s", st_sz, uid)
+                            raise HTTPException(status_code=502, detail="Unable to verify email size safety")
+
+                        if not sz_data:
+                            # Message does not exist in mailbox (missing mail)
+                            raise HTTPException(status_code=404, detail="No selected emails could be exported")
+
+                        sz: int | None = None
+                        for item in sz_data:
+                            meta = item[0] if isinstance(item, tuple) else item
+                            if isinstance(meta, bytes):
+                                u = _uid_from_fetch_meta(meta)
+                                if u == uid:
+                                    sz = _size_from_fetch_meta(meta)
+                                    break
+
+                        if sz is None or sz < 0:
+                            logger.warning("Single message size preflight returned invalid/malformed size for uid=%s", uid)
+                            raise HTTPException(status_code=502, detail="Invalid email size metadata from server")
+
+                        if sz > max_message_bytes or sz > max_total_bytes:
+                            raise HTTPException(status_code=413, detail="Export too large")
+
+                        # Preflight size safety established; perform RFC822 fetch.
+                        st, msg_data = _imap_uid_fetch(conn, uid_b, "(UID RFC822)")
+                        if st == "OK" and msg_data:
+                            for meta, raw in _group_uid_fetch_records(msg_data):
+                                if _uid_from_fetch_meta(meta) == uid and raw:
+                                    message_bytes = bytes(raw)
+                                    break
+                except HTTPException:
+                    raise
+                except Exception as exc:
+                    logger.warning("Single email export failed for uid=%s folder=%s: %s", uid, folder, exc)
+
+            if not message_bytes:
+                raise HTTPException(status_code=404, detail="No selected emails could be exported")
+
+            if len(message_bytes) > max_message_bytes or len(message_bytes) > max_total_bytes:
+                raise HTTPException(status_code=413, detail="Export too large")
+
+            return StreamingResponse(
+                io.BytesIO(message_bytes),
+                media_type="message/rfc822",
+                headers={
+                    "Content-Disposition": f"attachment; filename=\"email.eml\"; filename*=UTF-8''{quote(filename, safe='')}",
+                    "X-Exported-Emails": "1",
+                    "X-Skipped-Emails": "0",
+                },
+            )
+
+        seen_items: set[tuple[str | None, str, str]] = set()
+        remote_groups: dict[tuple[str | None, str], dict[str, tuple[str, dict]]] = {}
+        spooled_file = tempfile.SpooledTemporaryFile(max_size=spool_threshold_bytes, mode="w+b")
+        response_started = False
+        try:
+            with zipfile.ZipFile(spooled_file, "w", compression=zipfile.ZIP_DEFLATED) as zf:
+                for row in messages:
+                    if not isinstance(row, dict):
+                        skipped += 1
+                        continue
+                    uid = str(row.get("uid") or "").strip()
+                    if not uid:
+                        skipped += 1
+                        continue
+                    folder = str(row.get("folder") or "INBOX").strip() or "INBOX"
+                    account_id = str(row.get("account_id") or row.get("account") or "").strip() or None
+                    if account_id and account_id not in validated_accounts:
+                        _assert_owns_account(account_id, owner)
+                        validated_accounts.add(account_id)
+
+                    item_key = (account_id, folder, uid)
+                    if item_key in seen_items:
+                        skipped += 1
+                        continue
+                    seen_items.add(item_key)
+
+                    fixture = _fixture_email_read(uid, folder, owner)
+                    if fixture is not None and not fixture.get("error"):
+                        filename = _email_filename(row, uid)
+                        unique_name = filename
+                        suffix = 2
+                        while unique_name.lower() in used_names:
+                            unique_name = f"{Path(filename).stem}-{suffix}.eml"
+                            suffix += 1
+                        used_names.add(unique_name.lower())
+                        message_bytes = _fixture_message_bytes(fixture, uid, folder)
+                        if len(message_bytes) > max_message_bytes:
+                            raise HTTPException(status_code=413, detail="Export too large")
+                        total_bytes += len(message_bytes)
+                        if total_bytes > max_total_bytes:
+                            raise HTTPException(status_code=413, detail="Export too large")
+                        try:
+                            zf.writestr(f"{export_folder}/{unique_name}", message_bytes)
+                        except Exception as zip_err:
+                            logger.error("ZIP entry write failed for fixture: %s", zip_err)
+                            raise HTTPException(status_code=500, detail="Failed to create export archive")
+                        exported += 1
+                        continue
+                    if not re.fullmatch(r"\d+", uid):
+                        skipped += 1
+                        continue
+                    group = remote_groups.setdefault((account_id, folder), {})
+                    group.setdefault(uid, (_email_filename(row, uid), row))
+
+                preflight_error: HTTPException | None = None
+
+                # Fetch original RFC822 data in batches and store it byte-for-byte.
+                # Note on IMAP memory-safety: Python's standard imaplib buffers socket responses into
+                # memory before returning literals to application code. Preflight verification ensures
+                # we do not intentionally fetch messages exceeding limits, but an untrusted or non-compliant
+                # IMAP server sending unexpectedly large literals cannot be bounded at the transport level
+                # without custom socket streaming. Post-fetch checks guard application-level storage.
+                for (account_id, folder), rows_by_uid in remote_groups.items():
+                    rows = list(rows_by_uid.items())
+                    remaining_rows = len(rows)
+                    try:
+                        with _imap(account_id, owner=owner) as conn:
+                            conn.select(_q(folder), readonly=True)
+
+                            # Size preflight query to catch oversized exports early and bound batch sizes.
+                            uid_set_all = ",".join(uid for uid, _ in rows).encode()
+                            st_sz, sz_data = None, None
+                            try:
+                                st_sz, sz_data = _imap_uid_fetch(conn, uid_set_all, "(UID RFC822.SIZE)")
+                            except Exception as sz_err:
+                                logger.warning("IMAP size preflight query failed for folder=%s: %s", folder, sz_err)
+
+                            if st_sz != "OK":
+                                logger.warning("IMAP size preflight status=%s for folder=%s account=%s", st_sz, folder, account_id)
+                                preflight_error = HTTPException(status_code=502, detail="Unable to verify email size safety")
+                                skipped += remaining_rows
+                                remaining_rows = 0
+                                continue
+
+                            if not sz_data:
+                                # All requested UIDs in this mailbox are missing on server
+                                skipped += remaining_rows
+                                remaining_rows = 0
+                                continue
+
+                            reported_sizes: dict[str, int | None] = {}
+                            for item in sz_data:
+                                meta = item[0] if isinstance(item, tuple) else item
+                                if isinstance(meta, bytes):
+                                    u = _uid_from_fetch_meta(meta)
+                                    if u:
+                                        reported_sizes[u] = _size_from_fetch_meta(meta)
+
+                            # Categorize requested UIDs
+                            malformed_uids = [
+                                uid for uid, _ in rows
+                                if uid in reported_sizes and (reported_sizes[uid] is None or reported_sizes[uid] < 0)
+                            ]
+                            if malformed_uids:
+                                logger.warning("IMAP size preflight returned invalid/malformed size for uids=%s in folder=%s", malformed_uids, folder)
+                                preflight_error = HTTPException(status_code=502, detail="Invalid email size metadata from server")
+                                skipped += remaining_rows
+                                remaining_rows = 0
+                                continue
+
+                            # Identify missing mail (requested UIDs not returned in preflight)
+                            missing_uids = {uid for uid, _ in rows if uid not in reported_sizes}
+                            if missing_uids:
+                                skipped += len(missing_uids)
+                                remaining_rows -= len(missing_uids)
+                                rows = [item for item in rows if item[0] not in missing_uids]
+
+                            if not rows:
+                                continue
+
+                            size_by_uid: dict[str, int] = {
+                                uid: reported_sizes[uid] for uid, _ in rows
+                            }
+
+                            for u, s in size_by_uid.items():
+                                if s > max_message_bytes:
+                                    raise HTTPException(status_code=413, detail="Export too large")
+                            if total_bytes + sum(size_by_uid.values()) > max_total_bytes:
+                                raise HTTPException(status_code=413, detail="Export too large")
+
+                            # Form batches with at most 20 messages and capped at 10 MiB payload.
+                            batches: list[list[tuple[str, tuple[str, dict]]]] = []
+                            cur_batch: list[tuple[str, tuple[str, dict]]] = []
+                            cur_batch_bytes = 0
+                            max_batch_bytes = 10 * 1024 * 1024
+                            for item in rows:
+                                uid_val = item[0]
+                                sz = size_by_uid[uid_val]
+                                if cur_batch and (
+                                    len(cur_batch) >= 20 or
+                                    (cur_batch_bytes + sz > max_batch_bytes)
+                                ):
+                                    batches.append(cur_batch)
+                                    cur_batch = []
+                                    cur_batch_bytes = 0
+                                cur_batch.append(item)
+                                cur_batch_bytes += sz
+                            if cur_batch:
+                                batches.append(cur_batch)
+
+                            for batch in batches:
+                                wanted = {uid: filename for uid, (filename, _row) in batch}
+                                uid_set = ",".join(wanted).encode()
+                                status, msg_data = _imap_uid_fetch(conn, uid_set, "(UID RFC822)")
+                                if status != "OK" or not msg_data:
+                                    skipped += len(batch)
+                                    remaining_rows -= len(batch)
+                                    continue
+                                found: set[str] = set()
+                                for meta, raw_message in _group_uid_fetch_records(msg_data):
+                                    uid = _uid_from_fetch_meta(meta)
+                                    if not uid or uid not in wanted or not raw_message:
+                                        continue
+                                    if uid in found:
+                                        # P2: Reject duplicate IMAP response records for an already-processed UID
+                                        logger.debug("Ignoring duplicate IMAP FETCH record for uid=%s in folder=%s", uid, folder)
+                                        continue
+                                    filename = wanted[uid]
+                                    unique_name = filename
+                                    suffix = 2
+                                    while unique_name.lower() in used_names:
+                                        unique_name = f"{Path(filename).stem}-{suffix}.eml"
+                                        suffix += 1
+                                    used_names.add(unique_name.lower())
+                                    message_bytes = bytes(raw_message)
+                                    if len(message_bytes) > max_message_bytes:
+                                        raise HTTPException(status_code=413, detail="Export too large")
+                                    total_bytes += len(message_bytes)
+                                    if total_bytes > max_total_bytes:
+                                        raise HTTPException(status_code=413, detail="Export too large")
+                                    try:
+                                        zf.writestr(f"{export_folder}/{unique_name}", message_bytes)
+                                    except Exception as zip_err:
+                                        logger.error("ZIP entry write failed for remote message: %s", zip_err)
+                                        raise HTTPException(status_code=500, detail="Failed to create export archive")
+                                    found.add(uid)
+                                    exported += 1
+                                    remaining_rows -= 1
+                                missing = len(batch) - len(found)
+                                skipped += len(batch) - len(found)
+                                remaining_rows -= missing
+                    except HTTPException:
+                        raise
+                    except Exception as exc:
+                        skipped += remaining_rows
+                        logger.warning("Bulk email export skipped mailbox folder=%s account=%s count=%s: %s", folder, account_id, remaining_rows, exc)
+
+            if preflight_error is not None:
+                raise preflight_error
+
+            if exported <= 0:
+                raise HTTPException(status_code=404, detail="No selected emails could be exported")
+
+            spooled_file.seek(0)
+            zip_name = _safe_attachment_zip_name(f"{export_folder}.zip", "selected-emails.zip")
+            logger.info("Bulk email export complete owner=%s selected=%s exported=%s skipped=%s", owner, len(messages), exported, skipped)
+
+            def _iter_spooled(f, chunk_size=64 * 1024):
+                try:
+                    while True:
+                        chunk = f.read(chunk_size)
+                        if not chunk:
+                            break
+                        yield chunk
+                finally:
+                    f.close()
+
+            response_started = True
+            return StreamingResponse(
+                _iter_spooled(spooled_file),
+                media_type="application/zip",
+                headers={
+                    "Content-Disposition": f'attachment; filename="{zip_name}"',
+                    "X-Exported-Emails": str(exported),
+                    "X-Skipped-Emails": str(skipped),
+                },
+            )
+        finally:
+            if not response_started:
+                spooled_file.close()
+
     @router.get("/inline-image/{uid}")
     async def inline_image(
         uid: str,
@@ -4268,7 +4730,7 @@ def setup_email_routes():
                     continue
                 if not ct.lower().startswith("image/"):
                     raise HTTPException(status_code=415, detail="Content-ID is not an image")
-                target_dir = attachment_extract_dir(folder, uid)
+                target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
                 filepath = _extract_attachment_to_disk(msg, idx, target_dir)
                 if not filepath:
                     raise HTTPException(status_code=404, detail="Inline image not found")
@@ -4306,7 +4768,7 @@ def setup_email_routes():
             raw = msg_data[0][1]
             msg = email_mod.message_from_bytes(raw)
 
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             filepath = _extract_attachment_to_disk(msg, index, target_dir)
             if not filepath:
                 return {"error": f"Attachment index {index} not found"}
@@ -4634,7 +5096,9 @@ def setup_email_routes():
                 try:
                     content = filepath.read_text(encoding="utf-8", errors="replace")
                 except Exception as e:
-                    return {"error": f"Failed to read text file: {e}", "filename": base}
+                    # OSError text includes the server-side extraction path.
+                    logger.warning("Failed to read text attachment %s: %s", base, e)
+                    return {"error": "Failed to read text file", "filename": base}
                 doc_id = _create_markdown_doc(content, "Imported from email attachment")
                 return {"doc_id": doc_id, "filename": filepath.name}
 
@@ -4651,7 +5115,7 @@ def setup_email_routes():
             _row, att = fixture_att
             filename = str(att.get("filename") or f"attachment-{index}.txt")
             safe_name = re.sub(r"[^\w\s\-.]", "_", filename).strip() or f"attachment-{index}.txt"
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             target_dir.mkdir(parents=True, exist_ok=True)
             filepath = target_dir / safe_name
             filepath.write_bytes(str(att.get("content") or "").encode("utf-8"))
@@ -4665,7 +5129,7 @@ def setup_email_routes():
             raw = msg_data[0][1]
             msg = email_mod.message_from_bytes(raw)
 
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             filepath = _extract_attachment_to_disk(msg, index, target_dir)
             if not filepath:
                 return {"error": f"Attachment index {index} not found"}
@@ -5319,7 +5783,7 @@ def setup_email_routes():
                 return {"success": False, "error": "Email not found"}
             raw = msg_data[0][1]
             msg = email_mod.message_from_bytes(raw)
-            target_dir = attachment_extract_dir(folder, uid)
+            target_dir = attachment_extract_dir(folder, uid, owner=owner, account_id=account_id)
             filepath = _extract_attachment_to_disk(msg, index, target_dir)
             if not filepath:
                 return {"success": False, "error": f"Attachment index {index} not found"}
@@ -6769,6 +7233,12 @@ def setup_email_routes():
         if account_id:
             _assert_owns_account(account_id, owner)
 
+        for field, label in (("imap_host", "IMAP"), ("smtp_host", "SMTP")):
+            if data.get(field):
+                host_error = await run_in_threadpool(_check_mail_host, data[field], owner, label)
+                if host_error:
+                    return {"ok": False, "error": host_error}
+
         # Non-reply automation flags stay global. Away/auto-reply settings are
         # account-scoped when an account_id is supplied by the Email Settings UI.
         settings = _load_settings()
@@ -6909,6 +7379,11 @@ def setup_email_routes():
         smtp_port, port_err = _coerce_port(data.get("smtp_port"), 465)
         if port_err:
             return {"ok": False, "error": port_err}
+        for field, label in (("imap_host", "IMAP"), ("smtp_host", "SMTP")):
+            if data.get(field):
+                host_error = await run_in_threadpool(_check_mail_host, data[field], owner, label)
+                if host_error:
+                    return {"ok": False, "error": host_error}
         db = SessionLocal()
         try:
             _lock_email_account_owner_mutation(db, owner)
@@ -6954,6 +7429,11 @@ def setup_email_routes():
         """Update an email account. Passwords only overwrite if non-empty."""
         # Path param account_id — dep validated via Query, re-check the path-param value.
         _assert_owns_account(account_id, owner)
+        for field, label in (("imap_host", "IMAP"), ("smtp_host", "SMTP")):
+            if data.get(field):
+                host_error = await run_in_threadpool(_check_mail_host, data[field], owner, label)
+                if host_error:
+                    return {"ok": False, "error": host_error}
         from core.database import SessionLocal, EmailAccount
         db = SessionLocal()
         try:
@@ -7026,6 +7506,158 @@ def setup_email_routes():
         finally:
             db.close()
 
+    def _test_mail_connections(body: dict, owner: str) -> dict:
+        """Blocking IMAP/SMTP connection test for /accounts/test.
+
+        Runs in the threadpool: each probe can block for its full socket
+        timeout, which on the event loop stalled every other request. The
+        connections go through the outbound mail address policy for `owner`
+        (the user who chose the hosts)."""
+        imap_result = {"ok": False}
+        smtp_result = None
+
+        imap_host = (body.get("imap_host") or "").strip()
+        imap_port, imap_port_err = _coerce_port(body.get("imap_port"), 993)
+        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 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)
+            #   STARTTLS off + port 993 → IMAP4_SSL (implicit SSL, "IMAPS")
+            #   STARTTLS off + any other port → plain IMAP4 (no encryption)
+            # Without the last branch, local servers exposed on a non-993
+            # 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,
+                    "owner": owner,
+                }
+                if google_ssl_context:
+                    imap_kwargs["ssl_context"] = google_ssl_context
+                conn = _open_imap_connection(
+                    imap_host,
+                    imap_port,
+                    **imap_kwargs,
+                )
+                try:
+                    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()
+                    except Exception: pass
+            except Exception as e:
+                imap_result = {"ok": False, "error": _mail_connection_test_error("IMAP", imap_host, e)}
+
+        smtp_host = (body.get("smtp_host") or "").strip()
+        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_kwargs = (
+                        {"context": google_ssl_context}
+                        if google_ssl_context
+                        else {}
+                    )
+                    smtp = _PolicySMTP_SSL(
+                        smtp_host,
+                        smtp_port,
+                        timeout=10,
+                        block_private=_mail_private_blocked(owner),
+                        **smtp_kwargs,
+                    )
+                else:
+                    smtp = _PolicySMTP(smtp_host, smtp_port, timeout=10, block_private=_mail_private_blocked(owner))
+                    if smtp_security == "starttls":
+                        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}
+            except Exception as e:
+                smtp_result = {"ok": False, "error": _mail_connection_test_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"]),
+            "imap": imap_result,
+            "smtp": smtp_result,
+        }
+
     @router.post("/accounts/test")
     async def test_account_config(req: Request, owner: str = Depends(require_user)):
         """Try to actually connect to the provided IMAP (and optionally SMTP)
@@ -7089,148 +7721,7 @@ def setup_email_routes():
             # 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
-
-        imap_host = (body.get("imap_host") or "").strip()
-        imap_port, imap_port_err = _coerce_port(body.get("imap_port"), 993)
-        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 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)
-            #   STARTTLS off + port 993 → IMAP4_SSL (implicit SSL, "IMAPS")
-            #   STARTTLS off + any other port → plain IMAP4 (no encryption)
-            # Without the last branch, local servers exposed on a non-993
-            # 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,
-                    **imap_kwargs,
-                )
-                try:
-                    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()
-                    except Exception: pass
-            except Exception as e:
-                imap_result = {"ok": False, "error": _friendly_email_auth_error("IMAP", imap_host, e)}
-
-        smtp_host = (body.get("smtp_host") or "").strip()
-        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_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":
-                        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}
-            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"]),
-            "imap": imap_result,
-            "smtp": smtp_result,
-        }
+        return await run_in_threadpool(_test_mail_connections, body, owner)
 
     @router.post("/accounts/{account_id}/set-default")
     async def set_default_account(account_id: str, owner: str = Depends(require_user)):

routes/gallery/gallery_routes.py

@@ -1914,7 +1914,7 @@ def setup_gallery_routes() -> APIRouter:
                 raise HTTPException(504, "OpenAI inpaint timed out (300s)")
             except httpx.TransportError as exc:
                 logger.warning("inpaint_proxy: OpenAI endpoint unreachable: %s", exc)
-                raise HTTPException(502, f"Image server unreachable ({base}): {type(exc).__name__}")
+                raise HTTPException(502, f"Image server unreachable: {type(exc).__name__}")
 
         # Self-hosted diffusion server path. Newer Odysseus image
         # wrappers expose the OpenAI-compatible /v1/images/edits
@@ -2059,7 +2059,7 @@ def setup_gallery_routes() -> APIRouter:
             raise HTTPException(504, "Inpaint request timed out (240s)")
         except httpx.TransportError as exc:
             logger.warning("inpaint_proxy: image endpoint unreachable: %s", exc)
-            raise HTTPException(502, f"Image server unreachable ({base}): {type(exc).__name__}")
+            raise HTTPException(502, f"Image server unreachable: {type(exc).__name__}")
         except HTTPException:
             raise
         except Exception:
@@ -2563,7 +2563,13 @@ def setup_gallery_routes() -> APIRouter:
             from src.ai_interaction import do_generate_image
         except Exception:
             raise HTTPException(502, "Image generation is unavailable")
-        content = _json.dumps({"prompt": prompt, "size": size})
+        # The editor's model picker may choose a specific (cloud) image model;
+        # forward it so settings defaults only apply when none is selected.
+        model_choice = str((body or {}).get("model") or "").strip()
+        payload = {"prompt": prompt, "size": size}
+        if model_choice:
+            payload["model"] = model_choice
+        content = _json.dumps(payload)
         result = await do_generate_image(content, owner=user, incognito=True)
         if not isinstance(result, dict) or result.get("error"):
             msg = (result.get("error") if isinstance(result, dict) else None) or "Image generation failed"

routes/hwfit_routes.py

@@ -5,8 +5,9 @@ import shlex
 import subprocess
 from copy import deepcopy
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 
+from core.middleware import require_admin
 from core.platform_compat import run_ssh_command
 from routes._validators import validate_remote_host, validate_ssh_port
 
@@ -198,7 +199,11 @@ def _inspect_model_path(model_path: str, host: str = "", ssh_port: str = "") ->
 
 
 def setup_hwfit_routes():
-    router = APIRouter(prefix="/api/hwfit", tags=["hwfit"])
+    # Admin-gated like the Cookbook routes this backs (/api/cookbook/gpus does
+    # the same SSH host probing): `host`/`ssh_port` make the server SSH out with
+    # its own key to any syntactically valid host, and `model_path` probes the
+    # host filesystem. Single-user mode and internal tool calls still pass.
+    router = APIRouter(prefix="/api/hwfit", tags=["hwfit"], dependencies=[Depends(require_admin)])
 
     @router.get("/system")
     def get_system(host: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False):
@@ -218,7 +223,7 @@ def setup_hwfit_routes():
         fresh=true bypasses the hardware-detection cache."""
         from services.hwfit.hardware import detect_system
         from services.hwfit.fit import rank_models
-        from services.hwfit.models import get_models, model_catalog_path, refresh_dynamic_catalogs
+        from services.hwfit.models import get_models, refresh_dynamic_catalogs
         host, ssh_port = _validate_detection_target(host, ssh_port)
         system = deepcopy(detect_system(host=host, ssh_port=ssh_port, platform=platform, fresh=fresh))
         if system.get("error"):
@@ -233,7 +238,8 @@ def setup_hwfit_routes():
             return {
                 "system": system,
                 "models": [],
-                "error": f"Model catalog missing or empty: {model_catalog_path()}",
+                "error": "Model catalog is empty. Use Rescan in Cookbook while online to populate runtime model data; offline recommendations require a previously populated cache.",
+                **({"catalog_refresh": catalog_refresh} if catalog_refresh is not None else {}),
             }
 
         if ignore_detected_gpu:

routes/mcp/mcp_routes.py

@@ -114,6 +114,18 @@ def _mcp_oauth_redirect_uri() -> str:
     return REDIRECT_URI
 
 
+def _is_legacy_google_flow(identifier: str) -> bool:
+    """Determine whether an identifier is the server_id of a configured Google OAuth MCP server."""
+    if not identifier:
+        return False
+    db = SessionLocal()
+    try:
+        srv = db.query(McpServer).filter(McpServer.id == identifier).first()
+        return bool(srv and srv.oauth_config)
+    finally:
+        db.close()
+
+
 def setup_mcp_routes(mcp_manager: McpManager):
     """Setup MCP routes with the provided manager."""
 
@@ -487,10 +499,14 @@ def setup_mcp_routes(mcp_manager: McpManager):
             db.close()
 
     @router.get("/oauth/callback")
-    async def oauth_callback(code: str, state: str, request: Request):
+    async def oauth_callback(code: str = "", state: str = "", request: Request = None):
         """Handle OAuth callback. Generic MCP OAuth flows resolve via the
         pending-state registry; Google flows fall through to the legacy path."""
-        require_admin(request)
+        if not state:
+            return HTMLResponse(_oauth_result_page("Error", "No OAuth state parameter found in the callback URL."), status_code=400)
+        if not code:
+            return HTMLResponse(_oauth_result_page("Error", "No authorization code found in the URL. Make sure you copied the full URL from your browser."), status_code=400)
+
         from src.mcp_oauth import resolve_pending
         if resolve_pending(state, code):
             return HTMLResponse(_oauth_result_page(
@@ -498,8 +514,16 @@ def setup_mcp_routes(mcp_manager: McpManager):
                 "The MCP server is connecting. You can close this window and return to Odysseus.",
                 success=True,
             ))
-        # Legacy Google path: state is the server_id
-        return await _exchange_and_connect(state, code, request)
+
+        # Check if this callback belongs to a legacy Google flow vs expired generic flow
+        if _is_legacy_google_flow(state):
+            require_admin(request)
+            return await _exchange_and_connect(state, code, request)
+
+        return HTMLResponse(_oauth_result_page(
+            "Authorization Expired",
+            "This authorization round has expired. Press Reconnect in Odysseus and try again.",
+        ), status_code=400)
 
     @router.post("/oauth/exchange/{server_id}")
     async def oauth_exchange(server_id: str, request: Request, callback_url: str = Form(...)):
@@ -514,18 +538,28 @@ def setup_mcp_routes(mcp_manager: McpManager):
         except Exception:
             return HTMLResponse(_oauth_result_page("Error", "Invalid URL format."), status_code=400)
 
+        state = params.get("state", [None])[0]
+        if not state:
+            return HTMLResponse(_oauth_result_page("Error", "No OAuth state parameter found in the callback URL."), status_code=400)
+
         # Generic MCP OAuth: if the pasted URL carries a state we are waiting on,
         # resolve it directly (the background connect finishes the handshake).
-        state = params.get("state", [None])[0]
         from src.mcp_oauth import resolve_pending
-        if state and resolve_pending(state, code):
+        if resolve_pending(state, code):
             return HTMLResponse(_oauth_result_page(
                 "Authorization Successful",
                 "The MCP server is connecting. You can close this window and return to Odysseus.",
                 success=True,
             ))
 
-        return await _exchange_and_connect(server_id, code, request)
+        # Check if this exchange belongs to a legacy Google flow vs expired generic flow
+        if _is_legacy_google_flow(server_id):
+            return await _exchange_and_connect(server_id, code, request)
+
+        return HTMLResponse(_oauth_result_page(
+            "Authorization Expired",
+            "This authorization round has expired. Press Reconnect in Odysseus and try again.",
+        ), status_code=400)
 
     async def _exchange_and_connect(server_id: str, code: str, request: Request):
         """Exchange auth code for tokens and connect the MCP server."""

routes/model_routes.py

@@ -47,6 +47,7 @@ from src.endpoint_resolver import (
     build_headers,
 )
 from src.auth_helpers import _auth_disabled, effective_user, owner_filter
+from src.url_safety import check_outbound_url
 
 logger = logging.getLogger(__name__)
 
@@ -2172,6 +2173,19 @@ def setup_model_routes(model_discovery):
         # keep those container-local when the frontend marks them as such.
         base_url = _rewrite_loopback_for_docker(base_url, container_local=_truthy(container_local))
 
+        # SSRF hardening: validate the user-supplied URL before any outbound
+        # request. Local-first means loopback/LAN endpoints are allowed by
+        # default; non-HTTP(S) schemes and the cloud metadata range are always
+        # rejected. Set MODELENDPOINT_BLOCK_PRIVATE_IPS=true for full lockdown.
+        # Note: the outbound fetches rely on httpx's default
+        # follow_redirects=False — a redirect target is never dereferenced.
+        ok, reason = check_outbound_url(
+            base_url,
+            block_private=os.getenv("MODELENDPOINT_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+        )
+        if not ok:
+            raise HTTPException(400, f"Rejected endpoint URL: {reason}")
+
         is_featherless = _host_match(base_url, "featherless.ai") or _safe_detect_provider(base_url) == "featherless"
         # Auto-generate name from URL if not provided
         if not name.strip():
@@ -2408,6 +2422,16 @@ def setup_model_routes(model_discovery):
         from src.endpoint_resolver import resolve_url
         base_url = resolve_url(base_url)
         base_url = _rewrite_loopback_for_docker(base_url)
+        # SSRF hardening: validate the user-supplied URL before any outbound
+        # request. Local-first means loopback/LAN endpoints are allowed by
+        # default; non-HTTP(S) schemes and the cloud metadata range are always
+        # rejected. Set MODELENDPOINT_BLOCK_PRIVATE_IPS=true for full lockdown.
+        ok, reason = check_outbound_url(
+            base_url,
+            block_private=os.getenv("MODELENDPOINT_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+        )
+        if not ok:
+            raise HTTPException(400, f"Rejected endpoint URL: {reason}")
         requested_kind = _normalize_endpoint_kind(endpoint_kind)
         is_featherless = _host_match(base_url, "featherless.ai") or _safe_detect_provider(base_url) == "featherless"
         if is_featherless and requested_kind == "auto":
@@ -2898,6 +2922,16 @@ def setup_model_routes(model_discovery):
                             _new_base = _new_base[: -len(_suffix)].rstrip("/")
                     _new_base = _normalize_base(_new_base)
                     if _new_base:
+                        # SSRF hardening: PATCH rewrites the stored base_url, so
+                        # it needs the same guard as POST create/test — no
+                        # bypass through the update path. Follow-up fetches
+                        # rely on httpx's default follow_redirects=False.
+                        ok, reason = check_outbound_url(
+                            _new_base,
+                            block_private=os.getenv("MODELENDPOINT_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+                        )
+                        if not ok:
+                            raise HTTPException(400, f"Rejected endpoint URL: {reason}")
                         ep.base_url = _new_base
             else:
                 ep.is_enabled = not ep.is_enabled