fix(email): guard user-supplied IMAP/SMTP hosts against SSRF (#6355)

* fix(email): guard user-supplied IMAP/SMTP hosts against SSRF

POST /api/email/accounts/test opened raw imaplib/smtplib connections to
whatever host:port the request body named, and POST/PUT /api/email/accounts
and PUT /api/email/config persisted arbitrary hosts for the background
pollers to contact. All of these sit behind require_user/require_owner with
no admin gate, so any authenticated user could make the server connect to
loopback, LAN and cloud-metadata targets (169.254.169.254) and read the
first 200 characters of the remote banner back out of the friendly error.
THREAT_MODEL.md lists email as admin-only; the tool layer honours that but
the HTTP routes did not.

Add check_outbound_host() to src/url_safety.py, a bare-host companion to
check_outbound_url() with the same classification (link-local, multicast,
reserved and unspecified always rejected; private/loopback only with
block_private) plus localhost-name handling and an explicit policy for names
that do not resolve. Wire it into the four email routes via
_check_mail_host(): non-admin owners (and everyone when
EMAIL_BLOCK_PRIVATE_IPS=true) also lose private/loopback targets, admins
keep the local-first LAN/Dovecot setups. Unresolvable names pass through
because the connect that follows fails with the same DNS error.

Regression tests cover the helper's policy table and each route: the test
route rejects before any socket opens, save routes reject before touching
the DB, admins keep loopback unless locked down, and an update that fails
validation leaves the stored row untouched.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TwcwfY2uQMxNRsK3y6xBAp

* fix(email): harden IMAP/SMTP SSRF protection against rebinding, TOCTOU, and unvetted hosts

- P0: Implement safe_create_connection() in src/url_safety.py to resolve hostnames once, validate every returned address against SSRF policy, and connect directly to the pinned IP. This eliminates DNS rebinding and TOCTOU while preserving TLS SNI and certificate hostname verification across IMAP, IMAPS, STARTTLS, SMTP, and SMTP_SSL.
- P1: Ensure unresolvable or deferred hostnames cannot become an unrestricted future connection capability by enforcing runtime SSRF validation fail-closed on every connection, while maintaining compatibility with offline configurations.
- P2: Extend SSRF enforcement to the core connection layer (_open_imap_connection, _send_smtp_message, _imap_connect) using ContextVar owner scoping, protecting pre-existing DB rows, restored configs, background pollers, and scheduled operations.
- P3: Allow admin and single-user access to IPv6 loopback (::1, [::1]) for local mail servers (e.g. Dovecot) by checking loopback before is_reserved, while keeping link-local, cloud metadata (169.254.169.254, 100.100.100.200), and non-admin restrictions strictly enforced.
- Add comprehensive regression and adversarial test coverage in tests/test_email_account_host_ssrf.py.

* fix(email): reject unresolved saved hosts without stdlib monkeypatches

Use the existing explicit policy IMAP/SMTP classes and canonical connect_outbound_tcp helper. Remove process-wide stdlib assignments, context-variable plumbing, redundant aliases and resolver-signature retries. Reject unsafe or unresolved hosts before account/config persistence, keep DNS work off the event loop, return fixed client errors, identify Alibaba metadata accurately and reject invalid TCP ports. Validation: 654 relevant email, URL-safety and CodeQL regression tests passed, 3 skipped. Three unrelated browser checks were excluded after environment failures (missing Playwright browser and Chromium timeout). Running-app checks verified rejected create/update/config requests leave stored data unchanged, administrator LAN configuration still works, and metadata/non-admin loopback are blocked.

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: PR #6029 Reviewer <reviewer@odysseus-review.local>
Co-authored-by: Afonso Junior <chatgpt@yack.com.br>

src/url_safety.py

@@ -57,6 +57,10 @@ _SHARED_ADDRESS_SPACE_V4 = ipaddress.ip_network("100.64.0.0/10")
 # membership test excludes and which stays rejected as reserved.
 _NAT64_WELL_KNOWN_PREFIX_V6 = ipaddress.ip_network("64:ff9b::/96")
 
+_LOCALHOST_NAMES = frozenset({"localhost", "localhost.localdomain", "ip6-localhost", "ip6-loopback"})
+# Alibaba's metadata endpoint is shared address space, not link-local.
+_METADATA_IPS = {ipaddress.ip_address("100.100.100.200")}
+
 
 def _nat64_well_known_embedded_ipv4(
     ip: ipaddress._BaseAddress,
@@ -92,6 +96,8 @@ def _classify(ip: ipaddress._BaseAddress, *, block_private: bool) -> Optional[st
             if reason:
                 return f"NAT64 translated destination blocked: {reason}"
             return None
+    if ip in _METADATA_IPS:
+        return f"metadata endpoint blocked: {ip}"
     if ip.is_link_local:
         return f"link-local address blocked (SSRF metadata risk): {ip}"
     # IPv6 loopback is also reserved; its policy must match IPv4 loopback.
@@ -161,6 +167,58 @@ def check_outbound_url(
 
 
 
+def check_outbound_host(
+    host: str,
+    *,
+    block_private: bool = False,
+    resolver: Optional[Callable[[str], List[str]]] = None,
+) -> Tuple[bool, str]:
+    """Validate a bare mail hostname/IP, rejecting unresolved destinations.
+
+    Saving an unresolvable hostname is unsafe: another consumer of the stored
+    configuration may later connect after its DNS records change. Live mail
+    connections separately pin their sockets with connect_outbound_tcp.
+    """
+    if not isinstance(host, str):
+        return False, "host must be a string"
+    host = host.strip()
+    if not host:
+        return False, "host is required"
+    if any(ch.isspace() for ch in host) or any(ch in host for ch in "/\\@?#"):
+        return False, "host must be a bare hostname or IP address"
+    literal = host[1:-1] if host.startswith("[") and host.endswith("]") else host
+    try:
+        ip = ipaddress.ip_address(literal.split("%")[0])
+    except ValueError:
+        if ":" in literal:
+            return False, "host must be a bare hostname or IP address"
+    else:
+        reason = _classify(ip, block_private=block_private)
+        return (False, reason) if reason else (True, "ok")
+    name = host.lower().rstrip(".")
+    if name in _LOCALHOST_NAMES or name.endswith(".localhost"):
+        if block_private:
+            return False, f"private/shared/loopback address blocked: {host}"
+        return True, "ok"
+    try:
+        ips = (resolver or _default_resolver)(host)
+        if not ips:
+            return False, "host does not resolve to an IP"
+        saw_ip = False
+        for raw in ips:
+            # Malformed answers fail closed, including mixed valid/invalid ones.
+            ip = ipaddress.ip_address(raw.split("%")[0])
+            saw_ip = True
+            reason = _classify(ip, block_private=block_private)
+            if reason:
+                return False, reason
+    except Exception as exc:
+        return False, f"host does not resolve to valid IP addresses: {exc}"
+    if not saw_ip:
+        return False, "host does not resolve to an IP"
+    return True, "ok"
+
+
 class OutboundAddressBlocked(PermissionError):
     """A non-HTTP outbound host resolves into a disallowed address range."""
 
@@ -183,7 +241,15 @@ def connect_outbound_tcp(
     a rebinding name pass the check with a public answer and then connect to
     metadata/private space. TLS callers keep wrapping the returned socket with
     ``server_hostname=host``, so SNI and certificate checks are unchanged.
+
+    An injected resolver must implement socket.getaddrinfo's argument and
+    result contract. Resolver errors propagate without an unvetted retry.
     """
+    if isinstance(port, bool) or not isinstance(port, (int, str)):
+        raise ValueError("port must be an integer between 1 and 65535")
+    port = int(port)
+    if not 1 <= port <= 65535:
+        raise ValueError("port must be an integer between 1 and 65535")
     resolve = resolver or socket.getaddrinfo
     infos = resolve(host, port, 0, socket.SOCK_STREAM)
     for _family, _type, _proto, _canon, sockaddr in infos: