inbound-upstream-result: the upstream's answer in the Inbound log
test / test (push) Waiting to run

The owner, 2026-10-16: the Inbound log lacks the upstream's answer - "Не
понятна судьба письма: принято, отклонено, хост не доступен"; the result
in the row only, a page per message later if needed. Mockups 5348330 and
46b90d2, plan 295d72e.

- Migration 0012: inbound_spam_log gains queue_id, upstream_status ('',
  queued, held, sent, deferred, bounced; CHECK), upstream_relay,
  upstream_reply, upstream_at, and a partial index on queue_id. Schema 12.
- The inbound journal milter writes an accepted row with the queue id of
  end-of-message and "queued" ("held" when the filter had it held), the
  recipient as Postfix logs it (recipientAsLogged), after detaching older
  rows of a queue id Postfix uses again; no queue id, no result.
- The tailer: a delivery line no send-log row took goes to the journal -
  status, relay= ("" for none), the reply after the first status=, cut to
  255 printable bytes, the line's time; queued/held/deferred rows only,
  sent and bounced final. qmgr's expiry makes them bounced, keeping the
  last reply. cleanup's "removed (" detaches rows of a message never
  queued. Addresses in delivery lines are read by Postfix's quoting
  (loggedAddress): a quoted RCPT TO cannot end one early.
- The page: an Upstream column - a plain tag for queued and held, else a
  tag opening the host, the reply, when and what follows; "—" for refused
  mail and older rows. Help (word for word with the mockup), guide,
  architecture, security, schema-migrations, CHANGELOG.

Reviewed before the commit (technical and security, 2026-10-16); the plan
lists each finding and what it became.

