| """Low-level SMTP client for mailbox probing (RFC 5321). |
| |
| This is the core of the autonomous engine. Unlike a one-shot probe that opens a |
| TCP connection per address, `SmtpConnection` keeps a single session to a |
| receiving MX open and issues many `RCPT TO` probes over it (RSET between each). |
| Remote mail servers limit how often you may connect, and the TCP+TLS handshake |
| dominates cost — so connection reuse is what makes deep verification of a 10k |
| list feasible in minutes instead of hours. |
| |
| No message is ever transmitted: the dialogue stops after RCPT and issues RSET, |
| so nothing can be delivered or marked as spam. |
| |
| Every failure mode is returned as data (`RcptOutcome`), never raised, so one |
| dead host cannot abort a bulk job. |
| """ |
| from __future__ import annotations |
|
|
| import socket |
| from dataclasses import dataclass, field |
| from typing import Callable, Optional |
|
|
| from .mx_intel import is_greylist_code, is_permanent_reject, parse_rcpt_code |
|
|
| CRLF = "\r\n" |
|
|
|
|
| @dataclass |
| class RcptOutcome: |
| """Result of one RCPT TO probe.""" |
|
|
| code: Optional[int] = None |
| message: str = "" |
| transport_error: bool = False |
| error: Optional[str] = None |
|
|
| @property |
| def accepted(self) -> bool: |
| return self.code == 250 |
|
|
| @property |
| def permanent_reject(self) -> bool: |
| return is_permanent_reject(self.code) |
|
|
| @property |
| def retryable(self) -> bool: |
| return self.transport_error or is_greylist_code(self.code) |
|
|
|
|
| def _default_connect(host: str, port: int, timeout: float): |
| sock = socket.create_connection((host, port), timeout=timeout) |
| sock.settimeout(timeout) |
| return sock |
|
|
|
|
| class SmtpConnection: |
| """A single reusable SMTP session against one receiving host.""" |
|
|
| def __init__( |
| self, |
| host: str, |
| *, |
| port: int = 25, |
| timeout: float = 10.0, |
| helo_domain: str = "validator.local", |
| mail_from: str = "probe@validator.local", |
| connect: Optional[Callable[[str, int, float], object]] = None, |
| ): |
| self.host = host |
| self.port = port |
| self.timeout = timeout |
| self.helo_domain = helo_domain |
| self.mail_from = mail_from |
| self._connect = connect or _default_connect |
| self._sock = None |
| self.ready = False |
| self.server_code: Optional[int] = None |
| self.banner: str = "" |
| self._supports_pipelining = False |
|
|
| |
| def open(self) -> bool: |
| """Connect + banner + EHLO. Returns True when the session is usable.""" |
| try: |
| self._sock = self._connect(self.host, self.port, self.timeout) |
| except Exception as exc: |
| self._sock = None |
| self.banner = f"connect failed: {exc}" |
| self.ready = False |
| return False |
| try: |
| code, text = self._read_response() |
| self.banner = text |
| if code != 220: |
| self.ready = False |
| return False |
| self._send(f"EHLO {self.helo_domain}") |
| code, text = self._read_response() |
| if code != 250: |
| |
| self._send(f"HELO {self.helo_domain}") |
| code, text = self._read_response() |
| if code != 250: |
| self.ready = False |
| return False |
| self._supports_pipelining = "PIPELINING" in text.upper() |
| self.server_code = code |
| self.ready = True |
| return True |
| except Exception as exc: |
| self.banner = f"handshake failed: {exc}" |
| self.ready = False |
| return False |
|
|
| def _send(self, line: str) -> None: |
| if self._sock is None: |
| raise OSError("socket closed") |
| self._sock.sendall((line + CRLF).encode("utf-8")) |
|
|
| def _read_response(self) -> tuple[Optional[int], str]: |
| """Read one (possibly multiline) SMTP response. |
| |
| Multiline replies use `250-first`, `250-more`, `250 last` — the code is |
| repeated on every line and the final line has a space after the code. |
| Reading only one line would desynchronise the whole session. |
| """ |
| if self._sock is None: |
| raise OSError("socket closed") |
| lines: list[str] = [] |
| while True: |
| raw = self._sock.recv(4096) |
| if not raw: |
| raise OSError("connection closed by peer") |
| text = raw.decode("utf-8", "replace") |
| for line in text.split(CRLF): |
| if not line: |
| continue |
| lines.append(line) |
| |
| if len(line) >= 4 and line[3] == "-": |
| continue |
| joined = "\n".join(lines) |
| return parse_rcpt_code(line), joined |
| |
| if lines and len(lines[-1]) >= 4 and lines[-1][3] == " ": |
| joined = "\n".join(lines) |
| return parse_rcpt_code(lines[-1]), joined |
|
|
| |
| def probe(self, email: str) -> RcptOutcome: |
| """Ask the receiving server whether `email` is accepted.""" |
| if not self.ready or self._sock is None: |
| return RcptOutcome( |
| transport_error=True, error="session not ready", retryable=True |
| ) |
| try: |
| self._send(f"MAIL FROM:<{self.mail_from}>") |
| code, text = self._read_response() |
| if code is None or code >= 400: |
| return RcptOutcome( |
| code=code, |
| message=text, |
| transport_error=True, |
| error=f"MAIL FROM rejected: {text[:120]}", |
| ) |
|
|
| self._send(f"RCPT TO:<{email}>") |
| code, text = self._read_response() |
|
|
| |
| try: |
| self._send("RSET") |
| self._read_response() |
| except Exception: |
| pass |
|
|
| return RcptOutcome(code=code, message=text) |
| except Exception as exc: |
| return RcptOutcome(transport_error=True, error=str(exc)) |
|
|
| def quit(self) -> None: |
| try: |
| if self._sock is not None: |
| self._send("QUIT") |
| except Exception: |
| pass |
| finally: |
| self.close() |
|
|
| def close(self) -> None: |
| if self._sock is not None: |
| try: |
| self._sock.close() |
| except Exception: |
| pass |
| self._sock = None |
| self.ready = False |
|
|
| def __enter__(self) -> "SmtpConnection": |
| self.open() |
| return self |
|
|
| def __exit__(self, *exc) -> None: |
| self.quit() |
