Spaces:

sitcod3
/

modul-x

Running

Upload folder using huggingface_hub

0eff3fe verified 9 days ago

6.59 kB

"""Receiving-provider intelligence — the knowledge layer of the autonomous engine.
A raw SMTP probe returns different *meanings* depending on who is answering:
  - Google Workspace / consumer Gmail reject unknown mailboxes at RCPT with
    550 5.1.1, so a 250 there is real evidence the mailbox exists.
  - Microsoft 365 (and Mimecast/Proofpoint gateways in front of it) accept
    *every* recipient at SMTP time and emit the bounce asynchronously hours
    later. A 250 from those hosts proves nothing, so it must never be read as
    "mailbox verified".
  - Many smaller hosts are catch-all by accident (cPanel defaults).
Recognising the provider from its MX host lets the engine apply the correct
policy instead of treating every 250 as success. This is the standard technique
used by commercial validators, implemented here with no third-party API.
"""
from __future__ import annotations

import re
from dataclasses import dataclass

# MX host suffix -> (provider, accepts_all, trusts_rcpt)
#   accepts_all: accepts every recipient at RCPT (250 is not proof)
#   trusts_rcpt: a 250/550 at RCPT is meaningful evidence
_MX_PATTERNS: tuple[tuple[str, str, bool, bool], ...] = (
    ("google.com", "google", False, True),
    ("googlemail.com", "google", False, True),
    ("outlook.com", "microsoft365", True, False),
    ("protection.outlook.com", "microsoft365", True, False),
    ("office365.com", "microsoft365", True, False),
    ("mimecast.com", "mimecast", True, False),
    ("pphosted.com", "proofpoint", True, False),
    ("ppe-hosted.com", "proofpoint", True, False),
    ("proofpoint.com", "proofpoint", True, False),
    ("barracudanetworks.com", "barracuda", True, False),
    ("mailchannels.net", "mailchannels", True, False),
    ("emailsecurity.app", "emailsecurity", True, False),
    ("yahoodns.net", "yahoo", False, True),
    ("yahoo.com", "yahoo", False, True),
    ("aol.com", "yahoo", False, True),
    ("icloud.com", "apple", False, True),
    ("me.com", "apple", False, True),
    ("zoho.com", "zoho", False, True),
    ("zoho.eu", "zoho", False, True),
    ("protonmail.ch", "proton", False, True),
    ("proton.me", "proton", False, True),
    ("secureserver.net", "godaddy", False, False),
    ("registrar-servers.com", "namecheap", False, False),
    ("ionos.com", "ionos", False, False),
    ("1and1.com", "ionos", False, False),
    ("serverdata.net", "serverdata", False, False),
    ("hostgator.com", "hostgator", False, False),
    ("bluehost.com", "bluehost", False, False),
    ("siteground.com", "siteground", False, False),
    ("dreamhost.com", "dreamhost", False, False),
    ("networksolutions.com", "netsol", False, False),
    ("rackspace.com", "rackspace", False, False),
    ("emailsrvr.com", "rackspace", False, False),
    ("workmail.com", "amazon", False, False),
    ("fastmail.com", "fastmail", False, True),
    ("messagingengine.com", "fastmail", False, True),
    ("yandex.net", "yandex", False, True),
    ("qq.com", "tencent", False, True),
    ("163.com", "netease", False, True),
    ("amazonaws.com", "amazon", False, False),
    ("amazonses.com", "amazon", False, False),
)

_RCPT_CODE_RE = re.compile(r"^\s*(\d{3})")


@dataclass(frozen=True)
class MxPolicy:
    """How to interpret an SMTP response from a given receiving host."""

    provider: str
    trusts_rcpt: bool
    accepts_all: bool


def classify_mx(mx_host: str) -> MxPolicy:
    """Identify the receiving provider from its MX hostname.
    Unknown hosts get a neutral policy: not assumed malicious, but a lone 250
    is not accepted as proof — the catch-all probe must corroborate it.
    """
    host = (mx_host or "").strip().lower().rstrip(".")
    for suffix, provider, accepts_all, trusts_rcpt in _MX_PATTERNS:
        if host == suffix or host.endswith("." + suffix):
            return MxPolicy(
                provider=provider, trusts_rcpt=trusts_rcpt, accepts_all=accepts_all
            )
    return MxPolicy(provider="unknown", trusts_rcpt=False, accepts_all=False)


