Spaces:

sitcod3
/

modul-x

Running

modul-x / app /validator /mailbox_provider.py

FortyDN Dev

v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker

de745bf 2 days ago

Raw History Blame Contribute Delete

13 kB

"""Mailbox verification providers (Layer 5 catch-all + Layer 6 SMTP).
The engine is FULLY AUTONOMOUS (v3.0.0): mailbox verdicts come from its own
SMTP probing engine — no third-party verification API, no vendor key, no
per-check fee. The former MillionVerifier delegation was removed by design
decision: a stray key must never be able to re-route verification data (and
the full recipient list) through a vendor.
resolve_provider() picks the mailbox layer from settings:
    mailbox_provider = "auto"   → remote SMTP worker when SMTP_WORKER_URL is
                                  configured, else the built-in engine
    mailbox_provider = "local"  → built-in SMTP engine (this host)
    mailbox_provider = "remote" → the remote SMTP probe worker (spec §3.3)
RemoteSmtpWorkerProvider implements the four-worker architecture: the
orchestrator (this app) delegates the port-25 dialogue to a separate
deployment of the same image running on a host whose egress allows it
(see POST /api/v1/smtp/probe in routes.py). A shared secret
(SMTP_WORKER_TOKEN) guards the worker route.
Both providers embed a circuit breaker so a host with blocked port-25 egress
(or a dead worker) fails fast instead of burning the full SMTP timeout on
every address in a bulk job.
"""
from __future__ import annotations

import asyncio
import random
import time
from typing import Optional, Tuple

import httpx

from ..config import settings
from ..models import SMTPResult
from .smtp_verify import GIBBERISH_LOCAL, smtp_verifier


class MailboxProvider:
    """Common interface every mailbox provider implements."""

    name = "base"

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        raise NotImplementedError

    async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
        raise NotImplementedError


class AutonomousSmtpProvider(MailboxProvider):
    """The self-contained engine: own SMTP probing, no third-party service.
    Wraps the MX-grouped batch verifier (reused sessions, per-domain catch-all
    caching, greylist awareness). This is the default provider — the validator
    must work with no vendor account, key, or per-check fee.
    """

    name = "autonomous"

    def __init__(self, batch=None):
        if batch is None:
            from .mx_batch import CatchAllCache, MxBatchVerifier

            batch = MxBatchVerifier(
                timeout=settings.smtp_timeout,
                helo_domain=settings.smtp_helo_domain,
                mail_from=settings.smtp_mail_from,
                catch_all_cache=CatchAllCache(
                    ttl_seconds=settings.catch_all_ttl_seconds
                ),
            )
        self.batch = batch

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        results = self.batch.verify_batch([(email, mx_host)])
        return results.get(email) or SMTPResult(
            verified=None, error="engine returned no result"
        )

    async def catch_all(
        self, domain: str, mx_host: str
    ) -> Tuple[Optional[bool], float]:
        """Probe a gibberish local part to learn whether the domain accepts all."""
        gibberish = f"{GIBBERISH_LOCAL}@{domain}"
        results = self.batch.verify_batch([(gibberish, mx_host)])
        probe = results.get(gibberish)
        if probe is None:
            return None, 0.0
        if probe.verified is True:
            # The domain accepted a random address -> catch-all.
            return True, 0.9
        if probe.verified is False:
            return False, 0.9
        # Inconclusive: distinguish a detected catch-all from a dead probe.
        err = (probe.error or "").lower()
        if "catch_all" in err:
            return True, 0.95
        return None, 0.0

    async def verify_many(self, pairs):
        """Batched mailbox stage: one call for the whole list.
        Returns (email -> SMTPResult, domain -> catch-all flag). This is what
        lets the engine do one TCP session per domain instead of one per address.
        """
        pairs = [(e, mx) for e, mx in pairs if e and mx]
        if not pairs:
            return {}, {}
        # Prefer the concurrent path when the batch engine supports it.
        if hasattr(self.batch, "verify_batch_async"):
            results = await self.batch.verify_batch_async(pairs)
        else:
            results = self.batch.verify_batch(pairs)
        domains = {e.rsplit("@", 1)[-1].lower(): mx for e, mx in pairs}
        catch_all = {
            domain: self.batch.catch_all_cache.get(domain) for domain in domains
        }
        return results, catch_all


