diff --git a/server/listen/smtp_server/delivery_error.go b/server/listen/smtp_server/delivery_error.go new file mode 100644 index 0000000..ec289cb --- /dev/null +++ b/server/listen/smtp_server/delivery_error.go @@ -0,0 +1,94 @@ +package smtp_server + +import ( + "errors" + "net" + "net/textproto" + "sort" + "strings" + "unicode/utf8" + + smtp "github.com/emersion/go-smtp" +) + +const maxDeliverySMTPMessageBytes = 128 + +// downstreamDeliverySMTPError turns the synchronous downstream delivery result +// into the SMTP response returned to the submitting client. Unknown failures are +// temporary so a transient network problem does not permanently drop a message. +func downstreamDeliverySMTPError(deliveryErr error, domainErrors map[string]error) error { + if deliveryErr == nil { + return nil + } + + permanent := len(domainErrors) > 0 + for _, domainErr := range domainErrors { + if domainErr == nil || !isPermanentDeliveryError(domainErr) { + permanent = false + break + } + } + + if permanent { + return &smtp.SMTPError{ + Code: 550, + EnhancedCode: smtp.EnhancedCode{5, 0, 0}, + Message: deliverySMTPMessage("Downstream delivery permanently failed", deliveryErr, domainErrors), + } + } + + return &smtp.SMTPError{ + Code: 451, + EnhancedCode: smtp.EnhancedCode{4, 4, 0}, + Message: deliverySMTPMessage("Downstream delivery temporarily failed", deliveryErr, domainErrors), + } +} + +func deliverySMTPMessage(prefix string, deliveryErr error, domainErrors map[string]error) string { + details := make([]string, 0, len(domainErrors)) + domains := make([]string, 0, len(domainErrors)) + for domain := range domainErrors { + domains = append(domains, domain) + } + sort.Strings(domains) + for _, domain := range domains { + if domainErr := domainErrors[domain]; domainErr != nil { + details = append(details, domain+": "+domainErr.Error()) + } + } + if len(details) == 0 && deliveryErr != nil { + details = append(details, deliveryErr.Error()) + } + + message := prefix + if len(details) > 0 { + // SMTP replies must stay on one bounded line, even when a remote server + // returns malformed text. Retain useful detail for the submitting client. + message += ": " + strings.Join(strings.Fields(strings.Join(details, "; ")), " ") + } + if len(message) <= maxDeliverySMTPMessageBytes { + return message + } + + limit := maxDeliverySMTPMessageBytes - len("...") + for len(message) > limit { + _, size := utf8.DecodeLastRuneInString(message) + message = message[:len(message)-size] + } + return message + "..." +} + +func isPermanentDeliveryError(err error) bool { + var protocolErr *textproto.Error + if errors.As(err, &protocolErr) { + return protocolErr.Code >= 500 && protocolErr.Code <= 599 + } + + var smtpErr *smtp.SMTPError + if errors.As(err, &smtpErr) { + return smtpErr.Code >= 500 && smtpErr.Code <= 599 + } + + var dnsErr *net.DNSError + return errors.As(err, &dnsErr) && dnsErr.IsNotFound +} diff --git a/server/listen/smtp_server/delivery_error_test.go b/server/listen/smtp_server/delivery_error_test.go new file mode 100644 index 0000000..d5a06f7 --- /dev/null +++ b/server/listen/smtp_server/delivery_error_test.go @@ -0,0 +1,136 @@ +package smtp_server + +import ( + "errors" + "fmt" + "net" + "net/textproto" + "strings" + "testing" + + smtp "github.com/emersion/go-smtp" +) + +func TestDownstreamDeliverySMTPError(t *testing.T) { + tests := []struct { + name string + deliveryErr error + domainErrors map[string]error + wantCode int + wantEnhanced smtp.EnhancedCode + wantMessages []string + }{ + {name: "success"}, + { + name: "network failure is temporary", + deliveryErr: errors.New("delivery failed"), + domainErrors: map[string]error{"example.com": &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}}, + wantCode: 451, + wantEnhanced: smtp.EnhancedCode{4, 4, 0}, + wantMessages: []string{"connection refused"}, + }, + { + name: "missing domain details is temporary", + deliveryErr: errors.New("delivery failed"), + wantCode: 451, + wantEnhanced: smtp.EnhancedCode{4, 4, 0}, + wantMessages: []string{"delivery failed"}, + }, + { + name: "downstream 4xx is temporary", + deliveryErr: errors.New("delivery failed"), + domainErrors: map[string]error{ + "example.com": &textproto.Error{Code: 421, Msg: "try again later"}, + }, + wantCode: 451, + wantEnhanced: smtp.EnhancedCode{4, 4, 0}, + wantMessages: []string{"421", "try again later"}, + }, + { + name: "downstream 5xx is permanent", + deliveryErr: errors.New("delivery failed"), + domainErrors: map[string]error{ + "example.com": fmt.Errorf("remote rejected recipient: %w", &textproto.Error{Code: 550, Msg: "mailbox unavailable"}), + }, + wantCode: 550, + wantEnhanced: smtp.EnhancedCode{5, 0, 0}, + wantMessages: []string{"550", "mailbox unavailable"}, + }, + { + name: "nxdomain is permanent", + deliveryErr: errors.New("delivery failed"), + domainErrors: map[string]error{ + "missing.example": &net.OpError{Op: "dial", Net: "tcp", Err: &net.DNSError{Err: "no such host", Name: "missing.example", IsNotFound: true}}, + }, + wantCode: 550, + wantEnhanced: smtp.EnhancedCode{5, 0, 0}, + wantMessages: []string{"no such host"}, + }, + { + name: "mixed permanent and temporary failures are temporary", + deliveryErr: errors.New("delivery failed"), + domainErrors: map[string]error{ + "rejected.example": &textproto.Error{Code: 550, Msg: "mailbox unavailable"}, + "slow.example": &net.OpError{Op: "dial", Net: "tcp", Err: &net.DNSError{Err: "timeout", Name: "slow.example", IsTimeout: true}}, + }, + wantCode: 451, + wantEnhanced: smtp.EnhancedCode{4, 4, 0}, + wantMessages: []string{"550", "mailbox unavailable"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := downstreamDeliverySMTPError(tt.deliveryErr, tt.domainErrors) + if tt.wantCode == 0 { + if err != nil { + t.Fatalf("downstreamDeliverySMTPError() = %v, want nil", err) + } + return + } + + var smtpErr *smtp.SMTPError + if !errors.As(err, &smtpErr) { + t.Fatalf("downstreamDeliverySMTPError() type = %T, want *smtp.SMTPError", err) + } + if smtpErr.Code != tt.wantCode { + t.Errorf("SMTP code = %d, want %d", smtpErr.Code, tt.wantCode) + } + if smtpErr.EnhancedCode != tt.wantEnhanced { + t.Errorf("enhanced code = %v, want %v", smtpErr.EnhancedCode, tt.wantEnhanced) + } + for _, part := range tt.wantMessages { + if !strings.Contains(smtpErr.Message, part) { + t.Errorf("SMTP message = %q, want it to contain %q", smtpErr.Message, part) + } + } + if strings.ContainsAny(smtpErr.Message, "\r\n") { + t.Errorf("SMTP message must be one line, got %q", smtpErr.Message) + } + if len(smtpErr.Message) > maxDeliverySMTPMessageBytes { + t.Errorf("SMTP message is too long: %d bytes", len(smtpErr.Message)) + } + }) + } +} + +func TestDeliverySMTPMessageSanitizesAndBoundsDetails(t *testing.T) { + message := deliverySMTPMessage( + "Downstream delivery temporarily failed", + errors.New("fallback"), + map[string]error{ + "b.example": errors.New("line one\r\nline two"), + "a.example": errors.New(strings.Repeat("x", 200)), + }, + ) + + if strings.ContainsAny(message, "\r\n") { + t.Fatalf("SMTP message must be one line, got %q", message) + } + if len(message) > maxDeliverySMTPMessageBytes { + t.Fatalf("SMTP message is too long: %d bytes", len(message)) + } + if !strings.Contains(message, "a.example") { + t.Fatalf("SMTP message is not deterministically sorted: %q", message) + } +} diff --git a/server/listen/smtp_server/read_content.go b/server/listen/smtp_server/read_content.go index 844eeb7..10880e9 100644 --- a/server/listen/smtp_server/read_content.go +++ b/server/listen/smtp_server/read_content.go @@ -99,7 +99,7 @@ func (s *Session) Data(r io.Reader) error { } errMsg := "" - err, sendErr := send.Send(ctx, email) + deliveryErr, sendErr := send.Send(ctx, email) log.WithContext(ctx).Debugf("插件执行--SendAfter") @@ -115,8 +115,8 @@ func (s *Session) Data(r io.Reader) error { as3.Wait() log.WithContext(ctx).Debugf("插件执行--SendAfter") - if err != nil { - errMsg = err.Error() + if deliveryErr != nil { + errMsg = deliveryErr.Error() _, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =2 ,error=? where id = ? "), errMsg, email.MessageId) if err != nil { log.WithContext(ctx).Errorf("sql Error :%+v", err) @@ -126,6 +126,8 @@ func (s *Session) Data(r io.Reader) error { log.WithContext(ctx).Errorf("sql Error :%+v", err) } + return downstreamDeliverySMTPError(deliveryErr, sendErr) + } else { _, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =1 where id = ? "), email.MessageId) if err != nil { diff --git a/server/utils/send/send.go b/server/utils/send/send.go index 54dab8b..d7a7415 100644 --- a/server/utils/send/send.go +++ b/server/utils/send/send.go @@ -4,6 +4,7 @@ import ( "crypto/tls" "crypto/x509" "errors" + "fmt" "github.com/Jinnrry/pmail/config" "github.com/Jinnrry/pmail/dto/parsemail" "github.com/Jinnrry/pmail/models" @@ -14,6 +15,7 @@ import ( "github.com/Jinnrry/pmail/utils/smtp" log "github.com/sirupsen/logrus" "net" + "net/textproto" "strings" "sync" ) @@ -23,6 +25,19 @@ type mxDomain struct { mxHost string } +type temporaryMXFallbackError struct { + lookupErr error + fallbackErr error +} + +func (e *temporaryMXFallbackError) Error() string { + return fmt.Sprintf("temporary MX lookup failed (%v); fallback delivery failed: %v", e.lookupErr, e.fallbackErr) +} + +func (e *temporaryMXFallbackError) Unwrap() error { + return e.lookupErr +} + // Forward 转发邮件 func Forward(ctx *context.Context, e *parsemail.Email, forwardAddress string, user *models.User) error { @@ -70,6 +85,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai // 按域名整理 toByDomain := map[mxDomain][]*parsemail.User{} + mxLookupErrors := map[mxDomain]error{} for _, s := range to { args := strings.Split(s.EmailAddress, "@") if len(args) == 2 { @@ -82,13 +98,13 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai toByDomain[address] = append(toByDomain[address], s) } else { //查询dns mx记录 - mxInfo, err := net.LookupMX(args[1]) + mxInfo, lookupErr := net.LookupMX(args[1]) address := mxDomain{ domain: "smtp." + args[1], mxHost: "smtp." + args[1], } - if err != nil { - log.WithContext(ctx).Errorf(s.EmailAddress, "域名mx记录查询失败，检查邮箱是否存在！") + if lookupErr != nil { + log.WithContext(ctx).Errorf("%s 域名mx记录查询失败，检查邮箱是否存在！", s.EmailAddress) } if len(mxInfo) > 0 { address = mxDomain{ @@ -96,6 +112,9 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai mxHost: mxInfo[0].Host, } } + if lookupErr != nil { + mxLookupErrors[address] = lookupErr + } toByDomain[address] = append(toByDomain[address], s) } } else { @@ -105,6 +124,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai } var errEmailAddress []string + var errEmailAddressMu sync.Mutex errMap := sync.Map{} @@ -112,12 +132,25 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai for domain, tos := range toByDomain { domain := domain tos := tos + mxLookupErr := mxLookupErrors[domain] as.WaitProcess(func(p any) { + recordFailure := func(err error) { + err = deliveryFailureCause(mxLookupErr, err) + log.WithContext(ctx).Errorf("%v 邮件投递失败%+v", tos, err) + + errEmailAddressMu.Lock() + for _, user := range tos { + errEmailAddress = append(errEmailAddress, user.EmailAddress) + } + errEmailAddressMu.Unlock() + + errMap.Store(domain.domain, err) + } if domain.domain == "localhost" { err := smtp.SendMailUnsafe("", domain.mxHost+":25", nil, from, fromDomain, buildAddress(tos), data) if err != nil { - log.WithContext(ctx).Errorf("send error %s", err.Error()) + recordFailure(err) } return } @@ -143,6 +176,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai if err == nil { return } + if isPermanentSMTPResponse(err) { + recordFailure(err) + return + } log.WithContext(ctx).Infof("SMTP STARTTLS on 25 Send Error. %s", err.Error()) // 再试用587投递 @@ -150,6 +187,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai if err == nil { return } + if isPermanentSMTPResponse(err) { + recordFailure(err) + return + } log.WithContext(ctx).Infof("SMTPS on 587 Send Error. %s", err.Error()) // 再次尝试465投递 @@ -157,6 +198,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai if err == nil { return } + if isPermanentSMTPResponse(err) { + recordFailure(err) + return + } log.WithContext(ctx).Infof("SMTPS on 465 Send Error. %s", err.Error()) // 最后尝试非安全方式投递 @@ -166,13 +211,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai return } - if err != nil { - log.WithContext(ctx).Errorf("%v 邮件投递失败%+v", tos, err) - for _, user := range tos { - errEmailAddress = append(errEmailAddress, user.EmailAddress) - } - } - errMap.Store(domain.domain, err) + recordFailure(err) }, nil) } as.Wait() @@ -194,6 +233,32 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai return nil, orgMap } +func isPermanentSMTPResponse(err error) bool { + var protocolErr *textproto.Error + return errors.As(err, &protocolErr) && protocolErr.Code >= 500 && protocolErr.Code <= 599 +} + +func deliveryFailureCause(mxLookupErr, fallbackErr error) error { + if fallbackErr == nil || isPermanentSMTPResponse(fallbackErr) { + return fallbackErr + } + + var lookupDNSErr *net.DNSError + if !errors.As(mxLookupErr, &lookupDNSErr) || (!lookupDNSErr.IsTimeout && !lookupDNSErr.IsTemporary) { + return fallbackErr + } + + var networkErr net.Error + if !errors.As(fallbackErr, &networkErr) { + return fallbackErr + } + + return &temporaryMXFallbackError{ + lookupErr: mxLookupErr, + fallbackErr: fallbackErr, + } +} + func buildAddress(u []*parsemail.User) []string { var ret []string diff --git a/server/utils/send/send_test.go b/server/utils/send/send_test.go new file mode 100644 index 0000000..1be44d2 --- /dev/null +++ b/server/utils/send/send_test.go @@ -0,0 +1,80 @@ +package send + +import ( + "errors" + "fmt" + "net" + "net/textproto" + "testing" +) + +func TestIsPermanentSMTPResponse(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "nil"}, + {name: "network error", err: errors.New("connection refused")}, + {name: "temporary SMTP response", err: &textproto.Error{Code: 451, Msg: "try later"}}, + {name: "wrapped temporary SMTP response", err: fmt.Errorf("delivery: %w", &textproto.Error{Code: 421, Msg: "service unavailable"})}, + {name: "permanent SMTP response", err: &textproto.Error{Code: 550, Msg: "mailbox unavailable"}, want: true}, + {name: "wrapped permanent SMTP response", err: fmt.Errorf("delivery: %w", &textproto.Error{Code: 554, Msg: "transaction failed"}), want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isPermanentSMTPResponse(tt.err); got != tt.want { + t.Fatalf("isPermanentSMTPResponse() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestDeliveryFailureCausePreservesTemporaryMXError(t *testing.T) { + tests := []struct { + name string + lookupErr *net.DNSError + }{ + { + name: "SERVFAIL before fallback NXDOMAIN", + lookupErr: &net.DNSError{Err: "server misbehaving", Name: "example.com", IsTemporary: true}, + }, + { + name: "timeout before fallback NXDOMAIN", + lookupErr: &net.DNSError{Err: "i/o timeout", Name: "example.com", IsTimeout: true}, + }, + } + + fallbackErr := &net.OpError{ + Op: "dial", + Net: "tcp", + Err: &net.DNSError{Err: "no such host", Name: "smtp.example.com", IsNotFound: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := deliveryFailureCause(tt.lookupErr, fallbackErr) + + var dnsErr *net.DNSError + if !errors.As(err, &dnsErr) { + t.Fatalf("deliveryFailureCause() = %T, want wrapped DNS error", err) + } + if dnsErr.IsNotFound { + t.Fatalf("deliveryFailureCause() selected fallback NXDOMAIN: %v", dnsErr) + } + if !dnsErr.IsTemporary && !dnsErr.IsTimeout { + t.Fatalf("deliveryFailureCause() selected non-temporary DNS error: %v", dnsErr) + } + }) + } +} + +func TestDeliveryFailureCauseKeepsExplicitSMTPRejection(t *testing.T) { + lookupErr := &net.DNSError{Err: "server misbehaving", Name: "example.com", IsTemporary: true} + rejection := &textproto.Error{Code: 550, Msg: "mailbox unavailable"} + + if got := deliveryFailureCause(lookupErr, rejection); got != rejection { + t.Fatalf("deliveryFailureCause() = %v, want explicit SMTP rejection %v", got, rejection) + } +} diff --git a/server/utils/smtp/smtp.go b/server/utils/smtp/smtp.go index e07b176..ffad3b5 100644 --- a/server/utils/smtp/smtp.go +++ b/server/utils/smtp/smtp.go @@ -36,6 +36,8 @@ import ( var NoSupportSTARTTLSError = errors.New("smtp: server doesn't support STARTTLS") var EOFError = errors.New("EOF") +const outboundSMTPQuitWriteTimeout = 250 * time.Millisecond + // A Client represents a client connection to an SMTP server. type Client struct { // Text is the textproto.Conn used by the Client. It is exported to allow for @@ -341,6 +343,26 @@ func (c *Client) Data() (io.WriteCloser, error) { return &dataCloser{c, c.Text.DotWriter()}, nil } +func finishDelivery(c *Client, w io.WriteCloser, msg []byte) error { + if _, err := w.Write(msg); err != nil { + return err + } + if err := w.Close(); err != nil { + return err + } + + // The final 250 response to DATA transfers responsibility for the message. + // Send QUIT with a bounded write, but do not wait for 221: a failed or + // missing reply after acceptance must not block or trigger redelivery. + if err := c.conn.SetWriteDeadline(time.Now().Add(outboundSMTPQuitWriteTimeout)); err != nil { + log.Debugf("Could not bound SMTP QUIT write after message acceptance: %v", err) + } + if err := c.Text.PrintfLine("QUIT"); err != nil { + log.Debugf("SMTP QUIT failed after message acceptance: %v", err) + } + return nil +} + func SendMailWithTls(domain string, addr string, a smtp.Auth, from string, fromDomain string, to []string, msg []byte) error { if err := validateLine(from); err != nil { return err @@ -378,15 +400,7 @@ func SendMailWithTls(domain string, addr string, a smtp.Auth, from string, fromD if err != nil { return err } - _, err = w.Write(msg) - if err != nil { - return err - } - err = w.Close() - if err != nil { - return err - } - return c.Quit() + return finishDelivery(c, w, msg) } // SendMail connects to the server at addr, switches to TLS if @@ -464,15 +478,7 @@ func SendMail(domain string, addr string, a smtp.Auth, from string, fromDomain s if err != nil { return err } - _, err = w.Write(msg) - if err != nil { - return err - } - err = w.Close() - if err != nil { - return err - } - return c.Quit() + return finishDelivery(c, w, msg) } // SendMailUnsafe 无TLS加密的邮件发送方式 @@ -514,15 +520,7 @@ func SendMailUnsafe(domain string, addr string, a smtp.Auth, from string, fromDo if err != nil { return err } - _, err = w.Write(msg) - if err != nil { - return err - } - err = w.Close() - if err != nil { - return err - } - return c.Quit() + return finishDelivery(c, w, msg) } // Extension reports whether an extension is support by the server. diff --git a/server/utils/smtp/smtp_test.go b/server/utils/smtp/smtp_test.go new file mode 100644 index 0000000..34f9576 --- /dev/null +++ b/server/utils/smtp/smtp_test.go @@ -0,0 +1,182 @@ +package smtp + +import ( + "errors" + "fmt" + "io" + "net" + "net/textproto" + "strings" + "testing" + "time" +) + +func TestSendMailUnsafeUsesFinalDataResponseAsDeliveryResult(t *testing.T) { + tests := []struct { + name string + dataResponse string + holdWithoutQuitReply bool + wantCode int + maxDuration time.Duration + }{ + { + name: "accepted message returns while server withholds QUIT reply", + dataResponse: "250 2.0.0 queued", + holdWithoutQuitReply: true, + maxDuration: time.Second, + }, + { + name: "DATA rejection remains a delivery failure", + dataResponse: "550 5.0.0 rejected", + wantCode: 550, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + addr, serverDone := startSMTPTransactionServer(t, tt.dataResponse, tt.holdWithoutQuitReply) + + start := time.Now() + err := SendMailUnsafe( + "", + addr, + nil, + "sender@example.com", + "example.com", + []string{"recipient@example.net"}, + []byte("From: sender@example.com\r\nTo: recipient@example.net\r\nSubject: test\r\n\r\nbody\r\n"), + ) + duration := time.Since(start) + if tt.maxDuration > 0 && duration > tt.maxDuration { + t.Fatalf("SendMailUnsafe() took %v, want at most %v", duration, tt.maxDuration) + } + + if tt.wantCode == 0 { + if err != nil { + t.Fatalf("SendMailUnsafe() = %v, want nil", err) + } + } else { + var protocolErr *textproto.Error + if !errors.As(err, &protocolErr) { + t.Fatalf("SendMailUnsafe() error type = %T, want *textproto.Error", err) + } + if protocolErr.Code != tt.wantCode { + t.Fatalf("SMTP code = %d, want %d", protocolErr.Code, tt.wantCode) + } + } + + select { + case serverErr := <-serverDone: + if serverErr != nil { + t.Fatal(serverErr) + } + case <-time.After(2 * time.Second): + t.Fatal("fake SMTP server did not finish") + } + }) + } +} + +func startSMTPTransactionServer(t *testing.T, dataResponse string, holdWithoutQuitReply bool) (string, <-chan error) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + + done := make(chan error, 1) + go func() { + defer listener.Close() + + conn, err := listener.Accept() + if err != nil { + done <- err + return + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(2 * time.Second)) + + text := textproto.NewConn(conn) + defer text.Close() + + if err := text.PrintfLine("220 test ESMTP ready"); err != nil { + done <- err + return + } + if err := expectSMTPCommand(text, "EHLO "); err != nil { + done <- err + return + } + if err := text.PrintfLine("250 test"); err != nil { + done <- err + return + } + if err := expectSMTPCommand(text, "MAIL FROM:"); err != nil { + done <- err + return + } + if err := text.PrintfLine("250 2.1.0 sender accepted"); err != nil { + done <- err + return + } + if err := expectSMTPCommand(text, "RCPT TO:"); err != nil { + done <- err + return + } + if err := text.PrintfLine("250 2.1.5 recipient accepted"); err != nil { + done <- err + return + } + if err := expectSMTPCommand(text, "DATA"); err != nil { + done <- err + return + } + if err := text.PrintfLine("354 end data with ."); err != nil { + done <- err + return + } + body, err := io.ReadAll(text.DotReader()) + if err != nil { + done <- err + return + } + if !strings.Contains(string(body), "Subject: test") { + done <- fmt.Errorf("message body was not received: %q", body) + return + } + if err := text.PrintfLine("%s", dataResponse); err != nil { + done <- err + return + } + + if holdWithoutQuitReply { + if err := expectSMTPCommand(text, "QUIT"); err != nil { + done <- err + return + } + + // Keep the server side open without sending 221. The client must close + // the accepted transaction without waiting for a reply. + buf := make([]byte, 1) + if _, err := conn.Read(buf); err != io.EOF { + done <- fmt.Errorf("waiting for client close after QUIT: %w", err) + return + } + } + done <- nil + }() + + return listener.Addr().String(), done +} + +func expectSMTPCommand(conn *textproto.Conn, prefix string) error { + line, err := conn.ReadLine() + if err != nil { + return err + } + if !strings.HasPrefix(line, prefix) { + return fmt.Errorf("SMTP command = %q, want prefix %q", line, prefix) + } + return nil +}