diff --git a/server/consts/consts.go b/server/consts/consts.go index eb23722..c6f3d51 100644 --- a/server/consts/consts.go +++ b/server/consts/consts.go @@ -23,4 +23,7 @@ const ( // EmailStatusJunk 骚扰邮件 EmailStatusJunk int8 = 5 + + // EmailStatusDeliveryPending marks an outbound audit whose delivery result is not final. + EmailStatusDeliveryPending int8 = 6 ) diff --git a/server/db/init.go b/server/db/init.go index 832659e..77ed4da 100644 --- a/server/db/init.go +++ b/server/db/init.go @@ -2,6 +2,8 @@ package db import ( "fmt" + "time" + "github.com/Jinnrry/pmail/config" "github.com/Jinnrry/pmail/models" "github.com/Jinnrry/pmail/utils/context" @@ -9,8 +11,6 @@ import ( _ "github.com/go-sql-driver/mysql" _ "github.com/lib/pq" log "github.com/sirupsen/logrus" - _ "modernc.org/sqlite" - "time" "xorm.io/xorm" ) @@ -19,20 +19,20 @@ var Instance *xorm.Engine func Init(version string) error { dsn := config.Instance.DbDSN var err error + if config.Instance.DbType == config.DBTypeSQLite { + dsn, err = sqliteDSNWithReliabilityOptions(dsn) + if err != nil { + return errors.Wrap(err) + } + } switch config.Instance.DbType { case "mysql": Instance, err = xorm.NewEngine("mysql", dsn) - Instance.SetMaxOpenConns(100) - Instance.SetMaxIdleConns(10) case "sqlite": Instance, err = xorm.NewEngine("sqlite", dsn) - Instance.SetMaxOpenConns(1) - Instance.SetMaxIdleConns(1) case "postgres": Instance, err = xorm.NewEngine("postgres", dsn) - Instance.SetMaxOpenConns(100) - Instance.SetMaxIdleConns(10) default: return errors.New("Database Type Error!") } @@ -40,6 +40,13 @@ func Init(version string) error { log.Errorf("DB init Error! %s", err.Error()) return errors.Wrap(err) } + if config.Instance.DbType == config.DBTypeSQLite { + Instance.SetMaxOpenConns(1) + Instance.SetMaxIdleConns(1) + } else { + Instance.SetMaxOpenConns(100) + Instance.SetMaxIdleConns(10) + } Instance.SetConnMaxLifetime(30 * time.Minute) Instance.ShowSQL(config.Instance.LogLevel == "debug") diff --git a/server/db/retry.go b/server/db/retry.go new file mode 100644 index 0000000..ad7722e --- /dev/null +++ b/server/db/retry.go @@ -0,0 +1,174 @@ +package db + +import ( + stderrors "errors" + "fmt" + "net/url" + "strconv" + "strings" + "time" + "unicode" + + log "github.com/sirupsen/logrus" + "modernc.org/sqlite" + "xorm.io/xorm" +) + +const ( + sqliteBusyTimeoutMilliseconds = 1000 + transactionAttempts = 3 + transactionInitialBackoff = 50 * time.Millisecond +) + +// Transaction runs fn atomically. SQLite BUSY and LOCKED failures retry the +// entire transaction, so fn must contain only work that is safe to repeat. +func Transaction(engine *xorm.Engine, fn func(*xorm.Session) error) error { + return transactionWithRetry(engine, transactionAttempts, transactionInitialBackoff, fn) +} + +func transactionWithRetry(engine *xorm.Engine, attempts int, initialBackoff time.Duration, fn func(*xorm.Session) error) error { + if attempts < 1 { + attempts = 1 + } + + var err error + for attempt := 1; attempt <= attempts; attempt++ { + var retrySafe bool + err, retrySafe = runTransactionAttempt(engine, fn) + if err == nil { + return nil + } + if !retrySafe || engine.DriverName() != "sqlite" || !IsSQLiteBusyOrLocked(err) || attempt == attempts { + return err + } + time.Sleep(initialBackoff << (attempt - 1)) + } + return err +} + +func runTransaction(engine *xorm.Engine, fn func(*xorm.Session) error) error { + err, _ := runTransactionAttempt(engine, fn) + return err +} + +func runTransactionAttempt(engine *xorm.Engine, fn func(*xorm.Session) error) (error, bool) { + session := engine.NewSession() + defer session.Close() + + if err := session.Begin(); err != nil { + return fmt.Errorf("begin transaction: %w", err), true + } + if err := fn(session); err != nil { + if rollbackErr := session.Rollback(); rollbackErr != nil { + return stderrors.Join(err, fmt.Errorf("rollback transaction: %w", rollbackErr)), false + } + return err, true + } + if err := session.Commit(); err != nil { + // modernc.org/sqlite v1.46.1 and newer roll back internally when COMMIT + // fails, leaving the connection clean and the transaction safe to retry. + return fmt.Errorf("commit transaction: %w", err), true + } + return nil, false +} + +// IsSQLiteBusyOrLocked classifies SQLite errors by result code, including +// extended BUSY and LOCKED codes. Error message text is intentionally ignored. +func IsSQLiteBusyOrLocked(err error) bool { + var sqliteErr *sqlite.Error + if !stderrors.As(err, &sqliteErr) { + return false + } + primaryCode := sqliteErr.Code() & 0xff + return primaryCode == 5 || primaryCode == 6 +} + +func sqliteDSNWithReliabilityOptions(dsn string) (string, error) { + dsnPrefix, rawQuery, hasQuery := strings.Cut(dsn, "?") + query, err := url.ParseQuery(rawQuery) + if err != nil { + return "", fmt.Errorf("parse SQLite DSN: %w", err) + } + + pragmas := query["_pragma"] + timeoutIndex := -1 + for index, pragma := range pragmas { + if sqlitePragmaName(pragma) == "busy_timeout" { + if timeoutIndex >= 0 { + return "", fmt.Errorf("parse SQLite DSN: multiple busy_timeout pragmas are ambiguous") + } + timeoutIndex = index + } + } + if timeoutIndex >= 0 { + requested, err := sqliteBusyTimeoutValue(pragmas[timeoutIndex]) + if err != nil { + return "", fmt.Errorf("parse SQLite busy_timeout pragma %q: %w", pragmas[timeoutIndex], err) + } + if requested <= sqliteBusyTimeoutMilliseconds { + return dsn, nil + } + + log.Warnf( + "SQLite busy_timeout requested %dms exceeds the %dms maximum; using %dms", + requested, + sqliteBusyTimeoutMilliseconds, + sqliteBusyTimeoutMilliseconds, + ) + pragmas[timeoutIndex] = fmt.Sprintf("busy_timeout(%d)", sqliteBusyTimeoutMilliseconds) + query["_pragma"] = pragmas + return dsnPrefix + "?" + query.Encode(), nil + } + + separator := "?" + if hasQuery { + separator = "&" + if strings.HasSuffix(dsn, "?") || strings.HasSuffix(dsn, "&") { + separator = "" + } + } + option := url.Values{"_pragma": {fmt.Sprintf("busy_timeout(%d)", sqliteBusyTimeoutMilliseconds)}}.Encode() + return dsn + separator + option, nil +} + +func sqliteBusyTimeoutValue(pragma string) (int64, error) { + const name = "busy_timeout" + + normalized := strings.TrimSpace(pragma) + if len(normalized) < len(name) || !strings.EqualFold(normalized[:len(name)], name) { + return 0, fmt.Errorf("not a busy_timeout pragma") + } + remainder := strings.TrimSpace(normalized[len(name):]) + + var valueText string + switch { + case strings.HasPrefix(remainder, "("): + if !strings.HasSuffix(remainder, ")") { + return 0, fmt.Errorf("expected busy_timeout()") + } + valueText = strings.TrimSpace(remainder[1 : len(remainder)-1]) + case strings.HasPrefix(remainder, "="): + valueText = strings.TrimSpace(remainder[1:]) + default: + return 0, fmt.Errorf("expected busy_timeout() or busy_timeout = ") + } + if valueText == "" { + return 0, fmt.Errorf("timeout value is empty") + } + + timeout, err := strconv.ParseInt(valueText, 10, 64) + if err != nil { + return 0, fmt.Errorf("invalid timeout value %q: %w", valueText, err) + } + return timeout, nil +} + +func sqlitePragmaName(pragma string) string { + normalized := strings.ToLower(strings.TrimSpace(pragma)) + if end := strings.IndexFunc(normalized, func(character rune) bool { + return character == '(' || character == '=' || unicode.IsSpace(character) + }); end >= 0 { + normalized = normalized[:end] + } + return normalized +} diff --git a/server/db/retry_test.go b/server/db/retry_test.go new file mode 100644 index 0000000..f88f3ce --- /dev/null +++ b/server/db/retry_test.go @@ -0,0 +1,407 @@ +package db + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "path/filepath" + "strings" + "testing" + "time" + + log "github.com/sirupsen/logrus" + "modernc.org/sqlite" + "xorm.io/xorm" +) + +func TestSQLiteDSNOptions(t *testing.T) { + tests := []struct { + name string + dsn string + wantDSN string + wantBusyTimeout string + wantJournalMode string + wantTxLock string + }{ + { + name: "plain filename gets default timeout", + dsn: "/tmp/pmail.db", + wantBusyTimeout: "busy_timeout(1000)", + }, + { + name: "timeout at cap is preserved byte for byte", + dsn: "file:pmail.db?_txlock=deferred&_pragma=BUSY_TIMEOUT%20%3D%201000&mode=memory", + wantDSN: "file:pmail.db?_txlock=deferred&_pragma=BUSY_TIMEOUT%20%3D%201000&mode=memory", + wantBusyTimeout: "BUSY_TIMEOUT = 1000", + wantTxLock: "deferred", + }, + { + name: "explicit timeout is capped and other options are preserved", + dsn: "file:pmail.db?mode=memory&_pragma=journal_mode(WAL)&_pragma=busy_timeout%20%3D%20600000&_txlock=deferred", + wantBusyTimeout: "busy_timeout(1000)", + wantJournalMode: "journal_mode(WAL)", + wantTxLock: "deferred", + }, + { + name: "negative timeout is preserved byte for byte", + dsn: "file:pmail.db?cache=shared&_pragma=%20busy_timeout%20%28%20-1%20%29%20", + wantDSN: "file:pmail.db?cache=shared&_pragma=%20busy_timeout%20%28%20-1%20%29%20", + wantBusyTimeout: " busy_timeout ( -1 ) ", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := sqliteDSNWithReliabilityOptions(tt.dsn) + if err != nil { + t.Fatalf("sqliteDSNWithReliabilityOptions() error = %v", err) + } + if tt.wantDSN != "" && got != tt.wantDSN { + t.Errorf("sqliteDSNWithReliabilityOptions() = %q, want byte-for-byte %q", got, tt.wantDSN) + } + + queryText := "" + if _, after, ok := strings.Cut(got, "?"); ok { + queryText = after + } + query, err := url.ParseQuery(queryText) + if err != nil { + t.Fatalf("ParseQuery(%q): %v", queryText, err) + } + if got := query.Get("_txlock"); got != tt.wantTxLock { + t.Errorf("_txlock = %q, want %q", got, tt.wantTxLock) + } + if !containsString(query["_pragma"], tt.wantBusyTimeout) { + t.Errorf("_pragma = %q, want %q", query["_pragma"], tt.wantBusyTimeout) + } + if got := countPragma(query["_pragma"], "busy_timeout"); got != 1 { + t.Errorf("busy_timeout pragma count = %d, want 1: %q", got, query["_pragma"]) + } + if tt.wantJournalMode != "" && !containsString(query["_pragma"], tt.wantJournalMode) { + t.Errorf("_pragma = %q, want preserved %q", query["_pragma"], tt.wantJournalMode) + } + }) + } +} + +func TestSQLiteDSNRejectsInvalidBusyTimeout(t *testing.T) { + tests := []struct { + name string + dsn string + }{ + { + name: "duplicate timeout pragmas", + dsn: "file:pmail.db?_pragma=busy_timeout(10)&_pragma=BUSY_TIMEOUT%20%3D%2020", + }, + { + name: "missing timeout value", + dsn: "file:pmail.db?_pragma=busy_timeout", + }, + { + name: "nonnumeric timeout", + dsn: "file:pmail.db?_pragma=busy_timeout(forever)", + }, + { + name: "timeout with trailing junk", + dsn: "file:pmail.db?_pragma=busy_timeout(10)%20milliseconds", + }, + { + name: "overflowing timeout", + dsn: "file:pmail.db?_pragma=busy_timeout(9223372036854775808)", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := sqliteDSNWithReliabilityOptions(tt.dsn); err == nil { + t.Fatalf("sqliteDSNWithReliabilityOptions(%q) error = nil, want error", tt.dsn) + } + }) + } +} + +func TestSQLiteDSNWarnsWhenBusyTimeoutIsCapped(t *testing.T) { + logger := log.StandardLogger() + originalOutput := logger.Out + originalLevel := logger.GetLevel() + var output bytes.Buffer + logger.SetOutput(&output) + logger.SetLevel(log.WarnLevel) + t.Cleanup(func() { + logger.SetOutput(originalOutput) + logger.SetLevel(originalLevel) + }) + + if _, err := sqliteDSNWithReliabilityOptions("file:pmail.db?_pragma=busy_timeout(600000)"); err != nil { + t.Fatalf("sqliteDSNWithReliabilityOptions() error = %v", err) + } + for _, want := range []string{"600000ms", "1000ms"} { + if !strings.Contains(output.String(), want) { + t.Errorf("warning = %q, want it to contain %q", output.String(), want) + } + } +} + +func TestSQLiteBusyTimeoutAppliesToEveryConnection(t *testing.T) { + dsn, err := sqliteDSNWithReliabilityOptions(filepath.Join(t.TempDir(), "connections.db")) + if err != nil { + t.Fatal(err) + } + database, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + defer database.Close() + database.SetMaxOpenConns(2) + + first, err := database.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := database.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer second.Close() + + for index, connection := range []*sql.Conn{first, second} { + var timeout int + if err = connection.QueryRowContext(context.Background(), "PRAGMA busy_timeout").Scan(&timeout); err != nil { + t.Fatalf("connection %d: %v", index+1, err) + } + if timeout != sqliteBusyTimeoutMilliseconds { + t.Errorf("connection %d busy_timeout = %d, want %d", index+1, timeout, sqliteBusyTimeoutMilliseconds) + } + } +} + +func TestIsSQLiteBusyOrLockedUsesDriverCode(t *testing.T) { + engine, err := xorm.NewEngine("sqlite", filepath.Join(t.TempDir(), "busy.db")+"?_pragma=busy_timeout(1)") + if err != nil { + t.Fatal(err) + } + defer engine.Close() + if _, err = engine.Exec("CREATE TABLE audit (id INTEGER PRIMARY KEY, value TEXT)"); err != nil { + t.Fatal(err) + } + + busyErr := sqliteBusyError(t, engine.DataSourceName()) + var sqliteErr *sqlite.Error + if !errors.As(busyErr, &sqliteErr) { + t.Fatalf("error type = %T, want *sqlite.Error", busyErr) + } + if !IsSQLiteBusyOrLocked(fmt.Errorf("wrapped: %w", busyErr)) { + t.Fatalf("IsSQLiteBusyOrLocked(%v) = false", busyErr) + } + if IsSQLiteBusyOrLocked(errors.New("database is locked")) { + t.Fatal("plain error text must not be classified as SQLITE_BUSY") + } +} + +func TestTransactionRetriesWholeSQLiteTransaction(t *testing.T) { + engine, err := xorm.NewEngine("sqlite", filepath.Join(t.TempDir(), "retry.db")+"?_pragma=busy_timeout(10)") + if err != nil { + t.Fatal(err) + } + defer engine.Close() + engine.SetMaxOpenConns(1) + if _, err = engine.Exec("CREATE TABLE audit (id INTEGER PRIMARY KEY, value TEXT)"); err != nil { + t.Fatal(err) + } + busyErr := sqliteBusyError(t, engine.DataSourceName()) + + attempts := 0 + err = transactionWithRetry(engine, 3, time.Millisecond, func(session *xorm.Session) error { + attempts++ + if attempts == 1 { + return busyErr + } + _, insertErr := session.Exec("INSERT INTO audit(value) VALUES ('saved')") + return insertErr + }) + if err != nil { + t.Fatalf("transactionWithRetry() error = %v", err) + } + if attempts != 2 { + t.Fatalf("attempts = %d, want 2", attempts) + } + assertAuditRowCount(t, engine, 1) +} + +func TestTransactionRetriesSQLiteBusyCommit(t *testing.T) { + engine, blocker, readTx := newCommitBusyTest(t, 10) + + attempts := 0 + err := transactionWithRetry(engine, 2, 5*time.Millisecond, func(session *xorm.Session) error { + attempts++ + if attempts == 2 { + if rollbackErr := readTx.Rollback(); rollbackErr != nil { + return fmt.Errorf("release read transaction: %w", rollbackErr) + } + } + _, insertErr := session.Exec("INSERT INTO audit(value) VALUES ('saved')") + return insertErr + }) + _ = blocker.Close() + if err != nil { + t.Fatalf("transactionWithRetry() error = %v", err) + } + if attempts != 2 { + t.Fatalf("attempts = %d, want 2", attempts) + } + assertAuditRowCount(t, engine, 2) +} + +func TestPermanentSQLiteBusyCommitLeavesConnectionUsable(t *testing.T) { + engine, blocker, readTx := newCommitBusyTest(t, 5) + + attempts := 0 + err := transactionWithRetry(engine, 2, time.Millisecond, func(session *xorm.Session) error { + attempts++ + _, insertErr := session.Exec("INSERT INTO audit(value) VALUES ('rolled back')") + return insertErr + }) + if err == nil || !IsSQLiteBusyOrLocked(err) { + t.Fatalf("transactionWithRetry() error = %v, want SQLITE_BUSY", err) + } + if attempts != 2 { + t.Fatalf("attempts = %d, want 2", attempts) + } + if err = readTx.Rollback(); err != nil { + t.Fatal(err) + } + _ = blocker.Close() + + err = runTransaction(engine, func(session *xorm.Session) error { + _, insertErr := session.Exec("INSERT INTO audit(value) VALUES ('after busy')") + return insertErr + }) + if err != nil { + t.Fatalf("transaction after BUSY = %v", err) + } + assertAuditRowCount(t, engine, 2) +} + +func TestTransactionDoesNotClassifyBusyText(t *testing.T) { + engine, err := xorm.NewEngine("sqlite", filepath.Join(t.TempDir(), "text.db")) + if err != nil { + t.Fatal(err) + } + defer engine.Close() + + attempts := 0 + wantErr := errors.New("database is locked") + err = transactionWithRetry(engine, 3, time.Millisecond, func(*xorm.Session) error { + attempts++ + return wantErr + }) + if !errors.Is(err, wantErr) { + t.Fatalf("transactionWithRetry() error = %v, want %v", err, wantErr) + } + if attempts != 1 { + t.Fatalf("attempts = %d, want 1", attempts) + } +} + +func newCommitBusyTest(t *testing.T, timeoutMilliseconds int) (*xorm.Engine, *sql.DB, *sql.Tx) { + t.Helper() + path := filepath.Join(t.TempDir(), "commit-busy.db") + dsn := fmt.Sprintf("%s?_pragma=busy_timeout(%d)", path, timeoutMilliseconds) + engine, err := xorm.NewEngine("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = engine.Close() }) + engine.SetMaxOpenConns(1) + engine.SetMaxIdleConns(1) + if _, err = engine.Exec("CREATE TABLE audit (id INTEGER PRIMARY KEY, value TEXT)"); err != nil { + t.Fatal(err) + } + if _, err = engine.Exec("INSERT INTO audit(value) VALUES ('seed')"); err != nil { + t.Fatal(err) + } + + blocker, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + readTx, err := blocker.Begin() + if err != nil { + _ = blocker.Close() + t.Fatal(err) + } + var count int + if err = readTx.QueryRow("SELECT count(*) FROM audit").Scan(&count); err != nil { + _ = readTx.Rollback() + _ = blocker.Close() + t.Fatal(err) + } + t.Cleanup(func() { + _ = readTx.Rollback() + _ = blocker.Close() + }) + return engine, blocker, readTx +} + +func sqliteBusyError(t *testing.T, dsn string) error { + t.Helper() + first, err := xorm.NewEngine("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := xorm.NewEngine("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + defer second.Close() + + tx := first.NewSession() + defer tx.Close() + if err = tx.Begin(); err != nil { + t.Fatal(err) + } + if _, err = tx.Exec("INSERT INTO audit(value) VALUES ('rolled back')"); err != nil { + t.Fatal(err) + } + _, busyErr := second.Exec("INSERT INTO audit(value) VALUES ('blocked')") + _ = tx.Rollback() + if busyErr == nil { + t.Fatal("expected SQLITE_BUSY") + } + return busyErr +} + +func assertAuditRowCount(t *testing.T, engine *xorm.Engine, want int) { + t.Helper() + var count int + if _, err := engine.SQL("SELECT count(*) FROM audit").Get(&count); err != nil { + t.Fatal(err) + } + if count != want { + t.Fatalf("audit rows = %d, want %d", count, want) + } +} + +func containsString(values []string, want string) bool { + for _, value := range values { + if value == want { + return true + } + } + return false +} + +func countPragma(values []string, name string) int { + count := 0 + for _, value := range values { + if sqlitePragmaName(value) == name { + count++ + } + } + return count +} diff --git a/server/go.mod b/server/go.mod index 78f240a..4a873d6 100644 --- a/server/go.mod +++ b/server/go.mod @@ -21,9 +21,9 @@ require ( github.com/mileusna/spf v0.9.5 github.com/sirupsen/logrus v1.9.3 github.com/spf13/cast v1.7.1 - golang.org/x/crypto v0.37.0 - golang.org/x/text v0.24.0 - modernc.org/sqlite v1.37.0 + golang.org/x/crypto v0.43.0 + golang.org/x/text v0.30.0 + modernc.org/sqlite v1.46.1 xorm.io/builder v0.3.13 xorm.io/xorm v1.3.9 ) @@ -43,20 +43,20 @@ require ( github.com/miekg/dns v1.1.65 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect - github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect github.com/nxadm/tail v1.4.11 // indirect github.com/onsi/ginkgo v1.16.4 // indirect github.com/onsi/gomega v1.31.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/rogpeppe/go-internal v1.12.0 // indirect github.com/syndtr/goleveldb v1.0.0 // indirect - golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 // indirect - golang.org/x/mod v0.24.0 // indirect - golang.org/x/net v0.39.0 // indirect - golang.org/x/sync v0.13.0 // indirect - golang.org/x/sys v0.32.0 // indirect - golang.org/x/tools v0.32.0 // indirect - modernc.org/libc v1.63.0 // indirect + golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect + golang.org/x/mod v0.29.0 // indirect + golang.org/x/net v0.46.0 // indirect + golang.org/x/sync v0.17.0 // indirect + golang.org/x/sys v0.37.0 // indirect + golang.org/x/tools v0.38.0 // indirect + modernc.org/libc v1.67.6 // indirect modernc.org/mathutil v1.7.1 // indirect - modernc.org/memory v1.10.0 // indirect + modernc.org/memory v1.11.0 // indirect ) diff --git a/server/go.sum b/server/go.sum index 9d04eae..289a9b2 100644 --- a/server/go.sum +++ b/server/go.sum @@ -74,6 +74,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/css v1.0.1 h1:ntNaBIghp6JmvWnxbZKANoLyuXTPZ4cAMlo6RyhlbO8= github.com/gorilla/css v1.0.1/go.mod h1:BvnYkspnSzMmwRK+b8/xgNPLiIuNZr6vbZBTPQ2A3b0= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= @@ -101,8 +103,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= -github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= -github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU= github.com/nxadm/tail v1.4.11 h1:8feyoE3OzPrcshW5/MJ4sGESc5cqmGkGCWlco4l0bqY= @@ -143,16 +145,16 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= -golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= -golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 h1:R84qjqJb5nVJMxqWYb3np9L5ZsaDtB+a39EqjV0JSUM= -golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0/go.mod h1:S9Xr4PYopiDyqSyp5NjCrhFrqg6A5zA2E/iPHPhqnS8= +golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= +golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= +golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2wtOnGAHZWCHUM4KGzY= +golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= -golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU= -golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww= +golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= +golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -163,16 +165,16 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY= -golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E= +golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4= +golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= -golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -194,8 +196,8 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20= -golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= +golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= @@ -205,16 +207,16 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= -golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0= -golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU= +golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= +golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= -golang.org/x/tools v0.32.0 h1:Q7N1vhpkQv7ybVzLFtTjvQya2ewbwNDZzUgfXGqtMWU= -golang.org/x/tools v0.32.0/go.mod h1:ZxrU41P/wAbZD8EDa6dDCa6XfpkhJ7HFMjHJXfBDu8s= +golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ= +golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -236,26 +238,30 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -modernc.org/cc/v4 v4.26.0 h1:QMYvbVduUGH0rrO+5mqF/PSPPRZNpRtg2CLELy7vUpA= -modernc.org/cc/v4 v4.26.0/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= -modernc.org/ccgo/v4 v4.26.0 h1:gVzXaDzGeBYJ2uXTOpR8FR7OlksDOe9jxnjhIKCsiTc= -modernc.org/ccgo/v4 v4.26.0/go.mod h1:Sem8f7TFUtVXkG2fiaChQtyyfkqhJBg/zjEJBkmuAVY= -modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE= -modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ= +modernc.org/cc/v4 v4.27.1 h1:9W30zRlYrefrDV2JE2O8VDtJ1yPGownxciz5rrbQZis= +modernc.org/cc/v4 v4.27.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= +modernc.org/ccgo/v4 v4.30.1 h1:4r4U1J6Fhj98NKfSjnPUN7Ze2c6MnAdL0hWw6+LrJpc= +modernc.org/ccgo/v4 v4.30.1/go.mod h1:bIOeI1JL54Utlxn+LwrFyjCx2n2RDiYEaJVSrgdrRfM= +modernc.org/fileutil v1.3.40 h1:ZGMswMNc9JOCrcrakF1HrvmergNLAmxOPjizirpfqBA= +modernc.org/fileutil v1.3.40/go.mod h1:HxmghZSZVAz/LXcMNwZPA/DRrQZEVP9VX0V4LQGQFOc= modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= -modernc.org/libc v1.63.0 h1:wKzb61wOGCzgahQBORb1b0dZonh8Ufzl/7r4Yf1D5YA= -modernc.org/libc v1.63.0/go.mod h1:wDzH1mgz1wUIEwottFt++POjGRO9sgyQKrpXaz3x89E= +modernc.org/gc/v3 v3.1.1 h1:k8T3gkXWY9sEiytKhcgyiZ2L0DTyCQ/nvX+LoCljoRE= +modernc.org/gc/v3 v3.1.1/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.67.6 h1:eVOQvpModVLKOdT+LvBPjdQqfrZq+pC39BygcT+E7OI= +modernc.org/libc v1.67.6/go.mod h1:JAhxUVlolfYDErnwiqaLvUqc8nfb2r6S6slAgZOnaiE= modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= -modernc.org/memory v1.10.0 h1:fzumd51yQ1DxcOxSO+S6X7+QTuVU+n8/Aj7swYjFfC4= -modernc.org/memory v1.10.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8= modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= -modernc.org/sqlite v1.37.0 h1:s1TMe7T3Q3ovQiK2Ouz4Jwh7dw4ZDqbebSDTlSJdfjI= -modernc.org/sqlite v1.37.0/go.mod h1:5YiWv+YviqGMuGw4V+PNplcyaJ5v+vQd7TQOgkACoJM= +modernc.org/sqlite v1.46.1 h1:eFJ2ShBLIEnUWlLy12raN0Z1plqmFX9Qe3rjQTKt6sU= +modernc.org/sqlite v1.46.1/go.mod h1:CzbrU2lSB1DKUusvwGz7rqEKIq+NUd8GWuBBZDs9/nA= modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= diff --git a/server/listen/smtp_server/audit_persistence.go b/server/listen/smtp_server/audit_persistence.go new file mode 100644 index 0000000..cc300e7 --- /dev/null +++ b/server/listen/smtp_server/audit_persistence.go @@ -0,0 +1,215 @@ +package smtp_server + +import ( + "database/sql" + "encoding/json" + stderrors "errors" + "fmt" + "strings" + + "github.com/Jinnrry/pmail/config" + "github.com/Jinnrry/pmail/consts" + "github.com/Jinnrry/pmail/db" + "github.com/Jinnrry/pmail/dto/parsemail" + "github.com/Jinnrry/pmail/hooks" + "github.com/Jinnrry/pmail/hooks/framework" + "github.com/Jinnrry/pmail/models" + "github.com/Jinnrry/pmail/utils/array" + "github.com/Jinnrry/pmail/utils/async" + "github.com/Jinnrry/pmail/utils/context" + smtp "github.com/emersion/go-smtp" + log "github.com/sirupsen/logrus" + "xorm.io/builder" + "xorm.io/xorm" +) + +const deliveryPendingError = "delivery result pending; outcome unknown after interruption" + +type outgoingSender func(*context.Context, *parsemail.Email) (error, map[string]error) + +func (s *Session) deliverOutgoingEmail(email *parsemail.Email, sender outgoingSender) error { + ctx := s.Ctx + _, saved, err := saveEmail(ctx, email.Size, email, ctx.UserID, int(consts.EmailTypeSend), nil, true, true) + if err != nil { + log.WithContext(ctx).Errorf("CRITICAL audit_event=initial_persist_failed direction=outbound msg_id=%q error=%v", email.MsgID, err) + return localPersistenceSMTPError() + } + + deliveryErr, domainErrors := sender(ctx, email) + runSendAfterHooks(ctx, email, domainErrors) + + targetStatus := consts.EmailStatusSent + var errorText any + if deliveryErr != nil { + targetStatus = consts.EmailStatusFail + errorText = deliveryErr.Error() + } + + if err = persistDeliveryResult(ctx, saved.Id, targetStatus, errorText); err != nil { + log.WithContext(ctx).Errorf( + "CRITICAL audit_event=delivery_result_persist_failed email_id=%d msg_id=%q target_status=%d downstream_accepted=%t error=%v", + saved.Id, saved.MsgID, targetStatus, deliveryErr == nil, err, + ) + } + + // If the remote MX accepted DATA, returning a local 451 here would cause a + // duplicate delivery. The persistence failure is retained as pending state + // and a CRITICAL log; the SMTP result must still reflect the remote result. + return downstreamDeliverySMTPError(deliveryErr, domainErrors) +} + +func runSendAfterHooks(ctx *context.Context, email *parsemail.Email, domainErrors map[string]error) { + process := async.New(ctx) + for _, hook := range hooks.HookList { + if hook == nil { + continue + } + process.WaitProcess(func(value any) { + value.(framework.EmailHook).SendAfter(ctx, email, domainErrors) + }, hook) + } + process.Wait() +} + +func persistDeliveryResult(ctx *context.Context, emailID int, status int8, errorText any) error { + return db.Transaction(db.Instance, func(session *xorm.Session) error { + result, err := session.Exec(db.WithContext(ctx, `update email + set status=?, error=?, update_time=CURRENT_TIMESTAMP + where id=?`), status, errorText, emailID) + if err != nil { + return err + } + if err = requireOneAffected(result, "update email delivery result"); err != nil { + return err + } + + result, err = session.Exec(db.WithContext(ctx, `update user_email + set status=? where email_id=?`), status, emailID) + if err != nil { + return err + } + return requireOneAffected(result, "update user_email delivery result") + }) +} + +func requireOneAffected(result sql.Result, operation string) error { + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("%s: read affected rows: %w", operation, err) + } + if affected != 1 { + return fmt.Errorf("%s: affected %d rows, want 1", operation, affected) + } + return nil +} + +func localPersistenceSMTPError() error { + return &smtp.SMTPError{ + Code: 451, + EnhancedCode: smtp.EnhancedCode{4, 3, 0}, + Message: "Temporary local persistence failure", + } +} + +func persistInitialAudit(ctx *context.Context, modelEmail models.Email, userIDs []int) (*models.Email, error) { + var saved models.Email + err := db.Transaction(db.Instance, func(session *xorm.Session) error { + candidate := modelEmail + candidate.Id = 0 + if _, err := session.Insert(&candidate); err != nil { + return err + } + if candidate.Id <= 0 { + return stderrors.New("email insert returned no id") + } + for _, userID := range userIDs { + relation := models.UserEmail{EmailID: candidate.Id, UserID: userID, Status: candidate.Status} + if _, err := session.Insert(&relation); err != nil { + return err + } + } + saved = candidate + return nil + }) + if err != nil { + return nil, err + } + return &saved, nil +} + +func resolveIncomingUsers(ctx *context.Context, email *parsemail.Email, emailType int, reallyTo []string, SPFStatus, dkimStatus bool) ([]*models.User, bool, error) { + if emailType != int(consts.EmailTypeReceive) { + return nil, false, nil + } + + accounts := incomingAccounts(email, reallyTo) + var users []*models.User + if len(accounts) > 0 { + where, params, err := builder.ToSQL(builder.In("LOWER(account)", accounts)) + if err != nil { + return nil, false, err + } + if err = db.Instance.Table(&models.User{}).Where(where, params...).Find(&users); err != nil { + return nil, false, err + } + } + if len(users) > 0 { + return users, false, nil + } + + if DropUnknownRecipientEmails && + ((config.Instance.SpamFilterLevel == 1 && !SPFStatus && !dkimStatus) || + (config.Instance.SpamFilterLevel == 2 && !SPFStatus) || + (config.Instance.SpamFilterLevel == 3 && !dkimStatus)) { + log.WithContext(ctx).Infoln("垃圾邮件，拒信") + log.WithContext(ctx).Infof("收件人不存在且DKIM验证失败，丢弃邮件: %s -> %v", email.From.EmailAddress, accounts) + return nil, true, nil + } + + log.WithContext(ctx).Infof("收件人不存在但DKIM验证通过，转交管理员: %s -> %v", email.From.EmailAddress, accounts) + if err := db.Instance.Table(&models.User{}).Where("is_admin=1").Find(&users); err != nil { + return nil, false, err + } + if len(users) == 0 { + return nil, false, stderrors.New("no administrator available for unknown recipient") + } + return users, false, nil +} + +func incomingAccounts(email *parsemail.Email, reallyTo []string) []string { + var accounts []string + if len(reallyTo) > 0 { + for _, recipient := range reallyTo { + account := parsemail.BuilderUser(recipient) + if account == nil { + continue + } + name, domain := account.GetDomainAccount() + if array.InArray(domain, config.Instance.Domains) && name != "" { + accounts = append(accounts, strings.ToLower(name)) + } + } + return accounts + } + for _, recipients := range [][]*parsemail.User{email.To, email.Cc, email.Bcc} { + for _, user := range recipients { + account, _ := user.GetDomainAccount() + if account != "" { + accounts = append(accounts, strings.ToLower(account)) + } + } + } + return accounts +} + +func boolInt8(value bool) int8 { + if value { + return 1 + } + return 0 +} + +func json2string(value any) string { + data, _ := json.Marshal(value) + return string(data) +} diff --git a/server/listen/smtp_server/audit_persistence_test.go b/server/listen/smtp_server/audit_persistence_test.go new file mode 100644 index 0000000..940c83f --- /dev/null +++ b/server/listen/smtp_server/audit_persistence_test.go @@ -0,0 +1,562 @@ +package smtp_server + +import ( + "bytes" + "database/sql" + "errors" + "fmt" + "net" + "net/textproto" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Jinnrry/pmail/config" + "github.com/Jinnrry/pmail/consts" + "github.com/Jinnrry/pmail/db" + "github.com/Jinnrry/pmail/dto" + "github.com/Jinnrry/pmail/dto/parsemail" + "github.com/Jinnrry/pmail/hooks" + "github.com/Jinnrry/pmail/hooks/framework" + "github.com/Jinnrry/pmail/models" + listservice "github.com/Jinnrry/pmail/services/list" + "github.com/Jinnrry/pmail/utils/context" + smtp "github.com/emersion/go-smtp" + log "github.com/sirupsen/logrus" + _ "modernc.org/sqlite" + "xorm.io/xorm" +) + +func TestOutgoingAuditInsertFailureDoesNotDeliver(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + blocker := beginAuditReadTransaction(t, dsn) + + var deliveries atomic.Int32 + session := newAuditTestSession() + err := session.deliverOutgoingEmail(newAuditTestEmail("insert-busy@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + return nil, nil + }) + + assertSMTPCode(t, err, 451) + if got := deliveries.Load(); got != 0 { + t.Fatalf("downstream deliveries = %d, want 0", got) + } + assertAuditCounts(t, engine, 0, 0) + + if err := blocker.Rollback(); err != nil { + t.Fatal(err) + } + if err := session.deliverOutgoingEmail(newAuditTestEmail("insert-busy@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + return nil, nil + }); err != nil { + t.Fatalf("delivery after lock release failed: %v", err) + } + if got := deliveries.Load(); got != 1 { + t.Fatalf("downstream deliveries after recovery = %d, want 1", got) + } + assertAuditCounts(t, engine, 1, 1) + assertAuditStatuses(t, engine, consts.EmailStatusSent) +} + +func TestOutgoingAuditRetriesBusyInitialInsertBeforeDeliver(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + blocker := beginAuditReadTransaction(t, dsn) + released := make(chan struct{}) + go func() { + time.Sleep(35 * time.Millisecond) + _ = blocker.Rollback() + close(released) + }() + + var deliveries atomic.Int32 + err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("initial-retry@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + return nil, nil + }) + <-released + if err != nil { + t.Fatalf("deliverOutgoingEmail() error = %v", err) + } + if got := deliveries.Load(); got != 1 { + t.Fatalf("downstream deliveries = %d, want 1", got) + } + assertAuditCounts(t, engine, 1, 1) + assertAuditStatuses(t, engine, consts.EmailStatusSent) +} + +func TestOutgoingAuditRelationInsertRollsBackEmail(t *testing.T) { + engine, _ := newAuditTestEngine(t) + if _, err := engine.Exec(`CREATE TRIGGER fail_user_email_insert + BEFORE INSERT ON user_email + BEGIN + SELECT RAISE(ABORT, 'forced user_email insert failure'); + END`); err != nil { + t.Fatal(err) + } + + email := newAuditTestEmail("relation-failure@example.com") + _, _, err := saveEmail(newAuditTestSession().Ctx, email.Size, email, 1, int(consts.EmailTypeSend), nil, true, true) + if err == nil { + t.Fatal("saveEmail() error = nil, want relation insert failure") + } + if email.MessageId != 0 { + t.Fatalf("email.MessageId = %d, want 0 after rollback", email.MessageId) + } + assertAuditCounts(t, engine, 0, 0) +} + +func TestIncomingAuditRelationInsertRollsBackEmail(t *testing.T) { + engine, _ := newAuditTestEngine(t) + recipient := &models.User{Account: "recipient", Name: "Recipient"} + if _, err := engine.Insert(recipient); err != nil { + t.Fatal(err) + } + if _, err := engine.Exec(`CREATE TRIGGER fail_incoming_user_email_insert + BEFORE INSERT ON user_email + BEGIN + SELECT RAISE(ABORT, 'forced incoming user_email insert failure'); + END`); err != nil { + t.Fatal(err) + } + + from := &parsemail.User{Name: "External", EmailAddress: "external@example.net"} + email := &parsemail.Email{ + From: from, + Sender: from, + To: []*parsemail.User{{EmailAddress: "recipient@example.com"}}, + Text: []byte("body"), + Size: 4, + MsgID: "incoming-relation-failure@example.net", + } + _, _, err := saveEmail(&context.Context{}, email.Size, email, 0, int(consts.EmailTypeReceive), []string{"recipient@example.com"}, true, true) + if err == nil { + t.Fatal("saveEmail() error = nil, want incoming relation insert failure") + } + assertAuditCounts(t, engine, 0, 0) +} + +func TestOutgoingAuditRetriesBusyFinalUpdateAtomically(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + var blocker *sql.Tx + + session := newAuditTestSession() + err := session.deliverOutgoingEmail(newAuditTestEmail("final-retry@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + blocker = beginAuditReadTransaction(t, dsn) + go func() { + time.Sleep(35 * time.Millisecond) + _ = blocker.Rollback() + }() + return nil, nil + }) + if err != nil { + t.Fatalf("deliverOutgoingEmail() error = %v", err) + } + assertAuditStatuses(t, engine, consts.EmailStatusSent) + assertAuditError(t, engine, "", false) +} + +func TestOutgoingAuditFinalUpdateIsAtomic(t *testing.T) { + engine, _ := newAuditTestEngine(t) + if _, err := engine.Exec(fmt.Sprintf(`CREATE TRIGGER fail_user_email_success + BEFORE UPDATE OF status ON user_email + WHEN NEW.status = %d + BEGIN + SELECT RAISE(ABORT, 'forced user_email update failure'); + END`, consts.EmailStatusSent)); err != nil { + t.Fatal(err) + } + + session := newAuditTestSession() + err := session.deliverOutgoingEmail(newAuditTestEmail("atomic-update@example.com"), successfulAuditSender) + if err != nil { + t.Fatalf("accepted delivery must remain successful, got %v", err) + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) + assertAuditError(t, engine, deliveryPendingError, true) +} + +func TestAcceptedDeliveryWithPermanentAuditLockReturnsSuccessAndAlerts(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + var blocker *sql.Tx + var deliveries atomic.Int32 + + oldOutput := log.StandardLogger().Out + oldLevel := log.GetLevel() + var logs bytes.Buffer + log.SetOutput(&logs) + log.SetLevel(log.ErrorLevel) + t.Cleanup(func() { + log.SetOutput(oldOutput) + log.SetLevel(oldLevel) + }) + + err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("accepted-unknown@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + blocker = beginAuditReadTransaction(t, dsn) + return nil, nil + }) + if err != nil { + t.Fatalf("accepted delivery must remain successful, got %v", err) + } + if got := deliveries.Load(); got != 1 { + t.Fatalf("downstream deliveries = %d, want 1", got) + } + if !strings.Contains(logs.String(), "audit_event=delivery_result_persist_failed") { + t.Fatalf("missing CRITICAL persistence alert in logs: %s", logs.String()) + } + if err = blocker.Rollback(); err != nil { + t.Fatal(err) + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) + assertAuditError(t, engine, deliveryPendingError, true) +} + +func TestTemporaryDeliveryFailureWithUnpersistedResultCanRetryAsNewAttempt(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + var blocker *sql.Tx + var deliveries atomic.Int32 + session := newAuditTestSession() + messageID := "temporary-unpersisted@example.com" + + err := session.deliverOutgoingEmail(newAuditTestEmail(messageID), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + blocker = beginAuditReadTransaction(t, dsn) + return errors.New("delivery failed"), map[string]error{ + "example.net": &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}, + } + }) + assertSMTPCode(t, err, 451) + if err = blocker.Rollback(); err != nil { + t.Fatal(err) + } + + if err = session.deliverOutgoingEmail(newAuditTestEmail(messageID), func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + return nil, nil + }); err != nil { + t.Fatalf("retry after temporary failure: %v", err) + } + if got := deliveries.Load(); got != 2 { + t.Fatalf("downstream deliveries = %d, want 2", got) + } + assertAuditCounts(t, engine, 2, 2) + assertMessageIDCount(t, engine, messageID, 2) +} + +func TestOutgoingAuditPersistsDeliveryFailure(t *testing.T) { + tests := []struct { + name string + domainErr error + wantCode int + }{ + { + name: "temporary", + domainErr: &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}, + wantCode: 451, + }, + { + name: "permanent", + domainErr: &textproto.Error{Code: 550, Msg: "mailbox unavailable"}, + wantCode: 550, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + engine, _ := newAuditTestEngine(t) + deliveryErr := errors.New("delivery failed") + err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail(tt.name+"-failure@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + return deliveryErr, map[string]error{"example.net": tt.domainErr} + }) + assertSMTPCode(t, err, tt.wantCode) + assertAuditStatuses(t, engine, consts.EmailStatusFail) + assertAuditError(t, engine, deliveryErr.Error(), true) + }) + } +} + +func TestPermanentDeliveryFailureSurvivesFinalAuditLock(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + var blocker *sql.Tx + err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("permanent-failure@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { + blocker = beginAuditReadTransaction(t, dsn) + failure := &textproto.Error{Code: 550, Msg: "mailbox unavailable"} + return errors.New("delivery failed"), map[string]error{"example.net": failure} + }) + assertSMTPCode(t, err, 550) + if err = blocker.Rollback(); err != nil { + t.Fatal(err) + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) + assertAuditError(t, engine, deliveryPendingError, true) +} + +func TestSameMessageIDCreatesIndependentSubmissions(t *testing.T) { + engine, _ := newAuditTestEngine(t) + var deliveries atomic.Int32 + sender := func(*context.Context, *parsemail.Email) (error, map[string]error) { + deliveries.Add(1) + return nil, nil + } + + session := newAuditTestSession() + for i := 0; i < 2; i++ { + email := newAuditTestEmail("reused-message-id@example.com") + email.Text = []byte(fmt.Sprintf("body %d", i)) + if err := session.deliverOutgoingEmail(email, sender); err != nil { + t.Fatalf("submission %d: %v", i+1, err) + } + } + if got := deliveries.Load(); got != 2 { + t.Fatalf("downstream deliveries = %d, want 2", got) + } + assertAuditCounts(t, engine, 2, 2) + assertMessageIDCount(t, engine, "reused-message-id@example.com", 2) +} + +func TestSendAfterHookCompletesBeforeFinalPersistAndSMTPResult(t *testing.T) { + engine, _ := newAuditTestEngine(t) + hook := &blockingSendAfterHook{ + authenticationCaptureHook: &authenticationCaptureHook{}, + started: make(chan struct{}), + release: make(chan struct{}), + done: make(chan struct{}), + } + hooks.HookList = map[string]framework.EmailHook{"blocking": hook} + + result := make(chan error, 1) + go func() { + result <- newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("sync-hook@example.com"), successfulAuditSender) + }() + t.Cleanup(func() { + select { + case <-hook.release: + default: + close(hook.release) + } + }) + + select { + case <-hook.started: + case <-time.After(time.Second): + t.Fatal("SendAfter hook was not started") + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) + + select { + case err := <-result: + t.Fatalf("delivery returned before SendAfter hook completed: %v", err) + case <-time.After(50 * time.Millisecond): + } + close(hook.release) + + select { + case <-hook.done: + case <-time.After(time.Second): + t.Fatal("SendAfter hook did not finish") + } + select { + case err := <-result: + if err != nil { + t.Fatalf("deliverOutgoingEmail() error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("delivery did not return after SendAfter hook completed") + } + assertAuditStatuses(t, engine, consts.EmailStatusSent) +} + +func TestOutgoingPendingAuditMailboxVisibility(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session := newAuditTestSession() + email := newAuditTestEmail("pending-visibility@example.com") + _, saved, err := saveEmail(session.Ctx, email.Size, email, session.Ctx.UserID, int(consts.EmailTypeSend), nil, true, true) + if err != nil { + t.Fatal(err) + } + if saved.Status != consts.EmailStatusDeliveryPending { + t.Fatalf("saved status = %d, want %d", saved.Status, consts.EmailStatusDeliveryPending) + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) + + outgoing, _ := listservice.GetEmailList(session.Ctx, dto.SearchTag{ + Type: consts.EmailTypeSend, + Status: -1, + GroupId: -1, + }, "", false, 0, 10) + found := false + for _, item := range outgoing { + if item.Id == saved.Id { + found = true + if item.Status != consts.EmailStatusDeliveryPending || !item.Error.Valid || item.Error.String != deliveryPendingError { + t.Fatalf("outgoing pending audit = status:%d error:%#v", item.Status, item.Error) + } + } + } + if !found { + t.Fatalf("pending audit email %d is missing from web outgoing list", saved.Id) + } + + for _, mailbox := range []string{"INBOX", "Sent Messages"} { + for _, item := range listservice.GetUEListByUID(session.Ctx, mailbox, 0, 0, nil) { + if item.EmailID == saved.Id { + t.Fatalf("pending audit email %d unexpectedly appears in IMAP mailbox %q", saved.Id, mailbox) + } + } + } +} + +type blockingSendAfterHook struct { + *authenticationCaptureHook + started chan struct{} + release chan struct{} + done chan struct{} +} + +func (h *blockingSendAfterHook) SendAfter(*context.Context, *parsemail.Email, map[string]error) { + defer close(h.done) + close(h.started) + <-h.release +} + +func newAuditTestEngine(t *testing.T) (*xorm.Engine, string) { + t.Helper() + dsn := filepath.Join(t.TempDir(), "audit.db") + "?_pragma=busy_timeout(20)" + engine, err := xorm.NewEngine("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + engine.SetMaxOpenConns(1) + engine.SetMaxIdleConns(1) + if err = engine.Sync2(&models.User{}, &models.Email{}, &models.UserEmail{}); err != nil { + engine.Close() + t.Fatal(err) + } + if _, err = engine.Insert(&models.Email{Subject: "lock seed", MsgID: "lock-seed@example.com"}); err != nil { + engine.Close() + t.Fatal(err) + } + if _, err = engine.Insert(&models.UserEmail{UserID: 99, EmailID: 1}); err != nil { + engine.Close() + t.Fatal(err) + } + + oldDB := db.Instance + oldConfig := config.Instance + oldHooks := hooks.HookList + db.Instance = engine + config.Instance = &config.Config{DbType: config.DBTypeSQLite, Domain: "example.com", Domains: []string{"example.com"}} + hooks.HookList = nil + t.Cleanup(func() { + db.Instance = oldDB + config.Instance = oldConfig + hooks.HookList = oldHooks + _ = engine.Close() + }) + return engine, dsn +} + +func beginAuditReadTransaction(t *testing.T, dsn string) *sql.Tx { + t.Helper() + blocker, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = blocker.Close() }) + tx, err := blocker.Begin() + if err != nil { + t.Fatal(err) + } + var count int + if err = tx.QueryRow("SELECT count(*) FROM email").Scan(&count); err != nil { + _ = tx.Rollback() + t.Fatal(err) + } + return tx +} + +func newAuditTestSession() *Session { + return &Session{Ctx: &context.Context{UserID: 1, UserAccount: "sender", IsAdmin: true}} +} + +func newAuditTestEmail(messageID string) *parsemail.Email { + from := &parsemail.User{Name: "Sender", EmailAddress: "sender@example.com"} + return &parsemail.Email{ + From: from, + Sender: from, + To: []*parsemail.User{{EmailAddress: "recipient@example.net"}}, + Subject: "audit reliability", + Text: []byte("body"), + Size: 4, + MsgID: messageID, + } +} + +func successfulAuditSender(*context.Context, *parsemail.Email) (error, map[string]error) { + return nil, nil +} + +func assertSMTPCode(t *testing.T, err error, want int) { + t.Helper() + var smtpErr *smtp.SMTPError + if !errors.As(err, &smtpErr) { + t.Fatalf("error = %v (%T), want *smtp.SMTPError", err, err) + } + if smtpErr.Code != want { + t.Fatalf("SMTP code = %d, want %d", smtpErr.Code, want) + } +} + +func assertAuditCounts(t *testing.T, engine *xorm.Engine, wantEmail, wantUserEmail int) { + t.Helper() + var emailCount, userEmailCount int + if _, err := engine.SQL("SELECT count(*) FROM email WHERE id != 1").Get(&emailCount); err != nil { + t.Fatal(err) + } + if _, err := engine.SQL("SELECT count(*) FROM user_email WHERE email_id != 1").Get(&userEmailCount); err != nil { + t.Fatal(err) + } + if emailCount != wantEmail || userEmailCount != wantUserEmail { + t.Fatalf("audit counts = email:%d user_email:%d, want email:%d user_email:%d", emailCount, userEmailCount, wantEmail, wantUserEmail) + } +} + +func assertAuditStatuses(t *testing.T, engine *xorm.Engine, want int8) { + t.Helper() + var emailStatus, userEmailStatus int8 + if _, err := engine.SQL("SELECT status FROM email WHERE id != 1 ORDER BY id DESC LIMIT 1").Get(&emailStatus); err != nil { + t.Fatal(err) + } + if _, err := engine.SQL("SELECT status FROM user_email WHERE email_id != 1 ORDER BY email_id DESC LIMIT 1").Get(&userEmailStatus); err != nil { + t.Fatal(err) + } + if emailStatus != want || userEmailStatus != want { + t.Fatalf("audit statuses = email:%d user_email:%d, want %d", emailStatus, userEmailStatus, want) + } +} + +func assertAuditError(t *testing.T, engine *xorm.Engine, want string, wantValid bool) { + t.Helper() + var got sql.NullString + if _, err := engine.SQL("SELECT error FROM email WHERE id != 1 ORDER BY id DESC LIMIT 1").Get(&got); err != nil { + t.Fatal(err) + } + if got.String != want || got.Valid != wantValid { + t.Fatalf("audit error = %#v, want string %q valid %t", got, want, wantValid) + } +} + +func assertMessageIDCount(t *testing.T, engine *xorm.Engine, messageID string, want int) { + t.Helper() + var count int + if _, err := engine.SQL("SELECT count(*) FROM email WHERE msg_id=?", messageID).Get(&count); err != nil { + t.Fatal(err) + } + if count != want { + t.Fatalf("rows for Message-ID %q = %d, want %d", messageID, count, want) + } +} diff --git a/server/listen/smtp_server/read_content.go b/server/listen/smtp_server/read_content.go index 10880e9..2ea54f6 100644 --- a/server/listen/smtp_server/read_content.go +++ b/server/listen/smtp_server/read_content.go @@ -3,7 +3,6 @@ package smtp_server import ( "bytes" "database/sql" - "encoding/json" oerrors "errors" "io" "net" @@ -12,6 +11,7 @@ import ( "time" "github.com/Jinnrry/pmail/config" + "github.com/Jinnrry/pmail/consts" "github.com/Jinnrry/pmail/db" "github.com/Jinnrry/pmail/dto/parsemail" "github.com/Jinnrry/pmail/hooks" @@ -26,7 +26,6 @@ import ( "github.com/mileusna/spf" log "github.com/sirupsen/logrus" "github.com/spf13/cast" - . "xorm.io/builder" ) // DropUnknownRecipientEmails 是代码级别的功能开关 @@ -92,52 +91,7 @@ func (s *Session) Data(r io.Reader) error { return nil } - // 转发 - _, _, err := saveEmail(ctx, len(emailData), email, s.Ctx.UserID, 1, nil, true, true) - if err != nil { - log.WithContext(ctx).Errorf("Email Save Error %v", err) - } - - errMsg := "" - deliveryErr, sendErr := send.Send(ctx, email) - - log.WithContext(ctx).Debugf("插件执行--SendAfter") - - as3 := async.New(ctx) - for _, hook := range hooks.HookList { - if hook == nil { - continue - } - as3.WaitProcess(func(hk any) { - hk.(framework.EmailHook).SendAfter(ctx, email, sendErr) - }, hook) - } - as3.Wait() - log.WithContext(ctx).Debugf("插件执行--SendAfter") - - if deliveryErr != nil { - errMsg = deliveryErr.Error() - _, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =2 ,error=? where id = ? "), errMsg, email.MessageId) - if err != nil { - log.WithContext(ctx).Errorf("sql Error :%+v", err) - } - _, err = db.Instance.Exec(db.WithContext(ctx, "update user_email set status =2 where email_id = ? "), email.MessageId) - if err != nil { - log.WithContext(ctx).Errorf("sql Error :%+v", err) - } - - return downstreamDeliverySMTPError(deliveryErr, sendErr) - - } else { - _, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =1 where id = ? "), email.MessageId) - if err != nil { - log.WithContext(ctx).Errorf("sql Error :%+v", err) - } - _, err = db.Instance.Exec(db.WithContext(ctx, "update user_email set status =1 where email_id = ? "), email.MessageId) - if err != nil { - log.WithContext(ctx).Errorf("sql Error :%+v", err) - } - } + return s.deliverOutgoingEmail(email, send.Send) } else { // 收件 @@ -170,7 +124,11 @@ func (s *Session) Data(r io.Reader) error { email.Status = 3 } - users, dbEmail, _ := saveEmail(ctx, len(emailData), email, 0, 0, s.To, SPFStatus, dkimStatus) + users, dbEmail, saveErr := saveEmail(ctx, len(emailData), email, 0, 0, s.To, SPFStatus, dkimStatus) + if saveErr != nil { + log.WithContext(ctx).Errorf("CRITICAL audit_event=initial_persist_failed direction=inbound msg_id=%q error=%v", email.MsgID, saveErr) + return localPersistenceSMTPError() + } if email.MessageId > 0 { log.WithContext(ctx).Debugf("开始执行邮件规则！") @@ -214,28 +172,29 @@ func (s *Session) Data(r io.Reader) error { } func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserID int, emailType int, reallyTo []string, SPFStatus, dkimStatus bool) ([]*models.User, *models.Email, error) { - if emailType == 0 && email != nil && email.Authentication != nil { + if email == nil { + return nil, nil, oerrors.New("email is missing") + } + if email.From == nil { + return nil, nil, oerrors.New("email sender is missing") + } + if emailType == int(consts.EmailTypeReceive) && email.Authentication != nil { SPFStatus = email.Authentication.SPFPassed dkimStatus = email.Authentication.DKIMPassed } - var dkimV, spfV int8 - if dkimStatus { - dkimV = 1 - } - if SPFStatus { - spfV = 1 - } - log.WithContext(ctx).Debugf("开始入库！") - - if email == nil { - return nil, nil, nil + users, drop, err := resolveIncomingUsers(ctx, email, emailType, reallyTo, SPFStatus, dkimStatus) + if err != nil || drop { + return users, nil, err } msgID := email.MsgID if msgID == "" { msgID = parsemail.GenerateMsgID(config.Instance.Domain) } + if email.Size == 0 { + email.Size = size + } modelEmail := models.Email{ Type: cast.ToInt8(emailType), @@ -251,8 +210,8 @@ func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserI Sender: json2string(email.Sender), Attachments: json2string(email.Attachments), Size: email.Size, - SPFCheck: spfV, - DKIMCheck: dkimV, + SPFCheck: boolInt8(SPFStatus), + DKIMCheck: boolInt8(dkimStatus), SendUserID: sendUserID, SendDate: time.Now(), Status: cast.ToInt8(email.Status), @@ -261,108 +220,25 @@ func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserI MsgID: msgID, } - _, err := db.Instance.Insert(&modelEmail) - - if err != nil { - log.WithContext(ctx).Errorf("db insert error:%+v", err.Error()) - } - - if modelEmail.Id > 0 { - email.MessageId = cast.ToInt64(modelEmail.Id) - email.MsgID = modelEmail.MsgID - } - // 收信人信息 - var users []*models.User - - // 如果是收信 - if emailType == 0 { - // 找到收信人id - accounts := []string{} - // 优先取smtp协议中的收件人地址 - if len(reallyTo) > 0 { - for _, s := range reallyTo { - account := parsemail.BuilderUser(s) - if account != nil { - acc, domain := account.GetDomainAccount() - if array.InArray(domain, config.Instance.Domains) && acc != "" { - accounts = append(accounts, strings.ToLower(acc)) - } - } - } - } else { - for _, user := range append(append(email.To, email.Cc...), email.Bcc...) { - account, _ := user.GetDomainAccount() - if account != "" { - accounts = append(accounts, strings.ToLower(account)) - } - } - } - - /** 这里会导致索引失效，可以尝试对lower结果加索引 - PostgreSQL: CREATE INDEX idx_user_account_lower ON "user" (LOWER(account)); - MySQL8+: ALTER TABLE user ADD INDEX ((LOWER(account))); - */ - where, params, _ := ToSQL(In("LOWER(account)", accounts)) - - err = db.Instance.Table(&models.User{}).Where(where, params...).Find(&users) - if err != nil { - log.WithContext(ctx).Errorf("db Select error:%+v", err.Error()) - } - - if len(users) > 0 { - for _, user := range users { - ue := models.UserEmail{EmailID: modelEmail.Id, UserID: user.ID, Status: cast.ToInt8(email.Status)} - _, err = db.Instance.Insert(&ue) - if err != nil { - log.WithContext(ctx).Errorf("db insert error:%+v", err.Error()) - } - } - } else { - // 找不到收件人 - // 如果启用了保护功能，且 验证失败，则丢弃邮件 - // 验证通过的邮件即使收件人不存在也应交给管理员，避免误丢弃合法邮件 - - // 垃圾过滤 - if DropUnknownRecipientEmails && - ((config.Instance.SpamFilterLevel == 1 && !SPFStatus && !dkimStatus) || - (config.Instance.SpamFilterLevel == 2 && !SPFStatus) || - (config.Instance.SpamFilterLevel == 3 && !dkimStatus)) { - log.WithContext(ctx).Infoln("垃圾邮件，拒信") - // 直接删除已插入的邮件记录，不关联任何用户 - log.WithContext(ctx).Infof("收件人不存在且DKIM验证失败，丢弃邮件: %s -> %v", email.From.EmailAddress, accounts) - _, delErr := db.Instance.Delete(&models.Email{Id: modelEmail.Id}) - if delErr != nil { - log.WithContext(ctx).Errorf("db delete error:%+v", delErr.Error()) - } - return nil, nil, nil - } - - // DKIM 验证通过或未启用保护功能时，邮件丢给管理员账号 - log.WithContext(ctx).Infof("收件人不存在但DKIM验证通过，转交管理员: %s -> %v", email.From.EmailAddress, accounts) - - err = db.Instance.Table(&models.User{}).Where("is_admin=1").Find(&users) - for _, user := range users { - ue := models.UserEmail{EmailID: modelEmail.Id, UserID: user.ID, Status: cast.ToInt8(email.Status)} - _, err = db.Instance.Insert(&ue) - if err != nil { - log.WithContext(ctx).Errorf("db insert error:%+v", err.Error()) - } - } + var userIDs []int + if emailType == int(consts.EmailTypeReceive) { + for _, user := range users { + userIDs = append(userIDs, user.ID) } } else { - ue := models.UserEmail{EmailID: modelEmail.Id, UserID: ctx.UserID} - _, err = db.Instance.Insert(&ue) - if err != nil { - log.WithContext(ctx).Errorf("db insert error:%+v", err.Error()) - } + modelEmail.Status = consts.EmailStatusDeliveryPending + modelEmail.Error = sql.NullString{String: deliveryPendingError, Valid: true} + userIDs = []int{sendUserID} } - return users, &modelEmail, nil -} - -func json2string(d any) string { - by, _ := json.Marshal(d) - return string(by) + log.WithContext(ctx).Debugf("开始入库！") + saved, err := persistInitialAudit(ctx, modelEmail, userIDs) + if err != nil { + return users, nil, err + } + email.MessageId = cast.ToInt64(saved.Id) + email.MsgID = saved.MsgID + return users, saved, nil } func spfCheck(remoteAddress string, sender *parsemail.User, senderString string) bool {