class RemoteSmtpWorkerProvider(MailboxProvider):
    """Delegates the port-25 dialogue to a remote SMTP probe worker (§3.3).
    The worker is any deployment of this same image whose SMTP_WORKER_URL this
    orchestrator can reach. Egress requirement moves to the worker host; the
    orchestrator stays PaaS-deployable.
    """

    name = "remote"

    def __init__(
        self,
        url: Optional[str] = None,
        token: Optional[str] = None,
        timeout: Optional[int] = None,
        fail_threshold: Optional[int] = None,
        cooldown_seconds: Optional[int] = None,
    ):
        self.url = (url if url is not None else settings.smtp_worker_url).rstrip("/")
        self.token = token if token is not None else settings.smtp_worker_token
        self.timeout = timeout if timeout is not None else settings.smtp_worker_timeout
        self.fail_threshold = (
            fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
        )
        self.cooldown = (
            cooldown_seconds
            if cooldown_seconds is not None
            else settings.smtp_cooldown_seconds
        )
        self._consecutive_failures = 0
        self._open_until = 0.0

    # ---- circuit breaker (same contract as LocalSmtpProvider) --------------
    @property
    def circuit_open(self) -> bool:
        return time.monotonic() < self._open_until

    def record_failure(self) -> None:
        self._consecutive_failures += 1
        if self._consecutive_failures >= self.fail_threshold:
            self._open_until = time.monotonic() + self.cooldown

    def record_success(self) -> None:
        self._consecutive_failures = 0
        self._open_until = 0.0

    def _headers(self) -> dict:
        h = {"Content-Type": "application/json"}
        if self.token:
            h["X-Worker-Token"] = self.token
        return h

    async def _post(self, path: str, payload: dict):
        async with httpx.AsyncClient(timeout=self.timeout + 5) as client:
            resp = await client.post(
                f"{self.url}/smtp{path}", json=payload, headers=self._headers()
            )
            resp.raise_for_status()
            return resp.json()

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        if self.circuit_open:
            return SMTPResult(
                verified=None, error="SMTP worker unavailable (circuit open)"
            )
        try:
            data = await self._post(
                "/probe", {"email": email, "mx_host": mx_host}
            )
            result = SMTPResult(**data)
        except Exception as exc:
            self.record_failure()
            return SMTPResult(
                verified=None, error=f"smtp_worker_unreachable: {exc}"
            )
        self.record_success()
        return result

    async def catch_all(
        self, domain: str, mx_host: str
    ) -> Tuple[Optional[bool], float]:
        if self.circuit_open:
            return None, 0.0
        try:
            data = await self._post(
                "/probe", {"domain": domain, "mx_host": mx_host, "probe": "catch_all"}
            )
        except Exception:
            self.record_failure()
            return None, 0.0
        self.record_success()
        value = data.get("is_catch_all")
        confidence = float(data.get("confidence", 0.0) or 0.0)
        if value is None or confidence == 0.0:
            return None, 0.0
        return bool(value), confidence

    async def verify_many(self, pairs):
        pairs = [(e, mx) for e, mx in pairs if e and mx]
        if not pairs:
            return {}, {}
        if self.circuit_open:
            return (
                {
                    e: SMTPResult(
                        verified=None, error="SMTP worker unavailable (circuit open)"
                    )
                    for e, _ in pairs
                },
                {},
            )
        try:
            data = await self._post(
                "/probe/batch", {"pairs": [[e, mx] for e, mx in pairs]}
            )
            results = {
                e: SMTPResult(**row) for e, row in (data.get("results") or {}).items()
            }
            catch_all = {d: v for d, v in (data.get("catch_all") or {}).items()}
            self.record_success()
            return results, catch_all
        except Exception as exc:
            self.record_failure()
            return (
                {
                    e: SMTPResult(
                        verified=None, error=f"smtp_worker_unreachable: {exc}"
                    )
                    for e, _ in pairs
                },
                {},
            )