# Platforms where a CUSTOM domain requires a paid/business account. A
# disposable domain essentially never runs on these — they cost money and
# verify identity — while disposable domains live on cheap self-hosted or
# free-tier MX. When the MX is one of these, the domain-string classifier's
# opinion is unnecessary and, given an imbalanced training set, harmful: it
# flagged a legitimate small business (Google Workspace MX) as disposable.
_TRUSTED_MX_PROVIDERS = frozenset({
    "google",        # Google Workspace (paid for custom domains)
    "microsoft365",  # Microsoft 365 / Exchange Online
    "mimecast",      # enterprise email security gateway
    "proofpoint",    # enterprise + Essentials gateway (pphosted/ppe-hosted)
    "barracuda",     # enterprise email security gateway
    "mailchannels",  # mail infrastructure for hosting providers (paid)
    "emailsecurity", # hosted email security gateway
    "zoho",          # Zoho Mail (paid for custom domains)
    "proton",        # Proton Business
    "fastmail",      # Fastmail (paid custom domain)
    "apple",         # iCloud+ Custom Email Domain
    # Business hosting / mailbox providers: a paid, domain-verified account is
    # required, so their customers are real businesses (observed in the user's
    # 800-row export: ionos, namecheap, serverdata).
    "godaddy", "namecheap", "ionos", "serverdata", "hostgator",
    "bluehost", "siteground", "dreamhost", "netsol", "rackspace", "amazon",
})


def mx_is_trusted(mx_host: str) -> bool:
    """True when the MX runs on a paid enterprise/business mail platform."""
    return classify_mx(mx_host).provider in _TRUSTED_MX_PROVIDERS


def parse_rcpt_code(response: str) -> int | None:
    """Extract the numeric SMTP status from a response line, or None."""
    match = _RCPT_CODE_RE.match(response or "")
    return int(match.group(1)) if match else None


def is_greylist_code(code: int | None) -> bool:
    """4xx — temporary (greylisting / rate limit / mailbox busy)."""
    return code is not None and 400 <= code < 500


def is_permanent_reject(code: int | None) -> bool:
    """5xx — permanent failure (mailbox does not exist, domain refuses)."""
    return code is not None and 500 <= code < 600


