diff --git a/framework/module/streaming_delivery.go b/framework/module/streaming_delivery.go new file mode 100644 index 00000000..2ca0ffd5 --- /dev/null +++ b/framework/module/streaming_delivery.go @@ -0,0 +1,44 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package module + +import ( + "context" + "io" + + "github.com/emersion/go-message/textproto" +) + +// StreamingDelivery is an optional interface that may be implemented by the +// object returned by DeliveryTarget.StartDelivery to receive the message body +// while it is still being read from the source, without buffering it first. +type StreamingDelivery interface { + // CanStreamBody reports whether BodyStream can be used instead of Body for + // the current message. It is called after all AddRcpt calls. + CanStreamBody() bool + + // BodyStream is similar to Body of the regular Delivery interface except + // that the body is read exactly once, while the message source is still + // receiving it. + // + // If reading body fails, BodyStream must return an error and must make + // sure the partially read message is not delivered. The message source + // calls Abort after a failed BodyStream. + BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error +} diff --git a/internal/endpoint/smtp/session.go b/internal/endpoint/smtp/session.go index 11729151..2136f4dc 100644 --- a/internal/endpoint/smtp/session.go +++ b/internal/endpoint/smtp/session.go @@ -428,7 +428,7 @@ func (s *Session) Logout() error { return nil } -func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, error) { +func (s *Session) readHeader(r io.Reader) (textproto.Header, io.Reader, error) { limitr := limitReader(r, s.endp.maxHeaderBytes, &exterrors.SMTPError{ Code: 552, EnhancedCode: exterrors.EnhancedCode{5, 3, 4}, @@ -451,9 +451,26 @@ func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, err // the header size check is done. The message size will be checked by go-smtp limitr.Enabled = false - buf, err := s.endp.buffer(bufr) + return header, bufr, nil +} + +func (s *Session) bufferBody(body io.Reader) (buffer.Buffer, error) { + buf, err := s.endp.buffer(body) + if err != nil { + return nil, fmt.Errorf("I/O error while writing buffer: %w", err) + } + return buf, nil +} + +func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, error) { + header, body, err := s.readHeader(r) + if err != nil { + return textproto.Header{}, nil, err + } + + buf, err := s.bufferBody(body) if err != nil { - return textproto.Header{}, nil, fmt.Errorf("I/O error while writing buffer: %w", err) + return textproto.Header{}, nil, err } return header, buf, nil @@ -471,27 +488,39 @@ func (s *Session) Data(r io.Reader) error { return s.endp.wrapErr(s.msgMeta.ID, !s.opts.UTF8, "DATA", err) } - header, buf, err := s.prepareBody(r) + header, body, err := s.readHeader(r) if err != nil { return wrapErr(err) } - defer func() { - if err := buf.Remove(); err != nil { - s.log.Error("failed to remove buffered body", err) - } - - // go-smtp will call Reset, but it will call Abort if delivery is non-nil. - s.cleanSession() - }() if err := s.checkRoutingLoops(header); err != nil { return wrapErr(err) } - if err := s.delivery.Body(bodyCtx, header, buf); err != nil { + if streamDelivery, ok := s.delivery.(module.StreamingDelivery); ok && streamDelivery.CanStreamBody() { + err = streamDelivery.BodyStream(bodyCtx, header, body) + } else { + var buf buffer.Buffer + buf, err = s.bufferBody(body) + if err != nil { + return wrapErr(err) + } + defer func() { + if err := buf.Remove(); err != nil { + s.log.Error("failed to remove buffered body", err) + } + }() + + err = s.delivery.Body(bodyCtx, header, buf) + } + if err != nil { + // s.delivery is kept so that Reset aborts it. return wrapErr(err) } + // go-smtp will call Reset, but it will call Abort if delivery is non-nil. + defer s.cleanSession() + if err := s.delivery.Commit(bodyCtx); err != nil { return wrapErr(err) } diff --git a/internal/endpoint/smtp/smtp_test.go b/internal/endpoint/smtp/smtp_test.go index 485c7e9e..e5873248 100644 --- a/internal/endpoint/smtp/smtp_test.go +++ b/internal/endpoint/smtp/smtp_test.go @@ -451,6 +451,78 @@ func TestSMTPDelivery_AbortData(t *testing.T) { } } +func TestSMTPDelivery_StreamBody(t *testing.T) { + tgt := testutils.Target{Streaming: true} + endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) + defer func() { + assert.NoError(t, endp.Stop()) + }() + + cl, err := smtp.Dial("127.0.0.1:" + testPort) + if err != nil { + t.Fatal(err) + } + defer func() { + assert.NoError(t, cl.Close()) + }() + + err = submitMsg(t, cl, "sender@example.org", []string{"rcpt1@example.com", "rcpt2@example.com"}, testMsg) + if err != nil { + t.Fatal(err) + } + + if len(tgt.Messages) != 1 { + t.Fatal("Expected a message, got", len(tgt.Messages)) + } + msg := tgt.Messages[0] + if !msg.Streamed { + t.Error("Message body was not streamed") + } + testutils.CheckMsgID(t, &msg, "sender@example.org", []string{"rcpt1@example.com", "rcpt2@example.com"}, "") + if msg.Header.Get("Received") == "" { + t.Error("Missing Received field") + } +} + +func TestSMTPDelivery_StreamBody_AbortData(t *testing.T) { + tgt := testutils.Target{Streaming: true} + endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) + defer func() { + assert.NoError(t, endp.Stop()) + }() + + conn, err := net.Dial("tcp", "127.0.0.1:"+testPort) + if err != nil { + t.Fatal(err) + } + cl := smtp.NewClient(conn) + + if err := cl.Hello("mx.example.org"); err != nil { + t.Fatal(err) + } + if err := cl.Mail("sender@example.org", nil); err != nil { + t.Fatal(err) + } + if err := cl.Rcpt("test@example.com", &smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + if _, err := cl.Data(); err != nil { + t.Fatal(err) + } + // Send the header and a part of the body, then close the connection + // without sending the final dot. + if _, err := conn.Write([]byte(testMsg)); err != nil { + t.Fatal(err) + } + require.NoError(t, conn.Close()) + + time.Sleep(250 * time.Millisecond) + + if len(tgt.Messages) != 0 { + t.Fatal("Expected no messages, got", len(tgt.Messages)) + } +} + func TestSMTPDelivery_EmptyMessage(t *testing.T) { tgt := testutils.Target{} endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) diff --git a/internal/msgpipeline/bodystream_test.go b/internal/msgpipeline/bodystream_test.go new file mode 100644 index 00000000..d39cac4d --- /dev/null +++ b/internal/msgpipeline/bodystream_test.go @@ -0,0 +1,128 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package msgpipeline + +import ( + "context" + "testing" + + "github.com/emersion/go-smtp" + "github.com/foxcpp/maddy/framework/module" + "github.com/foxcpp/maddy/internal/modify" + "github.com/foxcpp/maddy/internal/testutils" +) + +func streamTestPipeline(t *testing.T, targets ...module.DeliveryTarget) *MsgPipeline { + return &MsgPipeline{ + msgpipelineCfg: msgpipelineCfg{ + perSource: map[string]sourceBlock{}, + defaultSource: sourceBlock{ + perRcpt: map[string]*rcptBlock{}, + defaultRcpt: &rcptBlock{ + targets: targets, + }, + }, + }, + Log: testutils.Logger(t, "msgpipeline"), + } +} + +func TestMsgPipeline_BodyStream(t *testing.T) { + target := testutils.Target{Streaming: true} + d := streamTestPipeline(t, &target) + + testutils.DoTestDeliveryStream(t, d, "sender@example.com", []string{"rcpt1@example.com", "rcpt2@example.com"}) + + if len(target.Messages) != 1 { + t.Fatalf("wrong amount of messages received, want %d, got %d", 1, len(target.Messages)) + } + testutils.CheckTestMessage(t, &target, 0, "sender@example.com", []string{"rcpt1@example.com", "rcpt2@example.com"}) + if !target.Messages[0].Streamed { + t.Error("message was not streamed") + } +} + +func TestMsgPipeline_CanStreamBody(t *testing.T) { + canStream := func(t *testing.T, d *MsgPipeline) bool { + t.Helper() + + delivery, err := d.StartDelivery(context.Background(), &module.MsgMetadata{ID: "test"}, "sender@example.com") + if err != nil { + t.Fatal(err) + } + defer func() { + if err := delivery.Abort(context.Background()); err != nil { + t.Error(err) + } + }() + if err := delivery.AddRcpt(context.Background(), "rcpt@example.com", smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + return delivery.(module.StreamingDelivery).CanStreamBody() + } + + t.Run("single streaming target", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + if !canStream(t, d) { + t.Error("expected streaming to be possible") + } + }) + t.Run("target without streaming", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{}) + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("two targets", func(t *testing.T) { + d := streamTestPipeline(t, + &testutils.Target{Streaming: true, InstName: "a"}, + &testutils.Target{Streaming: true, InstName: "b"}) + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("global check", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.globalChecks = []module.Check{&testutils.Check{}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("per-recipient check", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.defaultSource.defaultRcpt.checks = []module.Check{&testutils.Check{}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("source modifier", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.defaultSource.modifiers = modify.Group{Modifiers: []module.Modifier{testutils.Modifier{}}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("dmarc", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.doDMARC = true + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) +} diff --git a/internal/msgpipeline/msgpipeline.go b/internal/msgpipeline/msgpipeline.go index 6025bf06..6fb818cf 100644 --- a/internal/msgpipeline/msgpipeline.go +++ b/internal/msgpipeline/msgpipeline.go @@ -20,6 +20,7 @@ package msgpipeline import ( "context" + "io" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -393,19 +394,7 @@ func (dd *msgpipelineDelivery) Body(ctx context.Context, header textproto.Header } } - if dd.d.FirstPipeline { - // Add Received *after* checks to make sure they see the message literally - // how we received it BUT place it below any other field that might be - // added by applyResults (including Authentication-Results) - // per recommendation in RFC 7001, Section 4 (see GH issue #135). - received, err := target.GenerateReceived(ctx, dd.msgMeta, dd.d.Hostname, dd.msgMeta.OriginalFrom) - if err != nil { - return err - } - header.Add("Received", received) - } - - if err := dd.checkRunner.applyResults(dd.d.Hostname, &header); err != nil { + if err := dd.addTraceFields(ctx, &header); err != nil { return err } @@ -432,6 +421,67 @@ func (dd *msgpipelineDelivery) Body(ctx context.Context, header textproto.Header return nil } +func (dd *msgpipelineDelivery) addTraceFields(ctx context.Context, header *textproto.Header) error { + if dd.d.FirstPipeline { + // Add Received *after* checks to make sure they see the message literally + // how we received it BUT place it below any other field that might be + // added by applyResults (including Authentication-Results) + // per recommendation in RFC 7001, Section 4 (see GH issue #135). + received, err := target.GenerateReceived(ctx, dd.msgMeta, dd.d.Hostname, dd.msgMeta.OriginalFrom) + if err != nil { + return err + } + header.Add("Received", received) + } + + return dd.checkRunner.applyResults(dd.d.Hostname, header) +} + +// CanStreamBody implements module.StreamingDelivery. +// +// The body can be streamed only if nothing in the pipeline needs to read it +// before the target does, and there is exactly one target to read it. +func (dd *msgpipelineDelivery) CanStreamBody() bool { + if dd.checkRunner.doDMARC { + return false + } + if len(dd.d.globalChecks) != 0 || len(dd.sourceBlock.checks) != 0 { + return false + } + if len(dd.d.globalModifiers.Modifiers) != 0 || len(dd.sourceBlock.modifiers.Modifiers) != 0 { + return false + } + for blk := range dd.rcptModifiersState { + if len(blk.checks) != 0 || len(blk.modifiers.Modifiers) != 0 { + return false + } + } + + if len(dd.deliveries) != 1 { + return false + } + var streamDelivery module.StreamingDelivery + for _, delivery := range dd.deliveries { + streamDelivery, _ = delivery.Delivery.(module.StreamingDelivery) + } + return streamDelivery != nil && streamDelivery.CanStreamBody() +} + +// BodyStream implements module.StreamingDelivery. +func (dd *msgpipelineDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + if err := dd.addTraceFields(ctx, &header); err != nil { + return err + } + + for _, delivery := range dd.deliveries { + if err := delivery.Delivery.(module.StreamingDelivery).BodyStream(ctx, header, body); err != nil { + return err + } + dd.log.Debugf("delivery.BodyStream ok, Delivery object = %T", delivery.Delivery) + } + return nil +} + // statusCollector wraps StatusCollector and adds reverse translation // of recipients for all statuses.] // diff --git a/internal/target/ignore_error/ignore_error.go b/internal/target/ignore_error/ignore_error.go index 82efc65e..72058431 100644 --- a/internal/target/ignore_error/ignore_error.go +++ b/internal/target/ignore_error/ignore_error.go @@ -33,6 +33,7 @@ package ignore_error import ( "context" + "io" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -116,6 +117,34 @@ func (d *delivery) Body(ctx context.Context, header textproto.Header, body buffe return nil } +func (d *delivery) CanStreamBody() bool { + if d.wrapped == nil { + return true + } + streamDelivery, ok := d.wrapped.(module.StreamingDelivery) + return ok && streamDelivery.CanStreamBody() +} + +// BodyStream implements module.StreamingDelivery. +// +// Errors from the wrapped delivery are ignored as in Body, but errors from +// reading the body are not, since they come from the message source. +func (d *delivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + r := target.NewSourceReader(body) + if d.wrapped != nil { + if err := d.wrapped.(module.StreamingDelivery).BodyStream(ctx, header, r); err != nil && r.Err() == nil { + d.log.Error("ignored error from wrapped target", err) + } + } + + // Consume the rest of the body, as the buffered path does, so that + // the message source can still report a read error. + if r.Err() == nil { + _, _ = io.Copy(io.Discard, r) + } + return r.Err() +} + func (d *delivery) Abort(ctx context.Context) error { if d.wrapped == nil { return nil diff --git a/internal/target/ignore_error/ignore_error_test.go b/internal/target/ignore_error/ignore_error_test.go index 4f2b1897..a78e86ae 100644 --- a/internal/target/ignore_error/ignore_error_test.go +++ b/internal/target/ignore_error/ignore_error_test.go @@ -19,10 +19,17 @@ along with this program. If not, see . package ignore_error import ( + "context" "errors" + "io" + "strings" "testing" + "testing/iotest" + "github.com/emersion/go-message/textproto" + "github.com/emersion/go-smtp" "github.com/foxcpp/maddy/framework/config" + "github.com/foxcpp/maddy/framework/module" "github.com/foxcpp/maddy/internal/testutils" ) @@ -64,6 +71,86 @@ func TestIgnoredErrors(t *testing.T) { } } +func TestBodyStream(t *testing.T) { + inner := &testutils.Target{Streaming: true} + tgt := wrap(t, inner) + + testutils.DoTestDeliveryStream(t, tgt, "from@example.org", []string{"to@example.org"}) + + testutils.CheckTestMessage(t, inner, 0, "from@example.org", []string{"to@example.org"}) + if !inner.Messages[0].Streamed { + t.Error("message body was not streamed to the wrapped target") + } +} + +func TestBodyStreamNotSupported(t *testing.T) { + tgt := wrap(t, &testutils.Target{}) + + d, err := tgt.StartDelivery(context.Background(), &module.MsgMetadata{}, "from@example.org") + if err != nil { + t.Fatal(err) + } + if d.(module.StreamingDelivery).CanStreamBody() { + t.Error("CanStreamBody should be false if the wrapped delivery can't stream") + } +} + +func startStream(t *testing.T, tgt *Target) module.StreamingDelivery { + t.Helper() + + d, err := tgt.StartDelivery(context.Background(), &module.MsgMetadata{}, "from@example.org") + if err != nil { + t.Fatal(err) + } + if err := d.AddRcpt(context.Background(), "to@example.org", smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + streamDelivery := d.(module.StreamingDelivery) + if !streamDelivery.CanStreamBody() { + t.Fatal("CanStreamBody is false") + } + return streamDelivery +} + +func TestBodyStreamIgnoredError(t *testing.T) { + boom := errors.New("boom") + + for name, inner := range map[string]*testutils.Target{ + "StartDelivery": {Streaming: true, StartErr: boom}, + "BodyStream": {Streaming: true, BodyErr: boom}, + } { + t.Run(name, func(t *testing.T) { + d := startStream(t, wrap(t, inner)) + + body := strings.NewReader("foobar\r\n") + if err := d.BodyStream(context.Background(), textproto.Header{}, body); err != nil { + t.Errorf("wrapped error was not ignored: %v", err) + } + if body.Len() != 0 { + t.Errorf("body was not read to the end, %d bytes left", body.Len()) + } + }) + } +} + +func TestBodyStreamReadError(t *testing.T) { + readErr := errors.New("client went away") + + for name, inner := range map[string]*testutils.Target{ + "Passthrough": {Streaming: true}, + "BodyStream": {Streaming: true, BodyErr: errors.New("boom")}, + } { + t.Run(name, func(t *testing.T) { + d := startStream(t, wrap(t, inner)) + + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + if err := d.BodyStream(context.Background(), textproto.Header{}, body); !errors.Is(err, readErr) { + t.Errorf("expected the read error, got %v", err) + } + }) + } +} + func TestConfigureRequiresTarget(t *testing.T) { tgt := &Target{instName: "test", log: testutils.Logger(t, modName)} diff --git a/internal/target/queue/queue.go b/internal/target/queue/queue.go index 1aec8213..5c7633bf 100644 --- a/internal/target/queue/queue.go +++ b/internal/target/queue/queue.go @@ -600,8 +600,32 @@ func (qd *queueDelivery) AddRcpt(ctx context.Context, rcptTo string, _ smtp.Rcpt func (qd *queueDelivery) Body(ctx context.Context, header textproto.Header, body buffer.Buffer) error { defer trace.StartRegion(ctx, "queue/Body").End() + bodyReader, err := body.Open() + if err != nil { + return err + } + defer func() { + if err := bodyReader.Close(); err != nil { + qd.q.log.Error("bodyReader close failed", err) + } + }() + // Body buffer initially passed to us may not be valid after "delivery" to queue completes. // storeNewMessage returns a new buffer object created from message blob stored on disk. + return qd.store(header, bodyReader) +} + +func (qd *queueDelivery) CanStreamBody() bool { + return true +} + +func (qd *queueDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + defer trace.StartRegion(ctx, "queue/BodyStream").End() + + return qd.store(header, body) +} + +func (qd *queueDelivery) store(header textproto.Header, body io.Reader) error { storedBody, err := qd.q.storeNewMessage(qd.meta, header, body) if err != nil { return err @@ -752,7 +776,7 @@ func (q *Queue) readDiskQueue() error { return nil } -func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, body buffer.Buffer) (buffer.Buffer, error) { +func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, body io.Reader) (buffer.Buffer, error) { id := meta.MsgMeta.ID headerPath := filepath.Join(q.location, id+".header") @@ -771,17 +795,6 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo return nil, err } - bodyReader, err := body.Open() - if err != nil { - q.tryRemoveDanglingFile(id + ".header") - return nil, err - } - defer func() { - if err := bodyReader.Close(); err != nil { - q.log.Error("bodyReader close failed", err) - } - }() - bodyPath := filepath.Join(q.location, id+".body") bodyFile, err := os.Create(bodyPath) if err != nil { @@ -793,7 +806,8 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo } }() - if _, err := io.Copy(bodyFile, bodyReader); err != nil { + bodyLen, err := io.Copy(bodyFile, body) + if err != nil { q.tryRemoveDanglingFile(id + ".body") q.tryRemoveDanglingFile(id + ".header") return nil, err @@ -815,7 +829,7 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo queuedMsgs.WithLabelValues(q.name, q.location).Inc() - return buffer.FileBuffer{Path: bodyPath, LenHint: body.Len()}, nil + return buffer.FileBuffer{Path: bodyPath, LenHint: int(bodyLen)}, nil } func (q *Queue) updateMetadataOnDisk(meta *QueueMetadata) error { diff --git a/internal/target/queue/queue_test.go b/internal/target/queue/queue_test.go index 2ad9e821..4c51010d 100644 --- a/internal/target/queue/queue_test.go +++ b/internal/target/queue/queue_test.go @@ -30,6 +30,7 @@ import ( "reflect" "strings" "testing" + "testing/iotest" "time" "github.com/emersion/go-message/textproto" @@ -255,6 +256,50 @@ func TestQueueDelivery(t *testing.T) { checkQueueDir(t, q, []string{}) } +func TestQueueDelivery_BodyStream(t *testing.T) { + t.Parallel() + + dt := unreliableTarget{committed: make(chan testutils.Msg, 10)} + q := newTestQueue(t, &dt) + defer cleanQueue(t, q) + + testutils.DoTestDeliveryStream(t, q, "tester@example.com", []string{"tester1@example.org", "tester2@example.org"}) + + msg := readMsgChanTimeout(t, dt.committed, 5*time.Second) + require.NoError(t, q.Stop()) + + testutils.CheckMsgID(t, msg, "tester@example.com", []string{"tester1@example.org", "tester2@example.org"}, "") + + checkQueueDir(t, q, []string{}) +} + +func TestQueueDelivery_BodyStream_ReadErr(t *testing.T) { + t.Parallel() + + dt := unreliableTarget{ + committed: make(chan testutils.Msg, 10), + aborted: make(chan testutils.Msg, 10), + } + q := newTestQueue(t, &dt) + defer cleanQueue(t, q) + + ctx := context.Background() + delivery, err := q.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "tester@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "tester1@example.org", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + require.NoError(t, delivery.Abort(ctx)) + + require.NoError(t, q.Stop()) + require.Empty(t, dt.committed) + require.Empty(t, dt.aborted) + checkQueueDir(t, q, []string{}) +} + func TestQueueDelivery_PermanentFail_NonPartial(t *testing.T) { t.Parallel() diff --git a/internal/target/remote/connect.go b/internal/target/remote/connect.go index 3b9692d0..9f71184a 100644 --- a/internal/target/remote/connect.go +++ b/internal/target/remote/connect.go @@ -212,6 +212,10 @@ func (rd *remoteDelivery) closeConn(c *mxConn) { rd.rt.limits.ReleaseDest(c.domain) } + // Client is nil if the connection was already dropped. + if c.Client() == nil { + return + } if err := c.Close(); err != nil { rd.log.Error("client connection close failed", err) } diff --git a/internal/target/remote/remote.go b/internal/target/remote/remote.go index daf5a874..26828722 100644 --- a/internal/target/remote/remote.go +++ b/internal/target/remote/remote.go @@ -28,6 +28,7 @@ import ( "crypto/tls" "errors" "fmt" + "io" "net" "runtime/trace" "strings" @@ -442,6 +443,47 @@ func (rd *remoteDelivery) BodyNonAtomic(ctx context.Context, c module.StatusColl wg.Wait() } +// CanStreamBody implements module.StreamingDelivery. +// +// The body can be streamed only if it is sent over a single connection, +// that is, all recipients are in one domain. +func (rd *remoteDelivery) CanStreamBody() bool { + return len(rd.connections) == 1 +} + +// BodyStream implements module.StreamingDelivery. +func (rd *remoteDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + defer trace.StartRegion(ctx, "remote/BodyStream").End() + + if rd.msgMeta.Quarantine { + return &exterrors.SMTPError{ + Code: 550, + EnhancedCode: exterrors.EnhancedCode{5, 7, 0}, + Message: "Refusing to deliver quarantined message", + TargetName: "remote", + } + } + + for _, conn := range rd.connections { + r := target.NewSourceReader(body) + err := conn.Data(ctx, header, r) + conn.errored = err != nil + conn.lastUseAt = time.Now() + if r.Err() != nil { + // Sending QUIT would first terminate the DATA command and make the + // server accept the partial message. Drop the connection instead. + if err := conn.DirectClose(); err != nil { + rd.log.Error("failed to close SMTP connection", err) + } + return r.Err() + } + if err != nil { + return err + } + } + return nil +} + func (rd *remoteDelivery) Abort(ctx context.Context) error { return rd.Close() } diff --git a/internal/target/remote/remote_test.go b/internal/target/remote/remote_test.go index 06f0cb80..8e351341 100644 --- a/internal/target/remote/remote_test.go +++ b/internal/target/remote/remote_test.go @@ -21,12 +21,16 @@ package remote import ( "context" "crypto/tls" + "errors" "flag" + "io" "math/rand" "net" "os" "strconv" + "strings" "testing" + "testing/iotest" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -142,6 +146,103 @@ func TestRemoteDelivery(t *testing.T) { be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid"}) } +func TestRemoteDelivery_BodyStream(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + defer func() { + assert.NoError(t, tgt.Stop()) + }() + testutils.DoTestDeliveryStream(t, tgt, "test@example.com", []string{"test@example.invalid", "test2@example.invalid"}) + + be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid", "test2@example.invalid"}) +} + +func TestRemoteDelivery_BodyStream_MultipleConns(t *testing.T) { + _, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "example2.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + defer func() { + assert.NoError(t, tgt.Stop()) + }() + + ctx := context.Background() + delivery, err := tgt.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "test@example.invalid", smtp.RcptOptions{})) + require.True(t, delivery.(module.StreamingDelivery).CanStreamBody()) + require.NoError(t, delivery.AddRcpt(ctx, "test@example2.invalid", smtp.RcptOptions{})) + require.False(t, delivery.(module.StreamingDelivery).CanStreamBody()) + require.NoError(t, delivery.Abort(ctx)) +} + +func TestRemoteDelivery_BodyStream_ReadErr(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + tgt.connReuseLimit = 5 + defer func() { + assert.NoError(t, tgt.Stop()) + }() + + ctx := context.Background() + delivery, err := tgt.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "test@example.invalid", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + require.NoError(t, delivery.Abort(ctx)) + + testutils.WaitForConnsClose(t, srv) + require.Empty(t, be.Messages) + + // The dropped connection must not be returned to the pool. + testutils.DoTestDelivery(t, tgt, "test@example.com", []string{"test@example.invalid"}) + be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid"}) +} + func TestRemoteDelivery_NoMXFallback(t *testing.T) { tarpit := testutils.FailOnConn(t, "127.0.0.1:"+smtpPort) defer func() { diff --git a/internal/target/smtp/smtp_downstream.go b/internal/target/smtp/smtp_downstream.go index 823b1032..8fd91060 100644 --- a/internal/target/smtp/smtp_downstream.go +++ b/internal/target/smtp/smtp_downstream.go @@ -30,6 +30,7 @@ import ( "context" "crypto/tls" "fmt" + "io" "net" "runtime/trace" "time" @@ -176,6 +177,8 @@ type delivery struct { rcpts []string conn *smtpconn.C + + connDropped bool } // lmtpDelivery implements module.PartialDelivery @@ -301,6 +304,25 @@ func (d *delivery) Body(ctx context.Context, header textproto.Header, body buffe return d.u.moduleError(d.conn.Data(ctx, header, r)) } +func (d *delivery) CanStreamBody() bool { + return true +} + +func (d *delivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + r := target.NewSourceReader(body) + err := d.conn.Data(ctx, header, r) + if r.Err() != nil { + // Sending QUIT would first terminate the DATA command and make the + // server accept the partial message. Drop the connection instead. + d.connDropped = true + if err := d.conn.DirectClose(); err != nil { + d.log.Error("failed to close SMTP connection", err) + } + return r.Err() + } + return d.u.moduleError(err) +} + func (d *lmtpDelivery) BodyNonAtomic(ctx context.Context, sc module.StatusCollector, header textproto.Header, body buffer.Buffer) { r, err := body.Open() if err != nil { @@ -339,10 +361,16 @@ func (d *lmtpDelivery) BodyNonAtomic(ctx context.Context, sc module.StatusCollec } func (d *delivery) Abort(ctx context.Context) error { + if d.connDropped { + return nil + } return d.conn.Close() } func (d *delivery) Commit(ctx context.Context) error { + if d.connDropped { + return nil + } return d.conn.Close() } diff --git a/internal/target/smtp/smtp_downstream_test.go b/internal/target/smtp/smtp_downstream_test.go index e292f5e3..ad2505a6 100644 --- a/internal/target/smtp/smtp_downstream_test.go +++ b/internal/target/smtp/smtp_downstream_test.go @@ -19,16 +19,22 @@ along with this program. If not, see . package smtp_downstream import ( + "context" "errors" "flag" + "io" "math/rand" "os" "strconv" + "strings" "testing" + "testing/iotest" + "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" "github.com/foxcpp/maddy/framework/config" "github.com/foxcpp/maddy/framework/exterrors" + "github.com/foxcpp/maddy/framework/module" "github.com/foxcpp/maddy/internal/testutils" "github.com/stretchr/testify/require" ) @@ -276,6 +282,64 @@ func TestDownstreamDelivery_StartTLS_NoFallback(t *testing.T) { } } +func TestDownstreamDelivery_BodyStream(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+testPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + + mod := &Downstream{ + hostname: "mx.example.invalid", + endpoints: []config.Endpoint{ + { + Scheme: "tcp", + Host: "127.0.0.1", + Port: testPort, + }, + }, + log: testutils.Logger(t, "target.smtp"), + } + + testutils.DoTestDeliveryStream(t, mod, "test@example.invalid", []string{"rcpt@example.invalid"}) + be.CheckMsg(t, 0, "test@example.invalid", []string{"rcpt@example.invalid"}) +} + +func TestDownstreamDelivery_BodyStream_ReadErr(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+testPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + + mod := &Downstream{ + hostname: "mx.example.invalid", + endpoints: []config.Endpoint{ + { + Scheme: "tcp", + Host: "127.0.0.1", + Port: testPort, + }, + }, + log: testutils.Logger(t, "target.smtp"), + } + + ctx := context.Background() + delivery, err := mod.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.invalid") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "rcpt@example.invalid", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + + require.NoError(t, delivery.Abort(ctx)) + + testutils.WaitForConnsClose(t, srv) + require.Empty(t, be.Messages) +} + func TestMain(m *testing.M) { remoteSmtpPort := flag.String("test.smtpport", "random", "(maddy) SMTP port to use for connections in tests") flag.Parse() diff --git a/internal/target/source_reader.go b/internal/target/source_reader.go new file mode 100644 index 00000000..42aac1e2 --- /dev/null +++ b/internal/target/source_reader.go @@ -0,0 +1,47 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package target + +import "io" + +// SourceReader wraps the body passed to module.StreamingDelivery.BodyStream +// and records the error returned by the message source, so that the target +// can tell it apart from its own errors. +type SourceReader struct { + r io.Reader + err error +} + +func NewSourceReader(r io.Reader) *SourceReader { + return &SourceReader{r: r} +} + +func (r *SourceReader) Read(b []byte) (int, error) { + n, err := r.r.Read(b) + if err != nil && err != io.EOF { + r.err = err + } + return n, err +} + +// Err returns the last error other than io.EOF returned by the underlying +// reader. +func (r *SourceReader) Err() error { + return r.err +} diff --git a/internal/testutils/target.go b/internal/testutils/target.go index 221834ba..f8b04990 100644 --- a/internal/testutils/target.go +++ b/internal/testutils/target.go @@ -26,6 +26,7 @@ import ( "io" "reflect" "sort" + "strings" "testing" "github.com/emersion/go-message/textproto" @@ -42,6 +43,7 @@ type Msg struct { RcptTo []string Body []byte Header textproto.Header + Streamed bool } type Target struct { @@ -55,6 +57,9 @@ type Target struct { AbortErr error CommitErr error + // Streaming makes deliveries implement module.StreamingDelivery. + Streaming bool + InstName string } @@ -86,6 +91,10 @@ type testTargetDeliveryPartial struct { testTargetDelivery } +type testTargetDeliveryStreaming struct { + testTargetDelivery +} + func (dt *Target) StartDelivery(ctx context.Context, msgMeta *module.MsgMetadata, mailFrom string) (module.Delivery, error) { if dt.PartialBodyErr != nil { return &testTargetDeliveryPartial{ @@ -95,6 +104,14 @@ func (dt *Target) StartDelivery(ctx context.Context, msgMeta *module.MsgMetadata }, }, dt.StartErr } + if dt.Streaming { + return &testTargetDeliveryStreaming{ + testTargetDelivery: testTargetDelivery{ + tgt: dt, + msg: Msg{MsgMeta: msgMeta, MailFrom: mailFrom}, + }, + }, dt.StartErr + } return &testTargetDelivery{ tgt: dt, msg: Msg{MsgMeta: msgMeta, MailFrom: mailFrom}, @@ -173,6 +190,23 @@ func (dtd *testTargetDelivery) Body(ctx context.Context, header textproto.Header return err } +func (dtd *testTargetDeliveryStreaming) CanStreamBody() bool { + return true +} + +func (dtd *testTargetDeliveryStreaming) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + if dtd.tgt.BodyErr != nil { + return dtd.tgt.BodyErr + } + + dtd.msg.Header = header + dtd.msg.Streamed = true + + var err error + dtd.msg.Body, err = io.ReadAll(body) + return err +} + func (dtd *testTargetDelivery) Abort(ctx context.Context) error { return dtd.tgt.AbortErr } @@ -251,6 +285,54 @@ func DoTestDeliveryNonAtomic(t *testing.T, c module.StatusCollector, tgt module. return encodedID } +// DoTestDeliveryStream is similar to DoTestDelivery but passes the body using +// module.StreamingDelivery. +func DoTestDeliveryStream(t *testing.T, tgt module.DeliveryTarget, from string, to []string) string { + t.Helper() + + IDRaw := sha1.Sum([]byte(t.Name())) + encodedID := hex.EncodeToString(IDRaw[:]) + testCtx := context.Background() + + msgMeta := module.MsgMetadata{ + DontTraceSender: true, + ID: encodedID, + OriginalFrom: from, + } + delivery, err := tgt.StartDelivery(testCtx, &msgMeta, from) + if err != nil { + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + for _, rcpt := range to { + if err := delivery.AddRcpt(testCtx, rcpt, smtp.RcptOptions{}); err != nil { + if err := delivery.Abort(testCtx); err != nil { + t.Log("-- delivery.Abort:", err, exterrors.Fields(err)) + } + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + } + + streamDelivery, ok := delivery.(module.StreamingDelivery) + if !ok || !streamDelivery.CanStreamBody() { + t.Fatalf("Delivery object %T can't stream the body", delivery) + } + + hdr := textproto.Header{} + hdr.Add("B", "2") + hdr.Add("A", "1") + if err := streamDelivery.BodyStream(testCtx, hdr, strings.NewReader("foobar\r\n")); err != nil { + if err := delivery.Abort(testCtx); err != nil { + t.Log("-- delivery.Abort:", err, exterrors.Fields(err)) + } + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + if err := delivery.Commit(testCtx); err != nil { + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + + return encodedID +} + const DeliveryData = "A: 1\r\n" + "B: 2\r\n" + "\r\n" +