Add configurable external rules pipeline for ingress #19

Merged
thuanle merged 8 commits from feat/external-rules-pipeline into main 2026-04-28 07:14:45 +07:00
9 changed files with 69 additions and 31 deletions
Showing only changes of commit aa05e3f7c0 - Show all commits
+16 -1
View File
@@ -73,7 +73,7 @@ func (d *Dispatcher) Dispatch(task *database.Task) {
SessionID: task.ThreadID, SessionID: task.ThreadID,
History: history, History: history,
CallbackURL: callbackURL, CallbackURL: callbackURL,
Metadata: map[string]string{"task_uuid": task.TaskUUID}, Metadata: buildDispatchMetadata(task),
} }
// Update status to AI_PROCESSING before first attempt. // Update status to AI_PROCESSING before first attempt.
@@ -195,3 +195,18 @@ func (d *Dispatcher) buildHistory(threadID string) ([]historyEntry, error) {
return history, nil return history, nil
} }
// buildDispatchMetadata constructs the metadata map for OpenClaw dispatch,
// merging the task UUID with any DispatchContext from the rules pipeline.
func buildDispatchMetadata(task *database.Task) map[string]string {
metadata := map[string]string{"task_uuid": task.TaskUUID}
if task.DispatchContext != "" {
var ctx map[string]string
if err := json.Unmarshal([]byte(task.DispatchContext), &ctx); err == nil {
for k, v := range ctx {
metadata[k] = v
}
}
}
return metadata
}
+9
View File
@@ -31,6 +31,7 @@ type Config struct {
BridgeCallbackToken string BridgeCallbackToken string
SystemEmail string SystemEmail string
WhitelistEmails []string WhitelistEmails []string
BlocklistEmails []string
// Optional // Optional
IMAPProxyURL string IMAPProxyURL string
@@ -74,6 +75,14 @@ func Load() (*Config, error) {
} }
} }
if raw := os.Getenv("BLOCKLIST_EMAILS"); raw != "" {
for _, email := range strings.Split(raw, ",") {
if trimmed := strings.TrimSpace(email); trimmed != "" {
cfg.BlocklistEmails = append(cfg.BlocklistEmails, trimmed)
}
}
}
if cfg.ListenAddr == "" { if cfg.ListenAddr == "" {
cfg.ListenAddr = ":8080" cfg.ListenAddr = ":8080"
} }
+1
View File
@@ -24,6 +24,7 @@ type Task struct {
Sender string Sender string
Subject string Subject string
BodyPlain string BodyPlain string
DispatchContext string // JSON metadata from external rules pipeline.
AIResponse string AIResponse string
Status string `gorm:"index;not null"` Status string `gorm:"index;not null"`
AttemptOpenClaw int `gorm:"default:0"` AttemptOpenClaw int `gorm:"default:0"`
+36 -23
View File
@@ -3,6 +3,7 @@ package mail
import ( import (
"context" "context"
"crypto/tls" "crypto/tls"
"encoding/json"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
@@ -21,6 +22,7 @@ import (
"thuanle.me/claw-email-bridge/internal/config" "thuanle.me/claw-email-bridge/internal/config"
"thuanle.me/claw-email-bridge/internal/database" "thuanle.me/claw-email-bridge/internal/database"
"thuanle.me/claw-email-bridge/internal/logging" "thuanle.me/claw-email-bridge/internal/logging"
"thuanle.me/claw-email-bridge/internal/rules"
) )
// IMAPWatcher monitors an IMAP mailbox via IDLE and processes new emails. // IMAPWatcher monitors an IMAP mailbox via IDLE and processes new emails.
@@ -32,6 +34,8 @@ type IMAPWatcher struct {
dialIMAP func(addr string, options *imapclient.Options) (*imapclient.Client, error) dialIMAP func(addr string, options *imapclient.Options) (*imapclient.Client, error)
dialIMAPViaProxy func(addr, proxyURL string, options *imapclient.Options) (*imapclient.Client, error) dialIMAPViaProxy func(addr, proxyURL string, options *imapclient.Options) (*imapclient.Client, error)
rules *rules.Pipeline
// onReceived is called after a task is saved as RECEIVED. // onReceived is called after a task is saved as RECEIVED.
// This will be wired to the OpenClaw dispatch in a future step. // This will be wired to the OpenClaw dispatch in a future step.
OnReceived func(task *database.Task) OnReceived func(task *database.Task)
@@ -57,11 +61,20 @@ var socks5DialerFactory = proxy.SOCKS5
// NewIMAPWatcher creates a new IMAP watcher. // NewIMAPWatcher creates a new IMAP watcher.
func NewIMAPWatcher(cfg *config.Config, db *gorm.DB) *IMAPWatcher { func NewIMAPWatcher(cfg *config.Config, db *gorm.DB) *IMAPWatcher {
var externalRules []rules.Rule
if len(cfg.WhitelistEmails) > 0 {
externalRules = append(externalRules, rules.NewWhitelistRule(cfg.WhitelistEmails))
}
if len(cfg.BlocklistEmails) > 0 {
externalRules = append(externalRules, rules.NewBlocklistRule(cfg.BlocklistEmails))
}
return &IMAPWatcher{ return &IMAPWatcher{
cfg: cfg, cfg: cfg,
db: db, db: db,
dialIMAP: imapclient.DialTLS, dialIMAP: imapclient.DialTLS,
dialIMAPViaProxy: dialTLSViaSOCKS5, dialIMAPViaProxy: dialTLSViaSOCKS5,
rules: rules.NewPipeline(externalRules),
} }
} }
@@ -343,10 +356,15 @@ func (w *IMAPWatcher) processMessage(c *imapclient.Client, buf *imapclient.Fetch
return return
} }
// 2. Whitelist: skip if sender not in allowed list. // 2. External rules: evaluate operator-configurable policy.
if !w.isWhitelisted(sender) { ruleResult := w.rules.Evaluate(rules.EmailContext{
log.Info("imap: sender not whitelisted, ignoring", "sender", sender) Sender: sender,
w.saveIgnored(messageID, sender, subject, "not whitelisted") Subject: subject,
MessageID: messageID,
})
if !ruleResult.Accepted {
log.Info("imap: rejected by external rules", "sender", sender, "reason", ruleResult.Reason)
w.saveIgnored(messageID, sender, subject, ruleResult.Reason)
w.markSeen(c, buf.UID) w.markSeen(c, buf.UID)
return return
} }
@@ -404,13 +422,14 @@ func (w *IMAPWatcher) processMessage(c *imapclient.Client, buf *imapclient.Fetch
// 5. Save task with status RECEIVED. // 5. Save task with status RECEIVED.
taskUUID := uuid.New().String() taskUUID := uuid.New().String()
task := &database.Task{ task := &database.Task{
TaskUUID: taskUUID, TaskUUID: taskUUID,
ThreadID: threadID, ThreadID: threadID,
MessageID: messageID, MessageID: messageID,
Sender: sender, Sender: sender,
Subject: subject, Subject: subject,
BodyPlain: bodyPlain, BodyPlain: bodyPlain,
Status: database.StatusReceived, DispatchContext: marshalDispatchContext(ruleResult.Metadata),
Status: database.StatusReceived,
} }
taskLog := logging.TaskLogger(taskUUID, threadID, messageID) taskLog := logging.TaskLogger(taskUUID, threadID, messageID)
@@ -453,19 +472,13 @@ func (w *IMAPWatcher) saveIgnored(messageID, sender, subject, reason string) {
} }
} }
// isWhitelisted checks if the sender is in the allowed list. // marshalDispatchContext serializes rule metadata to JSON.
// Empty whitelist means all senders are allowed. func marshalDispatchContext(metadata map[string]string) string {
func (w *IMAPWatcher) isWhitelisted(sender string) bool { if len(metadata) == 0 {
if len(w.cfg.WhitelistEmails) == 0 { return ""
return true
} }
lower := strings.ToLower(sender) data, _ := json.Marshal(metadata)
for _, email := range w.cfg.WhitelistEmails { return string(data)
if strings.ToLower(email) == lower {
return true
}
}
return false
} }
// markSeen flags a message as \Seen in IMAP. // markSeen flags a message as \Seen in IMAP.
+1 -1
View File
@@ -27,4 +27,4 @@ func (p *Pipeline) Evaluate(ctx EmailContext) RuleResult {
} }
} }
return merged return merged
} }
+3 -3
View File
@@ -95,12 +95,12 @@ type stubRule struct {
result RuleResult result RuleResult
} }
func (s *stubRule) Name() string { return s.name } func (s *stubRule) Name() string { return s.name }
func (s *stubRule) Evaluate(_ EmailContext) RuleResult { return s.result } func (s *stubRule) Evaluate(_ EmailContext) RuleResult { return s.result }
type trackingRule struct { type trackingRule struct {
called bool called bool
} }
func (t *trackingRule) Name() string { return "tracker" } func (t *trackingRule) Name() string { return "tracker" }
func (t *trackingRule) Evaluate(_ EmailContext) RuleResult { t.called = true; return Accept() } func (t *trackingRule) Evaluate(_ EmailContext) RuleResult { t.called = true; return Accept() }
+1 -1
View File
@@ -38,4 +38,4 @@ func (r RuleResult) WithMetadata(key, value string) RuleResult {
type Rule interface { type Rule interface {
Name() string Name() string
Evaluate(ctx EmailContext) RuleResult Evaluate(ctx EmailContext) RuleResult
} }
+1 -1
View File
@@ -39,4 +39,4 @@ func TestEmailContext_Fields(t *testing.T) {
if ctx.Sender != "user@example.com" { if ctx.Sender != "user@example.com" {
t.Fatal("sender mismatch") t.Fatal("sender mismatch")
} }
} }
+1 -1
View File
@@ -27,4 +27,4 @@ func (r *WhitelistRule) Evaluate(ctx EmailContext) RuleResult {
return Accept() return Accept()
} }
return Reject("not whitelisted") return Reject("not whitelisted")
} }