From 28ec784ae21fa72f4a7ca6094af34d72e982ed0f Mon Sep 17 00:00:00 2001 From: Mix Date: Sun, 11 Oct 2026 22:18:14 +0300 Subject: [PATCH] inbound-antispam-panel: inbound journal milter on port 25 Step 3/12 of docs/plans/inbound-antispam-panel.md. The panel serves a second milter while the inbound relay is on (internal/milter/inbound.go, socket INBOUND_JOURNAL_MILTER_SOCKET, default /run/selfpost/inbound-journal.sock). build/postfix-config.sh puts it last on port 25, after the optional antispam milter, with default_action=accept; postfix-wrapper.sh waits for its socket. At end-of-message it writes one inbound_spam_log row per recipient at an inbound domain - other port-25 recipients (DMARC report addresses) get none - and always accepts. With a filter attached it reads the verdict from X-Rspamd-Action and X-Spamd-Result (action, score, up to six weighted symbols, only what the patterns recognise, 255 bytes); without one those headers are the sender's and are not read. The plan named X-Rspamd-Score, which rspamd does not have. The antispam compose fragment mounts rspamd/milter_headers.conf into the sidecar so that it adds the two headers (checked in rspamd 3.11: both routines exist and remove a sender's own headers of those names). Store: InsertInboundSpam, IsInboundDomain. Checked on the test server, uncommitted tree: go vet ./... ok, go test ./... ok (22 packages), make e2e ok - the relayed message of inbound_relay_delivers_and_refuses has its journal row (accept by selfpost) and the DMARC report sent through the same listener has none. Not yet seen in a real run: a verdict written by rspamd reaching this milter. The e2e stand has no filter, and the test server's only inbound domain has no upstream, so it accepts no mail. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 17 +- build/postfix-config.sh | 22 +- build/postfix-wrapper.sh | 14 +- cmd/panel/journal.go | 37 +++- cmd/panel/main.go | 15 +- deploy/antispam/docker-compose.antispam.yml | 3 + deploy/antispam/rspamd/milter_headers.conf | 11 + docs/architecture.md | 21 +- docs/guide.md | 9 + docs/plans/inbound-antispam-panel.md | 7 +- docs/roadmap.md | 2 +- internal/milter/inbound.go | 229 ++++++++++++++++++++ internal/milter/inbound_test.go | 224 +++++++++++++++++++ internal/store/inboundspam.go | 72 ++++++ internal/store/inboundspam_test.go | 61 ++++++ test/e2e/port25_config_check.go | 11 +- test/e2e/port25_features_test.go | 24 ++ 17 files changed, 748 insertions(+), 31 deletions(-) create mode 100644 deploy/antispam/rspamd/milter_headers.conf create mode 100644 internal/milter/inbound.go create mode 100644 internal/milter/inbound_test.go create mode 100644 internal/store/inboundspam.go create mode 100644 internal/store/inboundspam_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 5890273..4d9b0f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,22 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version - Schema: migration `0002_inbound_spam_log.sql`, the first after the 2.0 baseline - two empty tables for the inbound antispam journal and the filter lists (inbound-antispam-panel). A 2.0.x database is migrated when the panel - starts; nothing reads or writes the tables yet. + starts. +- **Inbound mail is recorded.** With the inbound relay on, the panel runs a + second milter, the last one on port 25, and writes a row for every recipient + of a message that was accepted for an inbound domain: client address, + envelope, subject and - with a spam filter attached - the filter's action, + score and heaviest symbols. It only watches: it refuses nothing and Postfix + accepts mail when it is down. Refused mail is not there yet (it comes from + the mail log, next), and there is no page for the journal yet. +- The antispam compose fragment mounts `antispam/rspamd/milter_headers.conf` + into the rspamd sidecar, which makes it write its verdict into + `X-Rspamd-Action` and `X-Spamd-Result`. **Copy `deploy/antispam/rspamd/` + next to the fragment before starting it** - Docker creates a directory in + place of a missing file and the sidecar then fails to read its + configuration. +- New setting `INBOUND_JOURNAL_MILTER_SOCKET` (default + `/run/selfpost/inbound-journal.sock`). ## [2.0.1] - 2026-10-11 diff --git a/build/postfix-config.sh b/build/postfix-config.sh index cdf2716..b46dbce 100644 --- a/build/postfix-config.sh +++ b/build/postfix-config.sh @@ -117,6 +117,9 @@ SUBMISSION_ENABLE="${SUBMISSION_ENABLE:-false}" INBOUND_RELAY_ENABLE="${INBOUND_RELAY_ENABLE:-false}" INBOUND_ANTISPAM_MILTER="${INBOUND_ANTISPAM_MILTER:-}" INBOUND_ANTISPAM_MILTER_ACTION="${INBOUND_ANTISPAM_MILTER_ACTION:-accept}" +# The panel's inbound journal milter (monitoring, fail-open), on port 25 only +# and only while the inbound relay is on. Matched by postfix-wrapper.sh. +INBOUND_JOURNAL_SOCK="${INBOUND_JOURNAL_MILTER_SOCKET:-/run/selfpost/inbound-journal.sock}" INBOUND_RATE_LIMIT_MESSAGES_PER_IP="${INBOUND_RATE_LIMIT_MESSAGES_PER_IP:-20}" INBOUND_MESSAGE_SIZE_LIMIT="${INBOUND_MESSAGE_SIZE_LIMIT:-26214400}" RELAY_DOMAINS_MAP="${POSTFIX_RELAY_DOMAINS:-/data/postfix/relay_domains}" @@ -357,7 +360,8 @@ if [ "${INBOUND_RELAY_ENABLE}" = "true" ] || [ "${DMARC_REPORTS_ENABLE}" = "true # would file inbound mail into the outbound send log under the *sender's* # domain — polluting Deliveries and letting a forged From: consume that # domain's level-2 rate-limit budget. So smtpd_milters is always overridden - # here: either the optional antispam milter, or nothing at all. + # here: the optional antispam milter, then - with the inbound relay on - the + # panel's inbound journal milter, or nothing at all. INBOUND_MILTERS="" if [ "${INBOUND_RELAY_ENABLE}" = "true" ] && [ -n "${INBOUND_ANTISPAM_MILTER}" ]; then case "${INBOUND_ANTISPAM_MILTER}" in @@ -376,6 +380,22 @@ if [ "${INBOUND_RELAY_ENABLE}" = "true" ] || [ "${DMARC_REPORTS_ENABLE}" = "true esac INBOUND_MILTERS="{${INBOUND_ANTISPAM_MILTER},default_action=${INBOUND_ANTISPAM_MILTER_ACTION}}" fi + if [ "${INBOUND_RELAY_ENABLE}" = "true" ]; then + # Last in the chain: Postfix hands a message to its milters one after + # another at end-of-message, so this one reads the verdict headers the + # filter added and sees only what the filter let through. It watches and + # never refuses, and must never hold up someone else's mail: accept when + # it is down. A comma, not a space, between the two - see + # master_cf_set_smtp_inet_o. + case "${INBOUND_JOURNAL_SOCK}" in + /[A-Za-z0-9._/-]*) ;; + *) + echo "FATAL: INBOUND_JOURNAL_MILTER_SOCKET must be an absolute path, got: ${INBOUND_JOURNAL_SOCK}" >&2 + exit 1 + ;; + esac + INBOUND_MILTERS="${INBOUND_MILTERS:+${INBOUND_MILTERS},}{unix:${INBOUND_JOURNAL_SOCK},default_action=accept}" + fi PORT25_RATE="${INBOUND_RATE_LIMIT_MESSAGES_PER_IP}" PORT25_SIZE="${INBOUND_MESSAGE_SIZE_LIMIT}" diff --git a/build/postfix-wrapper.sh b/build/postfix-wrapper.sh index d4dfc57..ed6f682 100644 --- a/build/postfix-wrapper.sh +++ b/build/postfix-wrapper.sh @@ -2,8 +2,9 @@ # Postfix start wrapper: solves the cold-start race where Postfix would try to # reach the milter sockets before they are listening. # -# It blocks until BOTH milter sockets — OpenDKIM and the panel's journal-milter -# — are present, then execs `postfix start-fg`. If they are not ready within the +# It blocks until the milter sockets — OpenDKIM, the panel's journal-milter +# and, with the inbound relay on, its inbound journal milter — are present, +# then execs `postfix start-fg`. If they are not ready within the # timeout it exits non-zero WITHOUT starting Postfix, so supervisord/Docker see # an explicit startup failure instead of a relay running blind. # @@ -16,8 +17,13 @@ JOURNAL_SOCK="${JOURNAL_MILTER_SOCKET:-/run/selfpost/journal.sock}" TIMEOUT="${MILTER_WAIT_TIMEOUT:-30}" INTERVAL=1 +SOCKS="$OPENDKIM_SOCK $JOURNAL_SOCK" +if [ "${INBOUND_RELAY_ENABLE:-false}" = "true" ]; then + SOCKS="$SOCKS ${INBOUND_JOURNAL_MILTER_SOCKET:-/run/selfpost/inbound-journal.sock}" +fi + elapsed=0 -for sock in "$OPENDKIM_SOCK" "$JOURNAL_SOCK"; do +for sock in $SOCKS; do while [ ! -S "$sock" ]; do if [ "$elapsed" -ge "$TIMEOUT" ]; then echo "postfix-wrapper: timed out after ${TIMEOUT}s waiting for milter socket $sock" >&2 @@ -29,5 +35,5 @@ for sock in "$OPENDKIM_SOCK" "$JOURNAL_SOCK"; do echo "postfix-wrapper: milter socket ready: $sock" done -echo "postfix-wrapper: both milter sockets ready, starting postfix" +echo "postfix-wrapper: milter sockets ready, starting postfix" exec postfix start-fg diff --git a/cmd/panel/journal.go b/cmd/panel/journal.go index bc206b9..05933c0 100644 --- a/cmd/panel/journal.go +++ b/cmd/panel/journal.go @@ -15,21 +15,42 @@ import ( // serveJournal opens the journal-milter Unix socket and runs the real milter // (architecture.md § Mail path), recording accepted messages into the send // log. Socket lifecycle (creation, stale cleanup, group permissions) lives -// here; the protocol handler lives in internal/milter. +// in listenMilter; the protocol handler lives in internal/milter. func serveJournal(ctx context.Context, cfg config, st *store.Store) error { - socketPath := cfg.journalSocket - if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil { + ln, err := listenMilter(cfg.journalSocket) + if err != nil { return err } + log.Printf("journal-milter listening on %s", cfg.journalSocket) + return milter.Serve(ctx, ln, st) +} + +// serveInboundJournal runs the inbound journal milter, the last milter on +// port 25 (internal/milter/inbound.go). It exists only while the inbound relay +// is on: build/postfix-config.sh attaches the socket under the same condition. +func serveInboundJournal(ctx context.Context, cfg config, st *store.Store) error { + ln, err := listenMilter(cfg.inboundJournalSocket) + if err != nil { + return err + } + log.Printf("inbound-journal-milter listening on %s", cfg.inboundJournalSocket) + return milter.ServeInbound(ctx, ln, st, cfg.inboundAntispamMilter != "") +} + +// listenMilter opens a milter's Unix socket where Postfix can reach it. +func listenMilter(socketPath string) (net.Listener, error) { + if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil { + return nil, err + } // Clear a stale socket left behind by an unclean shutdown, otherwise the // listen below fails with "address already in use". if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) { - return err + return nil, err } ln, err := net.Listen("unix", socketPath) if err != nil { - return err + return nil, err } // Postfix (user `postfix`) connects to this socket as a milter and needs // group access. The socket inherits group `selfpost` from the setgid parent @@ -37,9 +58,7 @@ func serveJournal(ctx context.Context, cfg config, st *store.Store) error { // it (connecting to a Unix socket needs write permission on the node). if err := os.Chmod(socketPath, 0o660); err != nil { ln.Close() - return err + return nil, err } - - log.Printf("journal-milter listening on %s", socketPath) - return milter.Serve(ctx, ln, st) + return ln, nil } diff --git a/cmd/panel/main.go b/cmd/panel/main.go index 0905ff3..90e5012 100644 --- a/cmd/panel/main.go +++ b/cmd/panel/main.go @@ -96,7 +96,10 @@ type config struct { // what attaches it; the panel only says so. See antispamAction. inboundAntispamMilter string inboundAntispamAction string - dmarcEnabled bool + // The inbound journal milter's socket; the default mirrors + // build/postfix-config.sh. Served only while the inbound relay is on. + inboundJournalSocket string + dmarcEnabled bool } func loadConfig() config { @@ -162,6 +165,7 @@ func loadConfig() config { inboundEnabled: os.Getenv("INBOUND_RELAY_ENABLE") == "true", inboundAntispamMilter: os.Getenv("INBOUND_ANTISPAM_MILTER"), inboundAntispamAction: antispamAction(os.Getenv("INBOUND_ANTISPAM_MILTER_ACTION")), + inboundJournalSocket: envDefault("INBOUND_JOURNAL_MILTER_SOCKET", "/run/selfpost/inbound-journal.sock"), // Optional DMARC aggregate ingest on port 25. Off unless exactly "true". dmarcEnabled: os.Getenv("DMARC_REPORTS_ENABLE") == "true", } @@ -294,7 +298,7 @@ func run() error { } var wg sync.WaitGroup - errc := make(chan error, 4) + errc := make(chan error, 5) roles := []struct { name string @@ -318,6 +322,13 @@ func run() error { }}, } + if cfg.inboundEnabled { + roles = append(roles, struct { + name string + fn func(context.Context) error + }{"inbound-journal-milter", func(ctx context.Context) error { return serveInboundJournal(ctx, cfg, st) }}) + } + for _, r := range roles { wg.Add(1) go func(name string, fn func(context.Context) error) { diff --git a/deploy/antispam/docker-compose.antispam.yml b/deploy/antispam/docker-compose.antispam.yml index c3f94f7..63194f4 100644 --- a/deploy/antispam/docker-compose.antispam.yml +++ b/deploy/antispam/docker-compose.antispam.yml @@ -27,3 +27,6 @@ services: image: rspamd/rspamd:3.11 restart: unless-stopped # No host ports: only SelfPost talks to the milter on the compose network. + volumes: + # The verdict headers SelfPost's inbound journal reads. + - ./antispam/rspamd/milter_headers.conf:/etc/rspamd/local.d/milter_headers.conf:ro diff --git a/deploy/antispam/rspamd/milter_headers.conf b/deploy/antispam/rspamd/milter_headers.conf new file mode 100644 index 0000000..4b57e74 --- /dev/null +++ b/deploy/antispam/rspamd/milter_headers.conf @@ -0,0 +1,11 @@ +# Mounted into the sidecar as /etc/rspamd/local.d/milter_headers.conf. +# +# rspamd writes its verdict into two headers of every message it lets through: +# the action it chose and the score with the symbols behind it. SelfPost's +# inbound journal reads them (Inbound > Log shows them next to the message), +# and the upstream mailbox server receives them with the mail. +# +# rspamd removes headers of these names that a sender wrote himself before it +# adds its own. Without this file the journal still lists accepted mail, with +# no verdict next to it. +use = ["x-spamd-result", "x-rspamd-action"]; diff --git a/docs/architecture.md b/docs/architecture.md index 2ee7383..4d78e62 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -27,7 +27,7 @@ Managed programs ([build/supervisord.conf](../build/supervisord.conf)): | Program | User | Priority | Role | |---|---|---|---| | `opendkim` | root → `opendkim` (`UserID` in opendkim.conf) | 100 | DKIM signing milter | -| `panel` | panel | 200 | HTTP UI + journal-milter + log-tailer + rate-limit recalc | +| `panel` | panel | 200 | HTTP UI + journal-milter + log-tailer + rate-limit recalc; with the inbound relay on also the inbound journal milter | | `postfix` | root (wrapper) | 300 | MTA — started only after both milter sockets exist | | `postfix-reload` | root | — | On-demand `postfix reload` (autostart off) | | `cert-reload` | root | 400 | Daily `postfix reload` for renewed TLS certs | @@ -74,10 +74,21 @@ smtp/inet` removes Debian's stock smtpd). Outbound delivery still uses the `smtpd_tls_security_level=encrypt`. - **25/smtp inet** — when `INBOUND_RELAY_ENABLE=true` and/or `DMARC_REPORTS_ENABLE=true`. No SASL, no OpenDKIM, no journal-milter — - `smtpd_milters` is always overridden on this service (to the antispam milter - or to nothing), because main.cf's submission chain would otherwise let a - signing outage defer relayed mail and would file inbound mail into the - outbound send log under the sender's domain. + `smtpd_milters` is always overridden on this service, because main.cf's + submission chain would otherwise let a signing outage defer relayed mail and + would file inbound mail into the outbound send log under the sender's + domain. With the inbound relay on the chain is the optional antispam milter, + then the panel's **inbound journal milter** (unix socket + `INBOUND_JOURNAL_MILTER_SOCKET`, default + `/run/selfpost/inbound-journal.sock`, `default_action=accept`); with DMARC + ingest alone it is empty. Postfix hands a message to its milters one after + another at end-of-message, each seeing what the one before added, so the + journal milter sees only mail the filter let through and reads the filter's + verdict headers (`X-Rspamd-Action`, `X-Spamd-Result`) off it. It writes one + `inbound_spam_log` row per recipient at an inbound domain - a DMARC report + address is not one - and never refuses anything + ([internal/milter/inbound.go](../internal/milter/inbound.go)). The headers + are read only while a filter is attached; refused mail never reaches it. **Inbound relay** (`INBOUND_RELAY_ENABLE=true`): accepts only `relay_domains` + `relay_recipient_maps` (`reject_unauth_destination`, `reject_unlisted_recipient`). Maps under `/data/postfix/` (`relay_domains`, diff --git a/docs/guide.md b/docs/guide.md index 991daed..f3827d0 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -209,6 +209,7 @@ expected to set; defaults match the code exactly. | `DMARC_MESSAGE_SIZE_LIMIT` | Maximum report message size in bytes when DMARC ingest is on. | `5242880` (5 MiB) | `.env` | | `INBOUND_ANTISPAM_MILTER` | Optional milter on the inbound relay listener only (requires `INBOUND_RELAY_ENABLE=true`; not 465/587 or DMARC-only port 25). Empty = off. Format `inet:host:port` or `unix:/path`. Example with [deploy/antispam/docker-compose.antispam.yml](../deploy/antispam/docker-compose.antispam.yml): `inet:antispam:11332`. The panel reads it to report on the *Spam filter* box of the inbound pages and to check that the filter answers (Server › Health). | *(empty)* | `.env` | | `INBOUND_ANTISPAM_MILTER_ACTION` | What Postfix does if that milter is down: `accept` (fail-open) or `tempfail` (defer). The *Spam filter* box states which. | `accept` | `.env` | +| `INBOUND_JOURNAL_MILTER_SOCKET` | Unix socket of the panel's inbound journal milter, the last milter on port 25 while the inbound relay is on. It records accepted inbound mail and never refuses any; there is no reason to change it. | `/run/selfpost/inbound-journal.sock` | compose | | `INBOUND_RATE_LIMIT_MESSAGES_PER_IP` | Coarse per-client-IP cap on inbound smtpd (`smtpd_client_message_rate_limit`). Uses the same window as `RATE_LIMIT_WINDOW_SECONDS`. | `20` | `.env` | | `INBOUND_MESSAGE_SIZE_LIMIT` | Maximum message size in bytes on inbound smtpd (`message_size_limit`). When both inbound relay and DMARC ingest are on, the shared listener uses the **greater** of this and `DMARC_MESSAGE_SIZE_LIMIT`. | `26214400` (25 MiB) | `.env` | | `RATE_LIMIT_MESSAGES_PER_IP` | Level-1 backstop: maximum messages one client IP may submit per window (Postfix `smtpd_client_message_rate_limit`). See [Rate limiting — level 1](#rate-limiting--level-1-ip-backstop). | `100` | `.env` | @@ -841,6 +842,14 @@ only sees SelfPost. Default action is fail-open (`accept`) so a down sidecar does not block backup-MX; set `INBOUND_ANTISPAM_MILTER_ACTION=tempfail` to defer instead. +The fragment mounts one file into the sidecar, +[deploy/antispam/rspamd/milter_headers.conf](../deploy/antispam/rspamd/milter_headers.conf): +it makes rspamd write its verdict - the action and the score with its symbols +- into two headers of every message it lets through. SelfPost records them +with the message, and the upstream receives them with the mail. Copy the +`antispam/rspamd/` directory next to the fragment; without the file accepted +mail is still recorded, with no verdict. + The panel only reports this setting; it does not change it. Every inbound domain page has a **Spam filter** box — on or off, what happens to mail while the filter is down, and that the filter belongs to the server, not to the diff --git a/docs/plans/inbound-antispam-panel.md b/docs/plans/inbound-antispam-panel.md index 99bb4f2..b9add80 100644 --- a/docs/plans/inbound-antispam-panel.md +++ b/docs/plans/inbound-antispam-panel.md @@ -57,8 +57,9 @@ smtpd_milters = { antispam, inbound-journal } ``` Antispam runs first; inbound-journal runs at end-of-message on the **accept** -path and reads rspamd-added headers (`X-Rspamd-Action`, `X-Rspamd-Score`, symbol -summary when present). +path and reads rspamd-added headers (`X-Rspamd-Action`, and `X-Spamd-Result` +for the score and symbols - rspamd has no `X-Rspamd-Score`; the sidecar adds +both through `deploy/antispam/rspamd/milter_headers.conf`). **Antispam rejects** do not reach end-of-message on the journal milter. v1 also tails `mail.log` for inbound `postfix/smtpd` **`milter-reject`** / **`reject`** @@ -244,7 +245,7 @@ Target version cut: **`2.1.0`** (MINOR), after `2.0.0`. One commit per step; - [x] Agree journal fields, retention, and RBAC (this plan § Decisions) — **Composer** - [x] Migration `0002_inbound_spam_log.sql` on the 2.0 baseline — **Composer** -- [ ] Inbound journal-milter + Postfix wiring — **Composer** +- [x] Inbound journal-milter + Postfix wiring — **Composer** - [ ] mail.log tailer: inbound reject rows — **Composer** - [ ] List CRUD + validation + atomic rspamd map sync — **Composer** - [ ] Panel: Inbound › Log and Inbound › Filter lists from the kit partials, mockups `in-log` / `in-filter-lists` (add client IP and engine columns mockup-first), evidence pairs — **Composer** diff --git a/docs/roadmap.md b/docs/roadmap.md index 68bfec3..105a677 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -30,7 +30,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md). |---|---|---|---|---| | preflight | Installation check page (`/preflight`) | candidate | — | [plans/preflight.md](plans/preflight.md) | | contributing | `CONTRIBUTING.md` | candidate | — | — | -| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | in progress | 2/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) | +| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | in progress | 3/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) | | inbound-quarantine | Inbound spam quarantine (hold / review / release) | candidate | — | [plans/inbound-quarantine.md](plans/inbound-quarantine.md) | | csrf-tokens | Per-session CSRF tokens on state-changing forms | candidate | — | — | | template-data-typing | Typed structs for template data instead of `map[string]any` | candidate | — | — | diff --git a/internal/milter/inbound.go b/internal/milter/inbound.go new file mode 100644 index 0000000..b30f395 --- /dev/null +++ b/internal/milter/inbound.go @@ -0,0 +1,229 @@ +package milter + +import ( + "context" + "log" + "net" + "net/textproto" + "regexp" + "strings" + + "github.com/emersion/go-milter" + + "github.com/mixeme/selfpost/internal/mailhdr" + "github.com/mixeme/selfpost/internal/store" +) + +// InboundStore is what the inbound journal milter needs: the journal itself +// and the list of inbound domains, since port 25 also takes mail that is not +// the relay's. *store.Store satisfies it; tests substitute a fake. +type InboundStore interface { + InsertInboundSpam(e store.InboundSpamEntry) error + IsInboundDomain(name string) (bool, error) +} + +// inboundSession is the inbound journal milter: the last milter on port 25, +// after the optional external filter. It writes one journal row per recipient +// of a message that reached end-of-message - that is, one the filter did not +// refuse; refusals never get this far and are read from mail.log instead. +// +// Like the send-log milter it only watches: every callback continues or +// accepts, and Postfix attaches the socket with default_action=accept, so +// neither an error here nor the panel being down holds up inbound mail. +type inboundSession struct { + milter.NoOpMilter + rec InboundStore + // filter says whether an external filter is attached in front of this + // milter. Without one the verdict headers below can only have been written + // by the sender, and are not read. + filter bool + + clientIP string // once per connection + + from string + rcpts []string + subject string + action string // X-Rspamd-Action + result string // X-Spamd-Result + seen bool // any verdict header of the filter +} + +func (s *inboundSession) Connect(host, family string, port uint16, addr net.IP, m *milter.Modifier) (milter.Response, error) { + if addr != nil { + s.clientIP = addr.String() + } + return milter.RespContinue, nil +} + +// MailFrom starts a message: a connection may carry several. +func (s *inboundSession) MailFrom(from string, m *milter.Modifier) (milter.Response, error) { + s.reset() + s.from = cleanAddress(from) + return milter.RespContinue, nil +} + +func (s *inboundSession) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) { + s.rcpts = append(s.rcpts, cleanAddress(rcpt)) + return milter.RespContinue, nil +} + +// Header keeps the first Subject and the filter's verdict. Postfix hands the +// message to its milters one after another at end-of-message, each seeing what +// the one before added, which is how the filter's headers arrive here. The +// filter replaces headers of these names, so the last one read is its own. +func (s *inboundSession) Header(name, value string, m *milter.Modifier) (milter.Response, error) { + switch textproto.CanonicalMIMEHeaderKey(name) { + case "Subject": + if s.subject == "" { + s.subject = mailhdr.DecodeSubject(value) + } + case "X-Rspamd-Action": + if s.filter { + s.action, s.seen = value, true + } + case "X-Spamd-Result": + if s.filter { + s.result, s.seen = value, true + } + } + return milter.RespContinue, nil +} + +// Body fires at end-of-message: the message is about to be queued for its +// upstream. +func (s *inboundSession) Body(m *milter.Modifier) (milter.Response, error) { + s.record() + s.reset() + return milter.RespAccept, nil +} + +func (s *inboundSession) Abort(m *milter.Modifier) error { + s.reset() + return nil +} + +func (s *inboundSession) reset() { + s.from, s.rcpts, s.subject = "", nil, "" + s.action, s.result, s.seen = "", "", false +} + +// record writes one row per recipient at an inbound domain. Other recipients +// of port 25 (a DMARC report address at the server's own name) are not the +// relay's mail and get no row. Failures are logged, never propagated. +func (s *inboundSession) record() { + e := store.InboundSpamEntry{ + ClientIP: s.clientIP, + From: s.from, + Subject: s.subject, + Decision: store.SpamDecisionAccept, + Engine: store.SpamEngineSelfPost, + } + if s.seen { + action, detail := rspamdVerdict(s.action, s.result) + e.Engine, e.Detail = store.SpamEngineRspamd, detail + if action == "quarantine" { + e.Decision = store.SpamDecisionQuarantine + } + } + for _, to := range s.rcpts { + domain := domainOf(to) + ok, err := s.rec.IsInboundDomain(domain) + if err != nil { + log.Printf("inbound-journal: look up %s: %v", domain, err) + continue + } + if !ok { + continue + } + e.InboundDomain, e.To = domain, to + if err := s.rec.InsertInboundSpam(e); err != nil { + log.Printf("inbound-journal: record %s -> %s: %v", s.from, to, err) + } + } +} + +// What the journal keeps of a verdict: enough to see why, on one table row. +const ( + verdictMaxSymbols = 6 + verdictMaxBytes = 255 +) + +var ( + // "default: False [7.20 / 15.00]" - the score and the reject threshold. + rspamdScore = regexp.MustCompile(`\[(-?[0-9]+(?:\.[0-9]+)?) / (-?[0-9]+(?:\.[0-9]+)?)\]`) + // "BAYES_SPAM(5.10)[100.00%]" - a symbol and its weight. + rspamdSymbol = regexp.MustCompile(`([A-Z][A-Z0-9_]{1,63})\((-?[0-9]+(?:\.[0-9]+)?)\)`) +) + +// rspamdVerdict reduces the filter's headers to an action and a one-line +// summary: the action, the score against the threshold, and the first few +// symbols that carried weight. It reads only what it recognises - the headers +// are the filter's when it is up, but the sender's own when it is not (it is +// attached fail-open), so nothing of them is kept verbatim except through +// these patterns. +func rspamdVerdict(actionHeader, resultHeader string) (action, detail string) { + action = strings.ToLower(strings.TrimSpace(actionHeader)) + for _, r := range action { + if (r < 'a' || r > 'z') && r != ' ' { + action = "" + break + } + } + if len(action) > 32 { + action = "" + } + + var parts []string + if action != "" { + parts = append(parts, action) + } + if m := rspamdScore.FindStringSubmatch(resultHeader); m != nil { + parts = append(parts, "score "+m[1]+" / "+m[2]) + } + detail = strings.Join(parts, ", ") + + var symbols []string + for _, m := range rspamdSymbol.FindAllStringSubmatch(resultHeader, -1) { + if strings.Trim(m[2], "-0.") == "" { // weight zero: informational + continue + } + symbols = append(symbols, m[1]+"("+m[2]+")") + if len(symbols) == verdictMaxSymbols { + break + } + } + if len(symbols) > 0 { + if detail != "" { + detail += "; " + } + detail += strings.Join(symbols, " ") + } + if len(detail) > verdictMaxBytes { + detail = detail[:verdictMaxBytes] + } + return action, detail +} + +// ServeInbound runs the inbound journal milter on ln until ctx is cancelled. +// filter says whether an external filter is attached in front of it +// (INBOUND_ANTISPAM_MILTER). It returns nil on a clean shutdown. +func ServeInbound(ctx context.Context, ln net.Listener, rec InboundStore, filter bool) error { + srv := &milter.Server{ + NewMilter: func() milter.Milter { return &inboundSession{rec: rec, filter: filter} }, + Actions: 0, // read-only + Protocol: milter.OptNoBody, // headers and end-of-message, not the body + } + + go func() { + <-ctx.Done() + _ = srv.Close() + }() + + if err := srv.Serve(ln); err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + return nil +} diff --git a/internal/milter/inbound_test.go b/internal/milter/inbound_test.go new file mode 100644 index 0000000..1ea49ee --- /dev/null +++ b/internal/milter/inbound_test.go @@ -0,0 +1,224 @@ +package milter + +import ( + "errors" + "net" + "strings" + "testing" + + "github.com/emersion/go-milter" + + "github.com/mixeme/selfpost/internal/store" +) + +// fakeInbound is the journal and the inbound-domain list of one test. +type fakeInbound struct { + domains map[string]bool + rows []store.InboundSpamEntry + insertErr error + lookupErr error +} + +func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error { + if f.insertErr != nil { + return f.insertErr + } + f.rows = append(f.rows, e) + return nil +} + +func (f *fakeInbound) IsInboundDomain(name string) (bool, error) { + if f.lookupErr != nil { + return false, f.lookupErr + } + return f.domains[name], nil +} + +// A header of the kind the documented rspamd sidecar adds. +const spamdResult = "default: False [7.20 / 15.00];\n\tBAYES_SPAM(5.10)[100.00%];\n\tMIME_GOOD(-0.10)[text/plain];\n\tARC_NA(0.00)[];\n\tRCVD_COUNT_ZERO(0.00)[0];\n\tR_DKIM_NA(2.20)[]" + +// message replays one message through s: envelope, headers in order, end. +func message(t *testing.T, s *inboundSession, from string, rcpts []string, headers [][2]string) milter.Response { + t.Helper() + if _, err := s.MailFrom(from, mods(nil)); err != nil { + t.Fatalf("MailFrom: %v", err) + } + for _, r := range rcpts { + if _, err := s.RcptTo(r, mods(nil)); err != nil { + t.Fatalf("RcptTo: %v", err) + } + } + for _, h := range headers { + if _, err := s.Header(h[0], h[1], mods(nil)); err != nil { + t.Fatalf("Header: %v", err) + } + } + resp, err := s.Body(mods(map[string]string{"i": "ABC123"})) + if err != nil { + t.Fatalf("Body: %v", err) + } + return resp +} + +func inboundSessionFor(t *testing.T, rec InboundStore, filter bool) *inboundSession { + t.Helper() + s := &inboundSession{rec: rec, filter: filter} + if _, err := s.Connect("mx.example.net", "tcp4", 0, net.ParseIP("203.0.113.7"), mods(nil)); err != nil { + t.Fatalf("Connect: %v", err) + } + return s +} + +// With no filter attached an accepted message is a plain row, one per +// recipient, decided by nobody. +func TestInboundRecordsAcceptedMessage(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, false) + resp := message(t, s, " SIZE=1200", + []string{"", "b@IN.example.com"}, + [][2]string{{"Subject", "=?utf-8?Q?Sch=C3=B6n?="}}) + if resp != milter.RespAccept { + t.Errorf("end-of-message response = %v, want accept", resp) + } + if len(rec.rows) != 2 { + t.Fatalf("rows = %+v, want one per recipient", rec.rows) + } + want := store.InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.7", + From: "sender@example.net", To: "a@in.example.com", Subject: "Schön", + Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost, + } + if rec.rows[0] != want { + t.Errorf("row = %+v\nwant %+v", rec.rows[0], want) + } + if rec.rows[1].To != "b@IN.example.com" || rec.rows[1].InboundDomain != "in.example.com" { + t.Errorf("second row = %+v", rec.rows[1]) + } +} + +// Port 25 also takes DMARC reports for the server's own name. They are not +// the relay's mail and the journal says nothing about them. +func TestInboundSkipsRecipientsOutsideInboundDomains(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, false) + message(t, s, "noreply-dmarc@google.com", + []string{"dmarc-example.com@mail.example.org", "a@in.example.com"}, nil) + if len(rec.rows) != 1 || rec.rows[0].To != "a@in.example.com" { + t.Errorf("rows = %+v, want the inbound recipient alone", rec.rows) + } +} + +// With the filter attached its verdict headers become the engine and detail. +func TestInboundReadsTheFiltersVerdict(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, true) + message(t, s, "sender@example.net", []string{"a@in.example.com"}, [][2]string{ + {"Subject", "Cheap pills"}, + {"X-Rspamd-Action", "add header"}, + {"X-Spamd-Result", spamdResult}, + }) + if len(rec.rows) != 1 { + t.Fatalf("rows = %+v", rec.rows) + } + got := rec.rows[0] + if got.Engine != store.SpamEngineRspamd || got.Decision != store.SpamDecisionAccept { + t.Errorf("engine %q, decision %q", got.Engine, got.Decision) + } + if want := "add header, score 7.20 / 15.00; BAYES_SPAM(5.10) MIME_GOOD(-0.10) R_DKIM_NA(2.20)"; got.Detail != want { + t.Errorf("detail = %q\nwant %q", got.Detail, want) + } +} + +func TestInboundQuarantineAction(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, true) + message(t, s, "sender@example.net", []string{"a@in.example.com"}, + [][2]string{{"x-rspamd-action", "quarantine"}}) + if len(rec.rows) != 1 || rec.rows[0].Decision != store.SpamDecisionQuarantine || rec.rows[0].Detail != "quarantine" { + t.Errorf("rows = %+v", rec.rows) + } +} + +// Without a filter in front, a verdict header is the sender's own writing. +func TestInboundIgnoresVerdictHeadersWithoutAFilter(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, false) + message(t, s, "sender@example.net", []string{"a@in.example.com"}, [][2]string{ + {"X-Rspamd-Action", "no action"}, + {"X-Spamd-Result", spamdResult}, + }) + if len(rec.rows) != 1 || rec.rows[0].Engine != store.SpamEngineSelfPost || rec.rows[0].Detail != "" { + t.Errorf("rows = %+v", rec.rows) + } +} + +// The filter is attached fail-open, so with it down the headers are again the +// sender's: nothing of them reaches the journal except through the patterns. +func TestRspamdVerdictKeepsOnlyWhatItRecognises(t *testing.T) { + action, detail := rspamdVerdict("", "default: False [1 / 15]; HTML lower_case(9.9) \x1b[31m") + if action != "" { + t.Errorf("action = %q, want it dropped", action) + } + if detail != "score 1 / 15" { + t.Errorf("detail = %q", detail) + } + + var many strings.Builder + for i := 0; i < 400; i++ { + many.WriteString("SYMBOL_WITH_A_LONG_NAME_" + strings.Repeat("X", 30) + "(1.00)[];") + } + if _, detail := rspamdVerdict("reject", many.String()); len(detail) > verdictMaxBytes || strings.Count(detail, "(") > verdictMaxSymbols { + t.Errorf("detail of %d bytes with %d symbols", len(detail), strings.Count(detail, "(")) + } + if action, _ := rspamdVerdict(strings.Repeat("a", 33), ""); action != "" { + t.Errorf("an action of 33 letters was kept") + } +} + +// A second message on the same connection starts clean. +func TestInboundResetsBetweenMessages(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, true) + message(t, s, "one@example.net", []string{"a@in.example.com"}, [][2]string{ + {"Subject", "First"}, {"X-Rspamd-Action", "add header"}, + }) + message(t, s, "two@example.net", []string{"b@in.example.com"}, nil) + if len(rec.rows) != 2 { + t.Fatalf("rows = %+v", rec.rows) + } + want := store.InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.7", + From: "two@example.net", To: "b@in.example.com", + Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost, + } + if rec.rows[1] != want { + t.Errorf("second row = %+v\nwant %+v", rec.rows[1], want) + } + + // An aborted transaction leaves nothing behind for the next one either. + if _, err := s.MailFrom("three@example.net", mods(nil)); err != nil { + t.Fatal(err) + } + if _, err := s.RcptTo("c@in.example.com", mods(nil)); err != nil { + t.Fatal(err) + } + if err := s.Abort(mods(nil)); err != nil { + t.Fatal(err) + } + if len(rec.rows) != 2 || len(s.rcpts) != 0 { + t.Errorf("after an abort: rows %d, pending recipients %v", len(rec.rows), s.rcpts) + } +} + +// The journal failing is never the mail's problem. +func TestInboundAcceptsWhenTheStoreFails(t *testing.T) { + for name, rec := range map[string]*fakeInbound{ + "insert": {domains: map[string]bool{"in.example.com": true}, insertErr: errors.New("boom")}, + "lookup": {lookupErr: errors.New("boom")}, + } { + s := inboundSessionFor(t, rec, false) + if resp := message(t, s, "sender@example.net", []string{"a@in.example.com"}, nil); resp != milter.RespAccept { + t.Errorf("%s failure: response = %v, want accept", name, resp) + } + } +} diff --git a/internal/store/inboundspam.go b/internal/store/inboundspam.go new file mode 100644 index 0000000..2101806 --- /dev/null +++ b/internal/store/inboundspam.go @@ -0,0 +1,72 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "strings" + "time" +) + +// Decisions of the inbound antispam journal. Kept in sync with the CHECK +// constraint of inbound_spam_log. +const ( + SpamDecisionAccept = "accept" + SpamDecisionReject = "reject" + SpamDecisionTempfail = "tempfail" + SpamDecisionQuarantine = "quarantine" +) + +// Who made a decision of the inbound antispam journal. Kept in sync with the +// CHECK constraint of inbound_spam_log. +const ( + SpamEngineRspamd = "rspamd" // the external filter + SpamEnginePostfix = "postfix" // Postfix's own policy: unknown recipient, size, rate + SpamEngineSelfPost = "selfpost" // nobody: relayed with no filter attached +) + +// InboundSpamEntry is one row of the inbound antispam journal: what happened +// to a message offered on port 25 for one recipient. Metadata only - the +// message itself is relayed, not stored. Subject and Detail may be empty. +type InboundSpamEntry struct { + InboundDomain string + ClientIP string + From string + To string + Subject string + Decision string + Engine string + Detail string +} + +// InsertInboundSpam records one journal row. It is called from the inbound +// journal milter's hot path and from the log tailer, so it returns any error +// for the caller to log; neither of them may hold up mail over it. +func (s *Store) InsertInboundSpam(e InboundSpamEntry) error { + _, err := s.db.Exec( + `INSERT INTO inbound_spam_log + (inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + e.InboundDomain, e.ClientIP, e.From, e.To, e.Subject, e.Decision, e.Engine, e.Detail, + time.Now().UTC().Format(time.RFC3339), + ) + if err != nil { + return fmt.Errorf("insert inbound_spam_log: %w", err) + } + return nil +} + +// IsInboundDomain reports whether name is a configured inbound domain. Port 25 +// also takes mail that is not the relay's - DMARC reports for the server's own +// name - and the journal keeps rows for inbound domains only. +func (s *Store) IsInboundDomain(name string) (bool, error) { + var one int + err := s.db.QueryRow("SELECT 1 FROM inbound_domains WHERE name = ?", strings.ToLower(name)).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("look up inbound domain: %w", err) + } + return true, nil +} diff --git a/internal/store/inboundspam_test.go b/internal/store/inboundspam_test.go new file mode 100644 index 0000000..92e5408 --- /dev/null +++ b/internal/store/inboundspam_test.go @@ -0,0 +1,61 @@ +package store + +import ( + "testing" + "time" +) + +func TestInsertInboundSpam(t *testing.T) { + st := openTestStore(t) + e := InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.7", + From: "sender@example.net", To: "a@in.example.com", Subject: "Hello", + Decision: SpamDecisionAccept, Engine: SpamEngineRspamd, Detail: "no action, score 0.10 / 15.00", + } + if err := st.InsertInboundSpam(e); err != nil { + t.Fatalf("InsertInboundSpam: %v", err) + } + var got InboundSpamEntry + var created string + err := st.db.QueryRow( + "SELECT inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at FROM inbound_spam_log", + ).Scan(&got.InboundDomain, &got.ClientIP, &got.From, &got.To, &got.Subject, &got.Decision, &got.Engine, &got.Detail, &created) + if err != nil { + t.Fatalf("read back: %v", err) + } + if got != e { + t.Errorf("row = %+v\nwant %+v", got, e) + } + if at, err := time.Parse(time.RFC3339, created); err != nil || time.Since(at) > time.Minute { + t.Errorf("created_at = %q (%v)", created, err) + } + + // Every decision and engine the code names is one the table takes. + for _, d := range []string{SpamDecisionAccept, SpamDecisionReject, SpamDecisionTempfail, SpamDecisionQuarantine} { + for _, eng := range []string{SpamEngineRspamd, SpamEnginePostfix, SpamEngineSelfPost} { + e.Decision, e.Engine = d, eng + if err := st.InsertInboundSpam(e); err != nil { + t.Errorf("decision %s by %s: %v", d, eng, err) + } + } + } + e.Decision = "bounce" + if err := st.InsertInboundSpam(e); err == nil { + t.Error("an unknown decision was stored") + } +} + +func TestIsInboundDomain(t *testing.T) { + st := openTestStore(t) + if _, err := st.AddInboundDomain("in.example.com"); err != nil { + t.Fatalf("AddInboundDomain: %v", err) + } + for name, want := range map[string]bool{ + "in.example.com": true, "IN.Example.COM": true, + "example.com": false, "": false, "other.example.com": false, + } { + if got, err := st.IsInboundDomain(name); err != nil || got != want { + t.Errorf("IsInboundDomain(%q) = %v, %v; want %v", name, got, err, want) + } + } +} diff --git a/test/e2e/port25_config_check.go b/test/e2e/port25_config_check.go index 25905fd..5ccef6a 100644 --- a/test/e2e/port25_config_check.go +++ b/test/e2e/port25_config_check.go @@ -47,9 +47,10 @@ var port25Cases = []port25Case{ {param: "smtpd_sasl_auth_enable", contains: "no", wantExact: true}, {param: "smtpd_relay_restrictions", contains: "reject_unauth_destination", wantExact: true}, {param: "smtpd_recipient_restrictions", contains: "reject_unauth_destination,reject_unlisted_recipient", wantExact: true}, - // No antispam milter configured: the submission chain (strict - // OpenDKIM + journal-milter) must not reach inbound mail. - {param: "smtpd_milters", contains: "", wantExact: true}, + // No antispam milter configured: the inbound journal milter + // alone. The submission chain (strict OpenDKIM + journal-milter) + // must not reach inbound mail. + {param: "smtpd_milters", contains: "{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true}, }, }, { @@ -72,7 +73,7 @@ var port25Cases = []port25Case{ }, check: []optionCheck{ {param: "smtpd_recipient_restrictions", contains: "check_recipient_access,texthash:/data/postfix/dmarc_recipients,reject_unauth_destination,reject_unlisted_recipient", wantExact: true}, - {param: "smtpd_milters", contains: "", wantExact: true}, + {param: "smtpd_milters", contains: "{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true}, }, // Both on: the report addresses sit on a relay domain, and the relay's // unlisted-recipient check runs after the restriction list. Unless their @@ -94,7 +95,7 @@ var port25Cases = []port25Case{ check: []optionCheck{ // Exact equality also covers the absence of OpenDKIM/journal-milter // sockets — anything else there would fail the comparison. - {param: "smtpd_milters", contains: "{inet:antispam:11332,default_action=accept}", wantExact: true}, + {param: "smtpd_milters", contains: "{inet:antispam:11332,default_action=accept},{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true}, }, }, } diff --git a/test/e2e/port25_features_test.go b/test/e2e/port25_features_test.go index 66ef830..18f4557 100644 --- a/test/e2e/port25_features_test.go +++ b/test/e2e/port25_features_test.go @@ -265,6 +265,30 @@ func testInboundRelay(t *testing.T, sc *scenario) { t.Fatalf("mail for the inbound domain never reached its upstream: %v\n%s", err, mailLogTail(h, "postfix/")) } + // The inbound journal milter saw it go through. No page shows the journal + // yet, so the row is read where it is kept. No filter is attached on the + // stand: the row is an accept by nobody. + want := fmt.Sprintf("%s|someone@stranger.e2e.test|%s|e2e %s|accept|selfpost", inboundDomain, inboundRecipient, token) + var rows string + for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(time.Second) { + rows, err = h.execIn("selfpost", "python3", "-c", + `import sqlite3; print("\n".join("|".join(r) for r in sqlite3.connect("file:/data/selfpost.db?mode=ro", uri=True).execute("select inbound_domain, from_addr, to_addr, subject, decision, engine from inbound_spam_log")))`) + if err != nil { + t.Fatalf("read the inbound journal: %v\n%s", err, rows) + } + if strings.Contains(rows, want) { + break + } + if time.Now().After(deadline) { + t.Fatalf("no journal row for the relayed message, want %q, journal:\n%s\n%s", want, rows, h.logs("selfpost")) + } + } + // A report mailed to the server's own name earlier in the run came through + // the same listener and is not the relay's mail. + if n := strings.Count(strings.TrimSpace(rows), "\n") + 1; n != 1 { + t.Fatalf("the journal holds %d rows, want the relayed message alone:\n%s", n, rows) + } + for _, to := range []string{"nobody@" + inboundDomain, "someone@elsewhere.e2e.test"} { res := deliverPort25("someone@stranger.e2e.test", to, message(to, uniqueToken("refused"))) if res.rcptErr == nil {