Spaces:

sitcod3
/

modul-x

Running

Upload folder using huggingface_hub

03c7b2c verified 11 days ago

10.7 kB

"""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}",
        )

	"""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}",
	)