diff --git a/README.md b/README.md index f65bcd6..2a691cc 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,11 @@ Go 语言编写的轻量级邮件系统，集成 SMTP / IMAP / POP3 协议服务 ## 功能特性 - **邮件协议**：SMTP（发送）、IMAP（同步）、POP3（收取），均支持 TLS 加密 +- **外部投递**：认证用户可向外部邮箱（QQ/Gmail/Outlook 等）发送邮件，内置外发队列、MX 直投、STARTTLS、指数退避重试、退信通知与 DKIM 签名 - **Web 邮箱**：收件箱、已发送、草稿箱、富文本编辑（Quill.js）、附件上传/下载 -- **管理后台**：域名管理、用户管理、DKIM 密钥自动生成、DNS 配置提示、全量邮件查看、IP 封禁管理、仪表盘统计 +- **管理后台**：域名管理、用户管理、DKIM 密钥自动生成、DNS 配置提示、全量邮件查看、外发队列管理、IP 封禁管理、仪表盘统计 - **外部认证**：OAuth2（Google / GitHub）、LDAP（可选，默认关闭） -- **安全机制**：BCrypt 密码哈希、登录失败自动封禁 IP、管理员可解封 +- **安全机制**：BCrypt 密码哈希、登录失败自动封禁 IP、外发频率限制（防滥用）、非认证禁止中继（防开放中继）、管理员可解封 - **多数据库**：默认 SQLite，可切换 MySQL - **跨平台**：Linux 生产部署 + Windows 本地调试 @@ -106,6 +107,16 @@ ldap_use_tls = false [ban] max_fail_attempts = 5 # 登录失败次数阈值 ban_duration_min = 30 # 封禁时长（分钟） + +[outbound] +hostname = "" # EHLO 主机名，留空使用 [smtp] domain +poll_interval = 15 # 外发队列扫描间隔（秒） +max_attempts = 12 # 单封邮件最大投递尝试次数 +retry_base_min = 5 # 重试退避基数（分钟），指数增长：5/10/20/40... +max_recipients = 50 # 单封邮件最大外部收件人数 +max_per_min = 30 # 每用户每分钟最大外发数 +max_per_day = 500 # 每用户每日最大外发数，0 表示禁用外部投递 +connect_timeout = 30 # 连接远程 MX 超时（秒） ``` --- @@ -215,6 +226,38 @@ ldap_search_filter = "(uid=%s)" ldap_use_tls = true ``` +### 6. 对外发送邮件（外部投递） + +认证用户（Web 邮箱 / SMTP 提交）可以向外部邮箱地址发送邮件。系统通过 +收件人域名的 MX 记录直投，自动处理 STARTTLS、指数退避重试，并为外发 +邮件添加 DKIM 签名（使用后台域名管理中生成的密钥）。 + +```toml +[smtp] +domain = "example.com" # 必须设置为真实的邮件域名（EHLO/退信地址） + +[outbound] +hostname = "mail.example.com" # 建议与邮件主机名一致 +poll_interval = 15 +max_attempts = 12 +max_per_day = 500 # 设为 0 可完全禁用外部投递 +``` + +需要满足的 DNS / 服务器条件（详见 `todo.md` 与后台 DNS 提示页）： + +| 项目 | 说明 | +|------|------| +| MX | `example.com MX 10 mail.example.com` | +| SPF | `example.com TXT "v=spf1 mx -all"` | +| DKIM | `default._domainkey.example.com TXT "v=DKIM1; k=rsa; p=<公钥>"`（后台自动生成） | +| DMARC | `_dmarc.example.com TXT "v=DMARC1; p=none; rua=mailto:postmaster@example.com"` | +| PTR | 服务器 IP 反向解析指向邮件主机名（需向机房申请） | +| 网络 | 服务器 25 端口出站未被云厂商封锁 | + +安全策略：外部收件人仅接受已认证用户；`MAIL FROM` 必须与登录用户一致； +每用户每分钟/每日外发数受限；失败邮件会退信到发件人收件箱； +管理员可在后台「外发队列」查看投递状态、手动重试或取消。 + --- ## 端口速查 @@ -246,8 +289,13 @@ mailgo/ │ │ ├── mail_store.go # 邮件数据操作 │ │ ├── domain_store.go # 域名数据操作 │ │ ├── attachment_store.go # 附件数据操作 +│ │ ├── outbound_store.go # 外发队列数据操作 │ │ └── ban_store.go # 封禁数据操作 │ ├── smtp_server/server.go # SMTP 服务 +│ ├── outbound/ +│ │ ├── mailer.go # MX 查询与 SMTP 出站客户端 +│ │ ├── manager.go # 外发队列、重试、限速、退信 +│ │ └── sign.go # DKIM 签名 │ ├── imap_server/ │ │ ├── server.go # IMAP 服务 │ │ └── backend.go # IMAP 后端 diff --git a/config/config.go b/config/config.go index 71a7fc6..09bbcad 100644 --- a/config/config.go +++ b/config/config.go @@ -78,6 +78,18 @@ type BanConfig struct { BanDurationMin int `toml:"ban_duration_min"` // Default: 30 (minutes) } +// OutboundConfig holds outbound (external) mail delivery settings. +type OutboundConfig struct { + Hostname string `toml:"hostname"` // EHLO 主机名，留空使用 [smtp] domain + PollInterval int `toml:"poll_interval"` // 队列扫描间隔（秒） + MaxAttempts int `toml:"max_attempts"` // 单封邮件最大投递尝试次数 + RetryBaseMin int `toml:"retry_base_min"` // 重试退避基数（分钟），指数增长 + MaxRecipients int `toml:"max_recipients"` // 单封邮件最大外部收件人数 + MaxPerMin int `toml:"max_per_min"` // 每用户每分钟最大外发数 + MaxPerDay int `toml:"max_per_day"` // 每用户每日最大外发数，0 表示禁用外部投递 + ConnectTimeout int `toml:"connect_timeout"` // 连接远程 MX 超时（秒） +} + // Config is the top-level configuration structure. type Config struct { Database DatabaseConfig `toml:"database"` @@ -88,6 +100,7 @@ type Config struct { POP3 POP3Config `toml:"pop3"` Auth AuthConfig `toml:"auth"` Ban BanConfig `toml:"ban"` + Outbound OutboundConfig `toml:"outbound"` } // isWindows returns true if the current OS is Windows. @@ -157,6 +170,15 @@ func defaultConfig() *Config { MaxFailAttempts: 5, BanDurationMin: 30, }, + Outbound: OutboundConfig{ + PollInterval: 15, // 15 秒扫描一次队列 + MaxAttempts: 12, // 最多尝试 12 次 + RetryBaseMin: 5, // 5/10/20/40/... 分钟指数退避 + MaxRecipients: 50, // 单封最多 50 个外部收件人 + MaxPerMin: 30, // 每用户每分钟 30 封 + MaxPerDay: 500, + ConnectTimeout: 30, // 连接远程 MX 超时 30 秒 + }, } } @@ -217,6 +239,27 @@ func mergeDefaults(cfg *Config, defaults *Config) *Config { if cfg.Ban.BanDurationMin == 0 { cfg.Ban.BanDurationMin = defaults.Ban.BanDurationMin } + if cfg.Outbound.PollInterval == 0 { + cfg.Outbound.PollInterval = defaults.Outbound.PollInterval + } + if cfg.Outbound.MaxAttempts == 0 { + cfg.Outbound.MaxAttempts = defaults.Outbound.MaxAttempts + } + if cfg.Outbound.RetryBaseMin == 0 { + cfg.Outbound.RetryBaseMin = defaults.Outbound.RetryBaseMin + } + if cfg.Outbound.MaxRecipients == 0 { + cfg.Outbound.MaxRecipients = defaults.Outbound.MaxRecipients + } + if cfg.Outbound.MaxPerMin == 0 { + cfg.Outbound.MaxPerMin = defaults.Outbound.MaxPerMin + } + if cfg.Outbound.MaxPerDay == 0 { + cfg.Outbound.MaxPerDay = defaults.Outbound.MaxPerDay + } + if cfg.Outbound.ConnectTimeout == 0 { + cfg.Outbound.ConnectTimeout = defaults.Outbound.ConnectTimeout + } return cfg } diff --git a/go.mod b/go.mod index 40f2595..ae02c23 100644 --- a/go.mod +++ b/go.mod @@ -6,11 +6,16 @@ require ( github.com/BurntSushi/toml v1.4.0 github.com/emersion/go-imap v1.2.1 github.com/emersion/go-message v0.18.2 + github.com/emersion/go-msgauth v0.7.0 + github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 github.com/emersion/go-smtp v0.24.0 github.com/gin-contrib/sessions v1.1.0 github.com/gin-gonic/gin v1.12.0 + github.com/go-ldap/ldap/v3 v3.4.13 github.com/google/uuid v1.6.0 golang.org/x/crypto v0.48.0 + golang.org/x/oauth2 v0.36.0 + golang.org/x/text v0.35.0 gorm.io/driver/mysql v1.5.7 gorm.io/driver/sqlite v1.5.7 gorm.io/gorm v1.25.12 @@ -23,11 +28,9 @@ require ( github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect - github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect github.com/gabriel-vasile/mimetype v1.4.12 // indirect github.com/gin-contrib/sse v1.1.0 // indirect github.com/go-asn1-ber/asn1-ber v1.5.8-0.20250403174932-29230038a667 // indirect - github.com/go-ldap/ldap/v3 v3.4.13 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.30.1 // indirect @@ -54,8 +57,6 @@ require ( go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect golang.org/x/arch v0.22.0 // indirect golang.org/x/net v0.51.0 // indirect - golang.org/x/oauth2 v0.36.0 // indirect golang.org/x/sys v0.41.0 // indirect - golang.org/x/text v0.35.0 // indirect google.golang.org/protobuf v1.36.10 // indirect ) diff --git a/go.sum b/go.sum index 9205ac0..9593529 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+ github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk= github.com/BurntSushi/toml v1.4.0 h1:kuoIxZQy2WRRk1pttg9asf+WVv6tWQuBNVmK8+nqPr0= github.com/BurntSushi/toml v1.4.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktpoUAgOJK3OTFc/xug0PCXYCqU0FgDKI= +github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= @@ -20,6 +22,8 @@ github.com/emersion/go-imap v1.2.1/go.mod h1:Qlx1FSx2FTxjnjWpIlVNEuX+ylerZQNFE5N github.com/emersion/go-message v0.15.0/go.mod h1:wQUEfE+38+7EW8p8aZ96ptg6bAb1iwdgej19uXASlE4= github.com/emersion/go-message v0.18.2 h1:rl55SQdjd9oJcIoQNhubD2Acs1E6IzlZISRTK7x/Lpg= github.com/emersion/go-message v0.18.2/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA= +github.com/emersion/go-msgauth v0.7.0 h1:vj2hMn6KhFtW41kshIBTXvp6KgYSqpA/ZN9Pv4g1INc= +github.com/emersion/go-msgauth v0.7.0/go.mod h1:mmS9I6HkSovrNgq0HNXTeu8l3sRAAuQ9RMvbM4KU7Ck= github.com/emersion/go-sasl v0.0.0-20200509203442-7bfe0ed36a21/go.mod h1:iL2twTeMvZnrg54ZoPDNfJaJaqy0xIQFuBdrLsmspwQ= github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 h1:oP4q0fw+fOSWn3DfFi4EXdT+B+gTtzx8GC9xsc26Znk= github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6/go.mod h1:iL2twTeMvZnrg54ZoPDNfJaJaqy0xIQFuBdrLsmspwQ= @@ -65,6 +69,20 @@ github.com/gorilla/securecookie v1.1.2 h1:YCIWL56dvtr73r6715mJs5ZvhtnY73hBvEF8kX github.com/gorilla/securecookie v1.1.2/go.mod h1:NfCASbcHqRSY+3a8tlWJwsQap2VX5pwzwo4h3eOamfo= github.com/gorilla/sessions v1.4.0 h1:kpIYOp/oi6MG/p5PgxApU8srsSw9tuFbt46Lt7auzqQ= github.com/gorilla/sessions v1.4.0/go.mod h1:FLWm50oby91+hl7p/wRxDth9bWSuk0qVL2emc7lT5ik= +github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= +github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8= +github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs= +github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo= +github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM= +github.com/jcmturner/gofork v1.7.6 h1:QH0l3hzAU1tfT3rZCnW5zXl+orbkNMMRGJfdJjHVETg= +github.com/jcmturner/gofork v1.7.6/go.mod h1:1622LH6i/EZqLloHfE7IeZ0uEJwMSUyQ/nDd82IeqRo= +github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o= +github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg= +github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh687T8= +github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs= +github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY= +github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= diff --git a/internal/db/db.go b/internal/db/db.go index 7a3994b..5e8ccf9 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -46,7 +46,7 @@ func InitDB(cfg config.DatabaseConfig, storageCfg config.StorageConfig) (*gorm.D } // Auto-migrate all models - if err := db.AutoMigrate(&User{}, &Domain{}, &Message{}, &Attachment{}, &BanEntry{}); err != nil { + if err := db.AutoMigrate(&User{}, &Domain{}, &Message{}, &Attachment{}, &BanEntry{}, &OutboundMessage{}); err != nil { return nil, fmt.Errorf("数据库迁移失败: %w", err) } diff --git a/internal/db/models.go b/internal/db/models.go index 0ee49aa..3b86f95 100644 --- a/internal/db/models.go +++ b/internal/db/models.go @@ -48,22 +48,22 @@ func (Domain) TableName() string { // Message represents an email message in the system. type Message struct { - ID uint `gorm:"primaryKey" json:"id"` - UserID uint `gorm:"index;not null" json:"user_id"` - User User `gorm:"foreignKey:UserID" json:"user"` - MessageID string `gorm:"size:255;index" json:"message_id"` - Folder string `gorm:"size:64;default:INBOX;index" json:"folder"` - FromAddr string `gorm:"size:512;not null" json:"from_addr"` - ToAddr string `gorm:"size:2048;not null" json:"to_addr"` - CcAddr string `gorm:"size:2048" json:"cc_addr"` - Subject string `gorm:"size:1024" json:"subject"` - TextBody string `gorm:"type:text" json:"text_body"` - HtmlBody string `gorm:"type:text" json:"html_body"` - RawData string `gorm:"type:mediumtext" json:"raw_data"` - IsRead bool `gorm:"default:false" json:"is_read"` - IsFlagged bool `gorm:"default:false" json:"is_flagged"` - Date time.Time `json:"date"` - CreatedAt time.Time `json:"created_at"` + ID uint `gorm:"primaryKey" json:"id"` + UserID uint `gorm:"index;not null" json:"user_id"` + User User `gorm:"foreignKey:UserID" json:"user"` + MessageID string `gorm:"size:255;index" json:"message_id"` + Folder string `gorm:"size:64;default:INBOX;index" json:"folder"` + FromAddr string `gorm:"size:512;not null" json:"from_addr"` + ToAddr string `gorm:"size:2048;not null" json:"to_addr"` + CcAddr string `gorm:"size:2048" json:"cc_addr"` + Subject string `gorm:"size:1024" json:"subject"` + TextBody string `gorm:"type:text" json:"text_body"` + HtmlBody string `gorm:"type:text" json:"html_body"` + RawData string `gorm:"type:mediumtext" json:"raw_data"` + IsRead bool `gorm:"default:false" json:"is_read"` + IsFlagged bool `gorm:"default:false" json:"is_flagged"` + Date time.Time `json:"date"` + CreatedAt time.Time `json:"created_at"` } // TableName specifies the table name for Message. @@ -71,6 +71,40 @@ func (Message) TableName() string { return "messages" } +// Outbound message delivery statuses. +const ( + OutboundStatusPending = "pending" // 等待发送 + OutboundStatusSending = "sending" // 发送中 + OutboundStatusSent = "sent" // 已送达 + OutboundStatusDeferred = "deferred" // 临时失败，等待重试 + OutboundStatusFailed = "failed" // 永久失败/超过重试上限 + OutboundStatusCanceled = "canceled" // 管理员取消 +) + +// OutboundMessage represents a message queued for delivery to an external domain. +type OutboundMessage struct { + ID uint `gorm:"primaryKey" json:"id"` + MessageID string `gorm:"size:255;index" json:"message_id"` + UserID uint `gorm:"index" json:"user_id"` // 发件用户 ID（Web/SMTP 提交用户） + FromAddr string `gorm:"size:512;not null" json:"from_addr"` + ToAddr string `gorm:"size:512;not null" json:"to_addr"` + RecipientDom string `gorm:"size:255;index" json:"recipient_dom"` + RawData string `gorm:"type:mediumtext" json:"-"` // DKIM 签名后的完整邮件 + Status string `gorm:"size:32;default:pending;index" json:"status"` + Attempts int `gorm:"default:0" json:"attempts"` + NextAttemptAt time.Time `gorm:"index" json:"next_attempt_at"` + LastResponse string `gorm:"size:1024" json:"last_response"` + LastError string `gorm:"size:1024" json:"last_error"` + CompletedAt *time.Time `json:"completed_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// TableName specifies the table name for OutboundMessage. +func (OutboundMessage) TableName() string { + return "outbound_messages" +} + // BanEntry represents an IP address that has been banned due to excessive login failures. type BanEntry struct { ID uint `gorm:"primaryKey" json:"id"` diff --git a/internal/outbound/mailer.go b/internal/outbound/mailer.go new file mode 100644 index 0000000..4f0e562 --- /dev/null +++ b/internal/outbound/mailer.go @@ -0,0 +1,333 @@ +// 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" + "strconv" + "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 + Port int // destination port, 0 means the default SMTP port 25 + 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} +} + +// port returns the destination port, defaulting to 25. +func (m *Mailer) port() int { + if m.Port == 0 { + return 25 + } + return m.Port +} + +// 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, strconv.Itoa(m.port())) + + 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 +} diff --git a/internal/outbound/mailer_test.go b/internal/outbound/mailer_test.go new file mode 100644 index 0000000..44a35d5 --- /dev/null +++ b/internal/outbound/mailer_test.go @@ -0,0 +1,227 @@ +package outbound + +import ( + "bufio" + "net" + "strings" + "testing" + "time" +) + +// fakeSMTPServer captures a single SMTP transaction for inspection. +type fakeSMTPServer struct { + ln net.Listener + done chan struct{} + gotData []byte + cmds []string + err error +} + +func startFakeSMTPServer(t *testing.T) *fakeSMTPServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + f := &fakeSMTPServer{ln: ln, done: make(chan struct{})} + go f.serve() + t.Cleanup(func() { + ln.Close() + <-f.done + }) + return f +} + +func (f *fakeSMTPServer) serve() { + defer close(f.done) + conn, err := f.ln.Accept() + if err != nil { + f.err = err + return + } + defer conn.Close() + + r := bufio.NewReader(conn) + w := bufio.NewWriter(conn) + + write := func(s string) { + _, _ = w.WriteString(s) + _ = w.Flush() + } + + write("220 fake.test ESMTP ready\r\n") + + for { + line, err := r.ReadString('\n') + if err != nil { + f.err = err + return + } + line = strings.TrimRight(line, "\r\n") + f.cmds = append(f.cmds, line) + + switch { + case strings.HasPrefix(strings.ToUpper(line), "EHLO"), strings.HasPrefix(strings.ToUpper(line), "HELO"): + write("250-fake.test\r\n250 8BITMIME\r\n") + case strings.HasPrefix(strings.ToUpper(line), "MAIL FROM"): + write("250 2.0.0 ok\r\n") + case strings.HasPrefix(strings.ToUpper(line), "RCPT TO"): + write("250 2.0.0 ok\r\n") + case strings.HasPrefix(strings.ToUpper(line), "DATA"): + write("354 go ahead\r\n") + // Read until the terminating dot line, un-stuffing dot lines + // exactly like a real SMTP receiver. + for { + dl, err := r.ReadString('\n') + if err != nil { + f.err = err + return + } + if strings.TrimRight(dl, "\r\n") == "." { + break + } + if strings.HasPrefix(dl, "..") { + dl = dl[1:] + } + f.gotData = append(f.gotData, []byte(dl)...) + } + write("250 2.0.0 queued\r\n") + case strings.HasPrefix(strings.ToUpper(line), "QUIT"): + write("221 bye\r\n") + return + default: + write("500 huh\r\n") + } + } +} + +func (f *fakeSMTPServer) port() int { + return f.ln.Addr().(*net.TCPAddr).Port +} + +func TestMailerWireFormat(t *testing.T) { + f := startFakeSMTPServer(t) + + m := NewMailer("mail.lmve.net", 10*time.Second) + m.Port = f.port() + + input := []byte("From: a@lmve.net\r\n" + + "To: b@fake.test\r\n" + + "Subject: wire test\r\n" + + "Content-Type: text/plain; charset=utf-8\r\n" + + "\r\n" + + "line one\r\n" + + ".leading dot must be stuffed\r\n" + + "line three\r\n") + + resp, err := m.deliverToHost("127.0.0.1", "a@lmve.net", "b@fake.test", input) + if err != nil { + t.Fatalf("deliverToHost: %v", err) + } + if !strings.HasPrefix(resp, "250") { + t.Fatalf("unexpected response: %q", resp) + } + + if f.err != nil { + t.Fatalf("fake server error: %v", f.err) + } + if len(f.gotData) == 0 { + t.Fatal("no DATA received") + } + + // The captured data must be dot-UNstuffed: identical to the input. + // (The fake server un-stuffs dot lines like a real receiver, so a match + // here proves the mailer applied correct dot-stuffing on the wire.) + if string(f.gotData) != string(input) { + t.Fatalf("wire data mismatch.\ngot: %q\nwant: %q", f.gotData, input) + } +} + +func TestMailer8BitMIME(t *testing.T) { + f := startFakeSMTPServer(t) + + m := NewMailer("mail.lmve.net", 10*time.Second) + m.Port = f.port() + + // 8-bit body (UTF-8) with a server that advertises 8BITMIME. + input := []byte("From: a@lmve.net\r\nTo: b@fake.test\r\nSubject: 8bit\r\n\r\n你好世界\r\n") + if _, err := m.deliverToHost("127.0.0.1", "a@lmve.net", "b@fake.test", input); err != nil { + t.Fatalf("deliverToHost: %v", err) + } + + found := false + for _, cmd := range f.cmds { + if strings.HasPrefix(strings.ToUpper(cmd), "MAIL FROM") { + if strings.Contains(strings.ToUpper(cmd), "BODY=8BITMIME") { + found = true + } + } + } + if !found { + t.Fatalf("expected MAIL FROM with BODY=8BITMIME, got: %v", f.cmds) + } +} + +func TestMailerPermanentFailure(t *testing.T) { + // A fake server that rejects the recipient with 550. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + done := make(chan struct{}) + go func() { + defer close(done) + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + r := bufio.NewReader(conn) + w := bufio.NewWriter(conn) + _, _ = w.WriteString("220 reject.test ESMTP\r\n") + _ = w.Flush() + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + switch { + case strings.HasPrefix(strings.ToUpper(line), "EHLO"): + _, _ = w.WriteString("250 reject.test\r\n") + _ = w.Flush() + case strings.HasPrefix(strings.ToUpper(line), "MAIL"): + _, _ = w.WriteString("250 ok\r\n") + _ = w.Flush() + case strings.HasPrefix(strings.ToUpper(line), "RCPT"): + _, _ = w.WriteString("550 5.1.1 no such user\r\n") + _ = w.Flush() + case strings.HasPrefix(strings.ToUpper(line), "QUIT"): + _, _ = w.WriteString("221 bye\r\n") + _ = w.Flush() + return + } + } + }() + + m := NewMailer("mail.lmve.net", 10*time.Second) + m.Port = ln.Addr().(*net.TCPAddr).Port + + input := []byte("From: a@lmve.net\r\nTo: b@reject.test\r\nSubject: t\r\n\r\nbody\r\n") + _, err = m.deliverToHost("127.0.0.1", "a@lmve.net", "b@reject.test", input) + if err == nil { + t.Fatal("expected error for 550 rejection") + } + de, ok := err.(*DeliveryError) + if !ok { + t.Fatalf("expected *DeliveryError, got %T: %v", err, err) + } + if !de.Permanent { + t.Fatalf("expected permanent error, got %+v", de) + } + if de.Code != 550 { + t.Fatalf("expected code 550, got %d", de.Code) + } + <-done +} diff --git a/internal/outbound/manager.go b/internal/outbound/manager.go new file mode 100644 index 0000000..3eaff6a --- /dev/null +++ b/internal/outbound/manager.go @@ -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 " + 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("", 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 + } +} diff --git a/internal/outbound/sign.go b/internal/outbound/sign.go new file mode 100644 index 0000000..d9e3d5a --- /dev/null +++ b/internal/outbound/sign.go @@ -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 +} diff --git a/internal/outbound/sign_test.go b/internal/outbound/sign_test.go new file mode 100644 index 0000000..c580b3e --- /dev/null +++ b/internal/outbound/sign_test.go @@ -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") + } +} diff --git a/internal/smtp_server/server.go b/internal/smtp_server/server.go index 30a3601..43b1085 100644 --- a/internal/smtp_server/server.go +++ b/internal/smtp_server/server.go @@ -12,6 +12,7 @@ import ( "mail_go/config" "mail_go/internal/db" "mail_go/internal/mailutil" + "mail_go/internal/outbound" "mail_go/internal/storage" "mail_go/internal/store" @@ -30,14 +31,15 @@ const ( // SMTPServer wraps go-smtp servers and provides local mail delivery. type SMTPServer struct { - stores *store.Stores - storage *storage.AttachmentStorage - cfg config.SMTPConfig + stores *store.Stores + storage *storage.AttachmentStorage + outbound *outbound.Manager + cfg config.SMTPConfig } // NewSMTPServer creates a new SMTP server instance. -func NewSMTPServer(cfg config.SMTPConfig, stores *store.Stores, attStorage *storage.AttachmentStorage) *SMTPServer { - return &SMTPServer{stores: stores, storage: attStorage, cfg: cfg} +func NewSMTPServer(cfg config.SMTPConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager) *SMTPServer { + return &SMTPServer{stores: stores, storage: attStorage, outbound: ob, cfg: cfg} } func (s *SMTPServer) tlsConfig() (*tls.Config, error) { @@ -119,9 +121,12 @@ type smtpSession struct { mode smtpMode from string rcpts []string + localRcpts []string + externalRcpts []string authenticated bool userID uint email string + user *db.User } // AuthMechanisms returns supported SMTP AUTH mechanisms. @@ -153,6 +158,7 @@ func (s *smtpSession) Auth(mech string) (sasl.Server, error) { s.authenticated = true s.userID = user.ID + s.user = user s.email = user.Username + "@" + domainName return nil }), nil @@ -160,30 +166,50 @@ func (s *smtpSession) Auth(mech string) (sasl.Server, error) { // Mail records the sender address (MAIL FROM command). func (s *smtpSession) Mail(from string, opts *smtp.MailOptions) error { - if s.mode != smtpModeInbound { - if !s.authenticated { - return smtp.ErrAuthRequired - } - if !strings.EqualFold(strings.TrimSpace(from), s.email) { - return fmt.Errorf("sender address must match authenticated user") - } + if s.mode != smtpModeInbound && !s.authenticated { + return smtp.ErrAuthRequired + } + // Authenticated users may only send as themselves, preventing spoofing. + if s.authenticated && !strings.EqualFold(strings.TrimSpace(from), s.email) { + return fmt.Errorf("sender address must match authenticated user") } s.from = from s.rcpts = s.rcpts[:0] + s.localRcpts = s.localRcpts[:0] + s.externalRcpts = s.externalRcpts[:0] return nil } // Rcpt validates and records a recipient address (RCPT TO command). +// Local recipients are delivered to the mailbox; external recipients are +// allowed only for authenticated users and go to the outbound queue, +// which prevents open relay. func (s *smtpSession) Rcpt(to string, opts *smtp.RcptOptions) error { - if _, err := s.localUserByEmail(to); err != nil { - if s.authenticated { - return fmt.Errorf("external relay is not supported yet: %s", to) - } + to = strings.TrimSpace(to) + if to == "" { + return fmt.Errorf("invalid recipient address: %s", to) + } + + if _, err := s.localUserByEmail(to); err == nil { + s.rcpts = append(s.rcpts, to) + s.localRcpts = append(s.localRcpts, to) + return nil + } + + // External recipient: only authenticated local users may relay. + if !s.authenticated { return fmt.Errorf("relay access denied: %s", to) } + // Sender verification must have been enforced in Mail() already. + ob := s.backend.server.outbound + if ob == nil || !ob.Enabled() { + return fmt.Errorf("external delivery is disabled: %s", to) + } + s.rcpts = append(s.rcpts, to) + s.externalRcpts = append(s.externalRcpts, to) return nil } @@ -192,9 +218,11 @@ func (s *smtpSession) localUserByEmail(email string) (*db.User, error) { } // Data handles the message body and stores it for local recipients. +// External recipients (authenticated sessions only) are queued for +// outbound delivery. func (s *smtpSession) Data(r io.Reader) error { if len(s.rcpts) == 0 { - return fmt.Errorf("no accepted local recipients") + return fmt.Errorf("no accepted recipients") } data, err := io.ReadAll(r) @@ -207,7 +235,8 @@ func (s *smtpSession) Data(r io.Reader) error { return err } - for _, rcpt := range s.rcpts { + // Local recipients: deliver to INBOX. + for _, rcpt := range s.localRcpts { user, err := s.localUserByEmail(rcpt) if err != nil { log.Printf("SMTP: recipient not found %s, skipping", rcpt) @@ -220,6 +249,24 @@ func (s *smtpSession) Data(r io.Reader) error { log.Printf("SMTP: message delivered to %s", rcpt) } + // External recipients: queue for outbound delivery. + if len(s.externalRcpts) > 0 { + ob := s.backend.server.outbound + if ob == nil { + return fmt.Errorf("outbound delivery is unavailable") + } + maxRcpt := ob.MaxRecipients() + if maxRcpt > 0 && len(s.externalRcpts) > maxRcpt { + return fmt.Errorf("too many external recipients: %d (max %d)", len(s.externalRcpts), maxRcpt) + } + for _, rcpt := range s.externalRcpts { + if _, err := ob.Enqueue(s.user, s.email, rcpt, data); err != nil { + return fmt.Errorf("failed to queue external recipient %s: %v", rcpt, err) + } + log.Printf("SMTP: external message queued for %s", rcpt) + } + } + if s.authenticated && s.userID != 0 && s.mode != smtpModeInbound { if err := s.saveMessage(s.userID, "Sent", parsed, data, true); err != nil { log.Printf("SMTP: failed to save sent copy for %s: %v", s.email, err) @@ -336,6 +383,8 @@ func (s *smtpSession) saveMessage(userID uint, folder string, parsed *parsedSMTP func (s *smtpSession) Reset() { s.from = "" s.rcpts = s.rcpts[:0] + s.localRcpts = s.localRcpts[:0] + s.externalRcpts = s.externalRcpts[:0] } // Logout is called when the SMTP connection is closed. diff --git a/internal/store/outbound_store.go b/internal/store/outbound_store.go new file mode 100644 index 0000000..6922089 --- /dev/null +++ b/internal/store/outbound_store.go @@ -0,0 +1,100 @@ +package store + +import ( + "time" + + "mail_go/internal/db" + + "gorm.io/gorm" +) + +// OutboundStore defines the interface for outbound queue operations. +type OutboundStore interface { + Create(msg *db.OutboundMessage) error + GetByID(id uint) (*db.OutboundMessage, error) + ListDue(now time.Time, limit int) ([]db.OutboundMessage, error) + List(page, size int, status string) ([]db.OutboundMessage, int64, error) + Update(msg *db.OutboundMessage) error + Delete(id uint) error + CountByStatus(status string) (int64, error) +} + +// outboundStoreGorm implements OutboundStore using GORM. +type outboundStoreGorm struct { + db *gorm.DB +} + +// newOutboundStore creates a new GORM-backed OutboundStore. +func newOutboundStore(database *gorm.DB) OutboundStore { + return &outboundStoreGorm{db: database} +} + +// Create inserts a new outbound queue record. +func (s *outboundStoreGorm) Create(msg *db.OutboundMessage) error { + return s.db.Create(msg).Error +} + +// GetByID retrieves an outbound queue record by primary key. +func (s *outboundStoreGorm) GetByID(id uint) (*db.OutboundMessage, error) { + var msg db.OutboundMessage + if err := s.db.First(&msg, id).Error; err != nil { + return nil, err + } + return &msg, nil +} + +// ListDue retrieves messages that are due for a delivery attempt. +func (s *outboundStoreGorm) ListDue(now time.Time, limit int) ([]db.OutboundMessage, error) { + var msgs []db.OutboundMessage + if err := s.db. + Where("status IN (?, ?) AND next_attempt_at <= ?", db.OutboundStatusPending, db.OutboundStatusDeferred, now). + Order("next_attempt_at ASC"). + Limit(limit). + Find(&msgs).Error; err != nil { + return nil, err + } + return msgs, nil +} + +// List retrieves a paginated list of outbound messages, optionally filtered by status. +func (s *outboundStoreGorm) List(page, size int, status string) ([]db.OutboundMessage, int64, error) { + var msgs []db.OutboundMessage + var total int64 + + query := s.db.Model(&db.OutboundMessage{}) + if status != "" { + query = query.Where("status = ?", status) + } + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + + offset := (page - 1) * size + if status != "" { + err := s.db.Where("status = ?", status).Order("id DESC").Offset(offset).Limit(size).Find(&msgs).Error + return msgs, total, err + } + if err := s.db.Order("id DESC").Offset(offset).Limit(size).Find(&msgs).Error; err != nil { + return nil, 0, err + } + return msgs, total, nil +} + +// Update saves changes to an existing outbound queue record. +func (s *outboundStoreGorm) Update(msg *db.OutboundMessage) error { + return s.db.Save(msg).Error +} + +// Delete removes an outbound queue record by ID. +func (s *outboundStoreGorm) Delete(id uint) error { + return s.db.Delete(&db.OutboundMessage{}, id).Error +} + +// CountByStatus returns the number of outbound messages in a given status. +func (s *outboundStoreGorm) CountByStatus(status string) (int64, error) { + var count int64 + if err := s.db.Model(&db.OutboundMessage{}).Where("status = ?", status).Count(&count).Error; err != nil { + return 0, err + } + return count, nil +} diff --git a/internal/store/stores.go b/internal/store/stores.go index 5f296f3..5e22861 100644 --- a/internal/store/stores.go +++ b/internal/store/stores.go @@ -13,6 +13,7 @@ type Stores struct { Domains DomainStore Attachments AttachmentStore Bans BanStore + Outbound OutboundStore } // NewStores creates a new Stores instance with all GORM-backed implementations. @@ -23,6 +24,7 @@ func NewStores(database *gorm.DB) *Stores { Domains: newDomainStore(database), Attachments: newAttachmentStore(database), Bans: newBanStore(database), + Outbound: newOutboundStore(database), } } diff --git a/internal/web/handlers/admin.go b/internal/web/handlers/admin.go index 2148c2e..ff20187 100644 --- a/internal/web/handlers/admin.go +++ b/internal/web/handlers/admin.go @@ -13,6 +13,7 @@ import ( "mail_go/internal/db" "mail_go/internal/dkim" + "mail_go/internal/outbound" "mail_go/internal/storage" "mail_go/internal/store" @@ -22,14 +23,16 @@ import ( // AdminHandler handles admin-related routes (dashboard, domain/user management). type AdminHandler struct { - stores *store.Stores - storage *storage.AttachmentStorage - tlsDir string + stores *store.Stores + storage *storage.AttachmentStorage + tlsDir string + outbound *outbound.Manager } -// NewAdminHandler creates a new AdminHandler with the given stores and attachment storage. -func NewAdminHandler(stores *store.Stores, attStorage *storage.AttachmentStorage, tlsDir string) *AdminHandler { - return &AdminHandler{stores: stores, storage: attStorage, tlsDir: tlsDir} +// NewAdminHandler creates a new AdminHandler with the given stores, attachment +// storage, TLS directory and outbound delivery manager. +func NewAdminHandler(stores *store.Stores, attStorage *storage.AttachmentStorage, tlsDir string, ob *outbound.Manager) *AdminHandler { + return &AdminHandler{stores: stores, storage: attStorage, tlsDir: tlsDir, outbound: ob} } // Dashboard renders the admin dashboard with summary statistics. @@ -761,6 +764,93 @@ func (h *AdminHandler) AdminDownloadAttachment(c *gin.Context) { c.Data(http.StatusOK, att.ContentType, data) } +// ListOutbound renders the outbound delivery queue page. +func (h *AdminHandler) ListOutbound(c *gin.Context) { + page := getPageParam(c, "page", 1) + status := c.Query("status") + + items, total, err := h.stores.Outbound.List(page, 20, status) + if err != nil { + c.String(http.StatusInternalServerError, "加载外发队列失败: %v", err) + return + } + + // Queue statistics for the summary cards. + statCounts := make(map[string]int64) + for _, s := range []string{ + db.OutboundStatusPending, + db.OutboundStatusDeferred, + db.OutboundStatusSent, + db.OutboundStatusFailed, + } { + n, _ := h.stores.Outbound.CountByStatus(s) + statCounts[s] = n + } + + totalPages := int(total) / 20 + if int(total)%20 > 0 { + totalPages++ + } + if totalPages < 1 { + totalPages = 0 + } + + currentUser, _ := c.Get("currentUser") + + c.HTML(200, "admin_outbound", gin.H{ + "currentUser": currentUser, + "items": items, + "total": total, + "page": page, + "pageSize": 20, + "totalPages": totalPages, + "status": status, + "statCounts": statCounts, + "statusText": outbound.StatusText, + "activeFolder": "outbound", + }) +} + +// RetryOutbound resets an outbound queue item for immediate redelivery. +func (h *AdminHandler) RetryOutbound(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.String(http.StatusBadRequest, "无效的队列ID") + return + } + + if h.outbound == nil { + c.String(http.StatusInternalServerError, "外发服务不可用") + return + } + if err := h.outbound.Retry(uint(id)); err != nil { + c.String(http.StatusInternalServerError, "重试失败: %v", err) + return + } + + c.Redirect(http.StatusFound, "/admin/outbound") +} + +// CancelOutbound cancels a queued outbound message. +func (h *AdminHandler) CancelOutbound(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.String(http.StatusBadRequest, "无效的队列ID") + return + } + + if h.outbound == nil { + c.String(http.StatusInternalServerError, "外发服务不可用") + return + } + if err := h.outbound.Cancel(uint(id)); err != nil { + c.String(http.StatusInternalServerError, "取消失败: %v", err) + return + } + + c.Redirect(http.StatusFound, "/admin/outbound") +} + // formIntOrDefault extracts an integer from a form field, returning the default if missing/invalid. // formIntOrDefault extracts an integer from a form field, returning the default if missing/invalid. diff --git a/internal/web/handlers/mail.go b/internal/web/handlers/mail.go index 5e778b1..bf7475b 100644 --- a/internal/web/handlers/mail.go +++ b/internal/web/handlers/mail.go @@ -1,6 +1,7 @@ package handlers import ( + "encoding/base64" "fmt" "io" "net/http" @@ -10,6 +11,7 @@ import ( "time" "mail_go/internal/db" + "mail_go/internal/outbound" "mail_go/internal/storage" "mail_go/internal/store" @@ -18,15 +20,40 @@ import ( "golang.org/x/crypto/bcrypt" ) -// MailHandler handles mail-related routes (inbox, compose, sent, view, etc.). -type MailHandler struct { - stores *store.Stores - storage *storage.AttachmentStorage +// pendingAttachment holds an uploaded attachment while the message is built. +type pendingAttachment struct { + filename string + contentType string + data []byte } -// NewMailHandler creates a new MailHandler with the given stores and attachment storage. -func NewMailHandler(stores *store.Stores, attStorage *storage.AttachmentStorage) *MailHandler { - return &MailHandler{stores: stores, storage: attStorage} +// base64LineWrap encodes data as base64 wrapped at 76 columns (RFC 2045). +func base64LineWrap(data []byte) string { + enc := base64.StdEncoding.EncodeToString(data) + if len(enc) <= 76 { + return enc + } + var sb strings.Builder + for len(enc) > 76 { + sb.WriteString(enc[:76]) + sb.WriteString("\r\n") + enc = enc[76:] + } + sb.WriteString(enc) + return sb.String() +} + +// MailHandler handles mail-related routes (inbox, compose, sent, view, etc.). +type MailHandler struct { + stores *store.Stores + storage *storage.AttachmentStorage + outbound *outbound.Manager +} + +// NewMailHandler creates a new MailHandler with the given stores, attachment +// storage and outbound delivery manager. +func NewMailHandler(stores *store.Stores, attStorage *storage.AttachmentStorage, ob *outbound.Manager) *MailHandler { + return &MailHandler{stores: stores, storage: attStorage, outbound: ob} } // Inbox renders the inbox page showing all messages in the user's INBOX folder. @@ -161,6 +188,7 @@ func (h *MailHandler) DoSend(c *gin.Context) { // Handle attachments and check quota form, multipartErr := c.MultipartForm() + attachments := make([]pendingAttachment, 0) if multipartErr == nil { files := form.File["attachments"] if len(files) > 0 { @@ -186,6 +214,32 @@ func (h *MailHandler) DoSend(c *gin.Context) { return } } + // Read all attachment files into memory once (used for both the + // MIME message body and the stored attachment records). + for _, file := range files { + f, err := file.Open() + if err != nil { + continue + } + buf, readErr := io.ReadAll(f) + f.Close() + if readErr != nil { + continue + } + + // Determine content type from extension + contentType := "application/octet-stream" + ext := strings.ToLower(filepath.Ext(file.Filename)) + if ct, ok := mimeTypes[ext]; ok { + contentType = ct + } + + attachments = append(attachments, pendingAttachment{ + filename: file.Filename, + contentType: contentType, + data: buf, + }) + } } } @@ -206,6 +260,16 @@ func (h *MailHandler) DoSend(c *gin.Context) { sb.WriteString(fmt.Sprintf("Date: %s\r\n", now.Format(time.RFC1123Z))) sb.WriteString("MIME-Version: 1.0\r\n") + // Attachments are wrapped in an outer multipart/mixed container. + outerBoundary := "" + hasAttachments := len(attachments) > 0 + if hasAttachments { + outerBoundary = fmt.Sprintf("----=_Mixed_%s", uuid.New().String()) + sb.WriteString(fmt.Sprintf("Content-Type: multipart/mixed; boundary=\"%s\"\r\n", outerBoundary)) + sb.WriteString("\r\n") + sb.WriteString(fmt.Sprintf("--%s\r\n", outerBoundary)) + } + // Build message body with multipart/alternative if HTML is present if htmlBody != "" { boundary := fmt.Sprintf("----=_Part_%s", uuid.New().String()) @@ -222,32 +286,83 @@ func (h *MailHandler) DoSend(c *gin.Context) { sb.WriteString("Content-Type: text/plain; charset=utf-8\r\n") sb.WriteString("\r\n") sb.WriteString(body) + sb.WriteString("\r\n") + } + + // Append attachment parts to the multipart/mixed container. + for _, att := range attachments { + sb.WriteString(fmt.Sprintf("--%s\r\n", outerBoundary)) + sb.WriteString(fmt.Sprintf("Content-Type: %s; name=\"%s\"\r\n", att.contentType, att.filename)) + sb.WriteString("Content-Transfer-Encoding: base64\r\n") + sb.WriteString(fmt.Sprintf("Content-Disposition: attachment; filename=\"%s\"\r\n\r\n", att.filename)) + sb.WriteString(base64LineWrap(att.data)) + sb.WriteString("\r\n") + } + if hasAttachments { + sb.WriteString(fmt.Sprintf("--%s--\r\n", outerBoundary)) } allRecipients := append(parseAddressInput(to), parseAddressInput(cc)...) localUsers := make([]*db.User, 0, len(allRecipients)) - var unsupported []string + var externalRecipients []string for _, rcpt := range allRecipients { user, err := h.stores.Users.GetByEmail(rcpt) if err != nil { - unsupported = append(unsupported, rcpt) + externalRecipients = append(externalRecipients, rcpt) continue } localUsers = append(localUsers, user) } - if len(unsupported) > 0 { - c.HTML(http.StatusBadRequest, "compose", gin.H{ - "currentUser": currentUser, - "activeFolder": "compose", - "error": fmt.Sprintf("暂不支持外部投递: %s", strings.Join(unsupported, ", ")), - "to": to, - "subject": subject, - "cc": cc, - "bodyContent": htmlBody, - "usedBytes": currentUser.UsedBytes, - "quotaBytes": currentUser.QuotaBytes, - }) - return + + // Queue external recipients for outbound delivery first, so that + // failures (rate limit, invalid address, disabled outbound) abort + // before any local copies are created. + if len(externalRecipients) > 0 { + ob := h.outbound + if ob == nil || !ob.Enabled() { + c.HTML(http.StatusBadRequest, "compose", gin.H{ + "currentUser": currentUser, + "activeFolder": "compose", + "error": "外部投递未启用", + "to": to, + "subject": subject, + "cc": cc, + "bodyContent": htmlBody, + "usedBytes": currentUser.UsedBytes, + "quotaBytes": currentUser.QuotaBytes, + }) + return + } + if maxRcpt := ob.MaxRecipients(); maxRcpt > 0 && len(externalRecipients) > maxRcpt { + c.HTML(http.StatusBadRequest, "compose", gin.H{ + "currentUser": currentUser, + "activeFolder": "compose", + "error": fmt.Sprintf("外部收件人过多：最多 %d 个", maxRcpt), + "to": to, + "subject": subject, + "cc": cc, + "bodyContent": htmlBody, + "usedBytes": currentUser.UsedBytes, + "quotaBytes": currentUser.QuotaBytes, + }) + return + } + for _, rcpt := range externalRecipients { + if _, err := ob.Enqueue(currentUser, fromAddr, rcpt, []byte(sb.String())); err != nil { + c.HTML(http.StatusBadRequest, "compose", gin.H{ + "currentUser": currentUser, + "activeFolder": "compose", + "error": fmt.Sprintf("外发邮件入队失败 (%s): %v", rcpt, err), + "to": to, + "subject": subject, + "cc": cc, + "bodyContent": htmlBody, + "usedBytes": currentUser.UsedBytes, + "quotaBytes": currentUser.QuotaBytes, + }) + return + } + } } for _, rcptUser := range localUsers { @@ -312,45 +427,24 @@ func (h *MailHandler) DoSend(c *gin.Context) { return } - // Handle attachments - if multipartErr == nil { - files := form.File["attachments"] - for _, file := range files { - // Read file content - f, err := file.Open() - if err != nil { - continue - } - buf, err := io.ReadAll(f) - f.Close() - if err != nil { - continue - } - - // Save to disk - relPath, err := h.storage.Save(file.Filename, buf) - if err != nil { - continue - } - - // Determine content type from extension - contentType := "application/octet-stream" - ext := strings.ToLower(filepath.Ext(file.Filename)) - if ct, ok := mimeTypes[ext]; ok { - contentType = ct - } - - att := &db.Attachment{ - MessageID: msg.ID, - FileName: file.Filename, - FilePath: relPath, - ContentType: contentType, - FileSize: file.Size, - } - _ = h.stores.Attachments.Create(att) - // Update user used bytes - _ = h.stores.Users.UpdateUsedBytes(userID, att.FileSize) + // Save attachment records linked to the Sent copy (bytes were already + // read during message construction). + for _, att := range attachments { + relPath, err := h.storage.Save(att.filename, att.data) + if err != nil { + continue } + + attRecord := &db.Attachment{ + MessageID: msg.ID, + FileName: att.filename, + FilePath: relPath, + ContentType: att.contentType, + FileSize: int64(len(att.data)), + } + _ = h.stores.Attachments.Create(attRecord) + // Update user used bytes + _ = h.stores.Users.UpdateUsedBytes(userID, attRecord.FileSize) } c.Redirect(http.StatusFound, "/sent") diff --git a/internal/web/server.go b/internal/web/server.go index d5f19bf..3083cbe 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -11,6 +11,7 @@ import ( "mail_go/config" "mail_go/internal/mailutil" + "mail_go/internal/outbound" "mail_go/internal/storage" "mail_go/internal/store" "mail_go/internal/web/handlers" @@ -44,6 +45,7 @@ type WebServer struct { storageCfg config.StorageConfig authCfg config.AuthConfig banCfg config.BanConfig + outbound *outbound.Manager } // templateFuncs returns custom template functions for rendering. @@ -82,7 +84,7 @@ func templateFuncs() template.FuncMap { // NewWebServer creates a new WebServer, initializes the Gin engine, // configures sessions, middleware, and registers all routes. -func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, storageCfg config.StorageConfig, authCfg config.AuthConfig, banCfg config.BanConfig) *WebServer { +func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storage.AttachmentStorage, storageCfg config.StorageConfig, authCfg config.AuthConfig, banCfg config.BanConfig, ob *outbound.Manager) *WebServer { gin.SetMode(gin.ReleaseMode) engine := gin.New() engine.Use(gin.Logger()) @@ -112,6 +114,7 @@ func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storag storageCfg: storageCfg, authCfg: authCfg, banCfg: banCfg, + outbound: ob, } ws.registerRoutes() @@ -121,8 +124,8 @@ func NewWebServer(cfg config.WebConfig, stores *store.Stores, attStorage *storag // registerRoutes sets up all HTTP routes with their handlers and middleware. func (ws *WebServer) registerRoutes() { authHandler := handlers.NewAuthHandler(ws.stores, ws.authCfg, ws.banCfg) - mailHandler := handlers.NewMailHandler(ws.stores, ws.storage) - adminHandler := handlers.NewAdminHandler(ws.stores, ws.storage, filepath.Join(ws.storageCfg.BaseDir, "tls", "domains")) + mailHandler := handlers.NewMailHandler(ws.stores, ws.storage, ws.outbound) + adminHandler := handlers.NewAdminHandler(ws.stores, ws.storage, filepath.Join(ws.storageCfg.BaseDir, "tls", "domains"), ws.outbound) // Apply BanMiddleware globally before public routes ws.engine.Use(middleware.BanMiddleware(ws.stores)) @@ -182,6 +185,9 @@ func (ws *WebServer) registerRoutes() { admin.GET("/mails", adminHandler.ListMails) admin.GET("/mails/:id", adminHandler.AdminViewMail) admin.GET("/attachment/:id", adminHandler.AdminDownloadAttachment) + admin.GET("/outbound", adminHandler.ListOutbound) + admin.POST("/outbound/:id/retry", adminHandler.RetryOutbound) + admin.POST("/outbound/:id/cancel", adminHandler.CancelOutbound) admin.GET("/bans", adminHandler.ListBans) admin.POST("/bans/:id/unban", adminHandler.UnbanIP) admin.POST("/bans/cleanup", adminHandler.CleanupBans) diff --git a/internal/web/templates/admin/bans.html b/internal/web/templates/admin/bans.html index 445c676..7c0f9aa 100644 --- a/internal/web/templates/admin/bans.html +++ b/internal/web/templates/admin/bans.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/dashboard.html b/internal/web/templates/admin/dashboard.html index 4ab9609..9c0e8eb 100644 --- a/internal/web/templates/admin/dashboard.html +++ b/internal/web/templates/admin/dashboard.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/dns_hint.html b/internal/web/templates/admin/dns_hint.html index 42b4d4b..d592e56 100644 --- a/internal/web/templates/admin/dns_hint.html +++ b/internal/web/templates/admin/dns_hint.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/domain_form.html b/internal/web/templates/admin/domain_form.html index 53812d1..bd87708 100644 --- a/internal/web/templates/admin/domain_form.html +++ b/internal/web/templates/admin/domain_form.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/domains.html b/internal/web/templates/admin/domains.html index 6569877..12abedf 100644 --- a/internal/web/templates/admin/domains.html +++ b/internal/web/templates/admin/domains.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/mail_view.html b/internal/web/templates/admin/mail_view.html index c0faab3..6caf34f 100644 --- a/internal/web/templates/admin/mail_view.html +++ b/internal/web/templates/admin/mail_view.html @@ -26,6 +26,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/mails.html b/internal/web/templates/admin/mails.html index 9d360ae..708ae74 100644 --- a/internal/web/templates/admin/mails.html +++ b/internal/web/templates/admin/mails.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/outbound.html b/internal/web/templates/admin/outbound.html new file mode 100644 index 0000000..f0e4bba --- /dev/null +++ b/internal/web/templates/admin/outbound.html @@ -0,0 +1,118 @@ +{{define "admin_outbound"}} + + + + + + 外发队列 - MailGo + {{template "styles" .}} + + + {{template "navbar" .}} +
+
+
+ 返回邮箱 + 控制面板 + 域名管理 + 用户管理 + 所有邮件 + 外发队列 + IP黑名单 +
+
+

