File size: 6,961 Bytes
f5afa05 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 | """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
# ---- transport -------------------------------------------------
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: # connection refused / blocked / DNS
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:
# Fall back to HELO for ancient servers.
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)
# "250-last" continues, "250 last" or a short line ends.
if len(line) >= 4 and line[3] == "-":
continue
joined = "\n".join(lines)
return parse_rcpt_code(line), joined
# A chunk may end mid-line; keep reading.
if lines and len(lines[-1]) >= 4 and lines[-1][3] == " ":
joined = "\n".join(lines)
return parse_rcpt_code(lines[-1]), joined
# ---- probing ---------------------------------------------------
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()
# Always reset so the next probe starts clean; never send DATA.
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() |