inbound-antispam-panel: refused inbound mail from mail.log; retention

Step 4/12 of docs/plans/inbound-antispam-panel.md. The log tailer writes an inbound_spam_log row for mail port 25 refused (internal/logtail/inbound.go): a "reject:" line of the port-25 smtpd is Postfix's own policy; a milter-reject / milter-hold / milter-discard line of cleanup is the filter's verdict. Only a recipient at an inbound domain is a row, so "Relay access denied" for a stranger's domain is not. The row is dated by its log line; the reply text is cut to 255 bytes of printable ASCII; a reply starting with 4 is a tempfail. To tell port 25 from 465 / 587 the listener now logs as postfix/inbound (syslog_name). cleanup logs under one name for every listener, so its verdict counts only for a queue id the port-25 smtpd was seen to open (4096 remembered). Lines are matched from the head, after timestamp and host: their tail is the sender's own text. Retention: PruneInboundSpam every six hours - older than the inbound_spam_log_retention_days setting (seeded from INBOUND_SPAM_LOG_RETENTION_DAYS, 90, range 7-365), then down to 10 000 rows. The setting has no field in the panel yet. Known limits, in architecture.md: a log line names one recipient, so a message refused at end-of-message is one row; a line re-read after an unclean stop would be a second row. Checked on the test server, uncommitted tree: go vet ./... ok, go test ./... ok (22 packages), make e2e ok - the unknown recipient at the inbound domain refused on port 25 has its row (reject by postfix), the stranger's domain has none. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-11 22:42:25 +03:00
parent 94e94490af
commit 27fd355e53
18 changed files with 744 additions and 25 deletions
@@ -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 <nobody@in.example.com>: Recipient address rejected: User unknown in relay recipient table; from=<a@example.net> to=<nobody@in.example.com> proto=ESMTP helo=<mx.example.net>"
+)
+
+// 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 <nobody@in.example.com>: 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 <a@nx.example>: Sender address rejected: Domain not found; from=<a@nx.example> to=<B@IN.example.com> proto=ESMTP helo=<x>", 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=<a@example.net> to=<b@in.example.com> proto=ESMTP helo=<mx.example.net>"
+
+	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=<a@example.net> to=<b@in.example.com> proto=ESMTP helo=<mx.example.net>", 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 <c@elsewhere.example>: Relay access denied; from=<a@example.net> to=<c@elsewhere.example> proto=ESMTP helo=<x>",
+		"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=<x>",
+		"a submission port":                    head + "postfix/smtpd[20]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 553 5.7.1 <b@in.example.com>: Sender address rejected: not owned by user app; from=<a@example.net> to=<b@in.example.com> proto=ESMTP helo=<x>",
+		"a delivery line":                      head + "postfix/smtp[26]: 41E862C00D9E: to=<b@in.example.com>, 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=<a@example.net> to=<b@in.example.com> proto=ESMTP helo=<x>",
+		"an address with a bracket in it":       head + "postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from x[203.0.113.9]: 550 no; from=<a> to=<v@in.example.com> proto=X helo=<y>@example.net> to=<b@elsewhere.example> proto=ESMTP helo=<x>",
+	} {
+		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=<a@example.net> to=<b@in.example.com> proto=ESMTP helo=<x>", 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())
+}