Spaces:

sitcod3
/

modul-x

Running

Upload folder using huggingface_hub

fabbde1 verified 14 days ago

5.34 kB

"""Outbound port-25 preflight.
Mailbox verification *requires* outbound TCP to the recipient's MX on port 25.
Most PaaS platforms block it — Hugging Face Spaces and AWS (EC2/Lambda) both
block 25 by default — so an engine that looks correct can still be unable to
produce a single real verdict.
This probe answers the only question that matters before trusting a host:
"can this machine reach an MX on port 25?" It is deliberately tiny, side-effect
free (banner + EHLO, then QUIT), and safe to run at boot or from CI.
"""
from __future__ import annotations

import socket
from dataclasses import asdict, dataclass
from typing import Callable, Optional

# A highly-available MX; if this is unreachable, port 25 egress is blocked.
DEFAULT_TARGET = ("gmail-smtp-in.l.google.com", 25)
DEFAULT_HELO = "validator.local"

_BLOCKED_ADVICE = (
    "Outbound port 25 is blocked. Hugging Face Spaces, AWS (EC2/Lambda) and most "
    "PaaS providers block it by default, so direct SMTP verification cannot run "
    "there. Fix: run this service on a VPS/host with port-25 egress (and set a "
    "PTR record + SMTP_HELO_DOMAIN so strict receivers accept the probes). "
    "Until then every mailbox row is honestly reported as smtp_unreachable."
)


@dataclass
class EgressReport:
    target: str
    reachable: bool
    banner: str = ""
    helo_accepted: bool = False
    diagnosis: str = ""
    advice: str = ""

    @property
    def ok(self) -> bool:
        return self.reachable

    def as_dict(self) -> dict:
        data = asdict(self)
        data["ok"] = self.ok
        return data


def _default_connect(host: str, port: int, timeout: float):
    sock = socket.create_connection((host, port), timeout=timeout)
    sock.settimeout(timeout)
    return sock


def check_port25_egress(
    connect: Optional[Callable[[str, int, float], object]] = None,
    target: tuple[str, int] = DEFAULT_TARGET,
    helo_domain: str = DEFAULT_HELO,
    timeout: float = 8.0,
) -> EgressReport:
    """Probe whether outbound port 25 works from this host."""
    host, port = target
    label = f"{host}:{port}"
    sock = None
    try:
        sock = (connect or _default_connect)(host, port, timeout)
    except Exception as exc:
        # A timeout / refusal / unreachable network on port 25 is the signature
        # of egress filtering. Say so explicitly — "timed out" alone sends the
        # operator hunting for a DNS or app bug instead of a firewall.
        kind = type(exc).__name__
        detail = str(exc)
        if isinstance(exc, socket.timeout) or "timed out" in detail.lower():
            diagnosis = f"connect timed out (port 25 is blocked/filtered): {detail}"
        elif isinstance(exc, ConnectionRefusedError) or "refused" in detail.lower():
            diagnosis = f"connection refused on port 25 (blocked): {detail}"
        elif isinstance(exc, OSError) and getattr(exc, "errno", None) == 101:
            diagnosis = f"network unreachable (port 25 egress blocked): {detail}"
        else:
            diagnosis = f"connect failed, port 25 likely blocked: {kind}: {detail}"
        return EgressReport(
            target=label,
            reachable=False,
            diagnosis=diagnosis,
            advice=_BLOCKED_ADVICE,
        )

    banner = ""
    helo_accepted = False
    try:
        try:
            raw = sock.recv(1024)  # type: ignore[attr-defined]
            banner = raw.decode("utf-8", "replace").strip() if raw else ""
        except Exception as exc:
            return EgressReport(
                target=label,
                reachable=False,
                diagnosis=f"no SMTP banner: {type(exc).__name__}: {exc}",
                advice=_BLOCKED_ADVICE,
            )

        try:
            sock.sendall(f"EHLO {helo_domain}\r\n".encode())  # type: ignore[attr-defined]
            reply = sock.recv(1024)  # type: ignore[attr-defined]
            helo_accepted = b"250" in reply
        except Exception:
            helo_accepted = False

        try:
            sock.sendall(b"QUIT\r\n")  # type: ignore[attr-defined]
        except Exception:
            pass
    finally:
        if sock is not None:
            try:
                sock.close()
            except Exception:
                pass

    if not banner.startswith("220"):
        return EgressReport(
            target=label,
            reachable=False,
            banner=banner,
            helo_accepted=helo_accepted,
            diagnosis=f"unexpected banner: {banner[:80]!r}",
            advice=_BLOCKED_ADVICE,
        )

    return EgressReport(
        target=label,
        reachable=True,
        banner=banner,
        helo_accepted=helo_accepted,
        diagnosis="port 25 reachable — real mailbox verification is possible here",
        advice=(
            "Set SMTP_HELO_DOMAIN to this host's real hostname and give that IP a "
            "matching PTR record; strict receivers reject generic HELO names."
        ),
    )