def should_retry(code: int | None, *, transport_error: bool = False) -> bool:
    """Whether the same address is worth a second attempt later.
    Only temporary conditions qualify: 4xx greylisting and transport failures.
    A 5xx or a 250 is final.
    """
    if transport_error:
        return True
    return is_greylist_code(code)

	"""Receiving-provider intelligence — the knowledge layer of the autonomous engine.

	A raw SMTP probe returns different meanings depending on who is answering:

	- Google Workspace / consumer Gmail reject unknown mailboxes at RCPT with
	550 5.1.1, so a 250 there is real evidence the mailbox exists.
	- Microsoft 365 (and Mimecast/Proofpoint gateways in front of it) accept
	every recipient at SMTP time and emit the bounce asynchronously hours
	later. A 250 from those hosts proves nothing, so it must never be read as
	"mailbox verified".
	- Many smaller hosts are catch-all by accident (cPanel defaults).

	Recognising the provider from its MX host lets the engine apply the correct
	policy instead of treating every 250 as success. This is the standard technique
	used by commercial validators, implemented here with no third-party API.
	"""
	from __future__ import annotations

	import re
	from dataclasses import dataclass

	# MX host suffix -> (provider, accepts_all, trusts_rcpt)
	# accepts_all: accepts every recipient at RCPT (250 is not proof)
	# trusts_rcpt: a 250/550 at RCPT is meaningful evidence
	_MX_PATTERNS: tuple[tuple[str, str, bool, bool], ...] = (
	("google.com", "google", False, True),
	("googlemail.com", "google", False, True),
	("outlook.com", "microsoft365", True, False),
	("protection.outlook.com", "microsoft365", True, False),
	("office365.com", "microsoft365", True, False),
	("mimecast.com", "mimecast", True, False),
	("pphosted.com", "proofpoint", True, False),
	("ppe-hosted.com", "proofpoint", True, False),
	("proofpoint.com", "proofpoint", True, False),
	("barracudanetworks.com", "barracuda", True, False),
	("mailchannels.net", "mailchannels", True, False),
	("emailsecurity.app", "emailsecurity", True, False),
	("yahoodns.net", "yahoo", False, True),
	("yahoo.com", "yahoo", False, True),
	("aol.com", "yahoo", False, True),
	("icloud.com", "apple", False, True),
	("me.com", "apple", False, True),
	("zoho.com", "zoho", False, True),
	("zoho.eu", "zoho", False, True),
	("protonmail.ch", "proton", False, True),
	("proton.me", "proton", False, True),
	("secureserver.net", "godaddy", False, False),
	("registrar-servers.com", "namecheap", False, False),
	("ionos.com", "ionos", False, False),
	("1and1.com", "ionos", False, False),
	("serverdata.net", "serverdata", False, False),
	("hostgator.com", "hostgator", False, False),
	("bluehost.com", "bluehost", False, False),
	("siteground.com", "siteground", False, False),
	("dreamhost.com", "dreamhost", False, False),
	("networksolutions.com", "netsol", False, False),
	("rackspace.com", "rackspace", False, False),
	("emailsrvr.com", "rackspace", False, False),
	("workmail.com", "amazon", False, False),
	("fastmail.com", "fastmail", False, True),
	("messagingengine.com", "fastmail", False, True),
	("yandex.net", "yandex", False, True),
	("qq.com", "tencent", False, True),
	("163.com", "netease", False, True),
	("amazonaws.com", "amazon", False, False),
	("amazonses.com", "amazon", False, False),
	)

	_RCPT_CODE_RE = re.compile(r"^\s*(\d{3})")


	@dataclass(frozen=True)
	class MxPolicy:
	"""How to interpret an SMTP response from a given receiving host."""

	provider: str
	trusts_rcpt: bool
	accepts_all: bool


	def classify_mx(mx_host: str) -> MxPolicy:
	"""Identify the receiving provider from its MX hostname.

	Unknown hosts get a neutral policy: not assumed malicious, but a lone 250
	is not accepted as proof — the catch-all probe must corroborate it.
	"""
	host = (mx_host or "").strip().lower().rstrip(".")
	for suffix, provider, accepts_all, trusts_rcpt in _MX_PATTERNS:
	if host == suffix or host.endswith("." + suffix):
	return MxPolicy(
	provider=provider, trusts_rcpt=trusts_rcpt, accepts_all=accepts_all
	)
	return MxPolicy(provider="unknown", trusts_rcpt=False, accepts_all=False)


	# Platforms where a CUSTOM domain requires a paid/business account. A
	# disposable domain essentially never runs on these — they cost money and
	# verify identity — while disposable domains live on cheap self-hosted or
	# free-tier MX. When the MX is one of these, the domain-string classifier's
	# opinion is unnecessary and, given an imbalanced training set, harmful: it
	# flagged a legitimate small business (Google Workspace MX) as disposable.
	_TRUSTED_MX_PROVIDERS = frozenset({
	"google", # Google Workspace (paid for custom domains)
	"microsoft365", # Microsoft 365 / Exchange Online
	"mimecast", # enterprise email security gateway
	"proofpoint", # enterprise + Essentials gateway (pphosted/ppe-hosted)
	"barracuda", # enterprise email security gateway
	"mailchannels", # mail infrastructure for hosting providers (paid)
	"emailsecurity", # hosted email security gateway
	"zoho", # Zoho Mail (paid for custom domains)
	"proton", # Proton Business
	"fastmail", # Fastmail (paid custom domain)
	"apple", # iCloud+ Custom Email Domain
	# Business hosting / mailbox providers: a paid, domain-verified account is
	# required, so their customers are real businesses (observed in the user's
	# 800-row export: ionos, namecheap, serverdata).
	"godaddy", "namecheap", "ionos", "serverdata", "hostgator",
	"bluehost", "siteground", "dreamhost", "netsol", "rackspace", "amazon",
	})


	def mx_is_trusted(mx_host: str) -> bool:
	"""True when the MX runs on a paid enterprise/business mail platform."""
	return classify_mx(mx_host).provider in _TRUSTED_MX_PROVIDERS


	def parse_rcpt_code(response: str) -> int \| None:
	"""Extract the numeric SMTP status from a response line, or None."""
	match = _RCPT_CODE_RE.match(response or "")
	return int(match.group(1)) if match else None


	def is_greylist_code(code: int \| None) -> bool:
	"""4xx — temporary (greylisting / rate limit / mailbox busy)."""
	return code is not None and 400 <= code < 500


	def is_permanent_reject(code: int \| None) -> bool:
	"""5xx — permanent failure (mailbox does not exist, domain refuses)."""
	return code is not None and 500 <= code < 600


	def should_retry(code: int \| None, *, transport_error: bool = False) -> bool:
	"""Whether the same address is worth a second attempt later.

	Only temporary conditions qualify: 4xx greylisting and transport failures.
	A 5xx or a 250 is final.
	"""
	if transport_error:
	return True
	return is_greylist_code(code)