diff --git a/README.md b/README.md index 5a284b8..6288b2b 100644 --- a/README.md +++ b/README.md @@ -131,19 +131,17 @@ https://nlnet.nl/project/Mox/. ## Roadmap -- Improve SMTP delivery from queue - Webmail improvements - HTTP-based API for sending messages and receiving delivery feedback - Calendaring with CalDAV/iCal - More IMAP extensions (PREVIEW, WITHIN, IMPORTANT, COMPRESS=DEFLATE, CREATE-SPECIAL-USE, SAVEDATE, UNAUTHENTICATE, REPLACE, QUOTA, NOTIFY, - MULTIAPPEND, OBJECTID, MULTISEARCH) + MULTIAPPEND, OBJECTID, MULTISEARCH, THREAD, SORT) - ARC, with forwarded email from trusted source - Forwarding (to an external address) - Add special IMAP mailbox ("Queue?") that contains queued but undelivered messages, updated with IMAP flags/keywords/tags and message headers. - Sieve for filtering (for now see Rulesets in the account config) -- Expose threading through IMAP extension - Autoresponder (out of office/vacation) - OAUTH2 support, for single sign on - Privilege separation, isolating parts of the application to more restricted diff --git a/apidiff/v0.0.10.txt b/apidiff/v0.0.10.txt index a161742..fea2dd4 100644 --- a/apidiff/v0.0.10.txt +++ b/apidiff/v0.0.10.txt @@ -47,6 +47,8 @@ Below are the incompatible changes between v0.0.9 and v0.0.10, per package. # scram # smtp +- SePol7ARCFail: removed +- SePol7MissingReqTLS: removed # smtpclient - Dial: changed from func(context.Context, *golang.org/x/exp/slog.Logger, Dialer, github.com/mjl-/mox/dns.IPDomain, []net.IP, int, map[string][]net.IP, []net.IP) (net.Conn, net.IP, error) to func(context.Context, *log/slog.Logger, Dialer, github.com/mjl-/mox/dns.IPDomain, []net.IP, int, map[string][]net.IP, []net.IP) (net.Conn, net.IP, error) diff --git a/dmarcdb/eval.go b/dmarcdb/eval.go index 0d920d7..dfee9b7 100644 --- a/dmarcdb/eval.go +++ b/dmarcdb/eval.go @@ -842,7 +842,7 @@ Period: %s - %s UTC continue } - qm := queue.MakeMsg(from.Path(), rcpt.address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil) + qm := queue.MakeMsg(from.Path(), rcpt.address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil, time.Now()) // Don't try as long as regular deliveries, and stop before we would send the // delayed DSN. Though we also won't send that due to IsDMARCReport. qm.MaxAttempts = 5 @@ -997,7 +997,7 @@ Submitting-URI: %s continue } - qm := queue.MakeMsg(fromAddr.Path(), rcpt.Address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil) + qm := queue.MakeMsg(fromAddr.Path(), rcpt.Address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil, time.Now()) // Don't try as long as regular deliveries, and stop before we would send the // delayed DSN. Though we also won't send that due to IsDMARCReport. qm.MaxAttempts = 5 diff --git a/gentestdata.go b/gentestdata.go index 71b90fd..5292a89 100644 --- a/gentestdata.go +++ b/gentestdata.go @@ -233,7 +233,7 @@ Accounts: const qmsg = "From: \r\nTo: \r\nSubject: test\r\n\r\nthe message...\r\n" _, err = fmt.Fprint(mf, qmsg) xcheckf(err, "writing message") - qm := queue.MakeMsg(mailfrom, rcptto, false, false, int64(len(qmsg)), "", prefix, nil) + qm := queue.MakeMsg(mailfrom, rcptto, false, false, int64(len(qmsg)), "", prefix, nil, time.Now()) err = queue.Add(ctxbg, c.log, "test0", mf, qm) xcheckf(err, "enqueue message") diff --git a/queue/direct.go b/queue/direct.go index 836db9c..0dfe1da 100644 --- a/queue/direct.go +++ b/queue/direct.go @@ -97,43 +97,9 @@ func ConnectionCounter() int64 { return connectionCounter.Load() } -// todo: rename function, perhaps put some of the params in a delivery struct so we don't pass all the params all the time? -func fail(ctx context.Context, qlog mlog.Log, m Msg, backoff time.Duration, permanent bool, remoteMTA dsn.NameIP, secodeOpt, errmsg, firstLine string, moreLines []string) { - // todo future: when we implement relaying, we should be able to send DSNs to non-local users. and possibly specify a null mailfrom. ../rfc/5321:1503 - // todo future: when we implement relaying, and a dsn cannot be delivered, and requiretls was active, we cannot drop the message. instead deliver to local postmaster? though ../rfc/8689:383 may intend to say the dsn should be delivered without requiretls? - // todo future: when we implement smtp dsn extension, parameter RET=FULL must be disregarded for messages with REQUIRETLS. ../rfc/8689:379 - - var smtpLines []string - if firstLine != "" { - smtpLines = append([]string{firstLine}, moreLines...) - } - - if permanent || m.MaxAttempts == 0 && m.Attempts >= 8 || m.MaxAttempts > 0 && m.Attempts >= m.MaxAttempts { - qlog.Errorx("permanent failure delivering from queue", errors.New(errmsg)) - deliverDSNFailure(ctx, qlog, m, remoteMTA, secodeOpt, errmsg, smtpLines) - - if err := queueDelete(context.Background(), m.ID); err != nil { - qlog.Errorx("deleting message from queue after permanent failure", err) - } - return - } - - qup := bstore.QueryDB[Msg](context.Background(), DB) - qup.FilterID(m.ID) - if _, err := qup.UpdateNonzero(Msg{LastError: errmsg, DialedIPs: m.DialedIPs}); err != nil { - qlog.Errorx("storing delivery error", err, slog.String("deliveryerror", errmsg)) - } - - if m.Attempts == 5 { - // We've attempted deliveries at these intervals: 0, 7.5m, 15m, 30m, 1h, 2u. - // Let sender know delivery is delayed. - qlog.Errorx("temporary failure delivering from queue, sending delayed dsn", errors.New(errmsg), slog.Duration("backoff", backoff)) - - retryUntil := m.LastAttempt.Add((4 + 8 + 16) * time.Hour) - deliverDSNDelay(ctx, qlog, m, remoteMTA, secodeOpt, errmsg, smtpLines, retryUntil) - } else { - qlog.Errorx("temporary failure delivering from queue", errors.New(errmsg), slog.Duration("backoff", backoff), slog.Time("nextattempt", m.NextAttempt)) - } +type msgResp struct { + msg *Msg + resp smtpclient.Response } // Delivery by directly dialing (MX) hosts for destination domain of message. @@ -144,7 +110,7 @@ func fail(ctx context.Context, qlog mlog.Log, m Msg, backoff time.Duration, perm // domain (MTA-STS), its policy type can be empty, in which case there is no // information (e.g. internal failure). hostResults are per-host details (DANE, one // per MX target). -func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, ourHostname dns.Domain, transportName string, m Msg, backoff time.Duration) (recipientDomainResult tlsrpt.Result, hostResults []tlsrpt.Result) { +func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, ourHostname dns.Domain, transportName string, msgs []*Msg, backoff time.Duration) (recipientDomainResult tlsrpt.Result, hostResults []tlsrpt.Result) { // High-level approach: // - Resolve domain to deliver to (CNAME), and determine hosts to try to deliver to (MX) // - Get MTA-STS policy for domain (optional). If present, only deliver to its @@ -160,14 +126,18 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // TLS verification and possibly without TLS at all, ignoring recipient domain/host // MTA-STS and DANE policies. + // For convenience, we use m0 to access properties that are shared over all + // messages we are delivering. + m0 := msgs[0] + // Resolve domain and hosts to attempt delivery to. // These next-hop names are often the name under which we find MX records. The // expanded name is different from the original if the original was a CNAME, // possibly a chain. If there are no MX records, it can be an IP or the host // directly. - origNextHop := m.RecipientDomain.Domain + origNextHop := m0.RecipientDomain.Domain ctx := mox.Shutdown - haveMX, origNextHopAuthentic, expandedNextHopAuthentic, expandedNextHop, hosts, permanent, err := smtpclient.GatherDestinations(ctx, qlog.Logger, resolver, m.RecipientDomain) + haveMX, origNextHopAuthentic, expandedNextHopAuthentic, expandedNextHop, hosts, permanent, err := smtpclient.GatherDestinations(ctx, qlog.Logger, resolver, m0.RecipientDomain) if err != nil { // If this is a DNSSEC authentication error, we'll collect it for TLS reporting. // Hopefully it's a temporary misconfiguration that is solve before we try to send @@ -181,12 +151,14 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale recipientDomainResult = tlsrpt.MakeResult(tlsrpt.NoPolicyFound, origNextHop, fd) recipientDomainResult.Summary.TotalFailureSessionCount++ } - - fail(ctx, qlog, m, backoff, permanent, dsn.NameIP{}, "", err.Error(), "", nil) + if permanent { + err = smtpclient.Error{Permanent: true, Err: err} + } + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, dsn.NameIP{}, err) return } - tlsRequiredNo := m.RequireTLS != nil && !*m.RequireTLS + tlsRequiredNo := m0.RequireTLS != nil && !*m0.RequireTLS // Check for MTA-STS policy and enforce it if needed. // We must check at the original next-hop, i.e. recipient domain, not following any @@ -202,7 +174,7 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale } else { qlog.Infox("mtasts lookup temporary error, aborting delivery attempt", err, slog.Any("domain", origNextHop)) recipientDomainResult.Summary.TotalFailureSessionCount++ - fail(ctx, qlog, m, backoff, false, dsn.NameIP{}, "", err.Error(), "", nil) + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, dsn.NameIP{}, err) return } } @@ -219,9 +191,7 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // RFC 5321 does not specify a clear algorithm, but common practice is probably // ../rfc/3974:268. var remoteMTA dsn.NameIP - var firstLine, secodeOpt, errmsg string - var moreLines []string // For additional SMTP response lines, included in DSN. - permanent = false + var lastErr = errors.New("no error") // Can be smtpclient.Error. nmissingRequireTLS := 0 // todo: should make distinction between host permanently not accepting the message, and the message not being deliverable permanently. e.g. a mx host may have a size limit, or not accept 8bitmime, while another host in the list does accept the message. same for smtputf8, ../rfc/6531:555 for _, h := range hosts { @@ -244,9 +214,7 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale qlog.Info("mx host does not match mta-sts policy in mode enforce, ignoring due to tls-required-no message header", slog.Any("host", h.Domain), slog.Any("policyhosts", policyHosts)) metricTLSRequiredNoIgnored.WithLabelValues("mtastsmx").Inc() } else { - errmsg = fmt.Sprintf("mx host %s does not match enforced mta-sts policy with hosts %s", h.Domain, strings.Join(policyHosts, ",")) - firstLine = "" - moreLines = nil + lastErr = fmt.Errorf("mx host %s does not match enforced mta-sts policy with hosts %s", h.Domain, strings.Join(policyHosts, ",")) qlog.Error("mx host does not match mta-sts policy in mode enforce, skipping", slog.Any("host", h.Domain), slog.Any("policyhosts", policyHosts)) recipientDomainResult.Summary.TotalFailureSessionCount++ continue @@ -284,13 +252,16 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // opportunistic TLS. var tlsDANE bool - var badTLS, ok bool - var hostResult tlsrpt.Result - permanent, tlsDANE, badTLS, secodeOpt, remoteIP, errmsg, firstLine, moreLines, hostResult, ok = deliverHost(nqlog, resolver, dialer, ourHostname, transportName, h, enforceMTASTS, haveMX, origNextHopAuthentic, origNextHop, expandedNextHopAuthentic, expandedNextHop, &m, tlsMode, tlsPKIX, &recipientDomainResult) + msgResps := make([]*msgResp, len(msgs)) + for i := range msgs { + msgResps[i] = &msgResp{msg: msgs[i]} + } + + result := deliverHost(nqlog, resolver, dialer, ourHostname, transportName, h, enforceMTASTS, haveMX, origNextHopAuthentic, origNextHop, expandedNextHopAuthentic, expandedNextHop, msgResps, tlsMode, tlsPKIX, &recipientDomainResult) var zerotype tlsrpt.PolicyType - if hostResult.Policy.Type != zerotype { - hostResults = append(hostResults, hostResult) + if result.hostResult.Policy.Type != zerotype { + hostResults = append(hostResults, result.hostResult) } // If we had a TLS-related failure when doing TLS, and we don't have a requirement @@ -302,7 +273,7 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // We don't fall back to plain text for DMARC reports. ../rfc/7489:1768 ../rfc/7489:2683 // We queue outgoing TLS reports with tlsRequiredNo, so reports can be delivered in // case of broken TLS. - if !ok && badTLS && (!enforceMTASTS && tlsMode == smtpclient.TLSOpportunistic && !tlsDANE && !m.IsDMARCReport || tlsRequiredNo) { + if result.err != nil && errors.Is(result.err, smtpclient.ErrTLS) && (!enforceMTASTS && tlsMode == smtpclient.TLSOpportunistic && !tlsDANE && !m0.IsDMARCReport || tlsRequiredNo) { metricPlaintextFallback.Inc() if tlsRequiredNo { metricTLSRequiredNoIgnored.WithLabelValues("badtls").Inc() @@ -312,24 +283,40 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale nqlog.Info("connecting again for delivery attempt without tls", slog.Bool("enforcemtasts", enforceMTASTS), slog.Bool("tlsdane", tlsDANE), - slog.Any("requiretls", m.RequireTLS)) - permanent, _, _, secodeOpt, remoteIP, errmsg, firstLine, moreLines, _, ok = deliverHost(nqlog, resolver, dialer, ourHostname, transportName, h, enforceMTASTS, haveMX, origNextHopAuthentic, origNextHop, expandedNextHopAuthentic, expandedNextHop, &m, smtpclient.TLSSkip, false, &tlsrpt.Result{}) + slog.Any("requiretls", m0.RequireTLS)) + result = deliverHost(nqlog, resolver, dialer, ourHostname, transportName, h, enforceMTASTS, haveMX, origNextHopAuthentic, origNextHop, expandedNextHopAuthentic, expandedNextHop, msgResps, smtpclient.TLSSkip, false, &tlsrpt.Result{}) } - if ok { - nqlog.Info("delivered from queue") - if err := queueDelete(context.Background(), m.ID); err != nil { - nqlog.Errorx("deleting message from queue after delivery", err) - } - return - } remoteMTA = dsn.NameIP{Name: h.XString(false), IP: remoteIP} - if permanent { - break + if result.err != nil { + lastErr = result.err + var cerr smtpclient.Error + if errors.As(result.err, &cerr) { + if cerr.Secode == smtp.SePol7MissingReqTLS30 { + nmissingRequireTLS++ + } + if cerr.Permanent { + break + } + } + continue } - if secodeOpt == smtp.SePol7MissingReqTLS { - nmissingRequireTLS++ + + delIDs := make([]int64, len(result.delivered)) + for i, mr := range result.delivered { + mqlog := nqlog.With(slog.Int64("msgid", mr.msg.ID), slog.Any("recipient", mr.msg.Recipient())) + mqlog.Info("delivered from queue") + delIDs[i] = mr.msg.ID } + if len(delIDs) > 0 { + if err := queueDelete(context.Background(), delIDs...); err != nil { + nqlog.Errorx("deleting messages from queue after delivery", err) + } + } + for _, mr := range result.failed { + fail(ctx, nqlog, []*Msg{mr.msg}, m0.DialedIPs, backoff, remoteMTA, smtpclient.Error(mr.resp)) + } + return } // In theory, we could make a failure permanent if we didn't find any mx host @@ -344,22 +331,43 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // If we failed due to requiretls not being satisfied, make the delivery permanent. // It is unlikely the recipient domain will implement requiretls during our retry // period. Best to let the sender know immediately. - if !permanent && nmissingRequireTLS > 0 && nmissingRequireTLS == len(hosts) { + if len(hosts) > 0 && nmissingRequireTLS == len(hosts) { qlog.Info("marking delivery as permanently failed because recipient domain does not implement requiretls") - permanent = true + err := smtpclient.Error{ + Permanent: true, + Code: smtp.C554TransactionFailed, + Secode: smtp.SePol7MissingReqTLS30, + Err: fmt.Errorf("destination servers do not support requiretls"), + } + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, remoteMTA, err) + return } - fail(ctx, qlog, m, backoff, permanent, remoteMTA, secodeOpt, errmsg, firstLine, moreLines) + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, remoteMTA, lastErr) return } -// deliverHost attempts to deliver m to host. Depending on tlsMode we'll do +type deliverResult struct { + tlsDANE bool + remoteIP net.IP + hostResult tlsrpt.Result + + // If err is set, no messages were delivered but delivered and failed are still + // nil. If err is not set, delivered and always add up to all msgs requested to be + // sent. All messages can be in failed. + delivered []*msgResp + failed []*msgResp + err error +} + +// deliverHost attempts to deliver msgs to host. All msgs must have the same +// delivery requirements (e.g. requiretls). Depending on tlsMode we'll do // opportunistic or required STARTTLS or skip TLS entirely. Based on tlsPKIX we do // PKIX/WebPKI verification (for MTA-STS). If we encounter DANE records, we verify // those. If the message has a message header "TLS-Required: No", we ignore TLS // verification errors. // -// deliverHost updates m.DialedIPs, which must be saved in case of failure to +// deliverHost updates DialedIPs of msgs, which must be saved in case of failure to // deliver. // // The haveMX and next-hop-authentic fields are used to determine if DANE is @@ -369,14 +377,24 @@ func deliverDirect(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // The returned hostResult holds TLSRPT reporting results for the connection // attempt. Its policy type can be the zero value, indicating there was no finding // (e.g. internal error). -func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, ourHostname dns.Domain, transportName string, host dns.IPDomain, enforceMTASTS, haveMX, origNextHopAuthentic bool, origNextHop dns.Domain, expandedNextHopAuthentic bool, expandedNextHop dns.Domain, m *Msg, tlsMode smtpclient.TLSMode, tlsPKIX bool, recipientDomainResult *tlsrpt.Result) (permanent, tlsDANE, badTLS bool, secodeOpt string, remoteIP net.IP, errmsg, firstLine string, moreLines []string, hostResult tlsrpt.Result, ok bool) { +// +// deliverHost may send a message multiple times: if the server doesn't accept +// multiple recipients for a message. +func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, ourHostname dns.Domain, transportName string, host dns.IPDomain, enforceMTASTS, haveMX, origNextHopAuthentic bool, origNextHop dns.Domain, expandedNextHopAuthentic bool, expandedNextHop dns.Domain, msgResps []*msgResp, tlsMode smtpclient.TLSMode, tlsPKIX bool, recipientDomainResult *tlsrpt.Result) (result deliverResult) { // About attempting delivery to multiple addresses of a host: ../rfc/5321:3898 - tlsRequiredNo := m.RequireTLS != nil && !*m.RequireTLS + m0 := msgResps[0].msg + tlsRequiredNo := m0.RequireTLS != nil && !*m0.RequireTLS + var tlsDANE bool + var remoteIP net.IP + var hostResult tlsrpt.Result start := time.Now() - var deliveryResult string defer func() { + result.tlsDANE = tlsDANE + result.remoteIP = remoteIP + result.hostResult = hostResult + mode := string(tlsMode) if tlsPKIX { mode += "+mtasts" @@ -384,28 +402,31 @@ func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, if tlsDANE { mode += "+dane" } - metricDelivery.WithLabelValues(fmt.Sprintf("%d", m.Attempts), transportName, mode, deliveryResult).Observe(float64(time.Since(start)) / float64(time.Second)) - log.Debug("queue deliverhost result", + + r := deliveryResult(result.err, len(result.delivered), len(result.failed)) + d := float64(time.Since(start)) / float64(time.Second) + metricDelivery.WithLabelValues(fmt.Sprintf("%d", m0.Attempts), transportName, mode, r).Observe(d) + + log.Debugx("queue deliverhost result", result.err, slog.Any("host", host), - slog.Int("attempt", m.Attempts), + slog.Int("attempt", m0.Attempts), + slog.String("result", r), + slog.Int("delivered", len(result.delivered)), + slog.Int("failed", len(result.failed)), slog.Any("tlsmode", tlsMode), slog.Bool("tlspkix", tlsPKIX), slog.Bool("tlsdane", tlsDANE), slog.Bool("tlsrequiredno", tlsRequiredNo), - slog.Bool("permanent", permanent), - slog.Bool("badtls", badTLS), - slog.String("secodeopt", secodeOpt), - slog.String("errmsg", errmsg), - slog.Bool("ok", ok), + slog.Bool("badtls", result.err != nil && errors.Is(result.err, smtpclient.ErrTLS)), slog.Duration("duration", time.Since(start))) }() // Open message to deliver. - f, err := os.Open(m.MessagePath()) + f, err := os.Open(m0.MessagePath()) if err != nil { - return false, false, false, "", nil, fmt.Sprintf("open message file: %s", err), "", nil, hostResult, false + return deliverResult{err: fmt.Errorf("open message file: %v", err)} } - msgr := store.FileMsgReader(m.MsgPrefix, f) + msgr := store.FileMsgReader(m0.MsgPrefix, f) defer func() { err := msgr.Close() log.Check(err, "closing message after delivery attempt") @@ -423,8 +444,10 @@ func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, if host.IsDomain() { tlsHostnames = []dns.Domain{host.Domain} } - if m.DialedIPs == nil { - m.DialedIPs = map[string][]net.IP{} + for _, mr := range msgResps { + if mr.msg.DialedIPs == nil { + mr.msg.DialedIPs = map[string][]net.IP{} + } } countResultFailure := func() { @@ -433,7 +456,7 @@ func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, } metricDestinations.Inc() - authentic, expandedAuthentic, expandedHost, ips, dualstack, err := smtpclient.GatherIPs(ctx, log.Logger, resolver, host, m.DialedIPs) + authentic, expandedAuthentic, expandedHost, ips, dualstack, err := smtpclient.GatherIPs(ctx, log.Logger, resolver, host, m0.DialedIPs) destAuthentic := err == nil && authentic && origNextHopAuthentic && (!haveMX || expandedNextHopAuthentic) && host.IsDomain() if !destAuthentic { log.Debugx("not attempting verification with dane", err, slog.Bool("authentic", authentic), slog.Bool("expandedauthentic", expandedAuthentic)) @@ -530,47 +553,48 @@ func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, // todo: for requiretls, should an MTA-STS policy in mode testing be treated as good enough for requiretls? let's be strict and assume not. // todo: ../rfc/8689:276 seems to specify stricter requirements on name in certificate than DANE (which allows original recipient domain name and cname-expanded name, and hints at following CNAME for MX targets as well, allowing both their original and expanded names too). perhaps the intent was just to say the name must be validated according to the relevant specifications? // todo: for requiretls, should we allow no usable dane records with requiretls? dane allows it, but doesn't seem in spirit of requiretls, so not allowing it. - if err == nil && m.RequireTLS != nil && *m.RequireTLS && !(tlsDANE && len(daneRecords) > 0) && !enforceMTASTS { + if err == nil && m0.RequireTLS != nil && *m0.RequireTLS && !(tlsDANE && len(daneRecords) > 0) && !enforceMTASTS { log.Info("verified tls is required, but destination has no usable dane records and no mta-sts policy, canceling delivery attempt to host") metricRequireTLSUnsupported.WithLabelValues("nopolicy").Inc() // Resond with proper enhanced status code. ../rfc/8689:301 - return false, tlsDANE, false, smtp.SePol7MissingReqTLS, remoteIP, "missing required tls verification mechanism", "", nil, hostResult, false + smtpErr := smtpclient.Error{ + Code: smtp.C554TransactionFailed, + Secode: smtp.SePol7MissingReqTLS30, + Err: fmt.Errorf("missing required tls verification mechanism"), + } + return deliverResult{err: smtpErr} } // Dial the remote host given the IPs if no error yet. var conn net.Conn if err == nil { - if m.DialedIPs == nil { - m.DialedIPs = map[string][]net.IP{} - } connectionCounter.Add(1) - conn, remoteIP, err = smtpclient.Dial(ctx, log.Logger, dialer, host, ips, 25, m.DialedIPs, mox.Conf.Static.SpecifiedSMTPListenIPs) + conn, remoteIP, err = smtpclient.Dial(ctx, log.Logger, dialer, host, ips, 25, m0.DialedIPs, mox.Conf.Static.SpecifiedSMTPListenIPs) } cancel() // Set error for metrics. - var result string + var dialResult string switch { case err == nil: - result = "ok" + dialResult = "ok" case errors.Is(err, os.ErrDeadlineExceeded), errors.Is(err, context.DeadlineExceeded): - result = "timeout" + dialResult = "timeout" case errors.Is(err, context.Canceled): - result = "canceled" + dialResult = "canceled" default: - result = "error" + dialResult = "error" } - metricConnection.WithLabelValues(result).Inc() + metricConnection.WithLabelValues(dialResult).Inc() if err != nil { log.Debugx("connecting to remote smtp", err, slog.Any("host", host)) - return false, tlsDANE, false, "", remoteIP, fmt.Sprintf("dialing smtp server: %v", err), "", nil, hostResult, false + return deliverResult{err: fmt.Errorf("dialing smtp server: %v", err)} } var mailFrom string - if m.SenderLocalpart != "" || !m.SenderDomain.IsZero() { - mailFrom = m.Sender().XString(m.SMTPUTF8) + if m0.SenderLocalpart != "" || !m0.SenderDomain.IsZero() { + mailFrom = m0.Sender().XString(m0.SMTPUTF8) } - rcptTo := m.Recipient().XString(m.SMTPUTF8) // todo future: get closer to timeouts specified in rfc? ../rfc/5321:3610 log = log.With(slog.Any("remoteip", remoteIP)) @@ -605,75 +629,102 @@ func deliverHost(log mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, } mox.Connections.Unregister(conn) }() - if err == nil && m.SenderAccount != "" { + if err == nil && m0.SenderAccount != "" { // Remember the STARTTLS and REQUIRETLS support for this recipient domain. // It is used in the webmail client, to show the recipient domain security mechanisms. // We always save only the last connection we actually encountered. There may be // multiple MX hosts, perhaps only some support STARTTLS and REQUIRETLS. We may not // be accurate for the whole domain, but we're only storing a hint. rdt := store.RecipientDomainTLS{ - Domain: m.RecipientDomain.Domain.Name(), + Domain: m0.RecipientDomain.Domain.Name(), STARTTLS: sc.TLSConnectionState() != nil, RequireTLS: sc.SupportsRequireTLS(), } - if err = updateRecipientDomainTLS(ctx, log, m.SenderAccount, rdt); err != nil { + if err = updateRecipientDomainTLS(ctx, log, m0.SenderAccount, rdt); err != nil { err = fmt.Errorf("storing recipient domain tls status: %w", err) } } - if err == nil { - // SMTP session is ready. Finally try to actually deliver. - has8bit := m.Has8bit - smtputf8 := m.SMTPUTF8 - var msg io.Reader = msgr - size := m.Size - if m.DSNUTF8 != nil && sc.Supports8BITMIME() && sc.SupportsSMTPUTF8() { - has8bit = true - smtputf8 = true - size = int64(len(m.DSNUTF8)) - msg = bytes.NewReader(m.DSNUTF8) - } - err = sc.Deliver(ctx, mailFrom, rcptTo, size, msg, has8bit, smtputf8, m.RequireTLS != nil && *m.RequireTLS) - } if err != nil { - log.Infox("delivery failed", err) - } - var cerr smtpclient.Error - switch { - case err == nil: - deliveryResult = "ok" - case errors.Is(err, os.ErrDeadlineExceeded), errors.Is(err, context.DeadlineExceeded): - deliveryResult = "timeout" - case errors.Is(err, context.Canceled): - deliveryResult = "canceled" - case errors.As(err, &cerr): - deliveryResult = "temperror" - if cerr.Permanent { - deliveryResult = "permerror" + if cerr, ok := err.(smtpclient.Error); ok { + // If we are being rejected due to policy reasons on the first + // attempt and remote has both IPv4 and IPv6, we'll give it + // another try. Our first IP may be in a block list, the address for + // the other family perhaps is not. + if cerr.Permanent && m0.Attempts == 1 && dualstack && strings.HasPrefix(cerr.Secode, "7.") { + cerr.Permanent = false + } + // If server does not implement requiretls, respond with that code. ../rfc/8689:301 + if errors.Is(cerr.Err, smtpclient.ErrRequireTLSUnsupported) { + cerr.Secode = smtp.SePol7MissingReqTLS30 + metricRequireTLSUnsupported.WithLabelValues("norequiretls").Inc() + } + err = cerr } - default: - deliveryResult = "error" + return deliverResult{err: err} } - if err == nil { - return false, tlsDANE, false, "", remoteIP, "", "", nil, hostResult, true - } else if cerr, ok := err.(smtpclient.Error); ok { - // If we are being rejected due to policy reasons on the first - // attempt and remote has both IPv4 and IPv6, we'll give it - // another try. Our first IP may be in a block list, the address for - // the other family perhaps is not. - permanent := cerr.Permanent - if permanent && m.Attempts == 1 && dualstack && strings.HasPrefix(cerr.Secode, "7.") { - permanent = false + + // SMTP session is ready. Finally try to actually deliver. + has8bit := m0.Has8bit + smtputf8 := m0.SMTPUTF8 + var msg io.Reader = msgr + resetReader := msgr.Reset + size := m0.Size + if m0.DSNUTF8 != nil && sc.Supports8BITMIME() && sc.SupportsSMTPUTF8() { + has8bit = true + smtputf8 = true + size = int64(len(m0.DSNUTF8)) + msg = bytes.NewReader(m0.DSNUTF8) + resetReader = func() { + msg = bytes.NewReader(m0.DSNUTF8) } - // If server does not implement requiretls, respond with that code. ../rfc/8689:301 - secode := cerr.Secode - if errors.Is(cerr.Err, smtpclient.ErrRequireTLSUnsupported) { - secode = smtp.SePol7MissingReqTLS - metricRequireTLSUnsupported.WithLabelValues("norequiretls").Inc() - } - return permanent, tlsDANE, errors.Is(cerr, smtpclient.ErrTLS), secode, remoteIP, cerr.Error(), cerr.Line, cerr.MoreLines, hostResult, false - } else { - return false, tlsDANE, errors.Is(cerr, smtpclient.ErrTLS), "", remoteIP, err.Error(), "", nil, hostResult, false } + + // Try to deliver messages. We'll do multiple transactions if the smtp server responds + // with "too many recipients". + todo := msgResps + var delivered, failed []*msgResp + for len(todo) > 0 { + resetReader() + + // SMTP server may limit number of recipients in single transaction. + n := len(todo) + if sc.ExtLimitRcptMax > 0 && sc.ExtLimitRcptMax < len(todo) { + n = sc.ExtLimitRcptMax + } + + rcpts := make([]string, n) + for i, mr := range todo[:n] { + rcpts[i] = mr.msg.Recipient().XString(m0.SMTPUTF8) + } + + resps, err := sc.DeliverMultiple(ctx, mailFrom, rcpts, size, msg, has8bit, smtputf8, m0.RequireTLS != nil && *m0.RequireTLS) + if err != nil && len(resps) == len(msgResps) { + // If error and it applies to all recipients, return a single error. + return deliverResult{err: err} + } + var ntodo []*msgResp + for i, mr := range todo[:n] { + if err != nil { + mr.resp = smtpclient.Response{Err: err} + failed = append(failed, mr) + } else if i > 0 && (resps[i].Code == smtp.C452StorageFull || resps[i].Code == smtp.C552MailboxFull) { + ntodo = append(ntodo, mr) + } else if resps[i].Code == smtp.C250Completed { + delivered = append(delivered, mr) + } else { + failed = append(failed, mr) + } + } + todo = append(ntodo, todo[n:]...) + + // We don't take LIMITS MAILMAX into account. Multiple MAIL commands are normal in + // SMTP. If the server doesn't support that, it will likely return a temporary + // error. So at least we'll try again. This would be quite unusual. And wasteful, + // because we would immediately dial again, do the TLS handshake, EHLO, etc. Let's + // implement such a limit when we see it in practice. + } + + return deliverResult{delivered: delivered, failed: failed} } // Update (overwite) last known starttls/requiretls support for recipient domain. diff --git a/queue/dsn.go b/queue/dsn.go index b99e4c2..2c1f46d 100644 --- a/queue/dsn.go +++ b/queue/dsn.go @@ -3,8 +3,10 @@ package queue import ( "bufio" "context" + "errors" "fmt" "log/slog" + "net" "os" "strings" "time" @@ -12,12 +14,15 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/mjl-/bstore" + "github.com/mjl-/mox/dns" "github.com/mjl-/mox/dsn" "github.com/mjl-/mox/message" "github.com/mjl-/mox/mlog" "github.com/mjl-/mox/mox-" "github.com/mjl-/mox/smtp" + "github.com/mjl-/mox/smtpclient" "github.com/mjl-/mox/store" ) @@ -30,6 +35,79 @@ var ( ) ) +// todo: rename function, perhaps put some of the params in a delivery struct so we don't pass all the params all the time? +func fail(ctx context.Context, qlog mlog.Log, msgs []*Msg, dialedIPs map[string][]net.IP, backoff time.Duration, remoteMTA dsn.NameIP, err error) { + // todo future: when we implement relaying, we should be able to send DSNs to non-local users. and possibly specify a null mailfrom. ../rfc/5321:1503 + // todo future: when we implement relaying, and a dsn cannot be delivered, and requiretls was active, we cannot drop the message. instead deliver to local postmaster? though ../rfc/8689:383 may intend to say the dsn should be delivered without requiretls? + // todo future: when we implement smtp dsn extension, parameter RET=FULL must be disregarded for messages with REQUIRETLS. ../rfc/8689:379 + + m0 := msgs[0] + + var smtpLines []string + var cerr smtpclient.Error + var permanent bool + var errmsg = err.Error() + var code int + var secodeOpt string + if errors.As(err, &cerr) { + if cerr.Line != "" { + smtpLines = append([]string{cerr.Line}, cerr.MoreLines...) + } + permanent = cerr.Permanent + code = cerr.Code + secodeOpt = cerr.Secode + } + qlog = qlog.With( + slog.Bool("permanent", permanent), + slog.Int("code", code), + slog.String("secode", secodeOpt), + ) + + ids := make([]int64, len(msgs)) + for i, m := range msgs { + ids[i] = m.ID + } + + if permanent || m0.MaxAttempts == 0 && m0.Attempts >= 8 || m0.MaxAttempts > 0 && m0.Attempts >= m0.MaxAttempts { + for _, m := range msgs { + qmlog := qlog.With(slog.Int64("msgid", m.ID), slog.Any("recipient", m.Recipient())) + qmlog.Errorx("permanent failure delivering from queue", err) + deliverDSNFailure(ctx, qmlog, *m, remoteMTA, secodeOpt, errmsg, smtpLines) + } + if err := queueDelete(context.Background(), ids...); err != nil { + qlog.Errorx("deleting messages from queue after permanent failure", err) + } + return + } + + // All messages should have the same DialedIPs, so we can update them all at once. + qup := bstore.QueryDB[Msg](context.Background(), DB) + qup.FilterIDs(ids) + if _, xerr := qup.UpdateNonzero(Msg{LastError: errmsg, DialedIPs: dialedIPs}); err != nil { + qlog.Errorx("storing delivery error", xerr, slog.String("deliveryerror", errmsg)) + } + + if m0.Attempts == 5 { + // We've attempted deliveries at these intervals: 0, 7.5m, 15m, 30m, 1h, 2u. + // Let sender know delivery is delayed. + + retryUntil := m0.LastAttempt.Add((4 + 8 + 16) * time.Hour) + for _, m := range msgs { + qmlog := qlog.With(slog.Int64("msgid", m.ID), slog.Any("recipient", m.Recipient())) + qmlog.Errorx("temporary failure delivering from queue, sending delayed dsn", err, slog.Duration("backoff", backoff)) + deliverDSNDelay(ctx, qmlog, *m, remoteMTA, secodeOpt, errmsg, smtpLines, retryUntil) + } + } else { + for _, m := range msgs { + qlog.Errorx("temporary failure delivering from queue", err, + slog.Int64("msgid", m.ID), + slog.Any("recipient", m.Recipient()), + slog.Duration("backoff", backoff), + slog.Time("nextattempt", m0.NextAttempt)) + } + } +} + func deliverDSNFailure(ctx context.Context, log mlog.Log, m Msg, remoteMTA dsn.NameIP, secodeOpt, errmsg string, smtpLines []string) { const subject = "mail delivery failed" message := fmt.Sprintf(` diff --git a/queue/queue.go b/queue/queue.go index 1f3d559..efc842a 100644 --- a/queue/queue.go +++ b/queue/queue.go @@ -5,6 +5,7 @@ package queue import ( "context" + "errors" "fmt" "io" "log/slog" @@ -81,9 +82,10 @@ type Msg struct { ID int64 // A message for multiple recipients will get a BaseID that is identical to the - // first Msg.ID queued. They may be delivered in a single SMTP transaction if they - // are going to the same mail server. For messages with a single recipient, this - // field will be 0. + // first Msg.ID queued. The message contents will be identical for each recipient, + // including MsgPrefix. If other properties are identical too, including recipient + // domain, multiple Msgs may be delivered in a single SMTP transaction. For + // messages with a single recipient, this field will be 0. BaseID int64 `bstore:"index"` Queued time.Time `bstore:"default now"` @@ -215,22 +217,21 @@ func Count(ctx context.Context) (int, error) { } // MakeMsg is a convenience function that sets the commonly used fields for a Msg. -func MakeMsg(sender, recipient smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, prefix []byte, requireTLS *bool) Msg { - now := time.Now() +func MakeMsg(sender, recipient smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, prefix []byte, requireTLS *bool, next time.Time) Msg { return Msg{ SenderLocalpart: sender.Localpart, SenderDomain: sender.IPDomain, RecipientLocalpart: recipient.Localpart, RecipientDomain: recipient.IPDomain, + RecipientDomainStr: formatIPDomain(recipient.IPDomain), Has8bit: has8bit, SMTPUTF8: smtputf8, Size: size, MessageID: messageID, MsgPrefix: prefix, RequireTLS: requireTLS, - Queued: now, - NextAttempt: now, - RecipientDomainStr: formatIPDomain(recipient.IPDomain), + Queued: time.Now(), + NextAttempt: next, } } @@ -354,8 +355,8 @@ func formatIPDomain(d dns.IPDomain) string { } var ( - kick = make(chan struct{}, 1) - deliveryResult = make(chan string, 1) + kick = make(chan struct{}, 1) + deliveryResults = make(chan string, 1) ) func queuekick() { @@ -489,7 +490,7 @@ func Start(resolver dns.Resolver, done chan struct{}) error { return case <-kick: case <-timer.C: - case domain := <-deliveryResult: + case domain := <-deliveryResults: delete(busyDomains, domain) } @@ -537,7 +538,16 @@ func launchWork(log mlog.Log, resolver dns.Resolver, busyDomains map[string]stru } q.FilterNotEqual("RecipientDomainStr", doms...) } - msgs, err := q.List() + var msgs []Msg + seen := map[string]bool{} + err := q.ForEach(func(m Msg) error { + dom := m.RecipientDomainStr + if _, ok := busyDomains[dom]; !ok && !seen[dom] { + seen[dom] = true + msgs = append(msgs, m) + } + return nil + }) if err != nil { log.Errorx("querying for work in queue", err) mox.Sleep(mox.Shutdown, 1*time.Second) @@ -545,24 +555,37 @@ func launchWork(log mlog.Log, resolver dns.Resolver, busyDomains map[string]stru } for _, m := range msgs { - busyDomains[formatIPDomain(m.RecipientDomain)] = struct{}{} + busyDomains[m.RecipientDomainStr] = struct{}{} go deliver(log, resolver, m) } return len(msgs) } // Remove message from queue in database and file system. -func queueDelete(ctx context.Context, msgID int64) error { - if err := DB.Delete(ctx, &Msg{ID: msgID}); err != nil { +func queueDelete(ctx context.Context, msgIDs ...int64) error { + err := DB.Write(ctx, func(tx *bstore.Tx) error { + for _, id := range msgIDs { + if err := tx.Delete(&Msg{ID: id}); err != nil { + return err + } + } + return nil + }) + if err != nil { return err } // If removing from database fails, we'll also leave the file in the file system. - p := mox.DataDirPath(filepath.Join("queue", store.MessagePath(msgID))) - if err := os.Remove(p); err != nil { - return fmt.Errorf("removing queue message from file system: %v", err) + var errs []string + for _, id := range msgIDs { + p := mox.DataDirPath(filepath.Join("queue", store.MessagePath(id))) + if err := os.Remove(p); err != nil { + errs = append(errs, fmt.Sprintf("%s: %v", p, err)) + } + } + if len(errs) > 0 { + return fmt.Errorf("removing message files from queue: %s", strings.Join(errs, "; ")) } - return nil } @@ -572,17 +595,16 @@ func queueDelete(ctx context.Context, msgID int64) error { func deliver(log mlog.Log, resolver dns.Resolver, m Msg) { ctx := mox.Shutdown - qlog := log.WithCid(mox.Cid()).With(slog.Any("from", m.Sender()), - slog.Any("recipient", m.Recipient()), - slog.Int("attempts", m.Attempts), - slog.Int64("msgid", m.ID)) + qlog := log.WithCid(mox.Cid()).With( + slog.Any("from", m.Sender()), + slog.Int("attempts", m.Attempts)) defer func() { - deliveryResult <- formatIPDomain(m.RecipientDomain) + deliveryResults <- formatIPDomain(m.RecipientDomain) x := recover() if x != nil { - qlog.Error("deliver panic", slog.Any("panic", x)) + qlog.Error("deliver panic", slog.Any("panic", x), slog.Int64("msgid", m.ID), slog.Any("recipient", m.Recipient())) debug.PrintStack() metrics.PanicInc(metrics.Queue) } @@ -600,6 +622,7 @@ func deliver(log mlog.Log, resolver dns.Resolver, m Msg) { backoff *= time.Duration(2) } m.Attempts++ + origNextAttempt := m.NextAttempt now := time.Now() m.LastAttempt = &now m.NextAttempt = now.Add(backoff) @@ -607,26 +630,30 @@ func deliver(log mlog.Log, resolver dns.Resolver, m Msg) { qup.FilterID(m.ID) update := Msg{Attempts: m.Attempts, NextAttempt: m.NextAttempt, LastAttempt: m.LastAttempt} if _, err := qup.UpdateNonzero(update); err != nil { - qlog.Errorx("storing delivery attempt", err) + qlog.Errorx("storing delivery attempt", err, slog.Int64("msgid", m.ID), slog.Any("recipient", m.Recipient())) return } - // Find route for transport to use for delivery attempt. - var transport config.Transport - var transportName string - if m.Transport != "" { - var ok bool - transport, ok = mox.Conf.Static.Transports[m.Transport] - if !ok { - var remoteMTA dsn.NameIP // Zero value, will not be included in DSN. ../rfc/3464:1027 - fail(ctx, qlog, m, backoff, false, remoteMTA, "", fmt.Sprintf("cannot find transport %q", m.Transport), "", nil) - return + resolveTransport := func(mm Msg) (string, config.Transport, bool) { + if mm.Transport != "" { + transport, ok := mox.Conf.Static.Transports[mm.Transport] + if !ok { + return "", config.Transport{}, false + } + return mm.Transport, transport, ok } - transportName = m.Transport - } else { - route := findRoute(m.Attempts-1, m) - transport = route.ResolvedTransport - transportName = route.Transport + route := findRoute(mm.Attempts, mm) + return route.Transport, route.ResolvedTransport, true + } + + // Find route for transport to use for delivery attempt. + m.Attempts-- + transportName, transport, transportOK := resolveTransport(m) + m.Attempts++ + if !transportOK { + var remoteMTA dsn.NameIP // Zero value, will not be included in DSN. ../rfc/3464:1027 + fail(ctx, qlog, []*Msg{&m}, m.DialedIPs, backoff, remoteMTA, fmt.Errorf("cannot find transport %q", m.Transport)) + return } if transportName != "" { @@ -634,8 +661,62 @@ func deliver(log mlog.Log, resolver dns.Resolver, m Msg) { qlog.Debug("delivering with transport") } + // Attempt to gather more recipients for this identical message, only with the same + // recipient domain, and under the same conditions (recipientdomain, attempts, + // requiretls, transport). ../rfc/5321:3759 + msgs := []*Msg{&m} + if m.BaseID != 0 { + err := DB.Write(mox.Shutdown, func(tx *bstore.Tx) error { + q := bstore.QueryTx[Msg](tx) + q.FilterNonzero(Msg{BaseID: m.BaseID, RecipientDomainStr: m.RecipientDomainStr, Attempts: m.Attempts - 1}) + q.FilterNotEqual("ID", m.ID) + q.FilterLessEqual("NextAttempt", origNextAttempt) + err := q.ForEach(func(xm Msg) error { + mrtls := m.RequireTLS != nil + xmrtls := xm.RequireTLS != nil + if mrtls != xmrtls || mrtls && *m.RequireTLS != *xm.RequireTLS { + return nil + } + tn, _, ok := resolveTransport(xm) + if ok && tn == transportName { + msgs = append(msgs, &xm) + } + return nil + }) + if err != nil { + return fmt.Errorf("looking up more recipients: %v", err) + } + + // Mark these additional messages as attempted too. + for _, mm := range msgs[1:] { + mm.Attempts++ + mm.NextAttempt = m.NextAttempt + mm.LastAttempt = m.LastAttempt + if err := tx.Update(mm); err != nil { + return fmt.Errorf("updating more message recipients for smtp transaction: %v", err) + } + } + return nil + }) + if err != nil { + qlog.Errorx("error finding more recipients for message, will attempt to send to single recipient", err) + msgs = msgs[:1] + } + } + if len(msgs) > 1 { + ids := make([]int64, len(msgs)) + rcpts := make([]smtp.Path, len(msgs)) + for i, m := range msgs { + ids[i] = m.ID + rcpts[i] = m.Recipient() + } + qlog.Debug("delivering to multiple recipients", slog.Any("msgids", ids), slog.Any("recipients", rcpts)) + } else { + qlog.Debug("delivering to single recipient", slog.Any("msgid", m.ID), slog.Any("recipient", m.Recipient())) + } + // We gather TLS connection successes and failures during delivery, and we store - // them in tlsrptb. Every 24 hours we send an email with a report to the recipient + // them in tlsrptdb. Every 24 hours we send an email with a report to the recipient // domains that opt in via a TLSRPT DNS record. For us, the tricky part is // collecting all reporting information. We've got several TLS modes // (opportunistic, DANE and/or MTA-STS (PKIX), overrides due to Require TLS). @@ -719,28 +800,28 @@ func deliver(log mlog.Log, resolver dns.Resolver, m Msg) { var dialer smtpclient.Dialer = &net.Dialer{} if transport.Submissions != nil { - deliverSubmit(qlog, resolver, dialer, m, backoff, transportName, transport.Submissions, true, 465) + deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submissions, true, 465) } else if transport.Submission != nil { - deliverSubmit(qlog, resolver, dialer, m, backoff, transportName, transport.Submission, false, 587) + deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submission, false, 587) } else if transport.SMTP != nil { // todo future: perhaps also gather tlsrpt results for submissions. - deliverSubmit(qlog, resolver, dialer, m, backoff, transportName, transport.SMTP, false, 25) + deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.SMTP, false, 25) } else { ourHostname := mox.Conf.Static.HostnameDomain if transport.Socks != nil { socksdialer, err := proxy.SOCKS5("tcp", transport.Socks.Address, nil, &net.Dialer{}) if err != nil { - fail(ctx, qlog, m, backoff, false, dsn.NameIP{}, "", fmt.Sprintf("socks dialer: %v", err), "", nil) + fail(ctx, qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer: %v", err)) return } else if d, ok := socksdialer.(smtpclient.Dialer); !ok { - fail(ctx, qlog, m, backoff, false, dsn.NameIP{}, "", "socks dialer is not a contextdialer", "", nil) + fail(ctx, qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer is not a contextdialer")) return } else { dialer = d } ourHostname = transport.Socks.Hostname } - recipientDomainResult, hostResults = deliverDirect(qlog, resolver, dialer, ourHostname, transportName, m, backoff) + recipientDomainResult, hostResults = deliverDirect(qlog, resolver, dialer, ourHostname, transportName, msgs, backoff) } } @@ -782,3 +863,30 @@ func routeMatchDomain(l []string, d dns.Domain) bool { } return false } + +// Returns string representing delivery result for err, and number of delivered and +// failed messages. +// +// Values: ok, okpartial, timeout, canceled, temperror, permerror, error. +func deliveryResult(err error, delivered, failed int) string { + var cerr smtpclient.Error + switch { + case err == nil: + if delivered == 0 { + return "error" + } else if failed > 0 { + return "okpartial" + } + return "ok" + case errors.Is(err, os.ErrDeadlineExceeded), errors.Is(err, context.DeadlineExceeded): + return "timeout" + case errors.Is(err, context.Canceled): + return "canceled" + case errors.As(err, &cerr): + if cerr.Permanent { + return "permerror" + } + return "temperror" + } + return "error" +} diff --git a/queue/queue_test.go b/queue/queue_test.go index f28e151..55dbfec 100644 --- a/queue/queue_test.go +++ b/queue/queue_test.go @@ -117,11 +117,11 @@ func TestQueue(t *testing.T) { var qm Msg - qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil) + qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) err = Add(ctxbg, pkglog, "mjl", mf, qm) tcheck(t, err, "add message to queue for delivery") - qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil) + qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) err = Add(ctxbg, pkglog, "mjl", mf, qm) tcheck(t, err, "add message to queue for delivery") @@ -162,7 +162,10 @@ func TestQueue(t *testing.T) { "mail.mox.example.": {"127.0.0.1"}, "submission.example.": {"127.0.0.1"}, }, - MX: map[string][]*net.MX{"mox.example.": {{Host: "mail.mox.example", Pref: 10}}}, + MX: map[string][]*net.MX{ + "mox.example.": {{Host: "mail.mox.example", Pref: 10}}, + "other.example.": {{Host: "mail.mox.example", Pref: 10}}, + }, } // Override dial function. We'll make connecting fail for now. dialed := make(chan struct{}, 1) @@ -199,7 +202,7 @@ func TestQueue(t *testing.T) { case <-timer.C: t.Fatalf("no dial within 1s") } - <-deliveryResult // Deliver sends here. + <-deliveryResults // Deliver sends here. _, err = OpenMessage(ctxbg, msg.ID+1) if err != bstore.ErrAbsent { @@ -227,12 +230,13 @@ func TestQueue(t *testing.T) { smtpdone := make(chan struct{}) - fakeSMTPServer := func(server net.Conn) { + nfakeSMTPServer := func(server net.Conn, rcpts, ntx int, onercpt bool, extensions []string) { defer func() { smtpdone <- struct{}{} }() - // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to cyclic dependencies. + // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to + // cyclic dependencies. fmt.Fprintf(server, "220 mail.mox.example\r\n") br := bufio.NewReader(server) @@ -247,16 +251,88 @@ func TestQueue(t *testing.T) { } readline("ehlo") - writeline("250 mail.mox.example") + writeline("250-mail.mox.example") + for _, ext := range extensions { + writeline("250-" + ext) + } + writeline("250 pipelining") + for tx := 0; tx < ntx; tx++ { + readline("mail") + writeline("250 ok") + for i := 0; i < rcpts; i++ { + readline("rcpt") + if onercpt && i > 0 { + writeline("552 ok") + } else { + writeline("250 ok") + } + } + readline("data") + writeline("354 continue") + reader := smtp.NewDataReader(br) + io.Copy(io.Discard, reader) + writeline("250 ok") + } + readline("quit") + writeline("221 ok") + } + fakeSMTPServer := func(server net.Conn) { + nfakeSMTPServer(server, 1, 1, false, nil) + } + fakeSMTPServer2Rcpts := func(server net.Conn) { + nfakeSMTPServer(server, 2, 1, false, nil) + } + fakeSMTPServerLimitRcpt1 := func(server net.Conn) { + nfakeSMTPServer(server, 1, 2, false, []string{"LIMITS RCPTMAX=1"}) + } + // Server that returns an error after first recipient. We expect another + // transaction to deliver the second message. + fakeSMTPServerRcpt1 := func(server net.Conn) { + defer func() { + smtpdone <- struct{}{} + }() + + // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to + // cyclic dependencies. + fmt.Fprintf(server, "220 mail.mox.example\r\n") + br := bufio.NewReader(server) + + readline := func(cmd string) { + line, err := br.ReadString('\n') + if err == nil && !strings.HasPrefix(strings.ToLower(line), cmd) { + panic(fmt.Sprintf("unexpected line %q, expected %q", line, cmd)) + } + } + writeline := func(s string) { + fmt.Fprintf(server, "%s\r\n", s) + } + + readline("ehlo") + writeline("250-mail.mox.example") + writeline("250 pipelining") + + readline("mail") + writeline("250 ok") + readline("rcpt") + writeline("250 ok") + readline("rcpt") + writeline("552 ok") + readline("data") + writeline("354 continue") + reader := smtp.NewDataReader(br) + io.Copy(io.Discard, reader) + writeline("250 ok") + readline("mail") writeline("250 ok") readline("rcpt") writeline("250 ok") readline("data") writeline("354 continue") - reader := smtp.NewDataReader(br) + reader = smtp.NewDataReader(br) io.Copy(io.Discard, reader) writeline("250 ok") + readline("quit") writeline("221 ok") } @@ -271,7 +347,8 @@ func TestQueue(t *testing.T) { attempt++ - // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to cyclic dependencies. + // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to + // cyclic dependencies. fmt.Fprintf(server, "220 mail.mox.example\r\n") br := bufio.NewReader(server) @@ -326,12 +403,13 @@ func TestQueue(t *testing.T) { return makeFakeSMTPSTARTTLSServer(&tls.Config{MaxVersion: tls.VersionTLS10, Certificates: []tls.Certificate{moxCert}}, 1, requiretls) } - fakeSubmitServer := func(server net.Conn) { + nfakeSubmitServer := func(server net.Conn, nrcpt int) { defer func() { smtpdone <- struct{}{} }() - // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to cyclic dependencies. + // We do a minimal fake smtp server. We cannot import smtpserver.Serve due to + // cyclic dependencies. fmt.Fprintf(server, "220 mail.mox.example\r\n") br := bufio.NewReader(server) br.ReadString('\n') // Should be EHLO. @@ -341,8 +419,10 @@ func TestQueue(t *testing.T) { fmt.Fprintf(server, "235 2.7.0 auth ok\r\n") br.ReadString('\n') // Should be MAIL FROM. fmt.Fprintf(server, "250 ok\r\n") - br.ReadString('\n') // Should be RCPT TO. - fmt.Fprintf(server, "250 ok\r\n") + for i := 0; i < nrcpt; i++ { + br.ReadString('\n') // Should be RCPT TO. + fmt.Fprintf(server, "250 ok\r\n") + } br.ReadString('\n') // Should be DATA. fmt.Fprintf(server, "354 continue\r\n") reader := smtp.NewDataReader(br) @@ -351,6 +431,12 @@ func TestQueue(t *testing.T) { br.ReadString('\n') // Should be QUIT. fmt.Fprintf(server, "221 ok\r\n") } + fakeSubmitServer := func(server net.Conn) { + nfakeSubmitServer(server, 1) + } + fakeSubmitServer2Rcpts := func(server net.Conn) { + nfakeSubmitServer(server, 2) + } testQueue := func(expectDSN bool, fakeServer func(conn net.Conn)) bool { t.Helper() @@ -367,10 +453,7 @@ func TestQueue(t *testing.T) { // Setting up a pipe. We'll start a fake smtp server on the server-side. And return the // client-side to the invocation dial, for the attempted delivery from the queue. server, client := net.Pipe() - for _, c := range pipes { - c.Close() - } - pipes = []net.Conn{server, client} + pipes = append(pipes, server, client) go fakeServer(server) _, wasNetDialer = dialer.(*net.Dialer) @@ -427,7 +510,7 @@ func TestQueue(t *testing.T) { case <-timer.C: t.Fatalf("no dial within 1s") } - <-deliveryResult // Deliver sends here. + <-deliveryResults // Deliver sends here. } launchWork(pkglog, resolver, map[string]struct{}{}) @@ -449,9 +532,48 @@ func TestQueue(t *testing.T) { t.Fatalf("expected net.Dialer as dialer") } + // Single delivery to two recipients at same domain, expecting single connection + // and single transaction. + qm0 := MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) + qml := []Msg{qm0, qm0} // Same NextAttempt. + err = Add(ctxbg, pkglog, "mjl", mf, qml...) + tcheck(t, err, "add messages to queue for delivery") + testDeliver(fakeSMTPServer2Rcpts) + + // Single enqueue to two recipients at different domain, expecting two connections. + otheraddr, _ := smtp.ParseAddress("mjl@other.example") + otherpath := otheraddr.Path() + t0 := time.Now() + qml = []Msg{ + MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, t0), + MakeMsg(path, otherpath, false, false, int64(len(testmsg)), "", nil, nil, t0), + } + err = Add(ctxbg, pkglog, "mjl", mf, qml...) + tcheck(t, err, "add messages to queue for delivery") + conns := ConnectionCounter() + testDeliver(fakeSMTPServer) + nconns := ConnectionCounter() + if nconns != conns+2 { + t.Errorf("saw %d connections, expected 2", nconns-conns) + } + + // Single enqueue with two recipients at same domain, but with smtp server that has + // LIMITS RCPTMAX=1, so we expect a single connection with two transactions. + qml = []Msg{qm0, qm0} + err = Add(ctxbg, pkglog, "mjl", mf, qml...) + tcheck(t, err, "add messages to queue for delivery") + testDeliver(fakeSMTPServerLimitRcpt1) + + // Single enqueue with two recipients at same domain, but smtp server sends 552 for + // 2nd recipient, so we expect a single connection with two transactions. + qml = []Msg{qm0, qm0} + err = Add(ctxbg, pkglog, "mjl", mf, qml...) + tcheck(t, err, "add messages to queue for delivery") + testDeliver(fakeSMTPServerRcpt1) + // Add a message to be delivered with submit because of its route. topath := smtp.Path{Localpart: "mjl", IPDomain: dns.IPDomain{Domain: dns.Domain{ASCII: "submit.example"}}} - qm = MakeMsg(path, topath, false, false, int64(len(testmsg)), "", nil, nil) + qm = MakeMsg(path, topath, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) err = Add(ctxbg, pkglog, "mjl", mf, qm) tcheck(t, err, "add message to queue for delivery") wasNetDialer = testDeliver(fakeSubmitServer) @@ -459,8 +581,17 @@ func TestQueue(t *testing.T) { t.Fatalf("expected net.Dialer as dialer") } + // Two messages for submission. + qml = []Msg{qm, qm} + err = Add(ctxbg, pkglog, "mjl", mf, qml...) + tcheck(t, err, "add messages to queue for delivery") + wasNetDialer = testDeliver(fakeSubmitServer2Rcpts) + if !wasNetDialer { + t.Fatalf("expected net.Dialer as dialer") + } + // Add a message to be delivered with submit because of explicitly configured transport, that uses TLS. - qml := []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") transportSubmitTLS := "submittls" @@ -509,7 +640,7 @@ func TestQueue(t *testing.T) { } // Add a message to be delivered with socks. - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") transportSocks := "socks" @@ -525,7 +656,7 @@ func TestQueue(t *testing.T) { // Add message to be delivered with opportunistic TLS verification. clearTLSResults(t) - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -539,7 +670,7 @@ func TestQueue(t *testing.T) { // Test fallback to plain text with TLS handshake fails. clearTLSResults(t) - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -559,7 +690,7 @@ func TestQueue(t *testing.T) { {Usage: adns.TLSAUsageDANEEE, Selector: adns.TLSASelectorSPKI, MatchType: adns.TLSAMatchTypeFull, CertAssoc: moxCert.Leaf.RawSubjectPublicKeyInfo}, }, } - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -580,7 +711,7 @@ func TestQueue(t *testing.T) { // Add message to be delivered with verified TLS and REQUIRETLS. yes := true - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -597,7 +728,7 @@ func TestQueue(t *testing.T) { {}, }, } - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -618,7 +749,7 @@ func TestQueue(t *testing.T) { {Usage: adns.TLSAUsageDANEEE, Selector: adns.TLSASelectorSPKI, MatchType: adns.TLSAMatchTypeFull, CertAssoc: make([]byte, sha256.Size)}, }, } - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -640,7 +771,7 @@ func TestQueue(t *testing.T) { // Check that message is delivered with TLS-Required: No and non-matching DANE record. no := false - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &no)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &no, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -651,7 +782,7 @@ func TestQueue(t *testing.T) { testDeliver(fakeSMTPSTARTTLSServer) // Check that message is delivered with TLS-Required: No and bad TLS, falling back to plain text. - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &no)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &no, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -662,7 +793,7 @@ func TestQueue(t *testing.T) { testDeliver(makeBadFakeSMTPSTARTTLSServer(true)) // Add message with requiretls that fails immediately due to no REQUIRETLS support in all servers. - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -677,7 +808,7 @@ func TestQueue(t *testing.T) { resolver.TLSA = nil // Add message with requiretls that fails immediately due to no verification policy for recipient domain. - qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes)} + qml = []Msg{MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, &yes, time.Now())} err = Add(ctxbg, pkglog, "mjl", mf, qml...) tcheck(t, err, "add message to queue for delivery") n, err = Kick(ctxbg, qml[0].ID, "", "", nil) @@ -692,7 +823,7 @@ func TestQueue(t *testing.T) { }) // Add another message that we'll fail to deliver entirely. - qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil) + qm = MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) err = Add(ctxbg, pkglog, "mjl", mf, qm) tcheck(t, err, "add message to queue for delivery") @@ -746,7 +877,7 @@ func TestQueue(t *testing.T) { defer comm.Unregister() for i := 1; i < 8; i++ { - go func() { <-deliveryResult }() // Deliver sends here. + go func() { <-deliveryResults }() // Deliver sends here. if i == 4 { resolver.AllAuthentic = true resolver.TLSA = map[string][]adns.TLSA{ @@ -781,7 +912,7 @@ func TestQueue(t *testing.T) { } // Trigger final failure. - go func() { <-deliveryResult }() // Deliver sends here. + go func() { <-deliveryResults }() // Deliver sends here. deliver(pkglog, resolver, msg) err = DB.Get(ctxbg, &msg) if err != bstore.ErrAbsent { @@ -883,7 +1014,7 @@ func TestQueueStart(t *testing.T) { mf := prepareFile(t) defer os.Remove(mf.Name()) defer mf.Close() - qm := MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil) + qm := MakeMsg(path, path, false, false, int64(len(testmsg)), "", nil, nil, time.Now()) err = Add(ctxbg, pkglog, "mjl", mf, qm) tcheck(t, err, "add message to queue for delivery") checkDialed(true) diff --git a/queue/submit.go b/queue/submit.go index 06c5a69..e081cc9 100644 --- a/queue/submit.go +++ b/queue/submit.go @@ -28,9 +28,12 @@ import ( // deliver via another SMTP server, e.g. relaying to a smart host, possibly // with authentication (submission). -func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, m Msg, backoff time.Duration, transportName string, transport *config.TransportSMTP, dialTLS bool, defaultPort int) { +func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Dialer, msgs []*Msg, backoff time.Duration, transportName string, transport *config.TransportSMTP, dialTLS bool, defaultPort int) { // todo: configurable timeouts + // For convenience, all messages share the same relevant values. + m0 := msgs[0] + port := transport.Port if port == 0 { port = defaultPort @@ -47,22 +50,27 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale tlsMode = smtpclient.TLSSkip tlsPKIX = false } + + // Prepare values for logging/metrics. They are updated for various error + // conditions later on. start := time.Now() - var deliveryResult string - var permanent bool - var secodeOpt string - var errmsg string - var success bool + var submiterr error // Of whole operation, nil for partial failure/success. + var delivered int + failed := len(msgs) // Reset and updated after smtp transaction. defer func() { - metricDelivery.WithLabelValues(fmt.Sprintf("%d", m.Attempts), transportName, string(tlsMode), deliveryResult).Observe(float64(time.Since(start)) / float64(time.Second)) - qlog.Debug("queue deliversubmit result", + r := deliveryResult(submiterr, delivered, failed) + d := float64(time.Since(start)) / float64(time.Second) + metricDelivery.WithLabelValues(fmt.Sprintf("%d", m0.Attempts), transportName, string(tlsMode), r).Observe(d) + + qlog.Debugx("queue deliversubmit result", submiterr, slog.Any("host", transport.DNSHost), slog.Int("port", port), - slog.Int("attempt", m.Attempts), - slog.Bool("permanent", permanent), - slog.String("secodeopt", secodeOpt), - slog.String("errmsg", errmsg), - slog.Bool("ok", success), + slog.Int("attempt", m0.Attempts), + slog.String("result", r), + slog.Int("delivered", delivered), + slog.Int("failed", failed), + slog.Any("tlsmode", tlsMode), + slog.Bool("tlspkix", tlsPKIX), slog.Duration("duration", time.Since(start))) }() @@ -75,25 +83,28 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale // If submit was done with REQUIRETLS extension for SMTP, we must verify TLS // certificates. If our submission connection is not configured that way, abort. - requireTLS := m.RequireTLS != nil && *m.RequireTLS + requireTLS := m0.RequireTLS != nil && *m0.RequireTLS if requireTLS && (tlsMode != smtpclient.TLSRequiredStartTLS && tlsMode != smtpclient.TLSImmediate || !tlsPKIX) { - errmsg = fmt.Sprintf("transport %s: message requires verified tls but transport does not verify tls", transportName) - fail(ctx, qlog, m, backoff, true, dsn.NameIP{}, smtp.SePol7MissingReqTLS, errmsg, "", nil) + submiterr = smtpclient.Error{ + Permanent: true, + Code: smtp.C554TransactionFailed, + Secode: smtp.SePol7MissingReqTLS30, + Err: fmt.Errorf("transport %s: message requires verified tls but transport does not verify tls", transportName), + } + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, dsn.NameIP{}, submiterr) return } dialctx, dialcancel := context.WithTimeout(ctx, 30*time.Second) defer dialcancel() - if m.DialedIPs == nil { - m.DialedIPs = map[string][]net.IP{} + if msgs[0].DialedIPs == nil { + msgs[0].DialedIPs = map[string][]net.IP{} + m0 = msgs[0] } - _, _, _, ips, _, err := smtpclient.GatherIPs(dialctx, qlog.Logger, resolver, dns.IPDomain{Domain: transport.DNSHost}, m.DialedIPs) + _, _, _, ips, _, err := smtpclient.GatherIPs(dialctx, qlog.Logger, resolver, dns.IPDomain{Domain: transport.DNSHost}, m0.DialedIPs) var conn net.Conn if err == nil { - if m.DialedIPs == nil { - m.DialedIPs = map[string][]net.IP{} - } - conn, _, err = smtpclient.Dial(dialctx, qlog.Logger, dialer, dns.IPDomain{Domain: transport.DNSHost}, ips, port, m.DialedIPs, mox.Conf.Static.SpecifiedSMTPListenIPs) + conn, _, err = smtpclient.Dial(dialctx, qlog.Logger, dialer, dns.IPDomain{Domain: transport.DNSHost}, ips, port, m0.DialedIPs, mox.Conf.Static.SpecifiedSMTPListenIPs) } addr := net.JoinHostPort(transport.Host, fmt.Sprintf("%d", port)) var result string @@ -114,8 +125,8 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale qlog.Check(err, "closing connection") } qlog.Errorx("dialing for submission", err, slog.String("remote", addr)) - errmsg = fmt.Sprintf("transport %s: dialing %s for submission: %v", transportName, addr, err) - fail(ctx, qlog, m, backoff, false, dsn.NameIP{}, "", errmsg, "", nil) + submiterr = fmt.Errorf("transport %s: dialing %s for submission: %w", transportName, addr, err) + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, dsn.NameIP{}, submiterr) return } dialcancel() @@ -165,13 +176,14 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale if err != nil { smtperr, ok := err.(smtpclient.Error) var remoteMTA dsn.NameIP + submiterr = fmt.Errorf("transport %s: establishing smtp session with %s for submission: %w", transportName, addr, err) if ok { remoteMTA.Name = transport.Host + smtperr.Err = submiterr + submiterr = smtperr } - qlog.Errorx("establishing smtp session for submission", err, slog.String("remote", addr)) - errmsg = fmt.Sprintf("transport %s: establishing smtp session with %s for submission: %v", transportName, addr, err) - secodeOpt = smtperr.Secode - fail(ctx, qlog, m, backoff, false, remoteMTA, secodeOpt, errmsg, smtperr.Line, smtperr.MoreLines) + qlog.Errorx("establishing smtp session for submission", submiterr, slog.String("remote", addr)) + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, remoteMTA, submiterr) return } defer func() { @@ -183,23 +195,23 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale var msgr io.ReadCloser var size int64 var req8bit, reqsmtputf8 bool - if len(m.DSNUTF8) > 0 && client.SupportsSMTPUTF8() { - msgr = io.NopCloser(bytes.NewReader(m.DSNUTF8)) + if len(m0.DSNUTF8) > 0 && client.SupportsSMTPUTF8() { + msgr = io.NopCloser(bytes.NewReader(m0.DSNUTF8)) reqsmtputf8 = true - size = int64(len(m.DSNUTF8)) + size = int64(len(m0.DSNUTF8)) } else { - req8bit = m.Has8bit // todo: not require this, but just try to submit? - size = m.Size + req8bit = m0.Has8bit // todo: not require this, but just try to submit? + size = m0.Size - p := m.MessagePath() + p := m0.MessagePath() f, err := os.Open(p) if err != nil { qlog.Errorx("opening message for delivery", err, slog.String("remote", addr), slog.String("path", p)) - errmsg = fmt.Sprintf("transport %s: opening message file for submission: %v", transportName, err) - fail(ctx, qlog, m, backoff, false, dsn.NameIP{}, "", errmsg, "", nil) + submiterr = fmt.Errorf("transport %s: opening message file for submission: %w", transportName, err) + fail(ctx, qlog, msgs, m0.DialedIPs, backoff, dsn.NameIP{}, submiterr) return } - msgr = store.FileMsgReader(m.MsgPrefix, f) + msgr = store.FileMsgReader(m0.MsgPrefix, f) defer func() { err := msgr.Close() qlog.Check(err, "closing message after delivery attempt") @@ -208,42 +220,48 @@ func deliverSubmit(qlog mlog.Log, resolver dns.Resolver, dialer smtpclient.Diale deliverctx, delivercancel := context.WithTimeout(context.Background(), time.Duration(60+size/(1024*1024))*time.Second) defer delivercancel() - err = client.Deliver(deliverctx, m.Sender().String(), m.Recipient().String(), size, msgr, req8bit, reqsmtputf8, requireTLS) - if err != nil { - qlog.Infox("delivery failed", err) + rcpts := make([]string, len(msgs)) + for i, m := range msgs { + rcpts[i] = m.Recipient().String() } - var cerr smtpclient.Error - switch { - case err == nil: - deliveryResult = "ok" - success = true - case errors.Is(err, os.ErrDeadlineExceeded), errors.Is(err, context.DeadlineExceeded): - deliveryResult = "timeout" - case errors.Is(err, context.Canceled): - deliveryResult = "canceled" - case errors.As(err, &cerr): - deliveryResult = "temperror" - if cerr.Permanent { - deliveryResult = "permerror" + rcptErrs, submiterr := client.DeliverMultiple(deliverctx, m0.Sender().String(), rcpts, size, msgr, req8bit, reqsmtputf8, requireTLS) + if submiterr != nil { + qlog.Infox("smtp transaction for delivery failed", submiterr) + } + failed = 0 // Reset, we are looking at the SMTP results below. + var delIDs []int64 + for i, m := range msgs { + qmlog := qlog.With( + slog.Int64("msgid", m.ID), + slog.Any("recipient", m.Recipient())) + + err := submiterr + if err == nil && len(rcptErrs) > i { + if rcptErrs[i].Code != smtp.C250Completed { + err = smtpclient.Error(rcptErrs[i]) + } } - default: - deliveryResult = "error" - } - if err != nil { - smtperr, ok := err.(smtpclient.Error) - var remoteMTA dsn.NameIP - if ok { - remoteMTA.Name = transport.Host + if err != nil { + smtperr, ok := err.(smtpclient.Error) + err = fmt.Errorf("transport %s: submitting message to %s: %w", transportName, addr, err) + var remoteMTA dsn.NameIP + if ok { + remoteMTA.Name = transport.Host + smtperr.Err = err + err = smtperr + } + qmlog.Errorx("submitting message", err, slog.String("remote", addr)) + fail(ctx, qmlog, []*Msg{m}, m0.DialedIPs, backoff, remoteMTA, err) + failed++ + } else { + delIDs = append(delIDs, m.ID) + qmlog.Info("delivered from queue with transport") + delivered++ } - qlog.Errorx("submitting email", err, slog.String("remote", addr)) - permanent = smtperr.Permanent - secodeOpt = smtperr.Secode - errmsg = fmt.Sprintf("transport %s: submitting email to %s: %v", transportName, addr, err) - fail(ctx, qlog, m, backoff, permanent, remoteMTA, secodeOpt, errmsg, smtperr.Line, smtperr.MoreLines) - return } - qlog.Info("delivered from queue with transport") - if err := queueDelete(context.Background(), m.ID); err != nil { - qlog.Errorx("deleting message from queue after delivery", err) + if len(delIDs) > 0 { + if err := queueDelete(context.Background(), delIDs...); err != nil { + qlog.Errorx("deleting message from queue after delivery", err) + } } } diff --git a/rfc/index.txt b/rfc/index.txt index 5a04ccc..e59ea98 100644 --- a/rfc/index.txt +++ b/rfc/index.txt @@ -97,7 +97,7 @@ https://www.iana.org/assignments/message-headers/message-headers.xhtml 8601 Yes - Message Header Field for Indicating Message Authentication Status 8689 Yes - SMTP Require TLS Option 8904 No - DNS Whitelist (DNSWL) Email Authentication Method Extension -9422 Partial - The LIMITS SMTP Service Extension +9422 Yes - The LIMITS SMTP Service Extension # SPF 4408 Yes Obs (by RFC 7208) Sender Policy Framework (SPF) for Authorizing Use of Domains in E-Mail, Version 1 diff --git a/smtp/codes.go b/smtp/codes.go index 315e565..65fcc8b 100644 --- a/smtp/codes.go +++ b/smtp/codes.go @@ -142,6 +142,6 @@ var ( SePol7RevDNSFail25 = "7.25" // ../rfc/7372:233 SePol7MultiAuthFails26 = "7.26" // ../rfc/7372:246 SePol7SenderHasNullMX27 = "7.27" // ../rfc/7505:246 - SePol7ARCFail = "7.29" // ../rfc/8617:1438 - SePol7MissingReqTLS = "7.30" // ../rfc/8689:448 + SePol7ARCFail29 = "7.29" // ../rfc/8617:1438 + SePol7MissingReqTLS30 = "7.30" // ../rfc/8689:448 ) diff --git a/smtpclient/client.go b/smtpclient/client.go index eb3e88b..310792d 100644 --- a/smtpclient/client.go +++ b/smtpclient/client.go @@ -28,6 +28,7 @@ package smtpclient import ( "bufio" + "bytes" "context" "crypto/tls" "crypto/x509" @@ -132,16 +133,20 @@ type Client struct { botched bool // If set, protocol is out of sync and no further commands can be sent. needRset bool // If set, a new delivery requires an RSET command. - remoteHelo string // From 220 greeting line. - extEcodes bool // Remote server supports sending extended error codes. - extStartTLS bool // Remote server supports STARTTLS. - ext8bitmime bool - extSize bool // Remote server supports SIZE parameter. Must only be used if > 0. - maxSize int64 // Max size of email message. - extPipelining bool // Remote server supports command pipelining. - extSMTPUTF8 bool // Remote server supports SMTPUTF8 extension. - extAuthMechanisms []string // Supported authentication mechanisms. - extRequireTLS bool // Remote supports REQUIRETLS extension. + remoteHelo string // From 220 greeting line. + extEcodes bool // Remote server supports sending extended error codes. + extStartTLS bool // Remote server supports STARTTLS. + ext8bitmime bool + extSize bool // Remote server supports SIZE parameter. Must only be used if > 0. + maxSize int64 // Max size of email message. + extPipelining bool // Remote server supports command pipelining. + extSMTPUTF8 bool // Remote server supports SMTPUTF8 extension. + extAuthMechanisms []string // Supported authentication mechanisms. + extRequireTLS bool // Remote supports REQUIRETLS extension. + ExtLimits map[string]string // For LIMITS extension, only if present and valid, with uppercase keys. + ExtLimitMailMax int // Max "MAIL" commands in a connection, if > 0. + ExtLimitRcptMax int // Max "RCPT" commands in a transaction, if > 0. + ExtLimitRcptDomainMax int // Max unique domains in a connection, if > 0. } // Error represents a failure to deliver a message. @@ -170,6 +175,8 @@ type Error struct { Err error } +type Response Error + // Unwrap returns the underlying Err. func (e Error) Unwrap() error { return e.Err @@ -744,6 +751,8 @@ func (c *Client) hello(ctx context.Context, tlsMode TLSMode, ehloHostname dns.Do } } else if strings.HasPrefix(s, "AUTH ") { c.extAuthMechanisms = strings.Split(s[len("AUTH "):], " ") + } else if strings.HasPrefix(s, "LIMITS ") { + c.ExtLimits, c.ExtLimitMailMax, c.ExtLimitRcptMax, c.ExtLimitRcptDomainMax = parseLimits([]byte(s[len("LIMITS"):])) } } } @@ -834,6 +843,79 @@ func (c *Client) hello(ctx context.Context, tlsMode TLSMode, ehloHostname dns.Do return } +// parse text after "LIMITS", including leading space. +func parseLimits(b []byte) (map[string]string, int, int, int) { + // ../rfc/9422:150 + var o int + // Read next " name=value". + pair := func() ([]byte, []byte) { + if o >= len(b) || b[o] != ' ' { + return nil, nil + } + o++ + + ns := o + for o < len(b) { + c := b[o] + if c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-' || c == '_' { + o++ + } else { + break + } + } + es := o + if ns == es || o >= len(b) || b[o] != '=' { + return nil, nil + } + o++ + vs := o + for o < len(b) { + c := b[o] + if c > 0x20 && c < 0x7f && c != ';' { + o++ + } else { + break + } + } + if vs == o { + return nil, nil + } + return b[ns:es], b[vs:o] + } + limits := map[string]string{} + var mailMax, rcptMax, rcptDomainMax int + for o < len(b) { + name, value := pair() + if name == nil { + // We skip the entire LIMITS extension for syntax errors. ../rfc/9422:232 + return nil, 0, 0, 0 + } + k := strings.ToUpper(string(name)) + if _, ok := limits[k]; ok { + // Not specified, but we treat duplicates as error. + return nil, 0, 0, 0 + } + limits[k] = string(value) + // For individual value syntax errors, we skip that value, leaving the default 0. + // ../rfc/9422:254 + switch string(name) { + case "MAILMAX": + if v, err := strconv.Atoi(string(value)); err == nil && v > 0 && len(value) <= 6 { + mailMax = v + } + case "RCPTMAX": + if v, err := strconv.Atoi(string(value)); err == nil && v > 0 && len(value) <= 6 { + rcptMax = v + } + case "RCPTDOMAINMAX": + if v, err := strconv.Atoi(string(value)); err == nil && v > 0 && len(value) <= 6 { + rcptDomainMax = v + } + } + } + return limits, mailMax, rcptMax, rcptDomainMax +} + func addrIP(addr net.Addr) string { if t, ok := addr.(*net.TCPAddr); ok { return t.IP.String() @@ -1019,15 +1101,39 @@ func (c *Client) TLSConnectionState() *tls.ConnectionState { // Returned errors can be of type Error, one of the Err-variables in this package // or other underlying errors, e.g. for i/o. Use errors.Is to check. func (c *Client) Deliver(ctx context.Context, mailFrom string, rcptTo string, msgSize int64, msg io.Reader, req8bitmime, reqSMTPUTF8, requireTLS bool) (rerr error) { + _, err := c.DeliverMultiple(ctx, mailFrom, []string{rcptTo}, msgSize, msg, req8bitmime, reqSMTPUTF8, requireTLS) + return err +} + +var errNoRecipientsPipelined = errors.New("no recipients accepted in pipelined transaction") +var errNoRecipients = errors.New("no recipients accepted in transaction") + +// DeliverMultiple is like Deliver, but attempts to deliver a message to multiple +// recipients. Errors about the entire transaction, such as i/o errors or error +// responses to the MAIL FROM or DATA commands, are returned by a non-nil rerr. If +// rcptTo has a single recipient, an error to the RCPT TO command is returned in +// rerr instead of rcptResps. Otherwise, the SMTP response for each recipient is +// returned in rcptResps. +// +// The caller should take extLimit* into account when sending. And recognize +// recipient response code "452" to mean that a recipient limit was reached, +// another transaction can be attempted immediately after instead of marking the +// delivery attempt as failed. Also code "552" must be treated like temporary error +// code "452" for historic reasons. +func (c *Client) DeliverMultiple(ctx context.Context, mailFrom string, rcptTo []string, msgSize int64, msg io.Reader, req8bitmime, reqSMTPUTF8, requireTLS bool) (rcptResps []Response, rerr error) { defer c.recover(&rerr) + if len(rcptTo) == 0 { + return nil, fmt.Errorf("need at least one recipient") + } + if c.origConn == nil { - return ErrClosed + return nil, ErrClosed } else if c.botched { - return ErrBotched + return nil, ErrBotched } else if c.needRset { if err := c.Reset(); err != nil { - return err + return nil, err } } @@ -1077,45 +1183,122 @@ func (c *Client) Deliver(ctx context.Context, mailFrom string, rcptTo string, ms // RCPT TO: ../rfc/5321:1916 // DATA: ../rfc/5321:1992 lineMailFrom := fmt.Sprintf("MAIL FROM:<%s>%s%s%s%s", mailFrom, mailSize, bodyType, smtputf8Arg, requiretlsArg) - lineRcptTo := fmt.Sprintf("RCPT TO:<%s>", rcptTo) // We are going into a transaction. We'll clear this when done. c.needRset = true if c.extPipelining { - c.cmds = []string{"mailfrom", "rcptto", "data"} + c.cmds = make([]string, 1+len(rcptTo)+1) + c.cmds[0] = "mailfrom" + for i := range rcptTo { + c.cmds[1+i] = "rcptto" + } + c.cmds[len(c.cmds)-1] = "data" c.cmdStart = time.Now() - // todo future: write in a goroutine to prevent potential deadlock if remote does not consume our writes before expecting us to read. could potentially happen with greylisting and a small tcp send window? - c.xbwriteline(lineMailFrom) - c.xbwriteline(lineRcptTo) - c.xbwriteline("DATA") - c.xflush() - // We read the response to RCPT TO and DATA without panic on read error. Servers + // Write and read in separte goroutines. Otherwise, writing a large recipient list + // could block when a server doesn't read more commands before we read their + // response. + errc := make(chan error, 1) + // Make sure we don't return before we're done writing to the connection. + defer func() { + if errc != nil { + <-errc + } + }() + go func() { + var b bytes.Buffer + b.WriteString(lineMailFrom) + b.WriteString("\r\n") + for _, rcpt := range rcptTo { + b.WriteString("RCPT TO:<") + b.WriteString(rcpt) + b.WriteString(">\r\n") + } + b.WriteString("DATA\r\n") + _, err := c.w.Write(b.Bytes()) + if err == nil { + err = c.w.Flush() + } + errc <- err + }() + + // Read response to MAIL FROM. + mfcode, mfsecode, mffirstLine, mfmoreLines := c.xread() + + // We read the response to RCPT TOs and DATA without panic on read error. Servers // may be aborting the connection after a failed MAIL FROM, e.g. outlook when it // has blocklisted your IP. We don't want the read for the response to RCPT TO to // cause a read error as it would result in an unhelpful error message and a // temporary instead of permanent error code. - mfcode, mfsecode, mffirstLine, mfmoreLines := c.xread() - rtcode, rtsecode, rtfirstLine, rtmoreLines, rterr := c.read() + // Read responses to RCPT TO. + rcptResps = make([]Response, len(rcptTo)) + nok := 0 + for i := 0; i < len(rcptTo); i++ { + code, secode, firstLine, moreLines, err := c.read() + // 552 should be treated as temporary historically, ../rfc/5321:3576 + permanent := code/100 == 5 && code != smtp.C552MailboxFull + rcptResps[i] = Response{permanent, code, secode, "rcptto", firstLine, moreLines, err} + if code == smtp.C250Completed { + nok++ + } + } + + // Read response to DATA. datacode, datasecode, datafirstLine, datamoreLines, dataerr := c.read() + writeerr := <-errc + errc = nil + + // If MAIL FROM failed, it's an error for the entire transaction. We may have been + // blocked. if mfcode != smtp.C250Completed { + if writeerr != nil || dataerr != nil { + c.botched = true + } c.xerrorf(mfcode/100 == 5, mfcode, mfsecode, mffirstLine, mfmoreLines, "%w: got %d, expected 2xx", ErrStatus, mfcode) } - if rterr != nil { - panic(rterr) - } - if rtcode != smtp.C250Completed { - c.xerrorf(rtcode/100 == 5, rtcode, rtsecode, rtfirstLine, rtmoreLines, "%w: got %d, expected 2xx", ErrStatus, rtcode) + + // If there was an i/o error writing the commands, there is no point continuing. + if writeerr != nil { + c.xbotchf(0, "", "", nil, "writing pipelined mail/rcpt/data: %w", writeerr) } + + // If the data command had an i/o or protocol error, it's also a failure for the + // entire transaction. if dataerr != nil { panic(dataerr) } + + // If we didn't have any successful recipient, there is no point in continuing. + if nok == 0 { + // Servers may return success for a DATA without valid recipients. Write a dot to + // end DATA and restore the connection to a known state. + // ../rfc/2920:328 + if datacode == smtp.C354Continue { + _, doterr := fmt.Fprintf(c.w, ".\r\n") + if doterr == nil { + doterr = c.w.Flush() + } + if doterr == nil { + _, _, _, _, doterr = c.read() + } + if doterr != nil { + c.botched = true + } + } + + if len(rcptTo) == 1 { + panic(Error(rcptResps[0])) + } + c.xerrorf(false, 0, "", "", nil, "%w", errNoRecipientsPipelined) + } + if datacode != smtp.C354Continue { c.xerrorf(datacode/100 == 5, datacode, datasecode, datafirstLine, datamoreLines, "%w: got %d, expected 354", ErrStatus, datacode) } + } else { c.cmds[0] = "mailfrom" c.cmdStart = time.Now() @@ -1125,12 +1308,35 @@ func (c *Client) Deliver(ctx context.Context, mailFrom string, rcptTo string, ms c.xerrorf(code/100 == 5, code, secode, firstLine, moreLines, "%w: got %d, expected 2xx", ErrStatus, code) } - c.cmds[0] = "rcptto" - c.cmdStart = time.Now() - c.xwriteline(lineRcptTo) - code, secode, firstLine, moreLines = c.xread() - if code != smtp.C250Completed { - c.xerrorf(code/100 == 5, code, secode, firstLine, moreLines, "%w: got %d, expected 2xx", ErrStatus, code) + rcptResps = make([]Response, len(rcptTo)) + nok := 0 + for i, rcpt := range rcptTo { + c.cmds[0] = "rcptto" + c.cmdStart = time.Now() + c.xwriteline(fmt.Sprintf("RCPT TO:<%s>", rcpt)) + code, secode, firstLine, moreLines = c.xread() + if i > 0 && (code == smtp.C452StorageFull || code == smtp.C552MailboxFull) { + // Remote doesn't accept more recipients for this transaction. Don't send more, give + // remaining recipients the same error result. + for j := i; j < len(rcptTo); j++ { + rcptResps[j] = Response{false, code, secode, "rcptto", firstLine, moreLines, fmt.Errorf("no more recipients accepted in transaction")} + } + break + } + var err error + if code == smtp.C250Completed { + nok++ + } else { + err = fmt.Errorf("%w: got %d, expected 2xx", ErrStatus, code) + } + rcptResps[i] = Response{code/100 == 5, code, secode, "rcptto", firstLine, moreLines, err} + } + + if nok == 0 { + if len(rcptTo) == 1 { + panic(Error(rcptResps[0])) + } + c.xerrorf(false, 0, "", "", nil, "%w", errNoRecipients) } c.cmds[0] = "data" diff --git a/smtpclient/client_test.go b/smtpclient/client_test.go index 87702f3..bba0e5f 100644 --- a/smtpclient/client_test.go +++ b/smtpclient/client_test.go @@ -39,6 +39,7 @@ func TestClient(t *testing.T) { mlog.SetConfig(map[string]slog.Level{"": mlog.LevelTrace}) type options struct { + // Server behaviour. pipelining bool ecodes bool maxSize int @@ -47,7 +48,11 @@ func TestClient(t *testing.T) { smtputf8 bool requiretls bool ehlo bool + auths []string // Allowed mechanisms. + nodeliver bool // For server, whether client will attempt a delivery. + + // Client behaviour. tlsMode TLSMode tlsPKIX bool roots *x509.CertPool @@ -55,9 +60,8 @@ func TestClient(t *testing.T) { need8bitmime bool needsmtputf8 bool needsrequiretls bool - auths []string // Allowed mechanisms. - - nodeliver bool // For server, whether client will attempt a delivery. + recipients []string // If nil, mjl@mox.example is used. + resps []Response // Checked only if non-nil. } // Make fake cert, and make it trusted. @@ -68,6 +72,13 @@ func TestClient(t *testing.T) { Certificates: []tls.Certificate{cert}, } + cleanupResp := func(resps []Response) []Response { + for i, r := range resps { + resps[i] = Response{Code: r.Code, Secode: r.Secode} + } + return resps + } + test := func(msg string, opts options, auth func(l []string, cs *tls.ConnectionState) (sasl.Client, error), expClientErr, expDeliverErr, expServerErr error) { t.Helper() @@ -89,6 +100,7 @@ func TestClient(t *testing.T) { }() fail := func(format string, args ...any) { err := fmt.Errorf("server: %w", fmt.Errorf(format, args...)) + log.Errorx("failure", err) if err != nil && expServerErr != nil && (errors.Is(err, expServerErr) || errors.As(err, reflect.New(reflect.ValueOf(expServerErr).Type()).Interface())) { err = nil } @@ -158,6 +170,7 @@ func TestClient(t *testing.T) { if opts.auths != nil { writeline("250-AUTH " + strings.Join(opts.auths, " ")) } + writeline("250-LIMITS MAILMAX=10 RCPTMAX=100 RCPTDOMAINMAX=1") writeline("250 UNKNOWN") // To be ignored. } @@ -255,8 +268,18 @@ func TestClient(t *testing.T) { if expClientErr == nil && !opts.nodeliver { readline("MAIL FROM:") writeline("250 ok") - readline("RCPT TO:") - writeline("250 ok") + n := len(opts.recipients) + if n == 0 { + n = 1 + } + for i := 0; i < n; i++ { + readline("RCPT TO:") + resp := "250 ok" + if i < len(opts.resps) { + resp = fmt.Sprintf("%d maybe", opts.resps[i].Code) + } + writeline(resp) + } readline("DATA") writeline("354 continue") reader := smtp.NewDataReader(br) @@ -269,8 +292,14 @@ func TestClient(t *testing.T) { readline("MAIL FROM:") writeline("250 ok") - readline("RCPT TO:") - writeline("250 ok") + for i := 0; i < n; i++ { + readline("RCPT TO:") + resp := "250 ok" + if i < len(opts.resps) { + resp = fmt.Sprintf("%d maybe", opts.resps[i].Code) + } + writeline(resp) + } readline("DATA") writeline("354 continue") reader = smtp.NewDataReader(br) @@ -294,6 +323,7 @@ func TestClient(t *testing.T) { }() fail := func(format string, args ...any) { err := fmt.Errorf("client: %w", fmt.Errorf(format, args...)) + log.Errorx("failure", err) result <- err panic("stop") } @@ -305,18 +335,26 @@ func TestClient(t *testing.T) { result <- nil return } - err = client.Deliver(ctx, "postmaster@mox.example", "mjl@mox.example", int64(len(msg)), strings.NewReader(msg), opts.need8bitmime, opts.needsmtputf8, opts.needsrequiretls) - if (err == nil) != (expDeliverErr == nil) || err != nil && !errors.Is(err, expDeliverErr) { - fail("first deliver: got err %v, expected %v", err, expDeliverErr) + rcptTo := opts.recipients + if len(rcptTo) == 0 { + rcptTo = []string{"mjl@mox.example"} + } + resps, err := client.DeliverMultiple(ctx, "postmaster@mox.example", rcptTo, int64(len(msg)), strings.NewReader(msg), opts.need8bitmime, opts.needsmtputf8, opts.needsrequiretls) + if (err == nil) != (expDeliverErr == nil) || err != nil && !errors.Is(err, expDeliverErr) && !reflect.DeepEqual(err, expDeliverErr) { + fail("first deliver: got err %#v (%s), expected %#v (%s)", err, err, expDeliverErr, expDeliverErr) + } else if opts.resps != nil && !reflect.DeepEqual(cleanupResp(resps), opts.resps) { + fail("first deliver: got resps %v, expected %v", resps, opts.resps) } if err == nil { err = client.Reset() if err != nil { fail("reset: %v", err) } - err = client.Deliver(ctx, "postmaster@mox.example", "mjl@mox.example", int64(len(msg)), strings.NewReader(msg), opts.need8bitmime, opts.needsmtputf8, opts.needsrequiretls) - if (err == nil) != (expDeliverErr == nil) || err != nil && !errors.Is(err, expDeliverErr) { - fail("second deliver: got err %v, expected %v", err, expDeliverErr) + resps, err = client.DeliverMultiple(ctx, "postmaster@mox.example", rcptTo, int64(len(msg)), strings.NewReader(msg), opts.need8bitmime, opts.needsmtputf8, opts.needsrequiretls) + if (err == nil) != (expDeliverErr == nil) || err != nil && !errors.Is(err, expDeliverErr) && !reflect.DeepEqual(err, expDeliverErr) { + fail("second deliver: got err %#v (%s), expected %#v (%s)", err, err, expDeliverErr, expDeliverErr) + } else if opts.resps != nil && !reflect.DeepEqual(cleanupResp(resps), opts.resps) { + fail("second: got resps %v, expected %v", resps, opts.resps) } } err = client.Close() @@ -369,9 +407,58 @@ test test(msg, options{ehlo: true, eightbitmime: true}, nil, nil, nil, nil) test(msg, options{ehlo: true, eightbitmime: false, need8bitmime: true, nodeliver: true}, nil, nil, Err8bitmimeUnsupported, nil) test(msg, options{ehlo: true, smtputf8: false, needsmtputf8: true, nodeliver: true}, nil, nil, ErrSMTPUTF8Unsupported, nil) - test(msg, options{ehlo: true, starttls: true, tlsMode: TLSRequiredStartTLS, tlsPKIX: true, tlsHostname: dns.Domain{ASCII: "mismatch.example"}, nodeliver: true}, nil, ErrTLS, nil, &net.OpError{}) // Server TLS handshake is a net.OpError with "remote error" as text. + + // Server TLS handshake is a net.OpError with "remote error" as text. + test(msg, options{ehlo: true, starttls: true, tlsMode: TLSRequiredStartTLS, tlsPKIX: true, tlsHostname: dns.Domain{ASCII: "mismatch.example"}, nodeliver: true}, nil, ErrTLS, nil, &net.OpError{}) + test(msg, options{ehlo: true, maxSize: len(msg) - 1, nodeliver: true}, nil, nil, ErrSize, nil) + // Multiple recipients, not pipelined. + multi1 := options{ + ehlo: true, + pipelining: true, + ecodes: true, + recipients: []string{"mjl@mox.example", "mjl2@mox.example", "mjl3@mox.example"}, + resps: []Response{ + {Code: smtp.C250Completed}, + {Code: smtp.C250Completed}, + {Code: smtp.C250Completed}, + }, + } + test(msg, multi1, nil, nil, nil, nil) + multi1.pipelining = true + test(msg, multi1, nil, nil, nil, nil) + + // Multiple recipients with 452 and other error, not pipelined + multi2 := options{ + ehlo: true, + ecodes: true, + recipients: []string{"xmjl@mox.example", "xmjl2@mox.example", "xmjl3@mox.example"}, + resps: []Response{ + {Code: smtp.C250Completed}, + {Code: smtp.C554TransactionFailed}, // Will continue when not pipelined. + {Code: smtp.C452StorageFull}, // Will stop sending further recipients. + }, + } + test(msg, multi2, nil, nil, nil, nil) + multi2.pipelining = true + test(msg, multi2, nil, nil, nil, nil) + multi2.pipelining = false + multi2.resps[2].Code = smtp.C552MailboxFull + test(msg, multi2, nil, nil, nil, nil) + multi2.pipelining = true + test(msg, multi2, nil, nil, nil, nil) + + // Single recipient with error and pipelining is an error. + multi3 := options{ + ehlo: true, + pipelining: true, + ecodes: true, + recipients: []string{"xmjl@mox.example"}, + resps: []Response{{Code: smtp.C452StorageFull}}, + } + test(msg, multi3, nil, nil, Error{Code: smtp.C452StorageFull, Command: "rcptto", Line: "452 maybe"}, nil) + authPlain := func(l []string, cs *tls.ConnectionState) (sasl.Client, error) { return sasl.NewClientPlain("test", "test"), nil } @@ -669,6 +756,69 @@ func TestErrors(t *testing.T) { panic(fmt.Errorf("got %#v, expected ErrStatus with Permanent", err)) } }) + + // If we try multiple recipients and first is 452, it is an error and a + // non-pipelined deliver will be aborted. + run(t, func(s xserver) { + s.writeline("220 mox.example") + s.readline("EHLO") + s.writeline("250 mox.example") + s.readline("MAIL FROM:") + s.writeline("250 ok") + s.readline("RCPT TO:") + s.writeline("451 not now") + s.readline("RCPT TO:") + s.writeline("451 not now") + s.readline("QUIT") + s.writeline("250 ok") + }, func(conn net.Conn) { + c, err := New(ctx, log.Logger, conn, TLSOpportunistic, false, localhost, zerohost, Opts{}) + if err != nil { + panic(err) + } + + msg := "" + _, err = c.DeliverMultiple(ctx, "postmaster@other.example", []string{"mjl@mox.example", "mjl@mox.example"}, int64(len(msg)), strings.NewReader(msg), false, false, false) + var xerr Error + if err == nil || !errors.Is(err, errNoRecipients) || !errors.As(err, &xerr) || xerr.Permanent { + panic(fmt.Errorf("got %#v (%s) expected errNoRecipients with non-Permanent", err, err)) + } + c.Close() + }) + + // If we try multiple recipients and first is 452, it is an error and a pipelined + // deliver will abort an allowed DATA. + run(t, func(s xserver) { + s.writeline("220 mox.example") + s.readline("EHLO") + s.writeline("250-mox.example") + s.writeline("250 PIPELINING") + s.readline("MAIL FROM:") + s.writeline("250 ok") + s.readline("RCPT TO:") + s.writeline("451 not now") + s.readline("RCPT TO:") + s.writeline("451 not now") + s.readline("DATA") + s.writeline("354 ok") + s.readline(".") + s.writeline("503 no recipient") + s.readline("QUIT") + s.writeline("250 ok") + }, func(conn net.Conn) { + c, err := New(ctx, log.Logger, conn, TLSOpportunistic, false, localhost, zerohost, Opts{}) + if err != nil { + panic(err) + } + + msg := "" + _, err = c.DeliverMultiple(ctx, "postmaster@other.example", []string{"mjl@mox.example", "mjl@mox.example"}, int64(len(msg)), strings.NewReader(msg), false, false, false) + var xerr Error + if err == nil || !errors.Is(err, errNoRecipientsPipelined) || !errors.As(err, &xerr) || xerr.Permanent { + panic(fmt.Errorf("got %#v (%s), expected errNoRecipientsPipelined with non-Permanent", err, err)) + } + c.Close() + }) } type xserver struct { @@ -740,6 +890,21 @@ func run(t *testing.T, server func(s xserver), client func(conn net.Conn)) { } } +func TestLimits(t *testing.T) { + check := func(s string, expLimits map[string]string, expMailMax, expRcptMax, expRcptDomainMax int) { + t.Helper() + limits, mailmax, rcptMax, rcptDomainMax := parseLimits([]byte(s)) + if !reflect.DeepEqual(limits, expLimits) || mailmax != expMailMax || rcptMax != expRcptMax || rcptDomainMax != expRcptDomainMax { + t.Errorf("bad limits, got %v %d %d %d, expected %v %d %d %d, for %q", limits, mailmax, rcptMax, rcptDomainMax, expLimits, expMailMax, expRcptMax, expRcptDomainMax, s) + } + } + check(" unknown=a=b -_1oK=xY", map[string]string{"UNKNOWN": "a=b", "-_1OK": "xY"}, 0, 0, 0) + check(" MAILMAX=123 OTHER=ignored RCPTDOMAINMAX=1 RCPTMAX=321", map[string]string{"MAILMAX": "123", "OTHER": "ignored", "RCPTDOMAINMAX": "1", "RCPTMAX": "321"}, 123, 321, 1) + check(" MAILMAX=invalid", map[string]string{"MAILMAX": "invalid"}, 0, 0, 0) + check(" invalid syntax", nil, 0, 0, 0) + check(" DUP=1 DUP=2", nil, 0, 0, 0) +} + // Just a cert that appears valid. SMTP client will not verify anything about it // (that is opportunistic TLS for you, "better some than none"). Let's enjoy this // one moment where it makes life easier. diff --git a/smtpserver/dsn.go b/smtpserver/dsn.go index 672eed4..7307971 100644 --- a/smtpserver/dsn.go +++ b/smtpserver/dsn.go @@ -3,6 +3,7 @@ package smtpserver import ( "context" "fmt" + "time" "github.com/mjl-/mox/dsn" "github.com/mjl-/mox/mlog" @@ -53,7 +54,7 @@ func queueDSN(ctx context.Context, log mlog.Log, c *conn, rcptTo smtp.Path, m ds if requireTLS { reqTLS = &requireTLS } - qm := queue.MakeMsg(smtp.Path{}, rcptTo, has8bit, smtputf8, int64(len(buf)), m.MessageID, nil, reqTLS) + qm := queue.MakeMsg(smtp.Path{}, rcptTo, has8bit, smtputf8, int64(len(buf)), m.MessageID, nil, reqTLS, time.Now()) qm.DSNUTF8 = bufUTF8 if err := queue.Add(ctx, c.log, "", f, qm); err != nil { return err diff --git a/smtpserver/server.go b/smtpserver/server.go index c095633..05ee88f 100644 --- a/smtpserver/server.go +++ b/smtpserver/server.go @@ -1834,7 +1834,13 @@ func (c *conn) cmdData(p *parser) { tlsComment := mox.TLSReceivedComment(c.log, tlsConn.ConnectionState()) recvHdr.Add(" ", tlsComment...) } - recvHdr.Add(" ", "for", "<"+rcptTo+">;", time.Now().Format(message.RFC5322Z)) + // We leave out an empty "for" clause. This is empty for messages submitted to + // multiple recipients, so the message stays identical and a single smtp + // transaction can deliver, only transferring the data once. + if rcptTo != "" { + recvHdr.Add(" ", "for", "<"+rcptTo+">;") + } + recvHdr.Add(" ", time.Now().Format(message.RFC5322Z)) return recvHdr.String() } @@ -1979,6 +1985,7 @@ func (c *conn) submit(ctx context.Context, recvHdrFor func(string) string, msgWr // We always deliver through the queue. It would be more efficient to deliver // directly, but we don't want to circumvent all the anti-spam measures. Accounts // on a single mox instance should be allowed to block each other. + now := time.Now() qml := make([]queue.Msg, len(c.recipients)) for i, rcptAcc := range c.recipients { if Localserve { @@ -1993,9 +2000,16 @@ func (c *conn) submit(ctx context.Context, recvHdrFor func(string) string, msgWr } } - xmsgPrefix := append([]byte(recvHdrFor(rcptAcc.rcptTo.String())), msgPrefix...) + // For multiple recipients, we don't make each message prefix unique, leaving out + // the "for" clause in the Received header. This allows the queue to deliver the + // messages in a single smtp transaction. + var rcptTo string + if len(c.recipients) == 1 { + rcptTo = rcptAcc.rcptTo.String() + } + xmsgPrefix := append([]byte(recvHdrFor(rcptTo)), msgPrefix...) msgSize := int64(len(xmsgPrefix)) + msgWriter.Size - qm := queue.MakeMsg(*c.mailFrom, rcptAcc.rcptTo, msgWriter.Has8bit, c.smtputf8, msgSize, messageID, xmsgPrefix, c.requireTLS) + qm := queue.MakeMsg(*c.mailFrom, rcptAcc.rcptTo, msgWriter.Has8bit, c.smtputf8, msgSize, messageID, xmsgPrefix, c.requireTLS, now) if !c.futureRelease.IsZero() { qm.NextAttempt = c.futureRelease qm.FutureReleaseRequest = c.futureReleaseRequest diff --git a/tlsrptsend/send.go b/tlsrptsend/send.go index 172e60b..4370191 100644 --- a/tlsrptsend/send.go +++ b/tlsrptsend/send.go @@ -589,7 +589,7 @@ Period: %s - %s UTC continue } - qm := queue.MakeMsg(from.Path(), rcpt.Address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil) + qm := queue.MakeMsg(from.Path(), rcpt.Address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil, time.Now()) // Don't try as long as regular deliveries, and stop before we would send the // delayed DSN. Though we also won't send that due to IsTLSReport. // ../rfc/8460:1077 diff --git a/webadmin/api.json b/webadmin/api.json index 430c179..776d587 100644 --- a/webadmin/api.json +++ b/webadmin/api.json @@ -3372,7 +3372,7 @@ }, { "Name": "BaseID", - "Docs": "A message for multiple recipients will get a BaseID that is identical to the first Msg.ID queued. They may be delivered in a single SMTP transaction if they are going to the same mail server. For messages with a single recipient, this field will be 0.", + "Docs": "A message for multiple recipients will get a BaseID that is identical to the first Msg.ID queued. The message contents will be identical for each recipient, including MsgPrefix. If other properties are identical too, including recipient domain, multiple Msgs may be delivered in a single SMTP transaction. For messages with a single recipient, this field will be 0.", "Typewords": [ "int64" ] diff --git a/webadmin/api.ts b/webadmin/api.ts index 6fc55c0..e282cd3 100644 --- a/webadmin/api.ts +++ b/webadmin/api.ts @@ -471,7 +471,7 @@ export interface ClientConfigsEntry { // queueing related fields. export interface Msg { ID: number - BaseID: number // A message for multiple recipients will get a BaseID that is identical to the first Msg.ID queued. They may be delivered in a single SMTP transaction if they are going to the same mail server. For messages with a single recipient, this field will be 0. + BaseID: number // A message for multiple recipients will get a BaseID that is identical to the first Msg.ID queued. The message contents will be identical for each recipient, including MsgPrefix. If other properties are identical too, including recipient domain, multiple Msgs may be delivered in a single SMTP transaction. For messages with a single recipient, this field will be 0. Queued: Date SenderAccount: string // Failures are delivered back to this local account. Also used for routing. SenderLocalpart: Localpart // Should be a local user and domain. diff --git a/webmail/api.go b/webmail/api.go index cbaa4d7..2e8d089 100644 --- a/webmail/api.go +++ b/webmail/api.go @@ -629,14 +629,21 @@ func (w Webmail) MessageSubmit(ctx context.Context, m SubmitMessage) { IPDomain: dns.IPDomain{Domain: fromAddr.Address.Domain}, } qml := make([]queue.Msg, len(recipients)) + now := time.Now() for i, rcpt := range recipients { - rcptMsgPrefix := recvHdrFor(rcpt.Pack(smtputf8)) + msgPrefix + // Don't use per-recipient unique message prefix when multiple recipients are + // present, or the queue cannot deliver it in a single smtp transaction. + var recvRcpt string + if len(recipients) == 1 { + recvRcpt = rcpt.Pack(smtputf8) + } + rcptMsgPrefix := recvHdrFor(recvRcpt) + msgPrefix msgSize := int64(len(rcptMsgPrefix)) + xc.Size toPath := smtp.Path{ Localpart: rcpt.Localpart, IPDomain: dns.IPDomain{Domain: rcpt.Domain}, } - qm := queue.MakeMsg(fromPath, toPath, has8bit, smtputf8, msgSize, messageID, []byte(rcptMsgPrefix), m.RequireTLS) + qm := queue.MakeMsg(fromPath, toPath, has8bit, smtputf8, msgSize, messageID, []byte(rcptMsgPrefix), m.RequireTLS, now) if m.FutureRelease != nil { ival := time.Until(*m.FutureRelease) if ival < 0 {