将SMTP信封收件人与授权快照持久保存，按收件人投递并在执行前复核凭据状态；保留原邮件日期，新增队列元数据迁移并补齐相关回归用例源码。
This commit is contained in:
parent
29eb85b4e6
commit
866504d7a0
@ -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: {
|
||||
|
||||
@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -77,6 +77,7 @@ type Email struct {
|
||||
ThreadReferences string
|
||||
EnvelopeFrom string
|
||||
EnvelopeTo []*User `json:"-"`
|
||||
DeliveryTrigger string `json:"-"`
|
||||
ReplyTo []*User
|
||||
From *User
|
||||
To []*User
|
||||
|
||||
@ -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
|
||||
})
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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" }
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user