diff --git a/backend/consts/consts.go b/backend/consts/consts.go index c6f3d51..5506a56 100644 --- a/backend/consts/consts.go +++ b/backend/consts/consts.go @@ -26,4 +26,7 @@ const ( // EmailStatusDeliveryPending marks an outbound audit whose delivery result is not final. EmailStatusDeliveryPending int8 = 6 + + // EmailStatusDeliveryUncertain 禁止自动重投，需人工核对接收结果。 + EmailStatusDeliveryUncertain int8 = 7 ) diff --git a/backend/controllers/email/send.go b/backend/controllers/email/send.go index 822bde0..ecca54b 100644 --- a/backend/controllers/email/send.go +++ b/backend/controllers/email/send.go @@ -10,17 +10,14 @@ import ( "git.pcux.cn/xuans/Smail/dto/parsemail" "git.pcux.cn/xuans/Smail/dto/response" "git.pcux.cn/xuans/Smail/hooks" - "git.pcux.cn/xuans/Smail/hooks/framework" "git.pcux.cn/xuans/Smail/i18n" "git.pcux.cn/xuans/Smail/models" "git.pcux.cn/xuans/Smail/services/access" + "git.pcux.cn/xuans/Smail/services/outbox" - "git.pcux.cn/xuans/Smail/utils/async" "git.pcux.cn/xuans/Smail/utils/context" - "git.pcux.cn/xuans/Smail/utils/send" "git.pcux.cn/xuans/Smail/utils/validation" log "github.com/sirupsen/logrus" - "github.com/spf13/cast" "mime" "net/http" "strings" @@ -274,61 +271,12 @@ func Send(ctx *context.Context, w http.ResponseWriter, req *http.Request) { } RecordUserContacts(ctx.UserID, contactList) - e.MessageId = cast.ToInt64(modelEmail.Id) - e.MsgID = modelEmail.MsgID - - async.New(ctx).Process(func(p any) { - errMsg := "" - err, sendErr := send.SendTriggered(ctx, e, "网页写信") - - log.WithContext(ctx).Debugf("插件执行--SendAfter") - - as2 := async.New(ctx) - for _, hook := range hooks.Snapshot() { - if hook == nil { - continue - } - as2.WaitProcess(func(hk any) { - hk.(framework.EmailHook).SendAfter(ctx, e, sendErr) - }, hook) - } - as2.Wait() - log.WithContext(ctx).Debugf("插件执行--SendAfter") - - if err != nil { - errMsg = err.Error() - _ = db.Transaction(db.Instance, func(session *xorm.Session) error { - if _, updateErr := session.Exec(db.WithContext(ctx, "update email set status = ?, error = ? where id = ?"), consts.EmailStatusFail, errMsg, modelEmail.Id); updateErr != nil { - log.WithContext(ctx).Errorf("update email status failed: %+v", updateErr) - return updateErr - } - if _, updateErr := session.Exec(db.WithContext(ctx, "update user_email set status = ? where email_id = ? and user_id = ?"), consts.EmailStatusFail, modelEmail.Id, ctx.UserID); updateErr != nil { - log.WithContext(ctx).Errorf("update user_email status failed: %+v", updateErr) - return updateErr - } - return nil - }) - } else { - _ = db.Transaction(db.Instance, func(session *xorm.Session) error { - if _, updateErr := session.Exec(db.WithContext(ctx, "update email set status = ?, error = '' where id = ?"), consts.EmailStatusSent, modelEmail.Id); updateErr != nil { - log.WithContext(ctx).Errorf("update email status failed: %+v", updateErr) - return updateErr - } - if _, updateErr := session.Exec(db.WithContext(ctx, "update user_email set status = ? where email_id = ? and user_id = ?"), consts.EmailStatusSent, modelEmail.Id, ctx.UserID); updateErr != nil { - log.WithContext(ctx).Errorf("update user_email status failed: %+v", updateErr) - return updateErr - } - return nil - }) - } - - }, nil) - response.NewSuccessResponse(i18n.GetText(ctx.Lang, "succ")).FPrint(w) } func persistOutgoingEmail(engine *xorm.Engine, userID int, email *models.Email) error { - return db.Transaction(engine, func(session *xorm.Session) error { + var saved models.Email + err := db.Transaction(engine, func(session *xorm.Session) error { candidate := *email candidate.Id = 0 candidate.Status = consts.EmailStatusDeliveryPending @@ -347,9 +295,16 @@ func persistOutgoingEmail(engine *xorm.Engine, userID int, email *models.Email) if _, err := session.Insert(&relation); err != nil { return err } - *email = candidate + if err := outbox.Enqueue(session, candidate); err != nil { + return err + } + saved = candidate return nil }) + if err == nil { + *email = saved + } + return err } func decodeAttachment(data string) (string, []byte, error) { diff --git a/backend/controllers/email/send_test.go b/backend/controllers/email/send_test.go index f0e6fd3..856e003 100644 --- a/backend/controllers/email/send_test.go +++ b/backend/controllers/email/send_test.go @@ -20,7 +20,7 @@ func TestSendTransactionalPersistenceConsistency(t *testing.T) { t.Fatalf("create test engine: %v", err) } engine.SetMaxOpenConns(1) - if err := engine.Sync2(&models.Email{}, &models.UserEmail{}); err != nil { + if err := engine.Sync2(&models.Email{}, &models.UserEmail{}, &models.OutboundDelivery{}); err != nil { t.Fatalf("sync test database: %v", err) } defer engine.Close() @@ -43,6 +43,7 @@ func TestSendTransactionalPersistenceConsistency(t *testing.T) { Status: consts.EmailStatusDeliveryPending, CreateTime: now, MsgID: "msg-123@example.com", + To: `[{"EmailAddress":"recipient@example.com"}]`, } err = persistOutgoingEmail(engine, ctx.UserID, &modelEmail) @@ -120,18 +121,18 @@ func TestOutboxExposesDeliveryStatus(t *testing.T) { } defer engine.Close() engine.SetMaxOpenConns(1) - if err := engine.Sync2(&models.Email{}, &models.UserEmail{}, &models.UserTrustedSender{}); err != nil { + if err := engine.Sync2(&models.Email{}, &models.UserEmail{}, &models.UserTrustedSender{}, &models.OutboundDelivery{}); err != nil { t.Fatal(err) } previous := db.Instance db.Instance = engine defer func() { db.Instance = previous }() ctx := &context.Context{UserID: 1} - email := models.Email{Subject: "pending", Type: consts.EmailTypeSend, Status: consts.EmailStatusSent} + email := models.Email{Subject: "pending", Type: consts.EmailTypeSend, Status: consts.EmailStatusSent, To: `[{"EmailAddress":"recipient@example.com"}]`} if err := persistOutgoingEmail(engine, ctx.UserID, &email); err != nil { t.Fatal(err) } - for _, status := range []int8{consts.EmailStatusDeliveryPending, consts.EmailStatusSent, consts.EmailStatusFail} { + for _, status := range []int8{consts.EmailStatusDeliveryPending, consts.EmailStatusSent, consts.EmailStatusFail, consts.EmailStatusDeliveryUncertain} { if _, err := engine.ID(email.Id).Cols("status").Update(&models.Email{Status: status}); err != nil { t.Fatal(err) } @@ -150,3 +151,28 @@ func TestOutboxExposesDeliveryStatus(t *testing.T) { } } } + +func TestPersistOutgoingEmailRollsBackQueueFailure(t *testing.T) { + engine, err := xorm.NewEngine("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + defer engine.Close() + engine.SetMaxOpenConns(1) + if err := engine.Sync2(&models.Email{}, &models.UserEmail{}, &models.OutboundDelivery{}); err != nil { + t.Fatal(err) + } + if _, err := engine.Exec("CREATE TRIGGER reject_queue BEFORE INSERT ON outbound_delivery BEGIN SELECT RAISE(ABORT, 'injected queue failure'); END"); err != nil { + t.Fatal(err) + } + email := models.Email{To: `[{"EmailAddress":"recipient@example.com"}]`} + if err := persistOutgoingEmail(engine, 1, &email); err == nil { + t.Fatal("queue failure ignored") + } + for _, model := range []any{&models.Email{}, &models.UserEmail{}, &models.OutboundDelivery{}} { + count, err := engine.Count(model) + if err != nil || count != 0 || email.Id != 0 { + t.Fatalf("rollback failed for %T: count=%d id=%d error=%v", model, count, email.Id, err) + } + } +} diff --git a/backend/db/migration/outbound_delivery.go b/backend/db/migration/outbound_delivery.go new file mode 100644 index 0000000..5a8a22b --- /dev/null +++ b/backend/db/migration/outbound_delivery.go @@ -0,0 +1,19 @@ +package migration + +func outboundDeliveryStatements() map[Dialect][]string { + return map[Dialect][]string{ + DialectSQLite: { + `CREATE TABLE IF NOT EXISTS "outbound_delivery" ("id" INTEGER PRIMARY KEY AUTOINCREMENT, "email_id" INTEGER NOT NULL, "recipient" TEXT NOT NULL, "state" TEXT NOT NULL, "attempts" INTEGER NOT NULL DEFAULT 0, "next_attempt" BIGINT NOT NULL DEFAULT 0, "last_error" TEXT NOT NULL DEFAULT '')`, + `CREATE INDEX IF NOT EXISTS "idx_outbound_delivery_email" ON "outbound_delivery"("email_id")`, + `CREATE INDEX IF NOT EXISTS "idx_outbound_delivery_due" ON "outbound_delivery"("state", "next_attempt")`, + }, + DialectMySQL: { + "CREATE TABLE IF NOT EXISTS `outbound_delivery` (`id` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, `email_id` BIGINT NOT NULL, `recipient` VARCHAR(320) NOT NULL, `state` VARCHAR(24) NOT NULL, `attempts` INT NOT NULL DEFAULT 0, `next_attempt` BIGINT NOT NULL DEFAULT 0, `last_error` TEXT NOT NULL, KEY `idx_outbound_delivery_email` (`email_id`), KEY `idx_outbound_delivery_due` (`state`, `next_attempt`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", + }, + DialectPostgres: { + `CREATE TABLE IF NOT EXISTS "outbound_delivery" ("id" BIGSERIAL PRIMARY KEY, "email_id" BIGINT NOT NULL, "recipient" VARCHAR(320) NOT NULL, "state" VARCHAR(24) NOT NULL, "attempts" INTEGER NOT NULL DEFAULT 0, "next_attempt" BIGINT NOT NULL DEFAULT 0, "last_error" TEXT NOT NULL DEFAULT '')`, + `CREATE INDEX IF NOT EXISTS "idx_outbound_delivery_email" ON "outbound_delivery"("email_id")`, + `CREATE INDEX IF NOT EXISTS "idx_outbound_delivery_due" ON "outbound_delivery"("state", "next_attempt")`, + }, + } +} diff --git a/backend/db/migration/registry.go b/backend/db/migration/registry.go index 1369aa6..db6201d 100644 --- a/backend/db/migration/registry.go +++ b/backend/db/migration/registry.go @@ -1,4 +1,4 @@ -package migration +package migration import ( "regexp" @@ -32,6 +32,7 @@ func DefaultMigrations() []Migration { DefineCustom(20, "expand_protocol_credential_ciphertext", "expand_protocol_credential_ciphertext_v1", applyProtocolCredentialCiphertextSchema), DefineCustom(21, "expand_trusted_sender_blacklist", "expand_trusted_sender_blacklist_v1", applyTrustedSenderBlacklistSchema), DefineForDialects(22, "create_client_credentials", clientCredentialStatements()), + DefineForDialects(23, "create_outbound_delivery", outboundDeliveryStatements()), } } diff --git a/backend/db/migration/registry_test.go b/backend/db/migration/registry_test.go index ed398b1..9f29c50 100644 --- a/backend/db/migration/registry_test.go +++ b/backend/db/migration/registry_test.go @@ -296,6 +296,7 @@ func TestDefaultMigrationChecksumsAreImmutable(t *testing.T) { 20: "e0bc01f4c2df742bc23c519c2a399cc1c3f341565dbed014509a8c63d68bf725", 21: "ac1a0b3be90128c0e0512d2cec858f8cd7e70abc5ff4734d7113b9674f08dbb0", 22: "1047cb22f5a32c8a09ba4ccedff8713a5225c14fdb98357db86a6e0723c30d8b", + 23: "5d80f2c2c583fc9a033fc320c0852c73bd6701ce6c8f12e8904deedfe0ba44b2", } migrations := DefaultMigrations() if len(migrations) != len(expected) { diff --git a/backend/models/outbound_delivery.go b/backend/models/outbound_delivery.go new file mode 100644 index 0000000..4080a22 --- /dev/null +++ b/backend/models/outbound_delivery.go @@ -0,0 +1,13 @@ +package models + +type OutboundDelivery struct { + ID int64 `xorm:"id pk autoincr"` + EmailID int `xorm:"email_id notnull index"` + Recipient string `xorm:"varchar(320) notnull"` + State string `xorm:"varchar(24) notnull"` + Attempts int `xorm:"notnull"` + NextAttempt int64 `xorm:"notnull"` + LastError string `xorm:"text"` +} + +func (OutboundDelivery) TableName() string { return "outbound_delivery" } diff --git a/backend/res_init/init.go b/backend/res_init/init.go index 20c678f..2e25d03 100644 --- a/backend/res_init/init.go +++ b/backend/res_init/init.go @@ -25,6 +25,7 @@ import ( "git.pcux.cn/xuans/Smail/services/dbmigration" "git.pcux.cn/xuans/Smail/services/externalemail" "git.pcux.cn/xuans/Smail/services/logstore" + "git.pcux.cn/xuans/Smail/services/outbox" pluginservice "git.pcux.cn/xuans/Smail/services/plugin" "git.pcux.cn/xuans/Smail/services/recovery" "git.pcux.cn/xuans/Smail/services/restart" @@ -47,6 +48,7 @@ var verificationRuntime *verification.Runtime var credentialService *credentials.Service var webAuthnRuntimeService *webauthnservice.Service var logRuntime *logstore.Runtime +var outboxRuntime *outbox.Runtime func startConfigManager() error { root := filepath.Join(config.ROOT_PATH, "config") @@ -215,6 +217,11 @@ func Init(serverVersion string) error { }() } hooks.Init(serverVersion) + outboxRuntime, err = outbox.Start(db.Instance) + if err != nil { + stopServices() + return fmt.Errorf("start persistent outbox: %w", err) + } httpRun, httpsRun, err := http_server.PrepareWebServers() if err != nil { stopServices() @@ -294,6 +301,14 @@ func stopServices() error { http_server.HttpStop() pop3_server.Stop() imap_server.Stop() + // 停止接收新请求后等待正在投递的任务写回，超时则保留数据库，避免提前关闭连接池。 + drainCtx, drainCancel := context.WithTimeout(context.Background(), 25*time.Second) + err := outboxRuntime.Stop(drainCtx) + drainCancel() + if err != nil { + return fmt.Errorf("outbox did not drain; database retained: %w", err) + } + outboxRuntime = nil hooks.Stop() // 先关闭业务入口，再撤下安全服务及会话清理器，避免新请求读取停止中的依赖。 stopVerificationRuntime() diff --git a/backend/services/dbmigration/maintenance_test.go b/backend/services/dbmigration/maintenance_test.go index 604171c..8db399e 100644 --- a/backend/services/dbmigration/maintenance_test.go +++ b/backend/services/dbmigration/maintenance_test.go @@ -104,3 +104,40 @@ func TestMaintenanceGuardDrainsRequestsAndKeepsRestartRoute(t *testing.T) { t.Fatal("cleanup wrote to frozen source") } } + +func TestMaintenanceDrainsBackgroundDeliveryBeforeFreeze(t *testing.T) { + executor := &PortableExecutor{} + service := NewService(&DatabaseExecutor{portable: executor}) + previous := Current() + SetCurrent(service) + t.Cleanup(func() { SetCurrent(previous) }) + entered, finish, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + go func() { + defer close(done) + _ = WhileWritable(func() error { + close(entered) + <-finish + return nil + }) + }() + <-entered + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + release, err := service.drainRequests(ctx) + release() + close(finish) + <-done + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("migration did not wait for delivery: %v", err) + } + release, err = service.drainRequests(context.Background()) + if err != nil { + t.Fatal(err) + } + executor.paused.Store(true) + release() + called := false + if err := WhileWritable(func() error { called = true; return nil }); err != nil || called { + t.Fatal("delivery started after source freeze") + } +} diff --git a/backend/services/outbox/runtime.go b/backend/services/outbox/runtime.go new file mode 100644 index 0000000..1b969e1 --- /dev/null +++ b/backend/services/outbox/runtime.go @@ -0,0 +1,172 @@ +package outbox + +import ( + stdcontext "context" + "errors" + "fmt" + "sync" + "time" + + "git.pcux.cn/xuans/Smail/dto/parsemail" + "git.pcux.cn/xuans/Smail/hooks" + "git.pcux.cn/xuans/Smail/models" + "git.pcux.cn/xuans/Smail/services/dbmigration" + "git.pcux.cn/xuans/Smail/utils/context" + "git.pcux.cn/xuans/Smail/utils/send" + "git.pcux.cn/xuans/Smail/utils/smtp" + log "github.com/sirupsen/logrus" + "xorm.io/xorm" +) + +type Runtime struct { + engine *xorm.Engine + stop chan struct{} + done chan struct{} + once sync.Once + deliver func(*context.Context, *parsemail.Email, string) error +} + +func Start(engine *xorm.Engine) (*Runtime, error) { + if err := recoverInterrupted(engine); err != nil { + return nil, err + } + r := &Runtime{engine: engine, stop: make(chan struct{}), done: make(chan struct{}), deliver: send.SendRecipientTriggered} + go r.loop() + return r, nil +} + +func (r *Runtime) Stop(ctx stdcontext.Context) error { + if r == nil { + return nil + } + r.once.Do(func() { close(r.stop) }) + select { + case <-r.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (r *Runtime) loop() { + defer close(r.done) + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-r.stop: + return + default: + } + var jobs []models.OutboundDelivery + err := r.engine.Where("state = ? AND next_attempt <= ?", queued, time.Now().Unix()).Asc("id").Limit(20).Find(&jobs) + if err != nil { + log.WithError(err).Error("读取持久投递队列失败") + } + for _, job := range jobs { + select { + case <-r.stop: + return + default: + } + if err := r.process(job); err != nil { + log.WithError(err).WithField("delivery_id", job.ID).Error("处理持久投递队列失败") + } + } + select { + case <-r.stop: + return + case <-ticker.C: + } + } +} + +func (r *Runtime) process(job models.OutboundDelivery) error { + // 迁移必须等待 SMTP 及结果落库结束，冻结源库后不再领取任务。 + return dbmigration.WhileWritable(func() error { return r.processWritable(job) }) +} + +func (r *Runtime) processWritable(job models.OutboundDelivery) error { + // 读取失败时尚未开始投递，保留排队状态供数据库恢复后继续处理。 + var stored models.Email + emailFound, err := r.engine.ID(job.EmailID).Get(&stored) + if err != nil { + return err + } + var user models.User + userFound := false + if emailFound { + userFound, err = r.engine.ID(stored.SendUserID).Get(&user) + if err != nil { + return err + } + } + owned, err := claim(r.engine, &job) + if err != nil || !owned { + return err + } + ctx := &context.Context{Context: stdcontext.Background()} + var message *parsemail.Email + deliveryErr := func() (result error) { + defer func() { + if value := recover(); value != nil { + result = &smtp.DeliveryUncertainError{Cause: fmt.Errorf("投递任务异常: %v", value)} + } + }() + if !emailFound { + return errors.New("待投递邮件已不存在") + } + if !userFound || user.Disabled != 0 { + return errors.New("发件账号不可用，已停止投递") + } + ctx.UserID, ctx.UserAccount, ctx.UserName, ctx.Role = user.ID, user.Account, user.Name, user.EffectiveRole() + message, err = parsemail.NewEmailFromModel(stored) + if err != nil { + return err + } + return r.deliver(ctx, message, job.Recipient) + }() + // 只重试结果落库，绝不因数据库故障再次调用 SMTP。 + var terminal bool + for attempt := 0; attempt < 3; attempt++ { + terminal, err = finish(r.engine, job, deliveryErr) + if err == nil { + break + } + time.Sleep(time.Second) + } + if err != nil { + return fmt.Errorf("投递结果写入失败，保留在途记录并禁止自动重投: %w", err) + } + if terminal && message != nil { + r.notify(ctx, message, job.EmailID) + } + return nil +} + +func (r *Runtime) notify(ctx *context.Context, message *parsemail.Email, emailID int) { + var jobs []models.OutboundDelivery + if err := r.engine.Where("email_id = ?", emailID).Find(&jobs); err != nil { + log.WithError(err).Error("读取投递结果用于插件通知失败") + return + } + failures := make(map[string]error) + for _, job := range jobs { + if job.State != accepted { + failures[job.Recipient] = errors.New(job.LastError) + } + } + for _, hook := range hooks.Snapshot() { + if hook == nil { + continue + } + func() { + defer func() { + if recover() != nil { + log.Error("投递完成插件发生异常") + } + }() + hook.SendAfter(ctx, message, failures) + }() + } +} diff --git a/backend/services/outbox/store.go b/backend/services/outbox/store.go new file mode 100644 index 0000000..4ec1d68 --- /dev/null +++ b/backend/services/outbox/store.go @@ -0,0 +1,179 @@ +package outbox + +import ( + "database/sql" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.pcux.cn/xuans/Smail/consts" + "git.pcux.cn/xuans/Smail/db" + "git.pcux.cn/xuans/Smail/dto/parsemail" + "git.pcux.cn/xuans/Smail/models" + "git.pcux.cn/xuans/Smail/services/logstore" + "git.pcux.cn/xuans/Smail/utils/send" + "git.pcux.cn/xuans/Smail/utils/smtp" + "xorm.io/xorm" +) + +const ( + queued = "queued" + running = "running" + accepted = "accepted" + failed = "failed" + uncertain = "uncertain" + maxAttempts = 5 + uncertainMessage = "投递结果不确定，已暂停自动重试，请先核对收件结果" +) + +// Enqueue 必须与邮件及用户关联在同一事务内保存，不保存验证码或恢复令牌。 +func Enqueue(session *xorm.Session, email models.Email) error { + message, err := parsemail.NewEmailFromModel(email) + if err != nil { + return err + } + seen := make(map[string]bool) + for _, user := range append(append(message.To, message.Cc...), message.Bcc...) { + if user == nil || user.EmailAddress == "" { + return errors.New("队列收件人不能为空") + } + // 本地部分保留大小写，仅合并完全相同的 SMTP 信封地址。 + address := user.EmailAddress + if at := strings.LastIndexByte(address, '@'); at >= 0 { + address = address[:at+1] + strings.ToLower(address[at+1:]) + } + if seen[address] { + continue + } + seen[address] = true + job := models.OutboundDelivery{EmailID: email.Id, Recipient: address, State: queued} + if _, err := session.Insert(&job); err != nil { + return err + } + } + if len(seen) == 0 { + return errors.New("队列必须包含收件人") + } + return nil +} + +func summarize(jobs []models.OutboundDelivery) (int8, string, bool) { + counts := make(map[string]int) + reasons := make(map[string]bool) + for _, job := range jobs { + counts[job.State]++ + if job.LastError != "" { + reasons[job.LastError] = true + } + } + if counts[queued]+counts[running] > 0 { + return consts.EmailStatusDeliveryPending, "", false + } + details := make([]string, 0, len(reasons)) + for reason := range reasons { + details = append(details, reason) + } + sort.Strings(details) + suffix := "" + if len(details) > 0 { + suffix = "；原因：" + strings.Join(details, "；") + } + if counts[uncertain] > 0 { + return consts.EmailStatusDeliveryUncertain, fmt.Sprintf("%s；已接收 %d，结果不确定 %d，失败 %d", uncertainMessage, counts[accepted], counts[uncertain], counts[failed]) + suffix, true + } + if len(jobs) > 0 && counts[accepted] == len(jobs) { + return consts.EmailStatusSent, "", true + } + return consts.EmailStatusFail, fmt.Sprintf("投递未全部完成；已接收 %d，失败 %d。已接收的收件人不会自动重投", counts[accepted], counts[failed]) + suffix, true +} + +func syncStatus(session *xorm.Session, emailID int) (bool, error) { + var jobs []models.OutboundDelivery + if err := session.Where("email_id = ?", emailID).Find(&jobs); err != nil { + return false, err + } + status, message, terminal := summarize(jobs) + if _, err := session.ID(emailID).Cols("status", "error").Update(&models.Email{Status: status, Error: sql.NullString{String: message, Valid: true}}); err != nil { + return false, err + } + // 不改动用户已放入回收站的关联状态。 + _, err := session.Where("email_id = ? AND status <> ?", emailID, consts.EmailStatusDel).Cols("status").Update(&models.UserEmail{Status: status}) + return terminal, err +} + +// recoverInterrupted 只在独占运行实例启动时执行，在途记录绝不重新排队。 +func recoverInterrupted(engine *xorm.Engine) error { + return db.Transaction(engine, func(session *xorm.Session) error { + // 旧版本未记录 SMTP 尝试边界，只标记待核对，不能当作未发送任务导入。 + var legacy []models.Email + if err := session.Where("type = ? AND status = ? AND id NOT IN (SELECT email_id FROM outbound_delivery)", consts.EmailTypeSend, consts.EmailStatusDeliveryPending).Find(&legacy); err != nil { + return err + } + for _, email := range legacy { + if _, err := session.ID(email.Id).Cols("status", "error").Update(&models.Email{Status: consts.EmailStatusDeliveryUncertain, Error: sql.NullString{String: uncertainMessage, Valid: true}}); err != nil { + return err + } + if _, err := session.Where("email_id = ? AND status = ?", email.Id, consts.EmailStatusDeliveryPending).Cols("status").Update(&models.UserEmail{Status: consts.EmailStatusDeliveryUncertain}); err != nil { + return err + } + } + var interrupted []models.OutboundDelivery + if err := session.Where("state = ?", running).Find(&interrupted); err != nil { + return err + } + if _, err := session.Where("state = ?", running).Cols("state", "last_error").Update(&models.OutboundDelivery{State: uncertain, LastError: uncertainMessage}); err != nil { + return err + } + seen := make(map[int]bool) + for _, job := range interrupted { + if !seen[job.EmailID] { + if _, err := syncStatus(session, job.EmailID); err != nil { + return err + } + seen[job.EmailID] = true + } + } + return nil + }) +} + +func claim(engine *xorm.Engine, job *models.OutboundDelivery) (bool, error) { + update := *job + update.State = running + update.Attempts++ + count, err := engine.ID(job.ID).Where("state = ? AND attempts = ? AND next_attempt <= ?", queued, job.Attempts, time.Now().Unix()).Cols("state", "attempts").Update(&update) + if err == nil && count == 1 { + *job = update + } + return count == 1, err +} + +func finish(engine *xorm.Engine, job models.OutboundDelivery, deliveryErr error) (bool, error) { + job.State, job.LastError, job.NextAttempt = accepted, "", 0 + if deliveryErr != nil { + // 不保存远端响应原文，避免凭据或恶意超长内容进入数据库和页面。 + job.State, job.LastError = failed, logstore.MailFailureReason(deliveryErr) + if smtp.IsDeliveryUncertain(deliveryErr) { + job.State = uncertain + job.LastError = uncertainMessage + } else if send.Retryable(deliveryErr) && job.Attempts < maxAttempts { + job.State = queued + job.NextAttempt = time.Now().Add(time.Minute * time.Duration(1<<(job.Attempts-1))).Unix() + } + } + terminal := false + err := db.Transaction(engine, func(session *xorm.Session) error { + count, err := session.ID(job.ID).Where("state = ? AND attempts = ?", running, job.Attempts).Cols("state", "last_error", "next_attempt").Update(&job) + if err != nil { + return err + } + if count != 1 { + return errors.New("投递任务状态已发生变化") + } + terminal, err = syncStatus(session, job.EmailID) + return err + }) + return terminal, err +} diff --git a/backend/services/outbox/store_test.go b/backend/services/outbox/store_test.go new file mode 100644 index 0000000..5406edc --- /dev/null +++ b/backend/services/outbox/store_test.go @@ -0,0 +1,340 @@ +package outbox + +import ( + stdcontext "context" + "errors" + "io" + "net/textproto" + "strings" + "sync/atomic" + "testing" + "time" + + "git.pcux.cn/xuans/Smail/consts" + "git.pcux.cn/xuans/Smail/db" + "git.pcux.cn/xuans/Smail/dto/parsemail" + "git.pcux.cn/xuans/Smail/models" + "git.pcux.cn/xuans/Smail/utils/context" + "git.pcux.cn/xuans/Smail/utils/smtp" + "xorm.io/xorm" +) + +func queueFixture(t *testing.T) (*xorm.Engine, models.Email, []models.OutboundDelivery) { + t.Helper() + engine, err := xorm.NewEngine("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + engine.SetMaxOpenConns(1) + t.Cleanup(func() { engine.Close() }) + if err := engine.Sync2(&models.Email{}, &models.UserEmail{}, &models.User{}, &models.OutboundDelivery{}); err != nil { + t.Fatal(err) + } + user := models.User{Account: "sender", Name: "sender", Password: "test", Role: "user"} + if _, err := engine.Insert(&user); err != nil { + t.Fatal(err) + } + email := models.Email{Type: consts.EmailTypeSend, Status: consts.EmailStatusDeliveryPending, SendUserID: user.ID, FromAddress: "sender@example.com", To: `[{"EmailAddress":"one@example.com"},{"EmailAddress":"two@example.com"}]`, Cc: `[{"EmailAddress":"one@EXAMPLE.com"}]`, MsgID: "stable@example.com"} + err = db.Transaction(engine, func(session *xorm.Session) error { + if _, err := session.Insert(&email); err != nil { + return err + } + if _, err := session.Insert(&models.UserEmail{UserID: user.ID, EmailID: email.Id, Status: consts.EmailStatusDeliveryPending}); err != nil { + return err + } + return Enqueue(session, email) + }) + if err != nil { + t.Fatal(err) + } + var jobs []models.OutboundDelivery + if err := engine.Asc("id").Find(&jobs); err != nil { + t.Fatal(err) + } + if len(jobs) != 2 { + t.Fatalf("duplicate recipient: %v", jobs) + } + return engine, email, jobs +} + +func assertStatus(t *testing.T, engine *xorm.Engine, emailID int, status int8) { + t.Helper() + var email models.Email + var relation models.UserEmail + if ok, err := engine.ID(emailID).Get(&email); err != nil || !ok { + t.Fatal(err) + } + if ok, err := engine.Where("email_id = ?", emailID).Get(&relation); err != nil || !ok { + t.Fatal(err) + } + if email.Status != status || relation.Status != status { + t.Fatalf("email=%d relation=%d want=%d", email.Status, relation.Status, status) + } +} + +func TestClaimAndPartialResults(t *testing.T) { + engine, email, jobs := queueFixture(t) + first := jobs[0] + if ok, err := claim(engine, &first); err != nil || !ok { + t.Fatal(err) + } + if ok, err := claim(engine, &jobs[0]); err != nil || ok { + t.Fatalf("duplicate claim: %v %v", ok, err) + } + if _, err := finish(engine, first, nil); err != nil { + t.Fatal(err) + } + assertStatus(t, engine, email.Id, consts.EmailStatusDeliveryPending) + second := jobs[1] + if ok, err := claim(engine, &second); err != nil || !ok { + t.Fatal(err) + } + if _, err := finish(engine, second, &textproto.Error{Code: 451, Msg: "later"}); err != nil { + t.Fatal(err) + } + var queuedJob models.OutboundDelivery + if _, err := engine.ID(second.ID).Get(&queuedJob); err != nil { + t.Fatal(err) + } + if queuedJob.State != queued || queuedJob.NextAttempt <= time.Now().Unix() { + t.Fatalf("retry state=%+v", queuedJob) + } + if ok, err := claim(engine, &queuedJob); err != nil || ok { + t.Fatal("retry was claimed before backoff elapsed") + } + if _, err := engine.ID(queuedJob.ID).Cols("next_attempt").Update(&models.OutboundDelivery{}); err != nil { + t.Fatal(err) + } + if ok, err := claim(engine, &queuedJob); err != nil || !ok { + t.Fatal(err) + } + if _, err := finish(engine, queuedJob, &textproto.Error{Code: 550, Msg: "rejected"}); err != nil { + t.Fatal(err) + } + assertStatus(t, engine, email.Id, consts.EmailStatusFail) + if ok, err := claim(engine, &first); err != nil || ok { + t.Fatal("accepted job was claimed again") + } +} + +func TestRestartDoesNotResendInterruptedOrAccepted(t *testing.T) { + engine, email, jobs := queueFixture(t) + for i := range jobs { + if ok, err := claim(engine, &jobs[i]); err != nil || !ok { + t.Fatal(err) + } + } + if _, err := finish(engine, jobs[0], nil); err != nil { + t.Fatal(err) + } + if err := recoverInterrupted(engine); err != nil { + t.Fatal(err) + } + assertStatus(t, engine, email.Id, consts.EmailStatusDeliveryUncertain) + var stored []models.OutboundDelivery + if err := engine.Asc("id").Find(&stored); err != nil { + t.Fatal(err) + } + if stored[0].State != accepted || stored[1].State != uncertain { + t.Fatalf("states=%+v", stored) + } + if err := recoverInterrupted(engine); err != nil { + t.Fatal(err) + } +} + +func TestUncertainDATAIsNeverRetried(t *testing.T) { + engine, _, jobs := queueFixture(t) + job := jobs[0] + if ok, err := claim(engine, &job); err != nil || !ok { + t.Fatal(err) + } + if _, err := finish(engine, job, &smtp.DeliveryUncertainError{Cause: io.EOF}); err != nil { + t.Fatal(err) + } + if ok, err := claim(engine, &job); err != nil || ok { + t.Fatal("uncertain job was claimed again") + } +} + +func TestResultFailureRollsBackAndNeverRedelivers(t *testing.T) { + engine, email, jobs := queueFixture(t) + if _, err := engine.Exec("CREATE TRIGGER reject_result BEFORE UPDATE ON user_email BEGIN SELECT RAISE(ABORT, 'injected failure'); END"); err != nil { + t.Fatal(err) + } + var calls atomic.Int32 + r := &Runtime{engine: engine, deliver: func(_ *context.Context, e *parsemail.Email, recipient string) error { + calls.Add(1) + if e.MsgID != "stable@example.com" || len(e.To) != 2 || recipient != "one@example.com" { + t.Fatal("envelope changed message headers") + } + return nil + }} + if err := r.process(jobs[0]); err == nil { + t.Fatal("result failure was ignored") + } + if err := r.process(jobs[0]); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 { + t.Fatalf("SMTP called %d times", calls.Load()) + } + assertStatus(t, engine, email.Id, consts.EmailStatusDeliveryPending) + var job models.OutboundDelivery + if _, err := engine.ID(jobs[0].ID).Get(&job); err != nil { + t.Fatal(err) + } + if job.State != running { + t.Fatalf("state=%s", job.State) + } +} + +func TestQueuedJobsSurviveRestartAndDrain(t *testing.T) { + engine, email, _ := queueFixture(t) + if err := recoverInterrupted(engine); err != nil { + t.Fatal(err) + } + var calls atomic.Int32 + r := &Runtime{engine: engine, stop: make(chan struct{}), done: make(chan struct{}), deliver: func(_ *context.Context, _ *parsemail.Email, _ string) error { calls.Add(1); return nil }} + go r.loop() + deadline := time.Now().Add(3 * time.Second) + for calls.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + ctx, cancel := stdcontext.WithTimeout(stdcontext.Background(), time.Second) + defer cancel() + if err := r.Stop(ctx); err != nil { + t.Fatal(err) + } + if calls.Load() != 2 { + t.Fatalf("calls=%d", calls.Load()) + } + assertStatus(t, engine, email.Id, consts.EmailStatusSent) +} + +func TestRetryLimitAndSafeFailureDetails(t *testing.T) { + engine, _, jobs := queueFixture(t) + job := jobs[0] + for attempt := 1; attempt <= maxAttempts; attempt++ { + if _, err := engine.ID(job.ID).Cols("next_attempt").Update(&models.OutboundDelivery{}); err != nil { + t.Fatal(err) + } + if ok, err := claim(engine, &job); err != nil || !ok { + t.Fatalf("claim %d: %v", attempt, err) + } + before := time.Now().Unix() + if _, err := finish(engine, job, &textproto.Error{Code: 451, Msg: "sensitive-remote-response"}); err != nil { + t.Fatal(err) + } + var refreshed models.OutboundDelivery + if found, err := engine.ID(job.ID).Get(&refreshed); err != nil || !found { + t.Fatal(err) + } + job = refreshed + if strings.Contains(job.LastError, "sensitive") || !strings.Contains(job.LastError, "451") { + t.Fatalf("unsafe or missing failure details: %s", job.LastError) + } + if attempt < maxAttempts { + minimum := before + int64(60*(1<<(attempt-1))) + if job.State != queued || job.NextAttempt < minimum || job.NextAttempt > minimum+2 { + t.Fatalf("unexpected backoff: %+v", job) + } + } else if job.State != failed || job.NextAttempt != 0 { + t.Fatalf("retry limit ignored: %+v", job) + } + } + status, message, terminal := summarize([]models.OutboundDelivery{job}) + if status != consts.EmailStatusFail || !terminal || !strings.Contains(message, "451") { + t.Fatalf("failure summary: %d %s %v", status, message, terminal) + } +} + +func TestLegacyPendingIsUncertainWithoutEnqueue(t *testing.T) { + engine, _, _ := queueFixture(t) + email := models.Email{Type: consts.EmailTypeSend, Status: consts.EmailStatusDeliveryPending} + if _, err := engine.Insert(&email); err != nil { + t.Fatal(err) + } + if _, err := engine.Insert(&models.UserEmail{UserID: 1, EmailID: email.Id, Status: consts.EmailStatusDeliveryPending}); err != nil { + t.Fatal(err) + } + if err := recoverInterrupted(engine); err != nil { + t.Fatal(err) + } + assertStatus(t, engine, email.Id, consts.EmailStatusDeliveryUncertain) + count, err := engine.Where("email_id = ?", email.Id).Count(&models.OutboundDelivery{}) + if err != nil || count != 0 { + t.Fatalf("legacy message enqueued: %d %v", count, err) + } +} + +func TestReadFailureDoesNotClaimOrDeliver(t *testing.T) { + engine, _, jobs := queueFixture(t) + if _, err := engine.Exec("ALTER TABLE email RENAME TO unavailable_email"); err != nil { + t.Fatal(err) + } + r := &Runtime{engine: engine, deliver: func(_ *context.Context, _ *parsemail.Email, _ string) error { + t.Fatal("SMTP called despite read failure") + return nil + }} + if err := r.process(jobs[0]); err == nil { + t.Fatal("read failure ignored") + } + var job models.OutboundDelivery + if _, err := engine.ID(jobs[0].ID).Get(&job); err != nil { + t.Fatal(err) + } + if job.State != queued || job.Attempts != 0 { + t.Fatalf("read failure consumed attempt: %+v", job) + } +} + +func TestStopTimeoutDoesNotStartAnotherDelivery(t *testing.T) { + engine, _, _ := queueFixture(t) + entered, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + r := &Runtime{engine: engine, stop: make(chan struct{}), done: make(chan struct{}), deliver: func(_ *context.Context, _ *parsemail.Email, _ string) error { + if calls.Add(1) == 1 { + close(entered) + } + <-release + return nil + }} + go r.loop() + <-entered + ctx, cancel := stdcontext.WithTimeout(stdcontext.Background(), 20*time.Millisecond) + defer cancel() + err := r.Stop(ctx) + close(release) + if !errors.Is(err, stdcontext.DeadlineExceeded) { + t.Fatalf("stop did not time out: %v", err) + } + if err := r.Stop(stdcontext.Background()); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 { + t.Fatalf("delivery started after stop: %d", calls.Load()) + } +} + +func TestDeliveryDoesNotRestoreTrashedRelation(t *testing.T) { + engine, email, jobs := queueFixture(t) + if _, err := engine.Where("email_id = ?", email.Id).Cols("status").Update(&models.UserEmail{Status: consts.EmailStatusDel}); err != nil { + t.Fatal(err) + } + for _, job := range jobs { + if ok, err := claim(engine, &job); err != nil || !ok { + t.Fatal(err) + } + if _, err := finish(engine, job, nil); err != nil { + t.Fatal(err) + } + } + var relation models.UserEmail + if _, err := engine.Where("email_id = ?", email.Id).Get(&relation); err != nil { + t.Fatal(err) + } + if relation.Status != consts.EmailStatusDel { + t.Fatal("delivery restored trashed message") + } +} diff --git a/backend/utils/send/routing.go b/backend/utils/send/routing.go new file mode 100644 index 0000000..7cb7fa6 --- /dev/null +++ b/backend/utils/send/routing.go @@ -0,0 +1,93 @@ +package send + +import ( + stdcontext "context" + "errors" + "fmt" + "net" + "net/textproto" + "sort" + "strings" + "time" + + "git.pcux.cn/xuans/Smail/utils/smtp" +) + +var errNullMX = errors.New("收件域名声明不接收邮件") + +type mxLookup func(stdcontext.Context, string) ([]*net.MX, error) + +func resolveMX(domain string, lookup mxLookup) ([]string, error) { + ctx, cancel := stdcontext.WithTimeout(stdcontext.Background(), 10*time.Second) + defer cancel() + records, err := lookup(ctx, domain) + if err != nil { + var dnsErr *net.DNSError + if !errors.As(err, &dnsErr) || !dnsErr.IsNotFound || dnsErr.IsTemporary || dnsErr.IsTimeout { + return nil, fmt.Errorf("查询收件域名 MX 失败: %w", err) + } + // 无 MX 时使用域名自身的地址记录，不猜测 smtp 子域名。 + return []string{domain}, nil + } + if len(records) == 0 { + return []string{domain}, nil + } + records = append([]*net.MX(nil), records...) + for _, record := range records { + if record == nil || record.Host == "" { + return nil, errors.New("收件域名 MX 记录无效") + } + if record.Host == "." { + return nil, errNullMX + } + } + sort.SliceStable(records, func(i, j int) bool { return records[i].Pref < records[j].Pref }) + hosts := make([]string, 0, len(records)) + seen := make(map[string]bool) + for _, record := range records { + host := strings.ToLower(strings.TrimSuffix(record.Host, ".")) + if !seen[host] { + hosts = append(hosts, host) + seen[host] = true + } + } + return hosts, nil +} + +func stopFallback(err error) bool { + return smtp.IsDeliveryUncertain(err) || isCertificateVerificationError(err) || isPermanentSMTPResponse(err) || errors.Is(err, smtp.ErrUnsafeRemoteAddress) +} + +// deliverMX 只在确认本次未接收时切换安全通道，DATA 不确定结果不进入下一条路由。 +func deliverMX(hosts []string, attempt func(host, port string) error) error { + var lastErr error + for _, host := range hosts { + for _, port := range []string{"25", "587", "465"} { + lastErr = attempt(host, port) + if lastErr == nil || stopFallback(lastErr) { + return lastErr + } + } + } + if lastErr == nil { + lastErr = errors.New("收件域名没有可用邮件服务器") + } + return lastErr +} + +// Retryable 仅允许有明确临时失败语义的结果进入持久队列重试。 +func Retryable(err error) bool { + if err == nil || stopFallback(err) { + return false + } + var response *textproto.Error + if errors.As(err, &response) { + return response.Code >= 400 && response.Code <= 499 + } + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return dnsErr.IsTimeout || dnsErr.IsTemporary + } + var networkErr net.Error + return errors.As(err, &networkErr) +} diff --git a/backend/utils/send/routing_test.go b/backend/utils/send/routing_test.go new file mode 100644 index 0000000..9215f60 --- /dev/null +++ b/backend/utils/send/routing_test.go @@ -0,0 +1,95 @@ +package send + +import ( + "context" + "crypto/tls" + "errors" + "io" + "net" + "net/textproto" + "reflect" + "testing" + + "git.pcux.cn/xuans/Smail/utils/smtp" +) + +func TestResolveMX(t *testing.T) { + for _, tt := range []struct { + name string + records []*net.MX + err error + want []string + wantErr bool + }{ + {name: "priority and duplicate", records: []*net.MX{{Host: "backup.example.", Pref: 20}, {Host: "primary.example.", Pref: 10}, {Host: "PRIMARY.example.", Pref: 30}}, want: []string{"primary.example", "backup.example"}}, + {name: "implicit MX", want: []string{"example.com"}}, + {name: "no MX record", err: &net.DNSError{IsNotFound: true}, want: []string{"example.com"}}, + {name: "temporary DNS", err: &net.DNSError{IsTemporary: true}, wantErr: true}, + {name: "timeout DNS", err: &net.DNSError{IsTimeout: true}, wantErr: true}, + {name: "null MX", records: []*net.MX{{Host: "."}}, wantErr: true}, + {name: "invalid MX", records: []*net.MX{nil}, wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + hosts, err := resolveMX("example.com", func(ctx context.Context, domain string) ([]*net.MX, error) { + if _, ok := ctx.Deadline(); !ok || domain != "example.com" { + t.Fatal("unbounded or wrong lookup") + } + return tt.records, tt.err + }) + if (err != nil) != tt.wantErr || !reflect.DeepEqual(hosts, tt.want) { + t.Fatalf("hosts=%v error=%v", hosts, err) + } + }) + } +} + +func TestDeliverMXFallbackAndStop(t *testing.T) { + for _, tt := range []struct { + name string + err error + calls int + }{ + {"network fallback", &net.OpError{Op: "dial", Err: io.EOF}, 4}, + {"temporary refusal fallback", &textproto.Error{Code: 451, Msg: "later"}, 4}, + {"permanent refusal", &textproto.Error{Code: 550, Msg: "rejected"}, 1}, + {"certificate failure", &tls.CertificateVerificationError{Err: io.EOF}, 1}, + {"uncertain DATA", &smtp.DeliveryUncertainError{Cause: io.EOF}, 1}, + {"private MX", smtp.ErrUnsafeRemoteAddress, 1}, + } { + t.Run(tt.name, func(t *testing.T) { + var calls []string + err := deliverMX([]string{"primary.example", "backup.example"}, func(host, port string) error { + calls = append(calls, host+":"+port) + if host == "backup.example" { + return nil + } + return tt.err + }) + if len(calls) != tt.calls { + t.Fatalf("calls=%v", calls) + } + if tt.calls == 4 && err != nil { + t.Fatal(err) + } + if tt.calls == 1 && !errors.Is(err, tt.err) { + t.Fatalf("error=%v", err) + } + }) + } +} + +func TestRetryable(t *testing.T) { + for _, tt := range []struct { + err error + want bool + }{ + {&textproto.Error{Code: 451}, true}, {&net.DNSError{IsTemporary: true}, true}, + {&net.DNSError{IsNotFound: true}, false}, {&textproto.Error{Code: 550}, false}, + {&smtp.DeliveryUncertainError{Cause: &net.DNSError{IsTimeout: true}}, false}, + {&tls.CertificateVerificationError{Err: io.EOF}, false}, {nil, false}, + } { + if got := Retryable(tt.err); got != tt.want { + t.Fatalf("Retryable(%v)=%v", tt.err, got) + } + } +} diff --git a/backend/utils/send/send.go b/backend/utils/send/send.go index 58a08a2..3fdc613 100644 --- a/backend/utils/send/send.go +++ b/backend/utils/send/send.go @@ -9,7 +9,6 @@ import ( "git.pcux.cn/xuans/Smail/models" "git.pcux.cn/xuans/Smail/services/logstore" "git.pcux.cn/xuans/Smail/utils/array" - "git.pcux.cn/xuans/Smail/utils/async" "git.pcux.cn/xuans/Smail/utils/consts" "git.pcux.cn/xuans/Smail/utils/context" "git.pcux.cn/xuans/Smail/utils/smtp" @@ -89,6 +88,23 @@ func Send(ctx *context.Context, e *parsemail.Email) (error, map[string]error) { // SendTriggered 显式标记业务发起方，不从邮件主题或内容推断来源。 func SendTriggered(ctx *context.Context, e *parsemail.Email, trigger string) (error, map[string]error) { + return sendRecipients(ctx, e, nil, trigger) +} + +// SendRecipientTriggered 保留原邮件头，只向当前队列收件人发送 SMTP 信封。 +func SendRecipientTriggered(ctx *context.Context, e *parsemail.Email, recipient string) error { + err, failures := sendRecipients(ctx, e, []*parsemail.User{{EmailAddress: recipient}}, "网页写信") + if err != nil { + for _, cause := range failures { + if cause != nil { + return cause + } + } + } + return err +} + +func sendRecipients(ctx *context.Context, e *parsemail.Email, recipients []*parsemail.User, trigger string) (error, map[string]error) { _, fromDomain := e.From.GetDomainAccount() b, err := e.BuildBytes(ctx, true) @@ -104,6 +120,9 @@ func SendTriggered(ctx *context.Context, e *parsemail.Email, trigger string) (er var to []*parsemail.User to = append(append(append(to, e.To...), e.Cc...), e.Bcc...) + if recipients != nil { + to = recipients + } return doSend(ctx, fromDomain, b, to, e.From.EmailAddress, trigger) @@ -176,8 +195,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai invalidErrors := make(map[string]error) // 按域名整理 - toByDomain := map[mxDomain][]*parsemail.User{} - mxLookupErrors := map[mxDomain]error{} + toByDomain := map[string][]*parsemail.User{} for _, s := range to { address, domain, addressErr := parseRecipientAddress(s) if addressErr != nil { @@ -189,41 +207,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai log.WithContext(ctx).WithError(addressErr).Error("invalid recipient address") continue } - args := []string{address, domain} - if len(args) == 2 { - if args[1] == consts.TEST_DOMAIN { - // 测试使用 - address := mxDomain{ - domain: "localhost", - mxHost: "127.0.0.1", - localTest: true, - } - toByDomain[address] = append(toByDomain[address], s) - } else { - //查询dns mx记录 - mxInfo, lookupErr := net.LookupMX(args[1]) - address := mxDomain{ - domain: "smtp." + args[1], - mxHost: "smtp." + args[1], - } - if lookupErr != nil { - log.WithContext(ctx).Errorf("%s 域名mx记录查询失败，检查邮箱是否存在！", s.EmailAddress) - } - if len(mxInfo) > 0 { - address = mxDomain{ - domain: args[1], - mxHost: mxInfo[0].Host, - } - } - if lookupErr != nil { - mxLookupErrors[address] = lookupErr - } - toByDomain[address] = append(toByDomain[address], s) - } - } else { - log.WithContext(ctx).Errorf("邮箱地址解析错误！ %s", s) - continue - } + toByDomain[domain] = append(toByDomain[domain], s) } errEmailAddress := append([]string(nil), invalidAddresses...) @@ -234,14 +218,14 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai errMap.Store(address, err) } - as := async.New(ctx) + var workers sync.WaitGroup for domain, tos := range toByDomain { domain := domain tos := tos - mxLookupErr := mxLookupErrors[domain] - as.WaitProcess(func(p any) { + workers.Add(1) + go func() { + defer workers.Done() recordFailure := func(err error) { - err = deliveryFailureCause(mxLookupErr, err) log.WithContext(ctx).Errorf("%v 邮件投递失败%+v", tos, err) errEmailAddressMu.Lock() @@ -250,11 +234,16 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai } errEmailAddressMu.Unlock() - errMap.Store(domain.domain, err) + errMap.Store(domain, err) } + defer func() { + if value := recover(); value != nil { + recordFailure(&smtp.DeliveryUncertainError{Cause: fmt.Errorf("投递任务异常: %v", value)}) + } + }() - if allowUnsafeSMTP(domain) { - err := smtp.SendMailUnsafe("localhost", domain.mxHost+":25", nil, from, fromDomain, buildAddress(tos), data) + if domain == consts.TEST_DOMAIN { + err := smtp.SendMailUnsafe("localhost", "127.0.0.1:25", nil, from, fromDomain, buildAddress(tos), data) if err != nil { recordFailure(err) } else { @@ -263,71 +252,26 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai return } - // 优先尝试25端口，starttls方式投递 - err := smtp.SendMail("", domain.mxHost+":25", nil, from, fromDomain, buildAddress(tos), data) - if err == nil { + hosts, err := resolveMX(domain, net.DefaultResolver.LookupMX) + if err != nil { + recordFailure(err) + return + } + err = deliverMX(hosts, func(host, port string) error { + address := net.JoinHostPort(host, port) + if port == "465" { + return smtp.SendMailWithTls(host, address, nil, from, fromDomain, buildAddress(tos), data) + } + return smtp.SendMail(host, address, nil, from, fromDomain, buildAddress(tos), data) + }) + if err != nil { + recordFailure(err) + } else { accepted.Add(int64(len(tos))) - return } - if isCertificateVerificationError(err) { - recordFailure(err) - return - } - if isPermanentSMTPResponse(err) { - recordFailure(err) - return - } - log.WithContext(ctx).WithFields(log.Fields{ - "action": "SMTP 邮件投递", - "result": "failure", - "object": domain.mxHost + ":25", - }).Infof("SMTP 25 端口 STARTTLS 投递失败: %s", err.Error()) - - // 587 使用 STARTTLS，只有 465 使用隐式 TLS。 - err = smtp.SendMail(strings.TrimSuffix(domain.mxHost, "."), domain.mxHost+":587", nil, from, fromDomain, buildAddress(tos), data) - if err == nil { - accepted.Add(int64(len(tos))) - return - } - if isCertificateVerificationError(err) { - recordFailure(err) - return - } - if isPermanentSMTPResponse(err) { - recordFailure(err) - return - } - log.WithContext(ctx).WithFields(log.Fields{ - "action": "SMTP 邮件投递", - "result": "failure", - "object": domain.mxHost + ":587", - }).Infof("SMTP 587 端口投递失败: %s", err.Error()) - - // 再次尝试465投递 - err = smtp.SendMailWithTls(strings.TrimSuffix(domain.mxHost, "."), domain.mxHost+":465", nil, from, fromDomain, buildAddress(tos), data) - if err == nil { - accepted.Add(int64(len(tos))) - return - } - if isCertificateVerificationError(err) { - recordFailure(err) - return - } - if isPermanentSMTPResponse(err) { - recordFailure(err) - return - } - log.WithContext(ctx).WithFields(log.Fields{ - "action": "SMTP 邮件投递", - "result": "failure", - "object": domain.mxHost + ":465", - }).Infof("SMTP 465 端口投递失败: %s", err.Error()) - - // 所有安全投递方式均失败后记录错误，不允许回退到明文连接。 - recordFailure(err) - }, nil) + }() } - as.Wait() + workers.Wait() orgMap := map[string]error{} errMap.Range(func(key, value any) bool { diff --git a/backend/utils/smtp/smtp.go b/backend/utils/smtp/smtp.go index b3d7c8a..9d2f6b0 100644 --- a/backend/utils/smtp/smtp.go +++ b/backend/utils/smtp/smtp.go @@ -45,6 +45,22 @@ var EOFError = errors.New("EOF") var ErrUnsafeRemoteAddress = errors.New("smtp: remote address is not public") var ErrUnsafePlaintextTarget = errors.New("smtp: plaintext delivery is restricted to localhost") +// DeliveryUncertainError 表示 DATA 已开始，但未取得明确接收或拒绝响应，禁止自动重投。 +type DeliveryUncertainError struct { + Cause error +} + +func (e *DeliveryUncertainError) Error() string { + return "投递结果不确定，请核对收件结果后再操作: " + e.Cause.Error() +} + +func (e *DeliveryUncertainError) Unwrap() error { return e.Cause } + +func IsDeliveryUncertain(err error) bool { + var uncertain *DeliveryUncertainError + return errors.As(err, &uncertain) +} + const ( outboundSMTPDialTimeout = 2 * time.Second outboundSMTPTLSHandshakeTimeout = 2 * time.Second @@ -554,10 +570,14 @@ func (c *Client) Data() (io.WriteCloser, error) { func finishDelivery(c *Client, w io.WriteCloser, msg []byte) error { if _, err := w.Write(msg); err != nil { - return err + return &DeliveryUncertainError{Cause: err} } if err := w.Close(); err != nil { - return err + var rejection *textproto.Error + if errors.As(err, &rejection) && rejection.Code >= 400 && rejection.Code <= 599 { + return err + } + return &DeliveryUncertainError{Cause: err} } // The final 250 response to DATA transfers responsibility for the message. diff --git a/backend/utils/smtp/smtp_test.go b/backend/utils/smtp/smtp_test.go index 69dd855..b1d2434 100644 --- a/backend/utils/smtp/smtp_test.go +++ b/backend/utils/smtp/smtp_test.go @@ -219,6 +219,10 @@ func startSMTPTransactionServer(t *testing.T, dataResponse string, holdWithoutQu done <- fmt.Errorf("message body was not received: %q", body) return } + if dataResponse == "" { + done <- nil + return + } if err := text.PrintfLine("%s", dataResponse); err != nil { done <- err return @@ -244,6 +248,17 @@ func startSMTPTransactionServer(t *testing.T, dataResponse string, holdWithoutQu return listener.Addr().String(), done } +func TestLostFinalDATAResponseIsUncertain(t *testing.T) { + addr, done := startSMTPTransactionServer(t, "", false) + err := SendMailUnsafe("localhost", addr, nil, "sender@example.com", "example.com", []string{"recipient@example.net"}, []byte("Subject: test\r\n\r\nbody\r\n")) + if !IsDeliveryUncertain(err) { + t.Fatalf("error=%v, want uncertain delivery", err) + } + if err := <-done; err != nil { + t.Fatal(err) + } +} + func expectSMTPCommand(conn *textproto.Conn, prefix string) error { line, err := conn.ReadLine() if err != nil { diff --git a/docs/生产整改验收清单.md b/docs/生产整改验收清单.md new file mode 100644 index 0000000..2f35cc0 --- /dev/null +++ b/docs/生产整改验收清单.md @@ -0,0 +1,63 @@ +# 生产整改验收清单 + +## 范围与结论 + +本清单对应 2026 年 9 月 17 日 `D:\Desktop\Smail` 生产审查整改，验收服务器为 `47.108.78.12`。应用和自动测试只在该服务器执行，SMTP 按用户要求使用远程 `127.0.0.1:25`，不将公网 25 端口连通列为本轮通过条件。 + +当前版本已经完成凭据配置、SMTP 判定、MX 路由和网页持久投递队列的代码整改及针对性远程测试，但仍不建议直接生产上线。测试范围、版本摘要和备份位置以《远程服务器测试记录》最新章节为准。 + +## 任务状态 + +| 编号 | 优先级 | 任务 | 状态 | 验收边界 | +| --- | --- | --- | --- | --- | +| PROD-001 | P0 | 移除脚本中的已知明文凭据 | [x] | 工作区已清理，历史泄露处置未结束 | +| PROD-002 | P0 | 轮换泄露凭据、核查访问记录 | [!] | 旧 SSH 密码仍有效，需要协调替代登录及轮换 | +| PROD-003 | P0 | 处置 Git 历史中的凭据 | [!] | 未授权重写历史或强推；轮换应先于历史清理 | +| PROD-004 | P0 | 持久化逐收件人投递、防不确定重投 | [x] | 相关自动测试通过，真实网页业务闭环待验收 | +| PROD-005 | P1 | MX、备用路由、DNS 错误分类 | [✓] | 针对性测试通过，不代表各外部邮件供应商兼容性 | +| PROD-006 | P0 | 可信 TLS 和正式域名验证 | [!] | 仍使用自签名测试证书，不关闭证书校验 | +| PROD-007 | P0 | 完整网页登录、发信和状态核对 | [!] | 需要有效测试账号和可信证书；不得绕过登录 | +| PROD-008 | P0 | 超级管理员恢复完整交互 | [!] | IPC 权限已复核，正式密码代理交互及旧会话撤销未验收 | +| PROD-009 | P1 | 备份与恢复演练 | [x] | 完整备份、归档及双库校验通过；异机恢复未演练 | +| PROD-010 | P1 | 内部 SMTP 与部署后安全回归 | [✓] | 五项真实收信、双库、产物、未认证 API 均通过 | +| PROD-011 | P1 | 全量测试、竞态、数据库兼容性和压测 | [ ] | 本轮仅七包 132 个顶层用例；未运行真实 MySQL/PostgreSQL | +| PROD-012 | P1 | 最新版本清库首次安装验收 | [ ] | 当前为保留数据升级，不能沿用旧版本清库结果 | + +状态含义：`[ ]` 未开始，`[~]` 进行中，`[x]` 已完成待复审，`[✓]` 指定范围复审通过，`[!]` 暂缓或存在阻塞。 + +## 凭据与 SSH + +SSH 脚本复用 `test/tools/remote_ssh.py`，强制核对已知主机密钥。不得为连接方便改回自动接受未知主机，也不要把真实值写入命令记录、仓库或文档。 + +| 环境变量 | 用途 | +| --- | --- | +| `SMAIL_SSH_HOST`、`SMAIL_SSH_PORT` | SSH 主机和端口 | +| `SMAIL_SSH_USER` | 登录账号 | +| `SMAIL_SSH_PASSWORD` | 临时注入的密码，优先协调迁移到密钥认证 | +| `SMAIL_SSH_KEY` | 私钥路径，不是私钥内容 | +| `SMAIL_SSH_KNOWN_HOSTS` | 额外的已核验主机密钥文件路径 | +| `SMAIL_TEST_ADMIN_PASSWORD` | 测试安装时使用的管理员密码 | +| `SMAIL_TEST_SMTP_PASSWORD` | SMTP 示例客户端凭据 | + +轮换前必须验证新的独立登录路径，避免锁死服务器访问；还应撤销关联会话、令牌及重复使用该密码的账号凭据，并审查异常登录。历史清理需要单独授权、备份及协作安排，当前未删除文件、重写历史或强推。删除源码中的密码不会让旧密码失效。 + +## 投递语义与运维边界 + +- 网页返回提交成功表示邮件和任务已经持久化，不表示收件箱已收到。SMTP 250 仅表示对端接收了投递责任。 +- 每收件人独立记录 `queued`、`running`、`accepted`、`failed` 或 `uncertain`。成功收件人不随其他收件人的失败重投。 +- 临时失败最多五次，间隔 1、2、4、8 分钟；明确拒绝、证书问题、危险目标和结果不确定停止自动重试。 +- DATA 结果不确定、重启时仍在途、旧版本没有可靠尝试记录的邮件必须先核对实际收件结果，再决定人工处理。当前没有已验收的人工重投界面，不要直接改库重排。 +- 数据库读取失败不领取任务；结果写库失败只重试写入三次，仍失败则保留在途状态并记录错误。恢复数据库后需要核查日志及受控重启，不能把它理解成无限自动恢复。 +- 当前仅允许单实例、单工作器使用同一队列库。没有验收多实例调度、吞吐容量、队列告警和故障期间的服务等级。 +- 现有邮件客户端 SMTP 提交、系统通知和其他调用方不自动等同于网页持久队列；分别验收其投递及失败处理。 +- 备用 MX 保留既有 25 STARTTLS、587 STARTTLS、465 TLS 顺序；禁止明文降级。各运营商的端口策略及可达性需要独立验证。 + +## 下一轮验收顺序 + +1. 验证替代 SSH 登录，完成泄露凭据轮换和访问审计，再另行协调历史清理。 +2. 部署覆盖实际域名的受信任证书，使用有效测试账号恢复桌面浏览器验收；移动端遵循指定插件流程。 +3. 从正常网页提交邮件，核对持久任务、部分成功、最终状态和实际收件结果；对重启、暂时拒绝、DATA 断连及数据库故障使用受控远程故障环境，不影响其他服务。 +4. 用受保护的本地管理命令完成恢复交互，核对新密码登录、旧密码拒绝、旧会话及客户端凭据撤销。不得增加普通 HTTP 或公网/LAN TCP 恢复入口。 +5. 在隔离远程目标完成备份恢复、全量与竞态测试、目标数据库兼容性和容量验证。最新版本重新清库安装前，先备份所有项目数据库并核实范围，不清理其他项目数据。 + +上述阻塞项解决后再做最终生产评审，不以构建成功、首页 200、内部 SMTP 250 或单次提交代替整体验收。 diff --git a/docs/远程服务器测试记录.md b/docs/远程服务器测试记录.md index 6c69af1..c72a90f 100644 --- a/docs/远程服务器测试记录.md +++ b/docs/远程服务器测试记录.md @@ -158,3 +158,76 @@ TLS 前未公布 AUTH，主动尝试明文 AUTH 返回 523。未认证外域邮 - Chrome DevTools 访问远程站点时被现有自签名证书拦截，停留在证书错误页；这与此前记录的受信任 TLS 阻塞一致，不能据此宣称网页渲染或真实点击验收通过。 因此，本轮审查确认服务、部署产物、数据库完整性及未认证接口边界正常；批量已读的真实网页登录、选择邮件、点击“标记为已读”、刷新页面和混合归属网页请求仍保持阻塞，待提供有效测试账号及受信任测试证书后继续。数据库没有执行清理，历史测试邮件和备份继续保留。 + +## 2026 年 9 月 17 日生产整改升级回归 + +本节记录后续生产整改版本，不覆盖前文历史证据。本轮应用及测试二进制仍只在 `47.108.78.12` 运行；本机仅进行源码检查、前端构建、Linux 交叉编译和 SSH 操作。本轮采用备份后保留双库升级，没有清库，不作为重新安装验收。 + +### 整改内容 + +- 测试及部署脚本统一读取环境变量中的 SSH 凭据，校验已知主机密钥，不再自动信任未知主机。网页安装及 SMTP 示例密码也改为环境变量输入。 +- 网页发信在同一事务内保存邮件、用户关联及每收件人的持久投递任务。进程重启后，尚未尝试的任务可以继续；已在途任务及旧版无队列的待投递邮件标记为“投递结果待核对”，不盲目重投。 +- 临时网络、DNS 或 SMTP 4xx 失败最多尝试五次，间隔为 1、2、4、8 分钟。明确拒绝、证书错误、私网目标和 DATA 结果不确定均不自动重试。 +- MX 按优先级排序并去重，支持备用 MX、无 MX 时使用域名自身及 Null MX 拒绝；保留既有安全端口回退。邮件头保留完整收件信息，队列任务只投递当前收件人的 SMTP 信封。 +- SMTP DATA 最终 250 后的 QUIT 异常不再误判失败；发送 DATA 后失去明确结果的任务暂停核对。结果写库失败只重试数据库写入，不重复调用 SMTP。 +- 后台投递接入在线数据库迁移排空边界；停机等待投递结束，不提前关闭仍在使用的数据库。投递状态更新不恢复已放入回收站的用户关联。 + +### 自动测试 + +七个相关 Linux 测试包在远程以 `smail` 账号执行，使用隔离内存或临时数据库，不改部署业务库。共 132 个顶层测试用例通过，日志保留在 `/var/tmp/smail-review-20260917/*.result.log`。 + +| 测试包 | 顶层用例数 | 结果 | +| --- | --- | --- | +| `outbox` | 10 | 通过 | +| `utils-send` | 11 | 通过 | +| `utils-smtp` | 8 | 通过 | +| `controllers-email` | 12 | 通过 | +| `db-migration` | 26 | 通过 | +| `res_init` | 14 | 通过 | +| `services-dbmigration` | 51 | 通过 | + +新增覆盖包括部分投递成功、防重复领取、退避及重试上限、DATA 不确定结果、结果写库故障、读取失败、旧待投递记录、重启恢复、停机超时、回收站状态、队列插入失败整事务回滚和迁移等待后台投递。这里的测试通过不等于真实网页登录发信闭环已通过。 + +前端生产构建和 Linux amd64 编译通过，仍有编辑器分块超过 500 kB 的构建提示。没有重新执行全部后端包、竞态检测、移动端、真实 MySQL 或 PostgreSQL 验证。 + +### 部署证据 + +2026 年 9 月 17 日 10:06 完成双程序升级，版本为 `remote-13bcad90e832`，源码指纹为 `13bcad90e832a9be78604854a1c91f72cfaf2a21770481c627e50ac361ce3b70`。 + +| 项目 | 结果 | +| --- | --- | +| `/opt/smail/smail` SHA-256 | `a3b4a0b0e75213bb7997fa55f789d98bca027e879c9dbfe986f7b1ee8c8230d9` | +| `/opt/smail/smail-admin` SHA-256 | `16e48a16d3f47ab78b42fb1aa854dfc32ec2ccd3a95694c6ccb93fc1da075e34` | +| 数据库迁移版本 | 23 | +| 两份 SQLite 完整性 | 均为 `ok` | +| 服务状态 | `active/running`，`NRestarts=0`，`ExecMainStatus=0` | +| 本地恢复 socket | `/run/smail/admin-control.sock`，`smail:smail 0600` | +| HTTPS 首页 | 显式测试 CA、主机名校验下返回 200 | +| 未认证用户信息、邮件列表接口 | 均返回业务码 403，无业务数据 | +| 本次检查时段日志 | 25 条 journal 记录，未发现 error、fatal 或 panic 标记 | + +完整停服备份为 `/var/backups/smail-before-outbox-20260917-100608`，归档 SHA-256 为 `a2ed56666e9e691a63dfc6734ebe2a1e9c7bf2fe4ceb81cb55028e66ea52d016`；包含原主程序、管理员工具、双库及完整部署归档。归档比对和数据库备份完整性检查通过，没有删除历史文件。此次部署成功，未实际演练失败回滚或异机恢复。 + +### 内部 SMTP 回归 + +运行号 `20260917-100912`，仅从远程连接 `127.0.0.1:25`。五项均使用 TLS 1.3，DATA 返回 250，并核对真实数据库归属、正文和附件字节： + +| 用例 | 邮件 ID | 结果 | +| --- | --- | --- | +| 中文纯文本 | 27 | 仅普通用户关联，正文一致 | +| HTML、中文附件及多收件人 | 28 | 管理员和普通用户关联，附件一致 | +| 管理员专属附件 | 29 | 仅管理员关联 | +| 未知本域收件人 | 30 | 过滤级别 0 下转管理员收件箱 | +| 未认证外域收件人 | 31 | 转管理员收件箱，无发信记录 | + +结果及测试前双库备份位于 `/var/backups/smail-internal-smtp-20260917-100912`。没有修改过滤策略、系统 CA、证书验证或 DNS；客户端仅显式信任已有测试证书。内部接收链路通过，不代表网页发信、外部投递或生产 TLS 验收通过。 + +### 结论与限制 + +- [✓] 本轮指定的七包测试、部署产物核对、双库完整性、内部 SMTP 和未认证接口边界复审通过。 +- [x] 凭据硬编码清理完成；工作区已知旧 SSH 密码扫描零命中，但旧密码仍有效，Git 历史仍含旧凭据，不能视为泄露处置完成。 +- [!] 新队列尚未完成真实网页登录发信闭环；部署后的队列表为空，不能用入站 SMTP 回归代替出站队列业务验收。 +- [!] 当前队列按单实例、单工作器运行，未压测。结果落库持续失败时保留在途任务，需要告警和受控重启后转为待核对；不得人工直接改回排队状态。 +- [!] 受信任 TLS、凭据轮换、正式管理员恢复交互、新密码登录与旧会话撤销、灾备恢复及全量兼容性验证仍未完成。 + +当前结论仍为：适合继续受控远程验收，不建议直接用于正式生产。后续门槛见《生产整改验收清单》。 diff --git a/frontend/src/views/List/List.vue b/frontend/src/views/List/List.vue index 00175cb..48c0b06 100644 --- a/frontend/src/views/List/List.vue +++ b/frontend/src/views/List/List.vue @@ -81,6 +81,11 @@ 待投递 + + + 投递结果待核对 + + 发送成功 @@ -115,7 +120,7 @@ - +

错误原因：

diff --git a/test/manual/smtp/sendEmailTest.py b/test/manual/smtp/sendEmailTest.py index c275c94..3f25a58 100644 --- a/test/manual/smtp/sendEmailTest.py +++ b/test/manual/smtp/sendEmailTest.py @@ -1,10 +1,11 @@ from email.mime.text import MIMEText import smtplib +import os msg = MIMEText('hello, send by Python...', 'plain', 'utf-8') from_addr = "admin@domain.com" -password = "admin" +password = os.environ["SMAIL_TEST_SMTP_PASSWORD"] to_addr = "admin@domain.com" smtp_server = "127.0.0.1" @@ -12,4 +13,4 @@ server = smtplib.SMTP(smtp_server, 25) server.starttls() server.login(from_addr, password) server.sendmail(from_addr, [to_addr], msg.as_string()) -server.quit() \ No newline at end of file +server.quit() diff --git a/test/run_deploy.py b/test/run_deploy.py index cf3d216..e07a86b 100644 --- a/test/run_deploy.py +++ b/test/run_deploy.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from tools.remote_ssh import connect_remote +connect_remote(ssh) sftp = ssh.open_sftp() print('Uploading out/smail to /opt/smail/smail.new ...') sftp.put('out/smail', '/opt/smail/smail.new') diff --git a/test/tools/add_dns.py b/test/tools/add_dns.py index f4b46ca..bfd07a4 100644 --- a/test/tools/add_dns.py +++ b/test/tools/add_dns.py @@ -1,8 +1,9 @@ import urllib.request, urllib.parse, hmac, hashlib, base64, time, json, paramiko, subprocess client = paramiko.SSHClient() -client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -client.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +client.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(client) stdin, stdout, stderr = client.exec_command('cat /opt/smail/ssl/dkim/dkim.priv') priv_key = stdout.read().decode('utf-8') client.close() diff --git a/test/tools/cat_remote_configs.py b/test/tools/cat_remote_configs.py index 84f9988..c2abe5a 100644 --- a/test/tools/cat_remote_configs.py +++ b/test/tools/cat_remote_configs.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) _, out, _ = ssh.exec_command('cat /opt/smail/config/smtp.json; echo ===; cat /opt/smail/config/email.json') print(out.read().decode('utf-8', errors='ignore')) ssh.close() diff --git a/test/tools/check_auth_codes.py b/test/tools/check_auth_codes.py index 0436ff6..c584fc5 100644 --- a/test/tools/check_auth_codes.py +++ b/test/tools/check_auth_codes.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command("sqlite3 /opt/smail/data/smail.db \"SELECT * FROM client_credential;\"") print('client_credential:\n', stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command("sqlite3 /opt/smail/data/smail.db \"SELECT * FROM protocol_credential;\"") diff --git a/test/tools/check_db_full.py b/test/tools/check_db_full.py index 1507a45..d6ef444 100644 --- a/test/tools/check_db_full.py +++ b/test/tools/check_db_full.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, type, subject, from_address, send_user_id, status, error, create_time FROM email WHERE id >= 20;"') print("Security questions:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, user_id, email_id, is_read, group_id, status, create FROM user_email WHERE email_id >= 20;"') diff --git a/test/tools/check_emails_10.py b/test/tools/check_emails_10.py index a4f29d2..4fc0aa5 100644 --- a/test/tools/check_emails_10.py +++ b/test/tools/check_emails_10.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, type, subject, from_address, [to], send_user_id, status FROM email WHERE id >= 10;"') print("Emails >= 10:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, user_id, email_id, is_read, group_id, status FROM user_email WHERE email_id >= 10;"') diff --git a/test/tools/check_err24.py b/test/tools/check_err24.py index d029cc9..7575b02 100644 --- a/test/tools/check_err24.py +++ b/test/tools/check_err24.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, error FROM email WHERE id = 24;"') print("Email 24 error:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT * FROM email WHERE id = 24;"') diff --git a/test/tools/check_external.py b/test/tools/check_external.py index 299cc3b..a1476df 100644 --- a/test/tools/check_external.py +++ b/test/tools/check_external.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command("sqlite3 /opt/smail/data/smail.db \"SELECT sql FROM sqlite_master WHERE name='external_email';\"") print('SCHEMA:\n', stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command("sqlite3 /opt/smail/data/smail.db \"SELECT * FROM external_email;\"") diff --git a/test/tools/check_inbox_query.py b/test/tools/check_inbox_query.py index d69b109..bde97cf 100644 --- a/test/tools/check_inbox_query.py +++ b/test/tools/check_inbox_query.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT e.id, e.subject, e.type, ue.status, ue.group_id, ue.is_read FROM email e LEFT JOIN user_email ue ON e.id = ue.email_id WHERE ue.user_id = 1 AND ue.status = 0 AND e.type = 0 AND (ue.group_id = 0 OR ue.group_id = 1) ORDER BY e.id DESC;"') print("Inbox query result:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT e.id, e.subject, e.type, e.status, e.error FROM email e WHERE e.send_user_id = 1 ORDER BY e.id DESC LIMIT 5;"') diff --git a/test/tools/check_log.py b/test/tools/check_log.py index ab29699..64c7836 100644 --- a/test/tools/check_log.py +++ b/test/tools/check_log.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command("tail -n 25 /opt/smail/log/stdout.log") print(stdout.read().decode("utf-8", errors="ignore")) stdin, stdout, stderr = ssh.exec_command("ps aux | grep smail") diff --git a/test/tools/check_log_send.py b/test/tools/check_log_send.py index 220428f..6d26368 100644 --- a/test/tools/check_log_send.py +++ b/test/tools/check_log_send.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('tail -n 40 /opt/smail/log/runtime-2026-09-14.log') print("Grep 6567567:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('tail -n 40 /opt/smail/log/stdout.log') diff --git a/test/tools/check_remote_db.py b/test/tools/check_remote_db.py index ba5e905..48231fd 100644 --- a/test/tools/check_remote_db.py +++ b/test/tools/check_remote_db.py @@ -1,7 +1,8 @@ import paramiko client = paramiko.SSHClient() -client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -client.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=8) +client.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(client) cmd = 'systemctl list-unit-files | grep -E "(mysql|mariadb|postgresql)"' stdin, stdout, stderr = client.exec_command(cmd) print("DB units:\n", stdout.read().decode('utf-8')) diff --git a/test/tools/check_remote_services.py b/test/tools/check_remote_services.py index 6ada672..c8a38c6 100644 --- a/test/tools/check_remote_services.py +++ b/test/tools/check_remote_services.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) _, out, _ = ssh.exec_command('ss -tulpn; ps aux | grep smail') print(out.read().decode('utf-8', errors='ignore')) ssh.close() \ No newline at end of file diff --git a/test/tools/check_remote_status.py b/test/tools/check_remote_status.py index 5e075fa..04a6727 100644 --- a/test/tools/check_remote_status.py +++ b/test/tools/check_remote_status.py @@ -1,8 +1,9 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!") +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) sftp = ssh.open_sftp() with sftp.open("/tmp/test_api.py", "w") as f: diff --git a/test/tools/check_row22.py b/test/tools/check_row22.py index 76a12a8..26c4deb 100644 --- a/test/tools/check_row22.py +++ b/test/tools/check_row22.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "date"') print("Security questions:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "date -u"') diff --git a/test/tools/check_srv.py b/test/tools/check_srv.py index 455edf8..7401c92 100644 --- a/test/tools/check_srv.py +++ b/test/tools/check_srv.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('date') print("Server Date:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('uptime') diff --git a/test/tools/check_ue.py b/test/tools/check_ue.py index d648dcc..f4011cc 100644 --- a/test/tools/check_ue.py +++ b/test/tools/check_ue.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT user_id, email_id FROM user_email WHERE email_id = 22;"') print("Security questions:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT user_id, email_id FROM user_email WHERE email_id = 23;"') diff --git a/test/tools/debug_ssh.py b/test/tools/debug_ssh.py index 9cdb79f..8ca519f 100644 --- a/test/tools/debug_ssh.py +++ b/test/tools/debug_ssh.py @@ -1,8 +1,9 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) def run(cmd): stdin, stdout, stderr = ssh.exec_command(cmd) diff --git a/test/tools/decrypt_external.py b/test/tools/decrypt_external.py index fa883be..4b06768 100644 --- a/test/tools/decrypt_external.py +++ b/test/tools/decrypt_external.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = '''import sqlite3, base64, glob from cryptography.hazmat.primitives.ciphers.aead import AESGCM diff --git a/test/tools/deploy_review.py b/test/tools/deploy_review.py new file mode 100644 index 0000000..764adf2 --- /dev/null +++ b/test/tools/deploy_review.py @@ -0,0 +1,121 @@ +"""仅在 47 服务器备份并升级 Smail，保留双库和失败证据。""" + +import argparse +import hashlib +import json +import os +from pathlib import Path +import shutil +import sqlite3 +import subprocess +import time + + +ROOT = Path('/opt/smail') +STAGING = Path('/var/tmp/smail-review-20260917') +DATABASES = ('smail.db', 'smail-state.db') + + +def run(*args): + return subprocess.check_output(args, text=True).strip() + + +def digest(path): + with path.open('rb') as stream: + return hashlib.file_digest(stream, 'sha256').hexdigest() + + +def snapshot(source, target): + with sqlite3.connect(f'file:{source}?mode=ro', uri=True) as original: + with sqlite3.connect(target) as copied: + original.backup(copied) + assert copied.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' + target.chmod(0o600) + + +def wait_ready(): + for _ in range(60): + result = subprocess.run([ + 'curl', '--silent', '--fail', '--max-time', '2', '--noproxy', '*', + '--cacert', str(ROOT / 'ssl/custom.crt'), '--resolve', + 'mail.solit.cn:443:127.0.0.1', 'https://mail.solit.cn/api/ping', + ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if result.returncode == 0 and run('systemctl', 'is-active', 'smail.service') == 'active': + return + time.sleep(1) + raise RuntimeError('升级后服务未就绪') + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--main-sha', required=True) + parser.add_argument('--admin-sha', required=True) + args = parser.parse_args() + os.umask(0o077) + assert os.geteuid() == 0 and ROOT.resolve() == ROOT and not ROOT.is_symlink() + assert run('hostname') == 'ikun2' + assert '172.31.134.42/20' in run('ip', '-4', 'address', 'show') + assert run('systemctl', 'show', 'smail.service', '--property=User', '--value') == 'smail' + assert run('systemctl', 'is-active', 'smail.service') == 'active' + expected = {'smail': args.main_sha, 'smail-admin': args.admin_sha} + for name, checksum in expected.items(): + path = STAGING / name + with path.open('rb') as binary: + header = binary.read(20) + assert header[:5] == b'\x7fELF\x02' and header[18:20] == b'\x3e\x00' + assert digest(path) == checksum + assert not (ROOT / name).is_symlink() + + backup = Path('/var/backups') / time.strftime('smail-before-outbox-%Y%m%d-%H%M%S') + backup.mkdir(mode=0o700) + installed = False + subprocess.run(['systemctl', 'stop', 'smail.service'], check=True) + try: + # 停服归档包含配置、密钥和日志，仅保存在远程受保护目录。 + archive = backup / 'smail.tar.gz' + subprocess.run(['tar', '-czf', str(archive), '-C', '/opt', 'smail'], check=True) + subprocess.run(['tar', '-dzf', str(archive), '-C', '/opt'], check=True) + archive.chmod(0o600) + for name in DATABASES: + source = ROOT / 'data' / name + assert source.resolve() == source and not source.is_symlink() + snapshot(source, backup / name) + for name in expected: + shutil.copy2(ROOT / name, backup / name) + installed = True + for name, checksum in expected.items(): + subprocess.run(['install', '-m', '0755', '-o', 'root', '-g', 'root', + str(STAGING / name), str(ROOT / name)], check=True) + assert digest(ROOT / name) == checksum + subprocess.run(['systemctl', 'start', 'smail.service'], check=True) + wait_ready() + for name in DATABASES: + with sqlite3.connect(f'file:{ROOT / "data" / name}?mode=ro', uri=True) as database: + assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' + with sqlite3.connect(f'file:{ROOT / "data/smail.db"}?mode=ro', uri=True) as database: + version = database.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0] + assert version == 23 + states = dict(database.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) + result = {'backup': str(backup), 'archive_sha256': digest(archive), + 'artifacts': expected, 'migration': version, 'queue_states': states, + 'service': run('systemctl', 'is-active', 'smail.service'), + 'admin_version': run(str(ROOT / 'smail-admin'), 'version'), + 'database_integrity': 'ok', 'database_reset': False} + (backup / 'result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8') + print(json.dumps(result, ensure_ascii=False, indent=2)) + except BaseException: + if installed: + subprocess.run(['systemctl', 'stop', 'smail.service'], check=True) + for name in DATABASES: + snapshot(ROOT / 'data' / name, backup / ('failed-' + name)) + snapshot(backup / name, ROOT / 'data' / name) + shutil.chown(ROOT / 'data' / name, user='smail', group='smail') + for name in expected: + shutil.copy2(backup / name, ROOT / name) + subprocess.run(['systemctl', 'start', 'smail.service'], check=True) + print('升级失败，备份及故障证据保留于：' + str(backup)) + raise + + +if __name__ == '__main__': + main() diff --git a/test/tools/deploy_run.py b/test/tools/deploy_run.py index ceba383..4820dc4 100644 --- a/test/tools/deploy_run.py +++ b/test/tools/deploy_run.py @@ -1,7 +1,8 @@ import paramiko, time ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s_in, s_out, s_err = ssh.exec_command("chmod +x /opt/smail/smail_new && pkill -9 -f smail || true") s_out.channel.recv_exit_status() time.sleep(1) diff --git a/test/tools/dump_emails.py b/test/tools/dump_emails.py index 7b61a1a..632f984 100644 --- a/test/tools/dump_emails.py +++ b/test/tools/dump_emails.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, subject, send_user_id, send_time FROM email ;"') print("Emails:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, user_id, email_id, is_read, is_trash FROM user_email ;"') diff --git a/test/tools/fix_remote_config_digest.py b/test/tools/fix_remote_config_digest.py index b5381b8..e2e64b8 100644 --- a/test/tools/fix_remote_config_digest.py +++ b/test/tools/fix_remote_config_digest.py @@ -2,8 +2,9 @@ import paramiko, time ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) cmd = """python3 -c ' import hashlib, json diff --git a/test/tools/get_dkim.py b/test/tools/get_dkim.py index 8f74e1c..8395c41 100644 --- a/test/tools/get_dkim.py +++ b/test/tools/get_dkim.py @@ -1,7 +1,8 @@ import paramiko client = paramiko.SSHClient() -client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -client.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +client.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(client) cmd = 'openssl rsa -in /opt/smail/ssl/dkim/dkim.priv -pubout -outform PEM' stdin, stdout, stderr = client.exec_command(cmd) lines = [l.strip() for l in stdout.read().decode('utf-8').splitlines() if '---' not in l and l.strip()] diff --git a/test/tools/inspect_schema.py b/test/tools/inspect_schema.py index c6b9b4e..dc9d884 100644 --- a/test/tools/inspect_schema.py +++ b/test/tools/inspect_schema.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) _, stdout, _ = ssh.exec_command('sqlite3 /opt/smail/data/smail.db ".schema client_credential"') print(stdout.read().decode()) _, stdout, _ = ssh.exec_command('sqlite3 /opt/smail/data/smail.db ".schema protocol_credential"') diff --git a/test/tools/query_arbitrary.py b/test/tools/query_arbitrary.py index cb3f8e1..949016d 100644 --- a/test/tools/query_arbitrary.py +++ b/test/tools/query_arbitrary.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, type, subject, from_address, send_user_id, status FROM email ORDER BY id ASC;"') print("Security questions:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT user_id, count(*), sum(consumed) FROM recovery_code GROUP BY user_id;"') diff --git a/test/tools/query_emails.py b/test/tools/query_emails.py index 686202b..4d5f154 100644 --- a/test/tools/query_emails.py +++ b/test/tools/query_emails.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, user_id, email_id, is_read, group_id, status FROM user_email ORDER BY id DESC LIMIT 10;"') print("Security questions:\n", stdout.read().decode("utf-8", errors="ignore").encode("utf-8")) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, type, subject, send_user_id, status FROM email ORDER BY id DESC LIMIT 10;"') diff --git a/test/tools/query_remote_details.py b/test/tools/query_remote_details.py index e650a3f..921e566 100644 --- a/test/tools/query_remote_details.py +++ b/test/tools/query_remote_details.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT user_id, question FROM security_question;"') print("Security questions:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT user_id, count(*), sum(consumed) FROM recovery_code GROUP BY user_id;"') diff --git a/test/tools/query_remote_tables.py b/test/tools/query_remote_tables.py index 01727eb..df1bc2c 100644 --- a/test/tools/query_remote_tables.py +++ b/test/tools/query_remote_tables.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db ".schema security_question"') print("Schema security_question:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db ".schema recovery_code"') diff --git a/test/tools/query_remote_testuser.py b/test/tools/query_remote_testuser.py index ffccfb8..1405a7a 100644 --- a/test/tools/query_remote_testuser.py +++ b/test/tools/query_remote_testuser.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT * FROM security_question WHERE user_id = 2;"') print("Testuser2 security question:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT * FROM recovery_code WHERE user_id = 2;"') diff --git a/test/tools/query_remote_users.py b/test/tools/query_remote_users.py index 878b15e..cdac4f5 100644 --- a/test/tools/query_remote_users.py +++ b/test/tools/query_remote_users.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db "SELECT id, account, name, role, is_admin, disabled FROM user;"') print("Users in DB:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('sqlite3 /opt/smail/data/smail.db ".tables"') diff --git a/test/tools/remote_review.py b/test/tools/remote_review.py new file mode 100644 index 0000000..1c03934 --- /dev/null +++ b/test/tools/remote_review.py @@ -0,0 +1,45 @@ +import argparse +import pathlib +import shlex +import sys + +import paramiko + +from remote_ssh import connect_remote + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--command") + parser.add_argument("--upload", nargs=2, action="append", default=[]) + args = parser.parse_args() + client = connect_remote(paramiko.SSHClient()) + try: + with client.open_sftp() as sftp: + for source, destination in args.upload: + target = pathlib.PurePosixPath(destination) + if (not target.is_absolute() or ".." in target.parts + or len(target.parts) < 5 or target.parts[1:3] != ("var", "tmp") + or not target.parts[3].startswith("smail-review-")): + raise ValueError("上传路径必须位于项目审查临时目录") + command = "install -d -m 0755 " + shlex.quote(str(target.parent)) + _, output, error = client.exec_command(command) + if output.channel.recv_exit_status() != 0: + raise RuntimeError(error.read().decode("utf-8", "replace")) + sftp.put(source, destination) + sftp.chmod(destination, 0o755 if destination.endswith(".test") else 0o644) + if args.command: + _, output, error = client.exec_command(args.command, timeout=900) + for line in output: + print(line, end="", flush=True) + errors = error.read().decode("utf-8", "replace") + if errors: + print(errors, file=sys.stderr) + return output.channel.recv_exit_status() + return 0 + finally: + client.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test/tools/remote_ssh.py b/test/tools/remote_ssh.py new file mode 100644 index 0000000..1110f4c --- /dev/null +++ b/test/tools/remote_ssh.py @@ -0,0 +1,28 @@ +import os + +import paramiko + + +def connect_remote(client): + host = os.environ.get("SMAIL_SSH_HOST", "47.108.78.12") + user = os.environ.get("SMAIL_SSH_USER", "root") + password = os.environ.get("SMAIL_SSH_PASSWORD") or None + key = os.environ.get("SMAIL_SSH_KEY") or None + client.load_system_host_keys() + known_hosts = os.environ.get("SMAIL_SSH_KNOWN_HOSTS") + if known_hosts: + client.load_host_keys(known_hosts) + client.set_missing_host_key_policy(paramiko.RejectPolicy()) + client.connect( + host, + port=int(os.environ.get("SMAIL_SSH_PORT", "22")), + username=user, + password=password, + key_filename=key, + look_for_keys=password is None, + allow_agent=password is None, + timeout=10, + auth_timeout=10, + banner_timeout=10, + ) + return client diff --git a/test/tools/review_tests.sh b/test/tools/review_tests.sh new file mode 100644 index 0000000..77bc780 --- /dev/null +++ b/test/tools/review_tests.sh @@ -0,0 +1,12 @@ +#!/bin/sh +set -eu +cd /var/tmp/smail-review-20260917 +for name in outbox utils-send utils-smtp controllers-email db-migration res_init services-dbmigration; do + if runuser -u smail -- "./$name.test" -test.v -test.timeout=180s > "$name.result.log" 2>&1; then + printf '%s PASS\n' "$name" + tail -n 3 "$name.result.log" + else + cat "$name.result.log" + exit 1 + fi +done diff --git a/test/tools/run_check_ext.py b/test/tools/run_check_ext.py index 09bc9d3..b9e2f34 100644 --- a/test/tools/run_check_ext.py +++ b/test/tools/run_check_ext.py @@ -2,8 +2,9 @@ sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_check_hex.py b/test/tools/run_check_hex.py index 82a8c75..62e4511 100644 --- a/test/tools/run_check_hex.py +++ b/test/tools/run_check_hex.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_check_hex2.py b/test/tools/run_check_hex2.py index 0e399d4..4df5b0b 100644 --- a/test/tools/run_check_hex2.py +++ b/test/tools/run_check_hex2.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_check_hex3.py b/test/tools/run_check_hex3.py index e17a7e6..484aaba 100644 --- a/test/tools/run_check_hex3.py +++ b/test/tools/run_check_hex3.py @@ -2,8 +2,9 @@ sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) _, out, err = ssh.exec_command('python3 /tmp/check_hex2.py') print('OUT:\n', out.read().decode('utf-8', errors='ignore')) diff --git a/test/tools/run_check_logs.py b/test/tools/run_check_logs.py index a08cf5e..fe9511f 100644 --- a/test/tools/run_check_logs.py +++ b/test/tools/run_check_logs.py @@ -2,8 +2,9 @@ sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_check_names.py b/test/tools/run_check_names.py index cb522f0..807159b 100644 --- a/test/tools/run_check_names.py +++ b/test/tools/run_check_names.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_check_users.py b/test/tools/run_check_users.py index fe3baef..00285f9 100644 --- a/test/tools/run_check_users.py +++ b/test/tools/run_check_users.py @@ -2,8 +2,9 @@ sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3 diff --git a/test/tools/run_decrypt_creds.py b/test/tools/run_decrypt_creds.py index 1db71fe..6fe0234 100644 --- a/test/tools/run_decrypt_creds.py +++ b/test/tools/run_decrypt_creds.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = ''' import sqlite3, base64, glob diff --git a/test/tools/run_visual_setup.py b/test/tools/run_visual_setup.py index 29bf1a1..b9980b5 100644 --- a/test/tools/run_visual_setup.py +++ b/test/tools/run_visual_setup.py @@ -32,11 +32,11 @@ with sync_playwright() as p: page.screenshot(path=os.path.join(test_log_dir, 'step2_admin_password.png')) # Step 2: 输入管理员密码 - print('输入管理员密码 10102442Lb!...') + print('输入管理员密码 [redacted]...') pwd_inputs = page.locator('.form input') print('密码页输入框数量:', pwd_inputs.count()) for i in range(pwd_inputs.count()): - pwd_inputs.nth(i).fill('10102442Lb!') + pwd_inputs.nth(i).fill(os.environ["SMAIL_TEST_ADMIN_PASSWORD"]) time.sleep(1) page.screenshot(path=os.path.join(test_log_dir, 'step2_password_filled.png')) diff --git a/test/tools/run_visual_wizard.py b/test/tools/run_visual_wizard.py index bd787ed..79c189c 100644 --- a/test/tools/run_visual_wizard.py +++ b/test/tools/run_visual_wizard.py @@ -38,7 +38,7 @@ with sync_playwright() as p: inputs = page.locator('.database-form input, .ctn input') print('找到输入框数量:', inputs.count()) for i in range(inputs.count()): - inputs.nth(i).fill('10102442Lb!') + inputs.nth(i).fill(os.environ["SMAIL_TEST_ADMIN_PASSWORD"]) time.sleep(1) page.screenshot(path=os.path.join(test_log_dir, 'step2_password_filled.png')) diff --git a/test/tools/send_test_email.py b/test/tools/send_test_email.py index 75724bd..328636e 100644 --- a/test/tools/send_test_email.py +++ b/test/tools/send_test_email.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) s = '''import smtplib, ssl from email.mime.text import MIMEText diff --git a/test/tools/test_echo.py b/test/tools/test_echo.py index 852f693..eb924fd 100644 --- a/test/tools/test_echo.py +++ b/test/tools/test_echo.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command("echo HELLO") print(stdout:, stdout.read().decode()) print(stderr:, stderr.read().decode()) diff --git a/test/tools/test_inbound_info.py b/test/tools/test_inbound_info.py index 281b76b..5265fc6 100644 --- a/test/tools/test_inbound_info.py +++ b/test/tools/test_inbound_info.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) stdin, stdout, stderr = ssh.exec_command('python3 -c "import sqlite3; conn=sqlite3.connect(\x27/opt/smail/data/smail.db\x27); c=conn.cursor(); print(c.execute(\x27SELECT id, user_id, email_id, is_read, is_trash FROM user_email ORDER BY id DESC LIMIT 5\x27).fetchall())"') print("User emails top 5:\n", stdout.read().decode()) stdin, stdout, stderr = ssh.exec_command('python3 -c "import sqlite3; conn=sqlite3.connect(\x27/opt/smail/data/smail.db\x27); c=conn.cursor(); print(c.execute(\x27SELECT id, subject, send_time FROM email ORDER BY id DESC LIMIT 5\x27).fetchall())"') diff --git a/test/tools/test_remote_setup.py b/test/tools/test_remote_setup.py index e47cecc..3af848a 100644 --- a/test/tools/test_remote_setup.py +++ b/test/tools/test_remote_setup.py @@ -1,7 +1,8 @@ import paramiko client = paramiko.SSHClient() -client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -client.connect('47.108.78.12', port=22, username='root', password='10102442Lb!', timeout=8) +client.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(client) client.exec_command('systemctl stop smail-init; pkill -9 smail') setup_script = '''#!/bin/bash diff --git a/test/tools/update_remote_email_config.py b/test/tools/update_remote_email_config.py index 5878b2f..75b177b 100644 --- a/test/tools/update_remote_email_config.py +++ b/test/tools/update_remote_email_config.py @@ -1,7 +1,8 @@ import paramiko ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) -ssh.connect("47.108.78.12", port=22, username="root", password="10102442Lb!", timeout=10) +ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) +from remote_ssh import connect_remote +connect_remote(ssh) cmd = """python3 -c ' import json