diff --git a/Makefile b/Makefile index 369cffc8..fc6e4864 100644 --- a/Makefile +++ b/Makefile @@ -60,6 +60,7 @@ create-env-files: \ env.d/development/backend.local \ env.d/development/frontend.local \ env.d/development/mta-in.local \ + env.d/development/mta-in-py.local \ env.d/development/mta-out.local \ env.d/development/socks-proxy.local .PHONY: create-env-files @@ -127,6 +128,17 @@ test-back-distroless: build-back-distroless ## build and smoke-test the distrole print(f'OK: Python {sys.version.split()[0]}, {ssl.OPENSSL_VERSION}')" .PHONY: test-back-distroless +build-pymta-distroless: ## build the pymta distroless production image + @docker build --target runtime-distroless-prod -t messages-pymta-distroless -f src/mta-in/Dockerfile.pymta src/mta-in/ +.PHONY: build-pymta-distroless + +test-pymta-distroless: build-pymta-distroless ## build and smoke-test the pymta distroless production image + @docker run --rm messages-pymta-distroless python -c " \ + import sys, ssl; \ + import pymta.settings; \ + print(f'OK: Python {sys.version.split()[0]}, {ssl.OPENSSL_VERSION}, pymta.settings loaded')" +.PHONY: test-pymta-distroless + down: ## stop and remove containers, networks, images, and volumes @$(COMPOSE) down .PHONY: down @@ -181,6 +193,7 @@ lint: \ lint-front \ typecheck-front \ lint-mta-in \ + lint-mta-in-py \ lint-mta-out .PHONY: lint @@ -228,12 +241,17 @@ lint-front: ## run the frontend linter @$(COMPOSE) run --rm frontend-tools npm run lint .PHONY: lint-front -lint-mta-in: ## lint mta-in python sources +lint-mta-in: ## lint mta-in python sources (Postfix milter implementation) $(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-in-test ruff format . #$(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-in-test ruff check . --fix #$(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-in-test pylint . .PHONY: lint-mta-in +lint-mta-in-py: ## lint mta-in python sources (pure-Python pymta implementation) + $(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-in-py-test ruff format . + $(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-in-py-test ruff check . --fix +.PHONY: lint-mta-in-py + lint-mta-out: ## lint mta-out python sources $(COMPOSE_RUN) --rm -e EXEC_CMD_ONLY=true mta-out-test ruff format . .PHONY: lint-mta-out @@ -245,6 +263,7 @@ test: \ test-back \ test-front \ test-mta-in \ + test-mta-in-py \ test-mta-out \ test-mpa \ test-socks-proxy @@ -285,10 +304,14 @@ test-front-amd64: ## run the frontend tests in amd64 $(COMPOSE) run --rm frontend-tools-amd64 npm run test -- $${args:-${1}} .PHONY: test-front-amd64 -test-mta-in: ## run the mta-in tests +test-mta-in: ## run the mta-in tests against the Postfix milter implementation @$(COMPOSE) run --build --rm mta-in-test .PHONY: test-mta-in +test-mta-in-py: ## run the mta-in tests against the pure-Python (aiosmtpd) implementation + @$(COMPOSE) run --build --rm mta-in-py-test +.PHONY: test-mta-in-py + test-mta-out: ## run the mta-out tests @$(COMPOSE) run --build --rm mta-out-test .PHONY: test-mta-out @@ -630,7 +653,7 @@ test-keycloak: ## run all Keycloak provider tests (builds JARs, brings up Keyclo @bin/test-keycloak .PHONY: test-keycloak -deps-lock-mta-in: ## lock the dependencies +deps-lock-mta-in: ## lock the dependencies for mta-in (shared between both implementations) @$(COMPOSE) run --rm --build mta-in-uv uv lock .PHONY: deps-lock-mta-in diff --git a/compose.yaml b/compose.yaml index fee03bdc..2f4bdd9e 100644 --- a/compose.yaml +++ b/compose.yaml @@ -263,6 +263,8 @@ services: - EXEC_CMD=true - MDA_API_BASE_URL=http://localhost:8000/api/mail/ - MTA_HOST=localhost + - MTA_PORT=25 + - MTA_IMPL=postfix command: pytest -vvs tests/ volumes: - ./src/mta-in:/app @@ -277,6 +279,71 @@ services: target: uv pull_policy: build + # ---- Pure-Python (aiosmtpd) inbound MTA -------------------------------- + # Runs side-by-side with the Postfix-based `mta-in` service on a different + # host port (8920 vs 8910). Both implementations share the same MDA + # contract, env vars, and test suite. Toggle which one is the public-facing + # MTA at the edge by switching the upstream pool. + + mta-in-py: + build: + context: src/mta-in + dockerfile: Dockerfile.pymta + target: runtime-distroless-prod + args: + DOCKER_USER: ${DOCKER_USER:-65532} + user: ${DOCKER_USER:-65532} + env_file: + - env.d/development/mta-in.defaults + - env.d/development/mta-in.local + - env.d/development/mta-in-py.defaults + - env.d/development/mta-in-py.local + ports: + - "8920:25" + - "9120:9100" # Prometheus metrics + # Defence-in-depth: pymta needs no on-disk writes at runtime. Read-only + # rootfs + dropped capabilities + no-new-privileges mirror the posture + # a production k8s pod-spec should run with. + read_only: true + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + tmpfs: + - /tmp:rw,noexec,nosuid,size=16m + depends_on: + - backend-dev + + mta-in-py-test: + profiles: + - tools + build: + context: src/mta-in + dockerfile: Dockerfile.pymta + target: runtime-dev + args: + DOCKER_USER: ${DOCKER_USER:-65532} + user: ${DOCKER_USER:-65532} + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + env_file: + - env.d/development/mta-in.defaults + - env.d/development/mta-in.local + - env.d/development/mta-in-py.defaults + - env.d/development/mta-in-py.local + environment: + - EXEC_CMD=true + - MDA_API_BASE_URL=http://localhost:8000/api/mail/ + - MTA_HOST=localhost + - MTA_PORT=25 + - MTA_IMPL=pymta + - MTA_METRICS_URL=http://localhost:9100/metrics + command: pytest -vvs tests/ + volumes: + - ./src/mta-in:/app + mta-out: build: context: src/mta-out diff --git a/env.d/development/mta-in-py.defaults b/env.d/development/mta-in-py.defaults new file mode 100644 index 00000000..4799230a --- /dev/null +++ b/env.d/development/mta-in-py.defaults @@ -0,0 +1,34 @@ +# Defaults for the pure-Python (aiosmtpd) inbound MTA. The Postfix-based +# `mta-in` service does not consume these. + +# PYMTA_HOSTNAME is intentionally left unset so the shared MYHOSTNAME env +# wins (matching the Postfix variant). Set explicitly to override. +PYMTA_SMTP_HOST=0.0.0.0 +PYMTA_SMTP_PORT=25 +PYMTA_METRICS_HOST=0.0.0.0 +PYMTA_METRICS_PORT=9100 +PYMTA_LOG_LEVEL=INFO + +# Match the existing Postfix dev config: per-IP cap off (the dev/test load +# all comes from loopback), large global cap for `test_connection_limits`. +PYMTA_MAX_SESSIONS_PER_IP=0 +PYMTA_MAX_SESSIONS_TOTAL=2000 + +# Inbound limits. +PYMTA_MAX_RECIPIENTS=100 +PYMTA_MAX_ENVELOPES_PER_CONNECTION=20 + +# Timeouts (seconds). +PYMTA_COMMAND_TIMEOUT=120 +PYMTA_DATA_TIMEOUT=600 + +# 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. +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. +PYMTA_ENABLE_PROXY_PROTOCOL=false diff --git a/src/mta-in/Dockerfile b/src/mta-in/Dockerfile index 15a5fedd..5e548dd0 100644 --- a/src/mta-in/Dockerfile +++ b/src/mta-in/Dockerfile @@ -37,8 +37,10 @@ COPY pyproject.toml uv.lock ./ ENV PATH="/venv/bin:$PATH" -# Install dependencies -RUN --mount=type=cache,target=/root/.cache/uv uv sync --locked --no-install-project --no-dev +# Install dependencies. The postfix extra pulls in pymilter (C extension needing +# libmilter-dev at install time); it lives behind an extra so the pure-Python +# pymta image can build without libmilter-dev. +RUN --mount=type=cache,target=/root/.cache/uv uv sync --locked --no-install-project --no-dev --extra postfix # ---- Base image with dependencies installed for development ---- FROM base-with-deps AS base-with-deps-dev diff --git a/src/mta-in/Dockerfile.pymta b/src/mta-in/Dockerfile.pymta new file mode 100644 index 00000000..612aae6a --- /dev/null +++ b/src/mta-in/Dockerfile.pymta @@ -0,0 +1,181 @@ +# Pure-Python inbound MTA image (aiosmtpd) — counterpart to ./Dockerfile +# (Postfix + milter). Both images share src/mta-in/pyproject.toml and live in +# parallel so the two implementations can be tested side-by-side. The Postfix +# image is the production default for now; this image will gradually replace +# it once parity is proven. +# +# Image layout mirrors src/backend/Dockerfile: +# * debian:trixie-slim base (not python:slim — Python comes from uv). +# * uv pinned by SHA256 digest for supply chain integrity. +# * uv-managed python-build-standalone (most C deps statically linked). +# * `python-runtime` stage strips pip/idle/tkinter/headers. +# * `runtime-prod` (slim) and `runtime-distroless-prod` (cc-debian13:nonroot) +# both available; the distroless variant is the security target. +# +# ---- Base OS ---- +FROM debian:trixie-slim AS base + +# Bump this to force an update of the apt repositories +ENV MIN_UPDATE_DATE="2026-06-05" + +RUN <` 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). + - 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. + +## 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. + +## Running + +```bash +# Default Postfix-based service make test-mta-in -``` - -You should also run lint before commit: - -``` make lint-mta-in -``` \ No newline at end of file + +# Pure-Python (aiosmtpd) service +make test-mta-in-py +make lint-mta-in-py +``` + +The shared test suite under `tests/` runs against both via the `MTA_HOST` / `MTA_PORT` env vars. A few tests assert implementation-specific behaviour and skip on the other impl (e.g. `tests/test_metrics.py` is pymta-only; the strict NUL-byte rejection test skips on Postfix). The fixtures `mta_impl`, `mta_address`, and `mta_metrics_url` in `tests/conftest.py` are how a test sees which impl it is running against. + +## pymta-specific env vars + +In addition to the shared `MDA_API_BASE_URL` / `MDA_API_SECRET` / `MDA_API_TIMEOUT` / `MAX_INCOMING_EMAIL_SIZE`, the pymta image reads: + +| Variable | Default | Purpose | +|---|---|---| +| `PYMTA_HOSTNAME` | `mta-in` | Banner / Received-header host name | +| `PYMTA_SMTP_HOST` / `PYMTA_SMTP_PORT` | `0.0.0.0` / `25` | SMTP listener bind | +| `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_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_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 | +| `PYMTA_ENABLE_SMTPUTF8` | `true` | Advertise SMTPUTF8 in EHLO | +| `ENABLE_PROXY_PROTOCOL` | unset | Set to `haproxy` to enable PROXY-protocol v1/v2 | diff --git a/src/mta-in/entrypoint.pymta.sh b/src/mta-in/entrypoint.pymta.sh new file mode 100755 index 00000000..0501d704 --- /dev/null +++ b/src/mta-in/entrypoint.pymta.sh @@ -0,0 +1,59 @@ +#!/bin/sh +# Entrypoint for the pure-Python (aiosmtpd) inbound MTA image. +# +# Two modes: +# - EXEC_CMD_ONLY=true: skip starting the server, exec the user command. Used +# by ad-hoc tooling like `ruff format .`. +# - EXEC_CMD=true: start the server in the background, then exec the user +# command. Used by the test runner to colocate pytest + pymta in a single +# container (mirroring the postfix mta-in-test workflow). +# - default: exec the server in the foreground. +set -eu + +if [ "${EXEC_CMD_ONLY:-false}" = "true" ]; then + exec "$@" +fi + +start_pymta() { + python -m pymta.server & + PYMTA_PID=$! + # Wait until the SMTP port accepts connections (max ~15s). Uses stdlib + # socket rather than nc so the runtime image doesn't need netcat just + # for this probe. + port="${PYMTA_SMTP_PORT:-25}" + for i in $(seq 1 30); do + if python -c "import socket, sys; s=socket.socket(); s.settimeout(0.5); s.connect(('127.0.0.1', int('$port'))); s.close()" 2>/dev/null; then + echo "pymta SMTP ready on port $port" + return 0 + fi + sleep 0.5 + done + echo "ERROR: pymta SMTP did not open port $port within 15s" >&2 + kill "$PYMTA_PID" 2>/dev/null || true + return 1 +} + +cleanup() { + if [ -n "${PYMTA_PID:-}" ]; then + kill "$PYMTA_PID" 2>/dev/null || true + # Give pymta a chance to flush logs / drain sessions before we exit. + wait "$PYMTA_PID" 2>/dev/null || true + fi +} +trap cleanup INT TERM + +if [ "${EXEC_CMD:-false}" = "true" ]; then + start_pymta + status=$? + if [ "$status" -ne 0 ]; then + echo "ERROR: pymta failed to start, not executing command" >&2 + cleanup + exit "$status" + fi + "$@" + status=$? + cleanup + exit $status +fi + +exec python -m pymta.server diff --git a/src/mta-in/pyproject.toml b/src/mta-in/pyproject.toml index a9b75eda..9b3f3e79 100644 --- a/src/mta-in/pyproject.toml +++ b/src/mta-in/pyproject.toml @@ -29,7 +29,9 @@ requires-python = ">=3.14.4,<4.0" dependencies = [ "requests==2.32.3", "PyJWT==2.10.1", - "pymilter==1.0.5" + "aiosmtpd==1.4.6", + "httpx==0.28.1", + "prometheus-client==0.24.1", ] [project.urls] @@ -39,9 +41,15 @@ dependencies = [ "Repository" = "https://github.com/suitenumerique/st-messages" [project.optional-dependencies] +# pymilter is a C extension that needs libmilter-dev at install time. Keep it +# behind an extra so the pure-Python pymta image can build without libmilter. +postfix = [ + "pymilter==1.0.5", +] dev = [ "pytest==8.3.5", "pytest-cov==6.0.0", + "pytest-asyncio==0.24.0", "fastapi==0.115.12", "uvicorn==0.34.1", "ruff==0.9.3", @@ -92,7 +100,11 @@ sections = { django=["django"] } extra-standard-library = ["tomllib"] [tool.ruff.lint.per-file-ignores] -"**/tests/*" = ["S", "SLF"] +"**/tests/*" = ["S", "SLF", "C405", "PLR0913"] +# The Postfix milter predates the pure-Python pymta module and intentionally +# uses bare except + print for stdout-based logging in the libmilter +# sync callback context. +"src/delivery_milter.py" = ["BLE001", "T201"] [tool.pytest.ini_options] addopts = [ diff --git a/src/mta-in/src/api/mda.py b/src/mta-in/src/api/mda.py index 20906fe7..be214880 100644 --- a/src/mta-in/src/api/mda.py +++ b/src/mta-in/src/api/mda.py @@ -33,6 +33,7 @@ def mda_api_call(path, content_type, body, metadata): now = datetime.datetime.now(datetime.timezone.utc) jwt_token = jwt.encode( { + **metadata, "exp": now + datetime.timedelta(seconds=MDA_API_JWT_TTL), # The channel is authenticated by the HMAC signature over the shared # MDA_API_SECRET; body_hash binds the token to its payload (sha256 of @@ -41,7 +42,6 @@ def mda_api_call(path, content_type, body, metadata): # No jti/nonce: retries (and urllib3's) resend the same token, and # the backend trusts the secret rather than tracking single use. "body_hash": hashlib.sha256(body).hexdigest(), - **metadata, }, MDA_API_SECRET, algorithm="HS256", diff --git a/src/mta-in/src/pymta/__init__.py b/src/mta-in/src/pymta/__init__.py new file mode 100644 index 00000000..e445234d --- /dev/null +++ b/src/mta-in/src/pymta/__init__.py @@ -0,0 +1,10 @@ +""" +Pure-Python inbound MTA built on aiosmtpd. + +Reception-side counterpart to the Postfix+milter implementation that lives in +``src/mta-in/src/delivery_milter.py``. Both share the same MDA REST contract +(``inbound/mta/check/`` + ``inbound/mta/deliver/``), and both run as a +stateless, queue-less SMTP front-end: each SMTP session blocks on the +synchronous delivery HTTP call and translates the outcome straight back to the +remote peer. +""" diff --git a/src/mta-in/src/pymta/address.py b/src/mta-in/src/pymta/address.py new file mode 100644 index 00000000..be98e506 --- /dev/null +++ b/src/mta-in/src/pymta/address.py @@ -0,0 +1,186 @@ +"""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 +(RFC 5321 §4.1.1.3), control characters (CRLF injection vector), overlong +local-parts or domains, and the common ``user@`` / ``@domain`` truncations. + +They never accept already-unbalanced quoting or angle brackets. +:func:`validate_envelope_address` accepts either the wrapped (````) +or unwrapped form; :func:`strip_brackets` runs unconditionally on entry. +""" + +from __future__ import annotations + +# Characters that must never appear unquoted in an envelope address. CR, LF, +# and NUL are the CRLF-injection and frame-confusion vectors. TAB is a header +# unfolding vector. ``%`` is included to keep us out of historical +# "percent-routing" relay tricks (RFC 1123 §5.2.16). DEL (0x7f) and bare +# space have no place in an address received from the wire. +_FORBIDDEN_CHARS = frozenset({"\r", "\n", "\x00", "\t", " ", "%", "\x7f"}) + + +class AddressError(ValueError): + """Raised when an envelope address fails validation. + + The ``reason`` attribute carries a short token suitable for a Prometheus + metric label and the ``smtp_code`` / ``smtp_text`` tuple gives the exact + SMTP reply the caller should send back to the peer. + """ + + def __init__(self, reason: str, smtp_text: str, smtp_code: int = 553): + super().__init__(smtp_text) + self.reason = reason + self.smtp_text = smtp_text + self.smtp_code = smtp_code + + +def strip_brackets(raw: str) -> str: + """Strip a single pair of surrounding angle brackets. + + Returns the input unchanged if there is no leading ``<``. Does not validate + that the address inside is well-formed. + """ + if not raw: + return raw + if raw.startswith("<") and raw.endswith(">"): + return raw[1:-1] + return raw + + +def validate_envelope_address( # noqa: PLR0912 + raw: str, + *, + allow_empty: bool, + max_local: int, + max_domain: int, +) -> str: + """Validate ``raw`` as an RFC 5321 envelope address. + + ``allow_empty`` controls whether the null sender ``<>`` is accepted. It + must be true for MAIL FROM and false for RCPT TO (RFC 5321 §4.5.5). + + Returns the cleaned address (lower-cased domain, original local-part) on + success, raises :class:`AddressError` on failure. + """ + address = strip_brackets(raw or "") + + if address == "": + if allow_empty: + return "" + raise AddressError( + reason="bad_address", + smtp_code=553, + smtp_text="5.1.3 Empty recipient address not allowed", + ) + + # ----- 1a. residual angle brackets --------------------------------------- + # strip_brackets only removes a balanced outer pair; any leftover '<' or '>' + # means the address is malformed (unbalanced or nested brackets). + if "<" in address or ">" in address: + raise AddressError( + reason="bad_address", + smtp_code=501, + smtp_text="5.1.3 Malformed address syntax", + ) + + # ----- 1b. control / CRLF / NUL injection -------------------------------- + bad = _FORBIDDEN_CHARS & set(address) + if bad: + raise AddressError( + reason="control_char", + smtp_code=501, + smtp_text="5.1.7 Address contains forbidden control characters", + ) + + # ----- 2. source routes @host1,@host2:user@host3 ----------------------- + # RFC 5321 §4.1.1.3 allows ignoring source routes; we reject outright. + if address.startswith("@"): + raise AddressError( + reason="source_route", + smtp_code=553, + smtp_text="5.1.3 Source routes are not accepted", + ) + + # ----- 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 + # a fertile parser-confusion ground. + if address.count("@") != 1: + raise AddressError( + reason="bad_address", + smtp_code=501, + smtp_text="5.1.3 Bad address syntax", + ) + + local, _, domain = address.partition("@") + + if not local or not domain: + raise AddressError( + reason="bad_address", + smtp_code=501, + smtp_text="5.1.3 Bad address syntax", + ) + + # ----- 3a. quoted local-parts ------------------------------------------- + # aiosmtpd's email-parser unwraps the quoted form (``"a"@b.com`` arrives + # with quotes intact in the local-part). The MDA's mailbox lookup + # normalises differently than our address validator, which is a + # parser-mismatch vector. Reject double quotes outright. + if '"' in local: + raise AddressError( + reason="bad_address", + smtp_code=553, + smtp_text="5.1.3 Quoted local-parts not accepted", + ) + + # ----- 3b. dot placement in unquoted local-part (RFC 5321 §4.1.2) ------- + # A leading dot, trailing dot, or two consecutive dots are illegal in an + # unquoted local-part. Different mailbox-lookup paths normalise these + # inconsistently, so reject at the gate. + if local.startswith(".") or local.endswith(".") or ".." in local: + raise AddressError( + reason="bad_address", + smtp_code=553, + smtp_text="5.1.3 Malformed local part", + ) + + # ----- 4. length limits (RFC 5321 §4.5.3.1) ----------------------------- + if len(local.encode("utf-8")) > max_local: + raise AddressError( + reason="oversize_local", + smtp_code=553, + smtp_text=f"5.1.3 Local part exceeds {max_local} octets", + ) + if len(domain.encode("utf-8")) > max_domain: + raise AddressError( + reason="oversize_domain", + smtp_code=553, + smtp_text=f"5.1.3 Domain part exceeds {max_domain} octets", + ) + + # ----- 5. domain shape --------------------------------------------------- + # Allow IDN/UTF-8 in the domain; reject empty labels, leading dot, label + # > 63 octets, and bare IP literals (the inbound path expects FQDNs from + # legitimate senders). + if domain.startswith("[") and domain.endswith("]"): + raise AddressError( + reason="address_literal", + smtp_code=501, + smtp_text="5.1.3 Address literals not accepted", + ) + if domain.startswith(".") or domain.endswith(".") or ".." in domain: + raise AddressError( + reason="bad_address", + smtp_code=501, + smtp_text="5.1.3 Malformed domain", + ) + for label in domain.split("."): + if not label or len(label.encode("utf-8")) > 63: + raise AddressError( + reason="bad_address", + smtp_code=501, + smtp_text="5.1.3 Malformed domain label", + ) + + return f"{local}@{domain.lower()}" diff --git a/src/mta-in/src/pymta/controller.py b/src/mta-in/src/pymta/controller.py new file mode 100644 index 00000000..57acc06f --- /dev/null +++ b/src/mta-in/src/pymta/controller.py @@ -0,0 +1,97 @@ +"""aiosmtpd Controller wired to our :class:`HardenedSMTP` factory. + +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. +""" + +from __future__ import annotations + +import asyncio +import logging +import ssl + +from aiosmtpd.controller import UnthreadedController + +from . import settings +from .handler import InboundHandler +from .limits import IPGate +from .smtp_protocol import HardenedSMTP + +logger = logging.getLogger(__name__) + + +def build_smtp_kwargs(*, tls_context: ssl.SSLContext | None) -> dict: + """Centralise the SMTP-class options driven by settings.""" + return { + "hostname": settings.PYMTA_HOSTNAME, + "ident": settings.PYMTA_IDENT, + "data_size_limit": settings.MAX_INCOMING_EMAIL_SIZE, + "enable_SMTPUTF8": settings.PYMTA_ENABLE_SMTPUTF8, + "timeout": settings.PYMTA_COMMAND_TIMEOUT, + # Per-verb call ceilings (defence against pipelining floods). Numbers + # come from "what a sane sender would ever do in one TCP session"; + # anything above means the peer is hammering us. + "command_call_limit": { + "EHLO": 4, + "HELO": 4, + "NOOP": 5, + "MAIL": settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION + 2, + "RCPT": settings.PYMTA_MAX_RECIPIENTS * settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION + + 10, + "DATA": settings.PYMTA_MAX_ENVELOPES_PER_CONNECTION + 2, + "RSET": 20, + "QUIT": 1, + # STARTTLS pinned explicitly so a future contributor cannot raise + # the "*" bucket and silently let a peer burn TLS handshakes by + # repeating EHLO/STARTTLS within one TCP session. + "STARTTLS": 2, + "*": 25, + }, + "proxy_protocol_timeout": ( + settings.PYMTA_PROXY_PROTOCOL_TIMEOUT if settings.PYMTA_ENABLE_PROXY_PROTOCOL else None + ), + "tls_context": tls_context, + # Plaintext AUTH on port 25 is unsafe; AUTH stays off entirely. + "auth_require_tls": True, + "auth_required": False, + "auth_exclude_mechanism": ("LOGIN", "PLAIN"), # ggignore + } + + +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. + """ + cert = settings.PYMTA_TLS_CERT_FILE + key = settings.PYMTA_TLS_KEY_FILE + if not cert or not key: + return None + ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + # As an MTA we accept any client identity; we just want our side encrypted. + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.load_cert_chain(cert, key) + return ctx + + +class HardenedController(UnthreadedController): + """:class:`UnthreadedController` returning a :class:`HardenedSMTP`.""" + + def __init__( + self, + handler: InboundHandler, + *, + ip_gate: IPGate, + hostname: str, + port: int, + loop: asyncio.AbstractEventLoop | None = None, + ): + self._ip_gate = ip_gate + tls_context = load_tls_context() + self._smtp_kwargs = build_smtp_kwargs(tls_context=tls_context) + super().__init__(handler, hostname=hostname, port=port, loop=loop) + + def factory(self) -> HardenedSMTP: # type: ignore[override] + return HardenedSMTP(self.handler, ip_gate=self._ip_gate, **self._smtp_kwargs) diff --git a/src/mta-in/src/pymta/handler.py b/src/mta-in/src/pymta/handler.py new file mode 100644 index 00000000..eae7f48f --- /dev/null +++ b/src/mta-in/src/pymta/handler.py @@ -0,0 +1,336 @@ +"""aiosmtpd handler implementing the queue-less inbound delivery flow. + +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 + 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/`` + and translates the MDA outcome back to a single SMTP reply line. + +The handler keeps no on-disk queue and no persistent envelope log: a 250 to +the peer means the MDA has already accepted the message; a 4xx means the +peer should retry later. +""" + +from __future__ import annotations + +import asyncio +import logging + +from . import metrics, settings +from .address import AddressError, validate_envelope_address +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. +_ENVELOPES_ATTR = "_pymta_envelopes" +_SOFT_ERRORS_ATTR = "_pymta_soft_errors" +_RCPT_MISSES_ATTR = "_pymta_rcpt_misses" + +# Sentinel for the RFC 5321 null sender (MAIL FROM:<>). aiosmtpd's +# ``smtp_RCPT`` rejects with 503 when ``envelope.mail_from`` is falsy, which +# would block legitimate bounces. We keep the sentinel internally and rewrite +# it back to the empty string when calling the MDA, matching the Postfix +# milter's existing wire contract. +NULL_SENDER_SENTINEL = "<>" + +# 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 _bump_envelopes(session) -> int: + n = _envelopes_count(session) + 1 + setattr(session, _ENVELOPES_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 _bump_rcpt_misses(session) -> int: + n = getattr(session, _RCPT_MISSES_ATTR, 0) + 1 + setattr(session, _RCPT_MISSES_ATTR, n) + return n + + +def _peer_ip(session) -> str | None: + 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) + peer = getattr(session, "peer", None) + if peer and len(peer) >= 1: + return str(peer[0]) + return None + + +def _peer_port(session) -> 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) + peer = getattr(session, "peer", None) + if peer and len(peer) >= 2: + return str(peer[1]) + return None + + +def _safe_hostname(raw: str | None, session=None) -> str | None: + """Return ``raw`` only if it is free of CRLF/NUL/TAB; otherwise None. + + The MDA receives this through a JWT claim; downstream consumers may + interpolate it into log lines or ``Received`` headers, so a control char + here is a header-injection vector. + + When ``session`` is supplied, a rejected hostname is counted and logged so + operators can spot floods of malformed HELO/EHLO greetings. + """ + if raw is None: + return None + if _FORBIDDEN_HOSTNAME_CHARS & set(raw): + metrics.SECURITY_REJECTIONS.labels(reason="bad_helo").inc() + logger.info("dropping HELO/EHLO with forbidden control chars from %s", _peer_ip(session)) + return None + return raw + + +class InboundHandler: + """One instance per server process; called concurrently from many sessions.""" + + def __init__(self, mda_client: MDAClient): + self.mda = mda_client + + # ------------------------------------------------------------------ EHLO + async def handle_EHLO(self, server, session, envelope, hostname, responses): + """Customize the EHLO response list. + + Strips any extension keyword we've decided not to expose on inbound + port 25: AUTH (would invite credential-stuffing or open relay if + misconfigured), CHUNKING/BDAT (smuggling parser-confusion surface), + and PIPELINING (announcing it advertises that we accept rapid command + coalescing; the actual per-verb rate cap lives in + ``controller.py:command_call_limit``). aiosmtpd already omits these + by default; we keep the filter as a guard against future regressions + or a contributor wiring an authenticator without re-reading the + security rationale. + """ + denied_verbs = {"AUTH", "CHUNKING", "BDAT", "PIPELINING"} + clean: list[str] = [] + for line in responses: + # line looks like "250-FOO bar" or "250 FOO bar". + after = line[4:] if len(line) > 4 else "" + verb = after.split(" ", 1)[0].upper() + if verb in denied_verbs: + metrics.SECURITY_REJECTIONS.labels(reason="auth_offered").inc() + logger.warning( + "stripping disallowed EHLO extension %r from %s — review the " + "SMTP configuration so it is not advertised in the first place", + verb, + _peer_ip(session), + ) + continue + clean.append(line) + + session.host_name = _safe_hostname(hostname, session=session) + return clean + + async def handle_HELO(self, server, session, envelope, hostname): + session.host_name = _safe_hostname(hostname, session=session) + return f"250 {server.hostname}" + + # ------------------------------------------------------------------ MAIL + async def handle_MAIL(self, server, session, envelope, address, mail_options): + try: + clean = validate_envelope_address( + address, + allow_empty=True, + max_local=settings.PYMTA_MAX_LOCAL_PART, + max_domain=settings.PYMTA_MAX_DOMAIN, + ) + except AddressError as err: + metrics.SECURITY_REJECTIONS.labels(reason=err.reason).inc() + return f"{err.smtp_code} {err.smtp_text}" + + # Honour MAIL FROM:... SIZE=N if announced — 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) + 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) + return "552 5.3.4 Message size exceeds fixed maximum" + + envelope.mail_from = clean if clean else NULL_SENDER_SENTINEL + envelope.mail_options.extend(mail_options or []) + return "250 2.1.0 OK" + + # ------------------------------------------------------------------ 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() + metrics.RCPT_TOTAL.labels(result="rejected_temp").inc() + return "421 4.7.0 Too many errors, goodbye" + + # Per-envelope recipient cap. + 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) + return "452 4.5.3 Too many recipients" + + try: + clean = validate_envelope_address( + address, + allow_empty=False, + max_local=settings.PYMTA_MAX_LOCAL_PART, + max_domain=settings.PYMTA_MAX_DOMAIN, + ) + except AddressError as err: + metrics.SECURITY_REJECTIONS.labels(reason=err.reason).inc() + metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() + _bump_soft_errors(session) + 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) + return "451 4.3.0 Recipient verification temporarily unavailable" + if not result.ok: + metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() + _bump_soft_errors(session) + return "550 5.1.1 Recipient verification failed" + + exists = bool(result.payload.get(clean, False)) + if not exists: + metrics.RCPT_TOTAL.labels(result="rejected_perm").inc() + _bump_soft_errors(session) + misses = _bump_rcpt_misses(session) + 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() + return "421 4.7.0 Too many unknown recipients, goodbye" + return "550 5.1.1 No such recipient" + + envelope.rcpt_tos.append(clean) + envelope.rcpt_options.extend(rcpt_options or []) + metrics.RCPT_TOTAL.labels(result="accepted").inc() + return "250 2.1.5 OK" + + # ------------------------------------------------------------------ DATA + async def handle_DATA(self, server, session, envelope): # noqa: PLR0911 + envelopes = _bump_envelopes(session) + 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) + 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. + 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) + return "554 5.6.0 NUL byte in message body" + + if len(content) > settings.MAX_INCOMING_EMAIL_SIZE: + # aiosmtpd already replies 552 itself when the in-flight DATA + # 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) + return "552 5.3.4 Message size exceeds fixed maximum" + + try: + sender = envelope.mail_from + if sender == NULL_SENDER_SENTINEL: + sender = "" + result: MDAResult = await asyncio.wait_for( + self.mda.deliver( + message=content, + sender=sender, + original_recipients=list(envelope.rcpt_tos), + client_address=_peer_ip(session), + client_port=_peer_port(session), + # We do not reverse-DNS ourselves: the MDA inserts its + # 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), + ), + timeout=settings.PYMTA_DATA_TIMEOUT, + ) + except TimeoutError: + metrics.MESSAGES_TOTAL.labels(result="rejected_temp").inc() + metrics.MESSAGE_BYTES.observe(len(content)) + _bump_soft_errors(session) + logger.warning( + "DATA deliver deadline exceeded (%ds) for peer %s", + settings.PYMTA_DATA_TIMEOUT, + _peer_ip(session), + ) + return "451 4.3.0 Delivery timed out, please retry" + + metrics.MESSAGE_BYTES.observe(len(content)) + + 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: + metrics.MESSAGES_TOTAL.labels(result="rejected_temp").inc() + _bump_soft_errors(session) + return "451 4.3.0 Delivery temporarily unavailable" + metrics.MESSAGES_TOTAL.labels(result="rejected_perm").inc() + _bump_soft_errors(session) + return "554 5.6.0 Message rejected by delivery agent" + + # ------------------------------------------------------------------ PROXY + async def handle_PROXY(self, server, session, envelope, proxy_data): + """Apply admission control once PROXY-protocol parsing is done. + + Routing the gate through here (rather than at SMTP-connect time) + means we count sessions against the real client IP carried in the + PROXY header, not against the load-balancer's IP. Without this, + every session behind HAProxy would be bucketed under one address + and ``PYMTA_MAX_SESSIONS_PER_IP`` would silently turn into a global + cap. + """ + real_ip = "unknown" + if proxy_data is not None and getattr(proxy_data, "src_addr", None): + real_ip = str(proxy_data.src_addr) + return await server.acquire_gate_post_proxy(real_ip) + + # ------------------------------------------------------------------ misc + async def handle_exception(self, error: BaseException) -> str: + # Never leak stack traces or internal hostnames in SMTP replies. + metrics.SECURITY_REJECTIONS.labels(reason="internal_error").inc() + metrics.DISCONNECTS_421.labels(reason="internal_error").inc() + logger.exception("Unhandled error in SMTP handler") + return "421 4.3.0 Internal error, please try again later" diff --git a/src/mta-in/src/pymta/limits.py b/src/mta-in/src/pymta/limits.py new file mode 100644 index 00000000..acb785bd --- /dev/null +++ b/src/mta-in/src/pymta/limits.py @@ -0,0 +1,118 @@ +"""Connection-level admission control for the pymta server. + +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 + 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 +where the whole load comes from the same loopback address. +""" + +from __future__ import annotations + +import asyncio +import logging +import time + +from . import metrics + +logger = logging.getLogger(__name__) + + +# Rolling window used by the per-IP rate cap. +_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 +# many distinct client IPs (each entry would otherwise live one full window +# beyond its last use). +_RATE_PRUNE_EVERY = 1000 + + +class TooManyConnections(Exception): + """Raised when the global, per-IP concurrent, or per-IP rate cap is hit.""" + + def __init__(self, scope: str): + super().__init__(scope) + self.scope = scope + + +class IPGate: + """Tracks live SMTP sessions per remote IP and globally. + + Acquisition does not block: if any cap is reached we raise immediately + so the caller can close the socket and reply ``421`` instead of holding + the connection open and amplifying the attack. + + The ``_try_acquire`` / ``_release`` pair is called from + :class:`pymta.smtp_protocol.HardenedSMTP` (post-PROXY when applicable). + """ + + def __init__( + self, + *, + max_total: int, + max_per_ip: int, + max_per_ip_per_minute: int = 0, + clock=time.monotonic, + ): + self.max_total = max_total + self.max_per_ip = max_per_ip + self.max_per_ip_per_minute = max_per_ip_per_minute + self._clock = clock + self._lock = asyncio.Lock() + self._per_ip: dict[str, int] = {} + self._total = 0 + # (count_in_window, window_start_monotonic) per IP + self._rate_per_ip: dict[str, tuple[int, float]] = {} + self._acquires_since_prune = 0 + + async def _try_acquire(self, ip: str) -> None: + async with self._lock: + if self.max_total and self._total >= self.max_total: + raise TooManyConnections("global") + if self.max_per_ip and self._per_ip.get(ip, 0) >= self.max_per_ip: + raise TooManyConnections("per_ip") + if self.max_per_ip_per_minute: + now = self._clock() + count, start = self._rate_per_ip.get(ip, (0, now)) + if now - start >= _RATE_WINDOW_SECONDS: + count, start = 0, now + if count >= self.max_per_ip_per_minute: + raise TooManyConnections("per_ip_rate") + self._rate_per_ip[ip] = (count + 1, start) + self._acquires_since_prune += 1 + if self._acquires_since_prune >= _RATE_PRUNE_EVERY: + self._prune_expired_rates(now) + self._acquires_since_prune = 0 + self._per_ip[ip] = self._per_ip.get(ip, 0) + 1 + self._total += 1 + metrics.SESSIONS_ACTIVE.inc() + metrics.SESSIONS_PER_IP.set(len(self._per_ip)) + + async def _release(self, ip: str) -> None: + async with self._lock: + new = self._per_ip.get(ip, 0) - 1 + if new <= 0: + self._per_ip.pop(ip, None) + else: + self._per_ip[ip] = new + self._total = max(0, self._total - 1) + metrics.SESSIONS_ACTIVE.dec() + metrics.SESSIONS_PER_IP.set(len(self._per_ip)) + + def _prune_expired_rates(self, now: float) -> None: + # Drop IPs whose window has fully elapsed; keeps the rate dict bounded + # to the set of IPs seen in roughly the last minute. + expired = [ + ip + for ip, (_count, start) in self._rate_per_ip.items() + if now - start >= _RATE_WINDOW_SECONDS + ] + for ip in expired: + del self._rate_per_ip[ip] diff --git a/src/mta-in/src/pymta/mda_async.py b/src/mta-in/src/pymta/mda_async.py new file mode 100644 index 00000000..22b1ba4a --- /dev/null +++ b/src/mta-in/src/pymta/mda_async.py @@ -0,0 +1,289 @@ +"""Async HTTP client for the MDA inbound API. + +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: + +* ``POST /inbound/mta/check/`` with ``application/json`` body + ``{"addresses": [...]}`` → returns ``{addr: bool}``. +* ``POST /inbound/mta/deliver/`` with ``message/rfc822`` body (the full + message bytes). The metadata (sender, recipients, client info, size) is + carried as JWT claims, not in the body. + +Every request is signed with a short-lived HS256 JWT whose body_hash claim +binds the JWT to the exact bytes being posted (replay-proofing). +""" + +from __future__ import annotations + +import datetime +import hashlib +import json +import logging +import time +from dataclasses import dataclass +from urllib.parse import urlparse + +import httpx +import jwt + +from . import metrics, settings + +logger = logging.getLogger(__name__) + +# Local development URLs are the only place we tolerate a plaintext MDA; +# anywhere else a leaked JWT secret on the wire is a credential incident. +_LOCAL_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "::1"}) +# Below this many bytes a shared HS256 secret is brute-forceable; refuse to +# even start the process rather than minting weak tokens. +_MIN_SECRET_LENGTH = 32 + + +@dataclass(frozen=True) +class MDAResult: + """Result of an MDA call. + + ``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. + """ + + ok: bool + temp_fail: bool + payload: dict + status_code: int + + +class MDAClient: + """Thin async wrapper over the MDA REST API. + + Lifetime: one instance per server process. Reuses one + :class:`httpx.AsyncClient` so the HTTP channel survives many SMTP + sessions (HTTP keep-alive); each individual SMTP transaction still + blocks on a synchronous MDA call so there is no on-disk queue. + """ + + def __init__( + self, + base_url: str | None = None, + secret: str | None = None, + timeout: int | None = None, + breaker_threshold: int | None = None, + breaker_cooldown: 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._breaker_threshold = ( + breaker_threshold + if breaker_threshold is not None + else settings.PYMTA_MDA_BREAKER_THRESHOLD + ) + self._breaker_cooldown = ( + breaker_cooldown + if breaker_cooldown is not None + else settings.PYMTA_MDA_BREAKER_COOLDOWN + ) + self._clock = clock + # Counts consecutive failures. Reset to 0 by any successful call. + self._consecutive_failures = 0 + # Monotonic-time deadline until which the breaker stays open. None + # when closed; a future timestamp when open. + self._open_until: float | None = None + self._client: httpx.AsyncClient | None = None + self._validate_credentials() + + def _validate_credentials(self) -> None: + """Warn loudly at startup about weak secret or 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. + """ + parsed = urlparse(self.base_url) + host = (parsed.hostname or "").lower() + if parsed.scheme == "http" and host not in _LOCAL_HOSTNAMES: + logger.warning( + "MDA_API_BASE_URL uses plaintext http:// for non-local host %r " + "(%r). The JWT bearer token will traverse the network in clear. " + "Configure https:// in production.", + host, + self.base_url, + ) + if self.secret and 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.", + len(self.secret), + _MIN_SECRET_LENGTH, + ) + + async def start(self) -> httpx.AsyncClient: + """Open the persistent HTTP client. Idempotent.""" + if self._client is None: + limits = httpx.Limits(max_keepalive_connections=20, max_connections=100) + self._client = httpx.AsyncClient(timeout=self.timeout, limits=limits) + return self._client + + async def close(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + + def _build_jwt(self, body: bytes, metadata: dict) -> str: + if not self.secret: + raise RuntimeError("MDA_API_SECRET is required to sign MDA API requests") + # Spread metadata FIRST so a stray metadata key named "exp" or + # "body_hash" cannot shadow the security-relevant claims. + claims = { + **metadata, + "exp": datetime.datetime.now(tz=datetime.UTC) + datetime.timedelta(seconds=60), + "body_hash": hashlib.sha256(body).hexdigest(), + } + return jwt.encode(claims, self.secret, algorithm="HS256") + + def _breaker_open(self) -> bool: + """True when the circuit is currently shedding traffic.""" + if self._open_until is None: + return False + if self._clock() >= self._open_until: + # Cool-down elapsed; let the next request probe upstream. + self._open_until = None + self._consecutive_failures = 0 + return False + return True + + def _record_failure(self) -> None: + if not self._breaker_threshold: + return + self._consecutive_failures += 1 + 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", + 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 + ) + self._consecutive_failures = 0 + + async def _post( + self, + path: str, + content_type: str, + body: bytes, + metadata: dict, + endpoint_label: str, + ) -> MDAResult: + if self._breaker_open(): + metrics.MDA_REQUEST_DURATION.labels( + endpoint=endpoint_label, result="breaker_open" + ).observe(0) + return MDAResult(ok=False, temp_fail=True, payload={}, status_code=0) + + client = self._client or await self.start() + + url = self.base_url + path.lstrip("/") + token = self._build_jwt(body, metadata) + headers = {"Content-Type": content_type, "Authorization": f"Bearer {token}"} + + start = self._clock() + try: + response = await client.post(url, content=body, headers=headers) + except httpx.TimeoutException: + metrics.MDA_REQUEST_DURATION.labels(endpoint=endpoint_label, result="timeout").observe( + self._clock() - start + ) + logger.warning("MDA %s timeout after %.2fs", endpoint_label, self._clock() - start) + self._record_failure() + return MDAResult(ok=False, temp_fail=True, payload={}, status_code=0) + except httpx.HTTPError: + metrics.MDA_REQUEST_DURATION.labels(endpoint=endpoint_label, result="error").observe( + self._clock() - start + ) + logger.exception("MDA %s transport error", endpoint_label) + self._record_failure() + return MDAResult(ok=False, temp_fail=True, payload={}, status_code=0) + + elapsed = self._clock() - start + + # JSON decode is best-effort; some error bodies may be HTML. + try: + payload = response.json() if response.content else {} + except json.JSONDecodeError: + payload = {} + + status = response.status_code + if status == 200: + metrics.MDA_REQUEST_DURATION.labels(endpoint=endpoint_label, result="ok").observe( + elapsed + ) + 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" + 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: + self._record_failure() + else: + self._record_success() + return MDAResult(ok=False, temp_fail=temp, payload=payload, status_code=status) + + async def check_recipient(self, address: str) -> MDAResult: + """Ask the MDA whether a single recipient mailbox exists.""" + body = json.dumps({"addresses": [address]}, separators=(",", ":")).encode("utf-8") + return await self._post( + "inbound/mta/check/", + "application/json", + body, + metadata={}, + endpoint_label="check", + ) + + async def deliver( # noqa: PLR0913 + self, + *, + message: bytes, + sender: str, + original_recipients: list[str], + client_address: str | None, + client_port: str | None, + client_hostname: str | None, + client_helo: str | None, + ) -> MDAResult: + """Push the complete message to the MDA for synchronous delivery.""" + metadata = { + "sender": sender, + "original_recipients": list(original_recipients), + "client_address": client_address, + "client_port": client_port, + "client_hostname": client_hostname, + "client_helo": client_helo, + "size": str(len(message)), + } + return await self._post( + "inbound/mta/deliver/", + "message/rfc822", + message, + metadata=metadata, + endpoint_label="deliver", + ) diff --git a/src/mta-in/src/pymta/metrics.py b/src/mta-in/src/pymta/metrics.py new file mode 100644 index 00000000..65398318 --- /dev/null +++ b/src/mta-in/src/pymta/metrics.py @@ -0,0 +1,104 @@ +"""Prometheus metrics for the pymta server. + +The metrics HTTP endpoint is started from :mod:`pymta.server`. Each metric +intentionally has a low cardinality (no email addresses, no client IPs in +labels) to keep the time-series space bounded. +""" + +import logging + +from prometheus_client import Counter, Gauge, Histogram, start_http_server + +logger = logging.getLogger(__name__) + + +_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 + labelnames=("result",), +) + +SESSIONS_ACTIVE = Gauge( + f"{_METRICS_NAMESPACE}_sessions_active", + "Currently active SMTP sessions (post-PROXY, pre-close).", +) + +SESSIONS_PER_IP = Gauge( + f"{_METRICS_NAMESPACE}_sessions_per_ip", + "Distinct remote IPs currently holding at least one session.", +) + +SESSION_DURATION = Histogram( + f"{_METRICS_NAMESPACE}_session_duration_seconds", + "Wall-clock time from accept to close of an SMTP session.", + buckets=(0.05, 0.1, 0.5, 1, 5, 10, 30, 60, 120, 300, 600), +) + +COMMANDS_TOTAL = Counter( + f"{_METRICS_NAMESPACE}_commands_total", + "SMTP commands processed, by verb and outcome class (2xx/4xx/5xx).", + labelnames=("verb", "class"), +) + +RCPT_TOTAL = Counter( + f"{_METRICS_NAMESPACE}_rcpt_total", + "RCPT TO outcomes.", + labelnames=("result",), # accepted | rejected_perm | rejected_temp +) + +MESSAGES_TOTAL = Counter( + f"{_METRICS_NAMESPACE}_messages_total", + "End-of-DATA delivery outcomes.", + labelnames=("result",), # delivered | rejected_perm | rejected_temp +) + +MESSAGE_BYTES = Histogram( + f"{_METRICS_NAMESPACE}_message_bytes", + "Size of received messages in bytes.", + buckets=(1024, 10_000, 100_000, 500_000, 1_000_000, 5_000_000, 10_000_000, 50_000_000), +) + +MDA_REQUEST_DURATION = Histogram( + f"{_METRICS_NAMESPACE}_mda_request_duration_seconds", + "Latency of MDA API calls.", + labelnames=( + "endpoint", + "result", + ), # endpoint: check|deliver, result: ok|http_5xx|timeout|error + buckets=(0.005, 0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10, 30), +) + +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 +) + + +SECURITY_REJECTIONS = Counter( + f"{_METRICS_NAMESPACE}_security_rejections_total", + "Requests rejected by an explicit hardening check, by reason.", + labelnames=("reason",), + # 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 +) + + +def start_metrics_server(host: str, port: int) -> None: + """Start the Prometheus exposition HTTP server in a daemon thread. + + ``prometheus_client.start_http_server`` already spawns a background + thread, so this just adds a log line. Pass ``port=0`` to skip. + """ + if port <= 0: + logger.info("Prometheus metrics endpoint disabled (PYMTA_METRICS_PORT=0)") + return + start_http_server(port, addr=host) + logger.info("Prometheus metrics endpoint listening on %s:%d/metrics", host, port) diff --git a/src/mta-in/src/pymta/server.py b/src/mta-in/src/pymta/server.py new file mode 100644 index 00000000..509d6bf1 --- /dev/null +++ b/src/mta-in/src/pymta/server.py @@ -0,0 +1,107 @@ +"""pymta entrypoint. + +Run with ``python -m pymta.server``. Starts: + +* the Prometheus exposition HTTP server (in a daemon thread), +* the SMTP listener (asyncio), + +and exits on SIGINT/SIGTERM with an orderly shutdown that closes the listener +and waits for in-flight sessions to finish. +""" + +from __future__ import annotations + +import asyncio +import logging +import signal +import sys + +from . import metrics, settings +from .controller import HardenedController +from .handler import InboundHandler +from .limits import IPGate +from .mda_async import MDAClient + +logger = logging.getLogger(__name__) + + +def _configure_logging() -> None: + logging.basicConfig( + level=getattr(logging, settings.PYMTA_LOG_LEVEL, logging.INFO), + format="%(asctime)s %(name)s %(levelname)s %(message)s", + stream=sys.stdout, + ) + + +async def _serve() -> None: + mda_client = MDAClient() + try: + await mda_client.start() + handler = InboundHandler(mda_client) + ip_gate = IPGate( + max_total=settings.PYMTA_MAX_SESSIONS_TOTAL, + max_per_ip=settings.PYMTA_MAX_SESSIONS_PER_IP, + max_per_ip_per_minute=settings.PYMTA_MAX_SESSIONS_PER_IP_PER_MINUTE, + ) + + controller = HardenedController( + handler, + ip_gate=ip_gate, + hostname=settings.PYMTA_SMTP_HOST, + port=settings.PYMTA_SMTP_PORT, + loop=asyncio.get_running_loop(), + ) + + # ``begin()`` is sync but only schedules; for a running loop we want to + # await ``_create_server`` directly so the loop drives it cleanly. + server = await controller._create_server() # noqa: SLF001 + controller.server = server + + logger.info( + "pymta SMTP listening on %s:%d (hostname=%s, proxy_protocol=%s, size=%d)", + settings.PYMTA_SMTP_HOST, + settings.PYMTA_SMTP_PORT, + settings.PYMTA_HOSTNAME, + settings.PYMTA_ENABLE_PROXY_PROTOCOL, + settings.MAX_INCOMING_EMAIL_SIZE, + ) + + stop = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, stop.set) + except NotImplementedError: + # Windows / restricted environments — no signal handler support. + pass + + try: + await stop.wait() + finally: + logger.info("shutting down pymta SMTP listener") + server.close() + try: + await asyncio.wait_for( + server.wait_closed(), timeout=settings.PYMTA_SHUTDOWN_TIMEOUT + ) + except TimeoutError: + logger.warning( + "graceful shutdown deadline (%ds) exceeded; in-flight " + "sessions abandoned", + settings.PYMTA_SHUTDOWN_TIMEOUT, + ) + finally: + await mda_client.close() + + +def main() -> None: + _configure_logging() + metrics.start_metrics_server(settings.PYMTA_METRICS_HOST, settings.PYMTA_METRICS_PORT) + try: + asyncio.run(_serve()) + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() diff --git a/src/mta-in/src/pymta/settings.py b/src/mta-in/src/pymta/settings.py new file mode 100644 index 00000000..62187a29 --- /dev/null +++ b/src/mta-in/src/pymta/settings.py @@ -0,0 +1,175 @@ +"""Environment-variable-driven settings for the pymta server.""" + +import os + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.environ.get(name, "").strip().lower() + if raw in ("1", "true", "yes", "on"): + return True + if raw in ("0", "false", "no", "off"): + return False + return default + + +def _env_int(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + return int(raw) + except ValueError as exc: + raise ValueError(f"Environment variable {name} must be an integer, got {raw!r}") from exc + + +def _env_str(name: str, default: str) -> str: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return raw + + +# --------------------------------------------------------------------------- +# 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) + +# 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) + + +# --------------------------------------------------------------------------- +# SMTP listener +# --------------------------------------------------------------------------- + +PYMTA_SMTP_HOST = _env_str("PYMTA_SMTP_HOST", "0.0.0.0") # noqa: S104 +PYMTA_SMTP_PORT = _env_int("PYMTA_SMTP_PORT", 25) + +# Banner / Received-header hostname. Matches Postfix's `myhostname`. +PYMTA_HOSTNAME = _env_str("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") + + +# --------------------------------------------------------------------------- +# Message-shape limits (security-critical) +# --------------------------------------------------------------------------- + +# Total RFC822 message size cap. Mirrors Postfix `message_size_limit`. +MAX_INCOMING_EMAIL_SIZE = _env_int("MAX_INCOMING_EMAIL_SIZE", 10_240_000) + +# RCPT TO per SMTP transaction. Mirrors Postfix `smtpd_recipient_limit=100`. +PYMTA_MAX_RECIPIENTS = _env_int("PYMTA_MAX_RECIPIENTS", 100) + +# Envelopes per TCP connection (one envelope = MAIL FROM..DATA cycle). +PYMTA_MAX_ENVELOPES_PER_CONNECTION = _env_int("PYMTA_MAX_ENVELOPES_PER_CONNECTION", 10) + +# 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) + + +# --------------------------------------------------------------------------- +# Timeouts & connection caps +# --------------------------------------------------------------------------- + +# Per-command idle timeout (seconds). Postfix default is 300 s; we tighten. +PYMTA_COMMAND_TIMEOUT = _env_int("PYMTA_COMMAND_TIMEOUT", 120) + +# 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) + +# 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) + +# Per-IP concurrent SMTP sessions. 0 disables the cap. +PYMTA_MAX_SESSIONS_PER_IP = _env_int("PYMTA_MAX_SESSIONS_PER_IP", 100) + +# Process-wide concurrent SMTP sessions. 0 disables. +PYMTA_MAX_SESSIONS_TOTAL = _env_int("PYMTA_MAX_SESSIONS_TOTAL", 1000) + +# Per-IP new-session rate, measured in a rolling 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) + +# 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) + +# 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) + + +# --------------------------------------------------------------------------- +# ESMTP feature toggles +# --------------------------------------------------------------------------- + +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) + + +# --------------------------------------------------------------------------- +# STARTTLS (opportunistic). When both files are set, STARTTLS is advertised. +# +# 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 +# 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 +# ssl module accepts a single combined PEM that way. Postfix-compatible. +# --------------------------------------------------------------------------- + +PYMTA_TLS_CERT_FILE = _env_str("PYMTA_TLS_CERT_FILE", "") +PYMTA_TLS_KEY_FILE = _env_str("PYMTA_TLS_KEY_FILE", "") + +# Postfix-style fallback. Only the first path in the comma-separated list is +# used (Postfix supports multiple for RSA+ECDSA dual-cert; pymta picks the +# first chain and lets the operator add SNI later if needed). +_chain_files = _env_str("STARTTLS_CHAIN_FILES", "") +if _chain_files and not PYMTA_TLS_CERT_FILE and not PYMTA_TLS_KEY_FILE: + _first_chain = _chain_files.split(",", 1)[0].strip() + PYMTA_TLS_CERT_FILE = _first_chain + PYMTA_TLS_KEY_FILE = _first_chain + + +# --------------------------------------------------------------------------- +# Prometheus metrics HTTP endpoint +# --------------------------------------------------------------------------- + +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) + + +# --------------------------------------------------------------------------- +# Logging +# --------------------------------------------------------------------------- + +PYMTA_LOG_LEVEL = _env_str("PYMTA_LOG_LEVEL", "INFO").upper() diff --git a/src/mta-in/src/pymta/smtp_protocol.py b/src/mta-in/src/pymta/smtp_protocol.py new file mode 100644 index 00000000..fd65477d --- /dev/null +++ b/src/mta-in/src/pymta/smtp_protocol.py @@ -0,0 +1,133 @@ +"""Hardened :class:`aiosmtpd.smtp.SMTP` subclass. + +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 + 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 + unconditionally to prevent address enumeration. +* ``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 +enabled, the check is deferred to the ``handle_PROXY`` hook so it sees the +real client address rather than the load-balancer's IP. +""" + +from __future__ import annotations + +import contextlib +import logging +import time + +from aiosmtpd.smtp import SMTP as BaseSMTP + +from . import metrics +from .limits import IPGate, TooManyConnections + +logger = logging.getLogger(__name__) + + +class HardenedSMTP(BaseSMTP): + """SMTP subclass that locks down VRFY/EXPN/AUTH and applies the IP gate.""" + + def __init__(self, *args, ip_gate: IPGate | None = None, **kwargs): + super().__init__(*args, **kwargs) + self._ip_gate: IPGate | None = ip_gate + # Track whether we are currently holding a slot in the gate so the + # release path runs at most once. + self._gate_held_ip: str | None = None + self._gate_started: float | None = None + + # ----------------------------------------------------------- verb lockdown + async def smtp_VRFY(self, arg: str) -> None: + await self.push("252 2.1.5 Cannot VRFY user; try RCPT to verify") + + async def smtp_EXPN(self, arg: str) -> None: + await self.push("502 5.7.0 EXPN disabled") + + async def smtp_AUTH(self, arg: str) -> None: + # AUTH is never advertised (no authenticator wired, auth_require_tls + # defaults to True), but reply explicitly anyway so a misconfigured + # scanner cannot mistake an absent reply for acceptance. + await self.push("502 5.7.0 AUTH not supported on inbound port 25") + + async def smtp_HELP(self, arg: str) -> None: + # Default aiosmtpd HELP enumerates implemented verbs (mild info leak). + await self.push("214 2.0.0 See https://www.rfc-editor.org/rfc/rfc5321") + + # ------------------------------------------------------------ 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, + so we gate before the SMTP dialogue starts. + * **PROXY protocol enabled** — 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. + """ + if self._ip_gate is not None and self._proxy_timeout is None: + if not await self._acquire_gate(self._wire_peer_ip()): + return + try: + await super()._handle_client() + finally: + await self._release_gate() + + async def acquire_gate_post_proxy(self, real_ip: str) -> bool: + """Acquire the gate using the IP parsed from a PROXY-protocol header. + + Called from :meth:`pymta.handler.InboundHandler.handle_PROXY`. Returns + ``True`` on success; on refusal sends 421 and closes the socket. + """ + if self._ip_gate is None: + return True + 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 + try: + await self._ip_gate._try_acquire(ip) # noqa: SLF001 + except TooManyConnections as exc: + metrics.CONNECTIONS_TOTAL.labels(result=f"rejected_{exc.scope}").inc() + metrics.DISCONNECTS_421.labels(reason=f"gate_{exc.scope}").inc() + logger.info("connection from %s refused: %s cap reached", ip, exc.scope) + with contextlib.suppress(OSError, ConnectionError): + await self.push("421 4.7.0 Too many connections, try again later") + # Best-effort drain so the 421 actually makes it out before the + # RST closes the socket. + if self._writer is not None: + await self._writer.drain() + if self.transport is not None: + self.transport.close() + return False + self._gate_held_ip = ip + self._gate_started = time.monotonic() + metrics.CONNECTIONS_TOTAL.labels(result="accepted").inc() + return True + + async def _release_gate(self) -> None: + if self._gate_held_ip is None or self._ip_gate is None: + return + ip, self._gate_held_ip = self._gate_held_ip, None + if self._gate_started is not None: + metrics.SESSION_DURATION.observe(time.monotonic() - self._gate_started) + self._gate_started = None + # _handle_client's finally runs in the event loop, so awaiting the + # release is safe and avoids the fire-and-forget bookkeeping leak we + # would have with create_task during shutdown. + await self._ip_gate._release(ip) # noqa: SLF001 + + def _wire_peer_ip(self) -> str: + peer = getattr(self.session, "peer", None) if self.session else None + if peer: + return str(peer[0]) + # All sessions without a wire peer collapse into one bucket; log so a + # spike here doesn't go invisible. + logger.warning("session has no transport peer; using 'unknown' bucket") + return "unknown" diff --git a/src/mta-in/tests/conftest.py b/src/mta-in/tests/conftest.py index 6a2bacb2..fc7ea39c 100644 --- a/src/mta-in/tests/conftest.py +++ b/src/mta-in/tests/conftest.py @@ -19,6 +19,17 @@ logger = logging.getLogger(__name__) MDA_API_SECRET = os.getenv("MDA_API_SECRET") MTA_HOST = os.getenv("MTA_HOST") +MTA_PORT = int(os.getenv("MTA_PORT", "25")) + +# When MTA_METRICS_URL is set (only by the pymta test runner) the metrics +# tests in tests/test_metrics.py become exercisable. The Postfix-based +# implementation has no Prometheus endpoint, so those tests skip on it. +MTA_METRICS_URL = os.getenv("MTA_METRICS_URL") + +# Tag tests with the implementation under test, exposed through the +# `mta_impl` fixture below. Useful for skipping the few tests that assert +# implementation-specific behaviour (e.g. metrics-shape). +MTA_IMPL = os.getenv("MTA_IMPL", "postfix") class MockAPIServer: @@ -153,31 +164,46 @@ def mock_api_server(): server.stop() -@pytest.fixture -def smtp_client(): - # Wait for Postfix to be ready - max_retries = 100 - for attempt in range(max_retries): +def _wait_for_mta(host: str, port: int, retries: int = 100) -> None: + for attempt in range(retries): try: - # First check if port is open with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.connect((MTA_HOST, 25)) - - # Then try SMTP connection - client = smtplib.SMTP(MTA_HOST, 25) - logger.info("SMTP connection established") - break - except (ConnectionRefusedError, smtplib.SMTPConnectError, socket.error) as e: - if attempt == max_retries - 1: + s.settimeout(1) + s.connect((host, port)) + return + except (ConnectionRefusedError, OSError) as e: + if attempt == retries - 1: raise if attempt % 20 == 0: - logger.warning( - f"SMTP connection attempt {attempt + 1} failed ({str(e)}), retrying in 1s..." - ) + logger.warning("SMTP port %s:%s not ready (%s); retrying...", host, port, e) time.sleep(0.1) + +@pytest.fixture +def smtp_client(): + _wait_for_mta(MTA_HOST, MTA_PORT) + client = smtplib.SMTP(MTA_HOST, MTA_PORT) + logger.info("SMTP connection established to %s:%s", MTA_HOST, MTA_PORT) yield client try: client.quit() except smtplib.SMTPServerDisconnected: pass + + +@pytest.fixture +def mta_impl() -> str: + """Identifier for the implementation under test: ``postfix`` or ``pymta``.""" + return MTA_IMPL + + +@pytest.fixture +def mta_address() -> tuple[str, int]: + """(host, port) of the inbound MTA under test.""" + return (MTA_HOST, MTA_PORT) + + +@pytest.fixture +def mta_metrics_url() -> str | None: + """URL to scrape for Prometheus metrics; None if unavailable.""" + return MTA_METRICS_URL diff --git a/src/mta-in/tests/test_address.py b/src/mta-in/tests/test_address.py new file mode 100644 index 00000000..82e2dc30 --- /dev/null +++ b/src/mta-in/tests/test_address.py @@ -0,0 +1,241 @@ +"""Unit tests for :mod:`pymta.address`. + +Exercises every rejection branch in ``validate_envelope_address`` plus the +positive happy paths. Pure-stdlib — no Docker stack needed. +""" + +from __future__ import annotations + +import pytest + +from pymta.address import AddressError, strip_brackets, validate_envelope_address + + +def _validate(address: str, *, allow_empty: bool = False) -> str: + return validate_envelope_address( + address, allow_empty=allow_empty, max_local=64, max_domain=255 + ) + + +# --------------------------------------------------------------------------- +# strip_brackets +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("", "user@example.com"), + ("user@example.com", "user@example.com"), + ("", ""), + ("<>", ""), + ], +) +def test_strip_brackets(raw, expected): + assert strip_brackets(raw) == expected + + +# --------------------------------------------------------------------------- +# Happy paths +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address", + [ + "user@example.com", + "", + "user.name@example.com", + "user+tag@sub.example.com", + "a@b.co", + "user@xn--bcher-kva.example", # IDN A-label + "user@münchen.de", # raw UTF-8 domain (SMTPUTF8) + ], +) +def test_valid_addresses_accepted(address): + out = _validate(address) + assert "@" in out + + +def test_domain_is_lowercased(): + assert _validate("User@EXAMPLE.COM") == "User@example.com" + + +def test_local_part_case_is_preserved(): + # Per RFC 5321 §2.3.11 local-parts are case-sensitive on the wire; we + # leave the decision to the MDA's normalisation rules. + assert _validate("User.Name@example.com").startswith("User.Name@") + + +def test_null_sender_allowed_when_enabled(): + assert _validate("", allow_empty=True) == "" + assert _validate("<>", allow_empty=True) == "" + + +def test_null_sender_rejected_for_rcpt(): + with pytest.raises(AddressError) as exc: + _validate("", allow_empty=False) + assert exc.value.reason == "bad_address" + assert exc.value.smtp_code == 553 + + +# --------------------------------------------------------------------------- +# Residual / unbalanced angle brackets +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address", + [ + "<>", + "", + "extra", + ], +) +def test_residual_brackets_rejected(address): + with pytest.raises(AddressError) as exc: + _validate(address) + assert exc.value.reason == "bad_address" + assert exc.value.smtp_code == 501 + + +# --------------------------------------------------------------------------- +# Control characters / CRLF injection +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address", + [ + "user\r@example.com", + "user\n@example.com", + "user\x00@example.com", + "user\t@example.com", + "user @example.com", + "user%@example.com", + "user\x7f@example.com", + ], +) +def test_control_chars_rejected(address): + with pytest.raises(AddressError) as exc: + _validate(address) + assert exc.value.reason == "control_char" + assert exc.value.smtp_code == 501 + + +# --------------------------------------------------------------------------- +# Source routes (RFC 5321 §4.1.1.3) +# --------------------------------------------------------------------------- + + +def test_source_route_rejected(): + with pytest.raises(AddressError) as exc: + _validate("@host1.example,@host2.example:user@host3.example") + assert exc.value.reason == "source_route" + + +# --------------------------------------------------------------------------- +# Quoted local-parts +# --------------------------------------------------------------------------- + + +def test_quoted_local_part_rejected(): + # The space-in-quote form would be caught earlier by the control-char + # check; use a tame quoted form so we land on the quoted-local-part rule. + with pytest.raises(AddressError) as exc: + _validate('"weird"@example.com') + assert exc.value.reason == "bad_address" + assert exc.value.smtp_code == 553 + + +# --------------------------------------------------------------------------- +# Dot-placement in unquoted local-part +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address", + [ + ".user@example.com", + "user.@example.com", + "user..name@example.com", + ], +) +def test_bad_dot_placement_rejected(address): + with pytest.raises(AddressError) as exc: + _validate(address) + assert exc.value.reason == "bad_address" + + +# --------------------------------------------------------------------------- +# @-count, missing local/domain +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "address", + [ + "noatsign", + "two@@example.com", + "user@host@other", + "@example.com", # caught earlier as source_route — covered above + "user@", + ], +) +def test_malformed_at_or_missing_parts_rejected(address): + with pytest.raises(AddressError): + _validate(address) + + +# --------------------------------------------------------------------------- +# Length limits +# --------------------------------------------------------------------------- + + +def test_overlong_local_part_rejected(): + long_local = "a" * 65 + with pytest.raises(AddressError) as exc: + _validate(f"{long_local}@example.com") + assert exc.value.reason == "oversize_local" + + +def test_overlong_domain_rejected(): + # 64-char label * 4 + dots = ~260 chars (>255 total domain cap). + long_domain = ".".join(["a" * 60] * 5) + with pytest.raises(AddressError) as exc: + _validate(f"user@{long_domain}") + assert exc.value.reason == "oversize_domain" + + +def test_overlong_label_rejected(): + too_long_label = "a" * 64 # one octet over RFC 1035 §2.3.4 + with pytest.raises(AddressError) as exc: + _validate(f"user@{too_long_label}.example") + assert exc.value.reason == "bad_address" + + +# --------------------------------------------------------------------------- +# Domain shape +# --------------------------------------------------------------------------- + + +def test_address_literal_rejected_with_dedicated_reason(): + with pytest.raises(AddressError) as exc: + _validate("user@[192.0.2.1]") + assert exc.value.reason == "address_literal" + assert "literal" in exc.value.smtp_text.lower() + + +@pytest.mark.parametrize( + "domain", + [ + ".example.com", + "example.com.", + "example..com", + ".", + ], +) +def test_malformed_domain_rejected(domain): + with pytest.raises(AddressError): + _validate(f"user@{domain}") diff --git a/src/mta-in/tests/test_email_delivery.py b/src/mta-in/tests/test_email_delivery.py index 19ff3a79..9dea5f18 100644 --- a/src/mta-in/tests/test_email_delivery.py +++ b/src/mta-in/tests/test_email_delivery.py @@ -46,7 +46,9 @@ def test_simple_email_delivery(mock_api_server, smtp_client): assert not email["email"].is_multipart() body = email["email"].get_payload() - # TODO: why the \n ? + # smtplib converts the bare \n in the source string to the SMTP-mandatory + # CRLF (\r\n) during DATA transmission, so the body received by the MDA + # carries \r\n line endings — not the \n we put in the MIMEText source. assert body == "This is a test email\r\n" @@ -91,7 +93,9 @@ def test_simple_email_delivery_with_multiple_recipients(mock_api_server, smtp_cl assert not email["email"].is_multipart() body = email["email"].get_payload() - # TODO: why the \n ? + # smtplib converts the bare \n in the source string to the SMTP-mandatory + # CRLF (\r\n) during DATA transmission, so the body received by the MDA + # carries \r\n line endings — not the \n we put in the MIMEText source. assert body == "This is a test email\r\n" mock_api_server.received_emails = [] @@ -119,7 +123,9 @@ def test_simple_email_delivery_with_multiple_recipients(mock_api_server, smtp_cl assert not email["email"].is_multipart() body = email["email"].get_payload() - # TODO: why the \n ? + # smtplib converts the bare \n in the source string to the SMTP-mandatory + # CRLF (\r\n) during DATA transmission, so the body received by the MDA + # carries \r\n line endings — not the \n we put in the MIMEText source. assert body == "This is a test email\r\n" diff --git a/src/mta-in/tests/test_handler.py b/src/mta-in/tests/test_handler.py new file mode 100644 index 00000000..ea3ee223 --- /dev/null +++ b/src/mta-in/tests/test_handler.py @@ -0,0 +1,182 @@ +"""Unit tests for :mod:`pymta.handler`. + +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. +""" + +from __future__ import annotations + +import types + +import pytest + +from pymta import settings +from pymta.handler import ( + _RCPT_MISSES_ATTR, + _SOFT_ERRORS_ATTR, + InboundHandler, + NULL_SENDER_SENTINEL, +) +from pymta.mda_async import MDAResult + + +class _FakeMDA: + """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 + ) + + async def check_recipient(self, address: str) -> MDAResult: + return self.check_result + + +def _session(): + return types.SimpleNamespace(host_name=None, peer=("203.0.113.5", 12345)) + + +def _envelope(): + return types.SimpleNamespace( + mail_from=None, rcpt_tos=[], mail_options=[], rcpt_options=[], content=b"" + ) + + +def _handler(mda=None) -> InboundHandler: + return InboundHandler(mda or _FakeMDA()) + + +# --------------------------------------------------------------------------- +# MAIL SIZE= path bumps the soft-error counter on both rejection branches. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mail_size_bad_value_bumps_soft_errors(): + session, envelope = _session(), _envelope() + reply = await _handler().handle_MAIL( + None, session, envelope, "", ["SIZE=not-a-number"] + ) + assert reply.startswith("501") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +@pytest.mark.asyncio +async def test_mail_size_oversize_bumps_soft_errors(): + session, envelope = _session(), _envelope() + too_big = settings.MAX_INCOMING_EMAIL_SIZE + 1 + reply = await _handler().handle_MAIL( + None, session, envelope, "", [f"SIZE={too_big}"] + ) + assert reply.startswith("552") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +# --------------------------------------------------------------------------- +# DATA negative paths bump the soft-error counter. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_data_nul_byte_bumps_soft_errors(): + session, envelope = _session(), _envelope() + envelope.content = b"Subject: x\r\n\r\nhello\x00world\r\n" + reply = await _handler().handle_DATA(None, session, envelope) + assert reply.startswith("554") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +@pytest.mark.asyncio +async def test_data_oversize_bumps_soft_errors(): + session, envelope = _session(), _envelope() + envelope.content = b"x" * (settings.MAX_INCOMING_EMAIL_SIZE + 10) + reply = await _handler().handle_DATA(None, session, envelope) + assert reply.startswith("552") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +@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) + reply = await _handler().handle_DATA(None, session, envelope) + assert reply.startswith("451") + assert getattr(session, _SOFT_ERRORS_ATTR) == 1 + + +# --------------------------------------------------------------------------- +# RCPT miss counter / dedicated cutoff (S3). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +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 + ) + 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"", [] + ) + assert reply.startswith("550"), reply + # Third miss hits the per-session cap and forces 421. + reply = await handler.handle_RCPT( + None, session, envelope, "", [] + ) + assert reply.startswith("421") + assert getattr(session, _RCPT_MISSES_ATTR) == 3 + + +@pytest.mark.asyncio +async def test_rcpt_existence_does_not_increment_miss_counter(): + mda = _FakeMDA( + check_result=MDAResult( + ok=True, + temp_fail=False, + payload={"hit@example.com": True}, + status_code=200, + ) + ) + handler, session, envelope = _handler(mda), _session(), _envelope() + reply = await handler.handle_RCPT( + None, session, envelope, "", [] + ) + assert reply.startswith("250") + assert getattr(session, _RCPT_MISSES_ATTR, 0) == 0 + + +# --------------------------------------------------------------------------- +# Hard-error budget cutoff still fires from the existing gate. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +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, "", [] + ) + assert reply.startswith("421") + + +# --------------------------------------------------------------------------- +# Null sender survives the round-trip via the sentinel. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_null_sender_round_trip_via_sentinel(): + session, envelope = _session(), _envelope() + reply = await _handler().handle_MAIL(None, session, envelope, "<>", []) + assert reply.startswith("250") + assert envelope.mail_from == NULL_SENDER_SENTINEL diff --git a/src/mta-in/tests/test_limits.py b/src/mta-in/tests/test_limits.py new file mode 100644 index 00000000..2c650ba3 --- /dev/null +++ b/src/mta-in/tests/test_limits.py @@ -0,0 +1,195 @@ +"""Unit tests for :class:`pymta.limits.IPGate`. + +Unlike the rest of the suite in this directory, these tests do NOT need a +running MTA — they exercise the gate object directly. They run via the same +``test-mta-in-py`` target but skip the SMTP integration fixtures. +""" + +from __future__ import annotations + +import pytest + +from pymta.limits import IPGate, TooManyConnections + + +class _FakeClock: + """Manually-advanced monotonic clock for deterministic rate-window tests.""" + + def __init__(self, start: float = 1000.0): + self.now = start + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +# --------------------------------------------------------------------------- +# Concurrent-cap behaviour (existing semantics). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_global_cap_blocks_when_total_reached(): + gate = IPGate(max_total=2, max_per_ip=0) + await gate._try_acquire("1.1.1.1") + await gate._try_acquire("2.2.2.2") + with pytest.raises(TooManyConnections) as exc: + await gate._try_acquire("3.3.3.3") + assert exc.value.scope == "global" + await gate._release("1.1.1.1") + await gate._release("2.2.2.2") + + +@pytest.mark.asyncio +async def test_per_ip_cap_blocks_same_ip_only(): + gate = IPGate(max_total=0, max_per_ip=2) + await gate._try_acquire("1.1.1.1") + await gate._try_acquire("1.1.1.1") + with pytest.raises(TooManyConnections) as exc: + await gate._try_acquire("1.1.1.1") + assert exc.value.scope == "per_ip" + # A different IP is still admitted. + await gate._try_acquire("2.2.2.2") + await gate._release("1.1.1.1") + await gate._release("1.1.1.1") + await gate._release("2.2.2.2") + + +@pytest.mark.asyncio +async def test_release_frees_slot_for_same_ip(): + gate = IPGate(max_total=0, max_per_ip=1) + await gate._try_acquire("1.1.1.1") + with pytest.raises(TooManyConnections): + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + # Slot freed — next acquire from the same IP succeeds. + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + + +@pytest.mark.asyncio +async def test_zero_disables_concurrency_caps(): + gate = IPGate(max_total=0, max_per_ip=0) + # Loopback test harness traffic comes from one IP; we must not throttle it. + for _ in range(50): + await gate._try_acquire("127.0.0.1") + for _ in range(50): + await gate._release("127.0.0.1") + + +@pytest.mark.asyncio +async def test_rate_cap_blocks_after_quota_in_window(): + clock = _FakeClock() + gate = IPGate(max_total=0, max_per_ip=0, max_per_ip_per_minute=3, clock=clock) + # Three quick session acquires from one IP — all release immediately so the + # concurrent cap can't be the thing blocking us; only the rate cap is. + for _ in range(3): + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + with pytest.raises(TooManyConnections) as exc: + await gate._try_acquire("1.1.1.1") + assert exc.value.scope == "per_ip_rate" + + +@pytest.mark.asyncio +async def test_rate_window_resets_after_60_seconds(): + clock = _FakeClock() + gate = IPGate(max_total=0, max_per_ip=0, max_per_ip_per_minute=2, clock=clock) + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + with pytest.raises(TooManyConnections): + await gate._try_acquire("1.1.1.1") + # Window closes — fresh budget. + clock.advance(60.1) + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + await gate._try_acquire("1.1.1.1") + + +@pytest.mark.asyncio +async def test_rate_cap_is_per_ip_not_global(): + clock = _FakeClock() + gate = IPGate(max_total=0, max_per_ip=0, max_per_ip_per_minute=2, clock=clock) + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + # A second IP gets its own bucket — must not be tarred by 1.1.1.1's spend. + await gate._try_acquire("2.2.2.2") + await gate._release("2.2.2.2") + + +@pytest.mark.asyncio +async def test_rate_cap_disabled_when_zero(): + clock = _FakeClock() + gate = IPGate(max_total=0, max_per_ip=0, max_per_ip_per_minute=0, clock=clock) + for _ in range(50): + await gate._try_acquire("1.1.1.1") + await gate._release("1.1.1.1") + # And the rate-tracking dict stays empty so loopback dev/test isn't + # paying memory for a feature it never uses. + assert gate._rate_per_ip == {} + + +@pytest.mark.asyncio +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 + limits._RATE_PRUNE_EVERY = 10 + try: + gate = IPGate(max_total=0, max_per_ip=0, max_per_ip_per_minute=1, clock=clock) + for i in range(9): + await gate._try_acquire(f"10.0.0.{i}") + await gate._release(f"10.0.0.{i}") + assert len(gate._rate_per_ip) == 9 + # All previous windows expire. + clock.advance(61.0) + # The 10th acquire triggers the prune sweep. + await gate._try_acquire("10.0.0.99") + await gate._release("10.0.0.99") + # Only the most recent entry survives; all stale ones are gone. + assert set(gate._rate_per_ip.keys()) == {"10.0.0.99"} + finally: + limits._RATE_PRUNE_EVERY = original + + +# --------------------------------------------------------------------------- +# Rate cap interacts cleanly with the concurrent caps. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_global_cap_takes_precedence_over_rate_cap(): + # When both would fire, the cheaper-to-evaluate global check should win + # so we don't bother accounting against the rate bucket for a session we + # were never going to admit anyway. + clock = _FakeClock() + gate = IPGate(max_total=1, max_per_ip=0, max_per_ip_per_minute=10, clock=clock) + await gate._try_acquire("1.1.1.1") + with pytest.raises(TooManyConnections) as exc: + await gate._try_acquire("2.2.2.2") + assert exc.value.scope == "global" + # The refused acquire must not have consumed 2.2.2.2's rate budget. + assert "2.2.2.2" not in gate._rate_per_ip + await gate._release("1.1.1.1") + + +@pytest.mark.asyncio +async def test_per_ip_concurrent_cap_takes_precedence_over_rate_cap(): + clock = _FakeClock() + gate = IPGate(max_total=0, max_per_ip=1, max_per_ip_per_minute=10, clock=clock) + await gate._try_acquire("1.1.1.1") + with pytest.raises(TooManyConnections) as exc: + await gate._try_acquire("1.1.1.1") + assert exc.value.scope == "per_ip" + # Only the first (admitted) acquire should have been billed to the bucket. + assert gate._rate_per_ip["1.1.1.1"][0] == 1 + await gate._release("1.1.1.1") diff --git a/src/mta-in/tests/test_mda_async.py b/src/mta-in/tests/test_mda_async.py new file mode 100644 index 00000000..b09fa10f --- /dev/null +++ b/src/mta-in/tests/test_mda_async.py @@ -0,0 +1,231 @@ +"""Unit tests for :class:`pymta.mda_async.MDAClient`. + +Stubs ``httpx.AsyncClient.post`` to drive every code path (timeout, transport +error, 5xx, 4xx, 200) and to verify the circuit breaker opens / resets as +expected. No real HTTP traffic, no Docker stack. +""" + +from __future__ import annotations + +import httpx +import pytest + +from pymta.mda_async import MDAClient + + +class _FakeClock: + def __init__(self, start: float = 1000.0): + self.now = start + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +class _StubAsyncClient: + """Stand-in for ``httpx.AsyncClient`` driven by a script of responses.""" + + def __init__(self, script): + self.script = list(script) + self.calls = 0 + + async def post(self, url, content=None, headers=None): + self.calls += 1 + action = self.script.pop(0) if self.script else None + if isinstance(action, Exception): + raise action + return action + + async def aclose(self): + pass + + +def _resp(status_code: int, body: bytes = b'{"ok": true}'): + return httpx.Response(status_code=status_code, content=body) + + +def _new_client(*, secret: str = "x" * 32, threshold: int = 3, cooldown: int = 30): + """Construct an MDAClient wired to fakes — no settings module mutation.""" + clock = _FakeClock() + client = MDAClient( + base_url="https://mda.example.invalid/api/", + secret=secret, + timeout=5, + breaker_threshold=threshold, + breaker_cooldown=cooldown, + clock=clock, + ) + return client, clock + + +# --------------------------------------------------------------------------- +# Single-shot result classification +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_timeout_returns_temp_fail(): + client, _ = _new_client() + client._client = _StubAsyncClient([httpx.TimeoutException("slow")]) + result = await client.check_recipient("user@example.com") + assert result.ok is False + assert result.temp_fail is True + assert result.status_code == 0 + + +@pytest.mark.asyncio +async def test_transport_error_returns_temp_fail(): + client, _ = _new_client() + client._client = _StubAsyncClient([httpx.ConnectError("no route")]) + result = await client.check_recipient("user@example.com") + assert result.temp_fail is True + + +@pytest.mark.asyncio +async def test_5xx_returns_temp_fail(): + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(503, b'{"detail":"upstream"}')]) + result = await client.check_recipient("user@example.com") + assert result.temp_fail is True + assert result.status_code == 503 + + +@pytest.mark.asyncio +async def test_4xx_returns_perm_fail(): + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(404, b'{"detail":"no"}')]) + result = await client.check_recipient("user@example.com") + assert result.ok is False + assert result.temp_fail is False + assert result.status_code == 404 + + +@pytest.mark.asyncio +async def test_200_returns_ok_with_payload(): + client, _ = _new_client() + client._client = _StubAsyncClient([_resp(200, b'{"user@example.com": true}')]) + result = await client.check_recipient("user@example.com") + assert result.ok is True + assert result.payload == {"user@example.com": True} + + +# --------------------------------------------------------------------------- +# Circuit breaker +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_breaker_opens_after_threshold_consecutive_failures(): + client, clock = _new_client(threshold=3, cooldown=30) + client._client = _StubAsyncClient( + [httpx.TimeoutException("a"), httpx.TimeoutException("b"), httpx.TimeoutException("c")] + ) + 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. + stub = client._client + result = await client.check_recipient("a@b") + assert result.temp_fail is True + assert stub.calls == 3, "breaker did not short-circuit" + + +@pytest.mark.asyncio +async def test_breaker_closes_after_cooldown(): + client, clock = _new_client(threshold=2, cooldown=30) + client._client = _StubAsyncClient( + [httpx.TimeoutException("a"), httpx.TimeoutException("b"), _resp(200)] + ) + await client.check_recipient("a@b") + await client.check_recipient("a@b") + # Breaker open — fast-fail. + await client.check_recipient("a@b") + assert client._open_until is not None + # Advance past cooldown — next call probes the network. + clock.advance(31.0) + result = await client.check_recipient("a@b") + assert result.ok is True + assert client._open_until is None + assert client._consecutive_failures == 0 + + +@pytest.mark.asyncio +async def test_success_resets_failure_counter(): + client, _ = _new_client(threshold=10) + client._client = _StubAsyncClient( + [httpx.TimeoutException("a"), httpx.TimeoutException("b"), _resp(200)] + ) + await client.check_recipient("a@b") + await client.check_recipient("a@b") + assert client._consecutive_failures == 2 + await client.check_recipient("a@b") + assert client._consecutive_failures == 0 + + +@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 + # liveness signal worth tripping the breaker. + client, _ = _new_client(threshold=2) + client._client = _StubAsyncClient([_resp(404), _resp(404), _resp(404)]) + for _ in range(3): + await client.check_recipient("a@b") + assert client._consecutive_failures == 0 + assert client._open_until is None + + +@pytest.mark.asyncio +async def test_breaker_disabled_when_threshold_zero(): + client, _ = _new_client(threshold=0) + client._client = _StubAsyncClient([httpx.TimeoutException("a")] * 5) + for _ in range(5): + await client.check_recipient("a@b") + assert client._open_until is None + + +# --------------------------------------------------------------------------- +# Credential / URL hygiene warnings (S1 + S2). +# --------------------------------------------------------------------------- + + +def test_non_local_http_url_logs_warning(caplog): + caplog.set_level("WARNING") + MDAClient(base_url="http://mda.example.com/api/", secret="x" * 32) + assert any("plaintext" in rec.message for rec in caplog.records) + + +def test_short_secret_logs_warning(caplog): + caplog.set_level("WARNING") + MDAClient(base_url="https://mda.example.com/api/", secret="too-short") + assert any("MDA_API_SECRET" in rec.message for rec in caplog.records) + + +def test_local_http_url_is_silent(caplog): + caplog.set_level("WARNING") + MDAClient(base_url="http://127.0.0.1:8000/api/", secret="x" * 32) + assert not any("plaintext" 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"} + ) + decoded = jwt.decode(token, client.secret, algorithms=["HS256"]) + # The real exp must be in the future, not 0. + assert decoded["exp"] != 0 + # The real body_hash must be the sha256 of `body`, not "deadbeef". + assert decoded["body_hash"] != "deadbeef" + # Sender (non-conflicting metadata) survives. + assert decoded["sender"] == "u@x" diff --git a/src/mta-in/tests/test_metrics.py b/src/mta-in/tests/test_metrics.py new file mode 100644 index 00000000..f09af7d6 --- /dev/null +++ b/src/mta-in/tests/test_metrics.py @@ -0,0 +1,86 @@ +"""Prometheus metrics tests — pymta-only. + +Skipped automatically when ``MTA_METRICS_URL`` is not set (i.e. when running +against the Postfix implementation, which has no Prometheus endpoint). +""" + +import logging +import smtplib +import urllib.request +from email.mime.text import MIMEText + +import pytest + +logger = logging.getLogger(__name__) + + +def _scrape(url: str) -> str: + with urllib.request.urlopen(url, timeout=5) as resp: + return resp.read().decode("utf-8", errors="replace") + + +def _metric_value(scrape_text: str, prefix: str) -> float: + """Sum every series that begins with ``prefix``, return the total. + + A ``prefix`` like ``pymta_messages_total{result="delivered"}`` matches a + single series. ``pymta_messages_total`` (no label selector) matches + every series of that metric. + """ + total = 0.0 + for line in scrape_text.splitlines(): + if line.startswith("#") or not line.strip(): + continue + # Format: `name{labels} value [timestamp]` or `name value` + if line.startswith(prefix): + parts = line.split() + # 2 tokens → name value; 3+ tokens → name value timestamp. + value = parts[-2] if len(parts) >= 3 else parts[-1] + try: + total += float(value) + except ValueError: + continue + return total + + +@pytest.fixture +def metrics_url(mta_metrics_url): + if not mta_metrics_url: + pytest.skip("MTA_METRICS_URL not set (only the pymta image exposes metrics)") + return mta_metrics_url + + +def test_metrics_endpoint_reachable(metrics_url): + text = _scrape(metrics_url) + assert "pymta_connections_total" in text + assert "pymta_messages_total" in text + + +def test_delivery_increments_messages_total(metrics_url, mock_api_server, smtp_client): + mock_api_server.add_mailbox("metrics-test@example.com") + + before = _metric_value(_scrape(metrics_url), 'pymta_messages_total{result="delivered"}') + + msg = MIMEText("metrics body\n") + msg["From"] = "sender@example.com" + msg["To"] = "metrics-test@example.com" + msg["Subject"] = "metrics" + smtp_client.send_message(msg) + mock_api_server.wait_for_email() + + after = _metric_value(_scrape(metrics_url), 'pymta_messages_total{result="delivered"}') + assert after >= before + 1, (before, after) + + +def test_rcpt_rejected_increments_rcpt_total(metrics_url, mock_api_server, smtp_client): + before = _metric_value(_scrape(metrics_url), 'pymta_rcpt_total{result="rejected_perm"}') + + # An RCPT that the MDA does not know about → permanent reject. + msg = MIMEText("body\n") + msg["From"] = "sender@example.com" + msg["To"] = "unknown-metrics@example.com" + msg["Subject"] = "rejected" + with pytest.raises(smtplib.SMTPRecipientsRefused): + smtp_client.send_message(msg) + + after = _metric_value(_scrape(metrics_url), 'pymta_rcpt_total{result="rejected_perm"}') + assert after >= before + 1, (before, after) diff --git a/src/mta-in/tests/test_security.py b/src/mta-in/tests/test_security.py new file mode 100644 index 00000000..00ffe9d6 --- /dev/null +++ b/src/mta-in/tests/test_security.py @@ -0,0 +1,381 @@ +"""Security / hardening tests, runnable against both the Postfix milter and +the pure-Python pymta implementations. + +These exercise the attack classes the inbound MTA is most exposed to on the +public internet: + +* SMTP smuggling (bare-LF / bare-CR EOD smuggling, RFC 5321 §2.3.8) +* CRLF / control-character injection in envelope addresses +* NUL bytes in DATA +* Source routes (RFC 5321 §4.1.1.3) +* VRFY / EXPN information disclosure +* AUTH leakage on port 25 +* Overlong local-parts / domains +* Pre-DATA SIZE oversize announcement +* Pipelined-command-before-banner (smuggling helper) +* Hard error limit / command flood +* Line-length cap + +Where the two implementations have different SMTP codes (Postfix uses some +504/521 codes pymta uses 502 for), tests assert on the response *class* +(2xx/4xx/5xx) instead of an exact code. +""" + +import logging +import os +import socket + +import pytest + +logger = logging.getLogger(__name__) + +MTA_HOST = os.getenv("MTA_HOST") +MTA_PORT = int(os.getenv("MTA_PORT", "25")) + + +def _raw_session(timeout: float = 5): + """Open a raw socket to the MTA, swallow the banner, and return it.""" + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(timeout) + s.connect((MTA_HOST, MTA_PORT)) + banner = _read_reply(s) + assert banner.startswith(b"220"), banner + return s + + +def _read_reply(s: socket.socket, max_bytes: int = 65536) -> bytes: + """Read one full SMTP reply, handling multi-line continuations. + + SMTP multi-line replies use ``xxx-`` on every line except the last, + which uses ``xxx ``. We accumulate bytes until either: + * the buffer ends with a final line ``xxx ...``, or + * the server closes the connection (yielding partial bytes). + """ + buf = b"" + while len(buf) < max_bytes: + try: + chunk = s.recv(4096) + except socket.timeout: + break + if not chunk: + break + buf += chunk + # Check the last full line for the "final" marker (digit + space). + lines = buf.splitlines() + last = lines[-1] if lines else b"" + if buf.endswith(b"\r\n") and len(last) >= 4 and last[3:4] == b" ": + break + return buf + + +def _send_cmd(s: socket.socket, cmd: bytes) -> bytes: + s.sendall(cmd) + return _read_reply(s) + + +# --------------------------------------------------------------------------- +# 1. AUTH must NEVER be offered on port 25 (inbound, no submission). +# --------------------------------------------------------------------------- + + +def test_auth_not_advertised_in_ehlo(): + s = _raw_session() + try: + resp = _send_cmd(s, b"EHLO example.com\r\n") + text = resp.decode("ascii", errors="replace").upper() + assert "AUTH" not in text, f"AUTH advertised on port 25!\n{text}" + finally: + s.close() + + +def test_auth_command_rejected(): + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + resp = _send_cmd(s, b"AUTH LOGIN\r\n") + # Must NOT be 235 (auth success) or 334 (continue), even on accident. + # Both 502 (not implemented) and 503 (bad sequence) are acceptable. + assert resp[:3] in (b"502", b"503", b"500"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 2. VRFY/EXPN address enumeration must be neutered. +# --------------------------------------------------------------------------- + + +def test_vrfy_does_not_confirm_existence(): + # No mock_api_server / mailbox setup needed — VRFY must reply identically + # whether the address exists or not. If it ever leaked existence the + # answer would differ even before the MDA is consulted. + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + # Both a real and a fake mailbox should produce the SAME class of reply + # so the attacker cannot tell them apart. + a = _send_cmd(s, b"VRFY known@example.com\r\n") + b = _send_cmd(s, b"VRFY does-not-exist@example.com\r\n") + assert a[:1] == b[:1], f"VRFY reply class differs: {a!r} vs {b!r}" + # And we should never confirm with 250 (which would mean "yes, exists"). + assert not a.startswith(b"250"), a + finally: + s.close() + + +def test_expn_disabled(): + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + resp = _send_cmd(s, b"EXPN postmaster\r\n") + assert resp[:3] in (b"502", b"500"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 3. SMTP smuggling — CVE-2023-51764 family. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "smuggle_bytes", + [ + # bare LF + b"\n.\r\n", + # bare CR + b"\r.\r\n", + # bare LF + bare LF EOD + b"\n.\n", + ], + ids=["LF-dot-CRLF", "CR-dot-CRLF", "LF-dot-LF"], +) +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. + """ + 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 + # silently pass without proving smuggling failed. + mock_api_server.add_mailbox("smuggled@example.com") + + s = _raw_session() + try: + _send_cmd(s, b"EHLO attacker.example\r\n") + _send_cmd(s, b"MAIL FROM:\r\n") + _send_cmd(s, b"RCPT TO:\r\n") + _send_cmd(s, b"DATA\r\n") + payload = ( + b"Subject: outer\r\n" + b"\r\n" + b"outer body" + smuggle_bytes + b"MAIL FROM:\r\n" + b"RCPT TO:\r\n" + b"DATA\r\n" + b"Subject: smuggled\r\n" + b"\r\n" + b"smuggled body\r\n.\r\n" + ) + s.sendall(payload) + # Final reply ends one or more responses; read until socket idle. + s.settimeout(3) + resp = b"" + try: + while True: + chunk = s.recv(4096) + if not chunk: + break + resp += chunk + if b"\r\n" in chunk and resp.rstrip().endswith((b"OK", b"delivery", b"later")): + break + except socket.timeout: + pass + try: + _send_cmd(s, b"QUIT\r\n") + except OSError: + pass + finally: + s.close() + + # The MDA must NOT have seen a second envelope with `smuggled@example.com`. + rcpts_seen = [ + addr + for em in mock_api_server.received_emails + for addr in em["metadata"]["original_recipients"] + ] + assert "smuggled@example.com" not in rcpts_seen, ( + f"SMTP smuggling succeeded! Recipients seen: {rcpts_seen}" + ) + + +# --------------------------------------------------------------------------- +# 4. NUL bytes in DATA must be rejected. +# --------------------------------------------------------------------------- + + +def test_nul_byte_in_body_rejected(mock_api_server, mta_impl): + if mta_impl == "postfix": + pytest.skip("Postfix accepts and silently normalizes NUL bytes; pymta is stricter") + mock_api_server.add_mailbox("test@example.com") + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + _send_cmd(s, b"MAIL FROM:\r\n") + _send_cmd(s, b"RCPT TO:\r\n") + _send_cmd(s, b"DATA\r\n") + s.sendall(b"Subject: NUL test\r\n\r\nhello\x00world\r\n.\r\n") + resp = _read_reply(s) + assert resp[:1] in (b"4", b"5"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 5. Control-character / CRLF injection in MAIL FROM and RCPT TO. +# --------------------------------------------------------------------------- + + +def test_tab_in_address_rejected(mta_impl): + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + # TAB inside the address is a header-unfolding vector. + resp = _send_cmd(s, b"MAIL FROM:\r\n") + assert resp[:1] in (b"4", b"5"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 6. Overlong local-parts / domains. +# --------------------------------------------------------------------------- + + +def test_overlong_local_part_rejected(mta_impl): + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + _send_cmd(s, b"MAIL FROM:\r\n") + long_local = b"a" * 200 + resp = _send_cmd(s, b"RCPT TO:<" + long_local + b"@example.com>\r\n") + # 4xx (Postfix milter tempfail path) or 5xx (pymta strict reject) both + # satisfy the security requirement: the address must not be delivered. + assert resp[:1] in (b"4", b"5"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 7. Pre-DATA SIZE oversize announcement must be rejected at MAIL FROM time. +# --------------------------------------------------------------------------- + + +def test_size_overlimit_rejected(mock_api_server, smtp_client): + s = _raw_session() + try: + _send_cmd(s, b"EHLO example.com\r\n") + # 1 GB announced — well above MAX_INCOMING_EMAIL_SIZE (30 MB). + resp = _send_cmd(s, b"MAIL FROM: SIZE=1000000000\r\n") + assert resp[:3] in (b"552", b"452", b"550"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 8. RSET resets the envelope state. +# --------------------------------------------------------------------------- + + +def test_rset_clears_envelope(mock_api_server, smtp_client): + mock_api_server.add_mailbox("test@example.com") + smtp_client.helo("example.com") + smtp_client.mail("a@example.com") + smtp_client.rcpt("test@example.com") + smtp_client.rset() + # After RSET, DATA without MAIL/RCPT must be refused. + code, _ = smtp_client.docmd("DATA") + assert code // 100 == 5, code + + +# --------------------------------------------------------------------------- +# 9. Hard-error limit / unknown-command flood eventually disconnects. +# --------------------------------------------------------------------------- + + +def test_unknown_command_flood_does_not_hang(mta_impl): + s = _raw_session(timeout=10) + try: + for i in range(200): + try: + s.sendall(f"GARBAGE{i}\r\n".encode()) + _read_reply(s) + except OSError: + # Server closed the connection — that is the expected defense. + return + pytest.fail("server accepted 200 unknown commands without disconnecting") + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 10. Line-length limit enforced. +# --------------------------------------------------------------------------- + + +def test_overlong_command_line_rejected(): + s = _raw_session() + try: + s.sendall(b"A" * 5000 + b"\r\n") + resp = _read_reply(s) + # Either the server returned a 4xx/5xx error, or it closed the socket + # without replying. Both are acceptable defences against a flooded + # parser; silently accepting the line is not. + assert not resp or resp[:1] in (b"4", b"5"), resp + finally: + s.close() + + +# --------------------------------------------------------------------------- +# 11. Reverse-path / null-sender accepted (bounces). +# --------------------------------------------------------------------------- + + +def test_null_sender_accepted(mock_api_server, smtp_client): + mock_api_server.add_mailbox("test@example.com") + # smtplib doesn't directly support empty MAIL FROM; use docmd. + code, _ = smtp_client.helo("example.com") + assert code == 250 + code, _ = smtp_client.docmd("MAIL FROM:<>") + assert code == 250, code + code, _ = smtp_client.docmd("RCPT TO:") + assert code == 250, code + + +def test_null_recipient_rejected(smtp_client): + smtp_client.helo("example.com") + smtp_client.docmd("MAIL FROM:") + code, _ = smtp_client.docmd("RCPT TO:<>") + assert code // 100 == 5, code + + +# --------------------------------------------------------------------------- +# 12. Missing EHLO/HELO before MAIL → 503. +# --------------------------------------------------------------------------- + + +def test_mail_without_helo_rejected(mta_impl): + # Postfix in our config has `smtpd_helo_required = no` (the Postfix + # default) so it accepts MAIL FROM without a prior HELO. pymta is strict + # by default. This is a documented behaviour gap, not a security bug — + # the MDA will reject malformed envelopes either way. + if mta_impl == "postfix": + pytest.skip("Postfix accepts MAIL FROM without HELO by default") + s = _raw_session() + try: + resp = _send_cmd(s, b"MAIL FROM:\r\n") + assert resp[:3] in (b"503", b"550"), resp + finally: + s.close() diff --git a/src/mta-in/tests/test_smtp_protocol.py b/src/mta-in/tests/test_smtp_protocol.py index 543b3511..941e15bd 100644 --- a/src/mta-in/tests/test_smtp_protocol.py +++ b/src/mta-in/tests/test_smtp_protocol.py @@ -74,27 +74,10 @@ def test_partial_writes(): assert response.startswith("250") -@pytest.mark.skip(reason="TODO review") -def test_pipelining_support(): - """Test SMTP command pipelining support""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.connect((MTA_HOST, 25)) - s.settimeout(2) - s.recv(1024) # Greeting - - # Send multiple commands at once - pipeline = ( - b"HELO example.com\r\nMAIL FROM:\r\nRCPT TO:\r\n" - ) - s.send(pipeline) - - # Should get multiple responses - responses = [] - for _ in range(3): - response = s.recv(1024).decode() - responses.append(response) - - assert all(r.startswith("250") for r in responses) +# Pipelining is deliberately NOT advertised in EHLO (see handle_EHLO) and +# command_call_limit enforces strict per-verb counts. A test that pipelines +# multiple commands at once would assert behaviour we *avoid* — leaving the +# placeholder skipped would only rot, so we don't keep one. @pytest.mark.skip(reason="Not supported for now") diff --git a/src/mta-in/uv.lock b/src/mta-in/uv.lock index df2f4f4f..224cc9d2 100644 --- a/src/mta-in/uv.lock +++ b/src/mta-in/uv.lock @@ -2,6 +2,19 @@ version = 1 revision = 3 requires-python = ">=3.14.4, <4.0" +[[package]] +name = "aiosmtpd" +version = "1.4.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "atpublic" }, + { name = "attrs" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c4/ca/b2b7cc880403ef24be77383edaadfcf0098f5d7b9ddbf3e2c17ef0a6af0d/aiosmtpd-1.4.6.tar.gz", hash = "sha256:5a811826e1a5a06c25ebc3e6c4a704613eb9a1bcf6b78428fbe865f4f6c9a4b8", size = 152775, upload-time = "2024-05-18T11:37:50.029Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/39/d401756df60a8344848477d54fdf4ce0f50531f6149f3b8eaae9c06ae3dc/aiosmtpd-1.4.6-py3-none-any.whl", hash = "sha256:72c99179ba5aa9ae0abbda6994668239b64a5ce054471955fe75f581d2592475", size = 154263, upload-time = "2024-05-18T11:37:47.877Z" }, +] + [[package]] name = "annotated-types" version = "0.7.0" @@ -32,6 +45,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/af/0f/3b8fdc946b4d9cc8cc1e8af42c4e409468c84441b933d037e101b3d72d86/astroid-3.3.11-py3-none-any.whl", hash = "sha256:54c760ae8322ece1abd213057c4b5bba7c49818853fc901ef09719a60dbf9dec", size = 275612, upload-time = "2025-07-13T18:04:21.07Z" }, ] +[[package]] +name = "atpublic" +version = "7.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a9/05/e2e131a0debaf0f01b8a1b586f5f11713f6affc3e711b406f15f11eafc92/atpublic-7.0.0.tar.gz", hash = "sha256:466ef10d0c8bbd14fd02a5fbd5a8b6af6a846373d91106d3a07c16d72d96b63e", size = 17801, upload-time = "2025-11-29T05:56:45.45Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/96/c0/271f3e1e3502a8decb8ee5c680dbed2d8dc2cd504f5e20f7ed491d5f37e1/atpublic-7.0.0-py3-none-any.whl", hash = "sha256:6702bd9e7245eb4e8220a3e222afcef7f87412154732271ee7deee4433b72b4b", size = 6421, upload-time = "2025-11-29T05:56:44.604Z" }, +] + +[[package]] +name = "attrs" +version = "26.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9a/8e/82a0fe20a541c03148528be8cac2408564a6c9a0cc7e9171802bc1d26985/attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32", size = 952055, upload-time = "2026-03-19T14:22:25.026Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" }, +] + [[package]] name = "certifi" version = "2026.1.4" @@ -158,6 +189,34 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] +[[package]] +name = "httpcore" +version = "1.0.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "certifi" }, + { name = "h11" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, +] + +[[package]] +name = "httpx" +version = "0.28.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "certifi" }, + { name = "httpcore" }, + { name = "idna" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, +] + [[package]] name = "idna" version = "3.11" @@ -221,6 +280,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, ] +[[package]] +name = "prometheus-client" +version = "0.24.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f0/58/a794d23feb6b00fc0c72787d7e87d872a6730dd9ed7c7b3e954637d8f280/prometheus_client-0.24.1.tar.gz", hash = "sha256:7e0ced7fbbd40f7b84962d5d2ab6f17ef88a72504dcf7c0b40737b43b2a461f9", size = 85616, upload-time = "2026-01-14T15:26:26.965Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/74/c3/24a2f845e3917201628ecaba4f18bab4d18a337834c1df2a159ee9d22a42/prometheus_client-0.24.1-py3-none-any.whl", hash = "sha256:150db128af71a5c2482b36e588fc8a6b95e498750da4b17065947c16070f4055", size = 64057, upload-time = "2026-01-14T15:26:24.42Z" }, +] + [[package]] name = "pydantic" version = "2.12.5" @@ -323,6 +391,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/30/3d/64ad57c803f1fa1e963a7946b6e0fea4a70df53c1a7fed304586539c2bac/pytest-8.3.5-py3-none-any.whl", hash = "sha256:c69214aa47deac29fad6c2a4f590b9c4a9fdb16a403176fe154b79c0b4d4d820", size = 343634, upload-time = "2025-03-02T12:54:52.069Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "0.24.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/52/6d/c6cf50ce320cf8611df7a1254d86233b3df7cc07f9b5f5cbcb82e08aa534/pytest_asyncio-0.24.0.tar.gz", hash = "sha256:d081d828e576d85f875399194281e92bf8a68d60d72d1a2faf2feddb6c46b276", size = 49855, upload-time = "2024-08-22T08:03:18.145Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/96/31/6607dab48616902f76885dfcf62c08d929796fc3b2d2318faf9fd54dbed9/pytest_asyncio-0.24.0-py3-none-any.whl", hash = "sha256:a811296ed596b69bf0b6f3dc40f83bcaf341b155a269052d82efa2b25ac7037b", size = 18024, upload-time = "2024-08-22T08:03:15.536Z" }, +] + [[package]] name = "pytest-cov" version = "6.0.0" @@ -381,8 +461,10 @@ name = "st-messages-mta-in" version = "0.8.0" source = { editable = "." } dependencies = [ + { name = "aiosmtpd" }, + { name = "httpx" }, + { name = "prometheus-client" }, { name = "pyjwt" }, - { name = "pymilter" }, { name = "requests" }, ] @@ -391,26 +473,34 @@ dev = [ { name = "fastapi" }, { name = "pylint" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "ruff" }, { name = "types-requests" }, { name = "uvicorn" }, ] +postfix = [ + { name = "pymilter" }, +] [package.metadata] requires-dist = [ + { name = "aiosmtpd", specifier = "==1.4.6" }, { name = "fastapi", marker = "extra == 'dev'", specifier = "==0.115.12" }, + { name = "httpx", specifier = "==0.28.1" }, + { name = "prometheus-client", specifier = "==0.24.1" }, { name = "pyjwt", specifier = "==2.10.1" }, { name = "pylint", marker = "extra == 'dev'", specifier = "==3.3.4" }, - { name = "pymilter", specifier = "==1.0.5" }, + { name = "pymilter", marker = "extra == 'postfix'", specifier = "==1.0.5" }, { name = "pytest", marker = "extra == 'dev'", specifier = "==8.3.5" }, + { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = "==0.24.0" }, { name = "pytest-cov", marker = "extra == 'dev'", specifier = "==6.0.0" }, { name = "requests", specifier = "==2.32.3" }, { name = "ruff", marker = "extra == 'dev'", specifier = "==0.9.3" }, { name = "types-requests", marker = "extra == 'dev'", specifier = "==2.32.0.20241016" }, { name = "uvicorn", marker = "extra == 'dev'", specifier = "==0.34.1" }, ] -provides-extras = ["dev"] +provides-extras = ["postfix", "dev"] [[package]] name = "starlette"