feat: 实现对外邮件投递（外发队列 + MX 直投 + DKIM + 退信 + 管理后台）
- 新增 internal/outbound 模块：MX 查询、SMTP 出站客户端（EHLO/STARTTLS/ MAIL/RCPT/DATA/QUIT）、4xx 临时失败与 5xx 永久失败分类、8BITMIME 支持 - 新增 outbound_messages 队列表与 OutboundStore，后台 worker 指数退避重试 - 永久失败/超限退信到发件人收件箱，包含原因与目标收件人 - 外发邮件使用域名 DKIM 私钥签名（go-msgauth） - SMTP 提交集成：认证用户可发外部收件人，MAIL FROM 必须等于登录用户邮箱， 未认证外部投递明确拒绝（防开放中继） - Web 发信集成：外部收件人自动入队，附件以 multipart/mixed + base64 编码 加入邮件正文 - 每用户每分钟/每日发送限速（max_per_day=0 可禁用外部投递） - 管理后台新增外发队列页面：状态统计、失败原因、手动重试/取消 - 新增 [outbound] 配置段并更新 README / todo.md
This commit is contained in:
28 files changed
+1657
-186
No files matched your search
@@ -0,0 +1,323 @@
|
||||
// Package outbound implements external (outbound) email delivery.
|
||||
//
|
||||
// Messages queued for external recipients are stored in the outbound_messages
|
||||
// table and delivered by the Manager's background worker: MX lookup, SMTP
|
||||
// transaction over port 25 with opportunistic STARTTLS, exponential backoff
|
||||
// retries, permanent-failure bounces and DKIM signing.
|
||||
package outbound
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/textproto"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DeliveryError wraps an SMTP delivery failure and records whether it is
|
||||
// permanent (5xx / NXDOMAIN / invalid address) or temporary (4xx / network /
|
||||
// timeout). Temporary failures are retried by the queue worker.
|
||||
type DeliveryError struct {
|
||||
Permanent bool
|
||||
Code int
|
||||
Msg string
|
||||
}
|
||||
|
||||
func (e *DeliveryError) Error() string {
|
||||
if e.Code > 0 {
|
||||
return fmt.Sprintf("%d %s", e.Code, e.Msg)
|
||||
}
|
||||
return e.Msg
|
||||
}
|
||||
|
||||
// newTempError creates a temporary delivery error.
|
||||
func newTempError(format string, args ...interface{}) *DeliveryError {
|
||||
return &DeliveryError{Permanent: false, Msg: fmt.Sprintf(format, args...)}
|
||||
}
|
||||
|
||||
// newPermError creates a permanent delivery error.
|
||||
func newPermError(format string, args ...interface{}) *DeliveryError {
|
||||
return &DeliveryError{Permanent: true, Msg: fmt.Sprintf(format, args...)}
|
||||
}
|
||||
|
||||
// Mailer performs direct MX delivery of a single message.
|
||||
type Mailer struct {
|
||||
Hostname string // EHLO hostname presented to remote servers
|
||||
ConnectTimeout time.Duration
|
||||
}
|
||||
|
||||
// NewMailer creates a Mailer with the given EHLO hostname and connect timeout.
|
||||
func NewMailer(hostname string, connectTimeout time.Duration) *Mailer {
|
||||
if hostname == "" {
|
||||
hostname = "localhost"
|
||||
}
|
||||
return &Mailer{Hostname: hostname, ConnectTimeout: connectTimeout}
|
||||
}
|
||||
|
||||
// Deliver sends one message to one recipient via the recipient domain's MX.
|
||||
// It returns the final SMTP response text on success and a *DeliveryError on
|
||||
// failure.
|
||||
func (m *Mailer) Deliver(from, to string, data []byte) (string, error) {
|
||||
at := strings.LastIndex(to, "@")
|
||||
if at < 0 || at == len(to)-1 {
|
||||
return "", newPermError("invalid recipient address: %s", to)
|
||||
}
|
||||
domain := strings.ToLower(strings.TrimSpace(to[at+1:]))
|
||||
|
||||
mxHosts, err := lookupMX(domain)
|
||||
if err != nil {
|
||||
var de *DeliveryError
|
||||
if errors.As(err, &de) {
|
||||
return "", de
|
||||
}
|
||||
return "", newTempError("MX lookup failed for %s: %v", domain, err)
|
||||
}
|
||||
|
||||
var lastErr *DeliveryError
|
||||
for _, host := range mxHosts {
|
||||
resp, err := m.deliverToHost(host, from, to, data)
|
||||
if err == nil {
|
||||
return resp, nil
|
||||
}
|
||||
var de *DeliveryError
|
||||
if errors.As(err, &de) {
|
||||
lastErr = de
|
||||
// A permanent failure from one MX applies to the whole message,
|
||||
// do not try other MX hosts.
|
||||
if de.Permanent {
|
||||
return "", de
|
||||
}
|
||||
continue
|
||||
}
|
||||
lastErr = newTempError("delivery to %s failed: %v", host, err)
|
||||
}
|
||||
if lastErr == nil {
|
||||
lastErr = newTempError("no MX hosts available for %s", domain)
|
||||
}
|
||||
return "", lastErr
|
||||
}
|
||||
|
||||
// smtpClient wraps a textproto connection to a remote SMTP server.
|
||||
type smtpClient struct {
|
||||
conn net.Conn
|
||||
txt *textproto.Conn
|
||||
host string
|
||||
exts map[string]string // advertised EHLO extensions (upper-case key -> params)
|
||||
}
|
||||
|
||||
func (c *smtpClient) Close() {
|
||||
if c.txt != nil {
|
||||
_ = c.txt.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// cmd sends a command and expects the given reply codes, returning the
|
||||
// response text. Codes other than expected are returned as a DeliveryError.
|
||||
func (c *smtpClient) cmd(expectCode int, format string, args ...interface{}) (int, string, error) {
|
||||
if err := c.txt.PrintfLine(format, args...); err != nil {
|
||||
return 0, "", newTempError("write to %s failed: %v", c.host, err)
|
||||
}
|
||||
code, msg, err := c.txt.ReadResponse(expectCode)
|
||||
if err != nil {
|
||||
return code, msg, classifyResponse(err, msg)
|
||||
}
|
||||
return code, msg, nil
|
||||
}
|
||||
|
||||
// classifyResponse converts a textproto error (wrong reply code) into a
|
||||
// DeliveryError, keeping the actual SMTP code and text.
|
||||
func classifyResponse(err error, fallback string) *DeliveryError {
|
||||
var protoErr *textproto.Error
|
||||
if errors.As(err, &protoErr) {
|
||||
return &DeliveryError{
|
||||
Permanent: protoErr.Code >= 500,
|
||||
Code: protoErr.Code,
|
||||
Msg: protoErr.Msg,
|
||||
}
|
||||
}
|
||||
if fallback != "" {
|
||||
return newTempError("%s", fallback)
|
||||
}
|
||||
return newTempError("%v", err)
|
||||
}
|
||||
|
||||
// hello sends EHLO and records the advertised extensions. If EHLO fails it
|
||||
// falls back to HELO for very old servers.
|
||||
func (c *smtpClient) hello(hostname string) error {
|
||||
if err := c.txt.PrintfLine("EHLO %s", hostname); err != nil {
|
||||
return newTempError("write EHLO to %s failed: %v", c.host, err)
|
||||
}
|
||||
code, msg, err := c.txt.ReadResponse(250)
|
||||
if err != nil {
|
||||
// Fall back to HELO.
|
||||
if err := c.txt.PrintfLine("HELO %s", hostname); err != nil {
|
||||
return newTempError("write HELO to %s failed: %v", c.host, err)
|
||||
}
|
||||
code, msg, err = c.txt.ReadResponse(250)
|
||||
if err != nil {
|
||||
return classifyResponse(err, msg)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
_ = code
|
||||
c.exts = map[string]string{}
|
||||
for _, line := range strings.Split(msg, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
parts := strings.SplitN(line, " ", 2)
|
||||
key := strings.ToUpper(parts[0])
|
||||
val := ""
|
||||
if len(parts) == 2 {
|
||||
val = parts[1]
|
||||
}
|
||||
c.exts[key] = val
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// deliverToHost performs a full SMTP transaction with a single MX host.
|
||||
func (m *Mailer) deliverToHost(host, from, to string, data []byte) (string, error) {
|
||||
addr := net.JoinHostPort(host, "25")
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), m.ConnectTimeout)
|
||||
defer cancel()
|
||||
|
||||
dialer := &net.Dialer{Timeout: m.ConnectTimeout}
|
||||
conn, err := dialer.DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return "", newTempError("connect to %s failed: %v", addr, err)
|
||||
}
|
||||
|
||||
c := &smtpClient{conn: conn, txt: textproto.NewConn(conn), host: host}
|
||||
defer c.Close()
|
||||
|
||||
// Read greeting (expect 220).
|
||||
if _, msg, err := c.txt.ReadResponse(220); err != nil {
|
||||
return "", classifyResponse(err, msg)
|
||||
}
|
||||
|
||||
if err := c.hello(m.Hostname); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Opportunistic STARTTLS (RFC 3207): only when the server advertises it.
|
||||
if _, ok := c.exts["STARTTLS"]; ok {
|
||||
if _, _, err := c.cmd(220, "STARTTLS"); err != nil {
|
||||
return "", err
|
||||
}
|
||||
tlsConn := tls.Client(conn, &tls.Config{
|
||||
ServerName: host,
|
||||
InsecureSkipVerify: true, // remote MX certificates often cannot be verified
|
||||
})
|
||||
if err := tlsConn.HandshakeContext(ctx); err != nil {
|
||||
return "", newTempError("TLS handshake with %s failed: %v", host, err)
|
||||
}
|
||||
c.txt = textproto.NewConn(tlsConn)
|
||||
if err := c.hello(m.Hostname); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
|
||||
// MAIL FROM with BODY=8BITMIME when the message contains 8-bit bytes and
|
||||
// the remote server supports it.
|
||||
mailCmd := "MAIL FROM:<%s>"
|
||||
if is8Bit(data) {
|
||||
if _, ok := c.exts["8BITMIME"]; ok {
|
||||
mailCmd = "MAIL FROM:<%s> BODY=8BITMIME"
|
||||
} else {
|
||||
return "", newPermError("%s does not advertise 8BITMIME and the message contains 8-bit data", host)
|
||||
}
|
||||
}
|
||||
if _, _, err := c.cmd(250, mailCmd, from); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, _, err := c.cmd(250, "RCPT TO:<%s>", to); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, _, err := c.cmd(354, "DATA"); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Write the message body with dot-stuffing.
|
||||
dw := c.txt.DotWriter()
|
||||
if _, err := dw.Write(data); err != nil {
|
||||
_ = dw.Close()
|
||||
return "", newTempError("writing message data to %s failed: %v", host, err)
|
||||
}
|
||||
if err := dw.Close(); err != nil {
|
||||
return "", newTempError("finalizing message data to %s failed: %v", host, err)
|
||||
}
|
||||
|
||||
code, msg, err := c.txt.ReadResponse(250)
|
||||
if err != nil {
|
||||
return "", classifyResponse(err, msg)
|
||||
}
|
||||
|
||||
// Best-effort QUIT.
|
||||
_ = c.txt.PrintfLine("QUIT")
|
||||
_, _, _ = c.txt.ReadResponse(221)
|
||||
|
||||
return fmt.Sprintf("%d %s", code, msg), nil
|
||||
}
|
||||
|
||||
// is8Bit reports whether the data contains any byte >= 0x80.
|
||||
func is8Bit(data []byte) bool {
|
||||
for _, b := range data {
|
||||
if b >= 0x80 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// lookupMX resolves the MX hosts for a domain, sorted by preference.
|
||||
// Per RFC 5321 section 5.1, when no MX record exists the domain itself is
|
||||
// used as an implicit MX with preference 0.
|
||||
func lookupMX(domain string) ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
mxs, err := net.DefaultResolver.LookupMX(ctx, domain)
|
||||
if err != nil {
|
||||
var dnsErr *net.DNSError
|
||||
if errors.As(err, &dnsErr) && dnsErr.IsNotFound {
|
||||
return nil, newPermError("domain does not exist: %s", domain)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(mxs) == 0 {
|
||||
// Implicit MX: fall back to the domain's A/AAAA records.
|
||||
ips, err := net.DefaultResolver.LookupIPAddr(ctx, domain)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
hosts := make([]string, 0, len(ips))
|
||||
for _, ip := range ips {
|
||||
hosts = append(hosts, ip.String())
|
||||
}
|
||||
if len(hosts) == 0 {
|
||||
return nil, fmt.Errorf("no MX or A records for %s", domain)
|
||||
}
|
||||
return hosts, nil
|
||||
}
|
||||
|
||||
sort.Slice(mxs, func(i, j int) bool { return mxs[i].Pref < mxs[j].Pref })
|
||||
hosts := make([]string, 0, len(mxs))
|
||||
for _, mx := range mxs {
|
||||
h := strings.TrimSuffix(mx.Host, ".")
|
||||
if h != "" {
|
||||
hosts = append(hosts, h)
|
||||
}
|
||||
}
|
||||
if len(hosts) == 0 {
|
||||
return nil, fmt.Errorf("no usable MX hosts for %s", domain)
|
||||
}
|
||||
return hosts, nil
|
||||
}
|
||||
@@ -0,0 +1,409 @@
|
||||
package outbound
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"net/mail"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"mail_go/config"
|
||||
"mail_go/internal/db"
|
||||
"mail_go/internal/store"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// Manager orchestrates the outbound delivery queue: enqueueing messages,
|
||||
// background delivery worker, exponential backoff retries, DKIM signing,
|
||||
// per-user rate limits and failure bounces.
|
||||
type Manager struct {
|
||||
cfg config.OutboundConfig
|
||||
hostname string // EHLO hostname
|
||||
mailer *Mailer
|
||||
stores *store.Stores
|
||||
|
||||
kick chan struct{}
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
once sync.Once
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
lim map[uint]*userWindow
|
||||
batch int
|
||||
}
|
||||
|
||||
// userWindow tracks a user's sending rate within fixed windows.
|
||||
type userWindow struct {
|
||||
minuteStart time.Time
|
||||
minuteCount int
|
||||
dayStart time.Time
|
||||
dayCount int
|
||||
}
|
||||
|
||||
// NewManager creates an outbound delivery Manager.
|
||||
// hostname is the EHLO name presented to remote servers (defaults to "localhost").
|
||||
func NewManager(cfg config.OutboundConfig, hostname string, stores *store.Stores) *Manager {
|
||||
m := &Manager{
|
||||
cfg: cfg,
|
||||
hostname: hostname,
|
||||
mailer: NewMailer(hostname, time.Duration(cfg.ConnectTimeout)*time.Second),
|
||||
stores: stores,
|
||||
kick: make(chan struct{}, 1),
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
lim: make(map[uint]*userWindow),
|
||||
batch: 50,
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// Start launches the background delivery worker.
|
||||
func (m *Manager) Start() {
|
||||
interval := time.Duration(m.cfg.PollInterval) * time.Second
|
||||
if interval <= 0 {
|
||||
interval = 15 * time.Second
|
||||
}
|
||||
|
||||
m.wg.Add(1)
|
||||
go func() {
|
||||
defer m.wg.Done()
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
log.Printf("outbound: delivery worker started (interval=%s, max_attempts=%d)", interval, m.cfg.MaxAttempts)
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
m.processDue()
|
||||
case <-m.kick:
|
||||
m.processDue()
|
||||
case <-m.stop:
|
||||
close(m.done)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop gracefully stops the delivery worker.
|
||||
func (m *Manager) Stop() {
|
||||
m.once.Do(func() {
|
||||
close(m.stop)
|
||||
})
|
||||
<-m.done
|
||||
m.wg.Wait()
|
||||
}
|
||||
|
||||
// kickWorker nudges the worker to scan the queue immediately.
|
||||
func (m *Manager) kickWorker() {
|
||||
select {
|
||||
case m.kick <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// Enabled reports whether external delivery is configured on.
|
||||
func (m *Manager) Enabled() bool {
|
||||
return m.cfg.MaxPerDay > 0
|
||||
}
|
||||
|
||||
// MaxRecipients returns the maximum number of external recipients allowed
|
||||
// per message (0 means unlimited).
|
||||
func (m *Manager) MaxRecipients() int {
|
||||
return m.cfg.MaxRecipients
|
||||
}
|
||||
|
||||
// Enqueue validates a sender/recipient pair, DKIM-signs the message once and
|
||||
// stores it in the outbound queue for background delivery. The recipient must
|
||||
// NOT be a local address — callers decide local vs external routing.
|
||||
// Returns a permanent-style error for invalid input or rate-limit violations.
|
||||
func (m *Manager) Enqueue(senderUser *db.User, from, to string, raw []byte) (*db.OutboundMessage, error) {
|
||||
if !m.Enabled() {
|
||||
return nil, fmt.Errorf("外部投递未启用")
|
||||
}
|
||||
|
||||
to = strings.TrimSpace(to)
|
||||
addr, err := mail.ParseAddress(to)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("收件人地址无效: %s", to)
|
||||
}
|
||||
to = addr.Address
|
||||
|
||||
at := strings.LastIndex(to, "@")
|
||||
if at < 0 || at == len(to)-1 {
|
||||
return nil, fmt.Errorf("收件人地址无效: %s", to)
|
||||
}
|
||||
recipientDom := strings.ToLower(to[at+1:])
|
||||
|
||||
// Local addresses must never enter the outbound queue.
|
||||
if _, err := m.stores.Users.GetByEmail(to); err == nil {
|
||||
return nil, fmt.Errorf("收件人 %s 是本地地址，应走本地投递", to)
|
||||
}
|
||||
|
||||
// Rate limiting per sender user.
|
||||
if senderUser != nil {
|
||||
if err := m.checkRateLimit(senderUser.ID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// DKIM-sign once with the sender domain's key.
|
||||
signed, err := m.signForSender(from, raw)
|
||||
if err != nil {
|
||||
log.Printf("outbound: DKIM signing failed for %s: %v", from, err)
|
||||
signed = raw
|
||||
}
|
||||
|
||||
item := &db.OutboundMessage{
|
||||
MessageID: fmt.Sprintf("<%s@outbound>", uuid.New().String()),
|
||||
UserID: userIDOrZero(senderUser),
|
||||
FromAddr: from,
|
||||
ToAddr: to,
|
||||
RecipientDom: recipientDom,
|
||||
RawData: string(signed),
|
||||
Status: db.OutboundStatusPending,
|
||||
Attempts: 0,
|
||||
NextAttemptAt: time.Now(),
|
||||
}
|
||||
if err := m.stores.Outbound.Create(item); err != nil {
|
||||
return nil, fmt.Errorf("写入外发队列失败: %w", err)
|
||||
}
|
||||
|
||||
log.Printf("outbound: queued %s -> %s (id=%d)", from, to, item.ID)
|
||||
m.kickWorker()
|
||||
return item, nil
|
||||
}
|
||||
|
||||
func userIDOrZero(u *db.User) uint {
|
||||
if u == nil {
|
||||
return 0
|
||||
}
|
||||
return u.ID
|
||||
}
|
||||
|
||||
// signForSender looks up the sender domain's DKIM key and signs the message.
|
||||
func (m *Manager) signForSender(from string, raw []byte) ([]byte, error) {
|
||||
at := strings.LastIndex(from, "@")
|
||||
if at < 0 || at == len(from)-1 {
|
||||
return raw, fmt.Errorf("无效发件人地址: %s", from)
|
||||
}
|
||||
domName := strings.ToLower(from[at+1:])
|
||||
|
||||
domain, err := m.stores.Domains.GetByName(domName)
|
||||
if err != nil {
|
||||
return raw, nil // domain not managed locally; send unsigned
|
||||
}
|
||||
return SignDKIM(raw, domain.Name, domain.DkimSelector, domain.DkimPrivateKey)
|
||||
}
|
||||
|
||||
// checkRateLimit enforces per-user minute/day sending limits.
|
||||
func (m *Manager) checkRateLimit(userID uint) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
w := m.lim[userID]
|
||||
if w == nil || now.Sub(w.dayStart) >= 24*time.Hour {
|
||||
w = &userWindow{minuteStart: now, dayStart: now}
|
||||
m.lim[userID] = w
|
||||
} else if now.Sub(w.minuteStart) >= time.Minute {
|
||||
w.minuteStart = now
|
||||
w.minuteCount = 0
|
||||
}
|
||||
|
||||
if m.cfg.MaxPerMin > 0 && w.minuteCount >= m.cfg.MaxPerMin {
|
||||
log.Printf("outbound: rate limit exceeded (per minute) for user %d", userID)
|
||||
return fmt.Errorf("发送频率超限：每分钟最多 %d 封", m.cfg.MaxPerMin)
|
||||
}
|
||||
if m.cfg.MaxPerDay > 0 && w.dayCount >= m.cfg.MaxPerDay {
|
||||
log.Printf("outbound: rate limit exceeded (per day) for user %d", userID)
|
||||
return fmt.Errorf("发送频率超限：每日最多 %d 封", m.cfg.MaxPerDay)
|
||||
}
|
||||
|
||||
w.minuteCount++
|
||||
w.dayCount++
|
||||
return nil
|
||||
}
|
||||
|
||||
// processDue attempts delivery of all due queue items.
|
||||
func (m *Manager) processDue() {
|
||||
items, err := m.stores.Outbound.ListDue(time.Now(), m.batch)
|
||||
if err != nil {
|
||||
log.Printf("outbound: loading due queue failed: %v", err)
|
||||
return
|
||||
}
|
||||
for i := range items {
|
||||
m.deliverOne(&items[i])
|
||||
}
|
||||
}
|
||||
|
||||
// deliverOne performs a single delivery attempt for a queue item.
|
||||
func (m *Manager) deliverOne(item *db.OutboundMessage) {
|
||||
// Mark as sending to avoid concurrent workers double-delivering.
|
||||
item.Status = db.OutboundStatusSending
|
||||
if err := m.stores.Outbound.Update(item); err != nil {
|
||||
log.Printf("outbound: update item %d to sending failed: %v", item.ID, err)
|
||||
return
|
||||
}
|
||||
|
||||
resp, err := m.mailer.Deliver(item.FromAddr, item.ToAddr, []byte(item.RawData))
|
||||
|
||||
now := time.Now()
|
||||
item.Attempts++
|
||||
|
||||
if err == nil {
|
||||
item.Status = db.OutboundStatusSent
|
||||
item.LastResponse = resp
|
||||
item.LastError = ""
|
||||
item.CompletedAt = &now
|
||||
if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
|
||||
log.Printf("outbound: update item %d to sent failed: %v", item.ID, saveErr)
|
||||
return
|
||||
}
|
||||
log.Printf("outbound: delivered %s -> %s (id=%d, attempts=%d)", item.FromAddr, item.ToAddr, item.ID, item.Attempts)
|
||||
return
|
||||
}
|
||||
|
||||
var de *DeliveryError
|
||||
permanent := false
|
||||
if ok := asDeliveryError(err, &de); ok {
|
||||
permanent = de.Permanent
|
||||
}
|
||||
item.LastResponse = ""
|
||||
item.LastError = err.Error()
|
||||
|
||||
if permanent || item.Attempts >= m.cfg.MaxAttempts {
|
||||
item.Status = db.OutboundStatusFailed
|
||||
item.CompletedAt = &now
|
||||
if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
|
||||
log.Printf("outbound: update item %d to failed failed: %v", item.ID, saveErr)
|
||||
return
|
||||
}
|
||||
log.Printf("outbound: permanent failure %s -> %s (id=%d): %v", item.FromAddr, item.ToAddr, item.ID, err)
|
||||
m.bounce(item, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Temporary failure: exponential backoff retry.
|
||||
item.Status = db.OutboundStatusDeferred
|
||||
backoff := time.Duration(m.cfg.RetryBaseMin) * time.Minute
|
||||
backoff <<= (item.Attempts - 1)
|
||||
if backoff > 24*time.Hour {
|
||||
backoff = 24 * time.Hour
|
||||
}
|
||||
item.NextAttemptAt = now.Add(backoff)
|
||||
if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
|
||||
log.Printf("outbound: update item %d to deferred failed: %v", item.ID, saveErr)
|
||||
return
|
||||
}
|
||||
log.Printf("outbound: temporary failure %s -> %s (id=%d, attempt=%d, retry in %s): %v",
|
||||
item.FromAddr, item.ToAddr, item.ID, item.Attempts, backoff, err)
|
||||
}
|
||||
|
||||
// bounce delivers a non-delivery notice to the sender's INBOX.
|
||||
func (m *Manager) bounce(item *db.OutboundMessage, deliveryErr error) {
|
||||
sender, err := m.stores.Users.GetByEmail(item.FromAddr)
|
||||
if err != nil {
|
||||
log.Printf("outbound: cannot bounce %s: sender is not a local user", item.FromAddr)
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
postmaster := "Mail Delivery System <postmaster@" + m.hostname + ">"
|
||||
subject := fmt.Sprintf("邮件投递失败: %s", item.ToAddr)
|
||||
|
||||
body := "这是一封系统退信通知。\r\n\r\n" +
|
||||
fmt.Sprintf("您的邮件未能投递到以下收件人：\r\n\r\n 收件人：%s\r\n 失败原因：%s\r\n 投递时间：%s\r\n 尝试次数：%d\r\n\r\n",
|
||||
item.ToAddr, deliveryErr.Error(), now.Format("2006-01-02 15:04:05"), item.Attempts) +
|
||||
"如果收件人地址无误，请稍后重试；连续失败可能表示收件地址不存在或对方服务器拒收。\r\n"
|
||||
|
||||
msg := &db.Message{
|
||||
UserID: sender.ID,
|
||||
MessageID: fmt.Sprintf("<bounce-%d@%s>", item.ID, m.hostname),
|
||||
Folder: "INBOX",
|
||||
FromAddr: postmaster,
|
||||
ToAddr: item.FromAddr,
|
||||
Subject: subject,
|
||||
TextBody: body,
|
||||
Date: now,
|
||||
IsRead: false,
|
||||
}
|
||||
if err := m.stores.Mails.Create(msg); err != nil {
|
||||
log.Printf("outbound: bounce message creation failed: %v", err)
|
||||
return
|
||||
}
|
||||
log.Printf("outbound: bounce delivered to %s for failed delivery of %s", item.FromAddr, item.ToAddr)
|
||||
}
|
||||
|
||||
// Retry resets a failed/deferred queue item for immediate redelivery.
|
||||
func (m *Manager) Retry(id uint) error {
|
||||
item, err := m.stores.Outbound.GetByID(id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
item.Status = db.OutboundStatusPending
|
||||
item.Attempts = 0
|
||||
item.LastError = ""
|
||||
item.LastResponse = ""
|
||||
item.CompletedAt = nil
|
||||
item.NextAttemptAt = time.Now()
|
||||
if err := m.stores.Outbound.Update(item); err != nil {
|
||||
return err
|
||||
}
|
||||
m.kickWorker()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Cancel marks a queue item as canceled by the administrator.
|
||||
func (m *Manager) Cancel(id uint) error {
|
||||
item, err := m.stores.Outbound.GetByID(id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if item.Status == db.OutboundStatusSent {
|
||||
return fmt.Errorf("已送达的邮件无法取消")
|
||||
}
|
||||
now := time.Now()
|
||||
item.Status = db.OutboundStatusCanceled
|
||||
item.LastError = "管理员取消"
|
||||
item.CompletedAt = &now
|
||||
return m.stores.Outbound.Update(item)
|
||||
}
|
||||
|
||||
// asDeliveryError extracts a *DeliveryError from an error chain.
|
||||
func asDeliveryError(err error, target **DeliveryError) bool {
|
||||
for err != nil {
|
||||
if de, ok := err.(*DeliveryError); ok {
|
||||
*target = de
|
||||
return true
|
||||
}
|
||||
type unwrapper interface{ Unwrap() error }
|
||||
u, ok := err.(unwrapper)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
err = u.Unwrap()
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// StatusText returns a human-readable Chinese label for a queue status.
|
||||
func StatusText(status string) string {
|
||||
switch status {
|
||||
case db.OutboundStatusPending:
|
||||
return "待发送"
|
||||
case db.OutboundStatusSending:
|
||||
return "发送中"
|
||||
case db.OutboundStatusSent:
|
||||
return "已送达"
|
||||
case db.OutboundStatusDeferred:
|
||||
return "等待重试"
|
||||
case db.OutboundStatusFailed:
|
||||
return "失败"
|
||||
case db.OutboundStatusCanceled:
|
||||
return "已取消"
|
||||
default:
|
||||
return status
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package outbound
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto"
|
||||
"crypto/x509"
|
||||
"encoding/pem"
|
||||
"fmt"
|
||||
|
||||
"github.com/emersion/go-msgauth/dkim"
|
||||
)
|
||||
|
||||
// SignDKIM signs a raw RFC 5322 message with the domain's DKIM private key.
|
||||
// When the private key is empty the message is returned unchanged (unsigned).
|
||||
// The signature covers the standard header fields present in the message.
|
||||
func SignDKIM(raw []byte, domainName, selector, privateKeyPEM string) ([]byte, error) {
|
||||
if privateKeyPEM == "" {
|
||||
return raw, nil
|
||||
}
|
||||
if domainName == "" {
|
||||
return raw, fmt.Errorf("DKIM domain is empty")
|
||||
}
|
||||
if selector == "" {
|
||||
selector = "default"
|
||||
}
|
||||
|
||||
block, _ := pem.Decode([]byte(privateKeyPEM))
|
||||
if block == nil {
|
||||
return raw, fmt.Errorf("invalid DKIM private key PEM for %s", domainName)
|
||||
}
|
||||
|
||||
signer, err := parseSigner(block)
|
||||
if err != nil {
|
||||
return raw, fmt.Errorf("parse DKIM private key for %s: %v", domainName, err)
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
options := &dkim.SignOptions{
|
||||
Domain: domainName,
|
||||
Selector: selector,
|
||||
Signer: signer,
|
||||
}
|
||||
if err := dkim.Sign(&out, bytes.NewReader(raw), options); err != nil {
|
||||
return raw, fmt.Errorf("DKIM signing for %s failed: %v", domainName, err)
|
||||
}
|
||||
return out.Bytes(), nil
|
||||
}
|
||||
|
||||
// parseSigner parses an RSA private key PEM block into a crypto.Signer.
|
||||
// Both PKCS#1 ("RSA PRIVATE KEY") and PKCS#8 ("PRIVATE KEY") are supported.
|
||||
func parseSigner(block *pem.Block) (crypto.Signer, error) {
|
||||
// PKCS#1 is what the domain management form stores.
|
||||
if key, err := x509.ParsePKCS1PrivateKey(block.Bytes); err == nil {
|
||||
return key, nil
|
||||
}
|
||||
|
||||
// PKCS#8 fallback for externally generated keys.
|
||||
parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
signer, ok := parsed.(crypto.Signer)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unsupported private key type %T", parsed)
|
||||
}
|
||||
return signer, nil
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package outbound
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
dkimgen "mail_go/internal/dkim"
|
||||
|
||||
msgauthdkim "github.com/emersion/go-msgauth/dkim"
|
||||
)
|
||||
|
||||
func TestSignDKIMAndVerify(t *testing.T) {
|
||||
privPEM, pubPEM, err := dkimgen.GenerateKeyPair()
|
||||
if err != nil {
|
||||
t.Fatalf("GenerateKeyPair: %v", err)
|
||||
}
|
||||
_ = pubPEM
|
||||
|
||||
raw := []byte("From: kevin@lmve.net\r\nTo: someone@example.com\r\nSubject: hello\r\n\r\nbody\r\n")
|
||||
|
||||
signed, err := SignDKIM(raw, "lmve.net", "default", privPEM)
|
||||
if err != nil {
|
||||
t.Fatalf("SignDKIM: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(signed), "DKIM-Signature") {
|
||||
t.Fatalf("signed message missing DKIM-Signature header")
|
||||
}
|
||||
|
||||
verifications, err := msgauthdkim.Verify(strings.NewReader(string(signed)))
|
||||
if err != nil {
|
||||
t.Fatalf("dkim.Verify: %v", err)
|
||||
}
|
||||
if len(verifications) != 1 {
|
||||
t.Fatalf("expected 1 signature, got %d", len(verifications))
|
||||
}
|
||||
if verifications[0].Domain != "lmve.net" {
|
||||
t.Fatalf("unexpected signature: %+v", verifications[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestSignDKIMEmptyKeyReturnsUnsigned(t *testing.T) {
|
||||
raw := []byte("From: kevin@lmve.net\r\nTo: someone@example.com\r\nSubject: hello\r\n\r\nbody\r\n")
|
||||
out, err := SignDKIM(raw, "lmve.net", "default", "")
|
||||
if err != nil {
|
||||
t.Fatalf("SignDKIM with empty key should not fail: %v", err)
|
||||
}
|
||||
if string(out) != string(raw) {
|
||||
t.Fatalf("message changed despite empty key")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user