diff --git a/deploy/env/mta-in-py.defaults b/deploy/env/mta-in-py.defaults index 4799230a..5ff64e64 100644 --- a/deploy/env/mta-in-py.defaults +++ b/deploy/env/mta-in-py.defaults @@ -18,17 +18,29 @@ PYMTA_MAX_SESSIONS_TOTAL=2000 PYMTA_MAX_RECIPIENTS=100 PYMTA_MAX_ENVELOPES_PER_CONNECTION=20 -# Timeouts (seconds). +# Timeouts (seconds). PYMTA_COMMAND_TIMEOUT is the idle gap allowed between +# complete commands; PYMTA_DATA_TIMEOUT is a separate total budget for the +# whole DATA phase (354 -> last body byte -> MDA deliver -> reply). PYMTA_COMMAND_TIMEOUT=120 -PYMTA_DATA_TIMEOUT=600 +PYMTA_DATA_TIMEOUT=300 # STARTTLS off by default in dev (no cert wired). PYMTA_TLS_CERT_FILE= PYMTA_TLS_KEY_FILE= -# SMTPUTF8 advertised — the MDA accepts UTF-8 envelope addresses. +# SMTPUTF8 advertised: the MDA accepts UTF-8 envelope addresses. PYMTA_ENABLE_SMTPUTF8=true -# PROXY protocol off in dev. In production set ENABLE_PROXY_PROTOCOL=haproxy -# (same env var the Postfix entrypoint already consumes) when behind HAProxy. +# PROXY protocol off in dev: the test suite connects to pymta directly, so the +# TCP peer IS the client. +# +# Behind a load balancer, set PYMTA_ENABLE_PROXY_PROTOCOL=true AND +# PYMTA_TRUSTED_PROXIES to the balancer's IPs/CIDRs. pymta refuses to start +# with PROXY protocol on and no allowlist. (The Postfix image drives the same +# feature from its own ENABLE_PROXY_PROTOCOL=haproxy; pymta ignores that name.) +# +# A balancer WITHOUT PROXY protocol is not a supported configuration: every +# session would bucket under the balancer's IP and every message would be +# stamped with it as the sender's address. PYMTA_ENABLE_PROXY_PROTOCOL=false +PYMTA_TRUSTED_PROXIES= diff --git a/src/mta-in/README.md b/src/mta-in/README.md index c8d223cb..1655488b 100644 --- a/src/mta-in/README.md +++ b/src/mta-in/README.md @@ -16,7 +16,7 @@ This directory ships **two** implementations in parallel. Both expose the same S | Image | `Dockerfile` | `Dockerfile.pymta` | | SMTP server | Postfix `smtpd` | `aiosmtpd 1.4.6` | | MDA glue | `src/delivery_milter.py` + `src/api/mda.py` (sync `requests`) | `src/pymta/*` (async `httpx`) | -| Prometheus metrics | — | `/metrics` on port `9100` | +| Prometheus metrics | none | `/metrics` on port `9100` | | Tests | `make test-mta-in` | `make test-mta-in-py` | | Lint | `make lint-mta-in` | `make lint-mta-in-py` | @@ -24,26 +24,53 @@ Both run as a stateless, queue-less SMTP front-end. After receiving an email thr - Validating each recipient with a REST API call to `{env.MDA_API_BASE_URL}/inbound/mta/check/` during the RCPT TO command. - Delivering the complete message via REST API call to `{env.MDA_API_BASE_URL}/inbound/mta/deliver/` during the DATA command. -- Translating the MDA outcome (200 + `status=ok` / 5xx / timeout) into a single SMTP reply line. +- Translating the MDA outcome into a single SMTP reply line. + +### Translating the MDA outcome + +Losing a legitimate message is worse than asking the sender to retry, so the permanent-rejection set is an explicit allow-list and everything else defers. + +The table below maps the response to `deliver/`; `check/` has its own table under it. + +| MDA `deliver/` response | SMTP reply | Why | +|---|---|---| +| `200` + `{"status": "ok"}` | `250` | Delivered to every recipient. | +| `400`, `413`, `415` | `554` / `5xx` | The message itself is unacceptable: unparseable, oversize, or wrong content type. A retry sends the same bytes. | +| `207 Multi-Status` | `451` | Partial delivery. The MDA cannot ask us to retry only the failed recipients, so the whole envelope must be retried. That duplicates for the recipients already served, which is preferable to losing the rest. Requires per-recipient delivery to be idempotent on the MDA side. | +| `401`, `403`, `429`, `404`, any other 4xx | `451` | Secret rotation, `exp` clock skew, throttling, a bad route. Operational events, not verdicts on the mail. | +| `5xx`, timeout, transport error | `451` | Upstream is unhealthy; also counts toward the circuit breaker. | +| `200` with an unrecognised body | `451` | Not proof of delivery. | + +Only 5xx and transport failures feed the circuit breaker. A `207` or a `401` is a complete answer from a healthy MDA. + +`check/` names **no** permanent statuses at all. Every entry above is a verdict on a *message*, and a recipient check carries none: a `400`, `413` or `415` there is a fault in the check request we built, so saying `554` would tell the sender a working address is permanently bad. The only permanent reply at RCPT TO comes from a `200` whose body says the mailbox does not exist. + +| MDA `check/` response | SMTP reply | Why | +|---|---|---| +| `200` + `{"": true}` | `250` | Mailbox exists. | +| `200` + `{"": false}` | `550` (`421` at the miss cutoff) | The MDA says there is no such mailbox. The only permanent rejection on this path. | +| `200` naming the address with anything but a bool | `451` | Not an answer about this mailbox. | +| `200` not naming the address at all | `451` | Empty body, unparseable body, a proxy's own 200 page, or a drifted response shape. | +| any non-200, timeout, transport error, open breaker | `451` | Nothing here is a verdict on the mailbox. | + +The last two rows are the reason `MDAResult.payload` is normalised to a dict and read strictly: the MDA returns one boolean per address it was asked about, so a missing key never legitimately means "no such mailbox" — it means we did not get the answer. Treating it as a miss would bounce real mail whenever something upstream of the MDA substituted the response. ### MDA wire contract Each MTA → MDA call is an HTTP `POST` carrying: -- **Body** — for `check/`, an `application/json` document `{"addresses": [...]}`; for `deliver/`, the full RFC 5322 message as `message/rfc822`. -- **Authorization** — `Bearer ` where the JWT is signed HS256 with `env.MDA_API_SECRET` and carries: - - `exp`: 60 s from issuance, anchored in UTC. - - `body_hash`: `sha256(body).hexdigest()` — binds the token to the exact bytes posted (replay-proof per-request). +- **Body**: for `check/`, an `application/json` document `{"addresses": [...]}`; for `deliver/`, the full RFC 5322 message as `message/rfc822`. +- **Authorization**: `Bearer ` where the JWT is signed HS256 with `env.MDA_API_SECRET` and carries: + - `exp`: `env.MDA_API_JWT_TTL` seconds from issuance, anchored in UTC. It has to cover the whole request *plus* clock skew against the MDA. + - `body_hash`: `sha256(body).hexdigest()`, which binds the token to the exact bytes posted, so a captured token cannot be reused to send *different* content. The MDA tracks no nonce, so replaying the same token with the same body until `exp` is not prevented; the two claims bound a captured token together, one in content and one in time. - Plus, for `deliver/`, envelope metadata claims (`sender`, `original_recipients`, `client_address`, `client_port`, `client_hostname`, `client_helo`, `size`). -- **Response** — `200 OK` + JSON for success; `4xx` for permanent reject; `5xx` for tempfail. Timeouts and transport errors are tempfail too. - -In production, run pymta with `MDA_API_BASE_URL=https://...` so the bearer token doesn't traverse the network in clear. The client logs a `WARNING` at startup if a non-local `http://` URL is configured. +- **Response**: see the outcome table above. ## When to use which Postfix is the production default. The pymta implementation is offered side-by-side so it can take over once parity is proven; it is easier to extend (no milter protocol, no C glue), gives us Prometheus metrics, and reduces the attack surface (no Postfix binary, no `libmilter`, no on-disk queue at all). -Switching production from one to the other only requires re-pointing the inbound public IP to the other container — the MDA back-end and the env vars are identical. PROXY-protocol passthrough is toggled by the same `ENABLE_PROXY_PROTOCOL=haproxy` env var on both. +Switching production from one to the other only requires re-pointing the inbound public IP to the other container. The MDA back-end and the env vars are identical, except that PROXY-protocol passthrough is `ENABLE_PROXY_PROTOCOL=haproxy` on Postfix and `PYMTA_ENABLE_PROXY_PROTOCOL=true` on pymta. ## Running @@ -65,22 +92,71 @@ In addition to the shared `MDA_API_BASE_URL` / `MDA_API_SECRET` / `MDA_API_TIMEO | Variable | Default | Purpose | |---|---|---| -| `PYMTA_HOSTNAME` | `mta-in` | Banner / Received-header host name | +| `PYMTA_HOSTNAME` | `$MYHOSTNAME`, else `mta-in` | Banner / Received-header host name | +| `PYMTA_IDENT` | `ESMTP` | Banner text after the hostname; kept version-less on purpose | | `PYMTA_SMTP_HOST` / `PYMTA_SMTP_PORT` | `0.0.0.0` / `25` | SMTP listener bind | +| `PYMTA_LOG_LEVEL` | `INFO` | Root log level | | `PYMTA_METRICS_HOST` / `PYMTA_METRICS_PORT` | `0.0.0.0` / `9100` | Prometheus endpoint (set port to 0 to disable) | | `PYMTA_MAX_RECIPIENTS` | `100` | RCPT TO cap per envelope | | `PYMTA_MAX_ENVELOPES_PER_CONNECTION` | `10` | Envelopes per TCP session | | `PYMTA_HARD_ERROR_LIMIT` | `50` | 4xx/5xx replies before forcing 421 + disconnect | | `PYMTA_MAX_RCPT_MISSES_PER_SESSION` | `10` | Unknown-mailbox lookups before 421 + disconnect | | `PYMTA_MAX_SESSIONS_PER_IP` | `100` (0 = off) | Per-IP concurrent session cap | -| `PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE` | `600` (0 = off) | Per-IP new-session rate cap (rolling 60 s window) | +| `PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE` | `600` (0 = off) | Per-IP new-session rate cap (fixed 60 s window) | | `PYMTA_MAX_SESSIONS_TOTAL` | `1000` (0 = off) | Process-wide concurrent session cap | -| `PYMTA_COMMAND_TIMEOUT` | `120` | Per-command idle timeout (s) | -| `PYMTA_DATA_TIMEOUT` | `600` | Total DATA-phase deadline (s) | +| `PYMTA_COMMAND_TIMEOUT` | `120` | Idle timeout (s) between complete commands; re-armed by each accepted command | +| `PYMTA_MAX_SESSION_SECONDS` | `1800` (0 = off) | Wall-clock ceiling (s) on one TCP session, armed at connect and never re-armed | +| `PYMTA_DATA_TIMEOUT` | `300` | Hard DATA-phase deadline (s): 354 → last body byte → MDA deliver → reply. Nothing in a DATA phase outlives it | +| `MDA_API_JWT_TTL` | `MDA_API_TIMEOUT + 90` | Lifetime (s) of the JWT signed for each MDA call | +| `PYMTA_TRUSTED_PROXIES` | empty | Comma-separated IPs/CIDRs allowed to send a PROXY header. Strongly recommended when PROXY protocol is on: empty means *every* peer's header is trusted, and startup only warns. Ignored when PROXY protocol is off | | `PYMTA_SHUTDOWN_TIMEOUT` | `25` | Drain deadline on SIGTERM before abandoning in-flight sessions (s) | | `PYMTA_MDA_BREAKER_THRESHOLD` | `10` (0 = off) | Consecutive MDA failures before short-circuiting to 451 | | `PYMTA_MDA_BREAKER_COOLDOWN` | `30` | Seconds the breaker stays open before probing the MDA again | | `PYMTA_TLS_CERT_FILE` / `PYMTA_TLS_KEY_FILE` | empty | STARTTLS cert + key paths (empty = STARTTLS off) | -| `STARTTLS_CHAIN_FILES` | empty | Postfix-compatible fallback — comma-separated PEM bundle(s); first bundle wins when `PYMTA_TLS_*` is unset | +| `STARTTLS_CHAIN_FILES` | empty | Postfix-compatible fallback: comma-separated PEM bundle(s); first bundle wins when `PYMTA_TLS_*` is unset | | `PYMTA_ENABLE_SMTPUTF8` | `true` | Advertise SMTPUTF8 in EHLO | -| `ENABLE_PROXY_PROTOCOL` | unset | Set to `haproxy` to enable PROXY-protocol v1/v2 | +| `PYMTA_ENABLE_PROXY_PROTOCOL` | `false` | Enable PROXY-protocol v1/v2. The Postfix image uses its own `ENABLE_PROXY_PROTOCOL=haproxy`; pymta does not read that name, so the two can share an env file | +| `PYMTA_PROXY_PROTOCOL_TIMEOUT` | `5` | Seconds to wait for the PROXY header before dropping the connection | + +Integer settings are range-checked at startup: those documented as `0 = off` accept zero, the rest reject it. `PYMTA_MAX_LOCAL_PART` and `PYMTA_MAX_DOMAIN` are **not** configurable. They are the RFC 5321 §4.5.3.1 constants (64 / 255 octets), and raising them would only widen the gap between what pymta accepts at RCPT and what the MDA can store. + +## Production checklist + +The defaults are tuned for the dev stack. Five things to set before pymta faces the internet; each has a legitimate reason to differ in dev, so the process does not enforce them. + +**1. Pick a supported topology, and isolate port 25.** There are exactly two: + +| | PROXY protocol | `PYMTA_TRUSTED_PROXIES` | Client IP comes from | +|---|---|---|---| +| **Behind a load balancer** | `PYMTA_ENABLE_PROXY_PROTOCOL=true` | **strongly recommended**: the balancer's IPs/CIDRs | the PROXY header | +| **Directly exposed** | off | ignored | the TCP peer | + +**A load balancer without PROXY protocol is not supported.** pymta would see only the balancer's address, so every session would bucket under one IP (`PYMTA_MAX_SESSIONS_PER_IP` silently becomes a second, much lower global cap), and every message would be stamped with the balancer's own IP as the sender's, in `Received` and in the envelope the MDA stores. There is no setting for this topology and no fallback that makes it work. + +An empty `PYMTA_TRUSTED_PROXIES` with PROXY protocol on means **every peer's header is trusted** — there is nothing left to filter on, exactly as if you had written `0.0.0.0/0`. It starts, with a `SECURITY:` warning on the first lines of the log, because deployments whose balancer addresses are not known at boot still have to run. In that mode the network isolation below is the *entire* trust boundary; name the balancer whenever you can. + +The one thing pymta does enforce: with PROXY protocol on it never falls back to the TCP peer for `client_address`, so a header-less connection (a v2 `LOCAL` health check) reports no client rather than naming the balancer. + +Network isolation has to come from the deployment. A PROXY header is trusted on the word of the peer that sent it, and it sets both the key for every per-IP cap and the `client_address` the MDA writes into `Received`. A peer that can open a TCP connection to port 25 directly (a container port published on a node, a second interface, a foothold on an internal network) can otherwise spread a forged source across the address space until only the global cap applies, and attribute its mail to any IP it names. Enforce the isolation in the NetworkPolicy or firewall as well as in the allowlist. + +**2. `MDA_API_BASE_URL` must be `https://`.** The MTA→MDA channel is authenticated by a bearer token; over plaintext to a non-local host, anyone on the path can read it. pymta logs a startup warning when it sees `http://` pointing anywhere but localhost. + +**3. `MDA_API_SECRET` should be at least 32 bytes of real entropy, and not the dev value.** It is an HS256 shared secret: short ones are brute-forceable offline from a single captured JWT, and `my-shared-secret-mda` is in the repo. pymta warns at startup below 32 bytes, and again if the secret is missing entirely. + +**4. Keep the metrics port off the interface that serves port 25.** `PYMTA_METRICS_HOST` defaults to `0.0.0.0` because the usual scrape paths (Prometheus hitting the pod IP, a compose port mapping) cannot reach a loopback-only listener. The exposition carries no addresses and no message content, so what leaks is operational: volumes, rejection reasons, breaker state. Restrict it with a NetworkPolicy, or set `PYMTA_METRICS_HOST=127.0.0.1` / `PYMTA_METRICS_PORT=0` where you can. + +**5. Size the session-slot budget.** `PYMTA_MAX_SESSIONS_TOTAL` slots are the contended resource, and `PYMTA_MAX_SESSION_SECONDS` bounds how long one connection can occupy one. + +Every other timeout is re-armed by peer activity, so a peer that stays marginally active can hold a session open for far longer than `PYMTA_COMMAND_TIMEOUT` suggests. The session cap is the one bound that is not re-armed. It does not prevent a distributed attacker from occupying slots, but it means blocking one frees the slots instead of leaving them held until the process restarts, and it stops `PYMTA_SHUTDOWN_TIMEOUT` (25 s) from severing long-lived sessions on every rollout. + +`PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE` does not constrain slot occupancy at any usable value: sustainable held slots per IP is `min(max_per_ip, rate_per_minute × session_minutes)`, so the concurrent cap binds first for any rate a real sender could tolerate. It defends against fast open/close churn (CPU, TLS handshakes, MDA recipient checks). + +**Do not tighten `PYMTA_MAX_SESSIONS_PER_IP` far.** Postfix's `default_destination_concurrency_limit` is 20, so a single default-configured relay sending a backlog sits at a cap of 20 and gets 421s on its 21st connection; busy relays raise that figure. The worst case for a tight cap is a queue flush: after pymta downtime or a tripped MDA breaker, every sender with a backlog retries at full concurrency, and a low cap extends the outage by rejecting the senders trying to drain into you. The gain in return is small, since the number of source IPs an attacker needs scales only linearly with the cap. 100 is a reasonable default, and the shipping Postfix config disables per-client limits entirely (`smtpd_client_event_limit_exceptions = static:all`), so it is already a tightening. + +Slot exhaustion on an *inbound* MTA delays mail rather than losing it: once the global cap is hit, new connections get `421` and close, and SMTP senders retry for days. Watch `pymta_sessions_active` and the upper buckets of `pymta_session_duration_seconds`; a healthy inbound MX has sessions measured in seconds. `PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE` refuses churn before the dialogue opens: the gate is acquired ahead of the `220` greeting, and TLS here is STARTTLS (an in-session command, not a handshake on accept), so a refused connection costs no asymmetric crypto and no MDA recipient check. What it still costs is the TCP accept, one protocol object, and the `421` write — small per connection, but paid inside this process. Only a cap upstream of it (firewall, load balancer, network ACL) drops the packet before that. + +Also review `PYMTA_DATA_TIMEOUT` against your real `MAX_INCOMING_EMAIL_SIZE`; it sets a floor on how slow a legitimate sender may be. The dev defaults file disables the per-IP cap entirely, because all local load comes from loopback. + +## Address normalisation is a cross-service contract + +pymta lower-cases the domain and preserves the local-part's case; the MDA matches mailboxes on the exact `(local_part, domain)` tuple. RCPT-check and deliver send the same string, so the two always agree with each other. But a mailbox registered lower-case will get a `550` for a mixed-case local-part. The sender sees the rejection rather than losing the message silently. Recorded here so the two services do not drift apart. diff --git a/src/mta-in/pyproject.toml b/src/mta-in/pyproject.toml index e9d8c500..1fce1979 100644 --- a/src/mta-in/pyproject.toml +++ b/src/mta-in/pyproject.toml @@ -96,7 +96,7 @@ select = [ [tool.ruff.lint.isort] section-order = ["future","standard-library","django","third-party","mta","first-party","local-folder"] -sections = { django=["django"] } +sections = { mta=["pymta", "api"], django=["django"] } extra-standard-library = ["tomllib"] [tool.ruff.lint.per-file-ignores] diff --git a/src/mta-in/src/delivery_milter.py b/src/mta-in/src/delivery_milter.py index e3fe65ea..a32e3abc 100644 --- a/src/mta-in/src/delivery_milter.py +++ b/src/mta-in/src/delivery_milter.py @@ -85,7 +85,7 @@ class DeliveryMilter(Milter.Base): self.rcpttos.append(clean_to) return Milter.CONTINUE - except Exception: # noqa: BLE001 + except Exception: # Exception during validation - temporary failure return Milter.TEMPFAIL @@ -150,7 +150,7 @@ class DeliveryMilter(Milter.Base): else: return Milter.TEMPFAIL - except Exception: # noqa: BLE001 + except Exception: return Milter.TEMPFAIL def close(self): @@ -165,7 +165,7 @@ class DeliveryMilter(Milter.Base): def main(): """Run the milter server""" - print("Starting delivery milter...") # noqa: T201 + print("Starting delivery milter...") # Set the socket for milter communication # Use Unix socket for better performance and security @@ -179,7 +179,7 @@ def main(): try: os.setgid(grp.getgrnam("postfix").gr_gid) except (KeyError, OSError) as e: - print(f"Warning: could not set gid to postfix: {e}", file=sys.stderr) # noqa: T201 + print(f"Warning: could not set gid to postfix: {e}", file=sys.stderr) os.umask(0o117) # Register our milter class @@ -189,15 +189,15 @@ def main(): flags = Milter.CHGBODY + Milter.CHGHDRS + Milter.ADDHDRS Milter.set_flags(flags) - print(f"Milter listening on {socket_path}") # noqa: T201 + print(f"Milter listening on {socket_path}") try: # Start the milter Milter.runmilter("delivery_milter", socket_path, timeout=240) except KeyboardInterrupt: - print("Milter shutting down...") # noqa: T201 - except Exception as e: # noqa: BLE001 - print(f"Milter error: {e}") # noqa: T201 + print("Milter shutting down...") + except Exception as e: + print(f"Milter error: {e}") sys.exit(1) diff --git a/src/mta-in/src/pymta/address.py b/src/mta-in/src/pymta/address.py index be98e506..71527d64 100644 --- a/src/mta-in/src/pymta/address.py +++ b/src/mta-in/src/pymta/address.py @@ -1,7 +1,7 @@ """RFC 5321 envelope-address validation. The functions in this module are intentionally strict: they reject anything -the inbound SMTP server should not have to deal with — source routes +the inbound SMTP server should not have to deal with: source routes (RFC 5321 §4.1.1.3), control characters (CRLF injection vector), overlong local-parts or domains, and the common ``user@`` / ``@domain`` truncations. @@ -104,7 +104,7 @@ def validate_envelope_address( # noqa: PLR0912 # ----- 3. exactly one unquoted '@' --------------------------------------- # Quoted local-parts could legally contain '@', but we don't accept those - # on the public inbound path — most senders never use them and they are + # on the public inbound path. Most senders never use them and they are # a fertile parser-confusion ground. if address.count("@") != 1: raise AddressError( diff --git a/src/mta-in/src/pymta/controller.py b/src/mta-in/src/pymta/controller.py index 57acc06f..1e568043 100644 --- a/src/mta-in/src/pymta/controller.py +++ b/src/mta-in/src/pymta/controller.py @@ -1,6 +1,6 @@ """aiosmtpd Controller wired to our :class:`HardenedSMTP` factory. -The Controller itself is unchanged structurally — all hardening lives inside +The Controller itself is unchanged structurally; all hardening lives inside :class:`HardenedSMTP` so the admission gate runs in the same coroutine that will dispatch SMTP verbs. """ @@ -62,7 +62,7 @@ def build_smtp_kwargs(*, tls_context: ssl.SSLContext | None) -> dict: def load_tls_context() -> ssl.SSLContext | None: """Build a TLS context from the configured cert/key, or None. - Returning None disables STARTTLS — aiosmtpd will not advertise it. + Returning None disables STARTTLS, so aiosmtpd will not advertise it. """ cert = settings.PYMTA_TLS_CERT_FILE key = settings.PYMTA_TLS_KEY_FILE diff --git a/src/mta-in/src/pymta/handler.py b/src/mta-in/src/pymta/handler.py index f3b7a08a..b5fc5a42 100644 --- a/src/mta-in/src/pymta/handler.py +++ b/src/mta-in/src/pymta/handler.py @@ -5,7 +5,7 @@ For each SMTP transaction the handler 1. validates EHLO syntax, 2. validates and stores MAIL FROM (allowing the null sender), 3. on RCPT TO: validates the address shape, then calls the MDA - ``inbound/mta/check/`` endpoint synchronously — RCPT is rejected with a + ``inbound/mta/check/`` endpoint synchronously. RCPT is rejected with a permanent 5xx if the mailbox does not exist, a 4xx if the check itself fails or times out, 4. on DATA: forwards the full message bytes to ``inbound/mta/deliver/`` @@ -19,7 +19,9 @@ peer should retry later. from __future__ import annotations import asyncio +import ipaddress import logging +import time from . import metrics, settings from .address import AddressError, validate_envelope_address @@ -28,9 +30,12 @@ from .mda_async import MDAClient, MDAResult logger = logging.getLogger(__name__) -# Per-session counters live on the Session object (one per TCP connection). -# aiosmtpd resets ``envelope`` after each DATA so we cannot stash counters -# there; we attach to ``session`` via setattr instead. +# Abuse counters are keyed to the TCP connection, so they live on the *server* +# (the per-connection SMTP protocol instance) rather than on ``session``. +# aiosmtpd resets ``envelope`` after each DATA, and rebuilds ``session`` from +# scratch on STARTTLS, so stashing them on the session would let a peer wipe its +# own enumeration and error budgets by issuing STARTTLS mid-stream. The server +# instance survives both. See _PROXY_SRC_ATTR below for the same reasoning. _ENVELOPES_ATTR = "_pymta_envelopes" _SOFT_ERRORS_ATTR = "_pymta_soft_errors" _RCPT_MISSES_ATTR = "_pymta_rcpt_misses" @@ -51,45 +56,84 @@ _PROXY_SRC_ATTR = "_pymta_proxy_src" # milter's existing wire contract. NULL_SENDER_SENTINEL = "<>" +# Slice of PYMTA_DATA_TIMEOUT the handler holds back from the MDA deliver call +# so it still has room to push a 451 before the protocol-level deadline closes +# the transport. Defined in settings beside the timeout it is subtracted from, +# which refuses to start a configuration where it would consume the whole +# budget. +_REPLY_RESERVE_SECONDS = float(settings.REPLY_RESERVE_SECONDS) + # Control characters that must never appear in an EHLO/HELO hostname or # anywhere else we'll log / pass into HTTP claims. CR, LF, NUL are the # CRLF-injection vectors; TAB is a header-unfolding vector. _FORBIDDEN_HOSTNAME_CHARS = frozenset({"\r", "\n", "\x00", "\t"}) -def _envelopes_count(session) -> int: - return getattr(session, _ENVELOPES_ATTR, 0) +def _counters(server, session): + """Return the object the per-connection counters hang off. + + The server when we have one (the normal path, since it outlives the STARTTLS + session rebuild); the session otherwise, which keeps the handler usable + from unit tests that pass ``server=None``. + """ + return session if server is None else server -def _bump_envelopes(session) -> int: - n = _envelopes_count(session) + 1 - setattr(session, _ENVELOPES_ATTR, n) +def _count(holder, attr: str) -> int: + return getattr(holder, attr, 0) + + +def _bump(holder, attr: str) -> int: + n = _count(holder, attr) + 1 + setattr(holder, attr, n) return n -def _bump_soft_errors(session) -> int: - n = getattr(session, _SOFT_ERRORS_ATTR, 0) + 1 - setattr(session, _SOFT_ERRORS_ATTR, n) - return n +def _hard_error_limit_reached(server, counters) -> bool: + """True when this connection has spent its ``PYMTA_HARD_ERROR_LIMIT``. + + Records the rejection and asks for the disconnect, so a caller only has to + return the 421. Bulk address enumeration and dictionary attacks otherwise + keep hammering a single TCP session. + """ + if _count(counters, _SOFT_ERRORS_ATTR) < settings.PYMTA_HARD_ERROR_LIMIT: + return False + metrics.SECURITY_REJECTIONS.labels(reason="hard_error_limit").inc() + metrics.DISCONNECTS_421.labels(reason="hard_error_limit").inc() + _request_disconnect(server) + return True -def _bump_rcpt_misses(session) -> int: - n = getattr(session, _RCPT_MISSES_ATTR, 0) + 1 - setattr(session, _RCPT_MISSES_ATTR, n) - return n +def _request_disconnect(server) -> None: + """Ask the protocol to close once the reply we are returning is on the wire. + + A 421 is a promise to hang up; aiosmtpd on its own would push the string + and keep serving the session. Silently a no-op for the unit-test fakes. + """ + requester = getattr(server, "request_disconnect", None) + if requester is not None: + requester() def _peer_ip(session, server=None) -> str | None: # Prefer the PROXY source captured at connect time and stashed on the # server: it is the only copy that survives the STARTTLS session rebuild # (see _PROXY_SRC_ATTR). Fall back to session.proxy_data for the pre-TLS - # window, then to the raw TCP peer when PROXY protocol is off. + # window, then to the raw TCP peer. + # + # That last fallback is only reachable with PROXY protocol off, where the + # wire peer IS the client. With it on, the wire peer is the balancer, and + # returning it would attribute the mail to our own infrastructure: a + # plausible-looking but wrong IP in the Received header. Report nothing + # instead; the MDA omits the trace when a part is missing. stashed = getattr(server, _PROXY_SRC_ATTR, None) if server is not None else None if stashed is not None and stashed[0]: return str(stashed[0]) proxy_data = getattr(session, "proxy_data", None) if proxy_data is not None and getattr(proxy_data, "src_addr", None): return str(proxy_data.src_addr) + if settings.PYMTA_ENABLE_PROXY_PROTOCOL: + return None peer = getattr(session, "peer", None) if peer and len(peer) >= 1: return str(peer[0]) @@ -103,12 +147,69 @@ def _peer_port(session, server=None) -> str | None: proxy_data = getattr(session, "proxy_data", None) if proxy_data is not None and getattr(proxy_data, "src_port", None) is not None: return str(proxy_data.src_port) + if settings.PYMTA_ENABLE_PROXY_PROTOCOL: + return None peer = getattr(session, "peer", None) if peer and len(peer) >= 2: return str(peer[1]) return None +def _remaining_data_budget(server) -> float: + """Seconds still available for the MDA deliver call. + + ``PYMTA_DATA_TIMEOUT`` is one budget covering the whole DATA phase, and the + protocol arms the transport at exactly that value, so the body receive has + already spent part of it, and we stop short of the rest. The reserve is + what buys the 451 on the timeout path: expiring at the same instant as the + transport would be a race we'd usually lose (its timer was scheduled first, + and equal deadlines fire in scheduling order), leaving the peer with a bare + disconnect instead of a reason to retry. + + Falls back to the full budget when the caller has no DATA start timestamp + (unit-test fakes). Raises ``TimeoutError`` when the receive already spent + the whole budget: issuing the deliver call anyway would overrun the + transport deadline the reserve exists to stay clear of, and the caller + already turns that exception into the 451. + """ + started = getattr(server, "data_phase_started", None) + if started is None: + return float(settings.PYMTA_DATA_TIMEOUT) + spent = time.monotonic() - started + remaining = settings.PYMTA_DATA_TIMEOUT - spent - _REPLY_RESERVE_SECONDS + if remaining <= 0: + raise TimeoutError("DATA budget spent before the deliver call") + return remaining + + +def _proxy_header_is_trusted(session) -> bool: + """True when the *wire* peer is allowed to speak PROXY protocol to us. + + aiosmtpd parses a PROXY header from whoever sent it and has no notion of + a trusted upstream. Since the header dictates both the rate-limit key and + the ``client_address`` the MDA records, an unfiltered one hands a direct + connector a free pass past every per-IP cap and the ability to attribute + its mail to any address it names. + + An *empty* allowlist means "no upstream named", and there is nothing left + to filter on: every header is trusted, exactly as if the allowlist were + 0.0.0.0/0. That is a deliberate escape hatch for deployments whose balancer + addresses are not known at boot, and it puts the whole trust boundary on + the network isolation; ``server._check_proxy_trust_config`` warns loudly + about it at startup. Naming the balancer is the safe configuration. + """ + if not settings.PYMTA_TRUSTED_PROXIES: + return True + peer = getattr(session, "peer", None) + if not peer: + return False + try: + wire_ip = ipaddress.ip_address(str(peer[0])) + except ValueError: + return False + return any(wire_ip in network for network in settings.PYMTA_TRUSTED_PROXIES) + + def _safe_hostname(raw: str | None, session=None) -> str | None: """Return ``raw`` only if it is free of CRLF/NUL/TAB; otherwise None. @@ -157,7 +258,7 @@ class InboundHandler: if verb in denied_verbs: metrics.SECURITY_REJECTIONS.labels(reason="auth_offered").inc() logger.warning( - "stripping disallowed EHLO extension %r from %s — review the " + "stripping disallowed EHLO extension %r from %s. Review the " "SMTP configuration so it is not advertised in the first place", verb, _peer_ip(session), @@ -165,6 +266,16 @@ class InboundHandler: continue clean.append(line) + # Re-mark the terminator. aiosmtpd appends "250 HELP" last, so today + # the line we drop is never the final one. But a multiline reply whose + # last line still reads "250-" leaves clients waiting forever for a + # continuation that will not come, and that is too sharp an edge to + # leave resting on an upstream implementation detail. + if clean: + last = clean[-1] + if last.startswith("250-"): + clean[-1] = "250 " + last[4:] + session.host_name = _safe_hostname(hostname, session=session) return clean @@ -174,6 +285,14 @@ class InboundHandler: # ------------------------------------------------------------------ MAIL async def handle_MAIL(self, server, session, envelope, address, mail_options): + counters = _counters(server, session) + + # Same hard-error budget as handle_RCPT: a peer can burn errors on + # MAIL alone (malformed senders, bad SIZE), so the guard has to sit on + # this verb too or the budget is trivially sidestepped. + if _hard_error_limit_reached(server, counters): + return "421 4.7.0 Too many errors, goodbye" + try: clean = validate_envelope_address( address, @@ -183,19 +302,20 @@ class InboundHandler: ) except AddressError as err: metrics.SECURITY_REJECTIONS.labels(reason=err.reason).inc() + _bump(counters, _SOFT_ERRORS_ATTR) return f"{err.smtp_code} {err.smtp_text}" - # Honour MAIL FROM:... SIZE=N if announced — fail fast before DATA. + # Honour MAIL FROM:... SIZE=N if announced, to fail fast before DATA. for opt in mail_options or []: if opt.upper().startswith("SIZE="): try: announced = int(opt.split("=", 1)[1]) except ValueError: - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "501 5.5.4 Bad SIZE parameter" if announced > settings.MAX_INCOMING_EMAIL_SIZE: metrics.SECURITY_REJECTIONS.labels(reason="oversize_announced").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "552 5.3.4 Message size exceeds fixed maximum" envelope.mail_from = clean if clean else NULL_SENDER_SENTINEL @@ -204,13 +324,10 @@ class InboundHandler: # ------------------------------------------------------------------ RCPT async def handle_RCPT(self, server, session, envelope, address, rcpt_options): # noqa: PLR0911 - # First gate: hard-error budget. Once the session has accumulated - # ``PYMTA_HARD_ERROR_LIMIT`` 4xx/5xx replies, send 421 and close so - # bulk address enumeration / dictionary attacks cannot keep hammering - # this single TCP session. - if getattr(session, _SOFT_ERRORS_ATTR, 0) >= settings.PYMTA_HARD_ERROR_LIMIT: - metrics.SECURITY_REJECTIONS.labels(reason="hard_error_limit").inc() - metrics.DISCONNECTS_421.labels(reason="hard_error_limit").inc() + counters = _counters(server, session) + + # First gate: hard-error budget, shared with handle_MAIL. + if _hard_error_limit_reached(server, counters): metrics.RCPT_TOTAL.labels(result="rejected_temp").inc() return "421 4.7.0 Too many errors, goodbye" @@ -218,7 +335,7 @@ class InboundHandler: if len(envelope.rcpt_tos) >= settings.PYMTA_MAX_RECIPIENTS: metrics.SECURITY_REJECTIONS.labels(reason="max_recipients").inc() metrics.RCPT_TOTAL.labels(result="rejected_temp").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "452 4.5.3 Too many recipients" try: @@ -231,27 +348,49 @@ class InboundHandler: except AddressError as err: metrics.SECURITY_REJECTIONS.labels(reason=err.reason).inc() metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return f"{err.smtp_code} {err.smtp_text}" result = await self.mda.check_recipient(clean) if result.temp_fail: metrics.RCPT_TOTAL.labels(result="rejected_temp").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "451 4.3.0 Recipient verification temporarily unavailable" + # Unreachable while ``check_recipient`` names no permanent statuses: + # every non-200 there comes back as a temp_fail above. Kept as the + # landing spot if one is ever added. if not result.ok: metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "550 5.1.1 Recipient verification failed" - exists = bool(result.payload.get(clean, False)) - if not exists: + # Only an explicit true/false for this exact address is a verdict. The + # MDA answers with one boolean per address it was asked about, so a + # missing key or any other type means the body is not the answer we + # asked for: an empty or unparseable body, a proxy's own 200 page, or a + # response shape that has drifted. Reading that as "no such mailbox" + # would bounce a working address on someone else's bug, so it defers + # like every other unusable check response. + verdict = result.payload.get(clean) + if not isinstance(verdict, bool): + metrics.RCPT_TOTAL.labels(result="rejected_temp").inc() + _bump(counters, _SOFT_ERRORS_ATTR) + logger.warning( + "MDA check returned HTTP %d with no usable verdict for %r (%s); deferring", + result.status_code, + clean, + type(verdict).__name__, + ) + return "451 4.3.0 Recipient verification temporarily unavailable" + + if not verdict: metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) - misses = _bump_rcpt_misses(session) + _bump(counters, _SOFT_ERRORS_ATTR) + misses = _bump(counters, _RCPT_MISSES_ATTR) if misses >= settings.PYMTA_MAX_RCPT_MISSES_PER_SESSION: metrics.SECURITY_REJECTIONS.labels(reason="max_rcpt_misses").inc() metrics.DISCONNECTS_421.labels(reason="max_rcpt_misses").inc() + _request_disconnect(server) return "421 4.7.0 Too many unknown recipients, goodbye" return "550 5.1.1 No such recipient" @@ -262,21 +401,22 @@ class InboundHandler: # ------------------------------------------------------------------ DATA async def handle_DATA(self, server, session, envelope): # noqa: PLR0911 - envelopes = _bump_envelopes(session) + counters = _counters(server, session) + envelopes = _bump(counters, _ENVELOPES_ATTR) if envelopes > settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION: metrics.SECURITY_REJECTIONS.labels(reason="max_envelopes").inc() metrics.MESSAGES_TOTAL.labels(result="rejected_temp").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "451 4.7.0 Too many messages this session" content: bytes = envelope.content or b"" # NUL bytes have no place in an RFC 5321 message and break downstream - # C parsers — reject before we pay the cost of the deliver call. + # C parsers. Reject before we pay the cost of the deliver call. if b"\x00" in content: metrics.SECURITY_REJECTIONS.labels(reason="nul_byte").inc() metrics.MESSAGES_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "554 5.6.0 NUL byte in message body" if len(content) > settings.MAX_INCOMING_EMAIL_SIZE: @@ -284,10 +424,14 @@ class InboundHandler: # exceeds data_size_limit, so reaching here is defensive only. metrics.SECURITY_REJECTIONS.labels(reason="oversize_announced").inc() metrics.MESSAGES_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "552 5.3.4 Message size exceeds fixed maximum" try: + # Before building the coroutine: a budget already spent raises + # here, and a coroutine created but never awaited would only add a + # RuntimeWarning to the log on the way to the same 451. + budget = _remaining_data_budget(server) sender = envelope.mail_from if sender == NULL_SENDER_SENTINEL: sender = "" @@ -302,16 +446,18 @@ class InboundHandler: # own Received header using metadata and can decide what # to do with the missing hostname. client_hostname=None, - client_helo=_safe_hostname(getattr(session, "host_name", None), session=session), + client_helo=_safe_hostname( + getattr(session, "host_name", None), session=session + ), ), - timeout=settings.PYMTA_DATA_TIMEOUT, + timeout=budget, ) except TimeoutError: metrics.MESSAGES_TOTAL.labels(result="rejected_temp").inc() metrics.MESSAGE_BYTES.observe(len(content)) - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) logger.warning( - "DATA deliver deadline exceeded (%ds) for peer %s", + "DATA deadline exceeded (%ds total for receive + deliver) for peer %s", settings.PYMTA_DATA_TIMEOUT, _peer_ip(session, server), ) @@ -322,12 +468,27 @@ class InboundHandler: if result.ok and result.payload.get("status") == "ok": metrics.MESSAGES_TOTAL.labels(result="delivered").inc() return "250 2.0.0 Message accepted for delivery" - if result.temp_fail: + + # Anything short of an unambiguous "ok" defers. In particular the MDA + # answers 207 when it delivered to some recipients and not others: it + # cannot tell us to retry just the stragglers, so the only way they + # ever arrive is a retry of the whole envelope. That duplicates for the + # recipients already served, which is the right trade against losing the rest, + # and the behaviour the Postfix milter has always had (anything but + # 200 + status=ok is TEMPFAIL). Permanent rejection is reserved for the + # statuses that mean *this message* is unacceptable; see + # ``mda_async._PERMANENT_STATUSES``. + if result.temp_fail or result.ok: metrics.MESSAGES_TOTAL.labels(result="rejected_temp").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) + logger.warning( + "deferring message: MDA deliver returned HTTP %d payload-status %r", + result.status_code, + result.payload.get("status"), + ) return "451 4.3.0 Delivery temporarily unavailable" metrics.MESSAGES_TOTAL.labels(result="rejected_perm").inc() - _bump_soft_errors(session) + _bump(counters, _SOFT_ERRORS_ATTR) return "554 5.6.0 Message rejected by delivery agent" # ------------------------------------------------------------------ PROXY @@ -340,7 +501,21 @@ class InboundHandler: every session behind HAProxy would be bucketed under one address and ``PYMTA_MAX_SESSIONS_PER_IP`` would silently turn into a global cap. + + Returning False makes aiosmtpd drop the connection without ever + starting the SMTP dialogue. """ + if not _proxy_header_is_trusted(session): + metrics.SECURITY_REJECTIONS.labels(reason="untrusted_proxy").inc() + metrics.CONNECTIONS_TOTAL.labels(result="rejected_untrusted_proxy").inc() + logger.warning( + "rejecting PROXY header from untrusted peer %r (claimed src=%s); " + "add it to PYMTA_TRUSTED_PROXIES if this is a real balancer", + getattr(session, "peer", None), + getattr(proxy_data, "src_addr", None), + ) + return False + real_ip = "unknown" if proxy_data is not None and getattr(proxy_data, "src_addr", None): real_ip = str(proxy_data.src_addr) diff --git a/src/mta-in/src/pymta/limits.py b/src/mta-in/src/pymta/limits.py index acb785bd..61f3aca4 100644 --- a/src/mta-in/src/pymta/limits.py +++ b/src/mta-in/src/pymta/limits.py @@ -5,12 +5,12 @@ The :class:`IPGate` enforces three ceilings on inbound TCP sessions: * a process-wide cap, defending against a generic flood; * a per-IP concurrent cap, defending against a single remote opening thousands of half-idle connections (aiosmtpd does not enforce any per-IP cap); -* a per-IP new-session rate cap (rolling 60s window), defending against fast +* a per-IP new-session rate cap (fixed 60s window), defending against fast open/close churn from one IP that never exceeds the concurrent cap but still costs CPU/TLS handshakes/MDA RCPT checks. All caps are skipped when set to 0, matching the existing Postfix default -(``smtpd_client_event_limit_exceptions = static:all``) — useful in dev/test +(``smtpd_client_event_limit_exceptions = static:all``), which is useful in dev/test where the whole load comes from the same loopback address. """ @@ -25,7 +25,10 @@ from . import metrics logger = logging.getLogger(__name__) -# Rolling window used by the per-IP rate cap. +# Window used by the per-IP rate cap. Fixed, not sliding: the counter resets +# wholesale once the window elapses, so a peer timed to straddle a boundary can +# land up to 2x the cap back to back. Acceptable here: this cap exists to +# bound sustained churn, and the concurrent caps handle the instantaneous side. _RATE_WINDOW_SECONDS = 60.0 # Opportunistic prune cadence for the rate-tracking dict: walk and drop # expired entries every Nth acquire. Bounds memory under PROXY-protocol with diff --git a/src/mta-in/src/pymta/mda_async.py b/src/mta-in/src/pymta/mda_async.py index 22b1ba4a..59fbfd4e 100644 --- a/src/mta-in/src/pymta/mda_async.py +++ b/src/mta-in/src/pymta/mda_async.py @@ -4,8 +4,8 @@ The Postfix milter uses ``requests`` (sync, see ``src/api/mda.py``). pymta runs inside an asyncio event loop, so blocking HTTP calls would freeze the whole SMTP server; we mirror the same JWT contract here on top of httpx. -The MDA contract — kept identical to the milter so both implementations stay -swap-compatible — is: +The MDA contract, kept identical to the milter so both implementations stay +swap-compatible, is: * ``POST /inbound/mta/check/`` with ``application/json`` body ``{"addresses": [...]}`` → returns ``{addr: bool}``. @@ -41,6 +41,31 @@ _LOCAL_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "::1"}) # even start the process rather than minting weak tokens. _MIN_SECRET_LENGTH = 32 +# HTTP statuses on which a message is rejected *permanently* (SMTP 5xx). The +# list is an allow-list, not a "4xx means permanent" rule, because the default +# has to be the safe one: deferring costs a retry, bouncing loses the mail. +# +# 400: the MDA could not parse the message, or the request was malformed. +# 413: over MAX_INCOMING_EMAIL_SIZE. Retrying sends the same oversized bytes. +# 415: wrong Content-Type. Ours to fix, but no retry will change it. +# +# Everything else defers: +# 207: Multi-Status. *Some* recipients were delivered, some were not. The +# MDA has no per-recipient reply channel back to us, so the only way +# the failed ones ever land is if the sending MTA retries the whole +# envelope. That re-delivers to the recipients who already succeeded, +# which is the correct trade against silently losing the rest. +# 401 / 403: secret rotation skew, or an `exp` the MDA's clock reads as +# past. A routine operational event must not bounce real mail. +# 404: a routing/deployment mistake, not a verdict on this message. +# 429: throttling. Retrying later is the intended response. +# +# Delivery only. Each entry is a verdict on the *message*, and a recipient +# check carries no message: a 400/413/415 there is a fault in the check request +# we built, so it defers like any other unexpected status rather than telling +# the sender the mailbox is permanently bad. +_PERMANENT_STATUSES = frozenset({400, 413, 415}) + @dataclass(frozen=True) class MDAResult: @@ -48,8 +73,16 @@ class MDAResult: ``ok`` is true iff the call returned HTTP 200 with a JSON body that the caller can rely on. ``temp_fail`` distinguishes "try again later" (network - error / 5xx / timeout) from a permanent rejection. ``payload`` is the - decoded JSON body when available. + error, timeout, 5xx, and every status the endpoint does not name as + permanent) from a permanent rejection; only ``deliver`` names any, via + :data:`_PERMANENT_STATUSES`. ``payload`` is the decoded JSON body when + available. + + ``payload`` is always a dict, so callers can ``.get()`` without guarding. + A body that is absent, unparseable, or a JSON value that is not an object + (a list, a string, ``null``) becomes ``{}``. An empty payload therefore + means "no usable answer", never a verdict: callers must not read a missing + key as a negative one. """ ok: bool @@ -67,18 +100,20 @@ class MDAClient: blocks on a synchronous MDA call so there is no on-disk queue. """ - def __init__( + def __init__( # noqa: PLR0913 self, base_url: str | None = None, secret: str | None = None, timeout: int | None = None, breaker_threshold: int | None = None, breaker_cooldown: int | None = None, + jwt_ttl: int | None = None, clock=time.monotonic, ): self.base_url = (base_url or settings.MDA_API_BASE_URL).rstrip("/") + "/" self.secret = secret or settings.MDA_API_SECRET self.timeout = timeout if timeout is not None else settings.MDA_API_TIMEOUT + self.jwt_ttl = jwt_ttl if jwt_ttl is not None else settings.MDA_API_JWT_TTL self._breaker_threshold = ( breaker_threshold if breaker_threshold is not None @@ -99,13 +134,13 @@ class MDAClient: self._validate_credentials() def _validate_credentials(self) -> None: - """Warn loudly at startup about weak secret or plaintext non-local MDA URL. + """Warn loudly at startup about a weak secret or a plaintext non-local MDA URL. - Warnings rather than hard failures because the shared dev secret - ``my-shared-secret-mda`` (20 chars) is intentionally short, and dev - deployments talk to the MDA over the docker bridge without TLS. The - log line gives a prod operator clear feedback to fix; promote to - ``RuntimeError`` here once prod has migrated to a stronger secret. + Warnings, not hard failures: the shared dev secret + ``my-shared-secret-mda`` (20 chars) is intentionally short and dev + deployments talk to the MDA over the docker bridge without TLS, so + refusing to start would block the normal local workflow. See the + production checklist in the README for what these should look like. """ parsed = urlparse(self.base_url) host = (parsed.hostname or "").lower() @@ -117,7 +152,12 @@ class MDAClient: host, self.base_url, ) - if self.secret and len(self.secret) < _MIN_SECRET_LENGTH: + if not self.secret: + logger.warning( + "MDA_API_SECRET is empty; every MDA call will fail to sign and " + "all mail will be deferred. Configure the shared secret." + ) + elif len(self.secret) < _MIN_SECRET_LENGTH: logger.warning( "MDA_API_SECRET is %d bytes; recommended minimum is %d. " "Short HS256 secrets are brute-forceable from a captured JWT.", @@ -144,7 +184,8 @@ class MDAClient: # "body_hash" cannot shadow the security-relevant claims. claims = { **metadata, - "exp": datetime.datetime.now(tz=datetime.UTC) + datetime.timedelta(seconds=60), + "exp": datetime.datetime.now(tz=datetime.UTC) + + datetime.timedelta(seconds=self.jwt_ttl), "body_hash": hashlib.sha256(body).hexdigest(), } return jwt.encode(claims, self.secret, algorithm="HS256") @@ -167,26 +208,24 @@ class MDAClient: if self._consecutive_failures >= self._breaker_threshold and self._open_until is None: self._open_until = self._clock() + self._breaker_cooldown logger.warning( - "MDA circuit breaker OPEN after %d consecutive failures; " - "fast-failing for %ds", + "MDA circuit breaker OPEN after %d consecutive failures; fast-failing for %ds", self._consecutive_failures, self._breaker_cooldown, ) def _record_success(self) -> None: if self._consecutive_failures and self._open_until is None: - logger.info( - "MDA recovered after %d consecutive failures", self._consecutive_failures - ) + logger.info("MDA recovered after %d consecutive failures", self._consecutive_failures) self._consecutive_failures = 0 - async def _post( + async def _post( # noqa: PLR0913 self, path: str, content_type: str, body: bytes, metadata: dict, endpoint_label: str, + permanent_statuses: frozenset[int] = frozenset(), ) -> MDAResult: if self._breaker_open(): metrics.MDA_REQUEST_DURATION.labels( @@ -220,11 +259,21 @@ class MDAClient: elapsed = self._clock() - start - # JSON decode is best-effort; some error bodies may be HTML. + # JSON decode is best-effort; some error bodies may be HTML. Anything + # that is not a JSON object collapses to {} so callers get a total + # ``.get()`` instead of an AttributeError on a list or None body. try: payload = response.json() if response.content else {} except json.JSONDecodeError: payload = {} + if not isinstance(payload, dict): + logger.warning( + "MDA %s returned HTTP %d with a non-object JSON body (%s)", + endpoint_label, + response.status_code, + type(payload).__name__, + ) + payload = {} status = response.status_code if status == 200: @@ -234,15 +283,30 @@ class MDAClient: self._record_success() return MDAResult(ok=True, temp_fail=False, payload=payload, status_code=status) - # 5xx → tempfail (counted as a breaker failure); 4xx → permanent reject - # (not counted — it's the MDA telling us the *request* was bad). - temp = status >= 500 - result_label = "http_5xx" if temp else "http_4xx" + # Two independent judgements here, kept separate: + # + # * temp_fail: what we tell the *sender*. Permanent only for the + # statuses that say "this message is bad"; everything else defers. + # * the circuit breaker: a *liveness* signal. Only 5xx (and the + # transport failures above) indicate the MDA is unhealthy. A 207 or + # a 401 is a complete answer from a healthy MDA, so it closes the + # breaker rather than opening it. + temp = status not in permanent_statuses + unhealthy = status >= 500 + if unhealthy: + result_label = "http_5xx" + else: + result_label = "http_defer" if temp else "http_perm" metrics.MDA_REQUEST_DURATION.labels(endpoint=endpoint_label, result=result_label).observe( elapsed ) - logger.warning("MDA %s returned HTTP %d", endpoint_label, status) - if temp: + logger.warning( + "MDA %s returned HTTP %d (%s)", + endpoint_label, + status, + "deferring" if temp else "rejecting permanently", + ) + if unhealthy: self._record_failure() else: self._record_success() @@ -286,4 +350,5 @@ class MDAClient: message, metadata=metadata, endpoint_label="deliver", + permanent_statuses=_PERMANENT_STATUSES, ) diff --git a/src/mta-in/src/pymta/metrics.py b/src/mta-in/src/pymta/metrics.py index 65398318..eda9e1e2 100644 --- a/src/mta-in/src/pymta/metrics.py +++ b/src/mta-in/src/pymta/metrics.py @@ -18,7 +18,8 @@ _METRICS_NAMESPACE = "pymta" CONNECTIONS_TOTAL = Counter( f"{_METRICS_NAMESPACE}_connections_total", "Total inbound TCP connections, by post-accept outcome.", - # accepted | rejected_per_ip | rejected_per_ip_rate | rejected_global | proxy_error + # accepted | rejected_per_ip | rejected_per_ip_rate | rejected_global | + # rejected_untrusted_proxy | proxy_error labelnames=("result",), ) @@ -68,7 +69,10 @@ MDA_REQUEST_DURATION = Histogram( labelnames=( "endpoint", "result", - ), # endpoint: check|deliver, result: ok|http_5xx|timeout|error + ), # endpoint: check|deliver + # result: ok | http_5xx | http_defer (non-5xx status we retry on) | + # http_perm (status that permanently rejects the message) | + # timeout | error | breaker_open buckets=(0.005, 0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10, 30), ) @@ -76,7 +80,7 @@ DISCONNECTS_421 = Counter( f"{_METRICS_NAMESPACE}_disconnects_421_total", "Sessions where pymta replied 421 and closed the TCP connection.", labelnames=("reason",), # gate_global | gate_per_ip | gate_per_ip_rate | - # hard_error_limit | internal_error + # hard_error_limit | max_rcpt_misses | max_session_seconds | internal_error ) @@ -87,7 +91,7 @@ SECURITY_REJECTIONS = Counter( # Known reasons: source_route, control_char, oversize_local, oversize_domain, # nul_byte, oversize_announced, max_recipients, max_envelopes, auth_offered, # bad_address, address_literal, bad_helo, hard_error_limit, max_rcpt_misses, - # internal_error + # untrusted_proxy, max_session_seconds, internal_error ) diff --git a/src/mta-in/src/pymta/server.py b/src/mta-in/src/pymta/server.py index 509d6bf1..9fd06e45 100644 --- a/src/mta-in/src/pymta/server.py +++ b/src/mta-in/src/pymta/server.py @@ -33,7 +33,56 @@ def _configure_logging() -> None: ) +def _check_proxy_trust_config() -> None: + """Warn loudly when the PROXY-protocol listener will believe any peer. + + Enabling PROXY protocol is normally the assertion that a balancer sits in + front of us, which makes the balancer's address a known fact. Naming it in + ``PYMTA_TRUSTED_PROXIES`` is strongly recommended and this warns when it is + absent, but it does not block startup: deployments where the balancer's + addresses are dynamic or simply unknown at boot still need to run, and + there the network isolation has to carry the whole weight instead. + + What the allowlist buys is that the header only decides the per-IP + rate-limit key and the ``client_address`` the MDA writes into ``Received`` + when a known balancer sent it. Without one, any peer that can open a TCP + connection to the SMTP port decides both. + + There are two supported topologies: PROXY protocol on, behind a balancer; + or PROXY protocol off, exposed directly. A balancer without PROXY protocol + is not supported, because pymta would attribute every session to the + balancer's own IP. + """ + if not settings.PYMTA_ENABLE_PROXY_PROTOCOL: + return + # A zero-prefix network (0.0.0.0/0, ::/0) matches every peer, so it is the + # empty allowlist wearing a disguise. Same posture, same warning. + catch_all = [net for net in settings.PYMTA_TRUSTED_PROXIES if net.prefixlen == 0] + if not settings.PYMTA_TRUSTED_PROXIES or catch_all: + why = ( + "PYMTA_TRUSTED_PROXIES is empty" + if not settings.PYMTA_TRUSTED_PROXIES + else f"PYMTA_TRUSTED_PROXIES contains {', '.join(str(n) for n in catch_all)}, " + "which matches every peer" + ) + logger.warning( + "SECURITY: PROXY protocol is enabled but %s, so a PROXY header is trusted " + "from any peer. Any host able to reach port %s directly can forge its " + "source IP past the per-IP caps and into the Received header. Set it to " + "the load balancer's IPs/CIDRs, and make sure the port is reachable only " + "from the balancer.", + why, + settings.PYMTA_SMTP_PORT, + ) + return + logger.info( + "PROXY protocol enabled; trusting headers only from %s", + ", ".join(str(net) for net in settings.PYMTA_TRUSTED_PROXIES), + ) + + async def _serve() -> None: + _check_proxy_trust_config() mda_client = MDAClient() try: await mda_client.start() @@ -72,7 +121,7 @@ async def _serve() -> None: try: loop.add_signal_handler(sig, stop.set) except NotImplementedError: - # Windows / restricted environments — no signal handler support. + # Windows / restricted environments: no signal handler support. pass try: @@ -86,8 +135,7 @@ async def _serve() -> None: ) except TimeoutError: logger.warning( - "graceful shutdown deadline (%ds) exceeded; in-flight " - "sessions abandoned", + "graceful shutdown deadline (%ds) exceeded; in-flight sessions abandoned", settings.PYMTA_SHUTDOWN_TIMEOUT, ) finally: diff --git a/src/mta-in/src/pymta/settings.py b/src/mta-in/src/pymta/settings.py index 62187a29..8abde90d 100644 --- a/src/mta-in/src/pymta/settings.py +++ b/src/mta-in/src/pymta/settings.py @@ -1,25 +1,52 @@ """Environment-variable-driven settings for the pymta server.""" +import ipaddress import os def _env_bool(name: str, default: bool) -> bool: + """Read a boolean setting, refusing spellings the code cannot honour. + + An unset or blank variable takes the default. Anything else must be a + recognised spelling: silently reading a typo as its opposite is how a + security toggle ends up off in production. + """ raw = os.environ.get(name, "").strip().lower() + if not raw: + return default if raw in ("1", "true", "yes", "on"): return True if raw in ("0", "false", "no", "off"): return False - return default + raise ValueError( + f"{name} is set to {raw!r}, which is not a recognised boolean. Use one of " + "1/true/yes/on or 0/false/no/off." + ) -def _env_int(name: str, default: int) -> int: +def _env_int(name: str, default: int, *, minimum: int) -> int: + """Read an integer setting, refusing values the code cannot honour. + + ``minimum`` is part of each setting's contract and is written out at every + call site: ``minimum=0`` marks the settings where zero means "disabled", + ``minimum=1`` the ones where zero is nonsense. Without the check those two + groups look identical from the env, and the nonsense values fail in ways + that are silent and wrong rather than loud. ``PYMTA_MAX_RECIPIENTS=0`` + would 452 every recipient, ``MAX_INCOMING_EMAIL_SIZE=0`` would disable + aiosmtpd's size cap while the handler rejected every message, and + ``PYMTA_DATA_TIMEOUT=0`` would quietly inherit the command timeout because + aiosmtpd reads a falsy duration as "unset". + """ raw = os.environ.get(name, "").strip() if not raw: return default try: - return int(raw) + value = int(raw) except ValueError as exc: raise ValueError(f"Environment variable {name} must be an integer, got {raw!r}") from exc + if value < minimum: + raise ValueError(f"Environment variable {name} must be >= {minimum}, got {value}") + return value def _env_str(name: str, default: str) -> str: @@ -29,20 +56,43 @@ def _env_str(name: str, default: str) -> str: return raw +def _env_token(name: str, default: str) -> str: + """Read a setting that gets interpolated into an SMTP reply line. + + The banner is ``220 {hostname} {ident}``, so a CR/LF/NUL here would append + extra lines to our greeting. These values come from the operator, not the + peer, so this is not a live vector; ``PYMTA_HOSTNAME`` does fall back to + the shared ``MYHOSTNAME``, which on k8s can come from the downward API. + """ + value = _env_str(name, default) + if any(c in value for c in ("\r", "\n", "\x00", "\t")): + raise ValueError(f"Environment variable {name} must not contain control characters") + return value + + # --------------------------------------------------------------------------- # MDA back-end (shared with the Postfix milter) # --------------------------------------------------------------------------- MDA_API_BASE_URL = _env_str("MDA_API_BASE_URL", "http://localhost:8000/api/v1.0/") MDA_API_SECRET = _env_str("MDA_API_SECRET", "") -MDA_API_TIMEOUT = _env_int("MDA_API_TIMEOUT", 30) +MDA_API_TIMEOUT = _env_int("MDA_API_TIMEOUT", 30, minimum=1) + +# Lifetime of the HS256 token signed for each MDA call. It must comfortably +# cover the whole request (bounded by MDA_API_TIMEOUT) *plus* whatever clock +# skew exists between this process and the MDA. An `exp` in the MDA's past is +# a 401, which under the classification below defers the mail instead of +# bouncing it, but still stalls delivery until the clocks agree. The body_hash +# claim keeps a captured token usable only for its exact request, so the extra +# margin costs nothing. +MDA_API_JWT_TTL = _env_int("MDA_API_JWT_TTL", MDA_API_TIMEOUT + 90, minimum=1) # Circuit-breaker: when this many consecutive MDA calls fail (timeout / 5xx / # transport error), pymta short-circuits subsequent calls for # ``PYMTA_MDA_BREAKER_COOLDOWN`` seconds and replies 451 directly. Prevents # SMTP sessions from stacking up against a dead MDA. Set to 0 to disable. -PYMTA_MDA_BREAKER_THRESHOLD = _env_int("PYMTA_MDA_BREAKER_THRESHOLD", 10) -PYMTA_MDA_BREAKER_COOLDOWN = _env_int("PYMTA_MDA_BREAKER_COOLDOWN", 30) +PYMTA_MDA_BREAKER_THRESHOLD = _env_int("PYMTA_MDA_BREAKER_THRESHOLD", 10, minimum=0) +PYMTA_MDA_BREAKER_COOLDOWN = _env_int("PYMTA_MDA_BREAKER_COOLDOWN", 30, minimum=0) # --------------------------------------------------------------------------- @@ -50,14 +100,14 @@ PYMTA_MDA_BREAKER_COOLDOWN = _env_int("PYMTA_MDA_BREAKER_COOLDOWN", 30) # --------------------------------------------------------------------------- PYMTA_SMTP_HOST = _env_str("PYMTA_SMTP_HOST", "0.0.0.0") # noqa: S104 -PYMTA_SMTP_PORT = _env_int("PYMTA_SMTP_PORT", 25) +PYMTA_SMTP_PORT = _env_int("PYMTA_SMTP_PORT", 25, minimum=1) # Banner / Received-header hostname. Matches Postfix's `myhostname`. -PYMTA_HOSTNAME = _env_str("PYMTA_HOSTNAME", _env_str("MYHOSTNAME", "mta-in")) +PYMTA_HOSTNAME = _env_token("PYMTA_HOSTNAME", _env_str("MYHOSTNAME", "mta-in")) # ESMTP banner ident (after the hostname). Kept short and version-less so we # don't broadcast "aiosmtpd X.Y.Z" to internet scanners. -PYMTA_IDENT = _env_str("PYMTA_IDENT", "ESMTP") +PYMTA_IDENT = _env_token("PYMTA_IDENT", "ESMTP") # --------------------------------------------------------------------------- @@ -65,17 +115,27 @@ PYMTA_IDENT = _env_str("PYMTA_IDENT", "ESMTP") # --------------------------------------------------------------------------- # Total RFC822 message size cap. Mirrors Postfix `message_size_limit`. -MAX_INCOMING_EMAIL_SIZE = _env_int("MAX_INCOMING_EMAIL_SIZE", 10_240_000) +MAX_INCOMING_EMAIL_SIZE = _env_int("MAX_INCOMING_EMAIL_SIZE", 10_240_000, minimum=1) # RCPT TO per SMTP transaction. Mirrors Postfix `smtpd_recipient_limit=100`. -PYMTA_MAX_RECIPIENTS = _env_int("PYMTA_MAX_RECIPIENTS", 100) +# +# Note the multiplication downstream: controller.py derives the per-verb +# ``command_call_limit`` for RCPT as (this x MAX_ENVELOPES_PER_CONNECTION) + 10, +# which on the defaults is 1010 commands before aiosmtpd force-closes. Raising +# either of these raises that ceiling as their product, and it is the ceiling +# on how many commands one connection can spend re-arming the idle timer. +PYMTA_MAX_RECIPIENTS = _env_int("PYMTA_MAX_RECIPIENTS", 100, minimum=1) # Envelopes per TCP connection (one envelope = MAIL FROM..DATA cycle). -PYMTA_MAX_ENVELOPES_PER_CONNECTION = _env_int("PYMTA_MAX_ENVELOPES_PER_CONNECTION", 10) +PYMTA_MAX_ENVELOPES_PER_CONNECTION = _env_int("PYMTA_MAX_ENVELOPES_PER_CONNECTION", 10, minimum=1) # RFC 5321 §4.5.3.1.1/.1.2: local-part ≤ 64 octets, domain ≤ 255 octets. -PYMTA_MAX_LOCAL_PART = _env_int("PYMTA_MAX_LOCAL_PART", 64) -PYMTA_MAX_DOMAIN = _env_int("PYMTA_MAX_DOMAIN", 255) +# Constants, not env vars: these are the protocol's numbers, not a deployment +# choice, and raising them would only widen the gap between what pymta accepts +# at RCPT and what the MDA can store. ``validate_envelope_address`` still takes +# them as arguments so the tests can probe the boundaries directly. +PYMTA_MAX_LOCAL_PART = 64 +PYMTA_MAX_DOMAIN = 255 # --------------------------------------------------------------------------- @@ -83,41 +143,87 @@ PYMTA_MAX_DOMAIN = _env_int("PYMTA_MAX_DOMAIN", 255) # --------------------------------------------------------------------------- # Per-command idle timeout (seconds). Postfix default is 300 s; we tighten. -PYMTA_COMMAND_TIMEOUT = _env_int("PYMTA_COMMAND_TIMEOUT", 120) +# +# aiosmtpd arms this deadline once per accepted command and only re-arms it +# when the *next* complete command line arrives. It is not re-armed while a +# command handler runs. So this is the ceiling on "peer connected / last +# command finished, nothing since". +PYMTA_COMMAND_TIMEOUT = _env_int("PYMTA_COMMAND_TIMEOUT", 120, minimum=1) -# Total deadline for the DATA phase (seconds), wrapping the bytes-receive loop -# plus the MDA delivery call. Defends against slowloris on the body. -PYMTA_DATA_TIMEOUT = _env_int("PYMTA_DATA_TIMEOUT", 600) +# Hard deadline for the DATA phase (seconds): 354 reply → last body byte → +# MDA deliver call → SMTP reply, as one budget. ``HardenedSMTP.smtp_DATA`` +# swaps the command deadline for this one while DATA runs, so a peer that +# dribbles the body cannot outlive it. Defends against slowloris on the body. +# Nothing in a DATA phase survives past it: the transport is armed at exactly +# this value, and the handler reserves a slice of it (see +# ``handler._REPLY_RESERVE_SECONDS``) so it can still answer 451 rather than +# vanishing mid-transaction. +# +# Sizing: this is a *floor* on how slow a legitimate sender may be. At the +# 10 MB default MAX_INCOMING_EMAIL_SIZE, 300 s means ~34 kB/s sustained. +# +# The slice the handler holds back, read by ``handler._REPLY_RESERVE_SECONDS``. +# No PYMTA_ prefix: that marks the env-backed settings, and this one is not +# configurable. It is a property of how the two deadlines nest, not something +# an operator sizes. It bounds the minimum below, because a DATA timeout at or +# under the reserve leaves the deliver call no budget at all and would defer +# every message. +REPLY_RESERVE_SECONDS = 10 +PYMTA_DATA_TIMEOUT = _env_int("PYMTA_DATA_TIMEOUT", 300, minimum=REPLY_RESERVE_SECONDS + 1) + +# Wall-clock ceiling on one TCP session (seconds), armed at connect and never +# re-armed. 0 disables. +# +# This is the only bound a peer cannot push back by staying busy: every other +# timeout is reset by activity, so a peer issuing one command just under +# PYMTA_COMMAND_TIMEOUT can hold a slot out of PYMTA_MAX_SESSIONS_TOTAL for as +# long as ``command_call_limit`` lets it keep issuing commands. +# +# It does not stop a distributed attacker, who can reconnect. It caps how long +# one connection can hold a slot, so blocking an abuser frees the slots instead +# of leaving them held until the process restarts. +# +# Sizing: 100 recipients plus a 10 MB body is under a minute for a real sender. +PYMTA_MAX_SESSION_SECONDS = _env_int("PYMTA_MAX_SESSION_SECONDS", 1800, minimum=0) # Maximum wall-clock seconds the server waits for in-flight sessions to drain -# after SIGTERM. Lower than k8s `terminationGracePeriodSeconds` so we exit -# cleanly before SIGKILL would interrupt an in-progress MDA deliver call. -PYMTA_SHUTDOWN_TIMEOUT = _env_int("PYMTA_SHUTDOWN_TIMEOUT", 25) +# after SIGTERM, before abandoning them. Sized to sit under k8s +# `terminationGracePeriodSeconds` so we choose the cut-off rather than having +# SIGKILL choose it. +# +# It sits far below PYMTA_DATA_TIMEOUT and PYMTA_MAX_SESSION_SECONDS, so a +# rollout does cut sessions mid-transaction rather than waiting out a slow DATA +# phase. The failure is one-directional: a peer cut off before our 250 retries, +# which risks a duplicate on an already-delivered message, not a loss. Do not +# raise this above `terminationGracePeriodSeconds`. +PYMTA_SHUTDOWN_TIMEOUT = _env_int("PYMTA_SHUTDOWN_TIMEOUT", 25, minimum=0) # Per-IP concurrent SMTP sessions. 0 disables the cap. -PYMTA_MAX_SESSIONS_PER_IP = _env_int("PYMTA_MAX_SESSIONS_PER_IP", 100) +PYMTA_MAX_SESSIONS_PER_IP = _env_int("PYMTA_MAX_SESSIONS_PER_IP", 100, minimum=0) # Process-wide concurrent SMTP sessions. 0 disables. -PYMTA_MAX_SESSIONS_TOTAL = _env_int("PYMTA_MAX_SESSIONS_TOTAL", 1000) +PYMTA_MAX_SESSIONS_TOTAL = _env_int("PYMTA_MAX_SESSIONS_TOTAL", 1000, minimum=0) -# Per-IP new-session rate, measured in a rolling 60s window. Defends against a +# Per-IP new-session rate, measured in a fixed 60s window. Defends against a # peer that churns through fast open/close cycles (which never exceed the # concurrent cap but still cost CPU/TLS handshakes/MDA RCPT checks). 0 disables. -PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE = _env_int("PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE", 600) +PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE = _env_int( + "PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE", 600, minimum=0 +) # Per-session soft-error budget. Mirrors Postfix `smtpd_hard_error_limit`: # once a session accumulates this many 4xx/5xx replies (typically over-limit # or unknown-recipient RCPTs), the next misbehaviour gets a 421 and the # connection closes. Defends against bulk address enumeration that lives in # one TCP session. -PYMTA_HARD_ERROR_LIMIT = _env_int("PYMTA_HARD_ERROR_LIMIT", 50) +PYMTA_HARD_ERROR_LIMIT = _env_int("PYMTA_HARD_ERROR_LIMIT", 50, minimum=1) # Per-session cap on unknown-mailbox lookups specifically. The hard-error # budget above covers the *aggregate* of all 4xx/5xx replies; this one # isolates enumeration: an attacker submitting valid-syntax addresses to # probe which exist gets cut off after this many ``no such recipient`` # replies, even if the soft-error counter is still below its limit. -PYMTA_MAX_RCPT_MISSES_PER_SESSION = _env_int("PYMTA_MAX_RCPT_MISSES_PER_SESSION", 10) +PYMTA_MAX_RCPT_MISSES_PER_SESSION = _env_int("PYMTA_MAX_RCPT_MISSES_PER_SESSION", 10, minimum=1) # --------------------------------------------------------------------------- @@ -126,12 +232,44 @@ PYMTA_MAX_RCPT_MISSES_PER_SESSION = _env_int("PYMTA_MAX_RCPT_MISSES_PER_SESSION" PYMTA_ENABLE_SMTPUTF8 = _env_bool("PYMTA_ENABLE_SMTPUTF8", True) -# PROXY protocol v1/v2 (HAProxy in front). Mirrors the Postfix -# ENABLE_PROXY_PROTOCOL=haproxy env knob. -PYMTA_ENABLE_PROXY_PROTOCOL = _env_str( - "ENABLE_PROXY_PROTOCOL", "" -).lower() == "haproxy" or _env_bool("PYMTA_ENABLE_PROXY_PROTOCOL", False) -PYMTA_PROXY_PROTOCOL_TIMEOUT = _env_int("PYMTA_PROXY_PROTOCOL_TIMEOUT", 5) +# PROXY protocol v1/v2 (HAProxy in front). The Postfix image toggles the same +# feature with its own ENABLE_PROXY_PROTOCOL=haproxy (see entrypoint.sh); pymta +# does not read that name, so the two services can share an env file without +# either one inheriting the other's switch. +PYMTA_ENABLE_PROXY_PROTOCOL = _env_bool("PYMTA_ENABLE_PROXY_PROTOCOL", False) +PYMTA_PROXY_PROTOCOL_TIMEOUT = _env_int("PYMTA_PROXY_PROTOCOL_TIMEOUT", 5, minimum=1) + + +# Comma-separated IPs / CIDRs allowed to send a PROXY header, matched against +# the *wire* peer (the TCP source, i.e. the load balancer). Anything the header +# claims is only as trustworthy as the peer that sent it: the claimed source IP +# becomes the key for every per-IP cap and the ``client_address`` the MDA bakes +# into the Received header. A peer that reaches port 25 directly could +# otherwise scatter a forged source across the address space (defeating +# PYMTA_MAX_SESSIONS_PER_IP / _PER_MINUTE) and attribute its mail to any IP it +# likes. +# +# Strongly recommended whenever PROXY protocol is enabled, because enabling it +# *is* the claim that a balancer sits in front, so the balancer's address is +# usually a known fact. Left empty there is nothing to match on and every peer's +# header is trusted; ``server.py`` warns loudly at startup rather than refusing +# to run, since some deployments only learn the balancer's addresses later. +def _parse_networks(raw: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]: + networks = [] + for chunk in raw.split(","): + entry = chunk.strip() + if not entry: + continue + try: + networks.append(ipaddress.ip_network(entry, strict=False)) + except ValueError as exc: + raise ValueError( + f"PYMTA_TRUSTED_PROXIES entry {entry!r} is not an IP address or CIDR" + ) from exc + return networks + + +PYMTA_TRUSTED_PROXIES = _parse_networks(_env_str("PYMTA_TRUSTED_PROXIES", "")) # --------------------------------------------------------------------------- @@ -139,7 +277,7 @@ PYMTA_PROXY_PROTOCOL_TIMEOUT = _env_int("PYMTA_PROXY_PROTOCOL_TIMEOUT", 5) # # Two ways to configure STARTTLS: # * pymta-native: ``PYMTA_TLS_CERT_FILE`` + ``PYMTA_TLS_KEY_FILE`` (two paths). -# * Postfix-style: ``STARTTLS_CHAIN_FILES`` — a comma-separated list of PEM +# * Postfix-style: ``STARTTLS_CHAIN_FILES``, a comma-separated list of PEM # bundle files (each bundle contains a private key followed by the cert # chain). pymta reads the first bundle in the list and loads it via # ``SSLContext.load_cert_chain(certfile=path, keyfile=path)``: Python's @@ -163,9 +301,16 @@ if _chain_files and not PYMTA_TLS_CERT_FILE and not PYMTA_TLS_KEY_FILE: # Prometheus metrics HTTP endpoint # --------------------------------------------------------------------------- +# Binds all interfaces by default because the usual scrape paths (a k8s +# Prometheus hitting the pod IP, a compose port mapping) cannot reach a +# loopback-only listener. The exposition is low-cardinality (no +# addresses, no client IPs), so the exposure is operational recon (volumes, +# rejection reasons, breaker state) rather than message data. It must still be +# fenced off from the interface that serves port 25 with a NetworkPolicy or +# firewall rule; set the host to 127.0.0.1, or the port to 0, where it isn't. PYMTA_METRICS_HOST = _env_str("PYMTA_METRICS_HOST", "0.0.0.0") # noqa: S104 # Set to 0 to disable the metrics HTTP server. -PYMTA_METRICS_PORT = _env_int("PYMTA_METRICS_PORT", 9100) +PYMTA_METRICS_PORT = _env_int("PYMTA_METRICS_PORT", 9100, minimum=0) # --------------------------------------------------------------------------- diff --git a/src/mta-in/src/pymta/smtp_protocol.py b/src/mta-in/src/pymta/smtp_protocol.py index fd65477d..2153410b 100644 --- a/src/mta-in/src/pymta/smtp_protocol.py +++ b/src/mta-in/src/pymta/smtp_protocol.py @@ -3,12 +3,12 @@ aiosmtpd's defaults are reasonable but its surface area still includes a few verbs we never want exposed on a public, inbound-only port-25 endpoint: -* ``AUTH`` — never offered (no authenticator wired) but we still reply 502 to +* ``AUTH``: never offered (no authenticator wired) but we still reply 502 to reject any attempt explicitly, so a misconfiguration cannot quietly become a relay. -* ``VRFY`` — RFC 5321 §3.5 lets us respond with a canned 252; we do so +* ``VRFY``: RFC 5321 §3.5 lets us respond with a canned 252; we do so unconditionally to prevent address enumeration. -* ``EXPN`` — explicit 502. +* ``EXPN``: explicit 502. We also fold connection-admission control into :meth:`_handle_client`. The gate is checked exactly once per accepted TCP session. When PROXY-protocol is @@ -23,8 +23,9 @@ import logging import time from aiosmtpd.smtp import SMTP as BaseSMTP +from aiosmtpd.smtp import syntax -from . import metrics +from . import metrics, settings from .limits import IPGate, TooManyConnections logger = logging.getLogger(__name__) @@ -40,6 +41,15 @@ class HardenedSMTP(BaseSMTP): # release path runs at most once. self._gate_held_ip: str | None = None self._gate_started: float | None = None + # Set by request_disconnect(); consumed by push() once the reply that + # announced the disconnect is on the wire. + self._disconnect_after_reply = False + # Monotonic timestamp of the 354 that opened the current DATA phase, + # or None outside DATA. Read by the handler to size its own deadline. + self.data_phase_started: float | None = None + # One-shot timer for the whole-session deadline. Armed on connect, + # never re-armed. See _arm_session_deadline. + self._session_deadline_handle = None # ----------------------------------------------------------- verb lockdown async def smtp_VRFY(self, arg: str) -> None: @@ -58,15 +68,126 @@ class HardenedSMTP(BaseSMTP): # Default aiosmtpd HELP enumerates implemented verbs (mild info leak). await self.push("214 2.0.0 See https://www.rfc-editor.org/rfc/rfc5321") + # ------------------------------------------------------ session deadline + def connection_made(self, transport) -> None: + super().connection_made(transport) + self._arm_session_deadline() + + def connection_lost(self, error) -> None: + if self._session_deadline_handle is not None: + self._session_deadline_handle.cancel() + self._session_deadline_handle = None + super().connection_lost(error) + + def _arm_session_deadline(self) -> None: + """Start the one bound a peer cannot push back by staying busy. + + Every other deadline is reset by peer activity: aiosmtpd re-arms its + idle timer on each accepted command line, so a peer that sends one + command just under ``PYMTA_COMMAND_TIMEOUT`` keeps the session alive + for as long as ``command_call_limit`` lets it issue commands, which is + many hours on the default budgets. + + Armed once per connection. Not re-armed on the STARTTLS transport swap + either: aiosmtpd calls ``connection_made`` a second time there, and + resetting the deadline there would give the peer a second full budget. + """ + if self._session_deadline_handle is not None: + return + limit = settings.PYMTA_MAX_SESSION_SECONDS + if limit <= 0: + return + self._session_deadline_handle = self.loop.call_later(limit, self._session_expired) + + def _session_expired(self) -> None: + self._session_deadline_handle = None + peer = getattr(self.session, "peer", None) if self.session else None + logger.info( + "session from %r exceeded PYMTA_MAX_SESSION_SECONDS (%ds); closing", + peer, + settings.PYMTA_MAX_SESSION_SECONDS, + ) + metrics.SECURITY_REJECTIONS.labels(reason="max_session_seconds").inc() + metrics.DISCONNECTS_421.labels(reason="max_session_seconds").inc() + # Announce before hanging up so the sender defers and retries rather + # than reading a bare reset as a hard failure. Runs as a task because + # push() is async and we are in a timer callback; writing from here is + # safe even mid-DATA (StreamWriter.write only appends to a buffer) and + # the connection is ending either way. + # The task is not retained: nothing awaits it, and the transport close + # it performs is what ends the connection. + self.loop.create_task(self._close_with_notice()) + + async def _close_with_notice(self) -> None: + with contextlib.suppress(OSError, ConnectionError): + await self.push("421 4.4.2 Session too long, closing connection") + if self.transport is not None: + self.transport.close() + + # ------------------------------------------------------- forced disconnect + def request_disconnect(self) -> None: + """Close the connection once the reply now being sent has gone out. + + aiosmtpd pushes whatever a handler hook returns and then loops back for + the next command. A ``421 ... goodbye`` from ``handle_RCPT`` is only a + string to it, so without this the session stays open and an enumerator + keeps probing until the much coarser per-verb ``command_call_limit`` + closes it. Closing from :meth:`push` rather than here means the 421 is + on the wire before the FIN, so the peer gets a reason. + """ + self._disconnect_after_reply = True + + async def push(self, status) -> None: + await super().push(status) + if self._disconnect_after_reply: + self._disconnect_after_reply = False + if self.transport is not None: + self.transport.close() + + async def handle_exception(self, error: Exception) -> str: + # The handler answers 421 ("closing transmission channel"); make that + # true. aiosmtpd would otherwise push it and carry on with a session + # whose state we no longer trust. + status = await super().handle_exception(error) + self.request_disconnect() + return status + + # ------------------------------------------------------------ DATA budget + @syntax("DATA") + async def smtp_DATA(self, arg: str) -> None: + """Run the DATA phase under a single total deadline. + + aiosmtpd arms its idle timer when a command line is dispatched and does + not re-arm it while the handler runs, so the whole of DATA (body + receive *and* the MDA deliver call) is charged to one + ``PYMTA_COMMAND_TIMEOUT``. That conflates two very different budgets: + 120 s is right for "peer went quiet at the command prompt" and too + tight for a 10 MB body plus a slow MDA, which would be torn down + mid-handler with no reply at all. + + Swap in the DATA budget for the duration. The transport is armed at + exactly ``PYMTA_DATA_TIMEOUT``, so nothing in a DATA phase outlives it; + the handler takes its reply reserve out of the same budget so it can + answer 451 first. + """ + self.data_phase_started = time.monotonic() + self._reset_timeout(settings.PYMTA_DATA_TIMEOUT) + try: + await super().smtp_DATA(arg) + finally: + self.data_phase_started = None + if self.transport is not None: + self._reset_timeout() + # ------------------------------------------------------------ gate wiring async def _handle_client(self) -> None: """Wrap aiosmtpd's per-connection dialogue with admission control. Two paths: - * **No PROXY protocol** — the immediate TCP peer is the real client, + * **No PROXY protocol**: the immediate TCP peer is the real client, so we gate before the SMTP dialogue starts. - * **PROXY protocol enabled** — gate is deferred to + * **PROXY protocol enabled**: the gate is deferred to :meth:`acquire_gate_post_proxy`, called from the handler's ``handle_PROXY`` hook once the real client IP has been parsed off the PROXY header. @@ -90,7 +211,7 @@ class HardenedSMTP(BaseSMTP): return await self._acquire_gate(real_ip) async def _acquire_gate(self, ip: str) -> bool: - assert self._ip_gate is not None # noqa: S101 — narrowing only; checked above + assert self._ip_gate is not None # noqa: S101 (narrowing only; checked above) try: await self._ip_gate._try_acquire(ip) # noqa: SLF001 except TooManyConnections as exc: diff --git a/src/mta-in/tests/test_handler.py b/src/mta-in/tests/test_handler.py index d255ba58..303bec2d 100644 --- a/src/mta-in/tests/test_handler.py +++ b/src/mta-in/tests/test_handler.py @@ -2,43 +2,63 @@ These tests cover the *session-state* invariants of the handler (counter bumps, gate paths). They run the handler against fake session/envelope/MDA -stand-ins — no Docker stack, no real SMTP traffic. +stand-ins: no Docker stack, no real SMTP traffic. """ from __future__ import annotations +import time import types -from ipaddress import ip_address +from ipaddress import ip_address, ip_network import pytest from pymta import settings from pymta.handler import ( + _ENVELOPES_ATTR, + _PROXY_SRC_ATTR, _RCPT_MISSES_ATTR, + _REPLY_RESERVE_SECONDS, _SOFT_ERRORS_ATTR, - InboundHandler, NULL_SENDER_SENTINEL, + InboundHandler, + _remaining_data_budget, ) from pymta.mda_async import MDAResult class _FakeMDA: - """Stand-in for MDAClient — returns whatever the test wires up.""" + """Stand-in for MDAClient. Returns whatever the test wires up. - def __init__(self, check_result: MDAResult | None = None): - self.check_result = check_result or MDAResult( - ok=True, temp_fail=False, payload={}, status_code=200 + ``check_result`` pins one verbatim result for every address; leave it unset + and the fake synthesises the real MDA shape instead, ``{address: + check_exists}``. A miss has to be an explicit ``False``, because the + handler treats a body that does not name the address as no answer at all. + """ + + def __init__( + self, + check_result: MDAResult | None = None, + deliver_result: MDAResult | None = None, + check_exists: bool = False, + ): + self.check_result = check_result + self.check_exists = check_exists + self.deliver_result = deliver_result or MDAResult( + ok=True, temp_fail=False, payload={"status": "ok"}, status_code=200 ) self.deliver_kwargs: dict | None = None async def check_recipient(self, address: str) -> MDAResult: - return self.check_result + if self.check_result is not None: + return self.check_result + return MDAResult( + ok=True, temp_fail=False, payload={address: self.check_exists}, status_code=200 + ) async def deliver(self, **kwargs) -> MDAResult: self.deliver_kwargs = kwargs - return MDAResult( - ok=True, temp_fail=False, payload={"status": "ok"}, status_code=200 - ) + return self.deliver_result def _session(): @@ -55,6 +75,15 @@ def _handler(mda=None) -> InboundHandler: return InboundHandler(mda or _FakeMDA()) +async def _deliver_with(mda) -> str: + """Run one complete DATA phase against ``mda`` and return the SMTP reply.""" + session, envelope = _session(), _envelope() + envelope.mail_from = "sender@example.com" + envelope.rcpt_tos = ["a@example.com"] + envelope.content = b"Subject: hi\r\n\r\nbody\r\n" + return await _handler(mda).handle_DATA(None, session, envelope) + + # --------------------------------------------------------------------------- # MAIL SIZE= path bumps the soft-error counter on both rejection branches. # --------------------------------------------------------------------------- @@ -107,7 +136,7 @@ async def test_data_oversize_bumps_soft_errors(): @pytest.mark.asyncio async def test_data_max_envelopes_bumps_soft_errors(): session, envelope = _session(), _envelope() - setattr(session, "_pymta_envelopes", settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION) + setattr(session, _ENVELOPES_ATTR, settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION) reply = await _handler().handle_DATA(None, session, envelope) assert reply.startswith("451") assert getattr(session, _SOFT_ERRORS_ATTR) == 1 @@ -122,27 +151,120 @@ async def test_data_max_envelopes_bumps_soft_errors(): async def test_rcpt_miss_counter_triggers_421_at_limit(monkeypatch): # Tight limit so we don't have to do many round-trips. monkeypatch.setattr(settings, "PYMTA_MAX_RCPT_MISSES_PER_SESSION", 3) - mda = _FakeMDA( - check_result=MDAResult( - ok=True, temp_fail=False, payload={}, status_code=200 - ) # exists=False for every address - ) + mda = _FakeMDA(check_exists=False) handler, session, envelope = _handler(mda), _session(), _envelope() # First two misses get the normal 550. for i in range(2): - reply = await handler.handle_RCPT( - None, session, envelope, f"", [] - ) + reply = await handler.handle_RCPT(None, session, envelope, f"", []) assert reply.startswith("550"), reply # Third miss hits the per-session cap and forces 421. - reply = await handler.handle_RCPT( - None, session, envelope, "", [] - ) + reply = await handler.handle_RCPT(None, session, envelope, "", []) assert reply.startswith("421") assert getattr(session, _RCPT_MISSES_ATTR) == 3 +# --------------------------------------------------------------------------- +# A 200 that does not name the address is not a verdict. +# +# ``_post`` collapses an absent, unparseable, or non-object body to {}, so the +# handler cannot tell those apart from "the MDA said this mailbox is missing" +# unless it insists on the key. Reading a bodyless 200 (an ingress error page, +# a response shape that drifted) as a miss would answer 550 and have the +# sending MTA bounce mail to a working address. +# --------------------------------------------------------------------------- + + +UNUSABLE_CHECK_BODIES = [ + pytest.param({}, id="empty-body-or-unparseable"), + pytest.param({"detail": "ok"}, id="proxy-200-page"), + pytest.param({"other@example.com": True}, id="answers-a-different-address"), + pytest.param({"Rcpt@example.com": True}, id="case-drifted-key"), + pytest.param({"rcpt@example.com": None}, id="null-verdict"), + pytest.param({"rcpt@example.com": "false"}, id="stringified-verdict"), + pytest.param({"rcpt@example.com": 0}, id="numeric-verdict"), + pytest.param({"rcpt@example.com": {"exists": False}}, id="nested-verdict-shape"), + pytest.param({"rcpt@example.com": []}, id="empty-list-verdict"), +] + + +@pytest.mark.parametrize("payload", UNUSABLE_CHECK_BODIES) +@pytest.mark.asyncio +async def test_rcpt_defers_when_a_200_carries_no_verdict(payload): + mda = _FakeMDA( + check_result=MDAResult(ok=True, temp_fail=False, payload=payload, status_code=200) + ) + handler, session, envelope = _handler(mda), _session(), _envelope() + + reply = await handler.handle_RCPT(None, session, envelope, "", []) + + assert reply.startswith("451"), reply + assert envelope.rcpt_tos == [] + # Not a miss: an unusable body must not spend the unknown-recipient budget, + # or an MDA hiccup would hang up on a sender addressing real mailboxes. + assert getattr(session, _RCPT_MISSES_ATTR, 0) == 0 + + +@pytest.mark.parametrize( + "result", + [ + pytest.param(MDAResult(ok=False, temp_fail=True, payload={}, status_code=0), id="timeout"), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={}, status_code=0), id="breaker-open" + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={"detail": "no"}, status_code=400), + id="http-400", + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={"detail": "no"}, status_code=413), + id="http-413", + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={"detail": "no"}, status_code=415), + id="http-415", + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={}, status_code=401), id="http-401" + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={}, status_code=404), id="http-404" + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={}, status_code=429), id="http-429" + ), + pytest.param( + MDAResult(ok=False, temp_fail=True, payload={}, status_code=503), id="http-503" + ), + ], +) +@pytest.mark.asyncio +async def test_rcpt_defers_on_every_unsuccessful_check(result): + # No check failure is ever a verdict on the mailbox: 400/413/415 included, + # since there they describe the request we built, not the address. + mda = _FakeMDA(check_result=result) + handler, session, envelope = _handler(mda), _session(), _envelope() + + reply = await handler.handle_RCPT(None, session, envelope, "", []) + + assert reply.startswith("451"), reply + assert envelope.rcpt_tos == [] + assert getattr(session, _RCPT_MISSES_ATTR, 0) == 0 + + +@pytest.mark.asyncio +async def test_rcpt_explicit_false_is_still_a_permanent_miss(): + # The guard above must not soften a real answer: an explicit False is the + # MDA saying the mailbox does not exist, and that stays a 550. + mda = _FakeMDA(check_exists=False) + handler, session, envelope = _handler(mda), _session(), _envelope() + + reply = await handler.handle_RCPT(None, session, envelope, "", []) + + assert reply.startswith("550"), reply + assert getattr(session, _RCPT_MISSES_ATTR) == 1 + + @pytest.mark.asyncio async def test_rcpt_existence_does_not_increment_miss_counter(): mda = _FakeMDA( @@ -154,13 +276,48 @@ async def test_rcpt_existence_does_not_increment_miss_counter(): ) ) handler, session, envelope = _handler(mda), _session(), _envelope() - reply = await handler.handle_RCPT( - None, session, envelope, "", [] - ) + reply = await handler.handle_RCPT(None, session, envelope, "", []) assert reply.startswith("250") assert getattr(session, _RCPT_MISSES_ATTR, 0) == 0 +# --------------------------------------------------------------------------- +# The DATA budget is one deadline shared by receive and deliver. +# --------------------------------------------------------------------------- + + +def test_data_budget_without_a_start_timestamp_is_the_full_budget(): + assert _remaining_data_budget(types.SimpleNamespace()) == float(settings.PYMTA_DATA_TIMEOUT) + + +def test_data_budget_subtracts_time_already_spent(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_DATA_TIMEOUT", 120) + server = types.SimpleNamespace(data_phase_started=time.monotonic() - 20) + remaining = _remaining_data_budget(server) + assert 120 - 20 - _REPLY_RESERVE_SECONDS - 1 < remaining <= 120 - 20 - _REPLY_RESERVE_SECONDS + + +def test_data_budget_raises_once_spent(monkeypatch): + # A slow receive that ate the whole budget must not still be granted a + # deliver call: that would overrun the transport deadline the reserve is + # there to stay clear of, costing the peer its 451. + monkeypatch.setattr(settings, "PYMTA_DATA_TIMEOUT", 60) + server = types.SimpleNamespace(data_phase_started=time.monotonic() - 60) + with pytest.raises(TimeoutError): + _remaining_data_budget(server) + + +@pytest.mark.asyncio +async def test_data_replies_451_when_the_budget_is_already_spent(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_DATA_TIMEOUT", 60) + server = _FakeServer() + server.data_phase_started = time.monotonic() - 60 + session, envelope = _session(), _envelope() + envelope.content = b"From: a@example.com\r\n\r\nhi\r\n" + reply = await _handler().handle_DATA(server, session, envelope) + assert reply.startswith("451") + + # --------------------------------------------------------------------------- # Hard-error budget cutoff still fires from the existing gate. # --------------------------------------------------------------------------- @@ -171,12 +328,41 @@ async def test_hard_error_limit_blocks_further_rcpts(monkeypatch): monkeypatch.setattr(settings, "PYMTA_HARD_ERROR_LIMIT", 2) handler, session, envelope = _handler(), _session(), _envelope() setattr(session, _SOFT_ERRORS_ATTR, 2) - reply = await handler.handle_RCPT( - None, session, envelope, "", [] - ) + reply = await handler.handle_RCPT(None, session, envelope, "", []) assert reply.startswith("421") +@pytest.mark.asyncio +async def test_mail_malformed_sender_bumps_soft_errors(): + # Without this the hard-error budget is sidestepped by burning errors on + # MAIL FROM alone, which never counted towards it. + session, envelope = _session(), _envelope() + reply = await _handler().handle_MAIL(None, session, envelope, "", []) + assert reply.startswith("501") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +@pytest.mark.asyncio +async def test_hard_error_limit_blocks_further_mails(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_HARD_ERROR_LIMIT", 2) + handler, session, envelope = _handler(), _session(), _envelope() + setattr(session, _SOFT_ERRORS_ATTR, 2) + reply = await handler.handle_MAIL(None, session, envelope, "", []) + assert reply.startswith("421") + assert envelope.mail_from is None + + +@pytest.mark.asyncio +async def test_hard_error_limit_on_mail_requests_disconnect(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_HARD_ERROR_LIMIT", 1) + server = _FakeServer() + handler, session, envelope = _handler(), _session(), _envelope() + setattr(server, _SOFT_ERRORS_ATTR, 1) + reply = await handler.handle_MAIL(server, session, envelope, "", []) + assert reply.startswith("421") + assert server.disconnect_requested is True + + # --------------------------------------------------------------------------- # Null sender survives the round-trip via the sentinel. # --------------------------------------------------------------------------- @@ -204,34 +390,39 @@ async def test_null_sender_round_trip_via_sentinel(): class _FakeServer: """Per-connection SMTP protocol stand-in (survives the STARTTLS swap).""" + def __init__(self): + self.disconnect_requested = False + self.data_phase_started = None + async def acquire_gate_post_proxy(self, ip: str) -> bool: return True + def request_disconnect(self) -> None: + self.disconnect_requested = True + @pytest.mark.asyncio -async def test_proxy_source_survives_starttls_and_reaches_mda(): +async def test_proxy_source_survives_starttls_and_reaches_mda(monkeypatch): real_client = ip_address("203.0.113.9") - lb_peer = ("10.89.0.2", 43154) # HAProxy/podman gateway — NOT the client + lb_peer = ("10.89.0.2", 43154) # HAProxy/podman gateway, NOT the client + + # Mirror a real deployment: PROXY protocol on, balancer allowlisted. + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network("10.89.0.0/24")]) server = _FakeServer() mda = _FakeMDA() handler = _handler(mda) # 1. PROXY header parsed on the plaintext connection, before STARTTLS. - proxy_data = types.SimpleNamespace( - src_addr=real_client, src_port=52000, version=2, protocol=1 - ) - session_pre_tls = types.SimpleNamespace( - host_name=None, peer=lb_peer, proxy_data=proxy_data - ) + proxy_data = types.SimpleNamespace(src_addr=real_client, src_port=52000, version=2, protocol=1) + session_pre_tls = types.SimpleNamespace(host_name=None, peer=lb_peer, proxy_data=proxy_data) gate = await handler.handle_PROXY(server, session_pre_tls, _envelope(), proxy_data) assert gate is True # 2. STARTTLS rebuilds the session: proxy_data gone, peer is the LB again. # Same server instance carries over. - session_post_tls = types.SimpleNamespace( - host_name=None, peer=lb_peer, proxy_data=None - ) + session_post_tls = types.SimpleNamespace(host_name=None, peer=lb_peer, proxy_data=None) # 3. DATA delivers using the post-TLS session. envelope = _envelope() @@ -244,3 +435,289 @@ async def test_proxy_source_survives_starttls_and_reaches_mda(): assert mda.deliver_kwargs is not None assert mda.deliver_kwargs["client_address"] == "203.0.113.9" assert mda.deliver_kwargs["client_port"] == "52000" + + +# --------------------------------------------------------------------------- +# Partial delivery (HTTP 207) must defer, never bounce. +# +# The MDA answers 207 when it delivered to some recipients and failed on +# others. It has no per-recipient reply channel back to us, so a permanent +# 554 tells the sending MTA to bounce the whole envelope and the stragglers +# are lost for good. 451 costs a duplicate for the recipients already served. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_partial_delivery_defers_instead_of_bouncing(): + mda = _FakeMDA( + deliver_result=MDAResult( + ok=False, + temp_fail=True, + payload={"status": "partial_success", "delivered": 1, "failed": 1}, + status_code=207, + ) + ) + handler, session, envelope = _handler(mda), _session(), _envelope() + envelope.mail_from = "sender@example.com" + envelope.rcpt_tos = ["a@example.com", "b@example.com"] + envelope.content = b"Subject: hi\r\n\r\nbody\r\n" + + reply = await handler.handle_DATA(None, session, envelope) + assert reply.startswith("451"), reply + + +@pytest.mark.asyncio +async def test_http_200_without_status_ok_defers(): + # A 200 whose body we don't recognise is not proof of delivery; deferring + # keeps the message alive while someone works out what the MDA meant. + mda = _FakeMDA( + deliver_result=MDAResult( + ok=True, temp_fail=False, payload={"status": "queued"}, status_code=200 + ) + ) + handler, session, envelope = _handler(mda), _session(), _envelope() + envelope.mail_from = "sender@example.com" + envelope.rcpt_tos = ["a@example.com"] + envelope.content = b"Subject: hi\r\n\r\nbody\r\n" + + reply = await handler.handle_DATA(None, session, envelope) + assert reply.startswith("451"), reply + + +@pytest.mark.parametrize("status", [400, 413, 415]) +@pytest.mark.asyncio +async def test_message_rejecting_status_still_bounces(status): + # 400/413/415 keep their permanent reject: retrying sends the same bytes. + mda = _FakeMDA( + deliver_result=MDAResult( + ok=False, temp_fail=False, payload={"detail": "unparseable"}, status_code=status + ) + ) + reply = await _deliver_with(mda) + assert reply.startswith("554"), reply + + +@pytest.mark.parametrize( + "payload", + [ + pytest.param({}, id="empty-body-or-unparseable"), + pytest.param({"status": None}, id="null-status"), + pytest.param({"status": "OK"}, id="wrong-case-status"), + pytest.param({"detail": "accepted"}, id="no-status-key"), + pytest.param({"status": "partial_success"}, id="partial-on-a-200"), + ], +) +@pytest.mark.asyncio +async def test_deliver_200_without_an_unambiguous_ok_defers(payload): + # 200 is not proof of delivery; only the exact ``{"status": "ok"}`` is. + # Every other body keeps the message alive for a retry. + mda = _FakeMDA( + deliver_result=MDAResult(ok=True, temp_fail=False, payload=payload, status_code=200) + ) + reply = await _deliver_with(mda) + assert reply.startswith("451"), reply + + +@pytest.mark.parametrize( + "status", + [ + pytest.param(0, id="timeout-transport-or-breaker"), + pytest.param(207, id="multi-status"), + pytest.param(401, id="jwt-rejected"), + pytest.param(403, id="forbidden"), + pytest.param(404, id="route-missing"), + pytest.param(429, id="throttled"), + pytest.param(418, id="unrecognised-4xx"), + pytest.param(500, id="mda-error"), + pytest.param(503, id="mda-unavailable"), + ], +) +@pytest.mark.asyncio +async def test_deliver_defers_on_every_status_outside_the_permanent_set(status): + mda = _FakeMDA( + deliver_result=MDAResult(ok=False, temp_fail=True, payload={}, status_code=status) + ) + reply = await _deliver_with(mda) + assert reply.startswith("451"), reply + + +# --------------------------------------------------------------------------- +# A 421 must actually hang up. +# +# aiosmtpd pushes whatever a hook returns and loops back for the next command, +# so "goodbye" is only a promise until the handler asks for the disconnect. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_rcpt_miss_cutoff_requests_disconnect(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_MAX_RCPT_MISSES_PER_SESSION", 1) + server = _FakeServer() + handler, session, envelope = _handler(), _session(), _envelope() + + reply = await handler.handle_RCPT(server, session, envelope, "", []) + assert reply.startswith("421") + assert server.disconnect_requested is True + + +@pytest.mark.asyncio +async def test_hard_error_cutoff_requests_disconnect(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_HARD_ERROR_LIMIT", 2) + server = _FakeServer() + setattr(server, _SOFT_ERRORS_ATTR, 2) + handler, session, envelope = _handler(), _session(), _envelope() + + reply = await handler.handle_RCPT(server, session, envelope, "", []) + assert reply.startswith("421") + assert server.disconnect_requested is True + + +# --------------------------------------------------------------------------- +# Abuse counters are keyed to the TCP connection, not the session object. +# +# aiosmtpd rebuilds `session` from scratch on STARTTLS. Counters living there +# would hand a peer a free budget reset for the price of one STARTTLS. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_rcpt_miss_budget_survives_the_starttls_session_rebuild(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_MAX_RCPT_MISSES_PER_SESSION", 3) + server = _FakeServer() + handler, envelope = _handler(), _envelope() + + # Two misses before STARTTLS. + for i in range(2): + reply = await handler.handle_RCPT( + server, _session(), envelope, f"", [] + ) + assert reply.startswith("550"), reply + + # STARTTLS hands us a brand-new session object; the budget must not reset. + reply = await handler.handle_RCPT(server, _session(), envelope, "", []) + assert reply.startswith("421"), reply + assert getattr(server, _RCPT_MISSES_ATTR) == 3 + + +# --------------------------------------------------------------------------- +# PROXY-protocol trust boundary. +# +# aiosmtpd parses a PROXY header from whoever sends it. The claimed source +# becomes the per-IP rate-limit key AND the client_address the MDA bakes into +# Received, so an unfiltered header is a free pass past both. +# --------------------------------------------------------------------------- + + +def _proxy_data(src="203.0.113.9"): + return types.SimpleNamespace(src_addr=ip_address(src), src_port=52000, version=2, protocol=1) + + +@pytest.mark.asyncio +async def test_proxy_header_from_untrusted_peer_is_refused(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network("10.89.0.0/24")]) + session = types.SimpleNamespace(host_name=None, peer=("198.51.100.7", 5555), proxy_data=None) + accepted = await _handler().handle_PROXY(_FakeServer(), session, _envelope(), _proxy_data()) + assert accepted is False + + +@pytest.mark.asyncio +async def test_proxy_header_from_trusted_peer_is_accepted(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network("10.89.0.0/24")]) + server = _FakeServer() + session = types.SimpleNamespace(host_name=None, peer=("10.89.0.2", 5555), proxy_data=None) + accepted = await _handler().handle_PROXY(server, session, _envelope(), _proxy_data()) + assert accepted is True + + +@pytest.mark.parametrize("allowlist", [[], [ip_network("0.0.0.0/0")]]) +@pytest.mark.asyncio +async def test_empty_allowlist_trusts_every_peer(monkeypatch, allowlist): + # "No upstream named" means there is nothing left to filter on, so the + # header is taken from anyone -- the same posture 0.0.0.0/0 spells out. + # server.py warns loudly at startup; the isolation is the trust boundary. + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", allowlist) + server = _FakeServer() + session = types.SimpleNamespace(host_name=None, peer=("198.51.100.7", 5555), proxy_data=None) + accepted = await _handler().handle_PROXY(server, session, _envelope(), _proxy_data()) + assert accepted is True + # And the claimed source is what the caps and Received are keyed on. + assert getattr(server, _PROXY_SRC_ATTR) == ("203.0.113.9", 52000) + + +# --------------------------------------------------------------------------- +# With PROXY protocol on, the wire peer is the balancer, never the client. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_no_proxy_source_reports_no_client_address(monkeypatch): + # A PROXY v2 LOCAL command (health check) carries no source. Falling back + # to session.peer would stamp the balancer's own IP into Received as if it + # were the sender. + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + mda = _FakeMDA() + server = _FakeServer() + session = types.SimpleNamespace( + host_name="client.test", peer=("10.89.0.2", 43154), proxy_data=None + ) + envelope = _envelope() + envelope.mail_from = "sender@example.com" + envelope.rcpt_tos = ["rcpt@example.com"] + envelope.content = b"Subject: hi\r\n\r\nbody\r\n" + + reply = await _handler(mda).handle_DATA(server, session, envelope) + + assert reply.startswith("250"), reply + assert mda.deliver_kwargs["client_address"] is None + assert mda.deliver_kwargs["client_port"] is None + + +@pytest.mark.asyncio +async def test_wire_peer_is_the_client_when_proxy_protocol_is_off(): + mda = _FakeMDA() + session = _session() # peer=("203.0.113.5", 12345) + envelope = _envelope() + envelope.mail_from = "sender@example.com" + envelope.rcpt_tos = ["rcpt@example.com"] + envelope.content = b"Subject: hi\r\n\r\nbody\r\n" + + await _handler(mda).handle_DATA(_FakeServer(), session, envelope) + + assert mda.deliver_kwargs["client_address"] == "203.0.113.5" + assert mda.deliver_kwargs["client_port"] == "12345" + + +# --------------------------------------------------------------------------- +# EHLO response filtering must leave a well-formed terminator. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_ehlo_strips_auth_and_keeps_a_final_line(): + responses = [ + "250-mta.example.com", + "250-SIZE 10240000", + "250-8BITMIME", + "250-AUTH ", + "250 HELP", + ] + clean = await _handler().handle_EHLO( + None, _session(), _envelope(), "client.example.com", responses + ) + assert not any(line[4:].upper().startswith("AUTH") for line in clean) + assert clean[-1].startswith("250 ") + assert all(line.startswith("250-") for line in clean[:-1]) + + +@pytest.mark.asyncio +async def test_ehlo_remarks_terminator_when_the_last_line_is_stripped(): + # Latent today (aiosmtpd always appends "250 HELP" last) but a reply left + # ending on a "250-" continuation hangs clients forever. + clean = await _handler().handle_EHLO( + None, + _session(), + _envelope(), + "client.example.com", + ["250-mta.example.com", "250-SIZE 10240000", "250 PIPELINING"], + ) + assert clean == ["250-mta.example.com", "250 SIZE 10240000"] diff --git a/src/mta-in/tests/test_hardened_smtp.py b/src/mta-in/tests/test_hardened_smtp.py new file mode 100644 index 00000000..7c6f7765 --- /dev/null +++ b/src/mta-in/tests/test_hardened_smtp.py @@ -0,0 +1,224 @@ +"""Unit tests for :class:`pymta.smtp_protocol.HardenedSMTP`. + +These exercise the two places where we step outside aiosmtpd's own control +flow: the forced disconnect behind a 421, and the DATA-phase deadline. Both +are driven against a fake transport: no sockets, no Docker stack. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from pymta import settings +from pymta.smtp_protocol import HardenedSMTP + + +class _FakeTransport: + def __init__(self): + self.closed = False + + def close(self) -> None: + self.closed = True + + def get_extra_info(self, name, default=None): + return ("203.0.113.5", 12345) if name == "peername" else default + + +class _FakeWriter: + def __init__(self): + self.written = bytearray() + + def write(self, data: bytes) -> None: + self.written.extend(data) + + async def drain(self) -> None: + pass + + +class _NullHandler: + """Handler with no hooks, so aiosmtpd falls back to its own replies.""" + + +def _server() -> HardenedSMTP: + """A connected-looking protocol instance without a real transport. + + Stands in for what ``connection_made`` would have built, minus the + ``_handle_client`` task and the stream plumbing that needs a live socket. + """ + smtp = HardenedSMTP(_NullHandler(), hostname="mta.test", timeout=120) + smtp.transport = _FakeTransport() + smtp._writer = _FakeWriter() + smtp.session = smtp._create_session() + smtp.envelope = smtp._create_envelope() + smtp._arm_session_deadline() + return smtp + + +# --------------------------------------------------------------------------- +# The 421 that actually hangs up. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_disconnect_waits_for_the_reply_to_go_out(): + # Closing eagerly would race the 421 off the wire and leave the peer with + # a bare TCP reset instead of a reason. + smtp = _server() + smtp.request_disconnect() + assert smtp.transport.closed is False + + await smtp.push("421 4.7.0 Too many errors, goodbye") + + assert bytes(smtp._writer.written).endswith(b"goodbye\r\n") + assert smtp.transport.closed is True + + +@pytest.mark.asyncio +async def test_disconnect_is_not_sticky(): + smtp = _server() + await smtp.push("250 2.1.0 OK") + assert smtp.transport.closed is False + + +@pytest.mark.asyncio +async def test_handle_exception_closes_the_session(): + smtp = _server() + status = await smtp.handle_exception(RuntimeError("boom")) + await smtp.push(status) + assert smtp.transport.closed is True + + +# --------------------------------------------------------------------------- +# DATA phase runs on its own budget. +# +# aiosmtpd arms its idle timer when a command line is dispatched and never +# re-arms it while the handler runs, so without this override the whole of +# DATA (body receive plus the MDA call) is charged to one +# PYMTA_COMMAND_TIMEOUT and the transport is torn down mid-handler. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_data_phase_swaps_in_the_data_deadline(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_DATA_TIMEOUT", 300) + + smtp = _server() + armed: list[float] = [] + loop = asyncio.get_running_loop() + real_call_later = loop.call_later + + def _record(delay, callback, *args): + armed.append(delay) + handle = real_call_later(delay, callback, *args) + handle.cancel() # never actually fire during the test + return handle + + monkeypatch.setattr(loop, "call_later", _record) + + # No RCPT recorded, so aiosmtpd's smtp_DATA bails out at "503 need RCPT" + # immediately, which is enough to prove the deadline is swapped and restored. + await smtp.smtp_DATA("") + + # PYMTA_DATA_TIMEOUT is the hard edge, armed verbatim. The handler's + # reply reserve comes out of this budget, it does not extend it. + assert armed[0] == 300, armed + assert armed[-1] == 120, armed + assert smtp.data_phase_started is None + + +@pytest.mark.asyncio +async def test_data_phase_start_is_published_for_the_handler(monkeypatch): + seen: list[float | None] = [] + + class _Recorder: + async def handle_DATA(self, server, session, envelope): + seen.append(server.data_phase_started) + return "250 2.0.0 OK" + + smtp = HardenedSMTP(_Recorder(), hostname="mta.test", timeout=120) + smtp.transport = _FakeTransport() + smtp._writer = _FakeWriter() + smtp.session = smtp._create_session() + smtp.envelope = smtp._create_envelope() + smtp.session.host_name = "client.test" + smtp.envelope.rcpt_tos.append("a@example.com") + + # Feed a one-line body followed by the end-of-data dot. + reader = asyncio.StreamReader() + reader.feed_data(b"Subject: x\r\n\r\nbody\r\n.\r\n") + reader.feed_eof() + smtp._reader = reader + + await smtp.smtp_DATA("") + + assert seen and seen[0] is not None + # Restored once DATA is over, so a stale timestamp cannot shrink the + # budget of a later envelope on the same connection. + assert smtp.data_phase_started is None + + +# --------------------------------------------------------------------------- +# Whole-session deadline. +# +# The one bound a peer cannot push back by staying busy: every other timeout +# is re-armed by activity, so a peer sending one command just under +# PYMTA_COMMAND_TIMEOUT rides command_call_limit for ~39 h on one connection. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_session_deadline_uses_the_configured_limit(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_MAX_SESSION_SECONDS", 1800) + smtp = HardenedSMTP(_NullHandler(), hostname="mta.test", timeout=120) + + armed: list[float] = [] + loop = asyncio.get_running_loop() + real_call_later = loop.call_later + + def _record(delay, callback, *args): + armed.append(delay) + handle = real_call_later(delay, callback, *args) + handle.cancel() + return handle + + monkeypatch.setattr(loop, "call_later", _record) + smtp._arm_session_deadline() + + assert armed == [1800] + assert smtp._session_deadline_handle is not None + + +@pytest.mark.asyncio +async def test_starttls_transport_swap_does_not_extend_the_deadline(monkeypatch): + # aiosmtpd calls connection_made a second time when STARTTLS replaces the + # transport. Re-arming there would hand the peer a fresh budget for the + # price of one STARTTLS. + monkeypatch.setattr(settings, "PYMTA_MAX_SESSION_SECONDS", 1800) + smtp = _server() + first = smtp._session_deadline_handle + assert first is not None + + smtp._arm_session_deadline() + assert smtp._session_deadline_handle is first + + +@pytest.mark.asyncio +async def test_session_deadline_disabled_by_zero(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_MAX_SESSION_SECONDS", 0) + smtp = HardenedSMTP(_NullHandler(), hostname="mta.test", timeout=120) + smtp._arm_session_deadline() + assert smtp._session_deadline_handle is None + + +@pytest.mark.asyncio +async def test_expired_session_announces_then_closes(monkeypatch): + monkeypatch.setattr(settings, "PYMTA_MAX_SESSION_SECONDS", 1800) + smtp = _server() + + smtp._session_expired() + await asyncio.sleep(0) # let the close task run + + assert bytes(smtp._writer.written).startswith(b"421 4.4.2") + assert smtp.transport.closed is True diff --git a/src/mta-in/tests/test_limits.py b/src/mta-in/tests/test_limits.py index 2c650ba3..be4a5d17 100644 --- a/src/mta-in/tests/test_limits.py +++ b/src/mta-in/tests/test_limits.py @@ -9,6 +9,7 @@ from __future__ import annotations import pytest +from pymta import limits from pymta.limits import IPGate, TooManyConnections @@ -139,7 +140,6 @@ async def test_rate_cap_disabled_when_zero(): async def test_rate_dict_prunes_expired_entries(): """The rate map must not grow without bound under churning client IPs.""" clock = _FakeClock() - from pymta import limits # Shrink the prune interval so the test doesn't have to call 1000 times. original = limits._RATE_PRUNE_EVERY diff --git a/src/mta-in/tests/test_mda_async.py b/src/mta-in/tests/test_mda_async.py index b09fa10f..6cfb8a8b 100644 --- a/src/mta-in/tests/test_mda_async.py +++ b/src/mta-in/tests/test_mda_async.py @@ -7,9 +7,13 @@ expected. No real HTTP traffic, no Docker stack. from __future__ import annotations +import datetime + import httpx +import jwt import pytest +from pymta import settings from pymta.mda_async import MDAClient @@ -47,7 +51,7 @@ def _resp(status_code: int, body: bytes = b'{"ok": true}'): def _new_client(*, secret: str = "x" * 32, threshold: int = 3, cooldown: int = 30): - """Construct an MDAClient wired to fakes — no settings module mutation.""" + """Construct an MDAClient wired to fakes, with no settings module mutation.""" clock = _FakeClock() client = MDAClient( base_url="https://mda.example.invalid/api/", @@ -92,14 +96,79 @@ async def test_5xx_returns_temp_fail(): assert result.status_code == 503 +async def _deliver(client): + return await client.deliver( + message=b"From: a@example.com\r\n\r\nbody\r\n", + sender="a@example.com", + original_recipients=["user@example.com"], + client_address="192.0.2.1", + client_port="2525", + client_hostname=None, + client_helo="relay.example.com", + ) + + +@pytest.mark.parametrize("status", [400, 413, 415]) @pytest.mark.asyncio -async def test_4xx_returns_perm_fail(): +async def test_message_rejecting_statuses_are_permanent_on_deliver(status): + # The only statuses that mean "this message is unacceptable, retrying + # cannot help": unparseable, oversize, wrong content type. client, _ = _new_client() - client._client = _StubAsyncClient([_resp(404, b'{"detail":"no"}')]) - result = await client.check_recipient("user@example.com") + client._client = _StubAsyncClient([_resp(status, b'{"detail":"no"}')]) + result = await _deliver(client) assert result.ok is False assert result.temp_fail is False - assert result.status_code == 404 + assert result.status_code == status + + +@pytest.mark.parametrize("status", [400, 413, 415]) +@pytest.mark.asyncio +async def test_message_rejecting_statuses_defer_on_recipient_check(status): + # A recipient check carries no message, so these statuses describe the + # check request we built, not the mailbox. Bouncing on our own bug would + # tell the sender a working address is permanently bad. + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(status, b'{"detail":"no"}')]) + result = await client.check_recipient("user@example.com") + assert result.ok is False + assert result.temp_fail is True + assert result.status_code == status + + +@pytest.mark.parametrize( + "status", + [ + 207, # Multi-Status: some recipients delivered, some not + 401, # secret rotation skew / clock skew on `exp` + 403, + 404, # MDA route missing: a deployment mistake, not a verdict + 429, # throttled + 418, # anything unrecognised defaults to the safe side + ], +) +@pytest.mark.asyncio +async def test_non_rejecting_statuses_defer(status): + # Losing mail is worse than a retry, so everything outside the explicit + # permanent set defers. + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(status, b'{"status":"partial_success"}')]) + result = await client.check_recipient("user@example.com") + assert result.ok is False + assert result.temp_fail is True + assert result.status_code == status + + +@pytest.mark.asyncio +async def test_deferring_statuses_do_not_trip_the_breaker(): + # A 207 or a 401 is a complete answer from a healthy MDA. Only 5xx and + # transport failures are liveness signals. + client, _ = _new_client(threshold=2) + client._client = _StubAsyncClient([_resp(207), _resp(401), _resp(429)]) + for _ in range(3): + result = await client.check_recipient("a@b") + assert result.temp_fail is True + assert client._consecutive_failures == 0 + assert client._open_until is None @pytest.mark.asyncio @@ -111,6 +180,49 @@ async def test_200_returns_ok_with_payload(): assert result.payload == {"user@example.com": True} +# --------------------------------------------------------------------------- +# ``payload`` is always a dict. +# +# Callers read it with ``.get()`` on the reply path of a live SMTP session; a +# body that is not a JSON object would otherwise raise AttributeError there and +# turn a readable defer into a 421 + disconnect. Collapsing to {} keeps the +# reply logic total, and the handlers treat {} as "no answer", never a verdict. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "body", + [ + pytest.param(b"", id="empty"), + pytest.param(b"200 OK", id="html-error-page"), + pytest.param(b"not json at all", id="garbage"), + pytest.param(b'{"user@example.com": true', id="truncated-json"), + pytest.param(b"[true, false]", id="json-list"), + pytest.param(b"null", id="json-null"), + pytest.param(b'"user@example.com"', id="json-string"), + pytest.param(b"42", id="json-number"), + pytest.param(b"true", id="json-bool"), + ], +) +@pytest.mark.parametrize("status", [200, 400, 500]) +@pytest.mark.asyncio +async def test_non_object_bodies_become_an_empty_payload(body, status): + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(status, body)]) + result = await client.check_recipient("user@example.com") + assert result.payload == {} + assert result.status_code == status + + +@pytest.mark.asyncio +async def test_non_object_body_on_deliver_also_becomes_an_empty_payload(): + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(200, b"[]")]) + result = await _deliver(client) + assert result.ok is True + assert result.payload == {} + + # --------------------------------------------------------------------------- # Circuit breaker # --------------------------------------------------------------------------- @@ -125,7 +237,7 @@ async def test_breaker_opens_after_threshold_consecutive_failures(): for _ in range(3): result = await client.check_recipient("a@b") assert result.temp_fail is True - # Breaker should now be open — the next call must NOT hit the network. + # Breaker should now be open; the next call must NOT hit the network. stub = client._client result = await client.check_recipient("a@b") assert result.temp_fail is True @@ -140,10 +252,10 @@ async def test_breaker_closes_after_cooldown(): ) await client.check_recipient("a@b") await client.check_recipient("a@b") - # Breaker open — fast-fail. + # Breaker open, so this fast-fails. await client.check_recipient("a@b") assert client._open_until is not None - # Advance past cooldown — next call probes the network. + # Advance past cooldown; the next call probes the network. clock.advance(31.0) result = await client.check_recipient("a@b") assert result.ok is True @@ -165,8 +277,8 @@ async def test_success_resets_failure_counter(): @pytest.mark.asyncio -async def test_4xx_does_not_count_as_breaker_failure(): - # 4xx means the MDA understood our request and rejected it — that's not a +async def test_non_5xx_does_not_count_as_breaker_failure(): + # 404 means the MDA understood our request and rejected it. That is not a # liveness signal worth tripping the breaker. client, _ = _new_client(threshold=2) client._client = _StubAsyncClient([_resp(404), _resp(404), _resp(404)]) @@ -208,20 +320,28 @@ def test_local_http_url_is_silent(caplog): assert not any("plaintext" in rec.message for rec in caplog.records) +def test_empty_secret_logs_warning(caplog, monkeypatch): + # Without this the misconfiguration only surfaces on the first MDA call, + # as a signing RuntimeError that defers every message. + # An empty `secret` argument falls back to the setting, which the docker + # test runner populates from the env, so blank it to exercise the + # unconfigured case rather than the dev secret. + monkeypatch.setattr(settings, "MDA_API_SECRET", "") + caplog.set_level("WARNING") + MDAClient(base_url="https://mda.example.com/api/", secret="") + assert any("MDA_API_SECRET is empty" in rec.message for rec in caplog.records) + + # --------------------------------------------------------------------------- # JWT claim ordering (B1). # --------------------------------------------------------------------------- def test_metadata_cannot_shadow_exp_or_body_hash(): - import jwt - client, _ = _new_client() body = b"hello" # Attacker-supplied metadata tries to overwrite security fields. - token = client._build_jwt( - body, {"exp": 0, "body_hash": "deadbeef", "sender": "u@x"} - ) + token = client._build_jwt(body, {"exp": 0, "body_hash": "deadbeef", "sender": "u@x"}) decoded = jwt.decode(token, client.secret, algorithms=["HS256"]) # The real exp must be in the future, not 0. assert decoded["exp"] != 0 @@ -229,3 +349,14 @@ def test_metadata_cannot_shadow_exp_or_body_hash(): assert decoded["body_hash"] != "deadbeef" # Sender (non-conflicting metadata) survives. assert decoded["sender"] == "u@x" + + +def test_jwt_ttl_is_configurable(): + # A fixed 60s expiry leaves no room for clock skew against the MDA, and a + # token the MDA reads as expired is a 401, which now defers rather than + # bounces, but still stalls the mail. + client = MDAClient(base_url="https://mda.example.invalid/api/", secret="x" * 32, jwt_ttl=600) + before = datetime.datetime.now(tz=datetime.UTC) + decoded = jwt.decode(client._build_jwt(b"body", {}), client.secret, algorithms=["HS256"]) + ttl = datetime.datetime.fromtimestamp(decoded["exp"], tz=datetime.UTC) - before + assert 590 <= ttl.total_seconds() <= 600 diff --git a/src/mta-in/tests/test_security.py b/src/mta-in/tests/test_security.py index 75f37067..84574787 100644 --- a/src/mta-in/tests/test_security.py +++ b/src/mta-in/tests/test_security.py @@ -153,8 +153,6 @@ def test_expn_disabled(): b"\r\n.\r", # the doubled-CR form Postfix accepted (SEC Consult, 2023) b"\r\r\n.\r\r\n", - # NUL ahead of the terminator, to unstick naive scanners - b"\x00\r\n.\r\n", ], ids=[ "LF-dot-CRLF", @@ -163,26 +161,20 @@ def test_expn_disabled(): "CRLF-dot-LF", "CRLF-dot-CR", "CRCRLF-dot-CRCRLF", - "NUL-CRLF-dot-CRLF", ], ) -def test_smtp_smuggling_does_not_split_messages( - mock_api_server, smuggle_bytes, mta_impl -): +def test_smtp_smuggling_does_not_split_messages(mock_api_server, smuggle_bytes): """A smuggling EOD variant must NOT split the envelope into two messages. The MDA must see at most ONE delivery, and the "smuggled" MAIL FROM/RCPT TO must appear as text inside that single message body — never as a separately-delivered envelope to an attacker-chosen recipient. + + Every variant here is a *malformed* terminator. A payload embedding a + genuine CRLF.CRLF is not a smuggling vector: both implementations end + DATA there and read what follows as a second transaction from the + directly-connected client, which is ordinary submission. """ - if mta_impl == "postfix" and smuggle_bytes.startswith(b"\x00"): - # Unlike the other variants, this payload embeds a *genuine* - # RFC 5321 terminator after the NUL. Postfix accepts and - # normalizes NUL bytes (see test_nul_byte_in_body_rejected), so - # it legitimately ends DATA there and reads what follows as a - # pipelined second transaction from the directly-connected - # client — ordinary submission, not smuggling. - pytest.skip("NUL-tolerant MTA: embedded CRLF.CRLF is a real terminator") mock_api_server.add_mailbox("victim@example.com") # Register the smuggled recipient too: otherwise an actual split would be # rejected at RCPT by the MDA (mailbox not found) and the test would diff --git a/src/mta-in/tests/test_server_config.py b/src/mta-in/tests/test_server_config.py new file mode 100644 index 00000000..767d099e --- /dev/null +++ b/src/mta-in/tests/test_server_config.py @@ -0,0 +1,104 @@ +"""Startup configuration checks in :mod:`pymta.server`. + +There are exactly two supported topologies: + +* PROXY protocol on, behind a balancer named in ``PYMTA_TRUSTED_PROXIES``; +* PROXY protocol off, exposed directly. + +A balancer *without* PROXY protocol is not supported, because pymta would +attribute every session to the balancer's own IP. PROXY protocol without an +allowlist starts, but only behind a loud warning: nothing is left to filter +headers on, so the network isolation carries the whole trust boundary. +""" + +from __future__ import annotations + +import importlib +import logging +from ipaddress import ip_network + +import pytest + +from pymta import settings +from pymta.server import _check_proxy_trust_config + + +def test_proxy_protocol_without_allowlist_starts_with_a_warning(monkeypatch, caplog): + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", []) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert "SECURITY" in caplog.text + assert "PYMTA_TRUSTED_PROXIES is empty" in caplog.text + + +def test_proxy_protocol_with_allowlist_starts_without_warning(monkeypatch, caplog): + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network("10.89.0.0/24")]) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert caplog.records == [] + + +@pytest.mark.parametrize("catch_all", ["0.0.0.0/0", "::/0"]) +def test_proxy_protocol_with_catch_all_allowlist_warns(monkeypatch, caplog, catch_all): + # Non-empty, so it clears the emptiness check, while trusting every peer + # exactly as much as no allowlist would: same posture, same warning. + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network(catch_all)]) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert "matches every peer" in caplog.text + + +def test_catch_all_warns_even_beside_a_real_network(monkeypatch, caplog): + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", True) + monkeypatch.setattr( + settings, + "PYMTA_TRUSTED_PROXIES", + [ip_network("10.89.0.0/24"), ip_network("0.0.0.0/0")], + ) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert "matches every peer" in caplog.text + + +def test_catch_all_is_irrelevant_without_proxy_protocol(monkeypatch, caplog): + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", False) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", [ip_network("0.0.0.0/0")]) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert caplog.records == [] + + +def test_allowlist_is_irrelevant_without_proxy_protocol(monkeypatch, caplog): + # Direct exposure: the wire peer is the client, no header is parsed. + monkeypatch.setattr(settings, "PYMTA_ENABLE_PROXY_PROTOCOL", False) + monkeypatch.setattr(settings, "PYMTA_TRUSTED_PROXIES", []) + with caplog.at_level(logging.WARNING): + _check_proxy_trust_config() + assert caplog.records == [] + + +@pytest.mark.parametrize( + "env, enabled", + [ + ("true", True), + ("false", False), + (None, False), + ], +) +def test_proxy_protocol_reads_only_the_pymta_name(monkeypatch, env, enabled): + # The Postfix image drives the same feature from its own + # ENABLE_PROXY_PROTOCOL=haproxy; pymta must not inherit it, so the two + # services can share an env file. + monkeypatch.setenv("ENABLE_PROXY_PROTOCOL", "haproxy") + if env is None: + monkeypatch.delenv("PYMTA_ENABLE_PROXY_PROTOCOL", raising=False) + else: + monkeypatch.setenv("PYMTA_ENABLE_PROXY_PROTOCOL", env) + try: + assert importlib.reload(settings).PYMTA_ENABLE_PROXY_PROTOCOL is enabled + finally: + monkeypatch.undo() + importlib.reload(settings) diff --git a/src/mta-in/tests/test_settings.py b/src/mta-in/tests/test_settings.py new file mode 100644 index 00000000..79da089e --- /dev/null +++ b/src/mta-in/tests/test_settings.py @@ -0,0 +1,114 @@ +"""Env-var parsing contracts in :mod:`pymta.settings`. + +The settings module runs once at import, so a bad value has to fail there, +loudly, rather than surfacing later as strange SMTP behaviour. +""" + +from __future__ import annotations + +import pytest + +from pymta import settings +from pymta.settings import _env_bool, _env_int, _env_str, _env_token + + +def test_int_reads_env_over_default(monkeypatch): + monkeypatch.setenv("PYMTA_TEST_INT", "42") + assert _env_int("PYMTA_TEST_INT", 7, minimum=1) == 42 + + +def test_int_falls_back_on_unset_and_blank(monkeypatch): + monkeypatch.delenv("PYMTA_TEST_INT", raising=False) + assert _env_int("PYMTA_TEST_INT", 7, minimum=1) == 7 + monkeypatch.setenv("PYMTA_TEST_INT", " ") + assert _env_int("PYMTA_TEST_INT", 7, minimum=1) == 7 + + +def test_int_rejects_non_numeric(monkeypatch): + monkeypatch.setenv("PYMTA_TEST_INT", "soon") + with pytest.raises(ValueError, match="must be an integer"): + _env_int("PYMTA_TEST_INT", 7, minimum=1) + + +@pytest.mark.parametrize("raw", ["0", "-1"]) +def test_int_rejects_values_below_the_minimum(monkeypatch, raw): + # The settings that treat 0 as "disabled" declare minimum=0; everywhere + # else 0 is nonsense that would otherwise fail silently, e.g. a + # PYMTA_MAX_RECIPIENTS of 0 would 452 every recipient. + monkeypatch.setenv("PYMTA_TEST_INT", raw) + with pytest.raises(ValueError, match="must be >= 1"): + _env_int("PYMTA_TEST_INT", 7, minimum=1) + + +def test_data_timeout_must_exceed_the_reply_reserve(monkeypatch): + # The handler subtracts REPLY_RESERVE_SECONDS from this budget. A DATA + # timeout at or below it leaves the deliver call nothing, which would defer + # every message rather than fail loudly at startup. + monkeypatch.setenv("PYMTA_TEST_INT", str(settings.REPLY_RESERVE_SECONDS)) + with pytest.raises(ValueError, match="must be >="): + _env_int("PYMTA_TEST_INT", 300, minimum=settings.REPLY_RESERVE_SECONDS + 1) + assert settings.PYMTA_DATA_TIMEOUT > settings.REPLY_RESERVE_SECONDS + + +def test_int_allows_zero_where_it_means_disabled(monkeypatch): + monkeypatch.setenv("PYMTA_TEST_INT", "0") + assert _env_int("PYMTA_TEST_INT", 7, minimum=0) == 0 + + +@pytest.mark.parametrize("raw", ["\r\n220 you are welcome here", "a\tb"]) +def test_token_rejects_control_characters(monkeypatch, raw): + # These land in the "220 {hostname} {ident}" banner; a CR/LF would append + # attacker-chosen lines to our own greeting, and a TAB folds if the value + # reaches a Received header. (NUL is also rejected but cannot be tested + # through the environment: putenv refuses to store one.) + monkeypatch.setenv("PYMTA_TEST_TOKEN", raw) + with pytest.raises(ValueError, match="control characters"): + _env_token("PYMTA_TEST_TOKEN", "mta-in") + + +def test_token_checks_the_default_too(monkeypatch): + # PYMTA_HOSTNAME defaults to $MYHOSTNAME, which on k8s can come from the + # downward API, so the fallback value needs the same check as the direct one. + monkeypatch.delenv("PYMTA_TEST_TOKEN", raising=False) + with pytest.raises(ValueError, match="control characters"): + _env_token("PYMTA_TEST_TOKEN", "bad\r\nvalue") + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("1", True), + ("true", True), + ("YES", True), + ("on", True), + ("0", False), + ("false", False), + ("no", False), + ("OFF", False), + ], +) +def test_bool_spellings(monkeypatch, raw, expected): + monkeypatch.setenv("PYMTA_TEST_BOOL", raw) + assert _env_bool("PYMTA_TEST_BOOL", not expected) is expected + + +def test_bool_unrecognised_value_is_refused(monkeypatch): + # A typo must not read as its opposite: PYMTA_ENABLE_PROXY_PROTOCOL=Ture + # silently disabling PROXY protocol is a security-relevant misconfiguration. + monkeypatch.setenv("PYMTA_TEST_BOOL", "maybe") + with pytest.raises(ValueError, match="not a recognised boolean"): + _env_bool("PYMTA_TEST_BOOL", True) + with pytest.raises(ValueError, match="not a recognised boolean"): + _env_bool("PYMTA_TEST_BOOL", False) + + +def test_bool_blank_and_missing_take_the_default(monkeypatch): + monkeypatch.setenv("PYMTA_TEST_BOOL", " ") + assert _env_bool("PYMTA_TEST_BOOL", True) is True + monkeypatch.delenv("PYMTA_TEST_BOOL", raising=False) + assert _env_bool("PYMTA_TEST_BOOL", False) is False + + +def test_str_treats_blank_as_unset(monkeypatch): + monkeypatch.setenv("PYMTA_TEST_STR", "") + assert _env_str("PYMTA_TEST_STR", "fallback") == "fallback"