From 2993b0b9460dd54b2b78890b2d75e4be56f322f1 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 4 May 2026 18:15:21 +0800 Subject: [PATCH] feat(messenger): enhance dynamic SMTP resolution and identity handling - Implemented dynamic SMTP provider resolution based on user/team context, improving email sending capabilities. - Updated the Send and SendT methods to utilize identity information from the context for dynamic provider selection. - Refactored cloud encryption and decryption methods to delegate to the setting package, streamlining cryptographic operations. - Enhanced team invitation email sending to include identity context, ensuring accurate user/team information is used. - Removed obsolete crypto helper functions, simplifying the codebase and improving maintainability. --- agent/robot/events/delivery.go | 4 + agent/robot/executor/standard/delivery.go | 12 +- messenger/messenger.go | 174 ++++++++++++++++++++-- openapi/setting/cloud.go | 83 +---------- openapi/user/team_invitation.go | 7 +- setting/crypto.go | 58 ++++++++ setting/identity.go | 8 + 7 files changed, 248 insertions(+), 98 deletions(-) create mode 100644 setting/crypto.go create mode 100644 setting/identity.go diff --git a/agent/robot/events/delivery.go b/agent/robot/events/delivery.go index fe1bef9f..6afff685 100644 --- a/agent/robot/events/delivery.go +++ b/agent/robot/events/delivery.go @@ -52,6 +52,10 @@ func (h *robotHandler) handleDelivery(ctx context.Context, ev *eventtypes.Event, return } + if ev.Auth != nil { + ctx = context.WithValue(ctx, "identity", ev.Auth) + } + deliveryCtx := &robottypes.DeliveryContext{ MemberID: payload.MemberID, ExecutionID: payload.ExecutionID, diff --git a/agent/robot/executor/standard/delivery.go b/agent/robot/executor/standard/delivery.go index 96bf9414..1e1f5483 100644 --- a/agent/robot/executor/standard/delivery.go +++ b/agent/robot/executor/standard/delivery.go @@ -7,6 +7,7 @@ import ( "time" "github.com/yaoapp/gou/model" + "github.com/yaoapp/gou/process" kunlog "github.com/yaoapp/kun/log" robotevents "github.com/yaoapp/yao/agent/robot/events" robottypes "github.com/yaoapp/yao/agent/robot/types" @@ -98,7 +99,16 @@ func (e *Executor) pushDeliveryEvent(ctx *robottypes.Context, exec *robottypes.E } } - _, err := event.Push(ctx.Context, robotevents.Delivery, robotevents.DeliveryPayload{ + eventCtx := ctx.Context + if ctx.Auth != nil { + eventCtx = event.WithAuth(eventCtx, &process.AuthorizedInfo{ + UserID: ctx.Auth.UserID, + TeamID: ctx.Auth.TeamID, + Subject: ctx.Auth.Subject, + }) + } + + _, err := event.Push(eventCtx, robotevents.Delivery, robotevents.DeliveryPayload{ ExecutionID: exec.ID, MemberID: exec.MemberID, TeamID: exec.TeamID, diff --git a/messenger/messenger.go b/messenger/messenger.go index 7f82c042..692482af 100644 --- a/messenger/messenger.go +++ b/messenger/messenger.go @@ -20,6 +20,7 @@ import ( "github.com/yaoapp/yao/messenger/providers/twilio" "github.com/yaoapp/yao/messenger/template" "github.com/yaoapp/yao/messenger/types" + "github.com/yaoapp/yao/setting" "github.com/yaoapp/yao/share" ) @@ -50,7 +51,18 @@ func Load(cfg config.Config) error { return err } if !exists { - log.Warn("[Messenger] messengers directory not found, skip loading messenger") + log.Warn("[Messenger] messengers directory not found, creating empty instance for dynamic resolution") + Instance = &Service{ + config: &types.Config{Global: types.GlobalConfig{ + RetryAttempts: 3, + RetryDelay: 2 * time.Second, + Timeout: 30 * time.Second, + }}, + providers: make(map[string]types.Provider), + providersByType: make(map[types.MessageType][]types.Provider), + channels: make(map[string]types.Channel), + defaults: make(map[string]string), + } return nil } @@ -236,8 +248,16 @@ func createMailgunProvider(config types.ProviderConfig) (types.Provider, error) return mailgun.NewMailgunProviderWithTemplateManager(config, template.Global) } -// Send sends a message using the specified channel or default provider +// Send sends a message using the specified channel or default provider. +// If ctx carries an Identity (key "identity"), dynamic SMTP resolution uses +// user/team scope; otherwise system-scope is tried. Falls back to static +// .yao providers when dynamic resolution yields nothing. func (m *Service) Send(ctx context.Context, channel string, message *types.Message) error { + id, _ := ctx.Value("identity").(setting.Identity) + if provider := m.resolveSettingProvider(id, message.Type); provider != nil { + return m.sendViaProvider(ctx, provider, message) + } + m.mutex.RLock() defer m.mutex.RUnlock() @@ -299,43 +319,43 @@ func (m *Service) SendWithProvider(ctx context.Context, providerName string, mes return fmt.Errorf("failed to send message after %d attempts: %w", maxAttempts, lastErr) } -// SendT sends a message using a template +// SendT sends a message using a template. +// Like Send, it tries dynamic SMTP resolution first (via ctx Identity), +// then falls back to static .yao providers. // messageType is optional - if not specified, the first available template type will be used func (m *Service) SendT(ctx context.Context, channel string, templateID string, data types.TemplateData, messageType ...types.MessageType) error { - m.mutex.RLock() - defer m.mutex.RUnlock() - // Determine which message type to use var msgType types.MessageType if len(messageType) > 0 { - // Use specified message type msgType = messageType[0] } else { - // Get available template types and use the first one availableTypes := template.Global.GetAvailableTypes(templateID) if len(availableTypes) == 0 { return fmt.Errorf("template not found: %s", templateID) } - // Convert TemplateType to MessageType msgType = templateTypeToMessageType(availableTypes[0]) } - // Get provider for this channel and message type + id, _ := ctx.Value("identity").(setting.Identity) + if provider := m.resolveSettingProvider(id, msgType); provider != nil { + templateType := messageTypeToTemplateType(msgType) + return provider.SendT(ctx, templateID, templateType, data) + } + + m.mutex.RLock() + defer m.mutex.RUnlock() + providerName := m.getProviderForChannel(channel, string(msgType)) if providerName == "" { return fmt.Errorf("no provider configured for channel %s with message type %s", channel, msgType) } - // Get the provider provider, exists := m.providers[providerName] if !exists { return fmt.Errorf("provider not found: %s", providerName) } - // Convert MessageType back to TemplateType templateType := messageTypeToTemplateType(msgType) - - // Call provider's SendT method return provider.SendT(ctx, templateID, templateType, data) } @@ -909,6 +929,132 @@ func parseChannelsConfig(channelsConfig map[string]interface{}, defaults map[str } } +// resolveSettingProvider tries to build a mailer.Provider from UI-configured +// SMTP settings stored in setting.Global. id may be nil – in that case +// system-scope settings are used. +func (m *Service) resolveSettingProvider(id setting.Identity, messageType types.MessageType) types.Provider { + if messageType != types.MessageTypeEmail { + return nil + } + if setting.Global == nil { + return nil + } + + userID, teamID := "", "" + if id != nil { + userID = id.GetUserID() + teamID = id.GetTeamID() + } + + cfg, err := setting.Global.GetMerged(userID, teamID, "smtp") + if err != nil || cfg == nil { + return nil + } + + if enabled, ok := cfg["enabled"].(bool); ok && !enabled { + return nil + } + + host, _ := cfg["host"].(string) + username, _ := cfg["username"].(string) + password, _ := cfg["password"].(string) + fromEmail, _ := cfg["from_email"].(string) + fromName, _ := cfg["from_name"].(string) + encryption, _ := cfg["encryption"].(string) + + if host == "" || username == "" { + return nil + } + + if password != "" { + password = setting.Decrypt(password) + } + + from := fromEmail + if from == "" { + from = username + } + if fromName != "" { + from = fromName + " <" + from + ">" + } + + useTLS, useSSL := false, false + switch encryption { + case "ssl": + useSSL = true + case "tls", "starttls": + useTLS = true + } + + port := 587 + switch v := cfg["port"].(type) { + case float64: + port = int(v) + case int: + port = v + case int64: + port = int(v) + } + + providerCfg := types.ProviderConfig{ + Name: "setting-smtp", Connector: "mailer", Enabled: true, + Options: map[string]interface{}{ + "smtp": map[string]interface{}{ + "host": host, "port": port, + "username": username, "password": password, + "from": from, "use_tls": useTLS, "use_ssl": useSSL, + }, + }, + } + + provider, err := mailer.NewMailerProvider(providerCfg) + if err != nil { + log.Warn("[Messenger] Failed to create dynamic SMTP provider: %v", err) + return nil + } + return provider +} + +// sendViaProvider validates the message and sends it through the given provider +// with the configured retry logic. +func (m *Service) sendViaProvider(ctx context.Context, provider types.Provider, message *types.Message) error { + if err := m.validateMessage(message); err != nil { + return fmt.Errorf("message validation failed: %w", err) + } + + maxAttempts := m.config.Global.RetryAttempts + if maxAttempts <= 0 { + maxAttempts = 1 + } + + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + select { + case <-ctx.Done(): + return fmt.Errorf("send cancelled: %w", ctx.Err()) + default: + } + + err := provider.Send(ctx, message) + if err == nil { + log.Info("[Messenger] Message sent via dynamic provider (attempt %d/%d)", attempt, maxAttempts) + return nil + } + + lastErr = err + if attempt < maxAttempts { + log.Warn("[Messenger] Dynamic send attempt %d/%d failed: %v", attempt, maxAttempts, err) + select { + case <-ctx.Done(): + return fmt.Errorf("send cancelled during retry: %w", ctx.Err()) + case <-time.After(m.config.Global.RetryDelay): + } + } + } + + return fmt.Errorf("failed to send message via dynamic provider after %d attempts: %w", maxAttempts, lastErr) +} + // validateMessage validates a message before sending func (m *Service) validateMessage(message *types.Message) error { if message == nil { diff --git a/openapi/setting/cloud.go b/openapi/setting/cloud.go index 91d762d1..40554b95 100644 --- a/openapi/setting/cloud.go +++ b/openapi/setting/cloud.go @@ -1,21 +1,14 @@ package setting import ( - "crypto/aes" - "crypto/cipher" - "crypto/rand" - "crypto/sha256" _ "embed" - "encoding/base64" "encoding/json" "fmt" - "io" "net/http" "strings" "time" "github.com/gin-gonic/gin" - "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/authorized" oauthTypes "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/response" @@ -356,40 +349,20 @@ func handleCloudRefresh(c *gin.Context) { } // --------------------------------------------------------------------------- -// Crypto helpers (AES-256-GCM, same scheme as llmprovider) +// Crypto helpers – delegates to setting.Encrypt / setting.Decrypt // --------------------------------------------------------------------------- func cloudEncrypt(plaintext string) string { - secret := config.Conf.DB.AESKey - if secret == "" { - return plaintext - } - enc, err := cloudEncryptString(plaintext, secret) - if err != nil { - return plaintext - } - return cloudEncPrefix + enc + return setting.Encrypt(plaintext) } func cloudDecrypt(value string) string { - if !strings.HasPrefix(value, cloudEncPrefix) { - return value - } - secret := config.Conf.DB.AESKey - if secret == "" { - return strings.TrimPrefix(value, cloudEncPrefix) - } - dec, err := cloudDecryptString(strings.TrimPrefix(value, cloudEncPrefix), secret) - if err != nil { - return value - } - return dec + return setting.Decrypt(value) } // DecryptValue decrypts a value encrypted by cloudEncrypt. -// Delegates to config.DecryptValue for the actual decryption. func DecryptValue(s string) string { - return config.DecryptValue(s) + return setting.Decrypt(s) } func cloudMaskKey(key string) string { @@ -403,51 +376,3 @@ func cloudMaskKey(key string) string { suffix := key[len(key)-cloudMaskChars:] return prefix + "..." + suffix } - -func cloudDeriveKey(secret string) []byte { - h := sha256.Sum256([]byte(secret)) - return h[:] -} - -func cloudEncryptString(plaintext, secret string) (string, error) { - key := cloudDeriveKey(secret) - block, err := aes.NewCipher(key) - if err != nil { - return "", err - } - gcm, err := cipher.NewGCM(block) - if err != nil { - return "", err - } - nonce := make([]byte, gcm.NonceSize()) - if _, err := io.ReadFull(rand.Reader, nonce); err != nil { - return "", err - } - ciphertext := gcm.Seal(nonce, nonce, []byte(plaintext), nil) - return base64.StdEncoding.EncodeToString(ciphertext), nil -} - -func cloudDecryptString(encoded, secret string) (string, error) { - key := cloudDeriveKey(secret) - data, err := base64.StdEncoding.DecodeString(encoded) - if err != nil { - return "", err - } - block, err := aes.NewCipher(key) - if err != nil { - return "", err - } - gcm, err := cipher.NewGCM(block) - if err != nil { - return "", err - } - nonceSize := gcm.NonceSize() - if len(data) < nonceSize { - return "", fmt.Errorf("ciphertext too short") - } - plaintext, err := gcm.Open(nil, data[:nonceSize], data[nonceSize:], nil) - if err != nil { - return "", err - } - return string(plaintext), nil -} diff --git a/openapi/user/team_invitation.go b/openapi/user/team_invitation.go index 6eee5892..a9caa708 100644 --- a/openapi/user/team_invitation.go +++ b/openapi/user/team_invitation.go @@ -21,6 +21,7 @@ import ( messengertypes "github.com/yaoapp/yao/messenger/types" "github.com/yaoapp/yao/openapi/oauth" "github.com/yaoapp/yao/openapi/oauth/authorized" + oauthTypes "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/response" "github.com/yaoapp/yao/openapi/utils" "github.com/yaoapp/yao/share" @@ -1221,11 +1222,9 @@ func teamInvitationCreate(ctx context.Context, userID, teamID string, invitation // Use the saved requestBaseURL and settings (not from invitationData, as they were lost in DB operation) // Send email asynchronously to improve user experience go func() { - // Use background context for async operation bgCtx := context.Background() + bgCtx = context.WithValue(bgCtx, "identity", &oauthTypes.AuthorizedInfo{UserID: userID, TeamID: teamID}) - // Use createdMember data (from database) for email sending - // This ensures we have the actual stored values including properly formatted timestamps emailData := maps.MapStrAny{} for k, v := range createdMember { emailData[k] = v @@ -1364,8 +1363,8 @@ func teamInvitationResend(ctx context.Context, userID, teamID, invitationID, req // Send new invitation email (asynchronously) go func() { - // Use background context for async operation bgCtx := context.Background() + bgCtx = context.WithValue(bgCtx, "identity", &oauthTypes.AuthorizedInfo{UserID: userID, TeamID: teamID}) err := sendTeamInvitationEmail(bgCtx, inviteeEmail, inviterName, teamName, newToken, invitationID, invitationData) if err != nil { log.Error("Failed to resend invitation email: %v", err) diff --git a/setting/crypto.go b/setting/crypto.go new file mode 100644 index 00000000..92aca28a --- /dev/null +++ b/setting/crypto.go @@ -0,0 +1,58 @@ +package setting + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "io" + "strings" + + "github.com/yaoapp/yao/config" +) + +const encPrefix = "enc:" + +// Encrypt encrypts a plaintext string using AES-256-GCM with the configured AES key. +// Returns the original string if no AES key is configured. +func Encrypt(plaintext string) string { + secret := config.Conf.DB.AESKey + if secret == "" { + return plaintext + } + enc, err := aesGCMEncrypt(plaintext, secret) + if err != nil { + return plaintext + } + return encPrefix + enc +} + +// Decrypt decrypts a value previously encrypted by Encrypt. +// Returns the original string if not encrypted or if decryption fails. +func Decrypt(value string) string { + return config.DecryptValue(value) +} + +// IsEncrypted returns true if the value has the encryption prefix. +func IsEncrypted(value string) bool { + return strings.HasPrefix(value, encPrefix) +} + +func aesGCMEncrypt(plaintext, secret string) (string, error) { + h := sha256.Sum256([]byte(secret)) + block, err := aes.NewCipher(h[:]) + if err != nil { + return "", err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return "", err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return "", err + } + ciphertext := gcm.Seal(nonce, nonce, []byte(plaintext), nil) + return base64.StdEncoding.EncodeToString(ciphertext), nil +} diff --git a/setting/identity.go b/setting/identity.go new file mode 100644 index 00000000..3284c115 --- /dev/null +++ b/setting/identity.go @@ -0,0 +1,8 @@ +package setting + +// Identity represents a scoped caller for setting resolution. +// GetMerged(userID, teamID, ns) uses these to cascade: system <- team <- user. +type Identity interface { + GetUserID() string + GetTeamID() string +}