Tests: store (results, finality, expiry, held, a reused id, the queue-id
index in every plan), milter (queue id, held, quoted recipient, no {i}),
logtail (relay and reply, a forged quoted recipient, Run hands the journal
what the send log did not take, cleanup's removal), view (the column, held,
null sender), e2e (both relayed messages sent with the sink's 250; an
upstream refusing connections deferred with the reason and no host, then
expired to bounced; the refused row has none). go vet + go test (26
packages) and make e2e on the test server for this tree.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mix
2026-10-16 16:35:09 +03:00
co-authored by Claude Opus 5.5
parent 46b90d2a8d
commit d51549385b
26 changed files with 966 additions and 72 deletions
+11
View File
@@ -5,6 +5,17 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version
## [Unreleased] ## [Unreleased]
### Added
- **Inbound › Log shows what the upstream answered** (inbound-upstream-result,
[the plan](docs/plans/inbound-upstream-result.md)): an *Upstream* column
beside the decision - *queued*, *held*, *sent*, *deferred* (a greylisting, an
upstream that cannot be reached), *bounced* (refused, or expired in the
queue) - opening the last reply, the host that gave it and when. Read from
the delivery lines of `mail.log`, one update of the row per attempt; the
lines of port-25 mail are still not stored. Rows logged before have no
result. Migration `0012`, schema 12.
## [2.9.1] - 2026-10-16 ## [2.9.1] - 2026-10-16
Log panes read on a phone. No migration; schema 11. Log panes read on a phone. No migration; schema 11.
+27 -2
View File
@@ -113,7 +113,12 @@ smtp/inet` removes Debian's stock smtpd). Outbound delivery still uses the
verdict headers (`X-Rspamd-Action`, `X-Spamd-Result`) off it. It writes one 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 `inbound_spam_log` row per recipient at an inbound domain - a DMARC report
address is not one - and never refuses anything address is not one - and never refuses anything
([internal/milter/inbound.go](../internal/milter/inbound.go)). The headers ([internal/milter/inbound.go](../internal/milter/inbound.go)). The row
keeps the message's queue id (`{i}` at end-of-message) and starts
`queued` - `held` when the filter had Postfix hold it - and the recipient
as Postfix logs it (`recipientAsLogged`); the log tailer puts the
upstream's answer on it (§ Log tailer, *The upstream's answer*). Older
rows of a queue id Postfix uses again are detached first. The headers
are read only while a filter is attached, and only next to the filter's are read only while a filter is attached, and only next to the filter's
signature: a third header, `X-SelfPost-Filter`, the HMAC-SHA256 of the signature: a third header, `X-SelfPost-Filter`, the HMAC-SHA256 of the
message's queue id under the key in `/data/antispam/filter.key` message's queue id under the key in `/data/antispam/filter.key`
@@ -585,6 +590,26 @@ inbound journal's. The tail of a line is the far end's words (a reply, a HELO
name), and a pattern that could start there let a stranger's HELO name file a name), and a pattern that could start there let a stranger's HELO name file a
delivery (audit 2026-10-12, M12). delivery (audit 2026-10-12, M12).
**The upstream's answer** on the inbound journal (inbound-upstream-result,
`2.10.0`). Mail accepted on port 25 is handed to its upstream by the same
`smtp` client as outbound mail, and its delivery lines look alike; what tells
them apart is the row. A delivery line no send-log row takes (`UpdateStatus`
changed nothing) is offered to the journal: the rows of its queue id and
recipient take the status, the host (`relay=`, `""` for `none` - a line
written without a session, Postfix 3.10.13, `src/smtp/smtp_trouble.c:273,
421`), the reply - the `(%s)` after the first `status=`, made printable and
cut to 255 bytes - and the line's time (`store.UpdateInboundUpstream`). Only
a `queued`, `held` or `deferred` row moves: `sent` and `bounced` are final, so
a line read twice after a restart changes nothing. The addresses are read by
Postfix's quoting (`loggedAddress`), so a quoted local part cannot end one
early. cleanup's `removed (canceled)` - a queue file never queued - detaches
the rows the milter had written for it. qmgr's expiry line, which names no
recipient, makes the rows of its queue id `bounced`, keeping the last reply
after "expired in the queue" (`ExpireInboundUpstream`). It is an update per
attempt, never a row: the journal's caps stand, and port-25 mail's lines are
still not stored (delivery-log-storage). A bounce Postfix sends to a stranger
is mail of its own queue id and matches no row.
**Read offset is persisted** (`logtail_state` table, in the baseline schema): the **Read offset is persisted** (`logtail_state` table, in the baseline schema): the
tailer stores its position plus a fingerprint of the log - the hash of its first tailer stores its position plus a fingerprint of the log - the hash of its first
512 bytes and the file's device and inode, so two logs that begin alike are two 512 bytes and the file's device and inode, so two logs that begin alike are two
@@ -1119,7 +1144,7 @@ full backups also archive the operator's `./certs` PEM files when present.
### Database ### Database
The schema is an embedded baseline, `internal/store/migrations/0001_init.sql`, The schema is an embedded baseline, `internal/store/migrations/0001_init.sql`,
and the migrations after it (`0002` to `0011` today; the next is `0012`), and the migrations after it (`0002` to `0012` today; the next is `0013`),
applied by `store.Open` in file order; `PRAGMA user_version` counts the files applied by `store.Open` in file order; `PRAGMA user_version` counts the files
applied and applied and
`PRAGMA application_id` is `0x53503230` ("SP20"). There is no 1.x migration `PRAGMA application_id` is `0x53503230` ("SP20"). There is no 1.x migration
+6 -1
View File
@@ -1188,7 +1188,12 @@ anyone on the host; they hold addresses, not secrets.
In the panel the lists are **Inbound › Filter lists** (global role), and what In the panel the lists are **Inbound › Filter lists** (global role), and what
port 25 decided about each message is **Inbound › Log**: accepted mail with the port 25 decided about each message is **Inbound › Log**: accepted mail with the
filter's verdict, refused mail with the reply it was given. A row of the log filter's verdict, refused mail with the reply it was given. For accepted mail
its *Upstream* column says what the upstream made of it - *queued* (not tried
yet), *held* (in the queue by the filter's verdict, until released), *sent*,
*deferred* (a greylisting, an upstream that cannot be reached: Postfix tries
again) or *bounced* (refused for good, or expired in the queue; the sender
was sent a bounce) - and opens the last reply and the host that gave it. Rows logged before 2.10.0 have no upstream result. A row of the log
leads to the lists with its sender filled in. Beside each entry of a list is leads to the lists with its sender filled in. Beside each entry of a list is
how often it decided a message, counted from the log - so over the days the how often it decided a message, counted from the log - so over the days the
log keeps - and *never* marks an entry that may have outlived its use. The log keeps - and *never* marks an entry that may have outlived its use. The
+37 -6
View File
@@ -54,13 +54,44 @@ every line, is for later, if the result is not enough.
architecture (journal and tailer), security (what is stored from the architecture (journal and tailer), security (what is stored from the
upstream and at what rate), CHANGELOG. upstream and at what rate), CHANGELOG.
## Reviews (2026-10-16)
A technical and a security review of the tree before the commit; what each
finding became:
- *A quoted RCPT TO* (both): the milter kept `"owner"@x` as given, every line
says `owner@x` - the row never matched. Now `recipientAsLogged`, as the
outbound milter does.
- *A forged line through a quoted local part* (security, medium): `to=<…>`
was read up to its first `>`; a local part in quotes may hold `>`, `,`,
`status=`. Port 25 refuses the one that would reach another row (an `@` in
the local part is a routed address: `src/trivial-rewrite/resolve.c:467`,
`src/smtpd/smtpd_check.c:1724`, `allow_untrusted_routing=no`), and the
pattern now reads addresses by Postfix's quoting all the same
(`loggedAddress`).
- *Held mail* (technical): a message the filter had held showed `queued`,
"not tried yet", for as long as it stayed held. Now `held`, a status of its
own (the CHECK of 0012, not shipped yet).
- *Rows that never close, and queue ids that repeat* (both): before writing
a message's rows the milter detaches older rows of its queue id; cleanup's
`removed (canceled)` detaches rows of a message never queued; an empty
`{i}` writes no result.
- *A bounce of null-sender mail* (technical): "The sender was sent a
bounce" is wrong there - Postfix does not bounce a bounce. The sheet says
so.
- Accepted, said in security.md: a row whose insert waited past the milter's
30 s timeout can miss its first line and stay `queued`. Not of this row:
an old send-log row of a reused queue id and the same recipient takes the
line first (sendlog.go, as before).
## Steps ## Steps
- [ ] Migration 0012, store: write the queue id, update by queue id and - [x] Migration 0012, store: write the queue id, update by queue id and
recipient, read the result back. recipient, read the result back.
- [ ] Milter: the queue id and `queued` on accepted rows. - [x] Milter: the queue id and `queued` on accepted rows.
- [ ] Tailer: delivery and expiry lines of port-25 mail to the journal. - [x] Tailer: delivery and expiry lines of port-25 mail to the journal.
- [ ] View, template, help; tests (store, milter, tailer, view). - [x] View, template, help; tests (store, milter, tailer, view).
- [ ] e2e: a relayed message's row shows `sent` with the sink's `250`; an - [x] e2e: a relayed message's row shows `sent` with the sink's `250`; an
upstream that refuses connections shows `deferred` with the reason. upstream that refuses connections shows `deferred` with the reason.
- [ ] Docs; roll out, read the server's journal; cut `2.10.0`. - [x] Docs. *Done (2026-10-16), with the reviews' findings above.*
- [ ] Roll out, read the server's journal; cut `2.10.0`.
+1 -1
View File
@@ -28,7 +28,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md).
| ID | Topic | Status | Progress | Plan | | ID | Topic | Status | Progress | Plan |
|---|---|---|---|---| |---|---|---|---|---|
| inbound-upstream-result | Inbound › Log shows what the upstream answered: sent, deferred, bounced, with its last reply | agreed (2026-10-16) | 0/6 | [plans/inbound-upstream-result.md](plans/inbound-upstream-result.md) | | inbound-upstream-result | Inbound › Log shows what the upstream answered: sent, deferred, bounced, with its last reply | agreed (2026-10-16); built, unreleased | 6/7 | [plans/inbound-upstream-result.md](plans/inbound-upstream-result.md) |
| health-setup-checks | Server › Health also checks the certificate's name, Postfix's greeting and the trusted proxy (was preflight) | agreed (2026-10-13) | — | — | | health-setup-checks | Server › Health also checks the certificate's name, Postfix's greeting and the trusted proxy (was preflight) | agreed (2026-10-13) | — | — |
| test-email | Send a test message from the panel (was preflight) | agreed (2026-10-13) | — | — | | test-email | Send a test message from the panel (was preflight) | agreed (2026-10-13) | — | — |
| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | shipped `2.1.0` | 12/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) | | inbound-antispam-panel | Inbound antispam journal + allow/deny lists | shipped `2.1.0` | 12/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) |
+5 -3
View File
@@ -17,8 +17,8 @@ When you add or review a migration, update this file in the same change.
| Field | Value | | Field | Value |
|---|---| |---|---|
| Chain head | `user_version = 11` | | Chain head | `user_version = 12` |
| Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql`, `0003_send_log_lines.sql`, `0004_notifications.sql`, `0005_send_log_rejected_index.sql`, `0006_sender_check.sql`, `0007_delivery_log_days.sql`, `0008_tenant_boundary.sql`, `0009_refusal_line.sql`, `0010_send_log_indexes.sql`, `0011_notify_addressees.sql` | | Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql`, `0003_send_log_lines.sql`, `0004_notifications.sql`, `0005_send_log_rejected_index.sql`, `0006_sender_check.sql`, `0007_delivery_log_days.sql`, `0008_tenant_boundary.sql`, `0009_refusal_line.sql`, `0010_send_log_indexes.sql`, `0011_notify_addressees.sql`, `0012_inbound_upstream.sql` |
| File mark | `PRAGMA application_id = 0x53503230` ("SP20") | | File mark | `PRAGMA application_id = 0x53503230` ("SP20") |
| Database file | `/data/selfpost.db` (bind mount) | | Database file | `/data/selfpost.db` (bind mount) |
| Compatibility | 2.x MINOR releases must boot a `2.0.0` data directory. **Nothing is carried over from 1.x.** | | Compatibility | 2.x MINOR releases must boot a `2.0.0` data directory. **Nothing is carried over from 1.x.** |
@@ -69,7 +69,8 @@ delivery-log-storage (`2.2.0`), `0004_notifications.sql` and
`0007_delivery_log_days.sql` and `0008_tenant_boundary.sql` of the 2026-10-12 `0007_delivery_log_days.sql` and `0008_tenant_boundary.sql` of the 2026-10-12
audit (`2.5.0`), `0009_refusal_line.sql` (`2.5.1`) and audit (`2.5.0`), `0009_refusal_line.sql` (`2.5.1`) and
`0010_send_log_indexes.sql` (`2.5.3`) and `0011_notify_addressees.sql` `0010_send_log_indexes.sql` (`2.5.3`) and `0011_notify_addressees.sql`
(`2.5.4`). (`2.5.4`), and `0012_inbound_upstream.sql` of inbound-upstream-result
(`2.10.0`).
--- ---
@@ -88,6 +89,7 @@ audit (`2.5.0`), `0009_refusal_line.sql` (`2.5.1`) and
| 9 | `0009_refusal_line.sql` | 2.5.1 | DDL | `send_log.refusal_line`: Postfix's line about a refusal at MAIL FROM (a level-2 limit, a client address off the list), which has no queue id; the log tailer puts it on the row; `send_log.recovered`, a refusal row the tailer wrote from `mail.log` because the panel stopped before writing it; `sender_check_days.messages` (messages deferred mid-message, counted apart from the connections turned away) and `sender_check_days.estimated` (a day whose downtime is an upper bound) (audit 2026-10-12, stage 2b) | | 9 | `0009_refusal_line.sql` | 2.5.1 | DDL | `send_log.refusal_line`: Postfix's line about a refusal at MAIL FROM (a level-2 limit, a client address off the list), which has no queue id; the log tailer puts it on the row; `send_log.recovered`, a refusal row the tailer wrote from `mail.log` because the panel stopped before writing it; `sender_check_days.messages` (messages deferred mid-message, counted apart from the connections turned away) and `sender_check_days.estimated` (a day whose downtime is an upper bound) (audit 2026-10-12, stage 2b) |
| 10 | `0010_send_log_indexes.sql` | 2.5.3 | DDL | Four indexes of `send_log`, each for a question that walked the table or sorted a domain's log (audit 2026-10-12, M2, stage 3): `idx_send_log_open`, partial, the queued and deferred rows by time (the tailer's sweep every 5 min); `idx_send_log_domain_id` and `idx_send_log_app_login`, a page of one domain's or one application's log in id order; `idx_send_log_refusal_line`, partial, whether a refusal line is already on a row. `TestSendLogQueriesUseTheirIndexes` checks the plans of these and of the other by-time and by-scope statements the store runs | | 10 | `0010_send_log_indexes.sql` | 2.5.3 | DDL | Four indexes of `send_log`, each for a question that walked the table or sorted a domain's log (audit 2026-10-12, M2, stage 3): `idx_send_log_open`, partial, the queued and deferred rows by time (the tailer's sweep every 5 min); `idx_send_log_domain_id` and `idx_send_log_app_login`, a page of one domain's or one application's log in id order; `idx_send_log_refusal_line`, partial, whether a refusal line is already on a row. `TestSendLogQueriesUseTheirIndexes` checks the plans of these and of the other by-time and by-scope statements the store runs |
| 11 | `0011_notify_addressees.sql` | 2.5.4 | DDL | `notify_addressees (key, address, told_at)`: who a subject's notification is owed to and who was handed it, so a copy that failed or was held back by the hourly ceiling goes out at a later check and a subject's end goes to those told (audit 2026-10-12, stage 4); deleted with its `notify_state` row | | 11 | `0011_notify_addressees.sql` | 2.5.4 | DDL | `notify_addressees (key, address, told_at)`: who a subject's notification is owed to and who was handed it, so a copy that failed or was held back by the hourly ceiling goes out at a later check and a subject's end goes to those told (audit 2026-10-12, stage 4); deleted with its `notify_state` row |
| 12 | `0012_inbound_upstream.sql` | 2.10.0 | DDL | `inbound_spam_log` gains `queue_id`, `upstream_status` (`''`, `queued`, `held`, `sent`, `deferred`, `bounced`; CHECK), `upstream_relay`, `upstream_reply` and `upstream_at`, and the partial index `idx_inbound_spam_log_queue_id` on rows that have a queue id: what the upstream answered about mail accepted on port 25, put on the row by the log tailer (inbound-upstream-result). Rows from before have none |
**Kind:** *DDL* — schema only; *data* — `INSERT`/`UPDATE` that must stay correct **Kind:** *DDL* — schema only; *data* — `INSERT`/`UPDATE` that must stay correct
for operators upgrading from older 2.x images. for operators upgrading from older 2.x images.
+22
View File
@@ -368,6 +368,28 @@ operator can act on it.
ask for it at all, an application with a client-address list is refused when ask for it at all, an application with a client-address list is refused when
no address is known, and neither milter replies anything but "continue" to a no address is known, and neither milter replies anything but "continue" to a
message it lets through. `test/e2e` sends two `MAIL FROM` on one connection. message it lets through. `test/e2e` sends two `MAIL FROM` on one connection.
- **The upstream's answer is an update of a row, not a row**
(inbound-upstream-result, `2.10.0`). A delivery line of mail accepted on
port 25 changes the result of that message's journal rows and adds
nothing: a stranger who sends more mail gets more rows from the milter, as
before, and each row at most one update per delivery attempt, at Postfix's
retry pace. The lines themselves of port-25 mail are still not kept. What
is written: the status and the time are Postfix's fields; the host is
Postfix's `relay=`, the reply the upstream's words - which may quote the
stranger's address - or Postfix's own about a host it could not reach, both
printable and cut to 255 bytes, shown in `sp-mono` cells like a refusal's
reply (escaped by html/template). The stranger's own text reaches the line
only as its `to=` and `orig_to=`, which is the key and nothing else: they
are read by Postfix's quoting (`loggedAddress`, `internal/logtail`), so a
quoted RCPT TO holding `>`, `,` or `status=` cannot end the address early
and pass fields of its own for another recipient's row - and one holding an
`@` is a routed address, which port 25 refuses as relaying
(`allow_untrusted_routing=no`). Short queue ids repeat (Postfix's default
`enable_long_queue_ids=no`): before writing a message's rows the milter
detaches older rows of its queue id, so a message that is gone cannot take
a new one's results. Accepted: a row whose insert waited longer than the
milter's 30 s timeout - the database held - can miss its first delivery
line and stay `queued`.
- **The inbound journal is written from what strangers send.** The inbound - **The inbound journal is written from what strangers send.** The inbound
journal milter is the last on port 25, read-only and attached fail-open; it journal milter is the last on port 25, read-only and attached fail-open; it
keeps of the filter's verdict only what its patterns recognise. **A verdict keeps of the filter's verdict only what its patterns recognise. **A verdict
+43 -2
View File
@@ -11,13 +11,47 @@ import (
) )
// InboundStore is what the tailer needs of the inbound antispam journal: // InboundStore is what the tailer needs of the inbound antispam journal:
// writing the rows for refused mail, knowing which domains are inbound, and // writing the rows for refused mail, knowing which domains are inbound,
// pruning. *store.Store satisfies it. // putting the upstream's answers on the rows of accepted mail, and pruning.
// *store.Store satisfies it.
type InboundStore interface { type InboundStore interface {
InsertInboundSpam(e store.InboundSpamEntry) error InsertInboundSpam(e store.InboundSpamEntry) error
InsertInboundSpamOnce(e store.InboundSpamEntry) (bool, error) InsertInboundSpamOnce(e store.InboundSpamEntry) (bool, error)
IsInboundDomain(name string) (bool, error) IsInboundDomain(name string) (bool, error)
PruneInboundSpam(cutoff time.Time, keep int) (int64, error) PruneInboundSpam(cutoff time.Time, keep int) (int64, error)
UpdateInboundUpstream(queueID, recipient, status, relay, reply string, at time.Time) (int64, error)
ExpireInboundUpstream(queueID string, at time.Time) (int64, error)
ForgetInboundQueueID(queueID string) error
}
// upstream puts a delivery line's result on the journal rows of a message
// accepted on port 25 (inbound-upstream-result). The tailer offers it every
// delivery line no send-log row took: the rows of that queue id and recipient
// take it, and a line of any other mail - the panel's own, a bounce Postfix
// sends back to a stranger - matches none. relay "none" is a line written
// without a session (Postfix 3.10.13, src/smtp/smtp_trouble.c:273, 421: the
// host is then only in the reply), kept as "". The reply is the upstream's
// text, or Postfix's own about a host it could not reach, cut and made
// printable like a refusal's reason. One update per attempt, never a row of
// its own: the journal's caps stand.
func (j *inboundJournal) upstream(line string, now time.Time, queueID, recipient, status string) {
at, rest := lineTime(line, now)
relay, reply := deliveryReply(rest)
if relay == "none" {
relay = ""
}
if _, err := j.st.UpdateInboundUpstream(queueID, recipient, status, printable(relay, inboundDetailMax),
printable(reply, store.UpstreamReplyMax), at); err != nil {
log.Printf("log-tailer: inbound journal: upstream %s/%s -> %s: %v", queueID, recipient, status, err)
}
}
// expired makes the rows of a port-25 message Postfix gave up on bounced.
func (j *inboundJournal) expired(line string, now time.Time, queueID string) {
at, _ := lineTime(line, now)
if _, err := j.st.ExpireInboundUpstream(queueID, at); err != nil {
log.Printf("log-tailer: inbound journal: expire %s: %v", queueID, err)
}
} }
// Inbound turns on the inbound half of the tailer: journal rows for mail that // Inbound turns on the inbound half of the tailer: journal rows for mail that
@@ -153,6 +187,13 @@ func (j *inboundJournal) handle(line string, now time.Time, catchingUp bool) {
} }
if m := inboundForgetRe.FindStringSubmatch(rest); m != nil { if m := inboundForgetRe.FindStringSubmatch(rest); m != nil {
delete(j.port25, m[1]+m[2]+m[3]) delete(j.port25, m[1]+m[2]+m[3])
// A queue file cleanup removed was never queued: rows the milter wrote
// for it will get no delivery line (store.ForgetInboundQueueID).
if m[3] != "" {
if err := j.st.ForgetInboundQueueID(m[3]); err != nil {
log.Printf("log-tailer: inbound journal: forget %s: %v", m[3], err)
}
}
return return
} }
m := inboundRejectRe.FindStringSubmatch(rest) m := inboundRejectRe.FindStringSubmatch(rest)
+34
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"strings" "strings"
"sync"
"testing" "testing"
"time" "time"
@@ -15,6 +16,39 @@ type fakeInboundStore struct {
rows []store.InboundSpamEntry rows []store.InboundSpamEntry
lookupErr error lookupErr error
pruned []int pruned []int
// upstream records the results put on rows, as
// "queue-id|recipient|status|relay|reply|time" or "queue-id|expired|time";
// Run writes it from its own goroutine.
mu sync.Mutex
upstream []string
}
func (f *fakeInboundStore) UpdateInboundUpstream(queueID, recipient, status, relay, reply string, at time.Time) (int64, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.upstream = append(f.upstream, strings.Join([]string{queueID, recipient, status, relay, reply, at.Format(time.RFC3339)}, "|"))
return 1, nil
}
func (f *fakeInboundStore) ExpireInboundUpstream(queueID string, at time.Time) (int64, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.upstream = append(f.upstream, queueID+"|expired|"+at.Format(time.RFC3339))
return 1, nil
}
func (f *fakeInboundStore) ForgetInboundQueueID(queueID string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.upstream = append(f.upstream, queueID+"|forgot")
return nil
}
func (f *fakeInboundStore) upstreamCalls() []string {
f.mu.Lock()
defer f.mu.Unlock()
return append([]string(nil), f.upstream...)
} }
func (f *fakeInboundStore) InsertInboundSpam(e store.InboundSpamEntry) error { func (f *fakeInboundStore) InsertInboundSpam(e store.InboundSpamEntry) error {
+47 -6
View File
@@ -85,8 +85,8 @@ const (
// calls it on every prune cycle so a panel change takes effect without restart. // calls it on every prune cycle so a panel change takes effect without restart.
type RetentionDays func() int type RetentionDays func() int
// deliveryRe matches a Postfix delivery line and captures queue-id, recipient // deliveryRe matches a Postfix delivery line and captures queue-id, recipient,
// and status, e.g. // orig_to, relay and status, e.g.
// //
// postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=…, dsn=2.0.0, status=sent (250 OK) // postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=…, dsn=2.0.0, status=sent (250 OK)
// //
@@ -116,7 +116,19 @@ type RetentionDays func() int
// controls. A bounce whose reply quoted "status=sent" would then be filed as a // controls. A bounce whose reply quoted "status=sent" would then be filed as a
// success. The real field is always the first one after to=<…>: in // success. The real field is always the first one after to=<…>: in
// log_adhoc() the reply is the "(%s)" after status=. // log_adhoc() the reply is the "(%s)" after status=.
var deliveryRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-Za-z]+): to=<(.*?)>, (?:orig_to=<(.*?)>, )?relay=.*?\bstatus=(\w+)`) var deliveryRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-Za-z]+): to=<(` + loggedAddress + `)>, (?:orig_to=<(` + loggedAddress + `)>, )?relay=([^,\s]*).*?\bstatus=(\w+)`)
// loggedAddress is an address as Postfix logs it, external form: a local part
// in quotes when it is not a plain dot-string - and then it may hold "<", ">",
// "," and "=", with only '"' and '\' escaped - or one with none of '"' and
// '>' (Postfix 3.10.13, src/global/quote_822_local.c,
// src/global/info_log_addr_form.c). Read by that grammar, a stranger's quoted
// RCPT TO cannot end the address early and pass "orig_to=<…>, relay=…,
// status=…" of his own as the line's fields (security review of
// inbound-upstream-result, 2026-10-16; a local part with "@" is a routed
// address, which port 25 refuses as relaying anyway: src/trivial-rewrite/
// resolve.c:467, src/smtpd/smtpd_check.c:1724, allow_untrusted_routing=no).
const loggedAddress = `"(?:[^"\\]|\\.)*"@[^>]*|[^>"]*`
// parseDelivery extracts (queue-id, recipient, status) from a mail.log line. // parseDelivery extracts (queue-id, recipient, status) from a mail.log line.
// ok is false for lines that are not recognised delivery results. The // ok is false for lines that are not recognised delivery results. The
@@ -133,7 +145,7 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
if m[3] != "" { if m[3] != "" {
recipient = m[3] recipient = m[3]
} }
switch m[4] { switch m[5] {
case "sent": case "sent":
status = store.StatusSent status = store.StatusSent
case "deferred": case "deferred":
@@ -146,6 +158,25 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
return m[1], recipient, status, true return m[1], recipient, status, true
} }
// deliveryReply returns the host and the reply of a delivery line (the line
// after its time stamp): relay= - which log_adhoc() writes right after to=
// and orig_to=, followed by a comma - and the "(%s)" after the first status=,
// to the end of the line (Postfix 3.10.13, src/global/log_adhoc.c). Taking the
// first status= keeps the reply whole when the far end's text holds one of
// its own. A line that is not a delivery line gives "", "".
func deliveryReply(rest string) (relay, reply string) {
loc := deliveryRe.FindStringSubmatchIndex(rest)
if loc == nil {
return "", ""
}
relay = rest[loc[8]:loc[9]]
tail := strings.TrimSpace(rest[loc[1]:])
if strings.HasPrefix(tail, "(") && strings.HasSuffix(tail, ")") {
reply = tail[1 : len(tail)-1]
}
return relay, reply
}
// expiryRe matches qmgr's line for a message it gave up on: // expiryRe matches qmgr's line for a message it gave up on:
// //
// postfix/qmgr[93]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender // postfix/qmgr[93]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender
@@ -225,18 +256,28 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
} }
keeper.add(line, now, catchingUp) keeper.add(line, now, catchingUp)
refusals.handle(line, now, Place{FP: tr.line.FP, End: tr.lineStart}) refusals.handle(line, now, Place{FP: tr.line.FP, End: tr.lineStart})
// A delivery or expiry line no send-log row takes may be of mail
// accepted on port 25: the inbound journal's rows take its result.
queueID, recipient, status, ok := parseDelivery(line) queueID, recipient, status, ok := parseDelivery(line)
if !ok { if !ok {
if queueID, ok := parseExpiry(line); ok { if queueID, ok := parseExpiry(line); ok {
if _, err := st.ExpireMessage(queueID); err != nil { n, err := st.ExpireMessage(queueID)
if err != nil {
log.Printf("log-tailer: expire %s: %v", queueID, err) log.Printf("log-tailer: expire %s: %v", queueID, err)
} }
if n == 0 && journal != nil {
journal.expired(line, now, queueID)
}
} }
return return
} }
if _, err := st.UpdateStatus(queueID, recipient, status); err != nil { n, err := st.UpdateStatus(queueID, recipient, status)
if err != nil {
log.Printf("log-tailer: update %s/%s -> %s: %v", queueID, recipient, status, err) log.Printf("log-tailer: update %s/%s -> %s: %v", queueID, recipient, status, err)
} }
if n == 0 && journal != nil {
journal.upstream(line, now, queueID, recipient, status)
}
}) })
} }
+6
View File
@@ -339,6 +339,9 @@ func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, e
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
c.calls = append(c.calls, queueID+"|"+recipient+"|"+status) c.calls = append(c.calls, queueID+"|"+recipient+"|"+status)
if strings.HasPrefix(queueID, "P25") {
return 0, nil // mail accepted on port 25: no send-log row
}
return 1, nil return 1, nil
} }
@@ -347,6 +350,9 @@ func (c *captureStore) ExpireMessage(queueID string) (int64, error) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
c.calls = append(c.calls, queueID+"|*|expired") c.calls = append(c.calls, queueID+"|*|expired")
if strings.HasPrefix(queueID, "P25") {
return 0, nil
}
return 1, nil return 1, nil
} }
+121
View File
@@ -0,0 +1,121 @@
package logtail
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
// The host and the reply of a delivery line, as log_adhoc() writes them: the
// relay= right after to=, the "(...)" after the first status= to the end of
// the line - also when the far end's reply holds a status= of its own.
func TestDeliveryReply(t *testing.T) {
for _, c := range []struct{ line, relay, reply string }{
{"selfpost postfix/smtp[315]: 82922153D147: to=<carrel@dev.mixdep.ru>, relay=mc-ent.mixfed.ru[188.242.125.61]:25, delay=498, delays=491/0.03/2.5/5, dsn=2.0.0, status=sent (250 2.0.0 Ok: queued as CE05D10261D)",
"mc-ent.mixfed.ru[188.242.125.61]:25", "250 2.0.0 Ok: queued as CE05D10261D"},
{"selfpost postfix/smtp[300]: D6E4F153D147: to=<a@in.example.com>, relay=none, delay=30, delays=0.1/0/30/0, dsn=4.4.1, status=deferred (connect to mx.up.example[198.51.100.5]:25: Connection refused)",
"none", "connect to mx.up.example[198.51.100.5]:25: Connection refused"},
{"selfpost postfix/smtp[312]: 1CACF153D147: to=<a@in.example.com>, orig_to=<A@in.example.com>, relay=mx.up.example[198.51.100.5]:25, conn_use=2, delay=1.9, dsn=5.7.1, status=bounced (host mx.up.example[198.51.100.5] said: 550 5.7.1 not status=sent (in reply to end of DATA command))",
"mx.up.example[198.51.100.5]:25", "host mx.up.example[198.51.100.5] said: 550 5.7.1 not status=sent (in reply to end of DATA command)"},
{"selfpost postfix/qmgr[286]: 1484E1539884: removed", "", ""},
} {
relay, reply := deliveryReply(c.line)
if relay != c.relay || reply != c.reply {
t.Errorf("deliveryReply(%q) =\n %q, %q\nwant\n %q, %q", c.line, relay, reply, c.relay, c.reply)
}
}
}
// The journal takes a delivery line's result with the host ("" for relay=none),
// the reply made printable and cut, and the line's own time.
func TestInboundJournalTakesTheUpstreamsAnswer(t *testing.T) {
j, st := inboundFixture()
now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC)
j.upstream("Oct 16 10:16:42 selfpost postfix/smtp[300]: P25A: to=<a@in.example.com>, relay=none, delay=30, dsn=4.4.1, status=deferred (connect to mx.up.example[198.51.100.5]:25: Connection refused\x01"+strings.Repeat("x", 300)+")",
now, "P25A", "a@in.example.com", "deferred")
calls := st.upstreamCalls()
if len(calls) != 1 {
t.Fatalf("calls = %q", calls)
}
f := strings.Split(calls[0], "|")
if f[0] != "P25A" || f[1] != "a@in.example.com" || f[2] != "deferred" || f[3] != "" {
t.Errorf("call = %q", calls[0])
}
if !strings.HasPrefix(f[4], "connect to mx.up.example[198.51.100.5]:25: Connection refused") || len(f[4]) > 255 || strings.ContainsRune(f[4], 1) {
t.Errorf("reply = %q (%d bytes)", f[4], len(f[4]))
}
if at, _ := time.Parse(time.RFC3339, f[5]); !at.Equal(time.Date(2026, 10, 16, 10, 16, 42, 0, time.Local).UTC()) {
t.Errorf("at = %s, want the line's time", f[5])
}
}
// Run hands the journal what no send-log row took: a delivery line and an
// expiry line of a port-25 message (captureStore takes no queue id with
// "P25"), and nothing of the mail the send log knows.
func TestRunPutsTheUpstreamsAnswerOnTheJournal(t *testing.T) {
old := pollInterval
pollInterval = 10 * time.Millisecond
t.Cleanup(func() { pollInterval = old })
path := filepath.Join(t.TempDir(), "mail.log")
if err := os.WriteFile(path, nil, 0o644); err != nil {
t.Fatal(err)
}
cs := &captureStore{}
_, in := inboundFixture()
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() {
done <- Run(ctx, path, cs, func() int { return 90 }, &Inbound{Store: in, Retention: func() int { return 90 }}, nil)
}()
time.Sleep(50 * time.Millisecond)
appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
appendLine(t, path, "host postfix/smtp[2]: P25A: to=<a@in.example.com>, relay=mx.up.example[198.51.100.5]:25, dsn=2.0.0, status=sent (250 2.0.0 Ok)")
appendLine(t, path, "host postfix/qmgr[3]: P25B: from=<s@example.net>, status=expired, returned to sender")
waitFor(t, func() bool { return len(in.upstreamCalls()) == 2 })
calls := in.upstreamCalls()
if !strings.HasPrefix(calls[0], "P25A|a@in.example.com|sent|mx.up.example[198.51.100.5]:25|250 2.0.0 Ok|") ||
!strings.HasPrefix(calls[1], "P25B|expired|") {
t.Errorf("journal calls = %q", calls)
}
cancel()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("Run did not return after cancel")
}
for _, c := range in.upstreamCalls() {
if strings.HasPrefix(c, "Q1|") {
t.Errorf("the journal was offered outbound mail: %q", c)
}
}
}
// A stranger's quoted RCPT TO may hold ">", "," and "=": read by Postfix's
// quoting, the address is his whole local part, and the line's own fields -
// the relay and status after it - are the ones taken, not those he wrote.
func TestDeliveryLineReadsAQuotedRecipientWhole(t *testing.T) {
line := `Oct 16 10:16:42 selfpost postfix/smtp[300]: P25F: to=<"a>, orig_to=<b@in.example.com>, relay=F, dsn=5.0.0, status=bounced (FORGED"@in.example.com>, relay=mx.up.example[198.51.100.5]:25, delay=1, dsn=2.0.0, status=sent (250 2.0.0 Ok)`
q, r, s, ok := parseDelivery(line)
if !ok || q != "P25F" || s != "sent" || r != `"a>, orig_to=<b@in.example.com>, relay=F, dsn=5.0.0, status=bounced (FORGED"@in.example.com` {
t.Errorf("parseDelivery = %q %q %q %v", q, r, s, ok)
}
_, rest := SplitTimestamp(line)
if relay, reply := deliveryReply(rest); relay != "mx.up.example[198.51.100.5]:25" || reply != "250 2.0.0 Ok" {
t.Errorf("deliveryReply = %q, %q", relay, reply)
}
}
// cleanup's "removed (" is a queue file that was never queued: the rows the
// milter wrote for it are detached from its queue id.
func TestInboundJournalForgetsAMessageCleanupRemoved(t *testing.T) {
j, st := inboundFixture()
j.handle("Oct 16 10:16:42 selfpost postfix/cleanup[52]: 4A1B2C3D: removed (canceled)", tailNow, false)
j.handle("Oct 16 10:16:42 selfpost postfix/qmgr[52]: 5A1B2C3D: removed", tailNow, false)
if calls := st.upstreamCalls(); len(calls) != 1 || calls[0] != "4A1B2C3D|forgot" {
t.Errorf("calls = %q", calls)
}
}
+30 -8
View File
@@ -23,6 +23,7 @@ import (
type InboundStore interface { type InboundStore interface {
InsertInboundSpam(e store.InboundSpamEntry) error InsertInboundSpam(e store.InboundSpamEntry) error
IsInboundDomain(name string) (bool, error) IsInboundDomain(name string) (bool, error)
ForgetInboundQueueID(queueID string) error
} }
// inboundSession is the inbound journal milter: the last milter on port 25, // inboundSession is the inbound journal milter: the last milter on port 25,
@@ -79,8 +80,11 @@ func (s *inboundSession) MailFrom(from string, m *milter.Modifier) (milter.Respo
return milter.RespContinue, nil return milter.RespContinue, nil
} }
// 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.
func (s *inboundSession) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) { func (s *inboundSession) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) {
s.rcpts = append(s.rcpts, cleanAddress(rcpt)) s.rcpts = append(s.rcpts, recipientAsLogged(rcpt))
return milter.RespContinue, nil return milter.RespContinue, nil
} }
@@ -156,11 +160,13 @@ func (s *inboundSession) signed(queueID string) bool {
// rest as session.Body in milter.go. // rest as session.Body in milter.go.
func (s *inboundSession) Body(m *milter.Modifier) (milter.Response, error) { func (s *inboundSession) Body(m *milter.Modifier) (milter.Response, error) {
// The queue id is known only now, and with it whether the verdict headers // The queue id is known only now, and with it whether the verdict headers
// read on the way here are the filter's. // read on the way here are the filter's. The rows keep it: the delivery
if !s.signed(macro(m, "i")) { // lines that put the upstream's answer on them name it (logtail).
queueID := macro(m, "i")
if !s.signed(queueID) {
s.seen = false s.seen = false
} }
s.record() s.record(queueID)
s.reset() s.reset()
return milter.RespContinue, nil return milter.RespContinue, nil
} }
@@ -175,10 +181,17 @@ func (s *inboundSession) reset() {
s.action, s.result, s.seen, s.sig = "", "", false, "" s.action, s.result, s.seen, s.sig = "", "", false, ""
} }
// record writes one row per recipient at an inbound domain. Other recipients // record writes one row per recipient at an inbound domain, with the queue id
// of port 25 (a DMARC report address at the server's own name) are not the // of the message and "queued" - "held" when the filter's verdict had Postfix
// relay's mail and get no row. Failures are logged, never propagated. // hold it (src/cleanup/cleanup_milter.c: a milter's HOLD puts the message in
func (s *inboundSession) record() { // the hold queue, which qmgr does not schedule). The message is about to be
// queued for its upstream, and the rows are written before Postfix goes on,
// so before any delivery line of it; older rows of a queue id Postfix uses
// again are detached first. With no queue id (none came) the rows have no
// result: nothing could report one. 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(queueID string) {
e := store.InboundSpamEntry{ e := store.InboundSpamEntry{
ClientIP: s.clientIP, ClientIP: s.clientIP,
From: s.from, From: s.from,
@@ -193,6 +206,15 @@ func (s *inboundSession) record() {
e.Decision = store.SpamDecisionQuarantine e.Decision = store.SpamDecisionQuarantine
} }
} }
if queueID != "" {
e.QueueID, e.Upstream = queueID, store.UpstreamQueued
if e.Decision == store.SpamDecisionQuarantine {
e.Upstream = store.UpstreamHeld
}
if err := s.rec.ForgetInboundQueueID(queueID); err != nil {
log.Printf("inbound-journal: detach older rows of %s: %v", queueID, err)
}
}
for _, to := range s.rcpts { for _, to := range s.rcpts {
domain := domainOf(to) domain := domainOf(to)
ok, err := s.rec.IsInboundDomain(domain) ok, err := s.rec.IsInboundDomain(domain)
+46
View File
@@ -17,6 +17,7 @@ type fakeInbound struct {
rows []store.InboundSpamEntry rows []store.InboundSpamEntry
insertErr error insertErr error
lookupErr error lookupErr error
forgot []string
} }
func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error { func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error {
@@ -27,6 +28,11 @@ func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error {
return nil return nil
} }
func (f *fakeInbound) ForgetInboundQueueID(queueID string) error {
f.forgot = append(f.forgot, queueID)
return nil
}
func (f *fakeInbound) IsInboundDomain(name string) (bool, error) { func (f *fakeInbound) IsInboundDomain(name string) (bool, error) {
if f.lookupErr != nil { if f.lookupErr != nil {
return false, f.lookupErr return false, f.lookupErr
@@ -100,6 +106,8 @@ func TestInboundRecordsAcceptedMessage(t *testing.T) {
InboundDomain: "in.example.com", ClientIP: "203.0.113.7", InboundDomain: "in.example.com", ClientIP: "203.0.113.7",
From: "sender@example.net", To: "a@in.example.com", Subject: "Schön", From: "sender@example.net", To: "a@in.example.com", Subject: "Schön",
Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost, Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost,
// The queue id of end-of-message, for the delivery lines to come.
QueueID: "ABC123", Upstream: store.UpstreamQueued,
} }
if rec.rows[0] != want { if rec.rows[0] != want {
t.Errorf("row = %+v\nwant %+v", rec.rows[0], want) t.Errorf("row = %+v\nwant %+v", rec.rows[0], want)
@@ -204,6 +212,7 @@ func TestInboundResetsBetweenMessages(t *testing.T) {
InboundDomain: "in.example.com", ClientIP: "203.0.113.7", InboundDomain: "in.example.com", ClientIP: "203.0.113.7",
From: "two@example.net", To: "b@in.example.com", From: "two@example.net", To: "b@in.example.com",
Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost, Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost,
QueueID: "ABC123", Upstream: store.UpstreamQueued,
} }
if rec.rows[1] != want { if rec.rows[1] != want {
t.Errorf("second row = %+v\nwant %+v", rec.rows[1], want) t.Errorf("second row = %+v\nwant %+v", rec.rows[1], want)
@@ -354,3 +363,40 @@ func TestFilterSignatureIsWhatTheFilterWrites(t *testing.T) {
t.Errorf("FilterSignature = %s", got) t.Errorf("FilterSignature = %s", got)
} }
} }
// The row is the delivery lines' to find: a recipient is kept as Postfix logs
// it, a held message starts held, older rows of a reused queue id are
// detached first - once per message - and a message with no queue id has no
// result, since nothing could ever report one.
func TestInboundRowsMeetTheirDeliveryLines(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{`<"owner"@in.example.com>`, "b@in.example.com"}, nil)
if len(rec.rows) != 2 || rec.rows[0].To != "owner@in.example.com" {
t.Fatalf("rows = %+v, want the quoted recipient as logged", rec.rows)
}
if len(rec.forgot) != 1 || rec.forgot[0] != "ABC123" {
t.Errorf("detached %q, want ABC123 once", rec.forgot)
}
rec.rows = nil
message(t, s, "sender@example.net", []string{"a@in.example.com"},
[][2]string{{"X-Rspamd-Action", "quarantine"}, testSig})
if len(rec.rows) != 1 || rec.rows[0].Upstream != store.UpstreamHeld {
t.Errorf("a held message's row = %+v, want held", rec.rows)
}
rec.rows, rec.forgot = nil, nil
if _, err := s.MailFrom("sender@example.net", mods(nil)); err != nil {
t.Fatal(err)
}
if _, err := s.RcptTo("a@in.example.com", mods(nil)); err != nil {
t.Fatal(err)
}
if _, err := s.Body(mods(nil)); err != nil {
t.Fatal(err)
}
if len(rec.rows) != 1 || rec.rows[0].QueueID != "" || rec.rows[0].Upstream != "" || len(rec.forgot) != 0 {
t.Errorf("with no queue id: rows %+v, detached %q", rec.rows, rec.forgot)
}
}
+103 -6
View File
@@ -25,6 +25,18 @@ const (
SpamEngineSelfPost = "selfpost" // nobody: relayed with no filter attached SpamEngineSelfPost = "selfpost" // nobody: relayed with no filter attached
) )
// What the upstream made of a message accepted on port 25, on its journal
// rows (migration 0012). Kept in sync with the CHECK constraint of
// inbound_spam_log; "" is a row with no result (refused mail, or written
// before 2.10.0). sent and bounced are final.
const (
UpstreamQueued = "queued" // accepted, not tried yet
UpstreamHeld = "held" // accepted into the hold queue by the filter's verdict, not tried until released
UpstreamSent = "sent" // the upstream took it
UpstreamDeferred = "deferred" // put off or not reached; Postfix tries again
UpstreamBounced = "bounced" // refused for good, or expired in the queue
)
// InboundSpamEntry is one row of the inbound antispam journal: what happened // InboundSpamEntry is one row of the inbound antispam journal: what happened
// to a message offered on port 25 for one recipient. Metadata only - the // 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. // message itself is relayed, not stored. Subject and Detail may be empty.
@@ -37,6 +49,11 @@ type InboundSpamEntry struct {
Decision string Decision string
Engine string Engine string
Detail string Detail string
// QueueID and Upstream are set on mail that was accepted: the queue id
// the delivery lines will name, and UpstreamQueued (UpstreamHeld for a
// message the filter had held).
QueueID string
Upstream string
// At is when it happened; zero means now. The log tailer dates a refusal // 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. // by its mail.log line, which it may read long after.
At time.Time At time.Time
@@ -56,10 +73,11 @@ func (s *Store) InsertInboundSpam(e InboundSpamEntry) error {
} }
_, err := s.db.Exec( _, err := s.db.Exec(
`INSERT INTO inbound_spam_log `INSERT INTO inbound_spam_log
(inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at, inbound_domain_id) (inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at, inbound_domain_id,
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`)`, queue_id, upstream_status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`, ?, ?)`,
e.InboundDomain, e.ClientIP, e.From, e.To, e.Subject, e.Decision, e.Engine, e.Detail, e.InboundDomain, e.ClientIP, e.From, e.To, e.Subject, e.Decision, e.Engine, e.Detail,
at.UTC().Format(time.RFC3339), e.InboundDomain, at.UTC().Format(time.RFC3339), e.InboundDomain, e.QueueID, e.Upstream,
) )
if err != nil { if err != nil {
return fmt.Errorf("insert inbound_spam_log: %w", err) return fmt.Errorf("insert inbound_spam_log: %w", err)
@@ -94,6 +112,74 @@ func (s *Store) InsertInboundSpamOnce(e InboundSpamEntry) (bool, error) {
return n > 0, nil return n > 0, nil
} }
// updateUpstreamQuery puts one delivery attempt's result on the rows of a
// queue id and recipient whose result is not final; expireUpstreamQuery makes
// those of a queue id bounced, saying it expired after the last reply. Both
// read the queue id through idx_inbound_spam_log_queue_id
// (TestInboundUpstreamUsesItsIndex).
const (
updateUpstreamQuery = `UPDATE inbound_spam_log
SET upstream_status = ?, upstream_relay = ?, upstream_reply = ?, upstream_at = ?
WHERE queue_id = ? AND queue_id <> '' AND to_addr = ? COLLATE NOCASE
AND upstream_status IN ('queued', 'held', 'deferred')`
expireUpstreamQuery = `UPDATE inbound_spam_log
SET upstream_status = 'bounced', upstream_at = ?,
upstream_reply = substr(CASE WHEN upstream_reply = '' THEN 'expired in the queue'
ELSE 'expired in the queue; last reply: ' || upstream_reply END, 1, ?)
WHERE queue_id = ? AND queue_id <> '' AND upstream_status IN ('queued', 'held', 'deferred')`
forgetQueueIDQuery = `UPDATE inbound_spam_log
SET queue_id = '',
upstream_status = CASE WHEN upstream_status IN ('queued', 'held', 'deferred') THEN '' ELSE upstream_status END
WHERE queue_id = ? AND queue_id <> ''`
)
// ForgetInboundQueueID detaches the journal rows of a queue id from it: those
// still open lose their result too, as nothing will ever report it. The
// milter calls it before it writes a new message's rows - Postfix reuses
// queue ids (enable_long_queue_ids=no, the default: microseconds and the
// queue file's inode, src/global/mail_queue.h), and only one live queue file
// has an id, so older rows of it are of a message that is gone and must not
// take the new one's results. The tailer calls it when cleanup removes a
// queue file it did not queue ("removed (canceled)"): the milter had written
// the rows, but the message was never queued.
func (s *Store) ForgetInboundQueueID(queueID string) error {
if _, err := s.db.Exec(forgetQueueIDQuery, queueID); err != nil {
return fmt.Errorf("forget inbound queue id %s: %w", queueID, err)
}
return nil
}
// UpstreamReplyMax is how much of an upstream's reply a row keeps.
const UpstreamReplyMax = 255
// UpdateInboundUpstream puts the result of one attempt to hand a message to
// its upstream on its journal rows - those of queueID and recipient (case
// folded, as Postfix logs the address it was given) - unless they hold a
// final one already: a line read twice after a restart, or an older one read
// after a newer, does not undo a sent. relay is the host that answered, ""
// when there was no session; reply is cut to UpstreamReplyMax by the caller.
// It returns how many rows took it: 0 for mail that is not port 25's. A
// message the filter had held takes its results once it is released.
func (s *Store) UpdateInboundUpstream(queueID, recipient, status, relay, reply string, at time.Time) (int64, error) {
res, err := s.db.Exec(updateUpstreamQuery, status, relay, reply, at.UTC().Format(time.RFC3339), queueID, recipient)
if err != nil {
return 0, fmt.Errorf("update inbound upstream %s: %w", queueID, err)
}
return res.RowsAffected()
}
// ExpireInboundUpstream makes the rows of a message Postfix gave up on
// bounced (qmgr's "status=expired, returned to sender", which names no
// recipient): the sender was sent a bounce. The last reply stays, after a
// note that the message expired.
func (s *Store) ExpireInboundUpstream(queueID string, at time.Time) (int64, error) {
res, err := s.db.Exec(expireUpstreamQuery, at.UTC().Format(time.RFC3339), UpstreamReplyMax, queueID)
if err != nil {
return 0, fmt.Errorf("expire inbound upstream %s: %w", queueID, err)
}
return res.RowsAffected()
}
// PruneInboundSpam deletes journal rows older than cutoff and then, if more // 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. // 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) { func (s *Store) PruneInboundSpam(cutoff time.Time, keep int) (int64, error) {
@@ -130,6 +216,12 @@ type InboundSpamRow struct {
Engine string Engine string
Detail string Detail string
CreatedAt time.Time CreatedAt time.Time
// The upstream's answer: its status ("" for none), the host that gave
// it, the reply and when.
Upstream string
UpstreamRelay string
UpstreamReply string
UpstreamAt time.Time
} }
// InboundSpamFilter narrows QueryInboundSpam and CountInboundSpam. Like // InboundSpamFilter narrows QueryInboundSpam and CountInboundSpam. Like
@@ -166,7 +258,8 @@ func (s *Store) QueryInboundSpam(filter InboundSpamFilter, limit, offset int) ([
where, args := inboundSpamWhere(filter) where, args := inboundSpamWhere(filter)
args = append(args, limit, offset) args = append(args, limit, offset)
rows, err := s.db.Query( rows, err := s.db.Query(
`SELECT id, inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at `SELECT id, inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at,
upstream_status, upstream_relay, upstream_reply, upstream_at
FROM inbound_spam_log`+where+` FROM inbound_spam_log`+where+`
ORDER BY created_at DESC, id DESC ORDER BY created_at DESC, id DESC
LIMIT ? OFFSET ?`, LIMIT ? OFFSET ?`,
@@ -181,13 +274,17 @@ func (s *Store) QueryInboundSpam(filter InboundSpamFilter, limit, offset int) ([
for rows.Next() { for rows.Next() {
var ( var (
row InboundSpamRow row InboundSpamRow
created string created, upd string
) )
if err := rows.Scan(&row.ID, &row.InboundDomain, &row.ClientIP, &row.From, &row.To, if err := rows.Scan(&row.ID, &row.InboundDomain, &row.ClientIP, &row.From, &row.To,
&row.Subject, &row.Decision, &row.Engine, &row.Detail, &created); err != nil { &row.Subject, &row.Decision, &row.Engine, &row.Detail, &created,
&row.Upstream, &row.UpstreamRelay, &row.UpstreamReply, &upd); err != nil {
return nil, fmt.Errorf("scan inbound_spam_log row: %w", err) return nil, fmt.Errorf("scan inbound_spam_log row: %w", err)
} }
row.CreatedAt, _ = time.Parse(time.RFC3339, created) row.CreatedAt, _ = time.Parse(time.RFC3339, created)
if upd != "" {
row.UpstreamAt, _ = time.Parse(time.RFC3339, upd)
}
out = append(out, row) out = append(out, row)
} }
return out, rows.Err() return out, rows.Err()
+141
View File
@@ -1,6 +1,7 @@
package store package store
import ( import (
"strings"
"testing" "testing"
"time" "time"
) )
@@ -344,3 +345,143 @@ func TestInsertInboundSpamOnce(t *testing.T) {
t.Errorf("rows = %d, %v; want 3", n, err) t.Errorf("rows = %d, %v; want 3", n, err)
} }
} }
// Accepted mail starts queued with its queue id; each delivery line moves it
// on until sent or bounced, which no later or repeated line undoes. A
// refusal, which has no queue id, and another recipient of the message are
// left alone; the queue id goes by its own index.
func TestInboundUpstream(t *testing.T) {
st := openTestStore(t)
accepted := InboundSpamEntry{
InboundDomain: "in.example.com", ClientIP: "203.0.113.7", From: "s@example.net", To: "a@in.example.com",
Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, QueueID: "4A1B2C3D", Upstream: UpstreamQueued,
}
other := accepted
other.To = "b@in.example.com"
refused := InboundSpamEntry{InboundDomain: "in.example.com", From: "s@example.net", To: "a@in.example.com",
Decision: SpamDecisionReject, Engine: SpamEnginePostfix, Detail: "550 5.1.1 unknown"}
for _, e := range []InboundSpamEntry{accepted, other, refused} {
if err := st.InsertInboundSpam(e); err != nil {
t.Fatal(err)
}
}
result := func(to string) InboundSpamRow {
t.Helper()
rows, err := st.QueryInboundSpam(InboundSpamFilter{}, 10, 0)
if err != nil {
t.Fatal(err)
}
for _, r := range rows {
if r.To == to && r.Decision == SpamDecisionAccept {
return r
}
}
t.Fatalf("no accepted row for %s", to)
return InboundSpamRow{}
}
if r := result("a@in.example.com"); r.Upstream != UpstreamQueued || !r.UpstreamAt.IsZero() {
t.Errorf("a new row = %q at %v, want queued and no time", r.Upstream, r.UpstreamAt)
}
at := time.Date(2026, 10, 16, 10, 16, 42, 0, time.UTC)
step := func(status, reply string, want int64) {
t.Helper()
n, err := st.UpdateInboundUpstream("4A1B2C3D", "A@IN.example.com", status, "mx.up.example[198.51.100.5]:25", reply, at)
if err != nil || n != want {
t.Fatalf("UpdateInboundUpstream(%s) = %d, %v; want %d", status, n, err, want)
}
at = at.Add(time.Minute)
}
step(UpstreamDeferred, "451 4.7.1 Greylisted", 1)
if r := result("a@in.example.com"); r.Upstream != UpstreamDeferred || r.UpstreamReply != "451 4.7.1 Greylisted" ||
r.UpstreamRelay != "mx.up.example[198.51.100.5]:25" || !r.UpstreamAt.Equal(time.Date(2026, 10, 16, 10, 16, 42, 0, time.UTC)) {
t.Errorf("after a deferral: %+v", r)
}
step(UpstreamSent, "250 2.0.0 Ok: queued as CE05D10261D", 1)
step(UpstreamDeferred, "451 4.7.1 Greylisted", 0) // read again after a restart
if r := result("a@in.example.com"); r.Upstream != UpstreamSent || r.UpstreamReply != "250 2.0.0 Ok: queued as CE05D10261D" {
t.Errorf("a sent row was moved back: %+v", r)
}
if r := result("b@in.example.com"); r.Upstream != UpstreamQueued {
t.Errorf("another recipient took the result: %+v", r)
}
if n, err := st.UpdateInboundUpstream("9999", "a@in.example.com", UpstreamSent, "", "250 ok", at); err != nil || n != 0 {
t.Errorf("an unknown queue id = %d, %v; want 0 (outbound mail)", n, err)
}
if n, err := st.UpdateInboundUpstream("", "a@in.example.com", UpstreamSent, "", "250 ok", at); err != nil || n != 0 {
t.Errorf("an empty queue id took %d rows, %v - a refusal has none", n, err)
}
// Expiry: the other recipient, still queued, bounces, and says so.
if _, err := st.UpdateInboundUpstream("4A1B2C3D", "b@in.example.com", UpstreamDeferred, "", "connect to mx.up.example[198.51.100.5]:25: Connection refused", at); err != nil {
t.Fatal(err)
}
if n, err := st.ExpireInboundUpstream("4A1B2C3D", at); err != nil || n != 1 {
t.Fatalf("ExpireInboundUpstream = %d, %v; want 1 (the sent row stays)", n, err)
}
if r := result("b@in.example.com"); r.Upstream != UpstreamBounced ||
r.UpstreamReply != "expired in the queue; last reply: connect to mx.up.example[198.51.100.5]:25: Connection refused" {
t.Errorf("after the expiry: %+v", r)
}
if err := st.InsertInboundSpam(InboundSpamEntry{InboundDomain: "in.example.com", To: "c@in.example.com",
Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, Upstream: "lost"}); err == nil {
t.Error("an unknown upstream status was stored")
}
for name, q := range map[string]string{"update": updateUpstreamQuery, "expire": expireUpstreamQuery} {
args := []any{"sent", "", "", "", "4A1B2C3D", "a@in.example.com"}
if name == "expire" {
args = []any{"", 255, "4A1B2C3D"}
}
if plan := explain(t, st, q, args...); !strings.Contains(plan, "USING INDEX idx_inbound_spam_log_queue_id") {
t.Errorf("%s does not use idx_inbound_spam_log_queue_id:\n%s", name, plan)
}
}
}
// A held message takes its result once released; a queue id Postfix uses
// again detaches the rows of the message that had it - the open ones lose
// their result, nothing reports it now - so the new message's lines and its
// expiry reach only its own rows.
func TestInboundUpstreamHeldAndReusedQueueID(t *testing.T) {
st := openTestStore(t)
old := InboundSpamEntry{InboundDomain: "in.example.com", To: "a@in.example.com",
Decision: SpamDecisionQuarantine, Engine: SpamEngineRspamd, QueueID: "4A1B", Upstream: UpstreamHeld}
if err := st.InsertInboundSpam(old); err != nil {
t.Fatal(err)
}
at := time.Date(2026, 10, 16, 10, 0, 0, 0, time.UTC)
if n, err := st.UpdateInboundUpstream("4A1B", "a@in.example.com", UpstreamDeferred, "", "451 later", at); err != nil || n != 1 {
t.Fatalf("a released held message: %d, %v", n, err)
}
if err := st.ForgetInboundQueueID("4A1B"); err != nil {
t.Fatal(err)
}
fresh := InboundSpamEntry{InboundDomain: "in.example.com", To: "a@in.example.com",
Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, QueueID: "4A1B", Upstream: UpstreamQueued}
if err := st.InsertInboundSpam(fresh); err != nil {
t.Fatal(err)
}
if n, err := st.ExpireInboundUpstream("4A1B", at); err != nil || n != 1 {
t.Fatalf("expiry of the reused id took %d rows, %v; want the new message's one", n, err)
}
rows, err := st.QueryInboundSpam(InboundSpamFilter{}, 10, 0)
if err != nil {
t.Fatal(err)
}
for _, r := range rows {
switch r.Decision {
case SpamDecisionQuarantine:
if r.Upstream != "" || r.UpstreamReply != "451 later" {
t.Errorf("the detached row = %+v, want no result and its last reply kept", r)
}
case SpamDecisionAccept:
if r.Upstream != UpstreamBounced {
t.Errorf("the new row = %+v, want bounced", r)
}
}
}
if plan := explain(t, st, forgetQueueIDQuery, "4A1B"); !strings.Contains(plan, "USING INDEX idx_inbound_spam_log_queue_id") {
t.Errorf("ForgetInboundQueueID does not use idx_inbound_spam_log_queue_id:\n%s", plan)
}
}
@@ -0,0 +1,20 @@
-- What the upstream answered, on the inbound journal's rows
-- (inbound-upstream-result, 2.10.0; owner, 2026-10-16). A row of mail
-- accepted on port 25 keeps the message's queue id, which the inbound journal
-- milter reads at end-of-message, and starts "queued" ("held" when the filter
-- had it held in the queue); the log tailer puts
-- each delivery attempt's result on it - the status, the host that answered
-- ("" when Postfix had no session), its reply and when - from the postfix/smtp
-- line of that queue id and recipient. sent and bounced are final. A refused
-- message has no queue id and no result, and neither has a row written before
-- this file.
ALTER TABLE inbound_spam_log ADD COLUMN queue_id TEXT NOT NULL DEFAULT '';
ALTER TABLE inbound_spam_log ADD COLUMN upstream_status TEXT NOT NULL DEFAULT ''
CHECK (upstream_status IN ('', 'queued', 'held', 'sent', 'deferred', 'bounced'));
ALTER TABLE inbound_spam_log ADD COLUMN upstream_relay TEXT NOT NULL DEFAULT '';
ALTER TABLE inbound_spam_log ADD COLUMN upstream_reply TEXT NOT NULL DEFAULT '';
ALTER TABLE inbound_spam_log ADD COLUMN upstream_at TEXT NOT NULL DEFAULT '';
-- The tailer finds a row by its queue id at every delivery line; only rows
-- of accepted mail have one.
CREATE INDEX idx_inbound_spam_log_queue_id ON inbound_spam_log(queue_id) WHERE queue_id <> '';
+6 -6
View File
@@ -57,7 +57,7 @@ func TestBaselineSchema(t *testing.T) {
"user_inbound_domains": "user_id inbound_domain_id", "user_inbound_domains": "user_id inbound_domain_id",
"dmarc_reports": "id domain reporter report_id period_begin period_end received_at contact_email policy_p policy_sp policy_pct policy_adkim policy_aspf pass_count fail_count recipient", "dmarc_reports": "id domain reporter report_id period_begin period_end received_at contact_email policy_p policy_sp policy_pct policy_adkim policy_aspf pass_count fail_count recipient",
"dmarc_report_records": "id report_row_id source_ip count disposition spf_result dkim_result header_from", "dmarc_report_records": "id report_row_id source_ip count disposition spf_result dkim_result header_from",
"inbound_spam_log": "id inbound_domain client_ip from_addr to_addr subject decision engine detail created_at inbound_domain_id", "inbound_spam_log": "id inbound_domain client_ip from_addr to_addr subject decision engine detail created_at inbound_domain_id queue_id upstream_status upstream_relay upstream_reply upstream_at",
"inbound_spam_list": "id list_type entry_type value note created_at", "inbound_spam_list": "id list_type entry_type value note created_at",
"send_log_lines": "id queue_id logged_at line", "send_log_lines": "id queue_id logged_at line",
"notify_state": "key phase count mark since sent_at", "notify_state": "key phase count mark since sent_at",
@@ -108,7 +108,7 @@ func TestBaselineSchema(t *testing.T) {
rows.Close() rows.Close()
wantIdx := []string{ wantIdx := []string{
"idx_applications_login_nocase", "idx_dmarc_report_records_report", "idx_dmarc_reports_domain", "idx_dmarc_reports_received", "idx_applications_login_nocase", "idx_dmarc_report_records_report", "idx_dmarc_reports_domain", "idx_dmarc_reports_received",
"idx_inbound_spam_log_created_at", "idx_inbound_spam_log_domain_created_at", "idx_inbound_spam_log_domain_id_created_at", "idx_inbound_spam_log_created_at", "idx_inbound_spam_log_domain_created_at", "idx_inbound_spam_log_domain_id_created_at", "idx_inbound_spam_log_queue_id",
"idx_send_log_app_login", "idx_send_log_app_login_created_at", "idx_send_log_created_at", "idx_send_log_domain", "idx_send_log_domain_id", "idx_send_log_domain_id_created_at", "idx_send_log_app_login", "idx_send_log_app_login_created_at", "idx_send_log_created_at", "idx_send_log_domain", "idx_send_log_domain_id", "idx_send_log_domain_id_created_at",
"idx_send_log_lines_queue_id", "idx_send_log_open", "idx_send_log_queue_id", "idx_send_log_refusal_line", "idx_send_log_rejected", "idx_send_log_lines_queue_id", "idx_send_log_open", "idx_send_log_queue_id", "idx_send_log_refusal_line", "idx_send_log_rejected",
"idx_sessions_expires_at", "idx_sessions_expires_at",
@@ -119,8 +119,8 @@ func TestBaselineSchema(t *testing.T) {
} }
var version, appID int64 var version, appID int64
if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 11 { if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 12 {
t.Errorf("user_version = %d, %v; want 11", version, err) t.Errorf("user_version = %d, %v; want 12", version, err)
} }
if err := st.db.QueryRow("PRAGMA application_id").Scan(&appID); err != nil || appID != applicationID { if err := st.db.QueryRow("PRAGMA application_id").Scan(&appID); err != nil || appID != applicationID {
t.Errorf("application_id = %#x, %v; want %#x", appID, err, applicationID) t.Errorf("application_id = %#x, %v; want %#x", appID, err, applicationID)
@@ -196,8 +196,8 @@ func TestOpenMigratesA20Database(t *testing.T) {
} }
defer st.Close() defer st.Close()
var version int var version int
if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 11 { if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 12 {
t.Errorf("user_version = %d, %v; want 11", version, err) t.Errorf("user_version = %d, %v; want 12", version, err)
} }
if ok, err := st.UserExists(); err != nil || !ok { if ok, err := st.UserExists(); err != nil || !ok {
t.Errorf("the user of the 2.0 database: exists = %v, %v", ok, err) t.Errorf("the user of the 2.0 database: exists = %v, %v", ok, err)
@@ -94,6 +94,7 @@ func (h *Handlers) HandleInboundLog(w http.ResponseWriter, r *http.Request) {
table[i] = out.NewRow(view.InLogRowInput{ table[i] = out.NewRow(view.InLogRowInput{
ID: row.ID, Listable: inbound.Listable(store.SpamEntryAddress, row.From), At: row.CreatedAt, From: row.From, Client: row.ClientIP, To: row.To, Subject: row.Subject, ID: row.ID, Listable: inbound.Listable(store.SpamEntryAddress, row.From), At: row.CreatedAt, From: row.From, Client: row.ClientIP, To: row.To, Subject: row.Subject,
Decision: row.Decision, Engine: row.Engine, Detail: row.Detail, Decision: row.Decision, Engine: row.Engine, Detail: row.Detail,
Upstream: row.Upstream, UpstreamRelay: row.UpstreamRelay, UpstreamReply: row.UpstreamReply, UpstreamAt: row.UpstreamAt,
}) })
} }
h.view.Render(w, http.StatusOK, "in-log", out.WithRows(table, page, lastPage)) h.view.Render(w, http.StatusOK, "in-log", out.WithRows(table, page, lastPage))
+34 -13
View File
@@ -448,29 +448,50 @@ func inLogFixture() *InLog {
p := NewInLog(admin(), 90, []string{"lists.example.org", "acme.io"}, p := NewInLog(admin(), 90, []string{"lists.example.org", "acme.io"},
InLogFilter{From: "2026-09-14", To: "2026-09-21"}) InLogFilter{From: "2026-09-14", To: "2026-09-21"})
var id int64 var id int64
row := func(clock, from, client, to, subject, decision, engine, detail string) InLogRow { clock := func(c string) time.Time {
at, err := time.Parse("2006-01-02 15:04:05", "2026-09-21 "+clock) at, err := time.Parse("2006-01-02 15:04:05", "2026-09-21 "+c)
if err != nil { if err != nil {
panic(err) panic(err)
} }
id++ return at
return p.NewRow(InLogRowInput{ID: id, Listable: true, At: at, From: from, Client: client, To: to, Subject: subject,
Decision: decision, Engine: engine, Detail: detail})
} }
return p.WithRows([]InLogRow{ // up is what the upstream answered: status, host, reply, when.
row("14:00:18", "t.berg@uni.example", "192.0.2.77", "dev@lists.example.org", "Re: [dev] patch series v3", type up struct{ status, relay, reply, at string }
"accept", "rspamd", "no action, score 0.40 / 15.00"), upRow := func(c, from, client, to, subject, decision, engine, detail string, u up) InLogRow {
row("13:59:02", "invoice@partner-billing.example", "203.0.113.140", "owner@lists.example.org", "Outstanding invoice — action required", id++
"accept", "rspamd", "add header, score 9.80 / 15.00; PHISHING(5.00) FORGED_SENDER(3.00) R_SPF_FAIL(1.00)"), in := InLogRowInput{ID: id, Listable: true, At: clock(c), From: from, Client: client, To: to, Subject: subject,
Decision: decision, Engine: engine, Detail: detail,
Upstream: u.status, UpstreamRelay: u.relay, UpstreamReply: u.reply}
if u.at != "" {
in.UpstreamAt = clock(u.at)
}
return p.NewRow(in)
}
row := func(c, from, client, to, subject, decision, engine, detail string) InLogRow {
return upRow(c, from, client, to, subject, decision, engine, detail, up{})
}
const host = "mx.upstream.example[198.51.100.5]:25"
sent := upRow("14:00:18", "t.berg@uni.example", "192.0.2.77", "dev@lists.example.org", "Re: [dev] patch series v3",
"accept", "rspamd", "no action, score 0.40 / 15.00",
up{"sent", host, "250 2.0.0 Ok: queued as CE05D10261D", "14:00:19"})
bounced := upRow("13:59:02", "invoice@partner-billing.example", "203.0.113.140", "owner@lists.example.org", "Outstanding invoice — action required",
"accept", "rspamd", "add header, score 9.80 / 15.00; PHISHING(5.00) FORGED_SENDER(3.00) R_SPF_FAIL(1.00)",
up{"bounced", host, "550 5.7.1 Message rejected as spam by the upstream filter", "13:59:04"})
rest := []InLogRow{
row("13:57:41", "promo@bulk.example", "198.51.100.90", "sales@lists.example.org", "", row("13:57:41", "promo@bulk.example", "198.51.100.90", "sales@lists.example.org", "",
"reject", "postfix", "550 5.1.1 <sales@lists.example.org>: Recipient address rejected: User unknown in relay recipient table"), "reject", "postfix", "550 5.1.1 <sales@lists.example.org>: Recipient address rejected: User unknown in relay recipient table"),
row("13:51:30", "billing@partner.example", "198.51.100.12", "owner@lists.example.org", "", row("13:51:30", "billing@partner.example", "198.51.100.12", "owner@lists.example.org", "",
"reject", "rspamd", "5.7.1 Spam message rejected"), "reject", "rspamd", "5.7.1 Spam message rejected"),
row("13:44:02", "hr@partner.example", "192.0.2.31", "jobs@acme.io", "", row("13:44:02", "hr@partner.example", "192.0.2.31", "jobs@acme.io", "",
"tempfail", "rspamd", "4.7.1 Try again later"), "tempfail", "rspamd", "4.7.1 Try again later"),
row("13:30:55", "noreply@github.example", "192.0.2.200", "security@lists.example.org", "[advisory] new report", upRow("13:30:55", "noreply@github.example", "192.0.2.200", "security@lists.example.org", "[advisory] new report",
"accept", "rspamd", "no action, score -98.10 / 15.00; SELFPOST_ALLOW_FROM(-100.00) R_DKIM_ALLOW(-0.20)"), "accept", "rspamd", "no action, score -98.10 / 15.00; SELFPOST_ALLOW_FROM(-100.00) R_DKIM_ALLOW(-0.20)",
}, 1, 38) up{"deferred", "", "connect to mx.upstream.example[198.51.100.5]:25: Connection refused", "13:45:12"}),
}
// The newest row is a message not tried yet; it is the seventh written.
queued := upRow("14:00:31", "ci@builds.example", "192.0.2.90", "dev@lists.example.org", "[ci] nightly build passed",
"accept", "rspamd", "no action, score 1.10 / 15.00", up{status: "queued"})
return p.WithRows(append([]InLogRow{queued, sent, bounced}, rest...), 1, 38)
} }
// inFilterListsFixture is the mockup's Filter lists: three senders denied, two // inFilterListsFixture is the mockup's Filter lists: three senders denied, two
+7 -2
View File
@@ -207,11 +207,16 @@ var inboundTopics = []HelpTopic{
}}, }},
{ID: HelpInboundLog, Title: "Inbound log", Body: []Text{ {ID: HelpInboundLog, Title: "Inbound log", Body: []Text{
Rich("One row per recipient of a message offered on port 25 for an inbound domain, with what was decided: ", Rich("One row per recipient of a message offered on port 25 for an inbound domain, with what was decided: ",
TagOf("accept", ""), " — handed on to the upstream; ", TagOf("reject", ""), " — refused for good; ", TagOf("accept", ""), " — taken for the upstream; ", TagOf("reject", ""), " — refused for good; ",
TagOf("tempfail", ""), " — the sender was asked to come back later; ", TagOf("quarantine", ""), " — held by the filter. ", TagOf("tempfail", ""), " — the sender was asked to come back later; ", TagOf("quarantine", ""), " — held by the filter. ",
Em("Why"), " names who decided. ", Code("rspamd"), " is the spam filter, with its action, the score against the reject threshold and the heaviest symbols; ", Em("Why"), " names who decided. ", Code("rspamd"), " is the spam filter, with its action, the score against the reject threshold and the heaviest symbols; ",
Code("postfix"), " is the relay's own rule — an unknown recipient, a sender domain that does not exist — with its reply; ", Code("postfix"), " is the relay's own rule — an unknown recipient, a sender domain that does not exist — with its reply; ",
Code("selfpost"), " means nobody judged the message: it was relayed with no filter attached. Accept says what this server did, not what the upstream did with the message afterwards.", Code("selfpost"), " means nobody judged the message: it was relayed with no filter attached. ",
Em("Upstream"), " says what became of an accepted message: ", TagOf("queued", ""), " — not tried yet; ",
TagOf("held", ""), " — held in the queue by the filter, not tried until it is released; ",
TagOf("sent", ""), " — the upstream took it; ", TagOf("deferred", ""),
" — the upstream put it off or could not be reached, and Postfix tries again; ", TagOf("bounced", ""),
" — the upstream refused it, or it waited in the queue too long, and the sender was sent a bounce, if it had a return address. It opens the last reply and the host that gave it. A refused message has no upstream result, and neither has a row logged before 2.10.0.",
Br(), Br(), Br(), Br(),
"Accepted mail is recorded as it passes; refused mail is read from ", Code("mail.log"), "Accepted mail is recorded as it passes; refused mail is read from ", Code("mail.log"),
", so its row has no subject, and a message refused after its text was sent is one row however many recipients it had. A refusal to relay for a domain that is not an inbound one is every scanner on the internet and is not recorded. Rows are kept for the number of days set in Settings, and never more than 10 000."), ", so its row has no subject, and a message refused after its text was sent is one row however many recipients it had. A refusal to relay for a domain that is not an inbound one is every scanner on the internet and is not recorded. Rows are kept for the number of days set in Settings, and never more than 10 000."),
+56 -6
View File
@@ -68,9 +68,9 @@ type InLog struct {
// InLogRow is one recipient of one message: when it was decided, who the mail // InLogRow is one recipient of one message: when it was decided, who the mail
// was from (and the address of the server that offered it) and for, its subject, // was from (and the address of the server that offered it) and for, its subject,
// the decision as a tag that opens the sheet saying who decided and why, and — // the decision as a tag that opens the sheet saying who decided and why, what
// for the global role, who alone edits the filter lists — the way to correct // the upstream made of it, and — for the global role, who alone edits the
// the decision. // filter lists — the way to correct the decision.
type InLogRow struct { type InLogRow struct {
Time string Time string
From string From string
@@ -78,6 +78,12 @@ type InLogRow struct {
To string To string
Subject string Subject string
Why Sheet Why Sheet
// Upstream is the upstream's answer, a tag that opens a sheet with the
// last reply; Pending is a message not tried yet - queued, or held by the
// filter - a plain tag. Neither for refused mail and rows from before
// 2.10.0, which show "—".
Upstream *Sheet
Pending *Tag
Action *Anchor Action *Anchor
} }
@@ -97,6 +103,12 @@ type InLogRowInput struct {
Decision string Decision string
Engine string Engine string
Detail string Detail string
// The upstream's answer (store.Upstream…): its status, "" for none, the
// host that gave it ("" when Postfix had no session), the reply and when.
Upstream string
UpstreamRelay string
UpstreamReply string
UpstreamAt time.Time
} }
// NewInLog builds the log for a viewer. domains are the choices of the domain // NewInLog builds the log for a viewer. domains are the choices of the domain
@@ -105,7 +117,7 @@ type InLogRowInput struct {
// change it and who edits the lists, is read from m. // change it and who edits the lists, is read from m.
func NewInLog(m Meta, retentionDays int, domains []string, filter InLogFilter) *InLog { func NewInLog(m Meta, retentionDays int, domains []string, filter InLogFilter) *InLog {
m.Title, m.Section, m.Page = "Inbound log", "inbound", "log" m.Title, m.Section, m.Page = "Inbound log", "inbound", "log"
lead := Rich("What port 25 decided about each message, one row per recipient. Accepted mail is recorded as it passes, refused mail is read from ", lead := Rich("What port 25 decided about each message and what the upstream answered, one row per recipient. Accepted mail is recorded as it passes, refused mail is read from ",
Code("mail.log"), "; the filter's verdict appears when the spam filter is on. Rows older than "+plural(int64(retentionDays), "day", "days")+" are deleted") Code("mail.log"), "; the filter's verdict appears when the spam filter is on. Rows older than "+plural(int64(retentionDays), "day", "days")+" are deleted")
if m.IsGlobal { if m.IsGlobal {
lead = append(lead, Inline{Text: " — change that in "}, Link("/server/settings", "Settings")) lead = append(lead, Inline{Text: " — change that in "}, Link("/server/settings", "Settings"))
@@ -152,12 +164,50 @@ func (p *InLog) NewRow(in InLogRowInput) InLogRow {
Rows: whyRows(in.Decision, in.Engine, in.Detail, in.Subject), Rows: whyRows(in.Decision, in.Engine, in.Detail, in.Subject),
}, },
} }
switch in.Upstream {
case "queued", "held":
r.Pending = &Tag{Status: in.Upstream}
case "sent", "deferred", "bounced":
r.Upstream = &Sheet{
ID: "up-" + strconv.FormatInt(in.ID, 10), Tag: Tag{Status: in.Upstream}, Title: "Upstream",
Rows: upstreamRows(in),
}
}
if p.IsGlobal && in.Listable { if p.IsGlobal && in.Listable {
r.Action = senderAction(in.Decision, in.Engine, in.From) r.Action = senderAction(in.Decision, in.Engine, in.From)
} }
return r return r
} }
// upstreamRows is the content of the Upstream sheet: the host that answered
// (none when Postfix could not open a session: the reply names the host
// then), the last reply, when it came, and what follows a deferral or a
// bounce - a bounce to the sender, unless the message had none to return to
// (Postfix does not bounce mail from the null sender).
func upstreamRows(in InLogRowInput) []SheetRow {
var rows []SheetRow
if in.UpstreamRelay != "" {
rows = append(rows, SheetRow{Label: "Host", Lines: []string{in.UpstreamRelay}, Mono: true})
}
if in.UpstreamReply != "" {
rows = append(rows, SheetRow{Label: "Reply", Lines: []string{in.UpstreamReply}, Mono: true})
}
if !in.UpstreamAt.IsZero() {
rows = append(rows, SheetRow{Label: "At", Lines: []string{FormatLogTime(in.UpstreamAt)}, Mono: true})
}
switch in.Upstream {
case "deferred":
rows = append(rows, SheetRow{Label: "Then", Lines: []string{"Postfix tries again; the message is in the queue"}})
case "bounced":
then := "The sender was sent a bounce"
if in.From == "" {
then = "No bounce: the message had no return address"
}
rows = append(rows, SheetRow{Label: "Then", Lines: []string{then}})
}
return rows
}
// whyRows is the content of a row's sheet: who decided, what they said, and // whyRows is the content of a row's sheet: who decided, what they said, and
// the subject whole, which the table clips. What was said has one of two // the subject whole, which the table clips. What was said has one of two
// shapes, and the decision says which. Mail that passed was written by the // shapes, and the decision says which. Mail that passed was written by the
@@ -226,9 +276,9 @@ func (p *InLog) WithRows(rows []InLogRow, page, last int) *InLog {
p.Foot.Link = &Anchor{Href: InLogHref(p.filter, page+1), Label: "Older →"} p.Foot.Link = &Anchor{Href: InLogHref(p.filter, page+1), Label: "Older →"}
} }
if p.IsGlobal { if p.IsGlobal {
p.Foot.End = Rich("A decision opens its reason · Allow / Deny sender opens ", Link(inboundFilterLists, "Filter lists"), " with the entry filled in") p.Foot.End = Rich("A decision opens its reason, an upstream result its last reply · Allow / Deny sender opens ", Link(inboundFilterLists, "Filter lists"), " with the entry filled in")
} else { } else {
p.Foot.End = Plain("A decision opens its reason") p.Foot.End = Plain("A decision opens its reason, an upstream result its last reply")
} }
return p return p
} }
+34 -5
View File
@@ -22,7 +22,18 @@ func TestInLogShowsTheRowsAndTheirFilters(t *testing.T) {
`<input class="input" type="date" name="from" value="2026-09-14" aria-label="From date">`, `<input class="input" type="date" name="from" value="2026-09-14" aria-label="From date">`,
`<input class="input" type="date" name="to" value="2026-09-21" aria-label="To date">`, `<input class="input" type="date" name="to" value="2026-09-21" aria-label="To date">`,
`href="/help#inbound-log" title="What the decisions mean"`, `href="/help#inbound-log" title="What the decisions mean"`,
`<th>Time, UTC</th><th>From</th><th>To</th><th>Subject</th><th>Decision</th><th></th></tr>`, `<th>Time, UTC</th><th>From</th><th>To</th><th>Subject</th><th>Decision</th><th>Upstream</th><th></th></tr>`,
// what the upstream answered: a tag that opens its last reply, a
// plain tag while not tried, "—" for refused mail
`<button type="button" class="tag is-success is-light" popovertarget="up-1">sent</button><div class="sp-sheet" id="up-1" popover>`,
`<span class="tag is-success is-light">sent</span><h2>Upstream</h2>`,
`<tr><th>Host</th><td class="sp-mono sp-small">mx.upstream.example[198.51.100.5]:25</td></tr><tr><th>Reply</th><td class="sp-mono sp-small">250 2.0.0 Ok: queued as CE05D10261D</td></tr><tr><th>At</th><td class="sp-mono sp-small">09-21 14:00:19</td></tr></tbody>`,
`<button type="button" class="tag is-danger is-light" popovertarget="up-2">bounced</button>`,
`<tr><th>Then</th><td>The sender was sent a bounce</td></tr>`,
`<button type="button" class="tag is-warning is-light" popovertarget="up-6">deferred</button>`,
`<tbody><tr><th>Reply</th><td class="sp-mono sp-small">connect to mx.upstream.example[198.51.100.5]:25: Connection refused</td></tr><tr><th>At</th><td class="sp-mono sp-small">09-21 13:45:12</td></tr><tr><th>Then</th><td>Postfix tries again; the message is in the queue</td></tr></tbody>`,
`<td><span class="tag is-light">queued</span></td>`,
`</div></div></div></div></td><td class="sp-muted">—</td><td class="sp-actions">`,
`09-21 14:00:18`, `09-21 14:00:18`,
`t.berg@<wbr>uni.example<br><span class="sp-mono sp-muted sp-small">192.0.2.77</span>`, `t.berg@<wbr>uni.example<br><span class="sp-mono sp-muted sp-small">192.0.2.77</span>`,
`dev@<wbr>lists.example.org`, `title="Re: [dev] patch series v3"`, `dev@<wbr>lists.example.org`, `title="Re: [dev] patch series v3"`,
@@ -36,14 +47,17 @@ func TestInLogShowsTheRowsAndTheirFilters(t *testing.T) {
`<tr><th>Verdict</th><td>add header, score 9.80 / 15.00</td></tr><tr><th>Symbols</th><td class="sp-mono sp-small">PHISHING(5.00)<br>FORGED_SENDER(3.00)<br>R_SPF_FAIL(1.00)</td></tr>`, `<tr><th>Verdict</th><td>add header, score 9.80 / 15.00</td></tr><tr><th>Symbols</th><td class="sp-mono sp-small">PHISHING(5.00)<br>FORGED_SENDER(3.00)<br>R_SPF_FAIL(1.00)</td></tr>`,
`<tr><th>Decided by</th><td>postfix</td></tr><tr><th>Reply</th><td class="sp-mono sp-small">550 5.1.1 &lt;sales@lists.example.org&gt;: Recipient address rejected: User unknown in relay recipient table</td></tr></tbody>`, `<tr><th>Decided by</th><td>postfix</td></tr><tr><th>Reply</th><td class="sp-mono sp-small">550 5.1.1 &lt;sales@lists.example.org&gt;: Recipient address rejected: User unknown in relay recipient table</td></tr></tbody>`,
`Page 1 of 38`, `<a href="/inbound/log?from=2026-09-14&amp;p=2&amp;to=2026-09-21">Older →</a>`, `Page 1 of 38`, `<a href="/inbound/log?from=2026-09-14&amp;p=2&amp;to=2026-09-21">Older →</a>`,
`A decision opens its reason · Allow / Deny sender opens <a href="/inbound/filter-lists">Filter lists</a> with the entry filled in`, `A decision opens its reason, an upstream result its last reply · Allow / Deny sender opens <a href="/inbound/filter-lists">Filter lists</a> with the entry filled in`,
) )
pageLacks(t, "Inbound log", out, `← Newer`, `data-poll`, `hx-get`, `<th>Why</th>`) pageLacks(t, "Inbound log", out, `← Newer`, `data-poll`, `hx-get`, `<th>Why</th>`)
// A message Postfix itself refused offers no way into the lists: one Deny // A message Postfix itself refused offers no way into the lists: one Deny
// for each of the three accepted messages, one Allow for each the filter // for each of the four accepted messages, one Allow for each the filter
// refused or put off. // refused or put off.
if deny, allow := strings.Count(out, `>Deny sender</a>`), strings.Count(out, `>Allow sender</a>`); deny != 3 || allow != 2 { if deny, allow := strings.Count(out, `>Deny sender</a>`), strings.Count(out, `>Allow sender</a>`); deny != 4 || allow != 2 {
t.Errorf("the fixture's six rows offer Deny %d times and Allow %d times, want 3 and 2", deny, allow) t.Errorf("the fixture's seven rows offer Deny %d times and Allow %d times, want 4 and 2", deny, allow)
}
if n := strings.Count(out, `<td class="sp-muted">—</td>`); n != 3 {
t.Errorf("%d rows have no upstream result, want the three refused", n)
} }
// Without the global role there is nowhere to change the retention, and // Without the global role there is nowhere to change the retention, and
@@ -308,3 +322,18 @@ func TestWhyRows(t *testing.T) {
} }
} }
} }
// A message the filter had held is a plain tag until it is released, and a
// bounce of mail from the null sender says none went back.
func TestInLogHeldAndABounceWithNoReturnAddress(t *testing.T) {
p := NewInLog(admin(), 90, nil, InLogFilter{})
at := time.Date(2026, 9, 21, 14, 0, 0, 0, time.UTC)
out := renderSignedIn(t, "in-log", p.WithRows([]InLogRow{
p.NewRow(InLogRowInput{ID: 1, At: at, From: "a@example.net", To: "b@acme.io", Decision: "quarantine", Engine: "rspamd", Upstream: "held"}),
p.NewRow(InLogRowInput{ID: 2, At: at, To: "b@acme.io", Decision: "accept", Engine: "selfpost",
Upstream: "bounced", UpstreamReply: "550 5.1.1 no such user", UpstreamAt: at}),
}, 1, 1))
pageHas(t, "Inbound log", out, `<td><span class="tag is-primary is-light">held</span></td>`,
`<tr><th>Then</th><td>No bounce: the message had no return address</td></tr>`)
pageLacks(t, "Inbound log", out, `The sender was sent a bounce`, `<th>Host</th>`)
}
+4 -3
View File
@@ -1,7 +1,8 @@
{{/* Inbound log (mockup in-log.html): what port 25 decided about each message {{/* Inbound log (mockup in-log.html): what port 25 decided about each message
offered for an inbound domain, one row per recipient, newest first, offered for an inbound domain, one row per recipient, newest first,
filtered by domain, decision and day. It does not poll. The decision is a filtered by domain, decision and day. It does not poll. The decision is a
tag that opens a sheet with who decided and why. The last cell is the way tag that opens a sheet with who decided and why; the upstream's answer one
with its last reply (queued or held, not tried yet, is a plain tag). The last cell is the way
to correct a decision in Filter lists, and only the global role, who alone to correct a decision in Filter lists, and only the global role, who alone
edits the lists, is given it. */}} edits the lists, is given it. */}}
{{define "content"}} {{define "content"}}
@@ -10,9 +11,9 @@
{{template "box_open" .Messages}} {{template "box_open" .Messages}}
{{if .Rows}} {{if .Rows}}
<div class="table-container"><table class="table is-fullwidth is-hoverable"> <div class="table-container"><table class="table is-fullwidth is-hoverable">
<thead><tr><th>Time, UTC</th><th>From</th><th>To</th><th>Subject</th><th>Decision</th><th></th></tr></thead> <thead><tr><th>Time, UTC</th><th>From</th><th>To</th><th>Subject</th><th>Decision</th><th>Upstream</th><th></th></tr></thead>
<tbody>{{range .Rows}} <tbody>{{range .Rows}}
<tr><td class="sp-mono sp-muted sp-nowrap sp-small">{{.Time}}</td><td>{{wbr_at .From}}{{if .Client}}<br><span class="sp-mono sp-muted sp-small">{{.Client}}</span>{{end}}</td><td>{{wbr_at .To}}</td>{{if .Subject}}<td class="sp-clip" title="{{.Subject}}">{{.Subject}}</td>{{else}}<td class="sp-clip sp-muted">—</td>{{end}}<td>{{template "sheet_tag" .Why}}</td><td class="sp-actions">{{if .Action}}<a href="{{.Action.Href}}">{{.Action.Label}}</a>{{end}}</td></tr>{{end}} <tr><td class="sp-mono sp-muted sp-nowrap sp-small">{{.Time}}</td><td>{{wbr_at .From}}{{if .Client}}<br><span class="sp-mono sp-muted sp-small">{{.Client}}</span>{{end}}</td><td>{{wbr_at .To}}</td>{{if .Subject}}<td class="sp-clip" title="{{.Subject}}">{{.Subject}}</td>{{else}}<td class="sp-clip sp-muted">—</td>{{end}}<td>{{template "sheet_tag" .Why}}</td>{{if .Upstream}}<td>{{template "sheet_tag" .Upstream}}</td>{{else if .Pending}}<td>{{template "tag" .Pending}}</td>{{else}}<td class="sp-muted">—</td>{{end}}<td class="sp-actions">{{if .Action}}<a href="{{.Action.Href}}">{{.Action.Label}}</a>{{end}}</td></tr>{{end}}
</tbody> </tbody>
</table></div> </table></div>
{{template "box_foot" .Foot}} {{template "box_foot" .Foot}}
+104
View File
@@ -0,0 +1,104 @@
package e2e
import (
"fmt"
"net/url"
"regexp"
"strings"
"testing"
"time"
)
// testInboundUpstreamDown: an inbound domain whose upstream refuses
// connections (the sink's address, a port nothing listens on). Port 25 takes
// the message; Postfix cannot hand it on and logs "relay=none ...
// status=deferred (connect to ...: Connection refused)" (Postfix 3.10.13,
// src/smtp/smtp_trouble.c:273, 421), which the Inbound log shows as deferred
// with that reason and no host. Then the message is made to expire - postsuper
// -e, postqueue -i, as testExpiredMessageIsClosed does - and its row is
// bounced, its last reply kept. The queue is emptied after: the bounce to the
// stranger has nowhere to go on the stand. Runs inside testInboundRelay.
func testInboundUpstreamDown(t *testing.T, sc *scenario, message func(to, token string) string) {
const domain = "down.e2e.test"
to := "user@" + domain
resp, body, err := sc.panel.postForm("/inbound/domains", url.Values{"name": {domain}})
if err != nil {
t.Fatal(err)
}
m := regexp.MustCompile(`^/inbound/domains/(\d+)$`).FindStringSubmatch(resp.Request.URL.Path)
if m == nil {
t.Fatalf("add inbound domain: landed on %s:\n%.800s", resp.Request.URL.Path, body)
}
base := "/inbound/domains/" + m[1]
for path, form := range map[string]url.Values{
base + "/upstream": {"host": {"10.77.0.11"}, "port": {"2526"}, "tls_mode": {"none"}},
base + "/recipients": {"recipient_mode": {"list"}, "addresses": {to}},
} {
resp, body, err := sc.panel.postForm(path, form)
if err != nil {
t.Fatal(err)
}
if resp.StatusCode != 200 || resp.Request.URL.Path != base {
t.Fatalf("POST %s: status %d, landed on %s:\n%.800s", path, resp.StatusCode, resp.Request.URL.Path, body)
}
}
defer h.execIn("selfpost", "postsuper", "-d", "ALL")
token := uniqueToken("inbound-down")
if res := deliverPort25("someone@stranger.e2e.test", to, message(to, token)); !res.ok() {
t.Fatalf("port 25 did not accept mail for %s: %v", to, res.firstErr())
}
row := func() (string, error) {
return asPanel(`import sqlite3,sys
r=sqlite3.connect("/data/selfpost.db").execute("select queue_id, upstream_status, upstream_relay, upstream_reply from inbound_spam_log where subject like ? order by id desc",("%"+sys.argv[1]+"%",)).fetchone()
print("" if r is None else "|".join(r))`, token)
}
var queueID string
if err := waitFor("the row to be deferred", 60*time.Second, time.Second, func() (bool, error) {
out, err := row()
if err != nil {
return false, err
}
f := strings.Split(out, "|")
if len(f) == 4 && f[1] == "deferred" {
if f[2] != "" || !strings.HasPrefix(f[3], "connect to 10.77.0.11[10.77.0.11]:2526: Connection refused") {
return false, fmt.Errorf("deferred with host %q, reply %q", f[2], f[3])
}
queueID = f[0]
return true, nil
}
return false, fmt.Errorf("row %q\n%s", out, mailLogTail(h, "postfix/"))
}); err != nil {
t.Fatal(err)
}
_, body, err = sc.panel.get("/inbound/log")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(body, `<span class="tag is-warning is-light">deferred</span><h2>Upstream</h2>`) ||
!strings.Contains(body, `<tr><th>Reply</th><td class="sp-mono sp-small">connect to 10.77.0.11[10.77.0.11]:2526: Connection refused</td></tr>`) ||
!strings.Contains(body, `Postfix tries again; the message is in the queue`) {
t.Fatalf("the Inbound log does not show the deferral:\n%.4000s", body)
}
for _, args := range [][]string{{"postsuper", "-e", queueID}, {"postqueue", "-i", queueID}} {
if out, err := h.execIn("selfpost", args...); err != nil {
t.Fatalf("%s: %v\n%s", strings.Join(args, " "), err, out)
}
}
if err := waitFor("the expired row to be bounced", 30*time.Second, 500*time.Millisecond, func() (bool, error) {
out, err := row()
if err != nil {
return false, err
}
f := strings.Split(out, "|")
if len(f) == 4 && f[1] == "bounced" &&
strings.HasPrefix(f[3], "expired in the queue; last reply: connect to 10.77.0.11[10.77.0.11]:2526") {
return true, nil
}
return false, fmt.Errorf("row %q\n%s", out, mailLogTail(h, "postfix/"))
}); err != nil {
t.Fatal(err)
}
}
+18
View File
@@ -399,6 +399,24 @@ func testInboundRelay(t *testing.T, sc *scenario) {
t.Fatalf("the Inbound log holds %d rows, want the two relayed messages and one refusal:\n%s", len(rows), strings.Join(rows, "\n")) t.Fatalf("the Inbound log holds %d rows, want the two relayed messages and one refusal:\n%s", len(rows), strings.Join(rows, "\n"))
} }
// What the upstream answered (inbound-upstream-result): both relayed
// messages are sent, with the host and the sink's reply.
for deadline := time.Now().Add(30 * time.Second); ; time.Sleep(time.Second) {
rows = logRows(`>sent</button>`)
if n := strings.Count(strings.Join(rows, "\n"), `>sent</button>`); n == 2 {
break
} else if time.Now().After(deadline) {
t.Fatalf("%d relayed messages show sent, want 2:\n%s\n%s", n, strings.Join(rows, "\n"), mailLogTail(h, "postfix/"))
}
}
rowWith(rows, "e2e "+token, `<span class="tag is-success is-light">sent</span><h2>Upstream</h2>`,
`<tr><th>Host</th><td class="sp-mono sp-small">10.77.0.11[10.77.0.11]:25</td></tr>`,
`<tr><th>Reply</th><td class="sp-mono sp-small">250 2.0.0 Ok`)
if row := rowWith(rows, "nobody@<wbr>"+inboundDomain); !strings.Contains(row, `<td class="sp-muted">—</td>`) {
t.Fatalf("the refused row has an upstream result:\n%s", row)
}
testInboundUpstreamDown(t, sc, message)
// The filter lists: an entry added in the panel is in the map file the // The filter lists: an entry added in the panel is in the map file the
// filter reads, and one removed is gone from it. Nothing reads the file on // filter reads, and one removed is gone from it. Nothing reads the file on
// this stand (no filter is attached), so the file is the whole effect. // this stand (no filter is attached), so the file is the whole effect.