fix(smtp): make audit persistence atomic (#358)
This commit is contained in:
0clickjacking0
GitHub
parent
1da3e178bf
commit
a6049bf052
@ -23,4 +23,7 @@ const (
|
||||
|
||||
// EmailStatusJunk 骚扰邮件
|
||||
EmailStatusJunk int8 = 5
|
||||
|
||||
// EmailStatusDeliveryPending marks an outbound audit whose delivery result is not final.
|
||||
EmailStatusDeliveryPending int8 = 6
|
||||
)
|
||||
|
||||
@ -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")
|
||||
|
||||
174
server/db/retry.go
Normal file
174
server/db/retry.go
Normal file
@ -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(<milliseconds>)")
|
||||
}
|
||||
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(<milliseconds>) or busy_timeout = <milliseconds>")
|
||||
}
|
||||
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
|
||||
}
|
||||
407
server/db/retry_test.go
Normal file
407
server/db/retry_test.go
Normal file
@ -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
|
||||
}
|
||||
@ -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
|
||||
)
|
||||
|
||||
@ -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=
|
||||
|
||||
215
server/listen/smtp_server/audit_persistence.go
Normal file
215
server/listen/smtp_server/audit_persistence.go
Normal file
@ -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)
|
||||
}
|
||||
562
server/listen/smtp_server/audit_persistence_test.go
Normal file
562
server/listen/smtp_server/audit_persistence_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
@ -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 {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user