class LocalSmtpProvider(MailboxProvider):
    """Direct port-25 SMTP (one dialogue per call) with a circuit breaker.
    Kept as the building block for single-probe paths and tests; bulk traffic
    goes through AutonomousSmtpProvider's session-reusing batch engine.
    """

    name = "local"

    def __init__(
        self,
        fail_threshold: Optional[int] = None,
        cooldown_seconds: Optional[int] = None,
    ):
        self.fail_threshold = (
            fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
        )
        self.cooldown = (
            cooldown_seconds if cooldown_seconds is not None else settings.smtp_cooldown_seconds
        )
        self._consecutive_failures = 0
        self._open_until = 0.0

    @property
    def circuit_open(self) -> bool:
        return time.monotonic() < self._open_until

    def circuit_open_after_cooldown(self) -> bool:
        """True only while still within the cooldown window (0s ⇒ closed)."""
        return time.monotonic() < self._open_until

    def record_failure(self) -> None:
        self._consecutive_failures += 1
        if self._consecutive_failures >= self.fail_threshold:
            self._open_until = time.monotonic() + self.cooldown

    def record_success(self) -> None:
        self._consecutive_failures = 0
        self._open_until = 0.0

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        if self.circuit_open:
            return SMTPResult(
                verified=None, error="SMTP unavailable (egress circuit open)"
            )
        result = await smtp_verifier.verify(email, mx_host, settings.use_proxies)
        # A connection-class failure (timeout/refused/unreachable) counts toward
        # opening the circuit; a real SMTP dialogue does not.
        if result.verified is None and result.error and any(
            k in result.error.lower()
            for k in ("timeout", "timed out", "refused", "unreachable", "unavailable")
        ):
            self.record_failure()
        elif result.smtp_code is not None or result.verified is not None:
            self.record_success()
        return result

    async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
        if self.circuit_open:
            return None, 0.0
        try:
            is_catch_all, confidence = await smtp_verifier.check_catch_all(domain, mx_host)
        except Exception:
            return None, 0.0
        # check_catch_all returns (False, 0.0) when the probe is inconclusive —
        # that means UNKNOWN, not "not catch-all". Confidence 0.0 is the signal.
        if confidence == 0.0:
            self.record_failure()
            return None, 0.0
        self.record_success()
        return is_catch_all, confidence


