szx/NewSzxcn-Email
1
0
Fork 0
Code Issues Pull Requests Actions 57 Packages Projects Releases Wiki Activity

Merge pull request #5 from LanQin996/feature/lanqin-submission

Browse Source 
Feature/lanqin submission
This commit is contained in:
LanQin_
2026-06-24 13:37:09 +08:00
committed by GitHub
parent f2457c63ee 65e4c4f6b5
commit 295a34881d
19 changed files with 2045 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+10
View File
@@ -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)
apps/api/cmd/server/main.go
+34
View File
@@ -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")
}
apps/api/go.mod
+2
View File
@@ -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
apps/api/go.sum
+4
View File
@@ -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=
apps/api/internal/app/app.go
+114
View File
@@ -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, &notnull, &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 {
apps/api/internal/app/app_test.go
+836 -6
View File
@@ -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 <admin@lanqin.local>",
"To: person@example.com",
"Bcc: hidden@example.com",
"Subject: Submission sent",
"Message-ID: <submission-sent@example.test>",
"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, "<submission-sent@example.test>").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: <relay-fail@example.test>\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, "<relay-fail@example.test>").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: <dedupe@example.test>\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, "<dedupe@example.test>").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, "<dedupe@example.test>").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 := "<orphan-sent-dedupe@example.test>"
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: <requeue@example.test>\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, "<requeue@example.test>"); 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, "<requeue@example.test>").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: <delivered-requeue@example.test>\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, "<delivered-requeue@example.test>").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 <team@lanqin.local>\r\nTo: person@example.com\r\nSubject: alias send-as\r\nMessage-ID: <alias-send-as@example.test>\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, "<alias-send-as@example.test>").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 <team-many@lanqin.local>\r\nTo: person@example.com\r\nSubject: alias send-as\r\nMessage-ID: <multi-alias-send-as@example.test>\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 <support@example.com>\r\nTo: person@example.com\r\nSubject: explicit send-as\r\nMessage-ID: <explicit-send-as@example.test>\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, "<explicit-send-as@example.test>").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", "<dup@example.test>", 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", "<dup@example.test>", 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='<dup@example.test>'`, 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='<dup@example.test>'`, 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: <starttls@example.test>\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: <smtps@example.test>\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")
}
}
apps/api/internal/app/config.go
+10
View File
@@ -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),
apps/api/internal/app/mail_handlers.go
+47 -13
View File
@@ -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) {
apps/api/internal/app/send_queue.go
+439
View File
@@ -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<max_attempts)) AND next_attempt_at<=? ORDER BY next_attempt_at, created_at LIMIT 20`, sendQueueStatusQueued, sendQueueStatusFailed, a.now().UTC().Format(time.RFC3339Nano))
if err != nil {
return err
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return err
}
ids = append(ids, id)
}
if err := rows.Err(); err != nil {
return err
}
sem := make(chan struct{}, sendQueueConcurrency)
done := make(chan struct{}, len(ids))
for _, id := range ids {
select {
case <-ctx.Done():
return ctx.Err()
case sem <- struct{}{}:
}
go func(id string) {
defer func() {
<-sem
done <- struct{}{}
}()
a.processSendQueueItem(ctx, id)
}(id)
}
for range ids {
select {
case <-ctx.Done():
return ctx.Err()
case <-done:
}
}
return nil
}
func (a *App) recoverStaleSendQueueItems(ctx context.Context) error {
cutoff := a.now().UTC().Add(-sendQueueStaleAfter).Format(time.RFC3339Nano)
rows, err := a.db.QueryContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE status=? AND updated_at<=? AND attempt_count<max_attempts LIMIT 20`, sendQueueStatusSending, cutoff)
if err != nil {
return err
}
defer rows.Close()
var items []sendQueueItem
for rows.Next() {
var item sendQueueItem
var recipientsJSON, mimeBase64 string
if err := rows.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
return err
}
item.Recipients = jsonDecodeSlice(recipientsJSON)
delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
if err != nil {
return err
}
if delivered {
items = append(items, item)
continue
}
mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
if err != nil {
return err
}
item.MIMEBytes = mimeBytes
items = append(items, item)
}
if err := rows.Err(); err != nil {
return err
}
now := a.now().UTC().Format(time.RFC3339Nano)
for _, item := range items {
delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
if err != nil {
return err
}
if delivered {
if err := a.markSendQueueDelivered(ctx, item); err != nil {
return err
}
continue
}
res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=? AND status=?`, sendQueueStatusFailed, now, "send attempt interrupted", now, item.ID, sendQueueStatusSending)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n > 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<max_attempts)) AND next_attempt_at<=?`, sendQueueStatusSending, now, id, sendQueueStatusQueued, sendQueueStatusFailed, now)
if err != nil {
return sendQueueItem{}, err
}
if n, _ := res.RowsAffected(); n == 0 {
return sendQueueItem{}, sql.ErrNoRows
}
var item sendQueueItem
var recipientsJSON, mimeBase64 string
row := a.db.QueryRowContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE id=?`, id)
if err := row.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
return sendQueueItem{}, err
}
item.Recipients = jsonDecodeSlice(recipientsJSON)
mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
if err != nil {
return sendQueueItem{}, err
}
item.MIMEBytes = mimeBytes
return item, nil
}
func (a *App) markSendQueueFailed(ctx context.Context, item sendQueueItem, sendErr error) {
now := a.now().UTC()
status := sendQueueStatusFailed
nextAttempt := now.Add(sendRetryDelay(item.AttemptCount))
if item.AttemptCount >= 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
}
apps/api/internal/app/submission.go
+497
View File
@@ -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}
}
deploy/.env.example
+10 -1
View File
@@ -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 同步
# =========================
deploy/README.md
+11 -4
View File
@@ -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
```
deploy/all-in-one/entrypoint.sh
+12 -2
View File
@@ -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"
deploy/api.Dockerfile
+1 -1
View File
@@ -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"]
deploy/docker-compose.stack.yml
+13 -3
View File
@@ -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
deploy/postfix/Dockerfile
+1 -1
View File
@@ -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"]
deploy/postfix/main.cf
+4 -6
View File
@@ -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
deploy/postfix/master.cf
-10
View File
@@ -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
deploy/postfix/sqlite-sender-bcc.cf
-2
View File
@@ -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'