diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d9b0f3..4516a9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version ## [Unreleased] +### Changed + +- **Port 25 logs as `postfix/inbound`.** Its lines in the mail log (Server › + System log) now read `postfix/inbound/smtpd[…]` where they read + `postfix/smtpd[…]`; the submission ports keep `postfix/smtpd`. This is how a + refusal on port 25 is told from one on 465 / 587. + ### Added - Schema: migration `0002_inbound_spam_log.sql`, the first after the 2.0 @@ -16,8 +23,14 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version 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. + accepts mail when it is down. There is no page for the journal yet. +- **Refused inbound mail is recorded too**, from the mail log: a recipient + Postfix refused at an inbound domain (unknown user, sender domain not + found), and a message the spam filter rejected, deferred, held or + discarded. The row is dated by its log line and carries the reply text. A + refusal to relay for a domain that is not an inbound one is not recorded. +- The inbound journal is pruned every six hours: rows older than 90 days + (`INBOUND_SPAM_LOG_RETENTION_DAYS`, 7-365) and anything beyond 10 000 rows. - 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/` diff --git a/build/postfix-config.sh b/build/postfix-config.sh index b46dbce..bbf5109 100644 --- a/build/postfix-config.sh +++ b/build/postfix-config.sh @@ -412,8 +412,12 @@ if [ "${INBOUND_RELAY_ENABLE}" = "true" ] || [ "${DMARC_REPORTS_ENABLE}" = "true fi fi + # syslog_name: this listener's lines read postfix/inbound/smtpd, the + # submission ports' postfix/smtpd. The log tailer files a refusal on port 25 + # into the inbound journal and has to tell the two apart. postconf -M "smtp/inet=smtp inet n - n - - smtpd" postconf -P \ + "smtp/inet/syslog_name=postfix/inbound" \ "smtp/inet/smtpd_sasl_auth_enable=no" \ "smtp/inet/smtpd_tls_auth_only=no" \ "smtp/inet/smtpd_sender_login_maps=" \ diff --git a/cmd/panel/main.go b/cmd/panel/main.go index 90e5012..1adc4d3 100644 --- a/cmd/panel/main.go +++ b/cmd/panel/main.go @@ -99,7 +99,9 @@ type config struct { // 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 + // Retention of the inbound journal until Server > Settings says otherwise. + inboundSpamLogRetentionDays int + dmarcEnabled bool } func loadConfig() config { @@ -166,6 +168,8 @@ func loadConfig() config { 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"), + + inboundSpamLogRetentionDays: envInt("INBOUND_SPAM_LOG_RETENTION_DAYS", store.InboundSpamLogRetentionDaysDefault), // Optional DMARC aggregate ingest on port 25. Off unless exactly "true". dmarcEnabled: os.Getenv("DMARC_REPORTS_ENABLE") == "true", } @@ -296,6 +300,22 @@ func run() error { if err := st.EnsureSendLogRetentionDays(cfg.retentionDays); err != nil { return fmt.Errorf("bootstrap send-log retention: %w", err) } + // With the inbound relay on, the tailer also writes the inbound journal's + // rows for refused mail and prunes that journal. + var inboundTail *logtail.Inbound + if cfg.inboundEnabled { + if err := st.EnsureInboundSpamLogRetentionDays(cfg.inboundSpamLogRetentionDays); err != nil { + return fmt.Errorf("bootstrap inbound journal retention: %w", err) + } + inboundTail = &logtail.Inbound{Store: st, Retention: func() int { + days, err := st.GetInboundSpamLogRetentionDays(cfg.inboundSpamLogRetentionDays) + if err != nil { + log.Printf("inbound journal retention: %v", err) + return cfg.inboundSpamLogRetentionDays + } + return days + }} + } var wg sync.WaitGroup errc := make(chan error, 5) @@ -314,7 +334,7 @@ func run() error { return cfg.retentionDays } return days - }) + }, inboundTail) }}, {"rate-limit-recalc", func(ctx context.Context) error { runAutoRateLimitRecalc(ctx, cfg, st) diff --git a/deploy/.env.example b/deploy/.env.example index c9c34a3..e6cbf45 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -30,6 +30,10 @@ DMARC_REPORTS_ENABLE=false # INBOUND_ANTISPAM_MILTER=inet:antispam:11332 # INBOUND_ANTISPAM_MILTER_ACTION=accept +# Days the inbound journal keeps a row (7-365) until the panel says otherwise; +# it never holds more than 10 000 rows. +# INBOUND_SPAM_LOG_RETENTION_DAYS=90 + # Coarse per-client-IP limit and message size on inbound smtpd (port 25). # INBOUND_RATE_LIMIT_MESSAGES_PER_IP=20 # INBOUND_MESSAGE_SIZE_LIMIT=26214400 diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 4e17146..b3d4bd6 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -47,6 +47,7 @@ services: # How long the send log keeps rows before the background sweep deletes # them — the main driver of /data growth over time. SEND_LOG_RETENTION_DAYS: "${SEND_LOG_RETENTION_DAYS:-90}" + INBOUND_SPAM_LOG_RETENTION_DAYS: "${INBOUND_SPAM_LOG_RETENTION_DAYS:-90}" # Sliding idle timeout for the panel login session, in days. PANEL_SESSION_IDLE_DAYS: "${PANEL_SESSION_IDLE_DAYS:-7}" # Resolvers the panel's PTR/SPF/DKIM/DMARC checks query directly, so they diff --git a/docs/architecture.md b/docs/architecture.md index 4d78e62..9632939 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -89,6 +89,19 @@ smtp/inet` removes Debian's stock smtpd). Outbound delivery still uses the 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. + Refusals come from `mail.log`: this listener logs as `postfix/inbound` + (`syslog_name`), and the log tailer turns a `reject:` line of + `postfix/inbound/smtpd` (Postfix's own policy) and a `milter-reject` / + `milter-hold` / `milter-discard` line of `cleanup` (the filter's verdict, + counted only for a queue id that listener opened) into a row, dated by the + line, when the recipient is at an inbound domain + ([internal/logtail/inbound.go](../internal/logtail/inbound.go)). A line + names one recipient, so a message refused at end-of-message is one row + whatever its recipient count, and a line read twice (the tailer saves its + place once a second; an unclean stop can lose that second) is two rows. The + journal is pruned every six hours: rows + older than the `inbound_spam_log_retention_days` setting, then down to + 10 000. **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 f3827d0..ae963f0 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_SPAM_LOG_RETENTION_DAYS` | How long the inbound journal keeps a row, 7-365 days; it also never holds more than 10 000 rows. Seeds the setting on the first start with the inbound relay on. | `90` | `.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` | diff --git a/docs/plans/inbound-antispam-panel.md b/docs/plans/inbound-antispam-panel.md index b9add80..1cf3cc1 100644 --- a/docs/plans/inbound-antispam-panel.md +++ b/docs/plans/inbound-antispam-panel.md @@ -246,7 +246,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** - [x] Inbound journal-milter + Postfix wiring — **Composer** -- [ ] mail.log tailer: inbound reject rows — **Composer** +- [x] 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** - [ ] Backup includes `/data/antispam/` — **Composer** diff --git a/docs/roadmap.md b/docs/roadmap.md index 1ae9f4b..4ebbb4b 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 | 3/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) | +| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | in progress | 4/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/logtail/inbound.go b/internal/logtail/inbound.go new file mode 100644 index 0000000..77aaeea --- /dev/null +++ b/internal/logtail/inbound.go @@ -0,0 +1,231 @@ +package logtail + +import ( + "context" + "log" + "regexp" + "strings" + "time" + + "github.com/mixeme/selfpost/internal/store" +) + +// InboundStore is what the tailer needs of the inbound antispam journal: +// writing the rows for refused mail, knowing which domains are inbound, and +// pruning. *store.Store satisfies it. +type InboundStore interface { + InsertInboundSpam(e store.InboundSpamEntry) error + IsInboundDomain(name string) (bool, error) + PruneInboundSpam(cutoff time.Time, keep int) (int64, error) +} + +// Inbound turns on the inbound half of the tailer: journal rows for mail that +// port 25 refused, and the journal's retention. Run takes nil when the inbound +// relay is off. +type Inbound struct { + Store InboundStore + Retention RetentionDays +} + +// The inbound journal milter writes a row for mail that was accepted; mail +// that was refused never reaches it, and what is known about it is one line +// of mail.log: +// +// postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from mx.example.net[203.0.113.9]: 550 5.1.1 : Recipient address rejected: User unknown in relay recipient table; from= to= proto=ESMTP helo= +// postfix/cleanup[52]: 4A1B2C3D: milter-reject: END-OF-MESSAGE from mx.example.net[203.0.113.9]: 5.7.1 Spam message rejected; from= to= proto=ESMTP helo= +// +// The first is Postfix's own policy, written by the port-25 smtpd, which +// build/postfix-config.sh names postfix/inbound so that its lines are told +// from those of the submission ports. The second is the filter's verdict at +// end-of-message, which cleanup(8) writes under one name for every listener - +// so it counts only for a queue id the port-25 smtpd was seen to open +// (inboundClientRe). +// +// Both are matched from the head of the line, after the timestamp and host +// name: the tail of such a line is the sender's own words (HELO name, +// addresses), and a pattern that could start anywhere would let him write +// journal rows. +var ( + inboundRejectRe = regexp.MustCompile(`^\S+ postfix/(inbound/smtpd|cleanup)\[\d+\]: ([0-9A-Za-z]+): (reject|milter-reject|milter-hold|milter-discard): \S+ from [^\[\s]*\[([0-9A-Fa-f:.]+)\]: (.*); from=<([^>]*)> to=<([^>]*)> proto=\S+(?: helo=<[^>]*>)?$`) + inboundClientRe = regexp.MustCompile(`^\S+ postfix/inbound/smtpd\[\d+\]: ([0-9A-Za-z]+): client=`) +) + +// A row holds one line of reason; Postfix's reply text carries the far end's +// bytes and is cut and reduced to printable ASCII like any of them. +const inboundDetailMax = 255 + +// inboundQueueIDs is how many port-25 queue ids are remembered while waiting +// for cleanup's verdict line on them. A message is in that state for the +// seconds its DATA takes; the cap only bounds a flood. +const inboundQueueIDs = 4096 + +// inboundJournal is the per-run state of the inbound half. +type inboundJournal struct { + st InboundStore + // port25 holds the queue ids the port-25 smtpd opened and cleanup has not + // reported on yet; order is their arrival, for dropping the oldest. + port25 map[string]struct{} + order []string +} + +func newInboundJournal(st InboundStore) *inboundJournal { + return &inboundJournal{st: st, port25: make(map[string]struct{})} +} + +// handle looks at one mail.log line and writes a journal row when it is a +// refusal of mail for an inbound domain. Errors are logged; the tailer goes on. +func (j *inboundJournal) handle(line string, now time.Time) { + at, rest := lineTime(line, now) + if m := inboundClientRe.FindStringSubmatch(rest); m != nil { + j.remember(m[1]) + return + } + m := inboundRejectRe.FindStringSubmatch(rest) + if m == nil { + return + } + program, queueID, kind, clientIP, reason, from, to := m[1], m[2], m[3], m[4], m[5], m[6], m[7] + if program == "cleanup" { + if _, ok := j.port25[queueID]; !ok { + return // a submission-port message, or one opened before this run + } + delete(j.port25, queueID) + } + + // Only mail for an inbound domain is the relay's. "Relay access denied" + // for a stranger's domain is every scanner on the internet, and a line + // with no recipient (size limit at MAIL) belongs to no domain. + domain := "" + if i := strings.LastIndexByte(to, '@'); i >= 0 { + domain = strings.ToLower(to[i+1:]) + } + if domain == "" { + return + } + ok, err := j.st.IsInboundDomain(domain) + if err != nil { + log.Printf("log-tailer: inbound journal: look up %s: %v", domain, err) + return + } + if !ok { + return + } + + e := store.InboundSpamEntry{ + InboundDomain: domain, + ClientIP: clientIP, + From: from, + To: to, + Decision: store.SpamDecisionReject, + Engine: store.SpamEngineRspamd, + Detail: printable(reason, inboundDetailMax), + At: at, + } + switch { + case kind == "reject": + e.Engine = store.SpamEnginePostfix + case kind == "milter-hold": + e.Decision = store.SpamDecisionQuarantine + } + // "450 4.7.1 …" from Postfix, "4.7.1 …" from a milter: either way a reply + // that starts with 4 asks the sender to come back. + if e.Decision == store.SpamDecisionReject && strings.HasPrefix(reason, "4") { + e.Decision = store.SpamDecisionTempfail + } + if err := j.st.InsertInboundSpam(e); err != nil { + log.Printf("log-tailer: inbound journal: record %s -> %s: %v", from, to, err) + } +} + +func (j *inboundJournal) remember(queueID string) { + if _, ok := j.port25[queueID]; ok { + return + } + if len(j.order) >= inboundQueueIDs { + delete(j.port25, j.order[0]) + j.order = j.order[1:] + } + j.port25[queueID] = struct{}{} + j.order = append(j.order, queueID) +} + +// printable cuts s to max bytes and replaces everything outside printable +// ASCII. +func printable(s string, max int) string { + if len(s) > max { + s = s[:max] + } + b := []byte(s) + for i, c := range b { + if c < 0x20 || c > 0x7e { + b[i] = '?' + } + } + return string(b) +} + +// lineTime reads the time at the head of a mail.log line and returns it with +// the rest of the line. A refusal is dated by its line and not by when the +// tailer got to it: after the panel was down the two are hours apart. The +// syslog format carries neither year nor zone - the year is taken from now +// (a stamp that would then lie in the future is last year's), the zone is the +// container's, which is what postlogd wrote it in. A line with no readable +// stamp is dated now. +func lineTime(line string, now time.Time) (time.Time, string) { + if m := isoStampRe.FindStringSubmatch(line); m != nil { + rest := strings.TrimSpace(line[len(m[0]):]) + if t, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(m[0])); err == nil { + return t.UTC(), rest + } + return now, rest + } + if m := syslogStampRe.FindStringSubmatch(line); m != nil { + rest := strings.TrimSpace(line[len(m[0]):]) + stamp := strings.Join(strings.Fields(m[1]), " ") + t, err := time.ParseInLocation("Jan 2 15:04:05", stamp, time.Local) + if err != nil { + return now, rest + } + t = t.AddDate(now.In(time.Local).Year(), 0, 0) + if t.After(now.Add(24 * time.Hour)) { + t = t.AddDate(-1, 0, 0) + } + return t.UTC(), rest + } + return now, line +} + +// The journal keeps at most this many rows whatever their age: refusals are +// written at the rate strangers knock on port 25. +const inboundSpamLogMaxRows = 10000 + +// inboundRetentionLoop prunes the inbound journal at once and then on the +// cadence of the send log. +func inboundRetentionLoop(ctx context.Context, in *Inbound) { + prune := func() { + days := in.Retention() + if days <= 0 { + days = store.InboundSpamLogRetentionDaysDefault + } + n, err := in.Store.PruneInboundSpam(time.Now().UTC().AddDate(0, 0, -days), inboundSpamLogMaxRows) + if err != nil { + log.Printf("log-tailer: inbound journal prune: %v", err) + return + } + if n > 0 { + log.Printf("log-tailer: pruned %d inbound journal rows (older than %d days or beyond %d rows)", n, days, inboundSpamLogMaxRows) + } + } + + prune() + t := time.NewTicker(retentionInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + prune() + } + } +} diff --git a/internal/logtail/inbound_test.go b/internal/logtail/inbound_test.go new file mode 100644 index 0000000..950fce2 --- /dev/null +++ b/internal/logtail/inbound_test.go @@ -0,0 +1,218 @@ +package logtail + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/mixeme/selfpost/internal/store" +) + +type fakeInboundStore struct { + domains map[string]bool + rows []store.InboundSpamEntry + lookupErr error + pruned []int +} + +func (f *fakeInboundStore) InsertInboundSpam(e store.InboundSpamEntry) error { + f.rows = append(f.rows, e) + return nil +} + +func (f *fakeInboundStore) IsInboundDomain(name string) (bool, error) { + if f.lookupErr != nil { + return false, f.lookupErr + } + return f.domains[name], nil +} + +func (f *fakeInboundStore) PruneInboundSpam(_ time.Time, keep int) (int64, error) { + f.pruned = append(f.pruned, keep) + return 0, nil +} + +func inboundFixture() (*inboundJournal, *fakeInboundStore) { + st := &fakeInboundStore{domains: map[string]bool{"in.example.com": true}} + return newInboundJournal(st), st +} + +var tailNow = time.Date(2026, 10, 11, 18, 0, 0, 0, time.UTC) + +const ( + head = "2026-10-11T12:00:05.123456+00:00 mail " + unknownRcp = head + "postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from mx.example.net[203.0.113.9]: 550 5.1.1 : Recipient address rejected: User unknown in relay recipient table; from= to= proto=ESMTP helo=" +) + +// Postfix's own refusal of a recipient at an inbound domain is a row, dated by +// its line. +func TestInboundRejectByPostfix(t *testing.T) { + j, st := inboundFixture() + j.handle(unknownRcp, tailNow) + if len(st.rows) != 1 { + t.Fatalf("rows = %+v", st.rows) + } + want := store.InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.9", + From: "a@example.net", To: "nobody@in.example.com", + Decision: store.SpamDecisionReject, Engine: store.SpamEnginePostfix, + Detail: "550 5.1.1 : Recipient address rejected: User unknown in relay recipient table", + At: time.Date(2026, 10, 11, 12, 0, 5, 123456000, time.UTC), + } + if st.rows[0] != want { + t.Errorf("row = %+v\nwant %+v", st.rows[0], want) + } +} + +func TestInboundTempfailByPostfix(t *testing.T) { + j, st := inboundFixture() + j.handle(head+"postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from unknown[2001:db8::9]: 450 4.1.8 : Sender address rejected: Domain not found; from= to= proto=ESMTP helo=", tailNow) + if len(st.rows) != 1 || st.rows[0].Decision != store.SpamDecisionTempfail || st.rows[0].ClientIP != "2001:db8::9" || st.rows[0].InboundDomain != "in.example.com" { + t.Errorf("rows = %+v", st.rows) + } +} + +// The filter's verdict is written by cleanup, under one name for every +// listener: it is the journal's only for a queue id port 25 opened. +func TestInboundMilterVerdictNeedsAPort25QueueID(t *testing.T) { + const verdict = "postfix/cleanup[52]: 4A1B2C3D: milter-reject: END-OF-MESSAGE from mx.example.net[203.0.113.9]: 5.7.1 Spam message rejected; from= to= proto=ESMTP helo=" + + j, st := inboundFixture() + // Opened by a submission port: postfix/smtpd, not postfix/inbound/smtpd. + j.handle(head+"postfix/smtpd[20]: 4A1B2C3D: client=app.example.com[203.0.113.4], sasl_method=PLAIN, sasl_username=app", tailNow) + j.handle(head+verdict, tailNow) + if len(st.rows) != 0 { + t.Fatalf("a submission-port verdict was journalled: %+v", st.rows) + } + + j.handle(head+"postfix/inbound/smtpd[41]: 4A1B2C3D: client=mx.example.net[203.0.113.9]", tailNow) + j.handle(head+verdict, tailNow) + if len(st.rows) != 1 { + t.Fatalf("rows = %+v", st.rows) + } + got := st.rows[0] + if got.Decision != store.SpamDecisionReject || got.Engine != store.SpamEngineRspamd || got.Detail != "5.7.1 Spam message rejected" { + t.Errorf("row = %+v", got) + } + // One verdict per message: the id is forgotten with it. + j.handle(head+verdict, tailNow) + if len(st.rows) != 1 { + t.Errorf("the same queue id was journalled twice") + } +} + +func TestInboundMilterDecisions(t *testing.T) { + for kind, c := range map[string]struct{ reply, decision string }{ + "milter-reject": {"4.7.1 Try again later", store.SpamDecisionTempfail}, + "milter-hold": {"milter triggers HOLD action", store.SpamDecisionQuarantine}, + "milter-discard": {"milter triggers DISCARD action", store.SpamDecisionReject}, + } { + j, st := inboundFixture() + j.handle(head+"postfix/inbound/smtpd[41]: Q1: client=mx.example.net[203.0.113.9]", tailNow) + j.handle(head+"postfix/cleanup[52]: Q1: "+kind+": END-OF-MESSAGE from mx.example.net[203.0.113.9]: "+c.reply+"; from= to= proto=ESMTP helo=", tailNow) + if len(st.rows) != 1 || st.rows[0].Decision != c.decision || st.rows[0].Engine != store.SpamEngineRspamd { + t.Errorf("%s: rows = %+v", kind, st.rows) + } + } +} + +// What is not the relay's mail, or is not a refusal at all, leaves no row. +func TestInboundIgnores(t *testing.T) { + for name, line := range map[string]string{ + "another domain (relay access denied)": head + "postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 554 5.7.1 : Relay access denied; from= to= proto=ESMTP helo=", + "no recipient (size limit at MAIL)": head + "postfix/inbound/smtpd[41]: NOQUEUE: reject: MAIL from x[203.0.113.9]: 552 5.3.4 Message size exceeds fixed limit; proto=ESMTP helo=", + "a submission port": head + "postfix/smtpd[20]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 553 5.7.1 : Sender address rejected: not owned by user app; from= to= proto=ESMTP helo=", + "a delivery line": head + "postfix/smtp[26]: 41E862C00D9E: to=, relay=mx[203.0.113.9]:25, dsn=2.0.0, status=sent (250 OK)", + "a connect line": head + "postfix/inbound/smtpd[41]: connect from mx.example.net[203.0.113.9]", + // The HELO name is the sender's; a line cannot be started from inside one. + "a refusal spelled inside another line": head + "postfix/inbound/smtpd[41]: warning: hostname x does not resolve: mail postfix/inbound/smtpd[1]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 no; from= to= proto=ESMTP helo=", + "an address with a bracket in it": head + "postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 no; from= to= proto=X helo=@example.net> to= proto=ESMTP helo=", + } { + j, st := inboundFixture() + j.handle(line, tailNow) + if len(st.rows) != 0 { + t.Errorf("%s: journalled %+v", name, st.rows) + } + } +} + +func TestInboundDetailIsCutAndPrintable(t *testing.T) { + j, st := inboundFixture() + j.handle(head+"postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 5.7.1 caf\xc3\xa9 \x1b[31m"+strings.Repeat("x", 400)+"; from= to= proto=ESMTP helo=", tailNow) + if len(st.rows) != 1 { + t.Fatalf("rows = %+v", st.rows) + } + d := st.rows[0].Detail + if len(d) != inboundDetailMax || !strings.HasPrefix(d, "550 5.7.1 caf?? ?[31mxxx") { + t.Errorf("detail (%d bytes) = %q", len(d), d) + } +} + +func TestInboundLookupFailureLeavesNoRow(t *testing.T) { + j, st := inboundFixture() + st.lookupErr = errors.New("boom") + j.handle(unknownRcp, tailNow) + if len(st.rows) != 0 { + t.Errorf("rows = %+v", st.rows) + } +} + +// The queue ids waiting for a verdict are bounded. +func TestInboundQueueIDsAreBounded(t *testing.T) { + j, _ := inboundFixture() + for i := 0; i < inboundQueueIDs+10; i++ { + j.remember("Q" + strings.Repeat("A", i%7) + string(rune('0'+i%10)) + time.Duration(i).String()) + } + if len(j.port25) != inboundQueueIDs || len(j.order) != inboundQueueIDs { + t.Errorf("remembered %d ids (%d in order), want %d", len(j.port25), len(j.order), inboundQueueIDs) + } +} + +func TestLineTime(t *testing.T) { + now := time.Date(2026, 1, 2, 10, 0, 0, 0, time.UTC) + for _, c := range []struct { + line string + want time.Time + rest string + }{ + {"2026-01-02T09:00:00.5+03:00 mail postfix/x: y", time.Date(2026, 1, 2, 6, 0, 0, 500000000, time.UTC), "mail postfix/x: y"}, + {"Jan 2 09:30:00 mail postfix/x: y", time.Date(2026, 1, 2, 9, 30, 0, 0, time.Local).UTC(), "mail postfix/x: y"}, + // December read in January is last year's. + {"Dec 31 23:59:00 mail postfix/x: y", time.Date(2025, 12, 31, 23, 59, 0, 0, time.Local).UTC(), "mail postfix/x: y"}, + {"no stamp here", now, "no stamp here"}, + } { + got, rest := lineTime(c.line, now) + if !got.Equal(c.want) || rest != c.rest { + t.Errorf("lineTime(%q) = %v, %q; want %v, %q", c.line, got, rest, c.want, c.rest) + } + } +} + +func TestInboundRetentionPrunesAtStart(t *testing.T) { + old := retentionInterval + retentionInterval = time.Hour + defer func() { retentionInterval = old }() + + st := &fakeInboundStore{} + ctx, cancel := contextWithCancel() + done := make(chan struct{}) + go func() { + inboundRetentionLoop(ctx, &Inbound{Store: st, Retention: func() int { return 0 }}) + close(done) + }() + deadline := time.Now().Add(2 * time.Second) + for len(st.pruned) == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + cancel() + <-done + if len(st.pruned) != 1 || st.pruned[0] != inboundSpamLogMaxRows { + t.Errorf("prune calls = %v", st.pruned) + } +} + +func contextWithCancel() (context.Context, context.CancelFunc) { + return context.WithCancel(context.Background()) +} diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go index b1e9cbd..8286572 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -107,9 +107,16 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) { // Run follows path and updates send-log statuses until ctx is cancelled, while // a background sweep prunes rows older than retention(). Reading resumes at // the offset the previous run persisted, so a restart parses the delivery lines -// written while the panel was down. It returns nil on a clean shutdown. -func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays) error { +// written while the panel was down. With inbound set it also writes the +// inbound journal's rows for refused mail and prunes that journal +// (inbound.go). It returns nil on a clean shutdown. +func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays, inbound *Inbound) error { go retentionLoop(ctx, st, retention) + var journal *inboundJournal + if inbound != nil { + go inboundRetentionLoop(ctx, inbound) + journal = newInboundJournal(inbound.Store) + } // The reconcile sweep must not run against a backlog the tailer has not // read yet: on a restart the log holds the very lines that resolve the rows @@ -119,6 +126,9 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa go reconcileLoop(ctx, st, caughtUp) return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) { + if journal != nil { + journal.handle(line, time.Now().UTC()) + } queueID, recipient, status, ok := parseDelivery(line) if !ok { return diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go index fa42b01..0d2059f 100644 --- a/internal/logtail/logtail_test.go +++ b/internal/logtail/logtail_test.go @@ -212,7 +212,7 @@ func TestFollowTailsAndRotates(t *testing.T) { cs := &captureStore{} ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- Run(ctx, path, cs, func() int { return 90 }) }() + go func() { done <- Run(ctx, path, cs, func() int { return 90 }, nil) }() // Give follow() time to open at EOF (it seeks to end immediately on start, // so the seed line above is ignored), then append a delivery line. @@ -295,7 +295,7 @@ func startRun(t *testing.T, path string, cs *captureStore) func() { t.Helper() ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- Run(ctx, path, cs, func() int { return 90 }) }() + go func() { done <- Run(ctx, path, cs, func() int { return 90 }, nil) }() // follow() opens and seeks on start; give it a moment before the caller // appends, so the append is not raced by the initial open. time.Sleep(50 * time.Millisecond) diff --git a/internal/store/inboundspam.go b/internal/store/inboundspam.go index 2101806..b4f5575 100644 --- a/internal/store/inboundspam.go +++ b/internal/store/inboundspam.go @@ -37,18 +37,25 @@ type InboundSpamEntry struct { Decision string Engine string Detail string + // At is when it happened; zero means now. The log tailer dates a refusal + // by its mail.log line, which it may read long after. + At time.Time } // 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 { + at := e.At + if at.IsZero() { + at = time.Now() + } _, 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), + at.UTC().Format(time.RFC3339), ) if err != nil { return fmt.Errorf("insert inbound_spam_log: %w", err) @@ -56,6 +63,24 @@ func (s *Store) InsertInboundSpam(e InboundSpamEntry) error { return nil } +// PruneInboundSpam deletes journal rows older than cutoff and then, if more +// than keep rows are left, the oldest of those. It returns how many went. +func (s *Store) PruneInboundSpam(cutoff time.Time, keep int) (int64, error) { + res, err := s.db.Exec("DELETE FROM inbound_spam_log WHERE created_at < ?", cutoff.UTC().Format(time.RFC3339)) + if err != nil { + return 0, fmt.Errorf("prune inbound_spam_log by age: %w", err) + } + n, _ := res.RowsAffected() + res, err = s.db.Exec( + `DELETE FROM inbound_spam_log WHERE id <= + (SELECT id FROM inbound_spam_log ORDER BY id DESC LIMIT 1 OFFSET ?)`, keep) + if err != nil { + return n, fmt.Errorf("prune inbound_spam_log by count: %w", err) + } + m, _ := res.RowsAffected() + return n + m, 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. diff --git a/internal/store/inboundspam_test.go b/internal/store/inboundspam_test.go index 92e5408..9b269da 100644 --- a/internal/store/inboundspam_test.go +++ b/internal/store/inboundspam_test.go @@ -59,3 +59,102 @@ func TestIsInboundDomain(t *testing.T) { } } } + +func TestInsertInboundSpamKeepsItsOwnTime(t *testing.T) { + st := openTestStore(t) + at := time.Date(2026, 10, 1, 9, 30, 0, 0, time.FixedZone("msk", 3*3600)) + err := st.InsertInboundSpam(InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.7", To: "a@in.example.com", + Decision: SpamDecisionReject, Engine: SpamEnginePostfix, At: at, + }) + if err != nil { + t.Fatalf("InsertInboundSpam: %v", err) + } + var created string + if err := st.db.QueryRow("SELECT created_at FROM inbound_spam_log").Scan(&created); err != nil || created != "2026-10-01T06:30:00Z" { + t.Errorf("created_at = %q, %v", created, err) + } +} + +func TestPruneInboundSpam(t *testing.T) { + st := openTestStore(t) + now := time.Now().UTC() + add := func(to string, age time.Duration) { + t.Helper() + err := st.InsertInboundSpam(InboundSpamEntry{ + InboundDomain: "in.example.com", ClientIP: "203.0.113.7", To: to, + Decision: SpamDecisionReject, Engine: SpamEnginePostfix, At: now.Add(-age), + }) + if err != nil { + t.Fatalf("InsertInboundSpam: %v", err) + } + } + left := func() string { + t.Helper() + rows, err := st.db.Query("SELECT to_addr FROM inbound_spam_log ORDER BY id") + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out string + for rows.Next() { + var to string + if err := rows.Scan(&to); err != nil { + t.Fatal(err) + } + out += to + " " + } + return out + } + + add("old", 100*24*time.Hour) + add("a", 3*time.Hour) + add("b", 2*time.Hour) + add("c", time.Hour) + + // Age first: nothing else goes while the rest fits. + if n, err := st.PruneInboundSpam(now.AddDate(0, 0, -90), 10); err != nil || n != 1 { + t.Fatalf("prune by age = %d, %v", n, err) + } + if got := left(); got != "a b c " { + t.Fatalf("left = %q", got) + } + // Then the cap, oldest first. + if n, err := st.PruneInboundSpam(now.AddDate(0, 0, -90), 2); err != nil || n != 1 { + t.Fatalf("prune by count = %d, %v", n, err) + } + if got := left(); got != "b c " { + t.Fatalf("left = %q", got) + } + if n, err := st.PruneInboundSpam(now.AddDate(0, 0, -90), 2); err != nil || n != 0 { + t.Errorf("a second prune removed %d, %v", n, err) + } +} + +func TestInboundSpamLogRetentionDays(t *testing.T) { + st := openTestStore(t) + if days, err := st.GetInboundSpamLogRetentionDays(30); err != nil || days != 30 { + t.Errorf("unset, env 30: %d, %v", days, err) + } + if days, err := st.GetInboundSpamLogRetentionDays(3); err != nil || days != InboundSpamLogRetentionDaysDefault { + t.Errorf("unset, env out of range: %d, %v", days, err) + } + if err := st.EnsureInboundSpamLogRetentionDays(30); err != nil { + t.Fatalf("Ensure: %v", err) + } + if err := st.EnsureInboundSpamLogRetentionDays(60); err != nil { + t.Fatalf("Ensure again: %v", err) + } + if days, _ := st.GetInboundSpamLogRetentionDays(90); days != 30 { + t.Errorf("after Ensure(30), Ensure(60): %d, want the first seed kept", days) + } + if err := st.SetInboundSpamLogRetentionDays(6); err == nil { + t.Error("6 days was accepted") + } + if err := st.SetInboundSpamLogRetentionDays(365); err != nil { + t.Errorf("365 days: %v", err) + } + if days, _ := st.GetInboundSpamLogRetentionDays(90); days != 365 { + t.Errorf("after Set(365): %d", days) + } +} diff --git a/internal/store/settings.go b/internal/store/settings.go index 22a61b4..ed4c6a9 100644 --- a/internal/store/settings.go +++ b/internal/store/settings.go @@ -77,6 +77,72 @@ func (s *Store) EnsureSendLogRetentionDays(envDefault int) error { return s.SetSendLogRetentionDays(sendLogRetentionFallback(envDefault)) } +const ( + // InboundSpamLogRetentionDaysKey is the settings table key for the + // inbound antispam journal's retention. Same range as the send log's. + InboundSpamLogRetentionDaysKey = "inbound_spam_log_retention_days" + InboundSpamLogRetentionDaysMin = 7 + InboundSpamLogRetentionDaysMax = 365 + // InboundSpamLogRetentionDaysDefault matches + // INBOUND_SPAM_LOG_RETENTION_DAYS when unset. + InboundSpamLogRetentionDaysDefault = 90 +) + +// ErrInboundSpamLogRetentionDaysOutOfRange is returned when the inbound +// journal's retention is outside the allowed panel range. +var ErrInboundSpamLogRetentionDaysOutOfRange = errors.New("inbound journal retention days out of range") + +// ValidateInboundSpamLogRetentionDays checks the panel-allowed window. +func ValidateInboundSpamLogRetentionDays(days int) error { + if days < InboundSpamLogRetentionDaysMin || days > InboundSpamLogRetentionDaysMax { + return fmt.Errorf("%w: must be between %d and %d days", ErrInboundSpamLogRetentionDaysOutOfRange, InboundSpamLogRetentionDaysMin, InboundSpamLogRetentionDaysMax) + } + return nil +} + +func inboundSpamLogRetentionFallback(envDefault int) int { + if ValidateInboundSpamLogRetentionDays(envDefault) == nil { + return envDefault + } + return InboundSpamLogRetentionDaysDefault +} + +// GetInboundSpamLogRetentionDays returns the effective retention window of +// the inbound journal. When the setting is missing or invalid, envDefault is +// used (INBOUND_SPAM_LOG_RETENTION_DAYS). +func (s *Store) GetInboundSpamLogRetentionDays(envDefault int) (int, error) { + raw, err := s.GetSetting(InboundSpamLogRetentionDaysKey) + if err != nil { + return 0, err + } + days, err := strconv.Atoi(raw) + if err != nil || ValidateInboundSpamLogRetentionDays(days) != nil { + return inboundSpamLogRetentionFallback(envDefault), nil + } + return days, nil +} + +// SetInboundSpamLogRetentionDays persists the inbound journal's retention. +func (s *Store) SetInboundSpamLogRetentionDays(days int) error { + if err := ValidateInboundSpamLogRetentionDays(days); err != nil { + return err + } + return s.SetSetting(InboundSpamLogRetentionDaysKey, strconv.Itoa(days)) +} + +// EnsureInboundSpamLogRetentionDays seeds the setting from envDefault when it +// has never been written. +func (s *Store) EnsureInboundSpamLogRetentionDays(envDefault int) error { + raw, err := s.GetSetting(InboundSpamLogRetentionDaysKey) + if err != nil { + return err + } + if raw != "" { + return nil + } + return s.SetInboundSpamLogRetentionDays(inboundSpamLogRetentionFallback(envDefault)) +} + // GetSetting returns a settings value or empty string when missing. func (s *Store) GetSetting(key string) (string, error) { var value string diff --git a/test/e2e/port25_config_check.go b/test/e2e/port25_config_check.go index 5ccef6a..5947f01 100644 --- a/test/e2e/port25_config_check.go +++ b/test/e2e/port25_config_check.go @@ -47,6 +47,9 @@ 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}, + // The log tailer tells this listener's refusals from the + // submission ports' by the name they are logged under. + {param: "syslog_name", contains: "postfix/inbound", wantExact: true}, // No antispam milter configured: the inbound journal milter // alone. The submission chain (strict OpenDKIM + journal-milter) // must not reach inbound mail. diff --git a/test/e2e/port25_features_test.go b/test/e2e/port25_features_test.go index 18f4557..f39ee0d 100644 --- a/test/e2e/port25_features_test.go +++ b/test/e2e/port25_features_test.go @@ -268,24 +268,27 @@ func testInboundRelay(t *testing.T, sc *scenario) { // 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")) + // journalHas waits for a row and returns the journal as it then stands. + journalHas := func(want string) string { + t.Helper() + 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) { + return strings.TrimSpace(rows) + } + if time.Now().After(deadline) { + t.Fatalf("no journal row %q, journal:\n%s\n%s", want, rows, mailLogTail(h, "postfix/")) + } } } + rows := journalHas(fmt.Sprintf("%s|someone@stranger.e2e.test|%s|e2e %s|accept|selfpost", inboundDomain, inboundRecipient, token)) // 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 { + if n := strings.Count(rows, "\n") + 1; n != 1 { t.Fatalf("the journal holds %d rows, want the relayed message alone:\n%s", n, rows) } @@ -295,4 +298,12 @@ func testInboundRelay(t *testing.T, sc *scenario) { t.Fatalf("port 25 accepted the recipient %s (first error: %v)", to, res.firstErr()) } } + + // The refusal of an unknown recipient at the inbound domain is read from + // the mail log into the journal; the refusal to relay for a stranger's + // domain is not the relay's mail and is not. + rows = journalHas(fmt.Sprintf("%s|someone@stranger.e2e.test|nobody@%s||reject|postfix", inboundDomain, inboundDomain)) + if n := strings.Count(rows, "\n") + 1; n != 2 { + t.Fatalf("the journal holds %d rows, want the relayed message and one refusal:\n%s", n, rows) + } }