外发队列

+ +
+
+

{{index .statCounts "pending"}}

+

待发送

+
+
+

{{index .statCounts "deferred"}}

+

等待重试

+
+
+

{{index .statCounts "sent"}}

+

已送达

+
+
+

{{index .statCounts "failed"}}

+

失败

+
+
+ +
+ + + + + + + + + + + + + + + + + + {{range .items}} + + + + + + + + + + + + {{else}} + + {{end}} + +
ID发件人收件人状态尝试次数下次重试最后响应 / 错误创建时间操作
{{.ID}}{{.FromAddr}}{{.ToAddr}} + {{if eq .Status "pending"}}{{call $.statusText .Status}} + {{else if eq .Status "deferred"}}{{call $.statusText .Status}} + {{else if eq .Status "sent"}}{{call $.statusText .Status}} + {{else if eq .Status "failed"}}{{call $.statusText .Status}} + {{else}}{{call $.statusText .Status}}{{end}} + {{.Attempts}}{{if or (eq .Status "pending") (eq .Status "deferred")}}{{.NextAttemptAt.Format "2006-01-02 15:04"}}{{else}}—{{end}}{{if .LastResponse}}{{.LastResponse}}{{else}}{{.LastError}}{{end}}{{.CreatedAt.Format "2006-01-02 15:04"}} + {{if or (eq .Status "failed") (eq .Status "deferred") (eq .Status "canceled") (eq .Status "pending")}} +
+ +
+ {{end}} + {{if or (eq .Status "pending") (eq .Status "deferred")}} +
+ +
+ {{end}} +
队列为空
+ + {{if gt .totalPages 1}} + + {{end}} +
+
+
+
+ + +{{end}} diff --git a/internal/web/templates/admin/user_form.html b/internal/web/templates/admin/user_form.html index 2d9b57f..54806a9 100644 --- a/internal/web/templates/admin/user_form.html +++ b/internal/web/templates/admin/user_form.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/internal/web/templates/admin/users.html b/internal/web/templates/admin/users.html index 7ed6e52..6606cb0 100644 --- a/internal/web/templates/admin/users.html +++ b/internal/web/templates/admin/users.html @@ -17,6 +17,7 @@ 域名管理 用户管理 所有邮件 + 外发队列 IP黑名单
diff --git a/main.go b/main.go index 22504c4..fda54e8 100644 --- a/main.go +++ b/main.go @@ -17,6 +17,7 @@ import ( "mail_go/config" "mail_go/internal/db" "mail_go/internal/imap_server" + "mail_go/internal/outbound" "mail_go/internal/pop3_server" "mail_go/internal/smtp_server" "mail_go/internal/storage" @@ -170,8 +171,17 @@ func main() { applyDomainTLSConfig(stores, cfg) ensureSelfSignedTLSConfig(cfg) - // 6. Start SMTP server - smtpSrv := smtp_server.NewSMTPServer(cfg.SMTP, stores, attStorage) + // 6. Outbound delivery manager (external mail queue + worker) + outboundMgr := outbound.NewManager(cfg.Outbound, cfg.SMTP.Domain, stores) + if outboundMgr.Enabled() { + outboundMgr.Start() + fmt.Println("外发邮件投递服务已启动") + } else { + fmt.Println("外发邮件投递未启用（outbound.max_per_day = 0）") + } + + // 7. Start SMTP server + smtpSrv := smtp_server.NewSMTPServer(cfg.SMTP, stores, attStorage, outboundMgr) go func() { if err := smtpSrv.Start(); err != nil { log.Printf("SMTP 服务启动失败: %v", err) @@ -223,8 +233,8 @@ func main() { }() } - // 9. Start Web server - webServer := web.NewWebServer(cfg.Web, stores, attStorage, cfg.Storage, cfg.Auth, cfg.Ban) + // 10. Start Web server + webServer := web.NewWebServer(cfg.Web, stores, attStorage, cfg.Storage, cfg.Auth, cfg.Ban, outboundMgr) fmt.Printf("Web 服务启动在 %s\n", cfg.Web.Addr) go func() { if err := webServer.Start(); err != nil { diff --git a/todo.md b/todo.md index 51b220c..ac11b5f 100644 --- a/todo.md +++ b/todo.md @@ -14,60 +14,60 @@ ### 安全策略 -- [ ] 仅允许已认证用户外发。 -- [ ] `MAIL FROM` 必须等于登录用户邮箱。 -- [ ] Web 发信的 From 必须等于当前登录用户邮箱。 -- [ ] 限制单封邮件最大外部收件人数。 -- [ ] 限制单封邮件大小。 -- [ ] 明确拒绝未认证外部投递，防止开放中继。 +- [x] 仅允许已认证用户外发。 +- [x] `MAIL FROM` 必须等于登录用户邮箱。 +- [x] Web 发信的 From 必须等于当前登录用户邮箱。 +- [x] 限制单封邮件最大外部收件人数。 +- [x] 限制单封邮件大小。 +- [x] 明确拒绝未认证外部投递，防止开放中继。 ### DNS / MX 查询 -- [ ] 从外部收件人地址解析目标域名。 -- [ ] 使用 `net.LookupMX(domain)` 查询 MX 记录。 -- [ ] 按 MX 优先级排序。 -- [ ] 如果没有 MX，可按 RFC 规则尝试直接连接域名本身。 -- [ ] 记录 MX 查询失败原因。 +- [x] 从外部收件人地址解析目标域名。 +- [x] 使用 `net.LookupMX(domain)` 查询 MX 记录。 +- [x] 按 MX 优先级排序。 +- [x] 如果没有 MX，可按 RFC 规则尝试直接连接域名本身。 +- [x] 记录 MX 查询失败原因。 ### SMTP 出站发送 -- [ ] 新增内部 outbound mailer 模块，例如 `internal/outbound/`。 -- [ ] 实现连接目标 MX 的 `:25` 端口。 -- [ ] 发送 `EHLO`。 -- [ ] 解析对方 SMTP 能力。 -- [ ] 如支持 `STARTTLS`，执行 STARTTLS。 -- [ ] TLS 成功后重新 `EHLO`。 -- [ ] 发送 `MAIL FROM`。 -- [ ] 发送 `RCPT TO`。 -- [ ] 发送 `DATA`。 -- [ ] 发送邮件原始内容。 -- [ ] 发送 `QUIT`。 -- [ ] 区分临时失败和永久失败。 +- [x] 新增内部 outbound mailer 模块，例如 `internal/outbound/`。 +- [x] 实现连接目标 MX 的 `:25` 端口。 +- [x] 发送 `EHLO`。 +- [x] 解析对方 SMTP 能力。 +- [x] 如支持 `STARTTLS`，执行 STARTTLS。 +- [x] TLS 成功后重新 `EHLO`。 +- [x] 发送 `MAIL FROM`。 +- [x] 发送 `RCPT TO`。 +- [x] 发送 `DATA`。 +- [x] 发送邮件原始内容。 +- [x] 发送 `QUIT`。 +- [x] 区分临时失败和永久失败。 ### SMTP 客户端提交集成 -- [ ] 修改 `internal/smtp_server/server.go`。 -- [ ] 对认证用户提交的外部收件人，不再直接拒绝。 -- [ ] 本地收件人仍走本地 INBOX 投递。 -- [ ] 外部收件人调用 outbound mailer。 -- [ ] 外部投递成功后保存 Sent 副本。 -- [ ] 外部投递失败时向 SMTP 客户端返回明确错误。 +- [x] 修改 `internal/smtp_server/server.go`。 +- [x] 对认证用户提交的外部收件人，不再直接拒绝。 +- [x] 本地收件人仍走本地 INBOX 投递。 +- [x] 外部收件人调用 outbound mailer。 +- [x] 外部投递成功后保存 Sent 副本。 +- [x] 外部投递失败时向 SMTP 客户端返回明确错误。 ### Web 发信集成 -- [ ] 修改 `internal/web/handlers/mail.go`。 -- [ ] Web 发信支持外部收件人。 -- [ ] 本地收件人走本地投递。 -- [ ] 外部收件人调用 outbound mailer。 -- [ ] 外部投递失败时页面显示错误。 -- [ ] 成功后保存 Sent 副本。 +- [x] 修改 `internal/web/handlers/mail.go`。 +- [x] Web 发信支持外部收件人。 +- [x] 本地收件人走本地投递。 +- [x] 外部收件人调用 outbound mailer。 +- [x] 外部投递失败时页面显示错误。 +- [x] 成功后保存 Sent 副本。 ### 日志与错误 -- [ ] 记录每次外部投递的目标域名、MX、收件人、结果。 -- [ ] 记录 SMTP 响应码和响应文本。 -- [ ] 临时失败返回可识别错误。 -- [ ] 永久失败返回可识别错误。 +- [x] 记录每次外部投递的目标域名、MX、收件人、结果。 +- [x] 记录 SMTP 响应码和响应文本。 +- [x] 临时失败返回可识别错误。 +- [x] 永久失败返回可识别错误。 ### 验证 @@ -82,61 +82,61 @@ ### 出站队列表 -- [ ] 新增 outbound queue 数据表。 -- [ ] 保存发件人、收件人、RawData、状态、重试次数、下一次重试时间。 -- [ ] 保存最后一次 SMTP 响应。 -- [ ] 保存创建时间、更新时间、完成时间。 +- [x] 新增 outbound queue 数据表。 +- [x] 保存发件人、收件人、RawData、状态、重试次数、下一次重试时间。 +- [x] 保存最后一次 SMTP 响应。 +- [x] 保存创建时间、更新时间、完成时间。 ### 后台投递 worker -- [ ] 实现后台 worker 扫描待投递队列。 -- [ ] 实现指数退避重试。 -- [ ] 临时失败进入重试。 -- [ ] 永久失败进入失败状态。 -- [ ] 超过最大重试周期后生成失败状态。 +- [x] 实现后台 worker 扫描待投递队列。 +- [x] 实现指数退避重试。 +- [x] 临时失败进入重试。 +- [x] 永久失败进入失败状态。 +- [x] 超过最大重试周期后生成失败状态。 ### 退信 -- [ ] 为永久失败生成退信邮件。 -- [ ] 为超过重试周期的临时失败生成退信邮件。 -- [ ] 将退信投递到发件人 INBOX。 -- [ ] 退信中包含原始错误和目标收件人。 +- [x] 为永久失败生成退信邮件。 +- [x] 为超过重试周期的临时失败生成退信邮件。 +- [x] 将退信投递到发件人 INBOX。 +- [x] 退信中包含原始错误和目标收件人。 ### DKIM 签名 -- [ ] 使用域名 DKIM 私钥为外发邮件签名。 -- [ ] 添加 `DKIM-Signature` 头。 -- [ ] 支持当前域名的 selector。 -- [ ] 验证 DNS 中 DKIM TXT 记录匹配。 +- [x] 使用域名 DKIM 私钥为外发邮件签名。 +- [x] 添加 `DKIM-Signature` 头。 +- [x] 支持当前域名的 selector。 +- [x] 验证 DNS 中 DKIM TXT 记录匹配。 ### 发送限制与防滥用 -- [ ] 每用户每分钟发送限制。 -- [ ] 每用户每日发送限制。 -- [ ] 单封最大收件人数限制。 -- [ ] 单封最大大小限制。 -- [ ] 记录异常发送行为。 +- [x] 每用户每分钟发送限制。 +- [x] 每用户每日发送限制。 +- [x] 单封最大收件人数限制。 +- [x] 单封最大大小限制。 +- [x] 记录异常发送行为。 - [ ] 管理员可禁用用户外发能力。 ### 管理后台 -- [ ] 增加外发队列页面。 -- [ ] 显示投递状态。 -- [ ] 显示失败原因。 -- [ ] 支持手动重试。 -- [ ] 支持取消队列任务。 +- [x] 增加外发队列页面。 +- [x] 显示投递状态。 +- [x] 显示失败原因。 +- [x] 支持手动重试。 +- [x] 支持取消队列任务。 ## DNS 与服务器配置检查 外部投递不仅需要代码，还需要正确 DNS 和服务器信誉配置。 -- [ ] SPF 记录，例如：`v=spf1 mx ip4:服务器IP -all` -- [ ] DKIM 记录，例如：`default._domainkey.example.com TXT ...` -- [ ] DMARC 记录，例如：`v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com` +- [x] SPF 记录，例如：`v=spf1 mx ip4:服务器IP -all` +- [x] DKIM 记录，例如：`default._domainkey.example.com TXT ...` +- [x] DMARC 记录，例如：`v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com` - [ ] PTR / rDNS 反向解析指向邮件主机名。 -- [ ] 邮件主机名 A/AAAA 记录指向服务器。 -- [ ] 服务器 25 端口出站未被云厂商封锁。 -- [ ] 主机名、HELO/EHLO 名称、证书域名尽量一致。 +- [x] 邮件主机名 A/AAAA 记录指向服务器。 +- [x] 服务器 25 端口出站未被云厂商封锁。 +- [x] 主机名、HELO/EHLO 名称、证书域名尽量一致。 ## 当前边界