diff --git a/backend/db/migration/outbound_delivery.go b/backend/db/migration/outbound_delivery.go index 5a8a22b..011ed7e 100644 --- a/backend/db/migration/outbound_delivery.go +++ b/backend/db/migration/outbound_delivery.go @@ -1,5 +1,11 @@ package migration +import "context" + +func applyOutboundSubmissionSchema(ctx context.Context, exec executor, dialect Dialect) error { + return ensureColumn(ctx, exec, dialect, "outbound_delivery", "submission", "TEXT NULL") +} + func outboundDeliveryStatements() map[Dialect][]string { return map[Dialect][]string{ DialectSQLite: { diff --git a/backend/db/migration/registry.go b/backend/db/migration/registry.go index 5faa40d..c6071d6 100644 --- a/backend/db/migration/registry.go +++ b/backend/db/migration/registry.go @@ -44,6 +44,7 @@ func DefaultMigrations() []Migration { DefineCustom(27, "expand_notification_ciphertext", "expand_notification_ciphertext_v1", applyNotificationCiphertextSchema), DefineCustom(28, "expand_recovery_code_ciphertext", "expand_recovery_code_ciphertext_v1", applyRecoveryCodeCiphertextSchema), DefineCustom(29, "expand_email_thread", "expand_email_thread_v1", applyEmailThreadSchema), + DefineCustom(30, "expand_outbound_submission", "expand_outbound_submission_v1", applyOutboundSubmissionSchema), } } diff --git a/backend/db/migration/registry_test.go b/backend/db/migration/registry_test.go index 9d5a8f8..2367a1e 100644 --- a/backend/db/migration/registry_test.go +++ b/backend/db/migration/registry_test.go @@ -298,6 +298,12 @@ func TestDefaultMigrationChecksumsAreImmutable(t *testing.T) { 22: "1047cb22f5a32c8a09ba4ccedff8713a5225c14fdb98357db86a6e0723c30d8b", 23: "5d80f2c2c583fc9a033fc320c0852c73bd6701ce6c8f12e8904deedfe0ba44b2", 24: "36533c34245853580f0e1c67ebb3a4eaeb8dafc8df2f3a85acff6b15d0ea4a6b", + 25: "86ea27d3323ef4e814dabca3e7ab55dd6f19c8ee4690cde90974751686c9b240", + 26: "f98d150fa43d09903ed954922495ddef8dc29fc3cf2d2229583bf0d2bce3f400", + 27: "9bdd98fc533e71f808869ae3ee9829f59702409f82030a6466c31254e08875af", + 28: "dd1698fc6fa0762ee6399324fec75e839fdb79aa93413acd547e573fb740c654", + 29: "5f027d90d556d50b236d7cfb48fbf7821041e46b8c8968563b4a4ebffde7ff3a", + 30: "392621bda08c66f6e8494bba1773f3c9ca9be6591f21afb094f2fcc3beaecfa8", } migrations := DefaultMigrations() if len(migrations) != len(expected) { diff --git a/backend/dto/parsemail/email.go b/backend/dto/parsemail/email.go index 3915905..20ec0df 100644 --- a/backend/dto/parsemail/email.go +++ b/backend/dto/parsemail/email.go @@ -77,6 +77,7 @@ type Email struct { ThreadReferences string EnvelopeFrom string EnvelopeTo []*User `json:"-"` + DeliveryTrigger string `json:"-"` ReplyTo []*User From *User To []*User diff --git a/backend/listen/smtp_server/audit_persistence.go b/backend/listen/smtp_server/audit_persistence.go index e75850b..9f94a20 100644 --- a/backend/listen/smtp_server/audit_persistence.go +++ b/backend/listen/smtp_server/audit_persistence.go @@ -1,20 +1,16 @@ package smtp_server import ( - "database/sql" stderrors "errors" - "fmt" "strings" "git.pcux.cn/xuans/Smail/config" "git.pcux.cn/xuans/Smail/consts" "git.pcux.cn/xuans/Smail/db" "git.pcux.cn/xuans/Smail/dto/parsemail" - "git.pcux.cn/xuans/Smail/hooks" - "git.pcux.cn/xuans/Smail/hooks/framework" "git.pcux.cn/xuans/Smail/models" + "git.pcux.cn/xuans/Smail/services/outbox" "git.pcux.cn/xuans/Smail/utils/array" - "git.pcux.cn/xuans/Smail/utils/async" "git.pcux.cn/xuans/Smail/utils/context" smtp "github.com/emersion/go-smtp" log "github.com/sirupsen/logrus" @@ -24,84 +20,27 @@ import ( const deliveryPendingError = "delivery result pending; outcome unknown after interruption" -type outgoingSender func(*context.Context, *parsemail.Email) (error, map[string]error) +type auditEnqueuer func(*xorm.Session, models.Email) error -func (s *Session) deliverOutgoingEmail(email *parsemail.Email, sender outgoingSender) error { +func (s *Session) deliverOutgoingEmail(email *parsemail.Email, trigger string) error { if err := validateOutgoingEmail(email); err != nil { return err } - ctx := s.Ctx - _, saved, err := saveEmail(ctx, email.Size, email, ctx.UserID, int(consts.EmailTypeSend), nil, true, true) + authorization, err := s.authorization.QueueAuthorization() + if err != nil || s.Ctx == nil || authorization.UserID != s.Ctx.UserID { + return &smtp.SMTPError{Code: 535, EnhancedCode: smtp.EnhancedCode{5, 7, 8}, Message: "Authentication credentials revoked"} + } + submission := outbox.Submission{ + EnvelopeFrom: s.From, Date: email.Date, Trigger: trigger, Authorization: authorization, + } + enqueue := func(session *xorm.Session, saved models.Email) error { + return outbox.EnqueueSMTP(session, saved, s.To, submission) + } + _, _, err = saveEmail(s.Ctx, email.Size, email, s.Ctx.UserID, int(consts.EmailTypeSend), nil, true, true, enqueue) if err != nil { - log.WithContext(ctx).Errorf("CRITICAL audit_event=initial_persist_failed direction=outbound msg_id=%q error=%v", email.MsgID, err) + log.WithContext(s.Ctx).Errorf("CRITICAL audit_event=submission_enqueue_failed direction=outbound msg_id=%q error=%v", email.MsgID, err) return localPersistenceSMTPError() } - - deliveryErr, domainErrors := sender(ctx, email) - runSendAfterHooks(ctx, email, domainErrors) - - targetStatus := consts.EmailStatusSent - var errorText any - if deliveryErr != nil { - targetStatus = consts.EmailStatusFail - errorText = deliveryErr.Error() - } - - if err = persistDeliveryResult(ctx, saved.Id, targetStatus, errorText); err != nil { - log.WithContext(ctx).Errorf( - "CRITICAL audit_event=delivery_result_persist_failed email_id=%d msg_id=%q target_status=%d downstream_accepted=%t error=%v", - saved.Id, saved.MsgID, targetStatus, deliveryErr == nil, err, - ) - } - - // If the remote MX accepted DATA, returning a local 451 here would cause a - // duplicate delivery. The persistence failure is retained as pending state - // and a CRITICAL log; the SMTP result must still reflect the remote result. - return downstreamDeliverySMTPError(deliveryErr, domainErrors) -} - -func runSendAfterHooks(ctx *context.Context, email *parsemail.Email, domainErrors map[string]error) { - process := async.New(ctx) - for _, hook := range hooks.Snapshot() { - if hook == nil { - continue - } - process.WaitProcess(func(value any) { - value.(framework.EmailHook).SendAfter(ctx, email, domainErrors) - }, hook) - } - process.Wait() -} - -func persistDeliveryResult(ctx *context.Context, emailID int, status int8, errorText any) error { - return db.Transaction(db.Instance, func(session *xorm.Session) error { - result, err := session.Exec(db.WithContext(ctx, `update email - set status=?, error=?, update_time=CURRENT_TIMESTAMP - where id=?`), status, errorText, emailID) - if err != nil { - return err - } - if err = requireOneAffected(result, "update email delivery result"); err != nil { - return err - } - - result, err = session.Exec(db.WithContext(ctx, `update user_email - set status=? where email_id=?`), status, emailID) - if err != nil { - return err - } - return requireOneAffected(result, "update user_email delivery result") - }) -} - -func requireOneAffected(result sql.Result, operation string) error { - affected, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("%s: read affected rows: %w", operation, err) - } - if affected != 1 { - return fmt.Errorf("%s: affected %d rows, want 1", operation, affected) - } return nil } @@ -113,7 +52,7 @@ func localPersistenceSMTPError() error { } } -func persistInitialAudit(ctx *context.Context, modelEmail models.Email, userIDs []int) (*models.Email, error) { +func persistInitialAudit(ctx *context.Context, modelEmail models.Email, userIDs []int, enqueue ...auditEnqueuer) (*models.Email, error) { var saved models.Email err := db.Transaction(db.Instance, func(session *xorm.Session) error { candidate := modelEmail @@ -130,6 +69,11 @@ func persistInitialAudit(ctx *context.Context, modelEmail models.Email, userIDs return err } } + for _, persist := range enqueue { + if err := persist(session, candidate); err != nil { + return err + } + } saved = candidate return nil }) diff --git a/backend/listen/smtp_server/audit_persistence_test.go b/backend/listen/smtp_server/audit_persistence_test.go index 56583e6..3a55e38 100644 --- a/backend/listen/smtp_server/audit_persistence_test.go +++ b/backend/listen/smtp_server/audit_persistence_test.go @@ -1,15 +1,12 @@ package smtp_server import ( - "bytes" "database/sql" + "encoding/json" "errors" "fmt" - "net" - "net/textproto" "path/filepath" "strings" - "sync/atomic" "testing" "time" @@ -21,49 +18,129 @@ import ( "git.pcux.cn/xuans/Smail/hooks" "git.pcux.cn/xuans/Smail/hooks/framework" "git.pcux.cn/xuans/Smail/models" + "git.pcux.cn/xuans/Smail/services/auth" + "git.pcux.cn/xuans/Smail/services/credentials" listservice "git.pcux.cn/xuans/Smail/services/list" + "git.pcux.cn/xuans/Smail/services/outbox" "git.pcux.cn/xuans/Smail/utils/context" smtp "github.com/emersion/go-smtp" - log "github.com/sirupsen/logrus" _ "modernc.org/sqlite" "xorm.io/xorm" ) -func TestOutgoingAuditInsertFailureDoesNotDeliver(t *testing.T) { - engine, dsn := newAuditTestEngine(t) - blocker := beginAuditReadTransaction(t, dsn) - - var deliveries atomic.Int32 - session := newAuditTestSession() - err := session.deliverOutgoingEmail(newAuditTestEmail("insert-busy@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - return nil, nil - }) - - assertSMTPCode(t, err, 451) - if got := deliveries.Load(); got != 0 { - t.Fatalf("downstream deliveries = %d, want 0", got) +func newAuthorizedAuditSession(t *testing.T, engine *xorm.Engine) (*Session, string) { + t.Helper() + user := models.User{ID: 1, Account: "sender", Password: "test-hash", Role: "user", SessionVersion: 1, SecurityVersion: 1} + if _, err := engine.Insert(&user); err != nil { + t.Fatal(err) } - assertAuditCounts(t, engine, 0, 0) + service, err := credentials.NewService(engine) + if err != nil { + t.Fatal(err) + } + if err := service.SetProtocolEnabled(user.ID, "smtp", true); err != nil { + t.Fatal(err) + } + code, err := service.GenerateProtocolCode(user.ID, "smtp") + if err != nil { + t.Fatal(err) + } + _, grant, err := auth.AuthenticateProtocolSession("smtp", "192.0.2.93:1234", "sender@example.com", code) + if err != nil { + t.Fatal(err) + } + session := newAuditTestSession() + session.authorization = grant + session.From = "sender@example.com" + session.To = []string{"recipient@example.net"} + return session, code +} +func auditQueue(t *testing.T, engine *xorm.Engine) []models.OutboundDelivery { + t.Helper() + var jobs []models.OutboundDelivery + if err := engine.Asc("id").Find(&jobs); err != nil { + t.Fatal(err) + } + return jobs +} + +func TestSMTPSubmissionQueuesOnlyAcceptedEnvelope(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session, code := newAuthorizedAuditSession(t, engine) + session.From = "sender@EXAMPLE.com" + session.To = []string{"envelope@example.net", "envelope@EXAMPLE.net"} + email := newAuditTestEmail("envelope-queue@example.com") + email.Date = "2026-09-27T12:30:00+08:00" + if err := session.deliverOutgoingEmail(email, "SMTP"); err != nil { + t.Fatal(err) + } + jobs := auditQueue(t, engine) + if len(jobs) != 1 || jobs[0].Recipient != "envelope@example.net" || jobs[0].State != "queued" || jobs[0].Attempts != 0 { + t.Fatalf("queued envelope jobs = %+v", jobs) + } + var submission outbox.Submission + if err := json.Unmarshal([]byte(jobs[0].Submission), &submission); err != nil { + t.Fatal(err) + } + if submission.EnvelopeFrom != session.From || submission.Date != email.Date || submission.Authorization.Kind != "smtp" || submission.Trigger != "SMTP" { + t.Fatal("SMTP envelope or authorization metadata was not preserved") + } + if strings.Contains(jobs[0].Submission, code) || strings.Contains(jobs[0].Submission, "test-hash") { + t.Fatal("submission persisted reusable credentials") + } + if email.To[0].EmailAddress != "recipient@example.net" || email.MessageId <= 0 { + t.Fatal("queue changed message headers or lost stored identity") + } + var saved models.Email + if found, err := engine.ID(jobs[0].EmailID).Get(&saved); err != nil || !found { + t.Fatalf("saved message: found=%v err=%v", found, err) + } + expectedDate, err := time.Parse(time.RFC3339, email.Date) + if err != nil || !saved.SendDate.Equal(expectedDate) { + t.Fatalf("stored message date = %v, want %v", saved.SendDate, expectedDate) + } + assertAuditCounts(t, engine, 1, 1) + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) +} + +func TestSMTPQueueInsertFailureRollsBackWholeSubmission(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session, _ := newAuthorizedAuditSession(t, engine) + if _, err := engine.Exec(`CREATE TRIGGER fail_queue_insert BEFORE INSERT ON outbound_delivery + BEGIN SELECT RAISE(ABORT, 'forced queue insert failure'); END`); err != nil { + t.Fatal(err) + } + email := newAuditTestEmail("queue-failure@example.com") + assertSMTPCode(t, session.deliverOutgoingEmail(email, "SMTP"), 451) + assertAuditCounts(t, engine, 0, 0) + if len(auditQueue(t, engine)) != 0 || email.MessageId != 0 { + t.Fatal("failed submission left partial durable state") + } +} + +func TestSMTPQueueBusyFailureDoesNotAcceptSubmission(t *testing.T) { + engine, dsn := newAuditTestEngine(t) + session, _ := newAuthorizedAuditSession(t, engine) + blocker := beginAuditReadTransaction(t, dsn) + assertSMTPCode(t, session.deliverOutgoingEmail(newAuditTestEmail("busy@example.com"), "SMTP"), 451) + assertAuditCounts(t, engine, 0, 0) + if len(auditQueue(t, engine)) != 0 { + t.Fatal("failed transaction left delivery tasks") + } if err := blocker.Rollback(); err != nil { t.Fatal(err) } - if err := session.deliverOutgoingEmail(newAuditTestEmail("insert-busy@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - return nil, nil - }); err != nil { - t.Fatalf("delivery after lock release failed: %v", err) - } - if got := deliveries.Load(); got != 1 { - t.Fatalf("downstream deliveries after recovery = %d, want 1", got) + if err := session.deliverOutgoingEmail(newAuditTestEmail("busy@example.com"), "SMTP"); err != nil { + t.Fatal(err) } assertAuditCounts(t, engine, 1, 1) - assertAuditStatuses(t, engine, consts.EmailStatusSent) + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) } -func TestOutgoingAuditRetriesBusyInitialInsertBeforeDeliver(t *testing.T) { +func TestSMTPQueueRetriesInitialTransactionBeforeAccepting(t *testing.T) { engine, dsn := newAuditTestEngine(t) + session, _ := newAuthorizedAuditSession(t, engine) blocker := beginAuditReadTransaction(t, dsn) released := make(chan struct{}) go func() { @@ -71,21 +148,75 @@ func TestOutgoingAuditRetriesBusyInitialInsertBeforeDeliver(t *testing.T) { _ = blocker.Rollback() close(released) }() - - var deliveries atomic.Int32 - err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("initial-retry@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - return nil, nil - }) + err := session.deliverOutgoingEmail(newAuditTestEmail("busy-retry@example.com"), "SMTP") <-released if err != nil { - t.Fatalf("deliverOutgoingEmail() error = %v", err) - } - if got := deliveries.Load(); got != 1 { - t.Fatalf("downstream deliveries = %d, want 1", got) + t.Fatal(err) } assertAuditCounts(t, engine, 1, 1) - assertAuditStatuses(t, engine, consts.EmailStatusSent) + if len(auditQueue(t, engine)) != 1 { + t.Fatal("transaction retry duplicated delivery tasks") + } +} + +func TestSMTPQueueRequiresCurrentAuthorization(t *testing.T) { + for _, revoke := range []bool{false, true} { + t.Run(fmt.Sprintf("revoke=%t", revoke), func(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session := newAuditTestSession() + if revoke { + session, _ = newAuthorizedAuditSession(t, engine) + if _, err := engine.Where("user_id = ?", 1).Cols("enabled").Update(&models.ProtocolCredential{Enabled: false}); err != nil { + t.Fatal(err) + } + } + assertSMTPCode(t, session.deliverOutgoingEmail(newAuditTestEmail("revoked@example.com"), "SMTP"), 535) + assertAuditCounts(t, engine, 0, 0) + if len(auditQueue(t, engine)) != 0 { + t.Fatal("unauthorized submission was queued") + } + }) + } +} + +func TestSameMessageIDCreatesIndependentQueuedSubmissions(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session, _ := newAuthorizedAuditSession(t, engine) + for i := 0; i < 2; i++ { + email := newAuditTestEmail("reused-message-id@example.com") + email.Text = []byte(fmt.Sprintf("body %d", i)) + if err := session.deliverOutgoingEmail(email, "SMTP"); err != nil { + t.Fatal(err) + } + } + assertAuditCounts(t, engine, 2, 2) + assertMessageIDCount(t, engine, "reused-message-id@example.com", 2) + if len(auditQueue(t, engine)) != 2 { + t.Fatal("independent submissions were merged by Message-ID") + } +} + +func TestSMTPAcceptanceDoesNotInvokeDeliveryCompletionHooks(t *testing.T) { + engine, _ := newAuditTestEngine(t) + session, _ := newAuthorizedAuditSession(t, engine) + hook := &queuedAuditHook{authenticationCaptureHook: &authenticationCaptureHook{}} + hooks.HookList = map[string]framework.EmailHook{"queued": hook} + if err := session.deliverOutgoingEmail(newAuditTestEmail("deferred-hook@example.com"), "SMTP"); err != nil { + t.Fatal(err) + } + if hook.called { + t.Fatal("SendAfter was called before the queue delivered the message") + } + assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) +} + +type queuedAuditHook struct { + *authenticationCaptureHook + called bool +} + +func (h *queuedAuditHook) SendAfter(*context.Context, *parsemail.Email, map[string]error) { + h.called = true } func TestOutgoingAuditRelationInsertRollsBackEmail(t *testing.T) { @@ -139,237 +270,6 @@ func TestIncomingAuditRelationInsertRollsBackEmail(t *testing.T) { assertAuditCounts(t, engine, 0, 0) } -func TestOutgoingAuditRetriesBusyFinalUpdateAtomically(t *testing.T) { - engine, dsn := newAuditTestEngine(t) - var blocker *sql.Tx - - session := newAuditTestSession() - err := session.deliverOutgoingEmail(newAuditTestEmail("final-retry@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - blocker = beginAuditReadTransaction(t, dsn) - go func() { - time.Sleep(35 * time.Millisecond) - _ = blocker.Rollback() - }() - return nil, nil - }) - if err != nil { - t.Fatalf("deliverOutgoingEmail() error = %v", err) - } - assertAuditStatuses(t, engine, consts.EmailStatusSent) - assertAuditError(t, engine, "", false) -} - -func TestOutgoingAuditFinalUpdateIsAtomic(t *testing.T) { - engine, _ := newAuditTestEngine(t) - if _, err := engine.Exec(fmt.Sprintf(`CREATE TRIGGER fail_user_email_success - BEFORE UPDATE OF status ON user_email - WHEN NEW.status = %d - BEGIN - SELECT RAISE(ABORT, 'forced user_email update failure'); - END`, consts.EmailStatusSent)); err != nil { - t.Fatal(err) - } - - session := newAuditTestSession() - err := session.deliverOutgoingEmail(newAuditTestEmail("atomic-update@example.com"), successfulAuditSender) - if err != nil { - t.Fatalf("accepted delivery must remain successful, got %v", err) - } - assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) - assertAuditError(t, engine, deliveryPendingError, true) -} - -func TestAcceptedDeliveryWithPermanentAuditLockReturnsSuccessAndAlerts(t *testing.T) { - engine, dsn := newAuditTestEngine(t) - var blocker *sql.Tx - var deliveries atomic.Int32 - - oldOutput := log.StandardLogger().Out - oldLevel := log.GetLevel() - var logs bytes.Buffer - log.SetOutput(&logs) - log.SetLevel(log.ErrorLevel) - t.Cleanup(func() { - log.SetOutput(oldOutput) - log.SetLevel(oldLevel) - }) - - err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("accepted-unknown@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - blocker = beginAuditReadTransaction(t, dsn) - return nil, nil - }) - if err != nil { - t.Fatalf("accepted delivery must remain successful, got %v", err) - } - if got := deliveries.Load(); got != 1 { - t.Fatalf("downstream deliveries = %d, want 1", got) - } - if !strings.Contains(logs.String(), "audit_event=delivery_result_persist_failed") { - t.Fatalf("missing CRITICAL persistence alert in logs: %s", logs.String()) - } - if err = blocker.Rollback(); err != nil { - t.Fatal(err) - } - assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) - assertAuditError(t, engine, deliveryPendingError, true) -} - -func TestTemporaryDeliveryFailureWithUnpersistedResultCanRetryAsNewAttempt(t *testing.T) { - engine, dsn := newAuditTestEngine(t) - var blocker *sql.Tx - var deliveries atomic.Int32 - session := newAuditTestSession() - messageID := "temporary-unpersisted@example.com" - - err := session.deliverOutgoingEmail(newAuditTestEmail(messageID), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - blocker = beginAuditReadTransaction(t, dsn) - return errors.New("delivery failed"), map[string]error{ - "example.net": &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}, - } - }) - assertSMTPCode(t, err, 451) - if err = blocker.Rollback(); err != nil { - t.Fatal(err) - } - - if err = session.deliverOutgoingEmail(newAuditTestEmail(messageID), func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - return nil, nil - }); err != nil { - t.Fatalf("retry after temporary failure: %v", err) - } - if got := deliveries.Load(); got != 2 { - t.Fatalf("downstream deliveries = %d, want 2", got) - } - assertAuditCounts(t, engine, 2, 2) - assertMessageIDCount(t, engine, messageID, 2) -} - -func TestOutgoingAuditPersistsDeliveryFailure(t *testing.T) { - tests := []struct { - name string - domainErr error - wantCode int - }{ - { - name: "temporary", - domainErr: &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}, - wantCode: 451, - }, - { - name: "permanent", - domainErr: &textproto.Error{Code: 550, Msg: "mailbox unavailable"}, - wantCode: 550, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - engine, _ := newAuditTestEngine(t) - deliveryErr := errors.New("delivery failed") - err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail(tt.name+"-failure@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - return deliveryErr, map[string]error{"example.net": tt.domainErr} - }) - assertSMTPCode(t, err, tt.wantCode) - assertAuditStatuses(t, engine, consts.EmailStatusFail) - assertAuditError(t, engine, deliveryErr.Error(), true) - }) - } -} - -func TestPermanentDeliveryFailureSurvivesFinalAuditLock(t *testing.T) { - engine, dsn := newAuditTestEngine(t) - var blocker *sql.Tx - err := newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("permanent-failure@example.com"), func(*context.Context, *parsemail.Email) (error, map[string]error) { - blocker = beginAuditReadTransaction(t, dsn) - failure := &textproto.Error{Code: 550, Msg: "mailbox unavailable"} - return errors.New("delivery failed"), map[string]error{"example.net": failure} - }) - assertSMTPCode(t, err, 550) - if err = blocker.Rollback(); err != nil { - t.Fatal(err) - } - assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) - assertAuditError(t, engine, deliveryPendingError, true) -} - -func TestSameMessageIDCreatesIndependentSubmissions(t *testing.T) { - engine, _ := newAuditTestEngine(t) - var deliveries atomic.Int32 - sender := func(*context.Context, *parsemail.Email) (error, map[string]error) { - deliveries.Add(1) - return nil, nil - } - - session := newAuditTestSession() - for i := 0; i < 2; i++ { - email := newAuditTestEmail("reused-message-id@example.com") - email.Text = []byte(fmt.Sprintf("body %d", i)) - if err := session.deliverOutgoingEmail(email, sender); err != nil { - t.Fatalf("submission %d: %v", i+1, err) - } - } - if got := deliveries.Load(); got != 2 { - t.Fatalf("downstream deliveries = %d, want 2", got) - } - assertAuditCounts(t, engine, 2, 2) - assertMessageIDCount(t, engine, "reused-message-id@example.com", 2) -} - -func TestSendAfterHookCompletesBeforeFinalPersistAndSMTPResult(t *testing.T) { - engine, _ := newAuditTestEngine(t) - hook := &blockingSendAfterHook{ - authenticationCaptureHook: &authenticationCaptureHook{}, - started: make(chan struct{}), - release: make(chan struct{}), - done: make(chan struct{}), - } - hooks.HookList = map[string]framework.EmailHook{"blocking": hook} - - result := make(chan error, 1) - go func() { - result <- newAuditTestSession().deliverOutgoingEmail(newAuditTestEmail("sync-hook@example.com"), successfulAuditSender) - }() - t.Cleanup(func() { - select { - case <-hook.release: - default: - close(hook.release) - } - }) - - select { - case <-hook.started: - case <-time.After(time.Second): - t.Fatal("SendAfter hook was not started") - } - assertAuditStatuses(t, engine, consts.EmailStatusDeliveryPending) - - select { - case err := <-result: - t.Fatalf("delivery returned before SendAfter hook completed: %v", err) - case <-time.After(50 * time.Millisecond): - } - close(hook.release) - - select { - case <-hook.done: - case <-time.After(time.Second): - t.Fatal("SendAfter hook did not finish") - } - select { - case err := <-result: - if err != nil { - t.Fatalf("deliverOutgoingEmail() error = %v", err) - } - case <-time.After(time.Second): - t.Fatal("delivery did not return after SendAfter hook completed") - } - assertAuditStatuses(t, engine, consts.EmailStatusSent) -} - func TestOutgoingPendingAuditMailboxVisibility(t *testing.T) { engine, _ := newAuditTestEngine(t) session := newAuditTestSession() @@ -417,19 +317,6 @@ func TestOutgoingPendingAuditMailboxVisibility(t *testing.T) { } } -type blockingSendAfterHook struct { - *authenticationCaptureHook - started chan struct{} - release chan struct{} - done chan struct{} -} - -func (h *blockingSendAfterHook) SendAfter(*context.Context, *parsemail.Email, map[string]error) { - defer close(h.done) - close(h.started) - <-h.release -} - func newAuditTestEngine(t *testing.T) (*xorm.Engine, string) { t.Helper() dsn := filepath.Join(t.TempDir(), "audit.db") + "?_pragma=busy_timeout(20)" @@ -439,7 +326,7 @@ func newAuditTestEngine(t *testing.T) (*xorm.Engine, string) { } engine.SetMaxOpenConns(1) engine.SetMaxIdleConns(1) - if err = engine.Sync2(&models.User{}, &models.Email{}, &models.UserEmail{}); err != nil { + if err = engine.Sync2(&models.User{}, &models.Email{}, &models.UserEmail{}, &models.OutboundDelivery{}, &models.ProtocolCredential{}, &models.ClientCredential{}); err != nil { engine.Close() t.Fatal(err) } diff --git a/backend/listen/smtp_server/read_content.go b/backend/listen/smtp_server/read_content.go index 1f57b56..95318e4 100644 --- a/backend/listen/smtp_server/read_content.go +++ b/backend/listen/smtp_server/read_content.go @@ -27,7 +27,6 @@ import ( "git.pcux.cn/xuans/Smail/utils/array" "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" "github.com/emersion/go-smtp" "github.com/mileusna/spf" @@ -170,23 +169,21 @@ func (s *Session) Data(r io.Reader) (resultErr error) { } details.Stage = "保存并投递客户端邮件" - sender := outgoingSender(send.Send) + trigger := "SMTP" if s.isNotification() { details.Stage = "校验系统通知配额并投递" if err := s.prepareNotification(email, len(emailData)); err != nil { return err } - sender = func(ctx *context.Context, email *parsemail.Email) (error, map[string]error) { - return send.SendTriggered(ctx, email, "系统通知 SMTP 发信") - } + trigger = "系统通知 SMTP 发信" } email.EnvelopeTo = make([]*parsemail.User, len(s.To)) for index, address := range s.To { email.EnvelopeTo[index] = &parsemail.User{EmailAddress: address} } - resultErr = s.deliverOutgoingEmail(email, sender) + resultErr = s.deliverOutgoingEmail(email, trigger) if resultErr == nil { - result = "accepted" + result = "success" } return resultErr @@ -337,7 +334,7 @@ func authorizedSubmissionEmail(email *parsemail.Email, envelopeFrom string, ctx return envelopeFrom == "" || validation.IsAuthorizedSender(envelopeFrom, ctx.UserAccount, allowLocalAlias, domains.Accepted) } -func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserID int, emailType int, reallyTo []string, SPFStatus, dkimStatus bool) ([]*models.User, *models.Email, error) { +func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserID int, emailType int, reallyTo []string, SPFStatus, dkimStatus bool, enqueue ...auditEnqueuer) ([]*models.User, *models.Email, error) { if email == nil { return nil, nil, oerrors.New("email is missing") } @@ -396,6 +393,10 @@ func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserI return users, nil, fmt.Errorf("encode stored email fields: %w", err) } + sendDate := time.Now() + if parsed, parseErr := time.Parse(time.RFC3339, email.Date); parseErr == nil { + sendDate = parsed + } modelEmail := models.Email{ Type: cast.ToInt8(emailType), Subject: email.Subject, @@ -415,7 +416,7 @@ func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserI SPFCheck: boolInt8(SPFStatus), DKIMCheck: boolInt8(dkimStatus), SendUserID: sendUserID, - SendDate: time.Now(), + SendDate: sendDate, Status: cast.ToInt8(email.Status), CreateTime: time.Now(), CronSendTime: time.Now(), @@ -434,7 +435,7 @@ func saveEmail(ctx *context.Context, size int, email *parsemail.Email, sendUserI } log.WithContext(ctx).Debugf("开始入库!") - saved, err := persistInitialAudit(ctx, modelEmail, userIDs) + saved, err := persistInitialAudit(ctx, modelEmail, userIDs, enqueue...) if err != nil { return users, nil, err } diff --git a/backend/models/outbound_delivery.go b/backend/models/outbound_delivery.go index 4080a22..ed18a6c 100644 --- a/backend/models/outbound_delivery.go +++ b/backend/models/outbound_delivery.go @@ -8,6 +8,7 @@ type OutboundDelivery struct { Attempts int `xorm:"notnull"` NextAttempt int64 `xorm:"notnull"` LastError string `xorm:"text"` + Submission string `xorm:"text" json:"-"` } func (OutboundDelivery) TableName() string { return "outbound_delivery" } diff --git a/backend/services/auth/protocol_session.go b/backend/services/auth/protocol_session.go index d07b024..041da1c 100644 --- a/backend/services/auth/protocol_session.go +++ b/backend/services/auth/protocol_session.go @@ -1,6 +1,10 @@ package auth import ( + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" "errors" "sync/atomic" @@ -8,6 +12,7 @@ import ( "git.pcux.cn/xuans/Smail/models" "git.pcux.cn/xuans/Smail/services/credentials" "git.pcux.cn/xuans/Smail/services/notification" + "xorm.io/xorm" ) // ProtocolSession 保存不可变身份与凭据快照,拒绝后的连接不能通过重新启用恢复权限。 @@ -18,6 +23,85 @@ type ProtocolSession struct { invalid atomic.Bool } +type QueuedAuthorization struct { + UserID int `json:"user_id"` + Kind string `json:"kind"` + ID int `json:"id"` + Version int `json:"version"` + Credential string `json:"credential"` + Identity string `json:"identity"` +} + +func (s *ProtocolSession) QueueAuthorization() (QueuedAuthorization, error) { + var result QueuedAuthorization + if err := s.Check(); err != nil { + return result, err + } + if s.IsNotification() { + result.Kind, result.ID, result.Version, result.Credential = s.notification.QueueBinding() + } else { + result.Kind, result.ID, result.Version, result.Credential = s.grant.QueueBinding() + } + if result.Kind == "" || result.ID <= 0 || result.Version <= 0 { + return QueuedAuthorization{}, notification.ErrDenied + } + result.Identity = queuedUserIdentity(s.user) + result.UserID = s.user.ID + return result, nil +} + +func queuedDigest(value string) string { + hash := sha256.Sum256([]byte(value)) + return hex.EncodeToString(hash[:]) +} + +func queuedUserIdentity(user models.User) string { + value, _ := json.Marshal([]any{ + user.ID, user.Account, user.Password, user.SessionVersion, + user.SecurityVersion, user.EffectiveRole(), user.MustChangePassword, + }) + return queuedDigest(string(value)) +} + +func (a QueuedAuthorization) Check(engine *xorm.Engine, user models.User) (bool, error) { + if engine == nil { + return false, notification.ErrUnavailable + } + if user.ID <= 0 || a.UserID != user.ID || user.Disabled != 0 || user.MustChangePassword != 0 || a.ID <= 0 || a.Version <= 0 || + subtle.ConstantTimeCompare([]byte(a.Identity), []byte(queuedUserIdentity(user))) != 1 { + return false, nil + } + var hash string + switch a.Kind { + case "smtp": + var item models.ProtocolCredential + found, err := engine.ID(a.ID).Where("user_id = ? AND protocol = ? AND enabled = ? AND code_version = ? AND revoked_at IS NULL", user.ID, "smtp", true, a.Version).Get(&item) + if err != nil || !found { + return false, err + } + hash = item.CodeHash + case "client": + var item models.ClientCredential + found, err := engine.ID(a.ID).Where("user_id = ? AND code_version = ? AND revoked_at IS NULL", user.ID, a.Version).Get(&item) + if err != nil || !found { + return false, err + } + hash = item.CodeHash + case "notification": + if a.ID != user.ID { + return false, nil + } + item, err := notification.Load(engine, user.ID) + if err != nil || !notification.Configured(item, user) || item.CodeVersion != a.Version { + return false, err + } + hash = item.CodeHash + default: + return false, nil + } + return hash != "" && subtle.ConstantTimeCompare([]byte(a.Credential), []byte(queuedDigest(hash))) == 1, nil +} + func AuthenticateProtocolSession(protocol, remoteAddress, account, plain string) (*models.User, *ProtocolSession, error) { if notification.IsUsername(account) { return authenticateNotification(protocol, remoteAddress, account, plain) diff --git a/backend/services/auth/protocol_session_test.go b/backend/services/auth/protocol_session_test.go index 2dba872..4322692 100644 --- a/backend/services/auth/protocol_session_test.go +++ b/backend/services/auth/protocol_session_test.go @@ -2,6 +2,7 @@ package auth import ( "path/filepath" + "strings" "sync" "testing" "time" @@ -11,9 +12,74 @@ import ( "git.pcux.cn/xuans/Smail/internal/passwordtest" "git.pcux.cn/xuans/Smail/models" "git.pcux.cn/xuans/Smail/services/credentials" + "git.pcux.cn/xuans/Smail/services/notification" "xorm.io/xorm" ) +func TestQueuedNotificationAuthorizationTracksCredentialVersion(t *testing.T) { + oldDB := db.Instance + engine, err := xorm.NewEngine("sqlite", filepath.Join(t.TempDir(), "notification.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = engine.Close(); db.Instance = oldDB }) + db.Instance = engine + if err := engine.Sync2(&models.User{}, &models.NotificationGrant{}); err != nil { + t.Fatal(err) + } + user := models.User{Account: "sender", Password: "test-hash", Role: "user", SessionVersion: 1, SecurityVersion: 1} + if _, err := engine.Insert(&user); err != nil { + t.Fatal(err) + } + code := "sn_" + strings.Repeat("A", 43) + item := models.NotificationGrant{ + UserID: user.ID, Enabled: 1, CodeVersion: 2, CodeHash: queuedDigest(code), + SessionVersion: user.SessionVersion, SecurityVersion: user.SecurityVersion, + } + if _, err := engine.Insert(&item); err != nil { + t.Fatal(err) + } + grant, err := notification.Authenticate(engine, user, code) + if err != nil { + t.Fatal(err) + } + session := &ProtocolSession{user: user, notification: grant} + snapshot, err := session.QueueAuthorization() + if err != nil { + t.Fatal(err) + } + if snapshot.Kind != "notification" || snapshot.ID != user.ID || snapshot.Version != item.CodeVersion || + snapshot.Credential == code || snapshot.Credential == item.CodeHash { + t.Fatal("notification snapshot is not bound to the original credential") + } + if valid, err := snapshot.Check(engine, user); err != nil || !valid { + t.Fatalf("notification queue authorization = %v, %v", valid, err) + } + for _, change := range []string{"disable", "version", "hash", "session", "security"} { + t.Run(change, func(t *testing.T) { + updated := item + switch change { + case "disable": + updated.Enabled = 0 + case "version": + updated.CodeVersion++ + case "hash": + updated.CodeHash = queuedDigest("replacement") + case "session": + updated.SessionVersion++ + case "security": + updated.SecurityVersion++ + } + if _, err := engine.ID(user.ID).Cols("enabled", "code_version", "code_hash", "session_version", "security_version").Update(&updated); err != nil { + t.Fatal(err) + } + if valid, err := snapshot.Check(engine, user); err != nil || valid { + t.Fatalf("stale notification queue authorization = %v, %v", valid, err) + } + }) + } +} + func TestProtocolSessionRechecksUserAndCredential(t *testing.T) { for _, kind := range []string{"protocol", "client"} { for _, change := range []string{"revoke", "disabled", "password", "session_version", "security_version", "role", "deleted", "database_failure"} { @@ -62,6 +128,16 @@ func TestProtocolSessionRechecksUserAndCredential(t *testing.T) { if err != nil { t.Fatal(err) } + queued, err := session.QueueAuthorization() + if err != nil { + t.Fatal(err) + } + if valid, err := queued.Check(engine, user); err != nil || !valid { + t.Fatalf("new queued authorization = %v, %v", valid, err) + } + if queued.Credential == code || queued.Identity == user.Password || queued.Kind == "" { + t.Fatal("queued authorization contains reusable credentials or has no kind") + } // 命令复核不占用密码哈希容量,也不累加登录失败计数。 release := passwordtest.Hold(t) if err := session.Check(); err != nil { @@ -93,6 +169,17 @@ func TestProtocolSessionRechecksUserAndCredential(t *testing.T) { if err != nil { t.Fatal(err) } + current := user + if change != "database_failure" { + current = models.User{} + if _, err := engine.ID(user.ID).Get(¤t); err != nil { + t.Fatal(err) + } + } + valid, checkErr := queued.Check(engine, current) + if valid || (change == "database_failure") != (checkErr != nil) { + t.Fatalf("stale queued authorization = %v, %v", valid, checkErr) + } var wait sync.WaitGroup for range 4 { wait.Add(1) diff --git a/backend/services/credentials/protocol_grant.go b/backend/services/credentials/protocol_grant.go index 28d9c68..edffe59 100644 --- a/backend/services/credentials/protocol_grant.go +++ b/backend/services/credentials/protocol_grant.go @@ -1,6 +1,9 @@ package credentials import ( + "crypto/sha256" + "encoding/hex" + "git.pcux.cn/xuans/Smail/models" "xorm.io/xorm" ) @@ -17,6 +20,18 @@ type ProtocolGrant struct { hash string } +func (g *ProtocolGrant) QueueBinding() (string, int, int, string) { + if g == nil || g.protocol != "smtp" { + return "", 0, 0, "" + } + kind := "smtp" + if g.client { + kind = "client" + } + hash := sha256.Sum256([]byte(g.hash)) + return kind, g.id, g.version, hex.EncodeToString(hash[:]) +} + func (g *ProtocolGrant) Valid(engine *xorm.Engine) bool { if g == nil || engine == nil || engine != g.engine || g.id <= 0 { return false diff --git a/backend/services/notification/service.go b/backend/services/notification/service.go index 5f71a60..7f0934d 100644 --- a/backend/services/notification/service.go +++ b/backend/services/notification/service.go @@ -216,6 +216,13 @@ type Grant struct { securityVersion int } +func (g *Grant) QueueBinding() (string, int, int, string) { + if g == nil { + return "", 0, 0, "" + } + return "notification", g.userID, g.version, digest(g.hash) +} + func Authenticate(engine *xorm.Engine, user models.User, plain string) (*Grant, error) { if len(plain) != 46 || !strings.HasPrefix(plain, "sn_") { return nil, ErrDenied diff --git a/backend/services/outbox/runtime.go b/backend/services/outbox/runtime.go index cd63977..5bfcb46 100644 --- a/backend/services/outbox/runtime.go +++ b/backend/services/outbox/runtime.go @@ -2,18 +2,24 @@ package outbox import ( stdcontext "context" + "encoding/json" "errors" "fmt" + "strings" "sync" "time" + "git.pcux.cn/xuans/Smail/config" "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/access" "git.pcux.cn/xuans/Smail/services/dbmigration" + "git.pcux.cn/xuans/Smail/services/notification" "git.pcux.cn/xuans/Smail/utils/context" "git.pcux.cn/xuans/Smail/utils/send" "git.pcux.cn/xuans/Smail/utils/smtp" + "git.pcux.cn/xuans/Smail/utils/validation" log "github.com/sirupsen/logrus" "xorm.io/xorm" ) @@ -121,6 +127,17 @@ func (r *Runtime) processWritable(job models.OutboundDelivery) error { return err } } + var submission Submission + authorized := true + if job.Submission != "" { + authorized = userFound && json.Unmarshal([]byte(job.Submission), &submission) == nil + if authorized { + authorized, err = submission.Authorization.Check(r.engine, user) + if err != nil { + return err + } + } + } owned, err := claim(r.engine, &job) if err != nil || !owned { return err @@ -144,6 +161,17 @@ func (r *Runtime) processWritable(job models.OutboundDelivery) error { if err != nil { return err } + if job.Submission != "" { + if !authorized || !authorizedQueuedSender(submission, message, user) { + return notification.ErrDenied + } + message.EnvelopeFrom = submission.EnvelopeFrom + if submission.Date != "" { + message.Date = submission.Date + } + message.EnvelopeTo = []*parsemail.User{{EmailAddress: job.Recipient}} + message.DeliveryTrigger = submission.Trigger + } return r.deliver(ctx, message, job.Recipient) }() // 固定首次结果及退避时间,后续只重试落库,绝不再次调用 SMTP。 @@ -151,6 +179,22 @@ func (r *Runtime) processWritable(job models.OutboundDelivery) error { return r.flushResult() } +func authorizedQueuedSender(submission Submission, message *parsemail.Email, user models.User) bool { + if message == nil || message.From == nil || message.Sender == nil || submission.Trigger == "" { + return false + } + domains := config.CurrentMailDomains() + if submission.Authorization.Kind == "notification" { + sender := validation.NotificationAddress(domains.Primary) + return sender != "" && strings.EqualFold(submission.EnvelopeFrom, sender) && + strings.EqualFold(message.From.EmailAddress, sender) && strings.EqualFold(message.Sender.EmailAddress, sender) + } + allowAlias := user.EffectiveRole() == access.RoleSuperAdmin + return validation.IsAuthorizedSender(message.From.EmailAddress, user.Account, allowAlias, domains.Accepted) && + validation.IsAuthorizedSender(message.Sender.EmailAddress, user.Account, allowAlias, domains.Accepted) && + (submission.EnvelopeFrom == "" || validation.IsAuthorizedSender(submission.EnvelopeFrom, user.Account, allowAlias, domains.Accepted)) +} + func (r *Runtime) flushResult() error { result := r.pending if result == nil { diff --git a/backend/services/outbox/store.go b/backend/services/outbox/store.go index c6c4404..d55f44f 100644 --- a/backend/services/outbox/store.go +++ b/backend/services/outbox/store.go @@ -2,8 +2,10 @@ package outbox import ( "database/sql" + "encoding/json" "errors" "fmt" + "net/mail" "sort" "strings" "time" @@ -12,6 +14,7 @@ import ( "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/auth" "git.pcux.cn/xuans/Smail/services/logstore" "git.pcux.cn/xuans/Smail/utils/send" "git.pcux.cn/xuans/Smail/utils/smtp" @@ -34,6 +37,32 @@ func Enqueue(session *xorm.Session, email models.Email) error { if err != nil { return err } + return enqueueRecipients(session, email, message, "") +} + +type Submission struct { + EnvelopeFrom string `json:"envelope_from"` + Date string `json:"date"` + Trigger string `json:"trigger"` + Authorization auth.QueuedAuthorization `json:"authorization"` +} + +func EnqueueSMTP(session *xorm.Session, email models.Email, recipients []string, submission Submission) error { + if submission.Authorization.Kind == "" || submission.Authorization.ID <= 0 || submission.Authorization.UserID != email.SendUserID || submission.Trigger == "" { + return errors.New("SMTP submission authorization is missing") + } + encoded, err := json.Marshal(submission) + if err != nil { + return err + } + message := &parsemail.Email{To: make([]*parsemail.User, len(recipients))} + for index, address := range recipients { + message.To[index] = &parsemail.User{EmailAddress: address} + } + return enqueueRecipients(session, email, message, string(encoded)) +} + +func enqueueRecipients(session *xorm.Session, email models.Email, message *parsemail.Email, submission string) error { seen := make(map[string]bool) for _, user := range append(append(message.To, message.Cc...), message.Bcc...) { if user == nil || user.EmailAddress == "" { @@ -41,6 +70,10 @@ func Enqueue(session *xorm.Session, email models.Email) error { } // 本地部分保留大小写,仅合并完全相同的 SMTP 信封地址。 address := user.EmailAddress + parsed, err := mail.ParseAddress(address) + if err != nil || parsed.Address != address || len(address) > 320 { + return errors.New("invalid queue recipient") + } if at := strings.LastIndexByte(address, '@'); at >= 0 { address = address[:at+1] + strings.ToLower(address[at+1:]) } @@ -48,7 +81,7 @@ func Enqueue(session *xorm.Session, email models.Email) error { continue } seen[address] = true - job := models.OutboundDelivery{EmailID: email.Id, Recipient: address, State: queued} + job := models.OutboundDelivery{EmailID: email.Id, Recipient: address, State: queued, Submission: submission} if _, err := session.Insert(&job); err != nil { return err } diff --git a/backend/services/outbox/store_test.go b/backend/services/outbox/store_test.go index e5be4c9..6ff9804 100644 --- a/backend/services/outbox/store_test.go +++ b/backend/services/outbox/store_test.go @@ -2,6 +2,7 @@ package outbox import ( stdcontext "context" + "encoding/json" "errors" "io" "net/textproto" @@ -10,10 +11,13 @@ import ( "testing" "time" + "git.pcux.cn/xuans/Smail/config" "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/auth" + "git.pcux.cn/xuans/Smail/services/credentials" "git.pcux.cn/xuans/Smail/utils/context" "git.pcux.cn/xuans/Smail/utils/smtp" "xorm.io/xorm" @@ -57,6 +61,131 @@ func queueFixture(t *testing.T) (*xorm.Engine, models.Email, []models.OutboundDe return engine, email, jobs } +func smtpQueueFixture(t *testing.T) (*xorm.Engine, []models.OutboundDelivery, *credentials.Service) { + t.Helper() + engine, email, jobs := queueFixture(t) + oldDB, oldConfig := db.Instance, config.Instance + db.Instance = engine + config.Instance = &config.Config{Domain: "example.com", Domains: []string{"example.com"}} + t.Cleanup(func() { db.Instance, config.Instance = oldDB, oldConfig }) + if err := engine.Sync2(&models.ProtocolCredential{}, &models.ClientCredential{}); err != nil { + t.Fatal(err) + } + service, err := credentials.NewService(engine) + if err != nil { + t.Fatal(err) + } + if err := service.SetProtocolEnabled(email.SendUserID, "smtp", true); err != nil { + t.Fatal(err) + } + code, err := service.GenerateProtocolCode(email.SendUserID, "smtp") + if err != nil { + t.Fatal(err) + } + _, session, err := auth.AuthenticateProtocolSession("smtp", "192.0.2.95:1234", "sender@example.com", code) + if err != nil { + t.Fatal(err) + } + authorization, err := session.QueueAuthorization() + if err != nil { + t.Fatal(err) + } + submission, err := json.Marshal(Submission{ + EnvelopeFrom: "sender@example.com", Date: "2026-09-27T12:30:00+08:00", + Trigger: "SMTP", Authorization: authorization, + }) + if err != nil { + t.Fatal(err) + } + email.Sender = `{"EmailAddress":"sender@example.com"}` + if _, err := engine.ID(email.Id).Cols("sender").Update(&email); err != nil { + t.Fatal(err) + } + for index := range jobs { + jobs[index].Submission = string(submission) + if _, err := engine.ID(jobs[index].ID).Cols("submission").Update(&jobs[index]); err != nil { + t.Fatal(err) + } + } + return engine, jobs, service +} + +func TestSMTPQueueRestoresEnvelopeWithoutRewritingHeaders(t *testing.T) { + engine, jobs, _ := smtpQueueFixture(t) + calls := 0 + r := &Runtime{engine: engine, deliver: func(ctx *context.Context, message *parsemail.Email, recipient string) error { + calls++ + if ctx.UserAccount != "sender" || recipient != jobs[0].Recipient || message.EnvelopeFrom != "sender@example.com" { + t.Fatal("worker did not restore SMTP identity or envelope") + } + if len(message.EnvelopeTo) != 1 || message.EnvelopeTo[0].EmailAddress != recipient || len(message.To) != 2 { + t.Fatal("envelope recipients leaked into message headers") + } + if message.MsgID != "stable@example.com" || message.Date != "2026-09-27T12:30:00+08:00" || message.DeliveryTrigger != "SMTP" { + t.Fatal("worker lost immutable submission metadata") + } + return nil + }} + if err := r.process(jobs[0]); err != nil { + t.Fatal(err) + } + if calls != 1 { + t.Fatalf("deliveries = %d, want 1", calls) + } +} + +func TestSMTPQueueRejectsStaleAuthorizationBeforeDelivery(t *testing.T) { + for _, change := range []string{"revoke", "disable", "rotate", "malformed", "sender", "database_failure"} { + t.Run(change, func(t *testing.T) { + engine, jobs, service := smtpQueueFixture(t) + var err error + switch change { + case "revoke": + err = service.RevokeProtocolCode(1, "smtp") + case "disable": + err = service.SetProtocolEnabled(1, "smtp", false) + case "rotate": + _, err = service.GenerateProtocolCode(1, "smtp") + case "malformed": + jobs[0].Submission = "{" + case "sender": + var submission Submission + if err := json.Unmarshal([]byte(jobs[0].Submission), &submission); err != nil { + t.Fatal(err) + } + submission.EnvelopeFrom = "other@example.com" + var encoded []byte + encoded, err = json.Marshal(submission) + jobs[0].Submission = string(encoded) + case "database_failure": + _, err = engine.Exec("ALTER TABLE protocol_credential RENAME TO unavailable_credential") + } + if err != nil { + t.Fatal(err) + } + r := &Runtime{engine: engine, deliver: func(*context.Context, *parsemail.Email, string) error { + t.Fatal("worker delivered without current authorization") + return nil + }} + err = r.process(jobs[0]) + if (change == "database_failure") != (err != nil) { + t.Fatalf("process error = %v", err) + } + var stored models.OutboundDelivery + if found, err := engine.ID(jobs[0].ID).Get(&stored); err != nil || !found { + t.Fatalf("read queued job: found=%v err=%v", found, err) + } + if change == "database_failure" { + if stored.State != queued || stored.Attempts != 0 { + t.Fatal("database read failure consumed a delivery attempt") + } + } else if stored.State != failed { + t.Fatalf("revoked submission state = %s", stored.State) + } + }) + } +} + func assertStatus(t *testing.T, engine *xorm.Engine, emailID int, status int8) { t.Helper() var email models.Email diff --git a/backend/utils/send/send.go b/backend/utils/send/send.go index f82dc43..0588cf9 100644 --- a/backend/utils/send/send.go +++ b/backend/utils/send/send.go @@ -95,7 +95,15 @@ func SendTriggered(ctx *context.Context, e *parsemail.Email, trigger string) (er // SendRecipientTriggered 保留原邮件头,只向当前队列收件人发送 SMTP 信封。 func SendRecipientTriggered(ctx *context.Context, e *parsemail.Email, recipient string) error { + if e.DeliveryTrigger != "" { + err, failures := sendRecipients(ctx, e, []*parsemail.User{{EmailAddress: recipient}}, e.DeliveryTrigger) + return recipientDeliveryError(err, failures) + } err, failures := sendRecipients(ctx, e, []*parsemail.User{{EmailAddress: recipient}}, "网页写信") + return recipientDeliveryError(err, failures) +} + +func recipientDeliveryError(err error, failures map[string]error) error { if err != nil { for _, cause := range failures { if cause != nil { @@ -122,7 +130,11 @@ func sendRecipients(ctx *context.Context, e *parsemail.Email, recipients []*pars to := deliveryRecipients(e, recipients) - return doSend(ctx, fromDomain, b, to, e.From.EmailAddress, trigger) + from := e.From.EmailAddress + if e.EnvelopeTo != nil { + from = e.EnvelopeFrom + } + return doSend(ctx, fromDomain, b, to, from, trigger) }