From c50314fb4cc68245b9ce78b5230f80b7801afd95 Mon Sep 17 00:00:00 2001 From: Mix Date: Sat, 17 Oct 2026 21:44:28 +0300 Subject: [PATCH] inbound journal: one row per recipient, and every row counts to the cap (audit M3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security audit 2026-10-17, finding M3, stage 1. inboundSession.RcptTo appended every RCPT (the outbound session de-duplicates), and the end of the message inserted a row per recipient one statement at a time inside Postfix's callback; the cut to 10 000 rows every 500 counted only the log tailer's rows. Postfix accepts a repeated recipient up to smtpd_recipient_limit (1 000) and sends each to the milter, so one message was a thousand rows, uncut until the six-hourly prune. Now: the inbound milter folds a repeated recipient (case-insensitive, as cleanup's been_here() with BH_FLAG_FOLD does, and as session.RcptTo does); a message's rows go in one transaction (store.InsertInboundSpamMessage, which replaces InsertInboundSpam in milter.InboundStore); the count and the cut moved from the tailer into the store (countInboundWrites, an atomic counter on Store) so the milter's rows, the tailer's and the tailer's start-up re-reads all count. store.InboundSpamLogMaxRows is the one cap. Tests: TestInboundRecipientRepeatedIsOneRow (1 000 RCPT in two cases, one row); TestInboundJournalIsCutWhicheverWriterWrites (a small cap: a message's rows and refusals together cross the mark, the table is cut); the tailer's own counting test went with its counter. go test ./internal/store ./internal/milter ./internal/logtail (Windows-only Follow failures aside). Also gofmt of a test line from the M2 commit. Docs: security.md § Mail path, architecture.md; CHANGELOG; plan. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 ++ docs/architecture.md | 3 +- docs/security-audit-2026-10-17.md | 7 ++- docs/security.md | 8 ++- internal/logtail/inbound.go | 26 ++------- internal/logtail/inbound_test.go | 14 +---- internal/milter/inbound.go | 27 ++++++--- internal/milter/inbound_test.go | 27 ++++++++- internal/store/inboundspam.go | 90 ++++++++++++++++++++++++++++-- internal/store/inboundspam_test.go | 57 +++++++++++++++++++ internal/store/store.go | 5 ++ 11 files changed, 216 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d54c08d..ce09e68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,11 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version milter before Postfix holds it to the login, so a client that never signed in, or another tenant's application, could use up a domain's limit. It is now the signed-in application's domain. +- **One accepted message is one Inbound › Log row per recipient** (audit + M3): a recipient repeated in RCPT TO, up to Postfix's thousand, was a row + each, and the journal milter's rows did not count toward the 10 000-row + cut between the six-hourly prunes. Repeats are one row, a message's rows + are written in one transaction, and both writers count toward the cut. - **A stranger's recipient no longer writes an Inbound › Log row** (audit M2): Postfix quotes a refused recipient in its reply with the quotes removed, so `RCPT TO:<"; from= to= …"@domain>` spelled a second diff --git a/docs/architecture.md b/docs/architecture.md index fd54c1b..b77d6f8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -149,7 +149,8 @@ smtp/inet` removes Debian's stock smtpd). Outbound delivery still uses the start are written only if the same row is not there already. The journal is pruned every six hours: rows older than the `inbound_spam_log_retention_days` setting, then down to - 10 000. + 10 000 - and down to 10 000 every 500 rows either writer adds in between + (the store counts them, `countInboundWrites`). **Filter lists** (`inbound_spam_list`): allow and deny entries for the external filter - sender address, sender domain, client address or network - validated and normalised by `internal/inbound` (`lists.go`) and written as diff --git a/docs/security-audit-2026-10-17.md b/docs/security-audit-2026-10-17.md index d42fb6e..e26b978 100644 --- a/docs/security-audit-2026-10-17.md +++ b/docs/security-audit-2026-10-17.md @@ -727,13 +727,16 @@ the argument. that an unauthenticated session reserves nothing. — **Opus**, 2 h. *Done 2026-10-17:* TestRateLimitIsNotReservedByAnotherSendersMailFrom; e2e `domain_limit_not_reserved_by_a_stranger`. -- [ ] **M3.** De-duplicate recipients in `inboundSession.RcptTo` (case-folded, +- [x] **M3.** De-duplicate recipients in `inboundSession.RcptTo` (case-folded, as the outbound session does); the milter's inserts count toward the prune trigger (a counter in the store beside `InsertInboundSpam`, the prune called from whichever side crosses 500); one transaction per message. Done when a unit test sends one message with the same recipient 1 000 times and sees one row, and 600 milter rows with no refusal trigger the - cut. — **Opus**, 2 h. + cut. — **Opus**, 2 h. *Done 2026-10-17:* the counter and the cut moved + into the store (`countInboundWrites`, both writers), + `InsertInboundSpamMessage`; TestInboundRecipientRepeatedIsOneRow, + TestInboundJournalIsCutWhicheverWriterWrites. - [ ] **W1.** Refuse a custom report address on the hostname; exact hosted address only in `AllowedRecipients`; a tag required for every recipient on the hostname in `checkReportDomain`; the existing test inverted; the diff --git a/docs/security.md b/docs/security.md index 9c7fd6b..771a637 100644 --- a/docs/security.md +++ b/docs/security.md @@ -463,7 +463,13 @@ operator can act on it. `2.14.4` the reply ended at the *first* `; from=<`, which a stranger's recipient could supply - security audit 2026-10-17, M2.) Only a recipient at an inbound domain makes a row. The journal is cut to - 10 000 rows every 500 rows the tailer writes, and by age every six hours. + 10 000 rows every 500 rows written - the tailer's refusals and the journal + milter's accepted recipients together, counted in the store - and by age + every six hours. The journal milter writes a recipient given twice, in any + case, once, and a message's rows in one transaction. (Until `2.14.4` it + wrote every repeated RCPT and its rows were not counted toward the cut: + one message could be a thousand rows, uncut for six hours - security + audit 2026-10-17, M3.) - **Filter lists** reach another container as files: values are whitelisted on the way in (address, domain, canonical IP or network; no `/0`, no IPv4-mapped network), the writer refuses a stored value with whitespace, a diff --git a/internal/logtail/inbound.go b/internal/logtail/inbound.go index 2010959..5fc27f4 100644 --- a/internal/logtail/inbound.go +++ b/internal/logtail/inbound.go @@ -146,15 +146,6 @@ const inboundDetailMax = 255 // cap bounds what is left - messages the client abandoned - and a flood. const inboundQueueIDs = 4096 -// inboundPruneEvery is how many rows the tailer writes before it has the -// journal cut back to its row cap, without waiting for the six-hourly prune. -// Refusals are written at the rate strangers send RCPT TO. Postfix's one -// limit on that, smtpd_client_recipient_rate_limit, is per client address and -// off by default (postconf(5), Postfix 3.10.13) and not set here, so between -// two prunes the journal would otherwise grow as fast as someone cared to -// make it. -const inboundPruneEvery = 500 - // inboundJournal is the per-run state of the inbound half. type inboundJournal struct { st InboundStore @@ -166,8 +157,6 @@ type inboundJournal struct { port25 map[string]uint64 order []queued seq uint64 - // written counts the rows since the journal was last cut to its cap. - written int } type queued struct { @@ -287,13 +276,9 @@ func (j *inboundJournal) handle(line string, now time.Time, catchingUp bool) { log.Printf("log-tailer: inbound journal: record %s -> %s: %v", from, to, err) return } - if j.written++; j.written >= inboundPruneEvery { - j.written = 0 - // The zero time removes nothing by age: this is the cap alone. - if _, err := j.st.PruneInboundSpam(time.Time{}, inboundSpamLogMaxRows); err != nil { - log.Printf("log-tailer: inbound journal: cut to %d rows: %v", inboundSpamLogMaxRows, err) - } - } + // Between the six-hourly prunes the store cuts the journal to its cap + // every so many rows, the journal milter's and these together + // (store.InboundSpamLogMaxRows). } func (j *inboundJournal) remember(queueID string) { @@ -354,9 +339,8 @@ func lineTime(line string, now time.Time) (time.Time, string) { 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 +// The journal keeps at most this many rows whatever their age. +const inboundSpamLogMaxRows = store.InboundSpamLogMaxRows // inboundRetentionLoop prunes the inbound journal at once and then on the // cadence of the send log. diff --git a/internal/logtail/inbound_test.go b/internal/logtail/inbound_test.go index de20f66..0a11312 100644 --- a/internal/logtail/inbound_test.go +++ b/internal/logtail/inbound_test.go @@ -305,7 +305,7 @@ func TestInboundHeloCannotWriteTheRow(t *testing.T) { func TestInboundRecipientCannotWriteTheRow(t *testing.T) { forged := `; from= to= proto=ESMTP helo=; ` for name, line := range map[string]string{ - "recipient unknown": head + `postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 5.1.1 <` + forged + `@in.example.com>: Recipient address rejected: User unknown in relay recipient table; from= to=<"` + forged + `"@in.example.com> proto=ESMTP helo=`, + "recipient unknown": head + `postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 5.1.1 <` + forged + `@in.example.com>: Recipient address rejected: User unknown in relay recipient table; from= to=<"` + forged + `"@in.example.com> proto=ESMTP helo=`, "forged sender also in MAIL FROM": head + `postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 5.1.1 <` + forged + `@in.example.com>: Recipient address rejected: User unknown in relay recipient table; from=<"x; from= to=<"` + forged + `"@in.example.com> proto=ESMTP helo=`, } { j, st := inboundFixture() @@ -362,18 +362,6 @@ func TestInboundStaleArrivalDoesNotDropALiveID(t *testing.T) { } } -// Between the six-hourly prunes the tailer has the journal cut to its cap -// itself, every so many rows: refusals come at the rate strangers send them. -func TestInboundCutsTheJournalAsItWrites(t *testing.T) { - j, st := inboundFixture() - for i := 0; i < inboundPruneEvery*2+10; i++ { - j.handle(unknownRcp, tailNow, false) - } - if len(st.pruned) != 2 || st.pruned[0] != inboundSpamLogMaxRows { - t.Errorf("cuts = %v, want two, to %d rows", st.pruned, inboundSpamLogMaxRows) - } -} - // The queue ids waiting for a verdict are bounded. func TestInboundQueueIDsAreBounded(t *testing.T) { j, _ := inboundFixture() diff --git a/internal/milter/inbound.go b/internal/milter/inbound.go index e5beba0..b59fb7d 100644 --- a/internal/milter/inbound.go +++ b/internal/milter/inbound.go @@ -21,7 +21,7 @@ import ( // 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 + InsertInboundSpamMessage(e store.InboundSpamEntry, rcpts []store.InboundRecipient) error IsInboundDomain(name string) (bool, error) ForgetInboundQueueID(queueID string) error } @@ -83,8 +83,19 @@ func (s *inboundSession) MailFrom(from string, m *milter.Modifier) (milter.Respo // RcptTo keeps the recipient as Postfix logs it, so that the delivery lines // that carry the upstream's answer find the row (recipientAsLogged): a quoted // local part that needs no quotes is written without them. +// +// A recipient given twice - in any case - is one row, as on the outbound side +// (session.RcptTo): Postfix accepts a repeated RCPT up to +// smtpd_recipient_limit and sends the milter each, and one message used to be +// a thousand rows (security audit 2026-10-17, M3). func (s *inboundSession) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) { - s.rcpts = append(s.rcpts, recipientAsLogged(rcpt)) + to := recipientAsLogged(rcpt) + for _, have := range s.rcpts { + if strings.EqualFold(have, to) { + return milter.RespContinue, nil + } + } + s.rcpts = append(s.rcpts, to) return milter.RespContinue, nil } @@ -215,6 +226,7 @@ func (s *inboundSession) record(queueID string) { log.Printf("inbound-journal: detach older rows of %s: %v", queueID, err) } } + var rows []store.InboundRecipient for _, to := range s.rcpts { domain := domainOf(to) ok, err := s.rec.IsInboundDomain(domain) @@ -222,14 +234,13 @@ func (s *inboundSession) record(queueID string) { 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) + if ok { + rows = append(rows, store.InboundRecipient{Domain: domain, To: to}) } } + if err := s.rec.InsertInboundSpamMessage(e, rows); err != nil { + log.Printf("inbound-journal: record %s -> %d recipients: %v", s.from, len(rows), err) + } } // What the journal keeps of a verdict: enough to see why, on one line. diff --git a/internal/milter/inbound_test.go b/internal/milter/inbound_test.go index 84666f7..4b89772 100644 --- a/internal/milter/inbound_test.go +++ b/internal/milter/inbound_test.go @@ -20,11 +20,14 @@ type fakeInbound struct { forgot []string } -func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error { +func (f *fakeInbound) InsertInboundSpamMessage(e store.InboundSpamEntry, rcpts []store.InboundRecipient) error { if f.insertErr != nil { return f.insertErr } - f.rows = append(f.rows, e) + for _, r := range rcpts { + e.InboundDomain, e.To = r.Domain, r.To + f.rows = append(f.rows, e) + } return nil } @@ -400,3 +403,23 @@ func TestInboundRowsMeetTheirDeliveryLines(t *testing.T) { t.Errorf("with no queue id: rows %+v, detached %q", rec.rows, rec.forgot) } } + +// A recipient given again - in any case - is one row: Postfix accepts a +// repeated RCPT up to smtpd_recipient_limit (1 000) and sends each, and one +// message used to be a thousand rows (security audit 2026-10-17, M3). +func TestInboundRecipientRepeatedIsOneRow(t *testing.T) { + rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}} + s := inboundSessionFor(t, rec, false) + rcpts := make([]string, 0, 1000) + for i := 0; i < 1000; i++ { + if i%2 == 0 { + rcpts = append(rcpts, "listed@in.example.com") + } else { + rcpts = append(rcpts, "LISTED@in.example.com") + } + } + message(t, s, "sender@example.net", rcpts, nil) + if len(rec.rows) != 1 || rec.rows[0].To != "listed@in.example.com" { + t.Errorf("rows = %d (%+v), want one, spelt as first given", len(rec.rows), rec.rows) + } +} diff --git a/internal/store/inboundspam.go b/internal/store/inboundspam.go index 5abbff9..72d3adc 100644 --- a/internal/store/inboundspam.go +++ b/internal/store/inboundspam.go @@ -63,9 +63,86 @@ type InboundSpamEntry struct { // inbound domain registered under the row's name now (audit 2026-10-12, L12). const inboundDomainID = `(SELECT d.id FROM inbound_domains d WHERE d.name = ?)` -// 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. +// InboundSpamLogMaxRows is how many rows the journal keeps whatever their +// age: refusals are written at the rate strangers knock on port 25, and +// accepted mail at the rate they send recipients. +const InboundSpamLogMaxRows = 10000 + +// inboundCapRows and inboundCapEvery are the cap and how many rows - from +// either writer - are written before the journal is cut back to it, without +// waiting for the six-hourly prune. Postfix's one limit on the rate of RCPT +// TO, smtpd_client_recipient_rate_limit, is per client address and off by +// default (postconf(5), Postfix 3.10.13) and not set here. Until 2.14.4 only +// the log tailer counted its rows; the journal milter's - one per accepted +// recipient - were cut only every six hours (security audit 2026-10-17, M3). +// Variables so that a test can make them small. +var ( + inboundCapRows = InboundSpamLogMaxRows + inboundCapEvery = int64(500) +) + +// countInboundWrites adds n written rows to the count and, once it reaches +// inboundCapEvery, cuts the journal to its cap. Two writers crossing the mark +// together may both cut; the second finds nothing to do. +func (s *Store) countInboundWrites(n int) error { + if n <= 0 || s.inboundWrites.Add(int64(n)) < inboundCapEvery { + return nil + } + s.inboundWrites.Store(0) + // The zero time removes nothing by age: this is the cap alone. + if _, err := s.PruneInboundSpam(time.Time{}, inboundCapRows); err != nil { + return fmt.Errorf("cut inbound_spam_log to %d rows: %w", inboundCapRows, err) + } + return nil +} + +// InboundRecipient is one recipient of an accepted message at an inbound +// domain: a row of InsertInboundSpamMessage. +type InboundRecipient struct { + Domain string + To string +} + +// InsertInboundSpamMessage records the rows of one accepted message, one per +// recipient, e with InboundDomain and To taken from each, in one transaction: +// the journal milter writes inside Postfix's end-of-message callback on the +// panel's one connection, and a thousand recipients are one commit, not a +// thousand. It returns any error for the caller to log. +func (s *Store) InsertInboundSpamMessage(e InboundSpamEntry, rcpts []InboundRecipient) error { + if len(rcpts) == 0 { + return nil + } + at := e.At + if at.IsZero() { + at = time.Now() + } + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("insert inbound_spam_log: begin: %w", err) + } + defer tx.Rollback() //nolint:errcheck // a no-op after Commit + for _, r := range rcpts { + if _, err := tx.Exec( + `INSERT INTO inbound_spam_log + (inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at, inbound_domain_id, + queue_id, upstream_status) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`, ?, ?)`, + r.Domain, e.ClientIP, e.From, r.To, e.Subject, e.Decision, e.Engine, e.Detail, + at.UTC().Format(time.RFC3339), r.Domain, e.QueueID, e.Upstream, + ); err != nil { + return fmt.Errorf("insert inbound_spam_log: %w", err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("insert inbound_spam_log: commit: %w", err) + } + return s.countInboundWrites(len(rcpts)) +} + +// InsertInboundSpam records one journal row. It is called from the log +// tailer (and was from the journal milter, which now writes a message's rows +// at once), so it returns any error for the caller to log; it may not hold up +// mail over it. func (s *Store) InsertInboundSpam(e InboundSpamEntry) error { at := e.At if at.IsZero() { @@ -82,7 +159,7 @@ func (s *Store) InsertInboundSpam(e InboundSpamEntry) error { if err != nil { return fmt.Errorf("insert inbound_spam_log: %w", err) } - return nil + return s.countInboundWrites(1) } // InsertInboundSpamOnce records a journal row unless the same row is there @@ -109,7 +186,10 @@ func (s *Store) InsertInboundSpamOnce(e InboundSpamEntry) (bool, error) { return false, fmt.Errorf("insert inbound_spam_log once: %w", err) } n, _ := res.RowsAffected() - return n > 0, nil + if n == 0 { + return false, nil + } + return true, s.countInboundWrites(1) } // updateUpstreamQuery puts one delivery attempt's result on the rows of a diff --git a/internal/store/inboundspam_test.go b/internal/store/inboundspam_test.go index 9b76622..9f48732 100644 --- a/internal/store/inboundspam_test.go +++ b/internal/store/inboundspam_test.go @@ -1,6 +1,7 @@ package store import ( + "fmt" "strings" "testing" "time" @@ -495,3 +496,59 @@ func TestInboundUpstreamHeldAndReusedQueueID(t *testing.T) { t.Errorf("ForgetInboundQueueID does not use idx_inbound_spam_log_queue_id:\n%s", plan) } } + +// The journal is cut to its cap every so many rows, whichever writer wrote +// them: the journal milter's rows of accepted mail count as the tailer's +// refusals do (security audit 2026-10-17, M3). One message's rows are one +// call. Here the cap is 5 rows and the cut every 4. +func TestInboundJournalIsCutWhicheverWriterWrites(t *testing.T) { + st := openTestStore(t) + rowsBefore, everyBefore := inboundCapRows, inboundCapEvery + inboundCapRows, inboundCapEvery = 5, 4 + t.Cleanup(func() { inboundCapRows, inboundCapEvery = rowsBefore, everyBefore }) + if _, err := st.AddInboundDomain("in.example.com"); err != nil { + t.Fatal(err) + } + count := func() int { + var n int + if err := st.db.QueryRow("SELECT COUNT(*) FROM inbound_spam_log").Scan(&n); err != nil { + t.Fatal(err) + } + return n + } + accepted := InboundSpamEntry{ClientIP: "203.0.113.7", From: "a@example.net", Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost} + var rcpts []InboundRecipient + for i := 0; i < 3; i++ { + rcpts = append(rcpts, InboundRecipient{Domain: "in.example.com", To: fmt.Sprintf("r%d@in.example.com", i)}) + } + // Three rows of one message, then two refusals: the fourth row crosses + // the mark, and the journal is cut to five. + if err := st.InsertInboundSpamMessage(accepted, rcpts); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err := st.InsertInboundSpam(InboundSpamEntry{InboundDomain: "in.example.com", ClientIP: "203.0.113.9", To: "x@in.example.com", + Decision: SpamDecisionReject, Engine: SpamEnginePostfix}); err != nil { + t.Fatal(err) + } + } + if n := count(); n != 5 { + t.Fatalf("after 5 rows, %d (no cut yet past the cap)", n) + } + // A message of six more crosses the mark again: cut to five. + rcpts = nil + for i := 0; i < 6; i++ { + rcpts = append(rcpts, InboundRecipient{Domain: "in.example.com", To: fmt.Sprintf("m%d@in.example.com", i)}) + } + if err := st.InsertInboundSpamMessage(accepted, rcpts); err != nil { + t.Fatal(err) + } + if n := count(); n != 5 { + t.Errorf("after a message of six over the mark, %d rows, want the cap of 5", n) + } + // The message's rows carry the domain by id, like the tailer's. + var missing int + if err := st.db.QueryRow("SELECT COUNT(*) FROM inbound_spam_log WHERE inbound_domain_id IS NULL").Scan(&missing); err != nil || missing != 0 { + t.Errorf("%d rows without a domain id (%v)", missing, err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 59d0a9c..9a80c3e 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -13,6 +13,7 @@ import ( "io/fs" "os" "sort" + "sync/atomic" _ "modernc.org/sqlite" // pure-Go SQLite driver (no cgo), keeps the static build ) @@ -23,6 +24,10 @@ var migrationsFS embed.FS // Store wraps the database connection pool. type Store struct { db *sql.DB + + // inboundWrites counts journal rows written since the journal was last + // cut to its row cap, by both of its writers (countInboundWrites). + inboundWrites atomic.Int64 } // Open opens (creating if needed) the SQLite database at path, enables WAL and