go-milter: build it patched not to keep every header

The security documentation's last open item of panel-dkim-signing (the
owner's decision, 2026-10-16): go-milter v0.4.1 added every header of a
session to a map of its own, emptied only on ABORT, and handed it to
Headers - which no milter of the panel reads. Postfix bounds one header,
not their number, so a message of nothing but headers up to the message
size limit was held whole by the library in the sender check, the signer
and the journal milter on 465/587 and in the inbound journal milter on port
25, where strangers send; the signer's 1 MiB refusal came after it was all
in.

third_party/go-milter is v0.4.1 from the module cache (BSD 2-Clause,
LICENSE kept) with session.go changed: CodeHeader hands the header to
Header and keeps nothing, CodeEOH hands Headers an empty map. PATCHES.md
says what and why, and how to take a new upstream version. go.mod replaces
the module with it; the Dockerfile copies third_party before "go mod
download". TestTheLibraryKeepsNoHeaders holds it through the library's own
client over two messages of one connection. Comments, security.md (the
accepted risk becomes a note), development.md's library table, the plan and
CHANGELOG follow.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -55,6 +55,11 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version
   it as quoted-printable for a server that does not announce 8BITMIME, after
   it was signed; `disable_mime_output_conversion=yes` now delivers it as
   signed. OpenDKIM's signatures were exposed to this too.