def resolve_provider() -> MailboxProvider:
    """Pick the mailbox layer.
    Fully autonomous by default: the built-in SMTP engine answers with no
    external dependency. A configured SMTP_WORKER_URL shifts the port-25
    dialogue to the dedicated worker host — nothing else changes.
    """
    mode = (settings.mailbox_provider or "auto").strip().lower()
    if mode == "remote":
        if settings.smtp_worker_url:
            return RemoteSmtpWorkerProvider()
        # Misconfiguration must not crash the API: degrade to the built-in
        # engine, which reports honest transport failures.
        return AutonomousSmtpProvider()
    if mode == "local":
        return AutonomousSmtpProvider()
    # auto → remote worker when configured, else built-in.
    if settings.smtp_worker_url:
        return RemoteSmtpWorkerProvider()
    return AutonomousSmtpProvider()

	"""Mailbox verification providers (Layer 5 catch-all + Layer 6 SMTP).

	The engine is FULLY AUTONOMOUS (v3.0.0): mailbox verdicts come from its own
	SMTP probing engine — no third-party verification API, no vendor key, no
	per-check fee. The former MillionVerifier delegation was removed by design
	decision: a stray key must never be able to re-route verification data (and
	the full recipient list) through a vendor.

	resolve_provider() picks the mailbox layer from settings:

	mailbox_provider = "auto" → remote SMTP worker when SMTP_WORKER_URL is
	configured, else the built-in engine
	mailbox_provider = "local" → built-in SMTP engine (this host)
	mailbox_provider = "remote" → the remote SMTP probe worker (spec §3.3)

	RemoteSmtpWorkerProvider implements the four-worker architecture: the
	orchestrator (this app) delegates the port-25 dialogue to a separate
	deployment of the same image running on a host whose egress allows it
	(see POST /api/v1/smtp/probe in routes.py). A shared secret
	(SMTP_WORKER_TOKEN) guards the worker route.

	Both providers embed a circuit breaker so a host with blocked port-25 egress
	(or a dead worker) fails fast instead of burning the full SMTP timeout on
	every address in a bulk job.
	"""
	from __future__ import annotations

	import asyncio
	import random
	import time
	from typing import Optional, Tuple

	import httpx

	from ..config import settings
	from ..models import SMTPResult
	from .smtp_verify import GIBBERISH_LOCAL, smtp_verifier


	class MailboxProvider:
	"""Common interface every mailbox provider implements."""

	name = "base"

	async def verify(self, email: str, mx_host: str) -> SMTPResult:
	raise NotImplementedError

	async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
	raise NotImplementedError


	class AutonomousSmtpProvider(MailboxProvider):
	"""The self-contained engine: own SMTP probing, no third-party service.

	Wraps the MX-grouped batch verifier (reused sessions, per-domain catch-all
	caching, greylist awareness). This is the default provider — the validator
	must work with no vendor account, key, or per-check fee.
	"""

	name = "autonomous"

	def __init__(self, batch=None):
	if batch is None:
	from .mx_batch import CatchAllCache, MxBatchVerifier

	batch = MxBatchVerifier(
	timeout=settings.smtp_timeout,
	helo_domain=settings.smtp_helo_domain,
	mail_from=settings.smtp_mail_from,
	catch_all_cache=CatchAllCache(
	ttl_seconds=settings.catch_all_ttl_seconds
	),
	)
	self.batch = batch

	async def verify(self, email: str, mx_host: str) -> SMTPResult:
	results = self.batch.verify_batch([(email, mx_host)])
	return results.get(email) or SMTPResult(
	verified=None, error="engine returned no result"
	)

	async def catch_all(
	self, domain: str, mx_host: str
	) -> Tuple[Optional[bool], float]:
	"""Probe a gibberish local part to learn whether the domain accepts all."""
	gibberish = f"{GIBBERISH_LOCAL}@{domain}"
	results = self.batch.verify_batch([(gibberish, mx_host)])
	probe = results.get(gibberish)
	if probe is None:
	return None, 0.0
	if probe.verified is True:
	# The domain accepted a random address -> catch-all.
	return True, 0.9
	if probe.verified is False:
	return False, 0.9
	# Inconclusive: distinguish a detected catch-all from a dead probe.
	err = (probe.error or "").lower()
	if "catch_all" in err:
	return True, 0.95
	return None, 0.0

	async def verify_many(self, pairs):
	"""Batched mailbox stage: one call for the whole list.

	Returns (email -> SMTPResult, domain -> catch-all flag). This is what
	lets the engine do one TCP session per domain instead of one per address.
	"""
	pairs = [(e, mx) for e, mx in pairs if e and mx]
	if not pairs:
	return {}, {}
	# Prefer the concurrent path when the batch engine supports it.
	if hasattr(self.batch, "verify_batch_async"):
	results = await self.batch.verify_batch_async(pairs)
	else:
	results = self.batch.verify_batch(pairs)
	domains = {e.rsplit("@", 1)[-1].lower(): mx for e, mx in pairs}
	catch_all = {
	domain: self.batch.catch_all_cache.get(domain) for domain in domains
	}
	return results, catch_all


	class RemoteSmtpWorkerProvider(MailboxProvider):
	"""Delegates the port-25 dialogue to a remote SMTP probe worker (§3.3).

	The worker is any deployment of this same image whose SMTP_WORKER_URL this
	orchestrator can reach. Egress requirement moves to the worker host; the
	orchestrator stays PaaS-deployable.
	"""

	name = "remote"

	def __init__(
	self,
	url: Optional[str] = None,
	token: Optional[str] = None,
	timeout: Optional[int] = None,
	fail_threshold: Optional[int] = None,
	cooldown_seconds: Optional[int] = None,
	):
	self.url = (url if url is not None else settings.smtp_worker_url).rstrip("/")
	self.token = token if token is not None else settings.smtp_worker_token
	self.timeout = timeout if timeout is not None else settings.smtp_worker_timeout
	self.fail_threshold = (
	fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
	)
	self.cooldown = (
	cooldown_seconds
	if cooldown_seconds is not None
	else settings.smtp_cooldown_seconds
	)
	self._consecutive_failures = 0
	self._open_until = 0.0

	# ---- circuit breaker (same contract as LocalSmtpProvider) --------------
	@property
	def circuit_open(self) -> bool:
	return time.monotonic() < self._open_until

	def record_failure(self) -> None:
	self._consecutive_failures += 1
	if self._consecutive_failures >= self.fail_threshold:
	self._open_until = time.monotonic() + self.cooldown

	def record_success(self) -> None:
	self._consecutive_failures = 0
	self._open_until = 0.0

	def _headers(self) -> dict:
	h = {"Content-Type": "application/json"}
	if self.token:
	h["X-Worker-Token"] = self.token
	return h

	async def _post(self, path: str, payload: dict):
	async with httpx.AsyncClient(timeout=self.timeout + 5) as client:
	resp = await client.post(
	f"{self.url}/smtp{path}", json=payload, headers=self._headers()
	)
	resp.raise_for_status()
	return resp.json()

	async def verify(self, email: str, mx_host: str) -> SMTPResult:
	if self.circuit_open:
	return SMTPResult(
	verified=None, error="SMTP worker unavailable (circuit open)"
	)
	try:
	data = await self._post(
	"/probe", {"email": email, "mx_host": mx_host}
	)
	result = SMTPResult(**data)
	except Exception as exc:
	self.record_failure()
	return SMTPResult(
	verified=None, error=f"smtp_worker_unreachable: {exc}"
	)
	self.record_success()
	return result

	async def catch_all(
	self, domain: str, mx_host: str
	) -> Tuple[Optional[bool], float]:
	if self.circuit_open:
	return None, 0.0
	try:
	data = await self._post(
	"/probe", {"domain": domain, "mx_host": mx_host, "probe": "catch_all"}
	)
	except Exception:
	self.record_failure()
	return None, 0.0
	self.record_success()
	value = data.get("is_catch_all")
	confidence = float(data.get("confidence", 0.0) or 0.0)
	if value is None or confidence == 0.0:
	return None, 0.0
	return bool(value), confidence

	async def verify_many(self, pairs):
	pairs = [(e, mx) for e, mx in pairs if e and mx]
	if not pairs:
	return {}, {}
	if self.circuit_open:
	return (
	{
	e: SMTPResult(
	verified=None, error="SMTP worker unavailable (circuit open)"
	)
	for e, _ in pairs
	},
	{},
	)
	try:
	data = await self._post(
	"/probe/batch", {"pairs": [[e, mx] for e, mx in pairs]}
	)
	results = {
	e: SMTPResult(**row) for e, row in (data.get("results") or {}).items()
	}
	catch_all = {d: v for d, v in (data.get("catch_all") or {}).items()}
	self.record_success()
	return results, catch_all
	except Exception as exc:
	self.record_failure()
	return (
	{
	e: SMTPResult(
	verified=None, error=f"smtp_worker_unreachable: {exc}"
	)
	for e, _ in pairs
	},
	{},
	)


	class LocalSmtpProvider(MailboxProvider):
	"""Direct port-25 SMTP (one dialogue per call) with a circuit breaker.

	Kept as the building block for single-probe paths and tests; bulk traffic
	goes through AutonomousSmtpProvider's session-reusing batch engine.
	"""

	name = "local"

	def __init__(
	self,
	fail_threshold: Optional[int] = None,
	cooldown_seconds: Optional[int] = None,
	):
	self.fail_threshold = (
	fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
	)
	self.cooldown = (
	cooldown_seconds if cooldown_seconds is not None else settings.smtp_cooldown_seconds
	)
	self._consecutive_failures = 0
	self._open_until = 0.0

	@property
	def circuit_open(self) -> bool:
	return time.monotonic() < self._open_until

	def circuit_open_after_cooldown(self) -> bool:
	"""True only while still within the cooldown window (0s ⇒ closed)."""
	return time.monotonic() < self._open_until

	def record_failure(self) -> None:
	self._consecutive_failures += 1
	if self._consecutive_failures >= self.fail_threshold:
	self._open_until = time.monotonic() + self.cooldown

	def record_success(self) -> None:
	self._consecutive_failures = 0
	self._open_until = 0.0

	async def verify(self, email: str, mx_host: str) -> SMTPResult:
	if self.circuit_open:
	return SMTPResult(
	verified=None, error="SMTP unavailable (egress circuit open)"
	)
	result = await smtp_verifier.verify(email, mx_host, settings.use_proxies)
	# A connection-class failure (timeout/refused/unreachable) counts toward
	# opening the circuit; a real SMTP dialogue does not.
	if result.verified is None and result.error and any(
	k in result.error.lower()
	for k in ("timeout", "timed out", "refused", "unreachable", "unavailable")
	):
	self.record_failure()
	elif result.smtp_code is not None or result.verified is not None:
	self.record_success()
	return result

	async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
	if self.circuit_open:
	return None, 0.0
	try:
	is_catch_all, confidence = await smtp_verifier.check_catch_all(domain, mx_host)
	except Exception:
	return None, 0.0
	# check_catch_all returns (False, 0.0) when the probe is inconclusive —
	# that means UNKNOWN, not "not catch-all". Confidence 0.0 is the signal.
	if confidence == 0.0:
	self.record_failure()
	return None, 0.0
	self.record_success()
	return is_catch_all, confidence


	def resolve_provider() -> MailboxProvider:
	"""Pick the mailbox layer.

	Fully autonomous by default: the built-in SMTP engine answers with no
	external dependency. A configured SMTP_WORKER_URL shifts the port-25
	dialogue to the dedicated worker host — nothing else changes.
	"""
	mode = (settings.mailbox_provider or "auto").strip().lower()
	if mode == "remote":
	if settings.smtp_worker_url:
	return RemoteSmtpWorkerProvider()
	# Misconfiguration must not crash the API: degrade to the built-in
	# engine, which reports honest transport failures.
	return AutonomousSmtpProvider()
	if mode == "local":
	return AutonomousSmtpProvider()
	# auto → remote worker when configured, else built-in.
	if settings.smtp_worker_url:
	return RemoteSmtpWorkerProvider()
	return AutonomousSmtpProvider()