"""MX/domain-grouped batch verification — the throughput core of the engine. Opens one SMTP session per domain (not per address), probes a high-entropy gibberish address once to learn whether the domain is catch-all, caches that domain-level fact, and only then probes the real mailboxes. Why this matters: - A TCP + EHLO handshake dominates the cost of a probe, so reusing one session for many RCPTs is the difference between a 10k list taking minutes vs hours. - Catch-all is a *domain* property, so probing it once per domain (and caching it for 24-72h) removes a redundant probe for every other address there. - On a known catch-all domain no per-address RCPT can distinguish a real mailbox from a fake one, so the honest answer is "unverifiable" — returned without burning a connection. Every remote failure is data, never an exception: one dead host cannot abort a bulk job. """ from __future__ import annotations import asyncio import random import time from dataclasses import dataclass from typing import Callable, Iterable, Optional from ..models import SMTPResult from .mx_intel import classify_mx from .smtp_client import RcptOutcome, SmtpConnection DEFAULT_CATCH_ALL_TTL = 72 * 3600 # 72h — catch-all status is stable @dataclass class CatchAllCache: """Per-domain catch-all memory. `None` means "probe could not decide" and is deliberately NOT cached — a failed probe must not be remembered as a decision in either direction. """ ttl_seconds: int = DEFAULT_CATCH_ALL_TTL def __post_init__(self) -> None: self._store: dict[str, tuple[bool, float]] = {} def put(self, domain: str, value: Optional[bool]) -> None: if value is None: return self._store[domain.lower()] = (bool(value), time.time() + self.ttl_seconds) def get(self, domain: str) -> Optional[bool]: item = self._store.get(domain.lower()) if item is None: return None value, expires = item if time.time() >= expires: self._store.pop(domain.lower(), None) return None return value def clear(self) -> None: self._store.clear() def _gibberish(domain: str) -> str: token = f"{random.getrandbits(48):012x}" return f"xq{token}@{domain}" class MxBatchVerifier: """Verify many (email, mx_host) pairs with domain-level session reuse.""" def __init__( self, connection_factory: Optional[Callable[..., object]] = None, *, timeout: float = 10.0, helo_domain: str = "validator.local", mail_from: str = "probe@validator.local", catch_all_cache: Optional[CatchAllCache] = None, max_connections_per_host: int = 2, fail_threshold: int = 3, cooldown_seconds: int = 300, bulk_concurrency: int = 20, ): self.timeout = timeout self.helo_domain = helo_domain self.mail_from = mail_from self.catch_all_cache = catch_all_cache or CatchAllCache() self.max_connections_per_host = max_connections_per_host self._factory = connection_factory or ( lambda host, **kw: SmtpConnection(host, **kw) ) # Egress circuit breaker. On a host that blocks outbound port 25 every # connect burns the full timeout, so a 10k job would take ~27h. After # `fail_threshold` consecutive connect failures we stop dialling for # `cooldown_seconds` and report rows unavailable immediately instead. self.fail_threshold = max(1, fail_threshold) self.cooldown_seconds = cooldown_seconds self.bulk_concurrency = max(1, bulk_concurrency) self._consecutive_failures = 0 self._open_until = 0.0 @property def circuit_open(self) -> bool: return time.monotonic() < self._open_until def _record_connect_failure(self) -> None: self._consecutive_failures += 1 if self._consecutive_failures >= self.fail_threshold: self._open_until = time.monotonic() + self.cooldown_seconds def _record_connect_success(self) -> None: self._consecutive_failures = 0 self._open_until = 0.0 def _new_connection(self, mx_host: str): return self._factory( mx_host, timeout=self.timeout, helo_domain=self.helo_domain, mail_from=self.mail_from, ) def verify_batch( self, pairs: Iterable[tuple[str, str]] ) -> dict[str, SMTPResult]: """Verify pairs of (email, mx_host), sequentially. Returns email -> result.""" groups = self._group(pairs) results: dict[str, SMTPResult] = {} for (domain, mx_host), emails in groups.items(): results.update(self._verify_domain(domain, mx_host, emails)) return results async def verify_batch_async( self, pairs: Iterable[tuple[str, str]] ) -> dict[str, SMTPResult]: """Verify many pairs with DOMAINS PROBED CONCURRENTLY. Blocking SMTP work runs in a thread pool bounded by `bulk_concurrency`. Sequential probing makes wall-clock the sum of every domain's latency (timeout x domains on a blocked host); real bulk throughput comes from parallelism across domains, not speed per domain. """ groups = self._group(pairs) if not groups: return {} semaphore = asyncio.Semaphore(self.bulk_concurrency) async def run_domain(domain: str, mx_host: str, emails: list[str]): async with semaphore: return await asyncio.to_thread( self._verify_domain, domain, mx_host, emails ) chunks = await asyncio.gather( *(run_domain(d, m, e) for (d, m), e in groups.items()), return_exceptions=True, ) results: dict[str, SMTPResult] = {} for chunk in chunks: if isinstance(chunk, Exception): continue results.update(chunk) # Any group that raised must still return a row per address. missing = [ (d, m, e) for (d, m), e in groups.items() if not any(addr in results for addr in e) ] for domain, mx_host, emails in missing: for email in emails: results.setdefault( email, SMTPResult( verified=None, error=f"smtp_unreachable: {mx_host} (probe aborted)", ), ) return results @staticmethod def _group( pairs: Iterable[tuple[str, str]] ) -> dict[tuple[str, str], list[str]]: """Group (email, mx_host) pairs by (domain, mx) — the session unit.""" groups: dict[tuple[str, str], list[str]] = {} for email, mx_host in pairs: if not email or not mx_host: continue domain = email.rsplit("@", 1)[-1].lower() groups.setdefault((domain, mx_host), []).append(email) return groups def _verify_domain( self, domain: str, mx_host: str, emails: list[str] ) -> dict[str, SMTPResult]: cached = self.catch_all_cache.get(domain) # Known catch-all: no per-address probe can confirm a mailbox here. if cached is True: return { e: SMTPResult(verified=None, error=f"catch_all (cached): {domain}") for e in emails } out: dict[str, SMTPResult] = {} # Fail fast while egress is known-bad: do not dial, do not wait. if self.circuit_open: return { e: SMTPResult( verified=None, error=f"smtp_unreachable: {mx_host} (egress circuit open)", ) for e in emails } conn = self._new_connection(mx_host) try: if not conn.open(): self._record_connect_failure() return { e: SMTPResult( verified=None, error=f"smtp_unreachable: {mx_host}", ) for e in emails } self._record_connect_success() if cached is None: ca = conn.probe(_gibberish(domain)) if ca.accepted: self.catch_all_cache.put(domain, True) for e in emails: out[e] = SMTPResult( verified=None, error=f"catch_all: {domain} accepts every recipient", ) return out if ca.permanent_reject: self.catch_all_cache.put(domain, False) # else inconclusive -> do not cache, fall through and probe. policy = classify_mx(mx_host) for email in emails: out[email] = self._interpret(conn.probe(email), policy, domain) except Exception as exc: # defensive: never let one group kill the job for email in emails: out.setdefault( email, SMTPResult(verified=None, error=f"probe_error: {exc}") ) finally: try: conn.quit() except Exception: pass return out @staticmethod def _interpret(outcome: RcptOutcome, policy, domain: str) -> SMTPResult: if outcome.transport_error: return SMTPResult( verified=None, error=f"transport_error: {outcome.error or 'unknown'}" ) if outcome.accepted: if policy.accepts_all: # Microsoft 365 / gateways accept everything and bounce later — # a 250 here is not proof of a mailbox. return SMTPResult( verified=None, error=f"accept_all_provider ({policy.provider}): rcpt not conclusive", ) return SMTPResult( verified=True, smtp_code=250, smtp_message=outcome.message[:400] ) if outcome.permanent_reject: return SMTPResult( verified=False, smtp_code=outcome.code, smtp_message=outcome.message[:400], ) # 4xx greylist or unexpected code: inconclusive, retryable. return SMTPResult( verified=None, smtp_code=outcome.code, error=f"inconclusive: {outcome.message[:200] or outcome.code}", )