def preflight_or_warn() -> EgressReport:
    """Run the preflight and print a one-line operational summary."""
    report = check_port25_egress()
    status = "OK" if report.ok else "BLOCKED"
    print(f"[EGRESS] port-25 preflight {status} — {report.target} — {report.diagnosis}")
    if not report.ok:
        print(f"[EGRESS] {report.advice}")
    return report

	"""Outbound port-25 preflight.

	Mailbox verification requires outbound TCP to the recipient's MX on port 25.
	Most PaaS platforms block it — Hugging Face Spaces and AWS (EC2/Lambda) both
	block 25 by default — so an engine that looks correct can still be unable to
	produce a single real verdict.

	This probe answers the only question that matters before trusting a host:
	"can this machine reach an MX on port 25?" It is deliberately tiny, side-effect
	free (banner + EHLO, then QUIT), and safe to run at boot or from CI.
	"""
	from __future__ import annotations

	import socket
	from dataclasses import asdict, dataclass
	from typing import Callable, Optional

	# A highly-available MX; if this is unreachable, port 25 egress is blocked.
	DEFAULT_TARGET = ("gmail-smtp-in.l.google.com", 25)
	DEFAULT_HELO = "validator.local"

	_BLOCKED_ADVICE = (
	"Outbound port 25 is blocked. Hugging Face Spaces, AWS (EC2/Lambda) and most "
	"PaaS providers block it by default, so direct SMTP verification cannot run "
	"there. Fix: run this service on a VPS/host with port-25 egress (and set a "
	"PTR record + SMTP_HELO_DOMAIN so strict receivers accept the probes). "
	"Until then every mailbox row is honestly reported as smtp_unreachable."
	)


	@dataclass
	class EgressReport:
	target: str
	reachable: bool
	banner: str = ""
	helo_accepted: bool = False
	diagnosis: str = ""
	advice: str = ""

	@property
	def ok(self) -> bool:
	return self.reachable

	def as_dict(self) -> dict:
	data = asdict(self)
	data["ok"] = self.ok
	return data


	def _default_connect(host: str, port: int, timeout: float):
	sock = socket.create_connection((host, port), timeout=timeout)
	sock.settimeout(timeout)
	return sock


	def check_port25_egress(
	connect: Optional[Callable[[str, int, float], object]] = None,
	target: tuple[str, int] = DEFAULT_TARGET,
	helo_domain: str = DEFAULT_HELO,
	timeout: float = 8.0,
	) -> EgressReport:
	"""Probe whether outbound port 25 works from this host."""
	host, port = target
	label = f"{host}:{port}"
	sock = None
	try:
	sock = (connect or _default_connect)(host, port, timeout)
	except Exception as exc:
	# A timeout / refusal / unreachable network on port 25 is the signature
	# of egress filtering. Say so explicitly — "timed out" alone sends the
	# operator hunting for a DNS or app bug instead of a firewall.
	kind = type(exc).__name__
	detail = str(exc)
	if isinstance(exc, socket.timeout) or "timed out" in detail.lower():
	diagnosis = f"connect timed out (port 25 is blocked/filtered): {detail}"
	elif isinstance(exc, ConnectionRefusedError) or "refused" in detail.lower():
	diagnosis = f"connection refused on port 25 (blocked): {detail}"
	elif isinstance(exc, OSError) and getattr(exc, "errno", None) == 101:
	diagnosis = f"network unreachable (port 25 egress blocked): {detail}"
	else:
	diagnosis = f"connect failed, port 25 likely blocked: {kind}: {detail}"
	return EgressReport(
	target=label,
	reachable=False,
	diagnosis=diagnosis,
	advice=_BLOCKED_ADVICE,
	)

	banner = ""
	helo_accepted = False
	try:
	try:
	raw = sock.recv(1024) # type: ignore[attr-defined]
	banner = raw.decode("utf-8", "replace").strip() if raw else ""
	except Exception as exc:
	return EgressReport(
	target=label,
	reachable=False,
	diagnosis=f"no SMTP banner: {type(exc).__name__}: {exc}",
	advice=_BLOCKED_ADVICE,
	)

	try:
	sock.sendall(f"EHLO {helo_domain}\r\n".encode()) # type: ignore[attr-defined]
	reply = sock.recv(1024) # type: ignore[attr-defined]
	helo_accepted = b"250" in reply
	except Exception:
	helo_accepted = False

	try:
	sock.sendall(b"QUIT\r\n") # type: ignore[attr-defined]
	except Exception:
	pass
	finally:
	if sock is not None:
	try:
	sock.close()
	except Exception:
	pass

	if not banner.startswith("220"):
	return EgressReport(
	target=label,
	reachable=False,
	banner=banner,
	helo_accepted=helo_accepted,
	diagnosis=f"unexpected banner: {banner[:80]!r}",
	advice=_BLOCKED_ADVICE,
	)

	return EgressReport(
	target=label,
	reachable=True,
	banner=banner,
	helo_accepted=helo_accepted,
	diagnosis="port 25 reachable — real mailbox verification is possible here",
	advice=(
	"Set SMTP_HELO_DOMAIN to this host's real hostname and give that IP a "
	"matching PTR record; strict receivers reject generic HELO names."
	),
	)


	def preflight_or_warn() -> EgressReport:
	"""Run the preflight and print a one-line operational summary."""
	report = check_port25_egress()
	status = "OK" if report.ok else "BLOCKED"
	print(f"[EGRESS] port-25 preflight {status} — {report.target} — {report.diagnosis}")
	if not report.ok:
	print(f"[EGRESS] {report.advice}")
	return report