diff --git a/README.md b/README.md index 524bdd7..d5650cf 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,16 @@ docker compose -f docker-compose.yml -f docker-compose.build.yml up -d --build - 云厂商常默认封禁 25 端口；无法收发公网邮件时先检查端口、安全组、防火墙与反向 DNS。 - SQLite 适合单机部署；多节点部署前需要迁移数据库，并同步调整 Postfix/Dovecot 查询配置。 +## SMTP 提交 + +- 第三方客户端的 SMTP 提交 `465/587` 由 LanQin API 进程处理。 +- 启用 SMTP 提交前必须配置 `LANQIN_TLS_CERT_FILE` / `LANQIN_TLS_KEY_FILE`；API 不会用 localhost 自签证书对外提供 465/587。 +- Postfix 只保留 `25` 端口，用于公网入站邮件和内部/外部 relay。 +- Webmail/API 和第三方客户端发信都会先写入 Sent，再进入发送队列。 +- 发送队列由 LanQin API 后台 worker relay 到 `LANQIN_SMTP_HOST:LANQIN_SMTP_PORT`，失败会记录审计并按退避策略重试。 +- v1 支持本人邮箱发信；如需 send-as，可使用启用的别名转发 source 指向本人邮箱，或在数据库中配置 `send_as_grants`。 +- 如果客户端随后又通过 IMAP APPEND 写入自己的 Sent 副本，Maildir 同步会按 Sent 文件夹内的 `Message-ID` 去重。 + ## License [MIT](./LICENSE) diff --git a/apps/api/cmd/server/main.go b/apps/api/cmd/server/main.go index 049491d..da98ec2 100644 --- a/apps/api/cmd/server/main.go +++ b/apps/api/cmd/server/main.go @@ -7,9 +7,12 @@ import ( "net/http" "os" "os/signal" + "strings" "syscall" "time" + smtpserver "github.com/emersion/go-smtp" + "lanqin-email-api/internal/app" ) @@ -29,6 +32,15 @@ func main() { Handler: svc.Router(), ReadHeaderTimeout: 10 * time.Second, } + submissionServers := &app.SubmissionServers{} + if strings.TrimSpace(cfg.SubmissionAddr) != "" || strings.TrimSpace(cfg.SubmissionTLSAddr) != "" { + tlsConfig, err := app.LoadServerTLSConfig(cfg) + if err != nil { + logger.Error("failed to initialize TLS config", "error", err) + os.Exit(1) + } + submissionServers = svc.NewSubmissionServers(tlsConfig) + } go func() { logger.Info("LanQin API listening", "addr", cfg.Addr) @@ -37,6 +49,24 @@ func main() { os.Exit(1) } }() + if submissionServers.Plain != nil { + go func() { + logger.Info("LanQin SMTP submission listening", "addr", cfg.SubmissionAddr) + if err := submissionServers.Plain.ListenAndServe(); err != nil && !errors.Is(err, smtpserver.ErrServerClosed) { + logger.Error("smtp submission server stopped unexpectedly", "error", err) + os.Exit(1) + } + }() + } + if submissionServers.TLS != nil { + go func() { + logger.Info("LanQin SMTP implicit TLS submission listening", "addr", cfg.SubmissionTLSAddr) + if err := submissionServers.TLS.ListenAndServeTLS(); err != nil && !errors.Is(err, smtpserver.ErrServerClosed) { + logger.Error("smtp tls submission server stopped unexpectedly", "error", err) + os.Exit(1) + } + }() + } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() @@ -48,5 +78,9 @@ func main() { logger.Error("server shutdown failed", "error", err) os.Exit(1) } + if err := submissionServers.Shutdown(shutdownCtx); err != nil { + logger.Error("smtp submission shutdown failed", "error", err) + os.Exit(1) + } logger.Info("server stopped") } diff --git a/apps/api/go.mod b/apps/api/go.mod index c31dc3e..f153bf9 100644 --- a/apps/api/go.mod +++ b/apps/api/go.mod @@ -13,6 +13,8 @@ require ( require ( github.com/aymerick/douceur v0.2.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect + github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect + github.com/emersion/go-smtp v0.24.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/gorilla/css v1.0.1 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect diff --git a/apps/api/go.sum b/apps/api/go.sum index 7ce8dc0..ff88e17 100644 --- a/apps/api/go.sum +++ b/apps/api/go.sum @@ -2,6 +2,10 @@ github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuP github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +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= +github.com/emersion/go-smtp v0.24.0 h1:g6AfoF140mvW0vLNPD/LuCBLEAdlxOjIXqbIkJIS6Wk= +github.com/emersion/go-smtp v0.24.0/go.mod h1:ZtRRkbTyp2XTHCA+BmyTFTrj8xY4I+b4McvHxCU2gsQ= github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw= github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8= github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo= diff --git a/apps/api/internal/app/app.go b/apps/api/internal/app/app.go index 5ff360d..5069cb4 100644 --- a/apps/api/internal/app/app.go +++ b/apps/api/internal/app/app.go @@ -74,6 +74,7 @@ func New(cfg Config, logger *slog.Logger) (*App, error) { if strings.TrimSpace(a.cfg.MaildirRoot) != "" { go a.maildirWorker(workerCtx) } + go a.sendQueueWorker(workerCtx) go a.smtpEventsCleanupWorker(workerCtx) return a, nil } @@ -233,6 +234,60 @@ func (a *App) migrate(ctx context.Context) error { `CREATE INDEX IF NOT EXISTS idx_messages_search ON messages(mailbox_id, subject, from_addr, from_name, snippet)`, `CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_mailbox_raw_path ON messages(mailbox_id, raw_path) WHERE raw_path <> '' AND mailbox_id IS NOT NULL`, `CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_unregistered_raw_path ON messages(raw_path) WHERE raw_path <> '' AND mailbox_id IS NULL`, + `CREATE TABLE IF NOT EXISTS sent_message_dedupe_keys ( + mailbox_id TEXT NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE, + folder_id TEXT NOT NULL REFERENCES folders(id) ON DELETE CASCADE, + message_id TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY(mailbox_id, folder_id, message_id) + )`, + `CREATE TABLE IF NOT EXISTS send_as_grants ( + id TEXT PRIMARY KEY, + mailbox_id TEXT NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE, + address TEXT NOT NULL, + display_name TEXT NOT NULL DEFAULT '', + enabled INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(mailbox_id, address) + )`, + `CREATE TABLE IF NOT EXISTS send_queue ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + mailbox_id TEXT NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE, + sent_message_id TEXT NOT NULL DEFAULT '', + message_id TEXT NOT NULL DEFAULT '', + source TEXT NOT NULL, + mail_from TEXT NOT NULL, + header_from TEXT NOT NULL, + recipients_json TEXT NOT NULL, + mime_base64 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + attempt_count INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 5, + next_attempt_at TEXT NOT NULL, + last_error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_send_queue_due ON send_queue(status, next_attempt_at, created_at)`, + `CREATE TABLE IF NOT EXISTS send_audit_events ( + id TEXT PRIMARY KEY, + queue_id TEXT NOT NULL DEFAULT '', + user_id TEXT NOT NULL DEFAULT '', + mailbox_id TEXT NOT NULL DEFAULT '', + sent_message_id TEXT NOT NULL DEFAULT '', + source TEXT NOT NULL, + event TEXT NOT NULL, + status TEXT NOT NULL, + mail_from TEXT NOT NULL DEFAULT '', + header_from TEXT NOT NULL DEFAULT '', + recipients_json TEXT NOT NULL DEFAULT '[]', + error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_send_audit_events_created ON send_audit_events(created_at)`, `CREATE TABLE IF NOT EXISTS attachments ( id TEXT PRIMARY KEY, message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE, @@ -370,12 +425,71 @@ func (a *App) migrate(ctx context.Context) error { if err := a.migratePermissionGroupLimits(ctx); err != nil { return err } + if err := a.migrateSendQueueMessageID(ctx); err != nil { + return err + } if err := a.ensureDefaultPermissionGroups(ctx); err != nil { return err } return nil } +func (a *App) migrateSendQueueMessageID(ctx context.Context) error { + rows, err := a.db.QueryContext(ctx, `PRAGMA table_info(send_queue)`) + if err != nil { + return err + } + hasMessageID := false + for rows.Next() { + var cid int + var name, typ string + var notnull int + var dflt any + var pk int + if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil { + rows.Close() + return err + } + if name == "message_id" { + hasMessageID = true + } + } + if err := rows.Close(); err != nil { + return err + } + if !hasMessageID { + if _, err := a.db.ExecContext(ctx, `ALTER TABLE send_queue ADD COLUMN message_id TEXT NOT NULL DEFAULT ''`); err != nil { + return err + } + } + if _, err := a.db.ExecContext(ctx, `DELETE FROM send_queue + WHERE id IN ( + SELECT id FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY mailbox_id, source, message_id + ORDER BY + CASE status + WHEN 'queued' THEN 0 + WHEN 'sending' THEN 1 + WHEN 'failed' THEN 2 + WHEN 'delivered' THEN 3 + ELSE 4 + END, + created_at DESC, + id DESC + ) AS row_num + FROM send_queue + WHERE message_id <> '' + ) + WHERE row_num > 1 + )`); err != nil { + return err + } + _, err = a.db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_send_queue_mailbox_source_message_id ON send_queue(mailbox_id, source, message_id) WHERE message_id <> ''`) + return err +} + func (a *App) migratePermissionGroupLimits(ctx context.Context) error { rows, err := a.db.QueryContext(ctx, `PRAGMA table_info(permission_groups)`) if err != nil { diff --git a/apps/api/internal/app/app_test.go b/apps/api/internal/app/app_test.go index 072d502..df9eba1 100644 --- a/apps/api/internal/app/app_test.go +++ b/apps/api/internal/app/app_test.go @@ -4,18 +4,31 @@ import ( "bufio" "bytes" "context" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" "encoding/base64" "encoding/json" + "encoding/pem" + "errors" "io" "log/slog" + "math/big" "net" "net/http" "net/http/httptest" + "net/textproto" "os" "path/filepath" "strings" "testing" "time" + + "github.com/emersion/go-sasl" + smtpclient "github.com/emersion/go-smtp" + "golang.org/x/crypto/bcrypt" ) func newTestApp(t *testing.T) *App { @@ -41,6 +54,63 @@ func newTestApp(t *testing.T) *App { return a } +func defaultAdminUserAndMailbox(t *testing.T, a *App) (*User, *Mailbox) { + t.Helper() + ctx := context.Background() + user, _, err := a.userByEmail(ctx, "admin@lanqin.local") + if err != nil { + t.Fatal(err) + } + mb, err := a.mailboxByAddress(ctx, "admin@lanqin.local") + if err != nil { + t.Fatal(err) + } + return user, mb +} + +func writeTestCertificateFiles(t *testing.T, hostname string) (string, string) { + t.Helper() + if strings.TrimSpace(hostname) == "" { + hostname = "localhost" + } + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + tmpl := x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{ + CommonName: hostname, + }, + NotBefore: now.Add(-time.Hour), + NotAfter: now.Add(24 * time.Hour), + KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + DNSNames: []string{hostname, "localhost"}, + } + der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + certPath := filepath.Join(t.TempDir(), "cert.pem") + keyPath := filepath.Join(filepath.Dir(certPath), "key.pem") + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)}) + if err := os.WriteFile(certPath, certPEM, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + t.Fatal(err) + } + return certPath, keyPath +} + func startFakeSMTP(t *testing.T) (string, string, <-chan string) { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") @@ -65,6 +135,30 @@ func startFakeSMTP(t *testing.T) (string, string, <-chan string) { return host, port, received } +func startCapturingSMTP(t *testing.T, capacity int) (string, string, <-chan string) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + received := make(chan string, capacity) + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go handleFakeSMTPConn(conn, received) + } + }() + host, port, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + return host, port, received +} + func handleFakeSMTPConn(conn net.Conn, received chan<- string) { defer conn.Close() reader := bufio.NewReader(conn) @@ -746,7 +840,7 @@ func TestCatchAllStoresUnregisteredMailForAdminOnly(t *testing.T) { } } -func TestMailSendReturnsSMTPFailure(t *testing.T) { +func TestMailSendQueuesSMTPFailureForRetry(t *testing.T) { a := newTestApp(t) a.cfg.SMTPHost = "127.0.0.1" a.cfg.SMTPPort = "1" @@ -763,12 +857,748 @@ func TestMailSendReturnsSMTPFailure(t *testing.T) { "subject": "smtp failure should surface", "text": "hello", } - var errBody map[string]any - if code := admin.do("POST", "/api/mail/send", payload, &errBody); code != http.StatusBadGateway { - t.Fatalf("smtp failure code=%d body=%v", code, errBody) + var sent MailMessage + if code := admin.do("POST", "/api/mail/send", payload, &sent); code != http.StatusCreated { + t.Fatalf("smtp queued send code=%d body=%+v", code, sent) } - if got, _ := errBody["error"].(string); !strings.Contains(got, "smtp delivery failed") { - t.Fatalf("error=%q", got) + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + var status, lastError string + if err := a.db.QueryRow(`SELECT status,last_error FROM send_queue WHERE sent_message_id=?`, sent.ID).Scan(&status, &lastError); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusFailed || lastError == "" { + t.Fatalf("queue status=%q lastError=%q", status, lastError) + } + var auditCount int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM send_audit_events WHERE sent_message_id=? AND event=?`, sent.ID, sendAuditRetry).Scan(&auditCount); err != nil { + t.Fatal(err) + } + if auditCount != 1 { + t.Fatalf("retry audit count=%d, want 1", auditCount) + } +} + +func TestMailSendRejectsUnauthorizedFrom(t *testing.T) { + a := newTestApp(t) + ts := httptest.NewServer(a.Router()) + defer ts.Close() + admin := &testClient{t: t, server: ts} + + var login map[string]any + if code := admin.do("POST", "/api/auth/login", map[string]string{"email": "admin@lanqin.local", "password": "ChangeMe123!"}, &login); code != http.StatusOK { + t.Fatalf("login code=%d body=%v", code, login) + } + var errBody map[string]any + if code := admin.do("POST", "/api/mail/send", map[string]any{ + "from": "attacker@example.com", + "to": []string{"person@example.com"}, + "subject": "bad from", + "text": "hello", + }, &errBody); code != http.StatusForbidden { + t.Fatalf("unauthorized from code=%d body=%v", code, errBody) + } +} + +func TestMailSendRollsBackSentCopyWhenQueueInsertFails(t *testing.T) { + a := newTestApp(t) + a.cfg.SMTPHost = "postfix" + a.cfg.SMTPPort = "25" + user, mb := defaultAdminUserAndMailbox(t, a) + if _, err := a.db.ExecContext(context.Background(), `DROP TABLE send_queue`); err != nil { + t.Fatal(err) + } + _, err := a.sendMailNow(context.Background(), user, mb, mailComposeInput{ + To: []string{"person@example.com"}, + Subject: "queue insert failure", + Text: "hello", + }) + if err == nil || !strings.Contains(err.Error(), "failed to enqueue delivery") { + t.Fatalf("sendMailNow error=%v, want enqueue failure", err) + } + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM messages WHERE mailbox_id=? AND subject=?`, mb.ID, "queue insert failure").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("sent copy should be removed after enqueue failure, count=%d", count) + } +} + +func TestSendQueueRecoversStaleSendingItems(t *testing.T) { + a := newTestApp(t) + host, port, received := startCapturingSMTP(t, 1) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + user, mb := defaultAdminUserAndMailbox(t, a) + now := a.now().UTC() + mimeBytes := []byte("From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: stale\r\n\r\nbody") + queueID, err := a.enqueueSend(context.Background(), sendQueueInput{ + UserID: user.ID, + MailboxID: mb.ID, + Source: sendSourceWebmail, + MailFrom: mb.Address, + HeaderFrom: mb.Address, + Recipients: []string{"person@example.com"}, + MIMEBytes: mimeBytes, + Now: now, + }) + if err != nil { + t.Fatal(err) + } + staleAt := now.Add(-sendQueueStaleAfter - time.Minute).Format(time.RFC3339Nano) + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,attempt_count=1,updated_at=? WHERE id=?`, sendQueueStatusSending, staleAt, queueID); err != nil { + t.Fatal(err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("recovered queue item was not relayed") + } + var status string + if err := a.db.QueryRow(`SELECT status FROM send_queue WHERE id=?`, queueID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusDelivered { + t.Fatalf("queue status=%q, want delivered", status) + } + var mimeBase64 string + if err := a.db.QueryRow(`SELECT mime_base64 FROM send_queue WHERE id=?`, queueID).Scan(&mimeBase64); err != nil { + t.Fatal(err) + } + if mimeBase64 != "" { + t.Fatal("delivered queue item should not retain raw MIME") + } +} + +func TestSendQueueStaleDeliveredMarkerDoesNotRedeliver(t *testing.T) { + a := newTestApp(t) + host, port, received := startCapturingSMTP(t, 1) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + user, mb := defaultAdminUserAndMailbox(t, a) + now := a.now().UTC() + mimeBytes := []byte("From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: marker\r\n\r\nbody") + queueID, err := a.enqueueSend(context.Background(), sendQueueInput{ + UserID: user.ID, + MailboxID: mb.ID, + Source: sendSourceWebmail, + MailFrom: mb.Address, + HeaderFrom: mb.Address, + Recipients: []string{"person@example.com"}, + MIMEBytes: mimeBytes, + Now: now, + }) + if err != nil { + t.Fatal(err) + } + staleAt := now.Add(-sendQueueStaleAfter - time.Minute).Format(time.RFC3339Nano) + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,attempt_count=1,updated_at=? WHERE id=?`, sendQueueStatusSending, staleAt, queueID); err != nil { + t.Fatal(err) + } + if err := a.writeSendQueueDeliveredMarker(queueID); err != nil { + t.Fatal(err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case body := <-received: + t.Fatalf("stale delivered marker should not redeliver, got %q", body) + case <-time.After(200 * time.Millisecond): + } + var status, mimeBase64 string + if err := a.db.QueryRow(`SELECT status,mime_base64 FROM send_queue WHERE id=?`, queueID).Scan(&status, &mimeBase64); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusDelivered { + t.Fatalf("queue status=%q, want delivered", status) + } + if mimeBase64 != "" { + t.Fatal("delivered marker recovery should clear raw MIME") + } + delivered, err := a.hasSendQueueDeliveredMarker(queueID) + if err != nil { + t.Fatal(err) + } + if delivered { + t.Fatal("delivered marker should be removed after database state is repaired") + } +} + +func TestSubmissionAuthRequiresMailboxPasswordAndSendPermission(t *testing.T) { + a := newTestApp(t) + user, mailbox, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatalf("authenticate submission: %v", err) + } + if user.Email != "admin@lanqin.local" || mailbox.Address != "admin@lanqin.local" { + t.Fatalf("unexpected auth user=%+v mailbox=%+v", user, mailbox) + } + if _, _, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "wrong-password"); err == nil { + t.Fatal("wrong password should fail") + } + + ctx := context.Background() + hash, err := bcrypt.GenerateFromPassword([]byte("Password123!"), bcrypt.DefaultCost) + if err != nil { + t.Fatal(err) + } + userID := newID("usr") + domainID := mustDefaultDomainID(t, a) + now := a.now().UTC().Format(time.RFC3339Nano) + if _, err := a.db.ExecContext(ctx, `INSERT INTO users(id,email,display_name,role,password_hash,disabled,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)`, userID, "nosend@lanqin.local", "No Send", "user", string(hash), 0, now, now); err != nil { + t.Fatal(err) + } + if _, err := a.db.ExecContext(ctx, `INSERT INTO mailboxes(id,user_id,domain_id,local_part,address,display_name,password_hash,quota_mb,status,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)`, newID("mb"), userID, domainID, "nosend", "nosend@lanqin.local", "No Send", string(hash), 1024, "active", now, now); err != nil { + t.Fatal(err) + } + if _, err := a.db.ExecContext(ctx, `UPDATE permission_groups SET permissions_json=?, updated_at=? WHERE id=?`, encodePermissions(withoutPermissions(regularUserDefaultPermissions(), PermissionMailSend)), now, PermissionGroupRegular); err != nil { + t.Fatal(err) + } + if _, _, err := a.authenticateSubmission(ctx, "nosend@lanqin.local", "Password123!"); err == nil { + t.Fatal("missing send permission should fail") + } + if _, err := a.db.ExecContext(ctx, `UPDATE users SET disabled=1 WHERE id=?`, userID); err != nil { + t.Fatal(err) + } + if _, _, err := a.authenticateSubmission(ctx, "nosend@lanqin.local", "Password123!"); err == nil { + t.Fatal("disabled owner should fail") + } +} + +func TestSubmissionSendsRelayAndStoresSentCopy(t *testing.T) { + a := newTestApp(t) + host, port, received := startCapturingSMTP(t, 2) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + raw := strings.Join([]string{ + "From: Admin ", + "To: person@example.com", + "Bcc: hidden@example.com", + "Subject: Submission sent", + "Message-ID: ", + "Date: Tue, 24 Jun 2025 10:00:00 +0000", + "MIME-Version: 1.0", + "Content-Type: text/plain; charset=utf-8", + "", + "hello from submission", + }, "\r\n") + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com", "hidden@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("submit smtp message: %v", err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case body := <-received: + if strings.Contains(strings.ToLower(body), "\r\nbcc:") || strings.Contains(body, "hidden@example.com") { + t.Fatalf("relay body leaked bcc: %s", body) + } + case <-time.After(2 * time.Second): + t.Fatal("relay message not received") + } + sentFolderID, err := a.ensureFolder(context.Background(), mb.ID, "Sent") + if err != nil { + t.Fatal(err) + } + var subject, bccJSON string + var read int + if err := a.db.QueryRow(`SELECT subject,bcc_addrs,is_read FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, "").Scan(&subject, &bccJSON, &read); err != nil { + t.Fatal(err) + } + if subject != "Submission sent" || read != 1 { + t.Fatalf("unexpected sent message subject=%q read=%d", subject, read) + } + if got := jsonDecodeSlice(bccJSON); len(got) != 1 || got[0] != "hidden@example.com" { + t.Fatalf("bcc json=%s", bccJSON) + } + var deliveredAudits int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM send_audit_events WHERE event=? AND status=?`, sendAuditDelivered, sendQueueStatusDelivered).Scan(&deliveredAudits); err != nil { + t.Fatal(err) + } + if deliveredAudits != 1 { + t.Fatalf("delivered audit count=%d, want 1", deliveredAudits) + } +} + +func TestSubmissionRejectsMismatchedSender(t *testing.T) { + a := newTestApp(t) + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: attacker@example.com\r\nTo: person@example.com\r\nSubject: nope\r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err == nil { + t.Fatal("mismatched header From should fail") + } + raw = "From: admin@lanqin.local, attacker@example.com\r\nTo: person@example.com\r\nSubject: nope\r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err == nil { + t.Fatal("multiple header From addresses should fail") + } + raw = "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: nope\r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, "attacker@example.com", []string{"person@example.com"}, strings.NewReader(raw)); err == nil { + t.Fatal("mismatched MAIL FROM should fail") + } +} + +func TestSerializeMessageUsesStableHeaderOrder(t *testing.T) { + header := textproto.MIMEHeader{ + "Subject": {"stable"}, + "From": {"admin@lanqin.local"}, + "Message": {"custom"}, + "X-Zebra": {"z"}, + "X-Answer": {"a"}, + } + first := string(serializeMessage(header, []byte("body"))) + for i := 0; i < 20; i++ { + if got := string(serializeMessage(header, []byte("body"))); got != first { + t.Fatalf("serializeMessage is not stable:\nfirst=%q\ngot=%q", first, got) + } + } + if !strings.HasPrefix(first, "From: admin@lanqin.local\r\n") { + t.Fatalf("unexpected header order: %q", first) + } +} + +func TestSubmissionRelayFailureKeepsSentCopyAndRetries(t *testing.T) { + a := newTestApp(t) + a.cfg.SMTPHost = "127.0.0.1" + a.cfg.SMTPPort = "1" + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: relay fail\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("submission should queue relay failure for retry: %v", err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM messages WHERE mailbox_id=? AND message_id=?`, mb.ID, "").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("sent copy should remain after queued relay failure, count=%d", count) + } + var status, lastError string + if err := a.db.QueryRow(`SELECT status,last_error FROM send_queue WHERE mailbox_id=? AND sent_message_id <> ''`, mb.ID).Scan(&status, &lastError); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusFailed || lastError == "" { + t.Fatalf("queue status=%q lastError=%q", status, lastError) + } +} + +func TestSubmissionSentCopyDedupesByMessageID(t *testing.T) { + a := newTestApp(t) + host, port, _ := startCapturingSMTP(t, 4) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: dedupe\r\nMessage-ID: \r\n\r\nbody" + for i := 0; i < 2; i++ { + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("submit %d: %v", i, err) + } + } + sentFolderID, err := a.ensureFolder(context.Background(), mb.ID, "Sent") + if err != nil { + t.Fatal(err) + } + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, "").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("sent copy count=%d, want 1", count) + } + var queueCount int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM send_queue WHERE mailbox_id=? AND message_id=?`, mb.ID, "").Scan(&queueCount); err != nil { + t.Fatal(err) + } + if queueCount != 1 { + t.Fatalf("send queue count=%d, want 1", queueCount) + } +} + +func TestInsertSentMessageOnceFailsWhenDedupeKeyHasNoMessage(t *testing.T) { + a := newTestApp(t) + _, mb := defaultAdminUserAndMailbox(t, a) + ctx := context.Background() + sentFolderID, err := a.ensureFolder(ctx, mb.ID, "Sent") + if err != nil { + t.Fatal(err) + } + messageID := "" + if err := a.insertSentDedupeKey(ctx, mb.ID, sentFolderID, messageID); err != nil { + t.Fatal(err) + } + now := a.now().UTC() + sentID, inserted, err := a.insertSentMessageOnce(ctx, storedMessage{ + MailboxID: mb.ID, + MessageUID: newID("uid"), + MessageID: messageID, + Subject: "orphan dedupe", + From: mb.Address, + To: []string{"person@example.com"}, + SentAt: now, + ReceivedAt: now, + BodyText: "body", + IsRead: true, + }, nil) + if !errors.Is(err, errSentDedupeExists) { + t.Fatalf("insertSentMessageOnce error=%v, want errSentDedupeExists", err) + } + if sentID != "" || inserted { + t.Fatalf("sentID=%q inserted=%v, want empty false", sentID, inserted) + } + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, messageID).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("orphan dedupe should not create sent message, count=%d", count) + } +} + +func TestSubmissionRequeuesTerminalFailedDuplicateMessageID(t *testing.T) { + a := newTestApp(t) + a.cfg.SMTPHost = "127.0.0.1" + a.cfg.SMTPPort = "1" + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: requeue\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,attempt_count=max_attempts,next_attempt_at=?,last_error='terminal' WHERE mailbox_id=? AND message_id=?`, sendQueueStatusFailed, a.now().UTC().Add(time.Hour).Format(time.RFC3339Nano), mb.ID, ""); err != nil { + t.Fatal(err) + } + + host, port, received := startCapturingSMTP(t, 1) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("requeued terminal failure was not relayed") + } + var status string + var attemptCount int + if err := a.db.QueryRow(`SELECT status,attempt_count FROM send_queue WHERE mailbox_id=? AND message_id=?`, mb.ID, "").Scan(&status, &attemptCount); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusDelivered || attemptCount != 1 { + t.Fatalf("queue status=%q attempts=%d, want delivered attempts=1", status, attemptCount) + } +} + +func TestSubmissionRequeuesDeliveredDuplicateMessageID(t *testing.T) { + a := newTestApp(t) + host, port, received := startCapturingSMTP(t, 2) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: resend\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("first delivery not received") + } + if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("requeued delivered message was not relayed") + } + var status string + var attemptCount int + if err := a.db.QueryRow(`SELECT status,attempt_count FROM send_queue WHERE mailbox_id=? AND message_id=?`, mb.ID, "").Scan(&status, &attemptCount); err != nil { + t.Fatal(err) + } + if status != sendQueueStatusDelivered || attemptCount != 1 { + t.Fatalf("queue status=%q attempts=%d, want delivered attempts=1", status, attemptCount) + } +} + +func TestSubmissionAllowsAuthorizedAliasSendAs(t *testing.T) { + a := newTestApp(t) + ctx := context.Background() + if _, err := a.db.ExecContext(ctx, `INSERT INTO aliases(id,domain_id,source,destination,enabled,created_at,updated_at) VALUES(?,?,?,?,?,?,?)`, newID("als"), mustDefaultDomainID(t, a), "team@lanqin.local", "admin@lanqin.local", 1, a.now().UTC().Format(time.RFC3339Nano), a.now().UTC().Format(time.RFC3339Nano)); err != nil { + t.Fatal(err) + } + user, mb, err := a.authenticateSubmission(ctx, "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: Team \r\nTo: person@example.com\r\nSubject: alias send-as\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(ctx, user, mb, "team@lanqin.local", []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("authorized alias send-as should submit: %v", err) + } + sentFolderID, err := a.ensureFolder(ctx, mb.ID, "Sent") + if err != nil { + t.Fatal(err) + } + var fromAddr string + if err := a.db.QueryRow(`SELECT from_addr FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, "").Scan(&fromAddr); err != nil { + t.Fatal(err) + } + if fromAddr != "team@lanqin.local" { + t.Fatalf("from_addr=%q, want alias", fromAddr) + } +} + +func TestSubmissionAllowsMultiDestinationAliasSendAs(t *testing.T) { + a := newTestApp(t) + ctx := context.Background() + if _, err := a.db.ExecContext(ctx, `INSERT INTO aliases(id,domain_id,source,destination,enabled,created_at,updated_at) VALUES(?,?,?,?,?,?,?)`, newID("als"), mustDefaultDomainID(t, a), "team-many@lanqin.local", "other@lanqin.local, admin@lanqin.local", 1, a.now().UTC().Format(time.RFC3339Nano), a.now().UTC().Format(time.RFC3339Nano)); err != nil { + t.Fatal(err) + } + user, mb, err := a.authenticateSubmission(ctx, "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + raw := "From: Team \r\nTo: person@example.com\r\nSubject: alias send-as\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(ctx, user, mb, "team-many@lanqin.local", []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("authorized multi-destination alias send-as should submit: %v", err) + } +} + +func TestSubmissionAllowsExplicitSendAsGrant(t *testing.T) { + a := newTestApp(t) + ctx := context.Background() + user, mb, err := a.authenticateSubmission(ctx, "admin@lanqin.local", "ChangeMe123!") + if err != nil { + t.Fatal(err) + } + now := a.now().UTC().Format(time.RFC3339Nano) + if _, err := a.db.ExecContext(ctx, `INSERT INTO send_as_grants(id,mailbox_id,address,display_name,enabled,created_at,updated_at) VALUES(?,?,?,?,?,?,?)`, newID("sag"), mb.ID, "support@example.com", "Support", 1, now, now); err != nil { + t.Fatal(err) + } + raw := "From: Support \r\nTo: person@example.com\r\nSubject: explicit send-as\r\nMessage-ID: \r\n\r\nbody" + if err := a.submitSMTPMessage(ctx, user, mb, "support@example.com", []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatalf("explicit send-as grant should submit: %v", err) + } + sentFolderID, err := a.ensureFolder(ctx, mb.ID, "Sent") + if err != nil { + t.Fatal(err) + } + var fromAddr, fromName string + if err := a.db.QueryRow(`SELECT from_addr,from_name FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, "").Scan(&fromAddr, &fromName); err != nil { + t.Fatal(err) + } + if fromAddr != "support@example.com" || fromName != "Support" { + t.Fatalf("from=%q name=%q, want explicit grant", fromAddr, fromName) + } +} + +func TestSentMessageDedupeTableExists(t *testing.T) { + a := newTestApp(t) + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM sqlite_master WHERE type='table' AND name='sent_message_dedupe_keys'`).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("sent message dedupe table count=%d, want 1", count) + } +} + +func TestSendQueueMessageIDMigrationDropsDuplicatesBeforeUniqueIndex(t *testing.T) { + a := newTestApp(t) + user, mb := defaultAdminUserAndMailbox(t, a) + if _, err := a.db.Exec(`DROP INDEX IF EXISTS idx_send_queue_mailbox_source_message_id`); err != nil { + t.Fatal(err) + } + if _, err := a.db.Exec(`INSERT INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + "dup_old", user.ID, mb.ID, "sent1", "", sendSourceSubmission, "admin@lanqin.local", "admin@lanqin.local", "[]", "bWVzc2FnZQ==", sendQueueStatusDelivered, a.now().UTC().Format(time.RFC3339Nano), "2026-06-24T00:00:00Z", "2026-06-24T00:00:00Z"); err != nil { + t.Fatal(err) + } + if _, err := a.db.Exec(`INSERT INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + "dup_keep", user.ID, mb.ID, "sent2", "", sendSourceSubmission, "admin@lanqin.local", "admin@lanqin.local", "[]", "bWVzc2FnZQ==", sendQueueStatusQueued, a.now().UTC().Format(time.RFC3339Nano), "2026-06-24T00:01:00Z", "2026-06-24T00:01:00Z"); err != nil { + t.Fatal(err) + } + if err := a.migrateSendQueueMessageID(context.Background()); err != nil { + t.Fatal(err) + } + var count int + if err := a.db.QueryRow(`SELECT COUNT(1) FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=''`, mb.ID, sendSourceSubmission).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("duplicate queue rows count=%d, want 1", count) + } + var keptID string + if err := a.db.QueryRow(`SELECT id FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=''`, mb.ID, sendSourceSubmission).Scan(&keptID); err != nil { + t.Fatal(err) + } + if keptID != "dup_keep" { + t.Fatalf("kept queue id=%q, want dup_keep", keptID) + } +} + +func TestSubmissionTLSConfigRequiresCertificateFiles(t *testing.T) { + a := newTestApp(t) + a.cfg.SubmissionAddr = ":587" + a.cfg.SubmissionTLSAddr = ":465" + if _, err := LoadServerTLSConfig(a.cfg); err == nil { + t.Fatal("submission TLS config should require certificate files") + } +} + +func TestSubmissionTLSConfigReloadsCertificateFiles(t *testing.T) { + a := newTestApp(t) + certPath, keyPath := writeTestCertificateFiles(t, "first.example.test") + a.cfg.TLSCertFile = certPath + a.cfg.TLSKeyFile = keyPath + tlsConfig, err := LoadServerTLSConfig(a.cfg) + if err != nil { + t.Fatal(err) + } + first, err := tlsConfig.GetCertificate(&tls.ClientHelloInfo{}) + if err != nil { + t.Fatal(err) + } + firstLeaf, err := x509.ParseCertificate(first.Certificate[0]) + if err != nil { + t.Fatal(err) + } + nextCertPath, nextKeyPath := writeTestCertificateFiles(t, "second.example.test") + nextCert, err := os.ReadFile(nextCertPath) + if err != nil { + t.Fatal(err) + } + nextKey, err := os.ReadFile(nextKeyPath) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(certPath, nextCert, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyPath, nextKey, 0o600); err != nil { + t.Fatal(err) + } + second, err := tlsConfig.GetCertificate(&tls.ClientHelloInfo{}) + if err != nil { + t.Fatal(err) + } + secondLeaf, err := x509.ParseCertificate(second.Certificate[0]) + if err != nil { + t.Fatal(err) + } + if firstLeaf.Subject.CommonName != "first.example.test" || secondLeaf.Subject.CommonName != "second.example.test" { + t.Fatalf("cert reload common names first=%q second=%q", firstLeaf.Subject.CommonName, secondLeaf.Subject.CommonName) + } +} + +func TestSubmissionServersAcceptStartTLSAndImplicitTLS(t *testing.T) { + a := newTestApp(t) + host, port, received := startCapturingSMTP(t, 2) + a.cfg.SMTPHost = host + a.cfg.SMTPPort = port + certPath, keyPath := writeTestCertificateFiles(t, "mail.example.test") + a.cfg.TLSCertFile = certPath + a.cfg.TLSKeyFile = keyPath + tlsConfig, err := LoadServerTLSConfig(a.cfg) + if err != nil { + t.Fatal(err) + } + + startServer := func(t *testing.T, implicit bool) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv := a.newSubmissionServer(ln.Addr().String(), tlsConfig) + go func() { + if implicit { + _ = srv.Serve(tls.NewListener(ln, tlsConfig)) + } else { + _ = srv.Serve(ln) + } + }() + t.Cleanup(func() { _ = srv.Shutdown(context.Background()) }) + return ln.Addr().String() + } + + raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: starttls\r\nMessage-ID: \r\n\r\nbody" + addr := startServer(t, false) + client, err := smtpclient.DialStartTLS(addr, &tls.Config{InsecureSkipVerify: true}) + if err != nil { + t.Fatal(err) + } + if err := client.Auth(sasl.NewPlainClient("", "admin@lanqin.local", "ChangeMe123!")); err != nil { + t.Fatal(err) + } + if err := client.SendMail("admin@lanqin.local", []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + _ = client.Close() + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("starttls relay not received") + } + + raw = "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: smtps\r\nMessage-ID: \r\n\r\nbody" + addr = startServer(t, true) + client, err = smtpclient.DialTLS(addr, &tls.Config{InsecureSkipVerify: true}) + if err != nil { + t.Fatal(err) + } + if err := client.Auth(sasl.NewPlainClient("", "admin@lanqin.local", "ChangeMe123!")); err != nil { + t.Fatal(err) + } + if err := client.SendMail("admin@lanqin.local", []string{"person@example.com"}, strings.NewReader(raw)); err != nil { + t.Fatal(err) + } + _ = client.Close() + if err := a.processDueSendQueue(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-received: + case <-time.After(2 * time.Second): + t.Fatal("implicit tls relay not received") } } diff --git a/apps/api/internal/app/config.go b/apps/api/internal/app/config.go index df2b6de..47f84d1 100644 --- a/apps/api/internal/app/config.go +++ b/apps/api/internal/app/config.go @@ -22,6 +22,11 @@ type Config struct { SMTPUsername string SMTPPassword string SMTPRequireTLS bool + SubmissionAddr string + SubmissionTLSAddr string + SubmissionMaxMessageMB int + TLSCertFile string + TLSKeyFile string MaildirRoot string MaildirScanSeconds int AllowInsecureHTTP bool @@ -55,6 +60,11 @@ func LoadConfig() Config { SMTPUsername: getenv("LANQIN_SMTP_USERNAME", ""), SMTPPassword: getenv("LANQIN_SMTP_PASSWORD", ""), SMTPRequireTLS: getenvBool("LANQIN_SMTP_REQUIRE_TLS", false), + SubmissionAddr: getenv("LANQIN_SUBMISSION_ADDR", ""), + SubmissionTLSAddr: getenv("LANQIN_SUBMISSION_TLS_ADDR", ""), + SubmissionMaxMessageMB: getenvInt("LANQIN_SUBMISSION_MAX_MESSAGE_MB", 35), + TLSCertFile: getenv("LANQIN_TLS_CERT_FILE", ""), + TLSKeyFile: getenv("LANQIN_TLS_KEY_FILE", ""), MaildirRoot: getenv("LANQIN_MAILDIR_ROOT", ""), MaildirScanSeconds: getenvInt("LANQIN_MAILDIR_SCAN_SECONDS", 30), AllowInsecureHTTP: getenvBool("LANQIN_ALLOW_INSECURE_HTTP", true), diff --git a/apps/api/internal/app/mail_handlers.go b/apps/api/internal/app/mail_handlers.go index 75bd7a9..55ddcf4 100644 --- a/apps/api/internal/app/mail_handlers.go +++ b/apps/api/internal/app/mail_handlers.go @@ -336,6 +336,8 @@ func (a *App) handleMailMessage(w http.ResponseWriter, r *http.Request) { type mailComposeInput struct { MailboxID string `json:"mailboxId"` + From string `json:"from"` + FromName string `json:"fromName"` To []string `json:"to"` CC []string `json:"cc"` BCC []string `json:"bcc"` @@ -358,6 +360,8 @@ type mailDraftInput struct { type scheduledSendPayload struct { MailboxID string `json:"mailboxId"` + From string `json:"from"` + FromName string `json:"fromName"` To []string `json:"to"` CC []string `json:"cc"` BCC []string `json:"bcc"` @@ -412,8 +416,8 @@ func (a *App) handleMailSend(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusTooManyRequests, err.Error()) return } - if strings.HasPrefix(err.Error(), "smtp delivery failed:") { - respondError(w, http.StatusBadGateway, err.Error()) + if errors.Is(err, errSenderNotAuthorized) { + respondError(w, http.StatusForbidden, err.Error()) return } respondError(w, http.StatusInternalServerError, err.Error()) @@ -426,6 +430,7 @@ var errNoRecipients = errors.New("at least one recipient is required") var errInvalidMIME = errors.New("invalid mime message") var errAttachmentTooLarge = errors.New("attachment size exceeds permission limit") var errSMTPRateLimited = errors.New("smtp send rate limit exceeded") +var errSenderNotAuthorized = errors.New("sender address is not authorized") func (a *App) sendMailNow(ctx context.Context, user *User, mb *Mailbox, req mailComposeInput) (*MailMessage, error) { if err := validateAttachmentLimit(req.Attachments, userLimits(user)); err != nil { @@ -448,9 +453,16 @@ func (a *App) sendMailNow(ctx context.Context, user *User, mb *Mailbox, req mail } now := a.now().UTC() + fromAddress, fromName, err := a.authorizedSender(ctx, mb, req.From) + if err != nil { + return nil, err + } + if strings.TrimSpace(req.FromName) != "" && normalizeEmail(req.From) == normalizeEmail(mb.Address) { + fromName = strings.TrimSpace(req.FromName) + } messageID := fmt.Sprintf("<%s@%s>", newID("msg"), strings.Split(mb.Address, "@")[1]) mimeBytes, err := BuildMIME(MIMEMessage{ - From: mb.Address, FromName: mb.DisplayName, To: req.To, CC: req.CC, BCC: req.BCC, Subject: req.Subject, Text: req.Text, HTML: req.HTML, MessageID: messageID, Date: now, Attachments: req.Attachments, + From: fromAddress, FromName: fromName, To: req.To, CC: req.CC, BCC: req.BCC, Subject: req.Subject, Text: req.Text, HTML: req.HTML, MessageID: messageID, Date: now, Attachments: req.Attachments, }) if err != nil { return nil, fmt.Errorf("%w: %v", errInvalidMIME, err) @@ -458,21 +470,21 @@ func (a *App) sendMailNow(ctx context.Context, user *User, mb *Mailbox, req mail if err := a.recordSMTPRate(ctx, user, mb); err != nil { return nil, err } - if a.cfg.SMTPHost != "" { - if err := a.sendSMTP(mb.Address, allRecipients, mimeBytes); err != nil { - return nil, fmt.Errorf("smtp delivery failed: %w", err) - } - } sentFolderID, err := a.ensureFolder(ctx, mb.ID, "Sent") if err != nil { return nil, fmt.Errorf("failed to load sent folder: %w", err) } - base := storedMessage{MailboxID: mb.ID, FolderID: sentFolderID, MessageUID: newID("uid"), MessageID: messageID, Subject: req.Subject, From: mb.Address, FromName: mb.DisplayName, To: req.To, CC: req.CC, BCC: req.BCC, SentAt: now, ReceivedAt: now, Snippet: snippetFrom(req.Text, req.HTML), BodyText: req.Text, BodyHTML: req.HTML, IsRead: true} + base := storedMessage{MailboxID: mb.ID, FolderID: sentFolderID, MessageUID: newID("uid"), MessageID: messageID, Subject: req.Subject, From: fromAddress, FromName: fromName, To: req.To, CC: req.CC, BCC: req.BCC, SentAt: now, ReceivedAt: now, Snippet: snippetFrom(req.Text, req.HTML), BodyText: req.Text, BodyHTML: req.HTML, IsRead: true} sentID, err := a.insertMessage(ctx, base, req.Attachments) if err != nil { return nil, fmt.Errorf("failed to store sent message: %w", err) } + a.recordSendAudit(ctx, sendAuditAccepted, sendQueueStatusQueued, sendAuditInput{UserID: user.ID, MailboxID: mb.ID, SentMessageID: sentID, Source: sendSourceWebmail, MailFrom: fromAddress, HeaderFrom: fromAddress, Recipients: allRecipients}) + if _, err := a.enqueueSend(ctx, sendQueueInput{UserID: user.ID, MailboxID: mb.ID, SentMessageID: sentID, MessageID: messageID, Source: sendSourceWebmail, MailFrom: fromAddress, HeaderFrom: fromAddress, Recipients: allRecipients, MIMEBytes: mimeBytes, Now: now}); err != nil { + a.deleteMessage(ctx, sentID) + return nil, fmt.Errorf("failed to enqueue delivery: %w", err) + } // Development/local-domain delivery: known local recipients go to their Inbox. // When catch-all is enabled, unknown local recipients are stored as unregistered @@ -925,7 +937,7 @@ func (a *App) handleScheduleSend(w http.ResponseWriter, r *http.Request) { } } now := a.now().UTC().Format(time.RFC3339Nano) - payload := scheduledSendPayload{MailboxID: compose.MailboxID, To: compose.To, CC: compose.CC, BCC: compose.BCC, Subject: compose.Subject, Text: compose.Text, HTML: compose.HTML, Attachments: compose.Attachments, DraftID: draftID} + payload := scheduledSendPayload{MailboxID: compose.MailboxID, From: compose.From, FromName: compose.FromName, To: compose.To, CC: compose.CC, BCC: compose.BCC, Subject: compose.Subject, Text: compose.Text, HTML: compose.HTML, Attachments: compose.Attachments, DraftID: draftID} item := ScheduledSend{ID: newID("sched"), MailboxID: mb.ID, DraftID: draftID, Subject: payload.Subject, To: payload.To, Snippet: snippetFrom(payload.Text, payload.HTML), SendAt: sendAt.UTC(), Status: "pending", CreatedAt: parseTime(now), UpdatedAt: parseTime(now)} if _, err := a.db.ExecContext(r.Context(), `INSERT INTO scheduled_sends(id,user_id,mailbox_id,draft_id,payload_json,send_at,status,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)`, item.ID, currentUser(r).ID, mb.ID, nullableString(draftID), jsonEncode(payload), item.SendAt.Format(time.RFC3339Nano), item.Status, now, now); err != nil { respondError(w, http.StatusInternalServerError, "failed to schedule send") @@ -1326,6 +1338,18 @@ func (a *App) messageByID(ctx context.Context, id string, includeBody bool) (*Ma } func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments []AttachmentInput) (string, error) { + return a.insertMessageWithDB(ctx, a.db, msg, attachments) +} + +type dbExecutor interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +type dbQueryer interface { + QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row +} + +func (a *App) insertMessageWithDB(ctx context.Context, db dbExecutor, msg storedMessage, attachments []AttachmentInput) (string, error) { id := newID("mail") now := a.now().UTC().Format(time.RFC3339Nano) hasAttachments := len(attachments) > 0 @@ -1343,13 +1367,14 @@ func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments folderID = msg.FolderID } recipientAddr := normalizeEmail(msg.RecipientAddr) - _, err := a.db.ExecContext(ctx, `INSERT INTO messages(id,mailbox_id,folder_id,recipient_addr,message_uid,message_id,subject,from_addr,from_name,to_addrs,cc_addrs,bcc_addrs,sent_at,received_at,snippet,body_text,body_html,is_read,is_starred,has_attachments,size_bytes,raw_path,created_at,updated_at) + _, err := db.ExecContext(ctx, `INSERT INTO messages(id,mailbox_id,folder_id,recipient_addr,message_uid,message_id,subject,from_addr,from_name,to_addrs,cc_addrs,bcc_addrs,sent_at,received_at,snippet,body_text,body_html,is_read,is_starred,has_attachments,size_bytes,raw_path,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, mailboxID, folderID, recipientAddr, msg.MessageUID, msg.MessageID, msg.Subject, msg.From, msg.FromName, jsonEncode(msg.To), jsonEncode(msg.CC), jsonEncode(msg.BCC), msg.SentAt.Format(time.RFC3339Nano), msg.ReceivedAt.Format(time.RFC3339Nano), msg.Snippet, msg.BodyText, msg.BodyHTML, boolInt(msg.IsRead), boolInt(msg.IsStarred), boolInt(hasAttachments), size, msg.RawPath, now, now) if err != nil { return "", err } for _, att := range attachments { - if err := a.storeAttachment(ctx, id, att); err != nil { + if err := a.storeAttachmentWithDB(ctx, db, id, att); err != nil { + a.deleteMessageFiles(ctx, id) return "", err } } @@ -1357,6 +1382,10 @@ func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments } func (a *App) storeAttachment(ctx context.Context, messageID string, input AttachmentInput) error { + return a.storeAttachmentWithDB(ctx, a.db, messageID, input) +} + +func (a *App) storeAttachmentWithDB(ctx context.Context, db dbExecutor, messageID string, input AttachmentInput) error { filename := filepath.Base(strings.TrimSpace(input.Filename)) if filename == "." || filename == "" { filename = "attachment.bin" @@ -1378,7 +1407,7 @@ func (a *App) storeAttachment(ctx context.Context, messageID string, input Attac if err := os.WriteFile(path, data, 0o600); err != nil { return err } - _, err = a.db.ExecContext(ctx, `INSERT INTO attachments(id,message_id,filename,content_type,size_bytes,storage_path,created_at) VALUES(?,?,?,?,?,?,?)`, id, messageID, filename, contentType, len(data), path, a.now().UTC().Format(time.RFC3339Nano)) + _, err = db.ExecContext(ctx, `INSERT INTO attachments(id,message_id,filename,content_type,size_bytes,storage_path,created_at) VALUES(?,?,?,?,?,?,?)`, id, messageID, filename, contentType, len(data), path, a.now().UTC().Format(time.RFC3339Nano)) return err } @@ -1548,6 +1577,11 @@ func (a *App) deleteMessageFiles(ctx context.Context, messageID string) { _ = os.RemoveAll(filepath.Join(a.cfg.DataDir, "attachments", messageID)) } +func (a *App) deleteMessage(ctx context.Context, messageID string) { + a.deleteMessageFiles(ctx, messageID) + _, _ = a.db.ExecContext(ctx, `DELETE FROM messages WHERE id=?`, messageID) +} + type messageSummaryScanner interface{ Scan(dest ...any) error } func scanAdminMessageSummary(row messageSummaryScanner) (MailMessage, error) { diff --git a/apps/api/internal/app/send_queue.go b/apps/api/internal/app/send_queue.go new file mode 100644 index 0000000..02a03ec --- /dev/null +++ b/apps/api/internal/app/send_queue.go @@ -0,0 +1,439 @@ +package app + +import ( + "context" + "database/sql" + "encoding/base64" + "errors" + "os" + "path/filepath" + "strings" + "time" +) + +const ( + sendQueueStatusQueued = "queued" + sendQueueStatusSending = "sending" + sendQueueStatusDelivered = "delivered" + sendQueueStatusFailed = "failed" + + sendAuditAccepted = "accepted" + sendAuditQueued = "queued" + sendAuditDelivered = "delivered" + sendAuditFailed = "failed" + sendAuditRetry = "retry" + + sendSourceWebmail = "webmail" + sendSourceSubmission = "submission" + + sendQueueStaleAfter = 15 * time.Minute + sendQueueConcurrency = 4 + + sendQueueDeliveredMarkerDir = "send_queue_delivered" +) + +type sendQueueInput struct { + UserID string + MailboxID string + SentMessageID string + MessageID string + Source string + MailFrom string + HeaderFrom string + Recipients []string + MIMEBytes []byte + Now time.Time +} + +type sendQueueItem struct { + ID string + UserID string + MailboxID string + SentMessageID string + MessageID string + Source string + MailFrom string + HeaderFrom string + Recipients []string + MIMEBytes []byte + AttemptCount int + MaxAttempts int +} + +func (a *App) enqueueSend(ctx context.Context, in sendQueueInput) (string, error) { + if strings.TrimSpace(a.cfg.SMTPHost) == "" { + return "", nil + } + now := in.Now.UTC() + if now.IsZero() { + now = a.now().UTC() + } + id := newID("snd") + messageID := strings.TrimSpace(in.MessageID) + mimeBase64 := base64.StdEncoding.EncodeToString(in.MIMEBytes) + recipientsJSON := jsonEncode(dedupeEmails(in.Recipients)) + _, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + id, in.UserID, in.MailboxID, in.SentMessageID, messageID, in.Source, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano)) + if err != nil { + return "", err + } + if strings.TrimSpace(messageID) != "" { + var existingID, status string + var attemptCount, maxAttempts int + if err := a.db.QueryRowContext(ctx, `SELECT id,status,attempt_count,max_attempts FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=?`, in.MailboxID, in.Source, messageID).Scan(&existingID, &status, &attemptCount, &maxAttempts); err != nil { + return "", err + } + if existingID != id { + if status == sendQueueStatusDelivered || (status == sendQueueStatusFailed && attemptCount >= maxAttempts) { + _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET user_id=?,sent_message_id=?,mail_from=?,header_from=?,recipients_json=?,mime_base64=?,status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=?`, + in.UserID, in.SentMessageID, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), existingID, status) + if err != nil { + return "", err + } + a.deleteSendQueueDeliveredMarker(existingID) + a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{ + QueueID: existingID, + UserID: in.UserID, + MailboxID: in.MailboxID, + SentMessageID: in.SentMessageID, + Source: in.Source, + MailFrom: in.MailFrom, + HeaderFrom: in.HeaderFrom, + Recipients: in.Recipients, + }) + } + return existingID, nil + } + } + a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{ + QueueID: id, + UserID: in.UserID, + MailboxID: in.MailboxID, + SentMessageID: in.SentMessageID, + Source: in.Source, + MailFrom: in.MailFrom, + HeaderFrom: in.HeaderFrom, + Recipients: in.Recipients, + }) + return id, nil +} + +func (a *App) sendQueueWorker(ctx context.Context) { + a.log.Info("send queue worker started") + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for { + if err := a.processDueSendQueue(ctx); err != nil { + a.log.Warn("send queue worker failed", "error", err) + } + select { + case <-ctx.Done(): + a.log.Info("send queue worker stopped") + return + case <-ticker.C: + } + } +} + +func (a *App) processDueSendQueue(ctx context.Context) error { + if strings.TrimSpace(a.cfg.SMTPHost) == "" { + return nil + } + if err := a.recoverStaleSendQueueItems(ctx); err != nil { + return err + } + rows, err := a.db.QueryContext(ctx, `SELECT id FROM send_queue WHERE (status=? OR (status=? AND attempt_count 0 { + a.recordSendAudit(ctx, sendAuditRetry, sendQueueStatusFailed, sendAuditInputFromQueue(item, "send attempt interrupted")) + } + } + return nil +} + +func (a *App) processSendQueueItem(ctx context.Context, id string) { + item, err := a.claimSendQueueItem(ctx, id) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + a.log.Warn("failed to claim send queue item", "id", id, "error", err) + } + return + } + if err := a.sendSMTP(item.MailFrom, item.Recipients, item.MIMEBytes); err != nil { + a.markSendQueueFailed(ctx, item, err) + return + } + if err := a.writeSendQueueDeliveredMarker(item.ID); err != nil { + a.log.Warn("failed to persist send queue delivered marker", "id", item.ID, "error", err) + } + if err := a.markSendQueueDelivered(ctx, item); err != nil { + a.log.Warn("failed to mark send queue delivered", "id", item.ID, "error", err) + return + } +} + +func (a *App) markSendQueueDelivered(ctx context.Context, item sendQueueItem) error { + now := a.now().UTC().Format(time.RFC3339Nano) + if _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,delivered_at=?,updated_at=?,last_error='',mime_base64='' WHERE id=?`, sendQueueStatusDelivered, now, now, item.ID); err != nil { + return err + } + a.deleteSendQueueDeliveredMarker(item.ID) + a.recordSendAudit(ctx, sendAuditDelivered, sendQueueStatusDelivered, sendAuditInputFromQueue(item, "")) + return nil +} + +func (a *App) claimSendQueueItem(ctx context.Context, id string) (sendQueueItem, error) { + now := a.now().UTC().Format(time.RFC3339Nano) + res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,attempt_count=attempt_count+1,updated_at=? WHERE id=? AND (status=? OR (status=? AND attempt_count= item.MaxAttempts { + nextAttempt = now.Add(365 * 24 * time.Hour) + } + _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=?`, status, nextAttempt.Format(time.RFC3339Nano), sendErr.Error(), now.Format(time.RFC3339Nano), item.ID) + if err != nil { + a.log.Warn("failed to mark send queue failed", "id", item.ID, "error", err) + } + event := sendAuditRetry + if item.AttemptCount >= item.MaxAttempts { + event = sendAuditFailed + } + a.recordSendAudit(ctx, event, status, sendAuditInputFromQueue(item, sendErr.Error())) +} + +func sendRetryDelay(attempt int) time.Duration { + if attempt < 1 { + attempt = 1 + } + delays := []time.Duration{30 * time.Second, 2 * time.Minute, 10 * time.Minute, time.Hour, 6 * time.Hour} + if attempt > len(delays) { + return delays[len(delays)-1] + } + return delays[attempt-1] +} + +type sendAuditInput struct { + QueueID string + UserID string + MailboxID string + SentMessageID string + Source string + MailFrom string + HeaderFrom string + Recipients []string + Error string +} + +func sendAuditInputFromQueue(item sendQueueItem, errorText string) sendAuditInput { + return sendAuditInput{ + QueueID: item.ID, + UserID: item.UserID, + MailboxID: item.MailboxID, + SentMessageID: item.SentMessageID, + Source: item.Source, + MailFrom: item.MailFrom, + HeaderFrom: item.HeaderFrom, + Recipients: item.Recipients, + Error: errorText, + } +} + +func (a *App) recordSendAudit(ctx context.Context, event, status string, in sendAuditInput) { + source := strings.TrimSpace(in.Source) + if source == "" { + source = "unknown" + } + _, err := a.db.ExecContext(ctx, `INSERT INTO send_audit_events(id,queue_id,user_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, newID("audit"), in.QueueID, in.UserID, in.MailboxID, in.SentMessageID, source, event, status, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), jsonEncode(dedupeEmails(in.Recipients)), in.Error, a.now().UTC().Format(time.RFC3339Nano)) + if err != nil { + a.log.Warn("failed to record send audit", "event", event, "error", err) + } +} + +func (a *App) sendQueueDeliveredMarkerPath(id string) string { + safeID := filepath.Base(strings.TrimSpace(id)) + if safeID == "" || safeID == "." { + safeID = "unknown" + } + return filepath.Join(a.cfg.DataDir, sendQueueDeliveredMarkerDir, safeID+".marker") +} + +func (a *App) writeSendQueueDeliveredMarker(id string) error { + path := a.sendQueueDeliveredMarkerPath(id) + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + tmp := filepath.Join(dir, filepath.Base(path)+"."+newID("tmp")) + if err := os.WriteFile(tmp, []byte(a.now().UTC().Format(time.RFC3339Nano)), 0o600); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +} + +func (a *App) hasSendQueueDeliveredMarker(id string) (bool, error) { + _, err := os.Stat(a.sendQueueDeliveredMarkerPath(id)) + if err == nil { + return true, nil + } + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + return false, err +} + +func (a *App) deleteSendQueueDeliveredMarker(id string) { + err := os.Remove(a.sendQueueDeliveredMarkerPath(id)) + if err != nil && !errors.Is(err, os.ErrNotExist) { + a.log.Warn("failed to remove send queue delivered marker", "id", id, "error", err) + } +} + +func (a *App) authorizedSender(ctx context.Context, mb *Mailbox, from string) (string, string, error) { + from = normalizeEmail(from) + if from == "" { + from = normalizeEmail(mb.Address) + } + if from == normalizeEmail(mb.Address) { + return normalizeEmail(mb.Address), mb.DisplayName, nil + } + var displayName string + var enabled int + err := a.db.QueryRowContext(ctx, `SELECT display_name,enabled FROM send_as_grants WHERE mailbox_id=? AND address=?`, mb.ID, from).Scan(&displayName, &enabled) + if err == nil { + if enabled == 0 { + return "", "", errSenderNotAuthorized + } + return from, strings.TrimSpace(displayName), nil + } + if !errors.Is(err, sql.ErrNoRows) { + return "", "", err + } + var aliasDestination string + err = a.db.QueryRowContext(ctx, `SELECT destination FROM aliases WHERE source=? AND enabled=1`, from).Scan(&aliasDestination) + if err == nil { + for _, destination := range strings.Split(aliasDestination, ",") { + if normalizeEmail(destination) == normalizeEmail(mb.Address) { + return from, mb.DisplayName, nil + } + } + } + return "", "", errSenderNotAuthorized +} diff --git a/apps/api/internal/app/submission.go b/apps/api/internal/app/submission.go new file mode 100644 index 0000000..27195a0 --- /dev/null +++ b/apps/api/internal/app/submission.go @@ -0,0 +1,497 @@ +package app + +import ( + "bytes" + "context" + "crypto/tls" + "database/sql" + "errors" + "fmt" + "io" + "log" + netmail "net/mail" + "net/textproto" + "sort" + "strings" + "time" + + "github.com/emersion/go-sasl" + smtpserver "github.com/emersion/go-smtp" + "golang.org/x/crypto/bcrypt" +) + +const ( + defaultSubmissionMaxRecipients = 200 +) + +type SubmissionServers struct { + Plain *smtpserver.Server + TLS *smtpserver.Server +} + +func (s *SubmissionServers) Shutdown(ctx context.Context) error { + if s == nil { + return nil + } + var errs []error + if s.Plain != nil { + if err := s.Plain.Shutdown(ctx); err != nil && !errors.Is(err, smtpserver.ErrServerClosed) { + errs = append(errs, err) + } + } + if s.TLS != nil { + if err := s.TLS.Shutdown(ctx); err != nil && !errors.Is(err, smtpserver.ErrServerClosed) { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +func (a *App) NewSubmissionServers(tlsConfig *tls.Config) *SubmissionServers { + return &SubmissionServers{ + Plain: a.newSubmissionServer(a.cfg.SubmissionAddr, tlsConfig), + TLS: a.newSubmissionServer(a.cfg.SubmissionTLSAddr, tlsConfig), + } +} + +func (a *App) newSubmissionServer(addr string, tlsConfig *tls.Config) *smtpserver.Server { + addr = strings.TrimSpace(addr) + if addr == "" { + return nil + } + s := smtpserver.NewServer(submissionBackend{app: a}) + s.Addr = addr + s.Domain = a.cfg.PublicHostname + s.TLSConfig = tlsConfig + s.AllowInsecureAuth = false + s.MaxRecipients = defaultSubmissionMaxRecipients + s.MaxMessageBytes = int64(a.cfg.SubmissionMaxMessageMB) * 1024 * 1024 + s.ReadTimeout = smtpSessionTimeout + s.WriteTimeout = smtpSessionTimeout + s.ErrorLog = log.New(submissionLogWriter{log: a.log}, "smtp/submission ", 0) + return s +} + +func LoadServerTLSConfig(cfg Config) (*tls.Config, error) { + certFile, keyFile := strings.TrimSpace(cfg.TLSCertFile), strings.TrimSpace(cfg.TLSKeyFile) + if certFile == "" || keyFile == "" { + return nil, errors.New("LANQIN_TLS_CERT_FILE and LANQIN_TLS_KEY_FILE are required when SMTP submission is enabled") + } + return &tls.Config{ + MinVersion: tls.VersionTLS12, + GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) { + cert, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return nil, err + } + return &cert, nil + }, + }, nil +} + +type submissionLogWriter struct { + log slogLogger +} + +func (w submissionLogWriter) Write(p []byte) (int, error) { + if w.log != nil { + w.log.Warn(strings.TrimSpace(string(p))) + } + return len(p), nil +} + +type slogLogger interface { + Warn(msg string, args ...any) +} + +type submissionBackend struct { + app *App +} + +func (b submissionBackend) NewSession(*smtpserver.Conn) (smtpserver.Session, error) { + return &submissionSession{app: b.app}, nil +} + +type submissionSession struct { + app *App + user *User + mailbox *Mailbox + mailFrom string + recipients []string +} + +func (s *submissionSession) AuthMechanisms() []string { + return []string{sasl.Plain} +} + +func (s *submissionSession) Auth(mech string) (sasl.Server, error) { + if !strings.EqualFold(mech, sasl.Plain) { + return nil, smtpserver.ErrAuthUnknownMechanism + } + return sasl.NewPlainServer(func(identity, username, password string) error { + user, mailbox, err := s.app.authenticateSubmission(context.Background(), username, password) + if err != nil { + return smtpserver.ErrAuthFailed + } + s.user, s.mailbox = user, mailbox + return nil + }), nil +} + +func (s *submissionSession) Mail(from string, _ *smtpserver.MailOptions) error { + if s.user == nil || s.mailbox == nil { + return smtpserver.ErrAuthRequired + } + from = normalizeEmail(from) + authorized, _, err := s.app.authorizedSender(context.Background(), s.mailbox, from) + if err != nil || from == "" || from != authorized { + return smtpError(553, smtpserver.EnhancedCode{5, 7, 1}, "sender must match authenticated mailbox") + } + s.mailFrom = from + s.recipients = nil + return nil +} + +func (s *submissionSession) Rcpt(to string, _ *smtpserver.RcptOptions) error { + if s.user == nil || s.mailbox == nil { + return smtpserver.ErrAuthRequired + } + to = normalizeEmail(to) + if to == "" || !strings.Contains(to, "@") { + return smtpError(501, smtpserver.EnhancedCode{5, 1, 3}, "invalid recipient") + } + s.recipients = append(s.recipients, to) + return nil +} + +func (s *submissionSession) Data(r io.Reader) error { + if s.user == nil || s.mailbox == nil { + return smtpserver.ErrAuthRequired + } + if s.mailFrom == "" || len(s.recipients) == 0 { + return smtpError(503, smtpserver.EnhancedCode{5, 5, 1}, "missing sender or recipients") + } + if err := s.app.submitSMTPMessage(context.Background(), s.user, s.mailbox, s.mailFrom, s.recipients, r); err != nil { + var smtpErr *smtpserver.SMTPError + if errors.As(err, &smtpErr) { + return smtpErr + } + return smtpError(451, smtpserver.EnhancedCode{4, 0, 0}, "message submission failed") + } + s.Reset() + return nil +} + +func (s *submissionSession) Reset() { + s.mailFrom = "" + s.recipients = nil +} + +func (s *submissionSession) Logout() error { + s.Reset() + return nil +} + +func (a *App) authenticateSubmission(ctx context.Context, username, password string) (*User, *Mailbox, error) { + address := normalizeEmail(username) + if address == "" { + return nil, nil, errors.New("missing username") + } + var mb Mailbox + var passwordHash, created string + row := a.db.QueryRowContext(ctx, `SELECT id,user_id,domain_id,local_part,address,display_name,password_hash,quota_mb,status,created_at + FROM mailboxes WHERE address=? AND status='active'`, address) + if err := row.Scan(&mb.ID, &mb.UserID, &mb.DomainID, &mb.LocalPart, &mb.Address, &mb.DisplayName, &passwordHash, &mb.QuotaMB, &mb.Status, &created); err != nil { + return nil, nil, err + } + if err := bcrypt.CompareHashAndPassword([]byte(passwordHash), []byte(password)); err != nil { + return nil, nil, err + } + mb.CreatedAt = parseTime(created) + user, err := a.userByID(ctx, mb.UserID) + if err != nil { + return nil, nil, err + } + if user.Disabled { + return nil, nil, errors.New("user disabled") + } + if !userHasPermission(user, PermissionMailSend) { + return nil, nil, errors.New("send permission required") + } + return user, &mb, nil +} + +func (a *App) submitSMTPMessage(ctx context.Context, user *User, mb *Mailbox, mailFrom string, recipients []string, r io.Reader) error { + if err := a.recordSMTPRate(ctx, user, mb); err != nil { + if errors.Is(err, errSMTPRateLimited) { + return smtpError(452, smtpserver.EnhancedCode{4, 7, 0}, err.Error()) + } + return err + } + raw, err := io.ReadAll(r) + if err != nil { + return err + } + prepared, msg, attachments, err := a.prepareSubmittedMessage(ctx, raw, mb, mailFrom, recipients) + if err != nil { + return err + } + msg.MailboxID = mb.ID + sentID, insertedSent, err := a.insertSentMessageOnce(ctx, msg, attachments) + if err != nil { + return err + } + a.recordSendAudit(ctx, sendAuditAccepted, sendQueueStatusQueued, sendAuditInput{UserID: user.ID, MailboxID: mb.ID, SentMessageID: sentID, Source: sendSourceSubmission, MailFrom: mailFrom, HeaderFrom: msg.From, Recipients: recipients}) + if sentID != "" { + if _, err := a.enqueueSend(ctx, sendQueueInput{UserID: user.ID, MailboxID: mb.ID, SentMessageID: sentID, MessageID: msg.MessageID, Source: sendSourceSubmission, MailFrom: mailFrom, HeaderFrom: msg.From, Recipients: recipients, MIMEBytes: prepared, Now: a.now().UTC()}); err != nil { + if insertedSent { + a.deleteMessage(ctx, sentID) + if sentFolderID, ferr := a.ensureFolder(ctx, mb.ID, "Sent"); ferr == nil { + a.deleteSentDedupeKey(ctx, mb.ID, sentFolderID, msg.MessageID) + } + } + return err + } + } + return nil +} + +func (a *App) prepareSubmittedMessage(ctx context.Context, raw []byte, mb *Mailbox, mailFrom string, recipients []string) ([]byte, storedMessage, []AttachmentInput, error) { + header, body, err := readMessageHeader(raw) + if err != nil { + return nil, storedMessage{}, nil, smtpError(554, smtpserver.EnhancedCode{5, 6, 0}, "invalid message") + } + fromAddress, fromName, ok := singleHeaderAddress(header.Get("From")) + if !ok || fromAddress == "" { + return nil, storedMessage{}, nil, smtpError(550, smtpserver.EnhancedCode{5, 7, 1}, "From header must contain exactly one address") + } + authAddress, fromName, err := a.authorizedSender(ctx, mb, fromAddress) + if err != nil || normalizeEmail(mailFrom) != authAddress || normalizeEmail(fromAddress) != authAddress { + return nil, storedMessage{}, nil, smtpError(553, smtpserver.EnhancedCode{5, 7, 1}, "sender must match authenticated mailbox") + } + now := a.now().UTC() + messageID := strings.TrimSpace(header.Get("Message-Id")) + if messageID == "" { + messageID = fmt.Sprintf("<%s@%s>", newID("msg"), domainPart(authAddress)) + header.Set("Message-ID", messageID) + } else { + header.Set("Message-ID", messageID) + } + sentAt := parseMailDate(header.Get("Date")) + if sentAt.IsZero() { + sentAt = now + header.Set("Date", sentAt.Format(time.RFC1123Z)) + } + header.Del("Bcc") + prepared := serializeMessage(header, body) + msg, attachments, err := a.parseMaildirMessage(prepared, authAddress) + if err != nil { + return nil, storedMessage{}, nil, smtpError(554, smtpserver.EnhancedCode{5, 6, 0}, "invalid message") + } + if msg.MessageID == "" { + msg.MessageID = messageID + } + if msg.SentAt.IsZero() { + msg.SentAt = sentAt + } + if msg.ReceivedAt.IsZero() { + msg.ReceivedAt = sentAt + } + msg.From = authAddress + msg.FromName = fromName + msg.To = dedupeEmails(msg.To) + msg.CC = dedupeEmails(msg.CC) + msg.BCC = deduceBCCRecipients(recipients, addressList(header.Get("To")), addressList(header.Get("Cc"))) + msg.IsRead = true + msg.RawPath = "" + if msg.Subject == "" { + msg.Subject = "(no subject)" + } + if msg.Snippet == "" { + msg.Snippet = snippetFrom(msg.BodyText, msg.BodyHTML) + } + return prepared, msg, attachments, nil +} + +func (a *App) insertSentMessageOnce(ctx context.Context, msg storedMessage, attachments []AttachmentInput) (string, bool, error) { + sentFolderID, err := a.ensureFolder(ctx, msg.MailboxID, "Sent") + if err != nil { + return "", false, err + } + msg.FolderID = sentFolderID + if msg.MessageUID == "" { + msg.MessageUID = newID("uid") + } + tx, err := a.db.BeginTx(ctx, nil) + if err != nil { + return "", false, err + } + committed := false + messageIDForCleanup := "" + defer func() { + if !committed { + _ = tx.Rollback() + if messageIDForCleanup != "" { + a.deleteMessageFiles(ctx, messageIDForCleanup) + } + } + }() + if msg.MessageID != "" { + existing, err := sentMessageIDByMessageID(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID) + if err == nil { + if err := a.insertSentDedupeKeyWithDB(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil && !errors.Is(err, errSentDedupeExists) { + return "", false, err + } + if err := tx.Commit(); err != nil { + return "", false, err + } + committed = true + return existing, false, nil + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return "", false, err + } + if err := a.insertSentDedupeKeyWithDB(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil { + if errors.Is(err, errSentDedupeExists) { + existing, qerr := sentMessageIDByMessageID(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID) + if qerr == nil { + if err := tx.Commit(); err != nil { + return "", false, err + } + committed = true + return existing, false, nil + } + if errors.Is(qerr, sql.ErrNoRows) { + return "", false, fmt.Errorf("sent dedupe key exists without sent message: %w", errSentDedupeExists) + } + return "", false, qerr + } + return "", false, err + } + } + id, err := a.insertMessageWithDB(ctx, tx, msg, attachments) + if err != nil { + return "", false, err + } + messageIDForCleanup = id + if err := tx.Commit(); err != nil { + return "", false, err + } + committed = true + return id, true, nil +} + +var errSentDedupeExists = errors.New("sent message already exists") + +func (a *App) insertSentDedupeKey(ctx context.Context, mailboxID, folderID, messageID string) error { + return a.insertSentDedupeKeyWithDB(ctx, a.db, mailboxID, folderID, messageID) +} + +func (a *App) insertSentDedupeKeyWithDB(ctx context.Context, db dbExecutor, mailboxID, folderID, messageID string) error { + if strings.TrimSpace(messageID) == "" { + return nil + } + res, err := db.ExecContext(ctx, `INSERT OR IGNORE INTO sent_message_dedupe_keys(mailbox_id,folder_id,message_id,created_at) VALUES(?,?,?,?)`, mailboxID, folderID, messageID, a.now().UTC().Format(time.RFC3339Nano)) + if err != nil { + return err + } + if rows, err := res.RowsAffected(); err == nil && rows == 0 { + return errSentDedupeExists + } + return nil +} + +func sentMessageIDByMessageID(ctx context.Context, db dbQueryer, mailboxID, folderID, messageID string) (string, error) { + var existing string + err := db.QueryRowContext(ctx, `SELECT id FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=? AND message_id <> '' LIMIT 1`, mailboxID, folderID, messageID).Scan(&existing) + return existing, err +} + +func (a *App) deleteSentDedupeKey(ctx context.Context, mailboxID, folderID, messageID string) { + if strings.TrimSpace(messageID) == "" { + return + } + _, _ = a.db.ExecContext(ctx, `DELETE FROM sent_message_dedupe_keys WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mailboxID, folderID, messageID) +} + +func readMessageHeader(raw []byte) (textproto.MIMEHeader, []byte, error) { + msg, err := netmail.ReadMessage(bytes.NewReader(raw)) + if err != nil { + return nil, nil, err + } + body, err := io.ReadAll(msg.Body) + if err != nil { + return nil, nil, err + } + return textproto.MIMEHeader(msg.Header), body, nil +} + +func serializeMessage(header textproto.MIMEHeader, body []byte) []byte { + var buf bytes.Buffer + keys := make([]string, 0, len(header)) + for key := range header { + keys = append(keys, key) + } + sort.SliceStable(keys, func(i, j int) bool { + return textproto.CanonicalMIMEHeaderKey(keys[i]) < textproto.CanonicalMIMEHeaderKey(keys[j]) + }) + for _, key := range keys { + values := header[key] + canonical := textproto.CanonicalMIMEHeaderKey(key) + for _, value := range values { + fmt.Fprintf(&buf, "%s: %s\r\n", canonical, strings.ReplaceAll(strings.ReplaceAll(value, "\r", ""), "\n", " ")) + } + } + buf.WriteString("\r\n") + buf.Write(body) + return buf.Bytes() +} + +func singleHeaderAddress(value string) (string, string, bool) { + value = strings.TrimSpace(value) + if value == "" { + return "", "", false + } + items, err := netmail.ParseAddressList(value) + if err != nil || len(items) != 1 { + decoded := decodeMIMEHeader(value) + items, err = netmail.ParseAddressList(decoded) + if err != nil || len(items) != 1 { + return "", "", false + } + } + item := items[0] + return normalizeEmail(item.Address), strings.TrimSpace(decodeMIMEHeader(item.Name)), true +} + +func deduceBCCRecipients(envelope, to, cc []string) []string { + visible := map[string]bool{} + for _, item := range append(to, cc...) { + if email := normalizeEmail(item); email != "" { + visible[email] = true + } + } + var out []string + seen := map[string]bool{} + for _, item := range envelope { + email := normalizeEmail(item) + if email == "" || visible[email] || seen[email] { + continue + } + seen[email] = true + out = append(out, email) + } + return out +} + +func domainPart(email string) string { + parts := strings.SplitN(normalizeEmail(email), "@", 2) + if len(parts) != 2 || parts[1] == "" { + return "lanqin.local" + } + return parts[1] +} + +func smtpError(code int, enhanced smtpserver.EnhancedCode, message string) *smtpserver.SMTPError { + return &smtpserver.SMTPError{Code: code, EnhancedCode: enhanced, Message: message} +} diff --git a/deploy/.env.example b/deploy/.env.example index 174bdd6..5ca83e2 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -28,7 +28,7 @@ LANQIN_PUBLIC_BASE_URL=https://mail.example.com # 邮件客户端 TLS 证书。用于 SMTP 465/587、IMAP 993、POP3 995。 # 生产环境请挂载真实证书，并确保域名包含 LANQIN_PUBLIC_HOSTNAME。 -# 留空时会使用容器自带 localhost 自签证书，第三方客户端会提示证书不匹配。 +# 留空时 Dovecot/Postfix 会使用容器自带 localhost 自签证书；LanQin API 的 SMTP submission 不会启用。 LANQIN_TLS_CERT_FILE= LANQIN_TLS_KEY_FILE= @@ -84,15 +84,24 @@ LANQIN_TURNSTILE_SECRET_KEY= # SMTP 发信 # ========================= # 单容器部署默认提交给容器内 Postfix。 +# Split stack 会由 docker-compose.stack.yml 默认覆盖为 postfix:25。 +# 如需在 split stack 使用外部 SMTP，可设置 LANQIN_STACK_SMTP_HOST / LANQIN_STACK_SMTP_PORT。 # 如果要走外部 SMTP，把 Host/Port/Username/Password 改成外部服务配置。 LANQIN_SMTP_HOST=127.0.0.1 LANQIN_SMTP_PORT=25 +LANQIN_STACK_SMTP_HOST= +LANQIN_STACK_SMTP_PORT= LANQIN_SMTP_USERNAME= LANQIN_SMTP_PASSWORD= # 外部 SMTP 要求 STARTTLS / TLS 时改 true；本机 Postfix 默认 false。 LANQIN_SMTP_REQUIRE_TLS=false +# 第三方客户端 SMTP 提交，由 LanQin API 监听 587/465；启用前必须配置可读 TLS 证书。 +LANQIN_SUBMISSION_ADDR= +LANQIN_SUBMISSION_TLS_ADDR= +LANQIN_SUBMISSION_MAX_MESSAGE_MB=35 + # ========================= # 收件 / Maildir 同步 # ========================= diff --git a/deploy/README.md b/deploy/README.md index d0bff83..9597cc1 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -128,18 +128,21 @@ docker compose -f docker-compose.stack.yml -f docker-compose.stack.build.yml up - Rspamd 会周期性从 SQLite 导出域名 DKIM 私钥到容器内 `/var/lib/rspamd/dkim`。 - Go API 是 Webmail 和管理后台入口；浏览器不直接连接 SMTP/IMAP/POP3。 - Go API 会读取 `LANQIN_MAILDIR_ROOT=/var/mail/vhosts`，周期扫描 Maildir，把 Postfix/Dovecot 入站邮件同步成 Webmail 索引。 -- 第三方客户端可通过 SMTP `465/587` 发信；Webmail 内的“已发送”由 Webmail API 发信流程写入。 +- 第三方客户端可通过 LanQin API 提供的 SMTP `465/587` 发信；Webmail/API 和第三方客户端的“已发送”都由 API 写入，外发投递进入发送队列并由 API worker relay/retry，客户端后续 IMAP APPEND 到 Sent 会按 `Message-ID` 去重。 +- send-as v1 支持本人邮箱、启用的别名转发 source 指向本人邮箱，或数据库表 `send_as_grants` 中显式授权的地址。 ## 邮件客户端 TLS 证书 Web 站点可以由宿主机 Nginx / 宝塔反代到容器 `80`，但 SMTP/IMAP/POP3 端口不会使用 Web 反代的证书。 -如果第三方客户端连接 `465/587/993/995` 时提示证书是 `localhost`，说明 Postfix/Dovecot 仍在使用容器自带的测试证书。 +如果第三方客户端连接 `993/995` 时提示证书是 `localhost`，说明 Dovecot 仍在使用容器自带的测试证书。LanQin API 的 SMTP `465/587` submission 不会使用自签测试证书；启用前必须配置可读的真实证书。 生产环境请把域名证书挂载进容器，并在 `.env` 指向证书文件： ```env LANQIN_TLS_CERT_FILE=/certs/fullchain.pem LANQIN_TLS_KEY_FILE=/certs/privkey.pem +LANQIN_SUBMISSION_ADDR=:587 +LANQIN_SUBMISSION_TLS_ADDR=:465 ``` 单容器示例： @@ -170,12 +173,16 @@ LANQIN_SMTP_PORT=25 LANQIN_SMTP_REQUIRE_TLS=false ``` -如果页面提示 `smtp delivery failed: EOF`，通常是 Postfix 会话被中断。优先检查： +Split stack 使用 `docker-compose.stack.yml` 时，API 容器默认会把 `LANQIN_SMTP_HOST` 覆盖为 `postfix`，让 Webmail 和 SMTP 提交都 relay 到 Postfix service。只有改用外部 SMTP 时才需要在 `.env` 明确填写 `LANQIN_STACK_SMTP_HOST` / `LANQIN_STACK_SMTP_PORT`。 + +如果发送队列里出现 relay 失败，通常是 Postfix 会话被中断或外部 SMTP 配置错误。优先检查： ```bash docker compose exec lanqin-email supervisorctl status -docker compose exec lanqin-email postconf -M smtp/inet submission/inet +docker compose exec lanqin-email postconf -M smtp/inet +# SMTP 提交 465/587 由 LanQin API 提供，不再由 Postfix 监听。 docker compose exec lanqin-email sqlite3 /data/lanqin.db "select key,value from system_settings where key like 'smtp%' order by key;" +docker compose exec lanqin-email sqlite3 /data/lanqin.db "select status,attempt_count,last_error from send_queue order by created_at desc limit 10;" docker compose logs --tail=200 lanqin-email ``` diff --git a/deploy/all-in-one/entrypoint.sh b/deploy/all-in-one/entrypoint.sh index 8b73053..1ca11d6 100644 --- a/deploy/all-in-one/entrypoint.sh +++ b/deploy/all-in-one/entrypoint.sh @@ -7,11 +7,14 @@ set -eu : "${LANQIN_ADDR:=127.0.0.1:8080}" : "${LANQIN_SMTP_HOST:=127.0.0.1}" : "${LANQIN_SMTP_PORT:=25}" +: "${LANQIN_SUBMISSION_ADDR:=}" +: "${LANQIN_SUBMISSION_TLS_ADDR:=}" +: "${LANQIN_SUBMISSION_MAX_MESSAGE_MB:=35}" : "${LANQIN_MAILDIR_ROOT:=/var/mail/vhosts}" : "${LANQIN_TLS_CERT_FILE:=}" : "${LANQIN_TLS_KEY_FILE:=}" -export LANQIN_DATA_DIR LANQIN_DB_PATH LANQIN_ADDR LANQIN_SMTP_HOST LANQIN_SMTP_PORT LANQIN_MAILDIR_ROOT +export LANQIN_DATA_DIR LANQIN_DB_PATH LANQIN_ADDR LANQIN_SMTP_HOST LANQIN_SMTP_PORT LANQIN_SUBMISSION_ADDR LANQIN_SUBMISSION_TLS_ADDR LANQIN_SUBMISSION_MAX_MESSAGE_MB LANQIN_MAILDIR_ROOT LANQIN_TLS_CERT_FILE LANQIN_TLS_KEY_FILE addgroup --system --gid 5000 vmail 2>/dev/null || true adduser --system --uid 5000 --gid 5000 --home /var/mail/vhosts --no-create-home vmail 2>/dev/null || true @@ -37,17 +40,24 @@ if [ -n "$LANQIN_TLS_CERT_FILE" ] || [ -n "$LANQIN_TLS_KEY_FILE" ]; then if [ -f "$LANQIN_TLS_CERT_FILE" ] && [ -f "$LANQIN_TLS_KEY_FILE" ]; then TLS_CERT="$LANQIN_TLS_CERT_FILE" TLS_KEY="$LANQIN_TLS_KEY_FILE" + : "${LANQIN_SUBMISSION_ADDR:=:587}" + : "${LANQIN_SUBMISSION_TLS_ADDR:=:465}" else echo "warning: LANQIN_TLS_CERT_FILE/LANQIN_TLS_KEY_FILE not readable; using snakeoil localhost certificate" >&2 fi fi +if [ -n "$LANQIN_SUBMISSION_ADDR$LANQIN_SUBMISSION_TLS_ADDR" ] && { [ "$TLS_CERT" = "/etc/ssl/certs/ssl-cert-snakeoil.pem" ] || [ "$TLS_KEY" = "/etc/ssl/private/ssl-cert-snakeoil.key" ]; }; then + echo "warning: SMTP submission disabled because LANQIN_TLS_CERT_FILE/LANQIN_TLS_KEY_FILE are not configured with readable certificate files" >&2 + LANQIN_SUBMISSION_ADDR="" + LANQIN_SUBMISSION_TLS_ADDR="" +fi +export LANQIN_SUBMISSION_ADDR LANQIN_SUBMISSION_TLS_ADDR postconf -e "myhostname = ${LANQIN_PUBLIC_HOSTNAME}" postconf -e "myorigin = ${LANQIN_PUBLIC_HOSTNAME}" postconf -e "smtpd_tls_cert_file = ${TLS_CERT}" postconf -e "smtpd_tls_key_file = ${TLS_KEY}" postconf -e "virtual_transport = lmtp:inet:127.0.0.1:24" -postconf -e "smtpd_sasl_path = inet:127.0.0.1:12345" postconf -e "milter_mail_macros = i {mail_addr} {client_addr} {client_name} {auth_authen}" postconf -e "smtpd_milters = inet:127.0.0.1:11332" postconf -e "non_smtpd_milters = inet:127.0.0.1:11332" diff --git a/deploy/api.Dockerfile b/deploy/api.Dockerfile index 19ac9ea..d8c99a2 100644 --- a/deploy/api.Dockerfile +++ b/deploy/api.Dockerfile @@ -17,5 +17,5 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ apt-get update && apt-get install -y --no-install-recommends ca-certificates tzdata WORKDIR /app COPY --from=build /out/lanqin-api /usr/local/bin/lanqin-api -EXPOSE 8080 +EXPOSE 8080 465 587 CMD ["lanqin-api"] diff --git a/deploy/docker-compose.stack.yml b/deploy/docker-compose.stack.yml index d88c6f3..5d85790 100644 --- a/deploy/docker-compose.stack.yml +++ b/deploy/docker-compose.stack.yml @@ -2,9 +2,21 @@ services: api: image: ${LANQIN_API_IMAGE:-ghcr.io/lanqin996/lanqin-email-api:latest} env_file: .env + environment: + LANQIN_SMTP_HOST: ${LANQIN_STACK_SMTP_HOST:-postfix} + LANQIN_SMTP_PORT: ${LANQIN_STACK_SMTP_PORT:-25} + LANQIN_SUBMISSION_ADDR: ${LANQIN_SUBMISSION_ADDR:-} + LANQIN_SUBMISSION_TLS_ADDR: ${LANQIN_SUBMISSION_TLS_ADDR:-} volumes: - - ./data:/data:ro + - ./data:/data - ./mail:/var/mail/vhosts:ro + # 生产环境如需第三方客户端校验证书，请取消下面挂载的注释，并在 .env 配置： + # LANQIN_TLS_CERT_FILE=/certs/fullchain.pem + # LANQIN_TLS_KEY_FILE=/certs/privkey.pem + # - /etc/letsencrypt/live/${LANQIN_PUBLIC_HOSTNAME}:/certs:ro + ports: + - "465:465" + - "587:587" depends_on: - dovecot - postfix @@ -36,8 +48,6 @@ services: # - /etc/letsencrypt:/etc/letsencrypt:ro ports: - "25:25" - - "465:465" - - "587:587" depends_on: - dovecot - rspamd diff --git a/deploy/postfix/Dockerfile b/deploy/postfix/Dockerfile index 8b2e1cb..4db9b69 100644 --- a/deploy/postfix/Dockerfile +++ b/deploy/postfix/Dockerfile @@ -11,5 +11,5 @@ COPY master.cf /etc/postfix/master.cf COPY sqlite-*.cf /etc/postfix/ COPY entrypoint.sh /entrypoint.sh RUN chmod +x /entrypoint.sh -EXPOSE 25 465 587 +EXPOSE 25 CMD ["/entrypoint.sh"] diff --git a/deploy/postfix/main.cf b/deploy/postfix/main.cf index 749be76..596115a 100644 --- a/deploy/postfix/main.cf +++ b/deploy/postfix/main.cf @@ -14,13 +14,11 @@ virtual_transport = lmtp:inet:dovecot:24 virtual_mailbox_base = /var/mail/vhosts smtpd_banner = $myhostname ESMTP LanQin Email -smtpd_relay_restrictions = permit_mynetworks, permit_sasl_authenticated, reject_unauth_destination -smtpd_recipient_restrictions = permit_mynetworks, permit_sasl_authenticated, reject_unauth_destination +smtpd_relay_restrictions = permit_mynetworks, reject_unauth_destination +smtpd_recipient_restrictions = permit_mynetworks, reject_unauth_destination -# Submission auth via Dovecot. -smtpd_sasl_type = dovecot -smtpd_sasl_path = inet:dovecot:12345 -smtpd_sasl_auth_enable = yes +# 465/587 提交由 LanQin API 处理；Postfix 25 只负责入站和内部 relay。 +smtpd_sasl_auth_enable = no smtpd_tls_cert_file = /etc/ssl/certs/ssl-cert-snakeoil.pem smtpd_tls_key_file = /etc/ssl/private/ssl-cert-snakeoil.key smtpd_tls_security_level = may diff --git a/deploy/postfix/master.cf b/deploy/postfix/master.cf index 245305a..92efe93 100644 --- a/deploy/postfix/master.cf +++ b/deploy/postfix/master.cf @@ -1,14 +1,4 @@ smtp inet n - n - - smtpd -submission inet n - n - - smtpd - -o syslog_name=postfix/submission - -o smtpd_tls_security_level=may - -o smtpd_sasl_auth_enable=yes - -o smtpd_relay_restrictions=permit_sasl_authenticated,reject -smtps inet n - n - - smtpd - -o syslog_name=postfix/smtps - -o smtpd_tls_wrappermode=yes - -o smtpd_sasl_auth_enable=yes - -o smtpd_relay_restrictions=permit_sasl_authenticated,reject pickup unix n - n 60 1 pickup cleanup unix n - n - 0 cleanup qmgr unix n - n 300 1 qmgr diff --git a/deploy/postfix/sqlite-sender-bcc.cf b/deploy/postfix/sqlite-sender-bcc.cf deleted file mode 100644 index 05cd7a9..0000000 --- a/deploy/postfix/sqlite-sender-bcc.cf +++ /dev/null @@ -1,2 +0,0 @@ -dbpath = /data/lanqin.db -query = SELECT local_part || '+Sent@' || substr(address, instr(address, '@') + 1) FROM mailboxes WHERE lower(address)=lower('%s') AND status='active'