+- **A message of nothing but headers was held whole by every milter.** The
+  milter library (go-milter v0.4.1) kept each header of a message in a map
+  of its own that no milter of the panel read - up to the message size
+  limit, on port 25 from strangers too. The panel builds it patched not to
+  (`third_party/go-milter`).
 - **A rotation of `mail.log` due at the container's start logged an error.**
   It runs before Postfix does, and its `postfix reload` failed ("the Postfix
   mail system is not running"); nothing was lost - the Postfix starting

build/Dockerfile

@@ -13,8 +13,11 @@ WORKDIR /src
 # Version stamped into both binaries; MUST match the image tag.
 ARG VERSION=dev
 
-# Module metadata first for layer caching.
+# Module metadata first for layer caching. third_party/go-milter is a module
+# go.mod replaces go-milter with (third_party/go-milter/PATCHES.md): the
+# download needs it.
 COPY go.mod go.sum ./
+COPY third_party ./third_party
 RUN go mod download
 
 COPY cmd ./cmd

docs/development.md

@@ -182,7 +182,7 @@ dependencies must be permissive or GPL-family (see
 
 | Package | Version | Repository | License |
 |---|---|---|---|
-| `github.com/emersion/go-milter` | v0.4.1 | <https://github.com/emersion/go-milter> | BSD-2-Clause |
+| `github.com/emersion/go-milter` | v0.4.1, patched in `third_party/go-milter` (its `PATCHES.md`; `replace` in `go.mod`) | <https://github.com/emersion/go-milter> | BSD-2-Clause |
 | `github.com/emersion/go-msgauth` | v0.7.0 (`dkim`, the signer) | <https://github.com/emersion/go-msgauth> | MIT |
 | `golang.org/x/crypto` | v0.58.0 | <https://github.com/golang/crypto> | BSD-3-Clause |
 | `modernc.org/sqlite` | v1.60.1 | <https://gitlab.com/cznic/sqlite> (mirror: <https://github.com/modernc-org/sqlite>) | BSD-3-Clause |

docs/plans/panel-dkim-signing.md

@@ -210,11 +210,10 @@ libmilter1.0.1.
 the reviews ran on stages 1-2 and their findings are fixed in `c2f63d2`,
 with the first rollout's incident (the queue's owner numbers); the
 documents in the commit after it. `Sender:` is held to the login like
-`From:` (security review F1; the owner's decision, 2026-10-16). Open:
+`From:` (security review F1; the owner's decision, 2026-10-16). go-milter
-go-milter v0.4.1 keeping
+v0.4.1 kept every header in a map of its own per session, on 465/587 and on
-every header in a map of its own per session (security.md, accepted risks)
+port 25 alike: the panel builds it patched not to
-- a header section the signer refuses past 1 MiB is still held by the
+(`third_party/go-milter/PATCHES.md`; the owner's decision, 2026-10-16).
-library, on 465/587 and on port 25 alike, as before 2.8.0.
 - [ ] On the test server: a day of the owner's mail; one message from an
   application through 465 and one notification to a real mailbox, the
   headers read (dkim=pass with the panel's `d=`).

docs/security.md

@@ -729,34 +729,21 @@ deferred item from the roadmap.
   key of another's is involved, as with a display name. Holding it too is a
   few lines in `internal/milter/sender.go` (`judge()`); the trigger to do it
   is a mail program that shows it in place of `From:`.
-- **A header section is bounded at 1 MiB by the signer, not by the milter
+- **A header section costs the signer up to 1 MiB.** The signer counts a
-  library.** The signer counts a message's headers as Postfix hands them
+  message's headers as Postfix hands them over and refuses a section of
-  over and refuses a section of more than 1 MiB with `552 5.3.4`
+  more than 1 MiB with `552 5.3.4` (`internal/milter/sign.go`,
-  (`internal/milter/sign.go`, `headerLimit`; `2.8.0`): it keeps no more
+  `headerLimit`; `2.8.0`): it keeps no more than that and drops its list
-  than that and drops its list once the block it signs is built, so its own
+  once the block it signs is built. No mail program writes such a section,
-  memory per message is a few mebibytes at most. No mail program writes such
+  and the refusal is permanent, so a sender is told rather than retried.
-  a section, and the refusal is permanent, so a sender is told rather than
+  The other milters keep a few headers each (From, Sender, Subject).
-  retried. What the cap does not bound: go-milter v0.4.1 adds every header
+  Until `2.8.0` the milter library itself added every header of a message
-  of a message to a map of its own in each milter it is sent to (session.go,
+  to a map of its own in each milter - up to the port's message size limit,
-  `CodeHeader`), which none of SelfPost's milters reads - the sender check,
+  from strangers on port 25 - which none of the milters read; the panel
-  the signer and the journal milter on 465/587, the inbound journal milter
+  builds go-milter v0.4.1 patched not to keep it
-  on port 25 - and that copy is bounded only by the port's message size
+  (`third_party/go-milter/PATCHES.md`; the owner's decision, 2026-10-16).
-  limit (Postfix's default 10 240 000 bytes on 465/587;
+  `deploy/docker-compose.yml` sets no memory limit: what is left - many
-  `INBOUND_MESSAGE_SIZE_LIMIT` or `DMARC_MESSAGE_SIZE_LIMIT` on port 25,
+  messages at once, each held to the bounds above - is the host's to
-  from strangers), since a refusal at the end of the headers comes after
+  size.
-  they are all in. A message of nothing but headers still costs the panel
-  about its size once per milter while it is in the milters, times the
-  messages Postfix runs at once (as many as its cleanup processes,
-  `default_process_limit`, 100), and `deploy/docker-compose.yml` sets no
-  memory limit, so the host's out-of-memory killer chooses what goes; a
-  panel killed is started again, and submissions wait meanwhile. The
-  library's copy is older than `2.8.0`: every milter of the panel has had
-  it. The fix, priced: a go-milter patched not to keep the map (vendored;
-  SelfPost reads headers only through `Header`), or a memory limit in the
-  compose file, which makes an exhausted panel a restart of the panel alone
-  rather than a victim the host picks. Trigger to revisit: the inbound relay
-  on an address strangers send to in volume, or a panel seen near the
-  host's memory.
 - **A message's page can show another message's log lines after a queue-id
   collision.** Postfix's short queue ids repeat: the id is the microsecond of
   arrival and the queue file's inode, and inodes come round within seconds.

go.mod

@@ -21,3 +21,6 @@ require (
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.12.1 // indirect
 )
+
+// go-milter with its header map dropped: third_party/go-milter/PATCHES.md.
+replace github.com/emersion/go-milter => ./third_party/go-milter

go.sum

@@ -2,8 +2,6 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/emersion/go-message v0.18.1 h1:tfTxIoXFSFRwWaZsgnqS1DSZuGpYGzSmCZD8SK3QA2E=
 github.com/emersion/go-message v0.18.1/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA=
-github.com/emersion/go-milter v0.4.1 h1:gLs9QD0zEHF8omgEw8M+aGz6iwBNpWLAcwgSur0ra4M=
-github.com/emersion/go-milter v0.4.1/go.mod h1:erCQVl0mH4SX9jEvwe+wyndit0rQtmvMLH86V6NGtkI=
 github.com/emersion/go-msgauth v0.7.0 h1:vj2hMn6KhFtW41kshIBTXvp6KgYSqpA/ZN9Pv4g1INc=
 github.com/emersion/go-msgauth v0.7.0/go.mod h1:mmS9I6HkSovrNgq0HNXTeu8l3sRAAuQ9RMvbM4KU7Ck=
 github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo=

internal/milter/library_test.go

@@ -0,0 +1,87 @@
+package milter
+
+import (
+	"context"
+	"net"
+	"net/textproto"
+	"sync"
+	"testing"
+
+	"github.com/emersion/go-milter"
+)
+
+// headerRecorder is a backend that notes what the library hands it.
+type headerRecorder struct {
+	milter.NoOpMilter
+	mu       *sync.Mutex
+	headers  *[]string
+	mapSizes *[]int
+}
+
+func (r headerRecorder) Header(name, value string, m *milter.Modifier) (milter.Response, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	*r.headers = append(*r.headers, name)
+	return milter.RespContinue, nil
+}
+
+func (r headerRecorder) Headers(h textproto.MIMEHeader, m *milter.Modifier) (milter.Response, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	*r.mapSizes = append(*r.mapSizes, len(h))
+	return milter.RespContinue, nil
+}
+
+// The go-milter the panel builds (third_party/go-milter/PATCHES.md) hands
+// every header to Header and keeps none: the map Headers gets is empty, on
+// the first message of a connection and on the next - upstream kept every
+// header of the connection until an ABORT.
+func TestTheLibraryKeepsNoHeaders(t *testing.T) {
+	var mu sync.Mutex
+	var headers []string
+	var mapSizes []int
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := &milter.Server{
+		NewMilter: func() milter.Milter { return headerRecorder{mu: &mu, headers: &headers, mapSizes: &mapSizes} },
+		Protocol:  milter.OptNoBody,
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() { <-ctx.Done(); _ = srv.Close() }()
+	go func() { _ = srv.Serve(ln) }()
+
+	cl := milter.NewClientWithOptions("tcp", ln.Addr().String(), milter.ClientOptions{})
+	defer cl.Close()
+	s, err := cl.Session()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	for i := 0; i < 2; i++ {
+		if _, err := s.Mail("<a@example.org>", nil); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := s.Rcpt("<b@example.net>", nil); err != nil {
+			t.Fatal(err)
+		}
+		for _, name := range []string{"From", "To", "X-Filler"} {
+			if _, err := s.HeaderField(name, " value"); err != nil {
+				t.Fatal(err)
+			}
+		}
+		if _, err := s.HeaderEnd(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(headers) != 6 {
+		t.Errorf("Header saw %v, want the three headers of each of two messages", headers)
+	}
+	if len(mapSizes) != 2 || mapSizes[0] != 0 || mapSizes[1] != 0 {
+		t.Errorf("Headers got maps of %v headers, want two empty ones", mapSizes)
+	}
+}

internal/milter/sender.go

@@ -331,10 +331,11 @@ func (s *senderSession) Header(name, value string, m *milter.Modifier) (milter.R
 // message, and src/smtpd/smtpd.c, common_post_message_handling(), says that
 // reply to the client.
 //
-// h is not read: go-milter v0.4.1 adds every header of the connection to it
+// h is not read, and is empty: upstream go-milter v0.4.1 added every header
-// and empties it only on ABORT (session.go, CodeHeader, CodeAbort), so on a
+// of the connection to it and emptied it only on ABORT, so on a second
-// second message it can still hold the first one's From:. The session keeps
+// message it could still hold the first one's From: (audit 2026-10-12, L9);
-// its own, emptied at each MAIL FROM (audit 2026-10-12, L9).
+// the copy SelfPost builds keeps none (third_party/go-milter/PATCHES.md). The
+// session keeps its own, emptied at each MAIL FROM.
 func (s *senderSession) Headers(h textproto.MIMEHeader, m *milter.Modifier) (milter.Response, error) {
 	field, shown, why, err := s.judge()
 	if err != nil {

internal/milter/sign.go

@@ -276,8 +276,7 @@ func (s *signSession) Header(name, value string, m *milter.Modifier) (milter.Res
 // What cannot be signed is deferred: the socket is attached with
 // default_action=tempfail, and so is every refusal here.
 //
-// h is not read: go-milter v0.4.1 keeps every header of the connection in it
+// h is not read, and is empty (see senderSession.Headers).
-// (see senderSession.Headers).
 func (s *signSession) Headers(h textproto.MIMEHeader, m *milter.Modifier) (milter.Response, error) {
 	if s.tooBig {
 		log.Printf("dkim-signer: login %q: a header section of more than %d bytes - refused", s.login, headerLimit)

third_party/go-milter/LICENSE

@@ -0,0 +1,25 @@
+BSD 2-Clause License
+
+Copyright (c) 2017 Bozhin Zafirov
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

third_party/go-milter/PATCHES.md

@@ -0,0 +1,33 @@
+# go-milter v0.4.1, as SelfPost builds it
+
+This is github.com/emersion/go-milter at v0.4.1 (BSD 2-Clause, `LICENSE`),
+copied from the Go module cache, with one change. The main module takes it
+through a `replace` in its `go.mod`; the module path stays the upstream one.
+Left out of the copy: `cmd/` (a command-line tool SelfPost does not build) and
+the protocol notes (`milter-protocol*.txt`).
+
+## The change: headers are not kept
+
+`session.go`, `Process()`, cases `CodeHeader` and `CodeEOH`. Upstream adds
+every header of a session to a map (`m.headers`) that it empties only on
+ABORT, and hands that map to the backend's `Headers` at the end of the
+headers. Postfix bounds one header (60000 bytes with milters) but not their
+number, so a message of nothing but headers, up to `message_size_limit`,
+was held whole by the library in every milter that receives headers - the
+sender check, the DKIM signer and the journal milter on 465/587, the inbound
+journal milter on port 25 - on top of what the milters keep themselves
+(security review of panel-dkim-signing, 2026-10-16).
+
+Patched, `CodeHeader` hands each header to the backend's `Header` and keeps
+nothing; `CodeEOH` hands `Headers` an empty map. No SelfPost milter reads
+that map: the sender check and the signer keep the headers they need
+themselves (internal/milter, `senderSession`, `signSession`), and the journal
+milters do not implement `Headers`.
+
+The client side (`client.go`, used by the unit tests) is unchanged.
+
+## Taking a new upstream version
+
+Copy the files again, apply the change above, and keep this note. Read the
+new `session.go` first: if upstream bounds or drops the map itself, the
+patch - and this copy - may go.

third_party/go-milter/README.md

@@ -0,0 +1,10 @@
+# go-milter
+
+[![GoDoc](https://godoc.org/github.com/emersion/go-milter?status.svg)](https://godoc.org/github.com/emersion/go-milter)
+[![builds.sr.ht status](https://builds.sr.ht/~emersion/go-milter/commits.svg)](https://builds.sr.ht/~emersion/go-milter/commits?)
+
+A Go library to write mail filters.
+
+## License
+
+BSD 2-Clause

third_party/go-milter/client.go

@@ -0,0 +1,665 @@
+package milter
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"time"
+
+	"github.com/emersion/go-message/textproto"
+)
+
+var ErrUnsupportedMilterVersion = fmt.Errorf("milter: negotiate: unsupported milter version")
+
+// Client is a wrapper for managing milter connections.
+//
+// Currently, it just creates new connections using provided Dialer.
+type Client struct {
+	opts    ClientOptions
+	network string
+	address string
+}
+
+type Dialer interface {
+	Dial(network string, addr string) (net.Conn, error)
+}
+
+type ClientOptions struct {
+	Dialer       Dialer
+	ReadTimeout  time.Duration
+	WriteTimeout time.Duration
+	ActionMask   OptAction
+	ProtocolMask OptProtocol
+}
+
+var defaultOptions = ClientOptions{
+	Dialer: &net.Dialer{
+		Timeout: 10 * time.Second,
+	},
+	ReadTimeout:  10 * time.Second,
+	WriteTimeout: 10 * time.Second,
+	ActionMask:   OptAddHeader | OptAddRcpt | OptChangeBody | OptChangeFrom | OptChangeHeader,
+	ProtocolMask: 0,
+}
+
+// NewDefaultClient creates a new Client object using default options.
+//
+// It uses 10 seconds for connection/read/write timeouts and allows milter to
+// send any actions supported by library.
+func NewDefaultClient(network, address string) *Client {
+	return NewClientWithOptions(network, address, defaultOptions)
+}
+
+// NewClientWithOptions creates a new Client object using provided options.
+//
+// You generally want to use options to restrict ActionMask to what your code
+// supports and ProtocolMask to what you intend to submit.
+//
+// If opts.Dialer is not set, empty net.Dialer object will be used.
+func NewClientWithOptions(network, address string, opts ClientOptions) *Client {
+	if opts.Dialer == nil {
+		opts.Dialer = &net.Dialer{}
+	}
+
+	return &Client{
+		opts:    opts,
+		network: network,
+		address: address,
+	}
+}
+
+func (c *Client) Session() (*ClientSession, error) {
+	s := &ClientSession{
+		readTimeout:           c.opts.ReadTimeout,
+		writeTimeout:          c.opts.WriteTimeout,
+		clientProtocolVersion: 6,
+	}
+
+	// TODO(foxcpp): Connection pooling.
+
+	conn, err := c.opts.Dialer.Dial(c.network, c.address)
+	if err != nil {
+		return nil, fmt.Errorf("milter: session create: %w", err)
+	}
+
+	s.conn = conn
+	if err := s.negotiate(c.opts.ActionMask, c.opts.ProtocolMask); err != nil {
+		return nil, err
+	}
+
+	return s, nil
+}
+
+func (c *Client) Close() error {
+	// Reserved for use in connection pooling.
+	return nil
+}
+
+type ClientSession struct {
+	conn net.Conn
+
+	// Bitmask of negotiated action options.
+	ActionOpts OptAction
+
+	// Bitmask of negotiated protocol options.
+	ProtocolOpts OptProtocol
+
+	needAbort bool
+
+	readTimeout  time.Duration
+	writeTimeout time.Duration
+	// Milter client version. Can be downgraded during negotiation
+	clientProtocolVersion uint32
+}
+
+// negotiate exchanges OPTNEG messages with the milter and sets s.mask to the
+// negotiated value.
+func (s *ClientSession) negotiate(actionMask OptAction, protoMask OptProtocol) error {
+	// Send our mask, get mask from milter..
+	msg := &Message{
+		Code: byte(CodeOptNeg), // TODO(foxcpp): Get rid of casts by changing msg.Code to have Code type
+		Data: make([]byte, 4*3),
+	}
+	binary.BigEndian.PutUint32(msg.Data, s.clientProtocolVersion)
+	binary.BigEndian.PutUint32(msg.Data[4:], uint32(actionMask))
+	binary.BigEndian.PutUint32(msg.Data[8:], uint32(protoMask))
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return fmt.Errorf("milter: negotiate: optneg write: %w", err)
+	}
+	msg, err := readPacket(s.conn, s.readTimeout)
+	if err != nil {
+		return fmt.Errorf("milter: negotiate: optneg read: %w", err)
+	}
+	if Code(msg.Code) != CodeOptNeg {
+		return fmt.Errorf("milter: negotiate: unexpected code: %v", rune(msg.Code))
+	}
+	if len(msg.Data) < 4*3 /* version + action mask + proto mask */ {
+		return fmt.Errorf("milter: negotiate: unexpected data size: %v", len(msg.Data))
+	}
+
+	milterVersion := binary.BigEndian.Uint32(msg.Data[:4])
+	milterActionMask := binary.BigEndian.Uint32(msg.Data[4:])
+	s.ActionOpts = OptAction(milterActionMask)
+	milterProtoMask := binary.BigEndian.Uint32(msg.Data[8:])
+	s.ProtocolOpts = OptProtocol(milterProtoMask)
+
+	// If milter advertises lower protocol version than we support, try to downgrade.
+	if milterVersion < s.clientProtocolVersion {
+		// Only downgrade if both sides support the same actions and protocols.
+		// The lowest supported milterVersion is 2.
+		if milterVersion >= 2 && actionMask&0x3f == actionMask && protoMask&0x7f == protoMask {
+			s.clientProtocolVersion = milterVersion
+		} else {
+			return ErrUnsupportedMilterVersion
+		}
+	}
+
+	s.needAbort = true
+
+	return nil
+}
+
+// ProtocolOption checks whether the option is set in negotiated options, that
+// is, requested by both sides.
+func (s *ClientSession) ProtocolOption(opt OptProtocol) bool {
+	return s.ProtocolOpts&opt != 0
+}
+
+// ActionOption checks whether the option is set in negotiated options, that
+// is, requested by both sides.
+func (s *ClientSession) ActionOption(opt OptAction) bool {
+	return s.ActionOpts&opt != 0
+}
+
+func (s *ClientSession) Macros(code Code, kv ...string) error {
+	// Note: kv is ...string with the expectation that the list of macro names
+	// will be static and not dynamically constructed.
+
+	msg := &Message{
+		Code: byte(CodeMacro),
+		Data: []byte{byte(code)},
+	}
+	for _, str := range kv {
+		msg.Data = appendCString(msg.Data, str)
+	}
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return fmt.Errorf("milter: macros: %w", err)
+	}
+
+	return nil
+}
+
+func appendUint16(dest []byte, val uint16) []byte {
+	dest = append(dest, 0x00, 0x00)
+	binary.BigEndian.PutUint16(dest[len(dest)-2:], val)
+	return dest
+}
+
+type Action struct {
+	Code ActionCode
+
+	// SMTP code if Code == ActReplyCode.
+	SMTPCode int
+	// Reply text if Code == ActReplyCode.
+	SMTPText string
+}
+
+func (s *ClientSession) readAction() (*Action, error) {
+	for {
+		msg, err := readPacket(s.conn, s.readTimeout)
+		if err != nil {
+			return nil, fmt.Errorf("action read: %w", err)
+		}
+		if msg.Code == 'p' /* progress */ {
+			continue
+		}
+		if ActionCode(msg.Code) != ActContinue {
+			s.needAbort = false
+		}
+
+		return parseAction(msg)
+	}
+}
+
+func parseAction(msg *Message) (*Action, error) {
+	act := &Action{
+		Code: ActionCode(msg.Code),
+	}
+	var err error
+
+	switch ActionCode(msg.Code) {
+	case ActAccept, ActContinue, ActDiscard, ActReject, ActTempFail:
+	case ActReplyCode:
+		if len(msg.Data) <= 4 {
+			return nil, fmt.Errorf("action read: unexpected data length: %v", len(msg.Data))
+		}
+		act.SMTPCode, err = strconv.Atoi(string(msg.Data[:3]))
+		if err != nil {
+			return nil, fmt.Errorf("action read: malformed SMTP code: %v", msg.Data[:3])
+		}
+		// There is 0x20 (' ') in between.
+		act.SMTPText = readCString(msg.Data[4:])
+	default:
+		return nil, fmt.Errorf("action read: unexpected code: %v", msg.Code)
+	}
+
+	return act, nil
+}
+
+// Conn sends the connection information to the milter.
+//
+// It should be called once per milter session (from Session to Close).
+func (s *ClientSession) Conn(hostname string, family ProtoFamily, port uint16, addr string) (*Action, error) {
+	if s.ProtocolOpts&OptNoConnect != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	msg := &Message{
+		Code: byte(CodeConn),
+	}
+	msg.Data = appendCString(msg.Data, hostname)
+	msg.Data = append(msg.Data, byte(family))
+	if family != FamilyUnknown {
+		if family == FamilyInet || family == FamilyInet6 {
+			msg.Data = appendUint16(msg.Data, port)
+		}
+		msg.Data = appendCString(msg.Data, addr)
+	}
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: conn: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoConnReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: conn: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+// Helo sends the HELO hostname to the milter.
+//
+// It should be called once per milter session (from Session to Close).
+func (s *ClientSession) Helo(helo string) (*Action, error) {
+	// Synthesise response as if server replied "go on" while in fact it does
+	// not support that message.
+	if s.ProtocolOpts&OptNoHelo != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	msg := &Message{
+		Code: byte(CodeHelo),
+		Data: appendCString(nil, helo),
+	}
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: helo: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoHeloReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: helo: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+func (s *ClientSession) Mail(sender string, esmtpArgs []string) (*Action, error) {
+	if s.ProtocolOpts&OptNoMailFrom != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	msg := &Message{
+		Code: byte(CodeMail),
+	}
+
+	msg.Data = appendCString(msg.Data, "<"+sender+">")
+	for _, arg := range esmtpArgs {
+		msg.Data = appendCString(msg.Data, arg)
+	}
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: mail: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoMailReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: mail: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+func (s *ClientSession) Rcpt(rcpt string, esmtpArgs []string) (*Action, error) {
+	if s.ProtocolOpts&OptNoRcptTo != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	msg := &Message{
+		Code: byte(CodeRcpt),
+	}
+
+	msg.Data = appendCString(msg.Data, "<"+rcpt+">")
+	for _, arg := range esmtpArgs {
+		msg.Data = appendCString(msg.Data, arg)
+	}
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: rcpt: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoRcptReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: rcpt: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+// HeaderField sends a single header field to the milter.
+//
+// Value should be the original field value without any unfolding applied.
+//
+// HeaderEnd() must be called after the last field.
+func (s *ClientSession) HeaderField(key, value string) (*Action, error) {
+	if s.ProtocolOpts&OptNoHeaders != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	msg := &Message{
+		Code: byte(CodeHeader),
+	}
+	msg.Data = appendCString(msg.Data, key)
+	msg.Data = appendCString(msg.Data, value)
+
+	if err := writePacket(s.conn, msg, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: header field: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoHeaderReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: header field: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+// HeaderEnd send the EOH (End-Of-Header) message to the milter.
+//
+// No HeaderField calls are allowed after this point.
+func (s *ClientSession) HeaderEnd() (*Action, error) {
+	if s.ProtocolOpts&OptNoEOH != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	if err := writePacket(s.conn, &Message{
+		Code: byte(CodeEOH),
+	}, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: header end: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoEOHReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: header end: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+// Header sends each field from textproto.Header followed by EOH unless
+// header messages are disabled during negotiation.
+func (s *ClientSession) Header(hdr textproto.Header) (*Action, error) {
+	for f := hdr.Fields(); f.Next(); {
+		act, err := s.HeaderField(f.Key(), f.Value())
+		if err != nil {
+			return nil, err
+		}
+
+		if act.Code != ActContinue {
+			return act, nil
+		}
+	}
+
+	return s.HeaderEnd()
+}
+
+// BodyChunk sends a single body chunk to the milter.
+//
+// It is callers responsibility to ensure every chunk is not bigger than
+// MaxBodyChunk.
+//
+// If OptSkip was specified during negotiation, caller should be ready to
+// handle return ActSkip and stop sending body chunks if it is returned.
+func (s *ClientSession) BodyChunk(chunk []byte) (*Action, error) {
+	if s.ProtocolOpts&OptNoBody != 0 {
+		return &Action{Code: ActContinue}, nil
+	}
+
+	// Callers tend to be irresponsible... /s
+	if len(chunk) > MaxBodyChunk {
+		return nil, fmt.Errorf("milter: body chunk: too big body chunk: %v", len(chunk))
+	}
+
+	if err := writePacket(s.conn, &Message{
+		Code: byte(CodeBody),
+		Data: chunk,
+	}, s.writeTimeout); err != nil {
+		return nil, fmt.Errorf("milter: body chunk: %w", err)
+	}
+
+	if !s.ProtocolOption(OptNoBodyReply) {
+		act, err := s.readAction()
+		if err != nil {
+			return nil, fmt.Errorf("milter: body chunk: %w", err)
+		}
+		return act, nil
+	}
+	return &Action{Code: ActContinue}, nil
+}
+
+// BodyReadFrom is a helper function that calls BodyChunk repeately to transmit entire
+// body from io.Reader and then calls End.
+//
+// See documentation for these functions for details.
+func (s *ClientSession) BodyReadFrom(r io.Reader) ([]ModifyAction, *Action, error) {
+	// It is problematic to use io.WriteCloser since we may need to report
+	// action after each write.
+
+	buf := make([]byte, MaxBodyChunk)
+	for {
+		n, err := r.Read(buf)
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, nil, err
+		}
+		if n == 0 {
+			break
+		}
+
+		act, err := s.BodyChunk(buf[:n])
+		if err != nil {
+			return nil, nil, err
+		}
+		if act.Code == ActSkip {
+			break
+		}
+		if act.Code != ActContinue {
+			return nil, act, nil
+		}
+	}
+
+	return s.End()
+}
+
+type ModifyAction struct {
+	Code ModifyActCode
+
+	// Recipient to add/remove if Code == ActAddRcpt or ActDelRcpt.
+	Rcpt string
+
+	// New envelope sender if Code = ActChangeFrom.
+	From string
+
+	// ESMTP arguments for envelope sender if Code = ActChangeFrom.
+	FromArgs []string
+
+	// Portion of body to be replaced if Code == ActReplBody.
+	Body []byte
+
+	// Index of the header field to be changed if Code = ActChangeHeader or Code = ActInsertHeader.
+	// Index is 1-based and is per value of HdrName.
+	// E.g. HeaderIndex = 3 and HdrName = "DKIM-Signature" mean "change third
+	// DKIM-Signature field". Order is the same as of HeaderField calls.
+	HeaderIndex uint32
+
+	// Header field name to be added/changed if Code == ActAddHeader or
+	// ActChangeHeader or ActInsertHeader.
+	HeaderName string
+
+	// Header field value to be added/changed if Code == ActAddHeader or
+	// ActChangeHeader or ActInsertHeader. If set to empty string - the field
+	// should be removed.
+	HeaderValue string
+
+	// Quarantine reason if Code == ActQuarantine.
+	Reason string
+}
+
+func parseModifyAct(msg *Message) (*ModifyAction, error) {
+	act := &ModifyAction{
+		Code: ModifyActCode(msg.Code),
+	}
+
+	switch ModifyActCode(msg.Code) {
+	case ActAddRcpt, ActDelRcpt:
+		act.Rcpt = readCString(msg.Data)
+	case ActQuarantine:
+		act.Reason = readCString(msg.Data)
+	case ActReplBody:
+		act.Body = msg.Data
+	case ActChangeFrom:
+		argv := bytes.Split(msg.Data, []byte{0x00})
+		act.From = string(argv[0])
+		for _, arg := range argv[1:] {
+			act.FromArgs = append(act.FromArgs, string(arg))
+		}
+	case ActChangeHeader, ActInsertHeader:
+		if len(msg.Data) < 4 {
+			return nil, fmt.Errorf("read modify action: missing header index")
+		}
+		act.HeaderIndex = binary.BigEndian.Uint32(msg.Data)
+
+		msg.Data = msg.Data[4:]
+		fallthrough
+	case ActAddHeader:
+		// TODO: Change readCString to return last index.
+		act.HeaderName = readCString(msg.Data)
+		nul := bytes.IndexByte(msg.Data, 0x00)
+		if nul == -1 {
+			return nil, fmt.Errorf("read modify action: missing NUL delimiter")
+		}
+		if nul == len(msg.Data) {
+			return nil, fmt.Errorf("read modify action: missing header value")
+		}
+		act.HeaderValue = readCString(msg.Data[nul+1:])
+	default:
+		return nil, fmt.Errorf("read modify action: unexpected message code: %v", msg.Code)
+	}
+
+	return act, nil
+}
+
+func (s *ClientSession) readModifyActs() (modifyActs []ModifyAction, act *Action, err error) {
+	for {
+		msg, err := readPacket(s.conn, s.readTimeout)
+		if err != nil {
+			return nil, nil, fmt.Errorf("action read: %w", err)
+		}
+		if msg.Code == 'p' /* progress */ {
+			continue
+		}
+
+		switch ModifyActCode(msg.Code) {
+		case ActAddRcpt, ActDelRcpt, ActReplBody, ActChangeHeader, ActInsertHeader,
+			ActAddHeader, ActChangeFrom, ActQuarantine:
+			modifyAct, err := parseModifyAct(msg)
+			if err != nil {
+				return nil, nil, err
+			}
+			modifyActs = append(modifyActs, *modifyAct)
+		default:
+			act, err = parseAction(msg)
+			if err != nil {
+				return nil, nil, err
+			}
+
+			return modifyActs, act, nil
+		}
+	}
+}
+
+// End sends the EOB message and resets session back to the state before Mail
+// call. The same ClientSession can be used to check another message arrived
+// within the same SMTP connection (Helo and Conn information is preserved).
+//
+// Close should be called to conclude session.
+func (s *ClientSession) End() ([]ModifyAction, *Action, error) {
+	if err := writePacket(s.conn, &Message{
+		Code: byte(CodeEOB),
+	}, s.writeTimeout); err != nil {
+		return nil, nil, fmt.Errorf("milter: end: %w", err)
+	}
+
+	modifyActs, act, err := s.readModifyActs()
+	if err != nil {
+		return nil, nil, fmt.Errorf("milter: end: %w", err)
+	}
+
+	return modifyActs, act, nil
+}
+
+// Abort sends Abort to the milter.
+//
+// This is called for an unexpected end to an email outside the milters
+// control.
+func (s *ClientSession) Abort() error {
+	return writePacket(s.conn, &Message{
+		Code: byte(CodeAbort),
+	}, s.writeTimeout)
+}
+
+// Close releases resources associated with the session.
+//
+// If there a milter sequence in progress - it is aborted.
+func (s *ClientSession) Close() error {
+	if s.needAbort {
+		_ = s.Abort()
+	}
+
+	if err := writePacket(s.conn, &Message{
+		Code: byte(CodeQuit),
+	}, s.writeTimeout); err != nil {
+		return fmt.Errorf("milter: close: %w", err)
+	}
+	return s.conn.Close()
+}

third_party/go-milter/client_test.go

@@ -0,0 +1,412 @@
+package milter
+
+import (
+	"bytes"
+	"net"
+	nettextproto "net/textproto"
+	"reflect"
+	"testing"
+
+	"github.com/emersion/go-message/textproto"
+)
+
+type MockMilter struct {
+	ConnResp Response
+	ConnMod  func(m *Modifier)
+	ConnErr  error
+
+	HeloResp Response
+	HeloMod  func(m *Modifier)
+	HeloErr  error
+
+	MailResp Response
+	MailMod  func(m *Modifier)
+	MailErr  error
+
+	RcptResp Response
+	RcptMod  func(m *Modifier)
+	RcptErr  error
+
+	HdrResp Response
+	HdrMod  func(m *Modifier)
+	HdrErr  error
+
+	HdrsResp Response
+	HdrsMod  func(m *Modifier)
+	HdrsErr  error
+
+	BodyChunkResp Response
+	BodyChunkMod  func(m *Modifier)
+	BodyChunkErr  error
+
+	BodyResp Response
+	BodyMod  func(m *Modifier)
+	BodyErr  error
+
+	AbortMod func(m *Modifier)
+	AbortErr error
+
+	// Info collected during calls.
+	Host   string
+	Family string
+	Port   uint16
+	Addr   net.IP
+
+	HeloValue string
+	From      string
+	Rcpt      []string
+	Hdr       nettextproto.MIMEHeader
+
+	Chunks [][]byte
+}
+
+func (mm *MockMilter) Connect(host string, family string, port uint16, addr net.IP, m *Modifier) (Response, error) {
+	if mm.ConnMod != nil {
+		mm.ConnMod(m)
+	}
+	mm.Host = host
+	mm.Family = family
+	mm.Port = port
+	mm.Addr = addr
+	return mm.ConnResp, mm.ConnErr
+}
+
+func (mm *MockMilter) Helo(name string, m *Modifier) (Response, error) {
+	if mm.HeloMod != nil {
+		mm.HeloMod(m)
+	}
+	mm.HeloValue = name
+	return mm.HeloResp, mm.HeloErr
+}
+
+func (mm *MockMilter) MailFrom(from string, m *Modifier) (Response, error) {
+	if mm.MailMod != nil {
+		mm.MailMod(m)
+	}
+	mm.From = from
+	return mm.MailResp, mm.MailErr
+}
+
+func (mm *MockMilter) RcptTo(rcptTo string, m *Modifier) (Response, error) {
+	if mm.RcptMod != nil {
+		mm.RcptMod(m)
+	}
+	mm.Rcpt = append(mm.Rcpt, rcptTo)
+	return mm.RcptResp, mm.RcptErr
+}
+
+func (mm *MockMilter) Header(name string, value string, m *Modifier) (Response, error) {
+	if mm.HdrMod != nil {
+		mm.HdrMod(m)
+	}
+	return mm.HdrResp, mm.HdrErr
+}
+
+func (mm *MockMilter) Headers(h nettextproto.MIMEHeader, m *Modifier) (Response, error) {
+	if mm.HdrsMod != nil {
+		mm.HdrsMod(m)
+	}
+	mm.Hdr = h
+	return mm.HdrsResp, mm.HdrsErr
+}
+
+func (mm *MockMilter) BodyChunk(chunk []byte, m *Modifier) (Response, error) {
+	if mm.BodyChunkMod != nil {
+		mm.BodyChunkMod(m)
+	}
+	mm.Chunks = append(mm.Chunks, chunk)
+	return mm.BodyChunkResp, mm.BodyChunkErr
+}
+
+func (mm *MockMilter) Body(m *Modifier) (Response, error) {
+	if mm.BodyMod != nil {
+		mm.BodyMod(m)
+	}
+	return mm.BodyResp, mm.BodyErr
+}
+
+func (mm *MockMilter) Abort(m *Modifier) error {
+	if mm.AbortMod != nil {
+		mm.AbortMod(m)
+	}
+	return mm.AbortErr
+}
+
+func TestMilterClient_UsualFlow(t *testing.T) {
+	mm := MockMilter{
+		ConnResp:      RespContinue,
+		HeloResp:      RespContinue,
+		MailResp:      RespContinue,
+		RcptResp:      RespContinue,
+		HdrResp:       RespContinue,
+		HdrsResp:      RespContinue,
+		BodyChunkResp: RespContinue,
+		BodyResp:      RespContinue,
+		BodyMod: func(m *Modifier) {
+			m.AddHeader("X-Bad", "very")
+			m.ChangeHeader(1, "Subject", "***SPAM***")
+			m.Quarantine("very bad message")
+		},
+	}
+	s := Server{
+		NewMilter: func() Milter {
+			return &mm
+		},
+		Actions: OptAddHeader | OptChangeHeader,
+	}
+	defer s.Close()
+	local, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Serve(local)
+
+	cl := NewClientWithOptions("tcp", local.Addr().String(), ClientOptions{
+		ActionMask: OptAddHeader | OptChangeHeader | OptQuarantine,
+	})
+	defer cl.Close()
+	session, err := cl.Session()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer session.Close()
+
+	assertAction := func(act *Action, err error, expectCode ActionCode) {
+		t.Helper()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if act.Code != expectCode {
+			t.Fatal("Unexpectedcode:", act.Code)
+		}
+	}
+
+	act, err := session.Conn("host", FamilyInet, 25565, "172.0.0.1")
+	assertAction(act, err, ActContinue)
+	if mm.Host != "host" {
+		t.Fatal("Wrong host:", mm.Host)
+	}
+	if mm.Family != "tcp4" {
+		t.Fatal("Wrong family:", mm.Family)
+	}
+	if mm.Port != 25565 {
+		t.Fatal("Wrong port:", mm.Port)
+	}
+	if mm.Addr.String() != "172.0.0.1" {
+		t.Fatal("Wrong IP:", mm.Addr)
+	}
+
+	if err := session.Macros(CodeHelo, "tls_version", "very old"); err != nil {
+		t.Fatal("Unexpected error", err)
+	}
+
+	act, err = session.Helo("helo_host")
+	assertAction(act, err, ActContinue)
+	if mm.HeloValue != "helo_host" {
+		t.Fatal("Wrong helo value:", mm.HeloValue)
+	}
+
+	act, err = session.Mail("from@example.org", []string{"A=B"})
+	assertAction(act, err, ActContinue)
+	if mm.From != "from@example.org" {
+		t.Fatal("Wrong MAIL FROM:", mm.From)
+	}
+
+	act, err = session.Rcpt("to1@example.org", []string{"A=B"})
+	assertAction(act, err, ActContinue)
+	act, err = session.Rcpt("to2@example.org", []string{"A=B"})
+	assertAction(act, err, ActContinue)
+	if !reflect.DeepEqual(mm.Rcpt, []string{"to1@example.org", "to2@example.org"}) {
+		t.Fatal("Wrong recipients:", mm.Rcpt)
+	}
+
+	hdr := textproto.Header{}
+	hdr.Add("From", "from@example.org")
+	hdr.Add("To", "to@example.org")
+	hdr.Add("x-empty-header", "")
+	act, err = session.Header(hdr)
+	assertAction(act, err, ActContinue)
+	if len(mm.Hdr) != 3 {
+		t.Fatal("Unexpected header length:", len(mm.Hdr))
+	}
+	if val := mm.Hdr.Get("From"); val != "from@example.org" {
+		t.Fatal("Wrong From header:", val)
+	}
+	if val := mm.Hdr.Get("To"); val != "to@example.org" {
+		t.Fatal("Wrong To header:", val)
+	}
+	if val := mm.Hdr.Get("x-empty-header"); val != "" {
+		t.Fatal("Wrong To header:", val)
+	}
+
+	modifyActs, act, err := session.BodyReadFrom(bytes.NewReader(bytes.Repeat([]byte{'A'}, 128000)))
+	assertAction(act, err, ActContinue)
+
+	if len(mm.Chunks) != 2 {
+		t.Fatal("Wrong amount of body chunks received")
+	}
+	if len(mm.Chunks[0]) > 65535 {
+		t.Fatal("Too big first chunk:", len(mm.Chunks[0]))
+	}
+	if totalLen := len(mm.Chunks[0]) + len(mm.Chunks[1]); totalLen < 128000 {
+		t.Fatal("Some body bytes lost:", totalLen)
+	}
+
+	expected := []ModifyAction{
+		{
+			Code:        ActAddHeader,
+			HeaderName:  "X-Bad",
+			HeaderValue: "very",
+		},
+		{
+			Code:        ActChangeHeader,
+			HeaderIndex: 1,
+			HeaderName:  "Subject",
+			HeaderValue: "***SPAM***",
+		},
+		{
+			Code:   ActQuarantine,
+			Reason: "very bad message",
+		},
+	}
+
+	if !reflect.DeepEqual(modifyActs, expected) {
+		t.Fatalf("Wrong modify actions, got %+v", modifyActs)
+	}
+}
+
+func TestMilterClient_AbortFlow(t *testing.T) {
+	macros := make(map[string]string)
+	mm := MockMilter{
+		ConnResp: RespContinue,
+		HeloResp: RespContinue,
+		HeloMod: func(m *Modifier) {
+			macros = m.Macros
+		},
+		AbortMod: func(m *Modifier) {
+			macros = m.Macros
+		},
+	}
+	s := Server{
+		NewMilter: func() Milter {
+			return &mm
+		},
+		Actions: OptAddHeader | OptChangeHeader,
+	}
+	defer s.Close()
+	local, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Serve(local)
+
+	cl := NewClientWithOptions("tcp", local.Addr().String(), ClientOptions{
+		ActionMask: OptAddHeader | OptChangeHeader | OptQuarantine,
+	})
+	defer cl.Close()
+	session, err := cl.Session()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer session.Close()
+
+	assertAction := func(act *Action, err error, expectCode ActionCode) {
+		t.Helper()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if act.Code != expectCode {
+			t.Fatal("Unexpectedcode:", act.Code)
+		}
+	}
+
+	act, err := session.Conn("host", FamilyInet, 25565, "172.0.0.1")
+	assertAction(act, err, ActContinue)
+	if mm.Host != "host" {
+		t.Fatal("Wrong host:", mm.Host)
+	}
+	if mm.Family != "tcp4" {
+		t.Fatal("Wrong family:", mm.Family)
+	}
+	if mm.Port != 25565 {
+		t.Fatal("Wrong port:", mm.Port)
+	}
+	if mm.Addr.String() != "172.0.0.1" {
+		t.Fatal("Wrong IP:", mm.Addr)
+	}
+
+	if err := session.Macros(CodeHelo, "tls_version", "very old"); err != nil {
+		t.Fatal("Unexpected error", err)
+	}
+
+	act, err = session.Helo("helo_host")
+	assertAction(act, err, ActContinue)
+	if mm.HeloValue != "helo_host" {
+		t.Fatal("Wrong helo value:", mm.HeloValue)
+	}
+	if v, ok := macros["tls_version"]; !ok || v != "very old" {
+		t.Fatal("Wrong tls_version macro value:", v)
+	}
+
+	err = session.Abort()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Validate macro values are preserved for the abort callback
+	if v, ok := macros["tls_version"]; !ok || v != "very old" {
+		t.Fatal("Wrong tls_version macro value: ", v)
+	}
+
+	act, err = session.Helo("repeated_helo_host")
+	assertAction(act, err, ActContinue)
+	if mm.HeloValue != "repeated_helo_host" {
+		t.Fatal("Wrong helo value:", mm.HeloValue)
+	}
+	if len(macros["tls_version"]) != 0 {
+		t.Fatal("Unexpected macro data:", macros)
+	}
+}
+
+// TestMilterClient_ImpossibleClientDowngrade tests that the client does not downgrade to v2
+// in case of a v6 bit set in the ActionMask.
+func TestMilterClient_ImpossibleClientDowngrade(t *testing.T) {
+	mm := MockMilter{
+		ConnResp:      RespContinue,
+		HeloResp:      RespContinue,
+		MailResp:      RespContinue,
+		RcptResp:      RespContinue,
+		HdrResp:       RespContinue,
+		HdrsResp:      RespContinue,
+		BodyChunkResp: RespContinue,
+		BodyResp:      RespContinue,
+		BodyMod: func(m *Modifier) {
+			m.AddHeader("X-Bad", "very")
+			m.ChangeHeader(1, "Subject", "***SPAM***")
+			m.Quarantine("very bad message")
+		},
+	}
+	s := Server{
+		NewMilter: func() Milter {
+			return &mm
+		},
+		Actions: OptAddHeader | OptChangeHeader,
+	}
+	defer s.Close()
+	local, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Serve(local)
+
+	cl := NewClientWithOptions("tcp", local.Addr().String(), ClientOptions{
+		// OptChangeFrom is only supported by a server with v6, but this server only supports v2.
+		// This should provoke a downgrade error.
+		ActionMask: OptAddHeader | OptChangeFrom,
+	})
+	defer cl.Close()
+	_, err = cl.Session()
+	if err != ErrUnsupportedMilterVersion {
+		t.Fatalf("Expected ErrUnsupportedMilterVersion, got %v", err)
+	}
+}

third_party/go-milter/cstrings.go

@@ -0,0 +1,33 @@
+package milter
+
+import (
+	"bytes"
+	"strings"
+)
+
+// NULL terminator
+const null = "\x00"
+
+// DecodeCStrings splits a C style strings into a Go slice
+func decodeCStrings(data []byte) []string {
+	if len(data) == 0 {
+		return nil
+	}
+	return strings.Split(strings.Trim(string(data), null), null)
+}
+
+// ReadCString reads and returns a C style string from []byte
+func readCString(data []byte) string {
+	pos := bytes.IndexByte(data, 0)
+	if pos == -1 {
+		return string(data)
+	}
+	return string(data[0:pos])
+}
+
+// appendCString appends a C style string to the buffer and returns it (like append does).
+func appendCString(dest []byte, s string) []byte {
+	dest = append(dest, []byte(s)...)
+	dest = append(dest, 0x00)
+	return dest
+}

third_party/go-milter/go.mod

@@ -0,0 +1,5 @@
+module github.com/emersion/go-milter
+
+go 1.12
+
+require github.com/emersion/go-message v0.18.1

third_party/go-milter/go.sum

@@ -0,0 +1,33 @@
+github.com/emersion/go-message v0.18.1 h1:tfTxIoXFSFRwWaZsgnqS1DSZuGpYGzSmCZD8SK3QA2E=
+github.com/emersion/go-message v0.18.1/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

third_party/go-milter/message.go

@@ -0,0 +1,68 @@
+package milter
+
+// Message represents a command sent from milter client
+type Message struct {
+	Code byte
+	Data []byte
+}
+
+type ActionCode byte
+
+const (
+	ActAccept    ActionCode = 'a' // SMFIR_ACCEPT
+	ActContinue  ActionCode = 'c' // SMFIR_CONTINUE
+	ActDiscard   ActionCode = 'd' // SMFIR_DISCARD
+	ActReject    ActionCode = 'r' // SMFIR_REJECT
+	ActTempFail  ActionCode = 't' // SMFIR_TEMPFAIL
+	ActReplyCode ActionCode = 'y' // SMFIR_REPLYCODE
+
+	// [v6]
+	ActSkip ActionCode = 's' // SMFIR_SKIP
+)
+
+type ModifyActCode byte
+
+const (
+	ActAddRcpt      ModifyActCode = '+' // SMFIR_ADDRCPT
+	ActDelRcpt      ModifyActCode = '-' // SMFIR_DELRCPT
+	ActReplBody     ModifyActCode = 'b' // SMFIR_ACCEPT
+	ActAddHeader    ModifyActCode = 'h' // SMFIR_ADDHEADER
+	ActChangeHeader ModifyActCode = 'm' // SMFIR_CHGHEADER
+	ActInsertHeader ModifyActCode = 'i' // SMFIR_INSHEADER
+	ActQuarantine   ModifyActCode = 'q' // SMFIR_QUARANTINE
+
+	// [v6]
+	ActChangeFrom ModifyActCode = 'e' // SMFIR_CHGFROM
+)
+
+type Code byte
+
+const (
+	CodeOptNeg Code = 'O' // SMFIC_OPTNEG
+	CodeMacro  Code = 'D' // SMFIC_MACRO
+	CodeConn   Code = 'C' // SMFIC_CONNECT
+	CodeQuit   Code = 'Q' // SMFIC_QUIT
+	CodeHelo   Code = 'H' // SMFIC_HELO
+	CodeMail   Code = 'M' // SMFIC_MAIL
+	CodeRcpt   Code = 'R' // SMFIC_RCPT
+	CodeHeader Code = 'L' // SMFIC_HEADER
+	CodeEOH    Code = 'N' // SMFIC_EOH
+	CodeBody   Code = 'B' // SMFIC_BODY
+	CodeEOB    Code = 'E' // SMFIC_BODYEOB
+	CodeAbort  Code = 'A' // SMFIC_ABORT
+	CodeData   Code = 'T' // SMFIC_DATA
+
+	// [v6]
+	CodeQuitNewConn Code = 'K' // SMFIC_QUIT_NC
+)
+
+const MaxBodyChunk = 65535
+
+type ProtoFamily byte
+
+const (
+	FamilyUnknown ProtoFamily = 'U' // SMFIA_UNKNOWN
+	FamilyUnix    ProtoFamily = 'L' // SMFIA_UNIX
+	FamilyInet    ProtoFamily = '4' // SMFIA_INET
+	FamilyInet6   ProtoFamily = '6' // SMFIA_INET6
+)

third_party/go-milter/milter.go

@@ -0,0 +1,57 @@
+// Package milter provides an interface to implement milter mail filters
+package milter
+
+// OptAction sets which actions the milter wants to perform.
+// Multiple options can be set using a bitmask.
+type OptAction uint32
+
+// Set which actions the milter wants to perform.
+const (
+	OptAddHeader    OptAction = 1 << 0 // SMFIF_ADDHDRS
+	OptChangeBody   OptAction = 1 << 1 // SMFIF_CHGBODY
+	OptAddRcpt      OptAction = 1 << 2 // SMFIF_ADDRCPT
+	OptRemoveRcpt   OptAction = 1 << 3 // SMFIF_DELRCPT
+	OptChangeHeader OptAction = 1 << 4 // SMFIF_CHGHDRS
+	OptQuarantine   OptAction = 1 << 5 // SMFIF_QUARANTINE
+
+	// [v6]
+	OptChangeFrom      OptAction = 1 << 6 // SMFIF_CHGFROM
+	OptAddRcptWithArgs OptAction = 1 << 7 // SMFIF_ADDRCPT_PAR
+	OptSetSymList      OptAction = 1 << 8 // SMFIF_SETSYMLIST
+)
+
+// OptProtocol masks out unwanted parts of the SMTP transaction.
+// Multiple options can be set using a bitmask.
+type OptProtocol uint32
+
+const (
+	OptNoConnect  OptProtocol = 1 << 0 // SMFIP_NOCONNECT
+	OptNoHelo     OptProtocol = 1 << 1 // SMFIP_NOHELO
+	OptNoMailFrom OptProtocol = 1 << 2 // SMFIP_NOMAIL
+	OptNoRcptTo   OptProtocol = 1 << 3 // SMFIP_NORCPT
+	OptNoBody     OptProtocol = 1 << 4 // SMFIP_NOBODY
+	OptNoHeaders  OptProtocol = 1 << 5 // SMFIP_NOHDRS
+	OptNoEOH      OptProtocol = 1 << 6 // SMFIP_NOEOH
+	OptNoUnknown  OptProtocol = 1 << 8 // SMFIP_NOUNKNOWN
+	OptNoData     OptProtocol = 1 << 9 // SMFIP_NODATA
+
+	// [v6] MTA supports ActSkip
+	OptSkip OptProtocol = 1 << 10 // SMFIP_SKIP
+	// [v6] Filter wants rejected RCPTs
+	OptRcptRej OptProtocol = 1 << 11 // SMFIP_RCPT_REJ
+
+	// Milter will not send action response for the following MTA messages
+	OptNoHeaderReply OptProtocol = 1 << 7 // SMFIP_NR_HDR, SMFIP_NOHREPL
+	// [v6]
+	OptNoConnReply    OptProtocol = 1 << 12 // SMFIP_NR_CONN
+	OptNoHeloReply    OptProtocol = 1 << 13 // SMFIP_NR_HELO
+	OptNoMailReply    OptProtocol = 1 << 14 // SMFIP_NR_MAIL
+	OptNoRcptReply    OptProtocol = 1 << 15 // SMFIP_NR_RCPT
+	OptNoDataReply    OptProtocol = 1 << 16 // SMFIP_NR_DATA
+	OptNoUnknownReply OptProtocol = 1 << 17 // SMFIP_NR_UNKN
+	OptNoEOHReply     OptProtocol = 1 << 18 // SMFIP_NR_EOH
+	OptNoBodyReply    OptProtocol = 1 << 19 // SMFIP_NR_BODY
+
+	// [v6]
+	OptHeaderLeadingSpace OptProtocol = 1 << 20 // SMFIP_HDR_LEADSPC
+)

third_party/go-milter/modifier.go

@@ -0,0 +1,96 @@
+// Modifier instance is provided to milter handlers to modify email messages
+
+package milter
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"net/textproto"
+)
+
+// postfix wants LF lines endings. Using CRLF results in double CR sequences.
+func crlfToLF(b []byte) []byte {
+	return bytes.ReplaceAll(b, []byte{'\r', '\n'}, []byte{'\n'})
+}
+
+// Modifier provides access to Macros, Headers and Body data to callback handlers. It also defines a
+// number of functions that can be used by callback handlers to modify processing of the email message
+type Modifier struct {
+	Macros  map[string]string
+	Headers textproto.MIMEHeader
+
+	writePacket func(*Message) error
+}
+
+// AddRecipient appends a new envelope recipient for current message
+func (m *Modifier) AddRecipient(r string) error {
+	data := []byte(fmt.Sprintf("<%s>", r) + null)
+	return m.writePacket(NewResponse('+', data).Response())
+}
+
+// DeleteRecipient removes an envelope recipient address from message
+func (m *Modifier) DeleteRecipient(r string) error {
+	data := []byte(fmt.Sprintf("<%s>", r) + null)
+	return m.writePacket(NewResponse('-', data).Response())
+}
+
+// ReplaceBody substitutes message body with provided body
+func (m *Modifier) ReplaceBody(body []byte) error {
+	body = crlfToLF(body)
+	return m.writePacket(NewResponse('b', body).Response())
+}
+
+// AddHeader appends a new email message header the message
+func (m *Modifier) AddHeader(name, value string) error {
+	var buffer bytes.Buffer
+	buffer.WriteString(name + null)
+	buffer.Write(crlfToLF([]byte(value)))
+	buffer.WriteString(null)
+	return m.writePacket(NewResponse('h', buffer.Bytes()).Response())
+}
+
+// Quarantine a message by giving a reason to hold it
+func (m *Modifier) Quarantine(reason string) error {
+	return m.writePacket(NewResponse('q', []byte(reason+null)).Response())
+}
+
+// ChangeHeader replaces the header at the specified position with a new one.
+// The index is per name.
+func (m *Modifier) ChangeHeader(index int, name, value string) error {
+	var buffer bytes.Buffer
+	if err := binary.Write(&buffer, binary.BigEndian, uint32(index)); err != nil {
+		return err
+	}
+	buffer.WriteString(name + null)
+	buffer.Write(crlfToLF([]byte(value)))
+	buffer.WriteString(null)
+	return m.writePacket(NewResponse('m', buffer.Bytes()).Response())
+}
+
+// InsertHeader inserts the header at the specified position
+func (m *Modifier) InsertHeader(index int, name, value string) error {
+	var buffer bytes.Buffer
+	if err := binary.Write(&buffer, binary.BigEndian, uint32(index)); err != nil {
+		return err
+	}
+	buffer.WriteString(name + null)
+	buffer.Write(crlfToLF([]byte(value)))
+	buffer.WriteString(null)
+	return m.writePacket(NewResponse('i', buffer.Bytes()).Response())
+}
+
+// ChangeFrom replaces the FROM envelope header with a new one
+func (m *Modifier) ChangeFrom(value string) error {
+	data := []byte(value + null)
+	return m.writePacket(NewResponse('e', data).Response())
+}
+
+// newModifier creates a new Modifier instance from milterSession
+func newModifier(s *milterSession) *Modifier {
+	return &Modifier{
+		Macros:      s.macros,
+		Headers:     s.headers,
+		writePacket: s.WritePacket,
+	}
+}

third_party/go-milter/response.go

@@ -0,0 +1,62 @@
+package milter
+
+// Response represents a response structure returned by callback
+// handlers to indicate how the milter server should proceed
+type Response interface {
+	Response() *Message
+	Continue() bool
+}
+
+// SimpleResponse type to define list of pre-defined responses
+type SimpleResponse byte
+
+// Response returns a Message object reference
+func (r SimpleResponse) Response() *Message {
+	return &Message{byte(r), nil}
+}
+
+// Continue to process milter messages only if current code is Continue
+func (r SimpleResponse) Continue() bool {
+	return ActionCode(r) == ActContinue
+}
+
+// Define standard responses with no data
+const (
+	RespAccept   = SimpleResponse(ActAccept)
+	RespContinue = SimpleResponse(ActContinue)
+	RespDiscard  = SimpleResponse(ActDiscard)
+	RespReject   = SimpleResponse(ActReject)
+	RespTempFail = SimpleResponse(ActTempFail)
+)
+
+// CustomResponse is a response instance used by callback handlers to indicate
+// how the milter should continue processing of current message
+type CustomResponse struct {
+	code byte
+	data []byte
+}
+
+// Response returns message instance with data
+func (c *CustomResponse) Response() *Message {
+	return &Message{c.code, c.data}
+}
+
+// Continue returns false if milter chain should be stopped, true otherwise
+func (c *CustomResponse) Continue() bool {
+	for _, q := range []ActionCode{ActAccept, ActDiscard, ActReject, ActTempFail} {
+		if c.code == byte(q) {
+			return false
+		}
+	}
+	return true
+}
+
+// NewResponse generates a new CustomResponse suitable for WritePacket
+func NewResponse(code byte, data []byte) *CustomResponse {
+	return &CustomResponse{code, data}
+}
+
+// NewResponseStr generates a new CustomResponse with string payload
+func NewResponseStr(code byte, data string) *CustomResponse {
+	return NewResponse(code, []byte(data+null))
+}

third_party/go-milter/server.go

@@ -0,0 +1,144 @@
+package milter
+
+import (
+	"errors"
+	"net"
+	"net/textproto"
+)
+
+// Milter protocol version implemented by the server.
+//
+// Note: Not exported as we might want to support multiple versions
+// transparently in the future.
+var serverProtocolVersion uint32 = 2
+
+// ErrServerClosed is returned by the Server's Serve method after a call to
+// Close.
+var ErrServerClosed = errors.New("milter: server closed")
+
+// Milter is an interface for milter callback handlers.
+type Milter interface {
+	// Connect is called to provide SMTP connection data for incoming message.
+	// Suppress with OptNoConnect.
+	Connect(host string, family string, port uint16, addr net.IP, m *Modifier) (Response, error)
+
+	// Helo is called to process any HELO/EHLO related filters. Suppress with
+	// OptNoHelo.
+	Helo(name string, m *Modifier) (Response, error)
+
+	// MailFrom is called to process filters on envelope FROM address. Suppress
+	// with OptNoMailFrom.
+	MailFrom(from string, m *Modifier) (Response, error)
+
+	// RcptTo is called to process filters on envelope TO address. Suppress with
+	// OptNoRcptTo.
+	RcptTo(rcptTo string, m *Modifier) (Response, error)
+
+	// Header is called once for each header in incoming message. Suppress with
+	// OptNoHeaders.
+	Header(name string, value string, m *Modifier) (Response, error)
+
+	// Headers is called when all message headers have been processed. Suppress
+	// with OptNoEOH.
+	Headers(h textproto.MIMEHeader, m *Modifier) (Response, error)
+
+	// BodyChunk is called to process next message body chunk data (up to 64KB
+	// in size). Suppress with OptNoBody.
+	BodyChunk(chunk []byte, m *Modifier) (Response, error)
+
+	// Body is called at the end of each message. All changes to message's
+	// content & attributes must be done here.
+	Body(m *Modifier) (Response, error)
+
+	// Abort is called is the current message has been aborted. All message data
+	// should be reset to prior to the Helo callback. Connection data should be
+	// preserved.
+	Abort(m *Modifier) error
+}
+
+// NoOpMilter is a dummy Milter implementation that does nothing.
+type NoOpMilter struct{}
+
+var _ Milter = NoOpMilter{}
+
+func (NoOpMilter) Connect(host string, family string, port uint16, addr net.IP, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) Helo(name string, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) MailFrom(from string, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) RcptTo(rcptTo string, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) Header(name string, value string, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) Headers(h textproto.MIMEHeader, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) BodyChunk(chunk []byte, m *Modifier) (Response, error) {
+	return RespContinue, nil
+}
+
+func (NoOpMilter) Body(m *Modifier) (Response, error) {
+	return RespAccept, nil
+}
+
+func (NoOpMilter) Abort(m *Modifier) error {
+	return nil
+}
+
+// Server is a milter server.
+type Server struct {
+	NewMilter func() Milter
+	Actions   OptAction
+	Protocol  OptProtocol
+
+	listeners []net.Listener
+	closed    bool
+}
+
+// Serve starts the server.
+func (s *Server) Serve(ln net.Listener) error {
+	defer ln.Close()
+
+	s.listeners = append(s.listeners, ln)
+
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			if s.closed {
+				return ErrServerClosed
+			}
+			return err
+		}
+
+		session := milterSession{
+			server:   s,
+			actions:  s.Actions,
+			protocol: s.Protocol,
+			conn:     conn,
+			backend:  s.NewMilter(),
+		}
+		go session.HandleMilterCommands()
+	}
+}
+
+func (s *Server) Close() error {
+	s.closed = true
+	for _, ln := range s.listeners {
+		if err := ln.Close(); err != nil {
+			return err
+		}
+	}
+	return nil
+}

third_party/go-milter/session.go

@@ -0,0 +1,269 @@
+package milter
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+	"log"
+	"net"
+	"net/textproto"
+	"strings"
+	"time"
+)
+
+var errCloseSession = errors.New("Stop current milter processing")
+
+// milterSession keeps session state during MTA communication
+type milterSession struct {
+	server   *Server
+	actions  OptAction
+	protocol OptProtocol
+	conn     net.Conn
+	headers  textproto.MIMEHeader
+	macros   map[string]string
+	backend  Milter
+}
+
+// ReadPacket reads incoming milter packet
+func (c *milterSession) ReadPacket() (*Message, error) {
+	return readPacket(c.conn, 0)
+}
+
+func readPacket(conn net.Conn, timeout time.Duration) (*Message, error) {
+	if timeout != 0 {
+		conn.SetReadDeadline(time.Now().Add(timeout))
+		defer conn.SetReadDeadline(time.Time{})
+	}
+
+	// read packet length
+	var length uint32
+	if err := binary.Read(conn, binary.BigEndian, &length); err != nil {
+		return nil, err
+	}
+
+	// read packet data
+	data := make([]byte, length)
+	if _, err := io.ReadFull(conn, data); err != nil {
+		return nil, err
+	}
+
+	// prepare response data
+	message := Message{
+		Code: data[0],
+		Data: data[1:],
+	}
+
+	return &message, nil
+}
+
+// WritePacket sends a milter response packet to socket stream
+func (m *milterSession) WritePacket(msg *Message) error {
+	return writePacket(m.conn, msg, 0)
+}
+
+func writePacket(conn net.Conn, msg *Message, timeout time.Duration) error {
+	if timeout != 0 {
+		conn.SetWriteDeadline(time.Now().Add(timeout))
+		defer conn.SetWriteDeadline(time.Time{})
+	}
+
+	buffer := bufio.NewWriter(conn)
+
+	// calculate and write response length
+	length := uint32(len(msg.Data) + 1)
+	if err := binary.Write(buffer, binary.BigEndian, length); err != nil {
+		return err
+	}
+
+	// write response code
+	if err := buffer.WriteByte(msg.Code); err != nil {
+		return err
+	}
+
+	// write response data
+	if _, err := buffer.Write(msg.Data); err != nil {
+		return err
+	}
+
+	// flush data to network socket stream
+	if err := buffer.Flush(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Process processes incoming milter commands
+func (m *milterSession) Process(msg *Message) (Response, error) {
+	switch Code(msg.Code) {
+	case CodeAbort:
+		// abort current message and start over
+		defer func() {
+			m.headers = nil
+			m.macros = nil
+		}()
+		return nil, m.backend.Abort(newModifier(m))
+
+	case CodeBody:
+		// body chunk
+		return m.backend.BodyChunk(msg.Data, newModifier(m))
+
+	case CodeConn:
+		// new connection, get hostname
+		hostname := readCString(msg.Data)
+		msg.Data = msg.Data[len(hostname)+1:]
+		// get protocol family
+		protocolFamily := msg.Data[0]
+		msg.Data = msg.Data[1:]
+		// get port
+		var port uint16
+		if protocolFamily == '4' || protocolFamily == '6' {
+			if len(msg.Data) < 2 {
+				return RespTempFail, nil
+			}
+			port = binary.BigEndian.Uint16(msg.Data)
+			msg.Data = msg.Data[2:]
+		}
+		// get address
+		address := readCString(msg.Data)
+		// convert address and port to human readable string
+		family := map[byte]string{
+			'U': "unknown",
+			'L': "unix",
+			'4': "tcp4",
+			'6': "tcp6",
+		}
+		// run handler and return
+		return m.backend.Connect(
+			hostname,
+			family[protocolFamily],
+			port,
+			net.ParseIP(address),
+			newModifier(m))
+
+	case CodeMacro:
+		// define macros
+		m.macros = make(map[string]string)
+		// convert data to Go strings
+		data := decodeCStrings(msg.Data[1:])
+		if len(data) != 0 {
+			if len(data)%2 == 1 {
+				data = append(data, "")
+			}
+
+			// store data in a map
+			for i := 0; i < len(data); i += 2 {
+				m.macros[data[i]] = data[i+1]
+			}
+		}
+		// do not send response
+		return nil, nil
+
+	case CodeEOB:
+		// call and return milter handler
+		return m.backend.Body(newModifier(m))
+
+	case CodeHelo:
+		// helo command
+		name := strings.TrimSuffix(string(msg.Data), null)
+		return m.backend.Helo(name, newModifier(m))
+
+	case CodeHeader:
+		// SelfPost patch (PATCHES.md): the header is handed to the backend
+		// and not kept. Upstream adds every header of the session to
+		// m.headers, emptied only on ABORT, and so held a whole header
+		// section - up to Postfix's message_size_limit - in memory for each
+		// connection; SelfPost's milters keep what they need themselves.
+		headerData := decodeCStrings(msg.Data)
+		// headers with an empty body appear as `text\x00\x00`, decodeCStrings will drop the empty body
+		if len(headerData) == 1 {
+			headerData = append(headerData, "")
+		}
+		if len(headerData) == 2 {
+			// call and return milter handler
+			return m.backend.Header(headerData[0], headerData[1], newModifier(m))
+		}
+
+	case CodeMail:
+		// envelope from address
+		from := readCString(msg.Data)
+		return m.backend.MailFrom(strings.Trim(from, "<>"), newModifier(m))
+
+	case CodeEOH:
+		// end of headers; SelfPost patch: no headers are kept, so the map
+		// handed over is always empty (PATCHES.md)
+		return m.backend.Headers(textproto.MIMEHeader{}, newModifier(m))
+
+	case CodeOptNeg:
+		// ignore request and prepare response buffer
+		var buffer bytes.Buffer
+		// prepare response data
+		for _, value := range []uint32{serverProtocolVersion, uint32(m.actions), uint32(m.protocol)} {
+			if err := binary.Write(&buffer, binary.BigEndian, value); err != nil {
+				return nil, err
+			}
+		}
+		// build and send packet
+		return NewResponse('O', buffer.Bytes()), nil
+
+	case CodeQuit:
+		// client requested session close
+		return nil, errCloseSession
+
+	case CodeRcpt:
+		// envelope to address
+		to := readCString(msg.Data)
+		return m.backend.RcptTo(strings.Trim(to, "<>"), newModifier(m))
+
+	case CodeData:
+		// data, ignore
+
+	default:
+		// print error and close session
+		log.Printf("Unrecognized command code: %c", msg.Code)
+		return nil, errCloseSession
+	}
+
+	// by default continue with next milter message
+	return RespContinue, nil
+}
+
+// HandleMilterComands processes all milter commands in the same connection
+func (m *milterSession) HandleMilterCommands() {
+	defer m.conn.Close()
+
+	for {
+		msg, err := m.ReadPacket()
+		if err != nil {
+			if err != io.EOF {
+				log.Printf("Error reading milter command: %v", err)
+			}
+			return
+		}
+
+		resp, err := m.Process(msg)
+		if err != nil {
+			if err != errCloseSession {
+				// log error condition
+				log.Printf("Error performing milter command: %v", err)
+			}
+			return
+		}
+
+		// ignore empty responses
+		if resp != nil {
+			// send back response message
+			if err = m.WritePacket(resp.Response()); err != nil {
+				log.Printf("Error writing packet: %v", err)
+				return
+			}
+
+			if !resp.Continue() {
+				// prepare backend for next message
+				m.backend = m.server.NewMilter()
+			}
+		}
+	}
+}