Merge pull request 'Add configurable external rules pipeline for ingress' (#19) from feat/external-rules-pipeline into main
Reviewed-on: #19 Reviewed-by: codex <39+codex@noreply.localhost>
This commit was merged in pull request #19.
This commit is contained in:
@@ -18,6 +18,7 @@ OPENCLAW_API_KEY=
|
|||||||
BRIDGE_CALLBACK_TOKEN=a-strong-random-token
|
BRIDGE_CALLBACK_TOKEN=a-strong-random-token
|
||||||
SYSTEM_EMAIL=bridge@example.com
|
SYSTEM_EMAIL=bridge@example.com
|
||||||
WHITELIST_EMAILS=user1@example.com,user2@example.com
|
WHITELIST_EMAILS=user1@example.com,user2@example.com
|
||||||
|
BLOCKLIST_EMAILS=
|
||||||
|
|
||||||
# === Optional ===
|
# === Optional ===
|
||||||
IMAP_PROXY_URL=
|
IMAP_PROXY_URL=
|
||||||
|
|||||||
@@ -3,7 +3,7 @@
|
|||||||
Tự động xử lý email bằng OpenClaw AI theo luồng:
|
Tự động xử lý email bằng OpenClaw AI theo luồng:
|
||||||
|
|
||||||
```
|
```
|
||||||
Ingress (IMAP) → Dispatch (OpenClaw) → Callback → Egress (SMTP reply)
|
Ingress (IMAP) → Safety Checks → External Rules → Apply Context → Dispatch (OpenClaw) → Callback → Egress (SMTP reply)
|
||||||
```
|
```
|
||||||
|
|
||||||
## Tính năng
|
## Tính năng
|
||||||
@@ -12,6 +12,7 @@ Ingress (IMAP) → Dispatch (OpenClaw) → Callback → Egress (SMTP reply)
|
|||||||
- **Threading** — giữ nguyên email thread qua `In-Reply-To` / `References`
|
- **Threading** — giữ nguyên email thread qua `In-Reply-To` / `References`
|
||||||
- **Idempotency** — chống duplicate theo `message_id`
|
- **Idempotency** — chống duplicate theo `message_id`
|
||||||
- **Anti-loop & Whitelist** — bỏ qua email từ chính hệ thống, chỉ xử lý sender được phép
|
- **Anti-loop & Whitelist** — bỏ qua email từ chính hệ thống, chỉ xử lý sender được phép
|
||||||
|
- **External Rules** — pipeline configurable: whitelist, blocklist, và rule metadata cho dispatch
|
||||||
- **Retry** — tối đa 3 lần với backoff 1s → 5s → 15s cho cả OpenClaw và SMTP
|
- **Retry** — tối đa 3 lần với backoff 1s → 5s → 15s cho cả OpenClaw và SMTP
|
||||||
- **Stateful** — SQLite lưu trạng thái pipeline, không mất dữ liệu khi restart
|
- **Stateful** — SQLite lưu trạng thái pipeline, không mất dữ liệu khi restart
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,30 @@
|
|||||||
|
---
|
||||||
|
type: ADR
|
||||||
|
id: "0002"
|
||||||
|
title: "Ingress uses a configurable external rules pipeline before dispatch"
|
||||||
|
status: active
|
||||||
|
date: 2026-04-28
|
||||||
|
---
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
The IMAP ingress flow previously hardcoded whitelist and idempotency checks directly in processMessage. There was no extension point for operator-defined policy such as sender blocking or context-based routing. The flow mixed internal safety rules (anti-loop) with configurable policy (whitelist).
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
Separate internal safety checks (anti-loop, idempotency) from external configurable rules. External rules run in a pipeline after safety checks. Each rule implements a `Rule` interface that evaluates `EmailContext` and returns `RuleResult` (accept/reject + metadata). Rule metadata is persisted as `DispatchContext` on the Task and passed to OpenClaw dispatch.
|
||||||
|
|
||||||
|
Pipeline order: anti-loop (internal) → idempotency (internal) → external rules → threading → save → dispatch.
|
||||||
|
|
||||||
|
## Options considered
|
||||||
|
|
||||||
|
- **Pipeline with Rule interface** (chosen): extensible, testable, each rule is isolated.
|
||||||
|
- **Keep everything in processMessage**: simpler but not configurable or testable.
|
||||||
|
- **Plugin-based rules via config files**: more flexible but over-engineered for current needs.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
New rules can be added by implementing the Rule interface and registering in config.
|
||||||
|
Whitelist and blocklist are now env-configurable and independently testable.
|
||||||
|
Future rules (domain routing, context selection) fit the same pipeline.
|
||||||
|
Task model has a new DispatchContext column (auto-migrated).
|
||||||
@@ -73,3 +73,4 @@ Optional. Capture important external input or review notes.
|
|||||||
| ID | Title | Status |
|
| ID | Title | Status |
|
||||||
|----|-------|--------|
|
|----|-------|--------|
|
||||||
| 0001 | IMAP thread resolution prefers In-Reply-To and falls back to References | active |
|
| 0001 | IMAP thread resolution prefers In-Reply-To and falls back to References | active |
|
||||||
|
| 0002 | Ingress uses a configurable external rules pipeline before dispatch | active |
|
||||||
|
|||||||
@@ -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,25 @@ func (d *Dispatcher) buildHistory(threadID string) ([]historyEntry, error) {
|
|||||||
|
|
||||||
return history, nil
|
return history, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExportBuildDispatchMetadata exports buildDispatchMetadata for testing.
|
||||||
|
func ExportBuildDispatchMetadata(task *database.Task) map[string]string {
|
||||||
|
return buildDispatchMetadata(task)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
|
if k != "task_uuid" {
|
||||||
|
metadata[k] = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return metadata
|
||||||
|
}
|
||||||
|
|||||||
@@ -12,6 +12,53 @@ import (
|
|||||||
"thuanle.me/claw-email-bridge/internal/database"
|
"thuanle.me/claw-email-bridge/internal/database"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func TestBuildDispatchMetadata_ForwardsContext(t *testing.T) {
|
||||||
|
task := &database.Task{
|
||||||
|
TaskUUID: "uuid-ctx-test",
|
||||||
|
DispatchContext: `{"rule":"whitelist","source":"env"}`,
|
||||||
|
}
|
||||||
|
metadata := ai_client.ExportBuildDispatchMetadata(task)
|
||||||
|
|
||||||
|
if metadata["task_uuid"] != "uuid-ctx-test" {
|
||||||
|
t.Errorf("expected task_uuid=uuid-ctx-test, got %q", metadata["task_uuid"])
|
||||||
|
}
|
||||||
|
if metadata["rule"] != "whitelist" {
|
||||||
|
t.Errorf("expected rule=whitelist, got %q", metadata["rule"])
|
||||||
|
}
|
||||||
|
if metadata["source"] != "env" {
|
||||||
|
t.Errorf("expected source=env, got %q", metadata["source"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildDispatchMetadata_ProtectsTaskUUID(t *testing.T) {
|
||||||
|
task := &database.Task{
|
||||||
|
TaskUUID: "original-uuid",
|
||||||
|
DispatchContext: `{"task_uuid":"malicious-override","extra":"data"}`,
|
||||||
|
}
|
||||||
|
metadata := ai_client.ExportBuildDispatchMetadata(task)
|
||||||
|
|
||||||
|
if metadata["task_uuid"] != "original-uuid" {
|
||||||
|
t.Errorf("task_uuid should not be overwritten, got %q", metadata["task_uuid"])
|
||||||
|
}
|
||||||
|
if metadata["extra"] != "data" {
|
||||||
|
t.Errorf("expected extra=data, got %q", metadata["extra"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildDispatchMetadata_NoContext(t *testing.T) {
|
||||||
|
task := &database.Task{
|
||||||
|
TaskUUID: "uuid-no-ctx",
|
||||||
|
}
|
||||||
|
metadata := ai_client.ExportBuildDispatchMetadata(task)
|
||||||
|
|
||||||
|
if len(metadata) != 1 {
|
||||||
|
t.Errorf("expected 1 key, got %d", len(metadata))
|
||||||
|
}
|
||||||
|
if metadata["task_uuid"] != "uuid-no-ctx" {
|
||||||
|
t.Errorf("expected task_uuid=uuid-no-ctx, got %q", metadata["task_uuid"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Test matrix #2: OpenClaw fail 2 lần, lần 3 thành công → COMPLETED, attempt_openclaw = 3.
|
// Test matrix #2: OpenClaw fail 2 lần, lần 3 thành công → COMPLETED, attempt_openclaw = 3.
|
||||||
func TestDispatch_RetryThenSuccess(t *testing.T) {
|
func TestDispatch_RetryThenSuccess(t *testing.T) {
|
||||||
tdb := database.NewTestDB(t)
|
tdb := database.NewTestDB(t)
|
||||||
|
|||||||
@@ -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"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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"`
|
||||||
|
|||||||
+41
-28
@@ -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,15 +356,7 @@ func (w *IMAPWatcher) processMessage(c *imapclient.Client, buf *imapclient.Fetch
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// 2. Whitelist: skip if sender not in allowed list.
|
// 2. Idempotency: skip if message_id already exists.
|
||||||
if !w.isWhitelisted(sender) {
|
|
||||||
log.Info("imap: sender not whitelisted, ignoring", "sender", sender)
|
|
||||||
w.saveIgnored(messageID, sender, subject, "not whitelisted")
|
|
||||||
w.markSeen(c, buf.UID)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// 3. Idempotency: skip if message_id already exists.
|
|
||||||
existing, err := database.FindByMessageID(w.db, messageID)
|
existing, err := database.FindByMessageID(w.db, messageID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("imap: db lookup failed", "error", err)
|
log.Error("imap: db lookup failed", "error", err)
|
||||||
@@ -363,6 +368,19 @@ func (w *IMAPWatcher) processMessage(c *imapclient.Client, buf *imapclient.Fetch
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 3. External rules: evaluate operator-configurable policy.
|
||||||
|
ruleResult := w.rules.Evaluate(rules.EmailContext{
|
||||||
|
Sender: sender,
|
||||||
|
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)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// 4. Threading: determine thread_id.
|
// 4. Threading: determine thread_id.
|
||||||
threadID := messageID
|
threadID := messageID
|
||||||
inReplyTo := ""
|
inReplyTo := ""
|
||||||
@@ -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.
|
||||||
|
|||||||
@@ -0,0 +1,26 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
import "strings"
|
||||||
|
|
||||||
|
// BlocklistRule rejects emails from configured senders.
|
||||||
|
type BlocklistRule struct {
|
||||||
|
emails map[string]bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewBlocklistRule creates a blocklist rule from a list of email addresses.
|
||||||
|
func NewBlocklistRule(emails []string) *BlocklistRule {
|
||||||
|
m := make(map[string]bool, len(emails))
|
||||||
|
for _, e := range emails {
|
||||||
|
m[strings.ToLower(e)] = true
|
||||||
|
}
|
||||||
|
return &BlocklistRule{emails: m}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *BlocklistRule) Name() string { return "blocklist" }
|
||||||
|
|
||||||
|
func (r *BlocklistRule) Evaluate(ctx EmailContext) RuleResult {
|
||||||
|
if r.emails[strings.ToLower(ctx.Sender)] {
|
||||||
|
return Reject("blocked sender")
|
||||||
|
}
|
||||||
|
return Accept()
|
||||||
|
}
|
||||||
@@ -0,0 +1,35 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
func TestBlocklistRule_SenderBlocked_Rejects(t *testing.T) {
|
||||||
|
r := NewBlocklistRule([]string{"spam@test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "spam@test.com"})
|
||||||
|
if result.Accepted {
|
||||||
|
t.Fatal("expected reject for blocked sender")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBlocklistRule_SenderNotBlocked_Accepts(t *testing.T) {
|
||||||
|
r := NewBlocklistRule([]string{"spam@test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "ok@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("expected accept for non-blocked sender")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBlocklistRule_EmptyList_Accepts(t *testing.T) {
|
||||||
|
r := NewBlocklistRule(nil)
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "anyone@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("empty blocklist should accept all")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBlocklistRule_CaseInsensitive(t *testing.T) {
|
||||||
|
r := NewBlocklistRule([]string{"Spam@Test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "spam@test.com"})
|
||||||
|
if result.Accepted {
|
||||||
|
t.Fatal("blocklist should be case-insensitive")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
// Pipeline evaluates a sequence of rules, stopping on first rejection.
|
||||||
|
type Pipeline struct {
|
||||||
|
rules []Rule
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewPipeline creates a rules pipeline from the given rules.
|
||||||
|
func NewPipeline(rules []Rule) *Pipeline {
|
||||||
|
return &Pipeline{rules: rules}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Evaluate runs all rules in order. Returns the first rejection or a merged accept.
|
||||||
|
func (p *Pipeline) Evaluate(ctx EmailContext) RuleResult {
|
||||||
|
merged := Accept()
|
||||||
|
for _, r := range p.rules {
|
||||||
|
result := r.Evaluate(ctx)
|
||||||
|
if !result.Accepted {
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
for k, v := range result.Metadata {
|
||||||
|
if merged.Metadata == nil {
|
||||||
|
merged.Metadata = map[string]string{k: v}
|
||||||
|
} else {
|
||||||
|
merged.Metadata[k] = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return merged
|
||||||
|
}
|
||||||
@@ -0,0 +1,106 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestPipeline_NoRules_Accepts(t *testing.T) {
|
||||||
|
p := NewPipeline(nil)
|
||||||
|
result := p.Evaluate(EmailContext{Sender: "anyone@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("expected accept with no rules")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPipeline_AllRulesAccept_Accepts(t *testing.T) {
|
||||||
|
allowAll := &stubRule{name: "allow-all", result: Accept()}
|
||||||
|
p := NewPipeline([]Rule{allowAll})
|
||||||
|
result := p.Evaluate(EmailContext{Sender: "a@b.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("expected accept")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPipeline_OneRuleRejects_Rejects(t *testing.T) {
|
||||||
|
rejector := &stubRule{name: "rejector", result: Reject("nope")}
|
||||||
|
p := NewPipeline([]Rule{rejector})
|
||||||
|
result := p.Evaluate(EmailContext{Sender: "a@b.com"})
|
||||||
|
if result.Accepted {
|
||||||
|
t.Fatal("expected reject")
|
||||||
|
}
|
||||||
|
if result.Reason != "nope" {
|
||||||
|
t.Fatalf("expected reason 'nope', got %s", result.Reason)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPipeline_MergesMetadata(t *testing.T) {
|
||||||
|
r1 := &stubRule{name: "r1", result: Accept().WithMetadata("a", "1")}
|
||||||
|
r2 := &stubRule{name: "r2", result: Accept().WithMetadata("b", "2")}
|
||||||
|
p := NewPipeline([]Rule{r1, r2})
|
||||||
|
result := p.Evaluate(EmailContext{})
|
||||||
|
if result.Metadata["a"] != "1" {
|
||||||
|
t.Fatal("missing metadata a")
|
||||||
|
}
|
||||||
|
if result.Metadata["b"] != "2" {
|
||||||
|
t.Fatal("missing metadata b")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPipeline_StopsOnFirstReject(t *testing.T) {
|
||||||
|
rejector := &stubRule{name: "rejector", result: Reject("stop")}
|
||||||
|
neverCalled := &trackingRule{}
|
||||||
|
p := NewPipeline([]Rule{rejector, neverCalled})
|
||||||
|
p.Evaluate(EmailContext{})
|
||||||
|
if neverCalled.called {
|
||||||
|
t.Fatal("expected second rule not to be called after rejection")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWhitelistRule_EmptyList_Accepts(t *testing.T) {
|
||||||
|
r := NewWhitelistRule(nil)
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "anyone@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("empty whitelist should accept all")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWhitelistRule_SenderInList_Accepts(t *testing.T) {
|
||||||
|
r := NewWhitelistRule([]string{"allowed@test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "allowed@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("expected accept for whitelisted sender")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWhitelistRule_SenderNotInList_Rejects(t *testing.T) {
|
||||||
|
r := NewWhitelistRule([]string{"allowed@test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "unknown@test.com"})
|
||||||
|
if result.Accepted {
|
||||||
|
t.Fatal("expected reject for non-whitelisted sender")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWhitelistRule_CaseInsensitive(t *testing.T) {
|
||||||
|
r := NewWhitelistRule([]string{"Allowed@Test.com"})
|
||||||
|
result := r.Evaluate(EmailContext{Sender: "allowed@test.com"})
|
||||||
|
if !result.Accepted {
|
||||||
|
t.Fatal("whitelist should be case-insensitive")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// stubs
|
||||||
|
|
||||||
|
type stubRule struct {
|
||||||
|
name string
|
||||||
|
result RuleResult
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *stubRule) Name() string { return s.name }
|
||||||
|
func (s *stubRule) Evaluate(_ EmailContext) RuleResult { return s.result }
|
||||||
|
|
||||||
|
type trackingRule struct {
|
||||||
|
called bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *trackingRule) Name() string { return "tracker" }
|
||||||
|
func (t *trackingRule) Evaluate(_ EmailContext) RuleResult { t.called = true; return Accept() }
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
// EmailContext contains the data available to rules for evaluation.
|
||||||
|
type EmailContext struct {
|
||||||
|
Sender string
|
||||||
|
Subject string
|
||||||
|
MessageID string
|
||||||
|
}
|
||||||
|
|
||||||
|
// RuleResult is the output of evaluating a single rule.
|
||||||
|
type RuleResult struct {
|
||||||
|
Accepted bool
|
||||||
|
Reason string
|
||||||
|
Metadata map[string]string
|
||||||
|
}
|
||||||
|
|
||||||
|
// Accept returns a passing RuleResult.
|
||||||
|
func Accept() RuleResult {
|
||||||
|
return RuleResult{Accepted: true}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Reject returns a failing RuleResult with a reason.
|
||||||
|
func Reject(reason string) RuleResult {
|
||||||
|
return RuleResult{Accepted: false, Reason: reason}
|
||||||
|
}
|
||||||
|
|
||||||
|
// WithMetadata attaches a key-value pair to the result.
|
||||||
|
func (r RuleResult) WithMetadata(key, value string) RuleResult {
|
||||||
|
if r.Metadata == nil {
|
||||||
|
r.Metadata = map[string]string{key: value}
|
||||||
|
} else {
|
||||||
|
r.Metadata[key] = value
|
||||||
|
}
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
|
||||||
|
// Rule evaluates an email against operator-configurable policy.
|
||||||
|
type Rule interface {
|
||||||
|
Name() string
|
||||||
|
Evaluate(ctx EmailContext) RuleResult
|
||||||
|
}
|
||||||
@@ -0,0 +1,42 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRuleResult_Accept(t *testing.T) {
|
||||||
|
r := Accept()
|
||||||
|
if !r.Accepted {
|
||||||
|
t.Fatal("expected Accepted=true")
|
||||||
|
}
|
||||||
|
if r.Reason != "" {
|
||||||
|
t.Fatal("expected no reason for accept")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRuleResult_Reject(t *testing.T) {
|
||||||
|
r := Reject("blocked sender")
|
||||||
|
if r.Accepted {
|
||||||
|
t.Fatal("expected Accepted=false")
|
||||||
|
}
|
||||||
|
if r.Reason != "blocked sender" {
|
||||||
|
t.Fatalf("expected reason 'blocked sender', got %s", r.Reason)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRuleResult_WithMetadata(t *testing.T) {
|
||||||
|
r := Accept().WithMetadata("routing", "high-priority")
|
||||||
|
if r.Metadata["routing"] != "high-priority" {
|
||||||
|
t.Fatalf("expected routing=high-priority, got %s", r.Metadata["routing"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEmailContext_Fields(t *testing.T) {
|
||||||
|
ctx := EmailContext{
|
||||||
|
Sender: "user@example.com",
|
||||||
|
Subject: "Test",
|
||||||
|
}
|
||||||
|
if ctx.Sender != "user@example.com" {
|
||||||
|
t.Fatal("sender mismatch")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
package rules
|
||||||
|
|
||||||
|
import "strings"
|
||||||
|
|
||||||
|
// WhitelistRule accepts emails from configured senders.
|
||||||
|
// An empty list accepts all senders.
|
||||||
|
type WhitelistRule struct {
|
||||||
|
emails map[string]bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewWhitelistRule creates a whitelist rule from a list of email addresses.
|
||||||
|
func NewWhitelistRule(emails []string) *WhitelistRule {
|
||||||
|
m := make(map[string]bool, len(emails))
|
||||||
|
for _, e := range emails {
|
||||||
|
m[strings.ToLower(e)] = true
|
||||||
|
}
|
||||||
|
return &WhitelistRule{emails: m}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *WhitelistRule) Name() string { return "whitelist" }
|
||||||
|
|
||||||
|
func (r *WhitelistRule) Evaluate(ctx EmailContext) RuleResult {
|
||||||
|
if len(r.emails) == 0 {
|
||||||
|
return Accept()
|
||||||
|
}
|
||||||
|
if r.emails[strings.ToLower(ctx.Sender)] {
|
||||||
|
return Accept()
|
||||||
|
}
|
||||||
|
return Reject("not whitelisted")
|
||||||
|
}
|
||||||
+11
-5
@@ -47,6 +47,7 @@ Bảng `tasks`:
|
|||||||
- `status` TEXT
|
- `status` TEXT
|
||||||
- `attempt_openclaw` INTEGER DEFAULT 0
|
- `attempt_openclaw` INTEGER DEFAULT 0
|
||||||
- `attempt_smtp` INTEGER DEFAULT 0
|
- `attempt_smtp` INTEGER DEFAULT 0
|
||||||
|
- `dispatch_context` TEXT (JSON metadata từ external rules pipeline, truyền qua OpenClaw dispatch)
|
||||||
- `last_error` TEXT NULL
|
- `last_error` TEXT NULL
|
||||||
- `next_retry_at` DATETIME NULL (chỉ cần nếu retry xử lý theo scheduler/worker)
|
- `next_retry_at` DATETIME NULL (chỉ cần nếu retry xử lý theo scheduler/worker)
|
||||||
- `created_at` DATETIME
|
- `created_at` DATETIME
|
||||||
@@ -68,12 +69,12 @@ Danh sách trạng thái hợp lệ:
|
|||||||
- Nếu có `IMAP_PROXY_URL` thì kết nối IMAP qua proxy; nếu không thì direct.
|
- Nếu có `IMAP_PROXY_URL` thì kết nối IMAP qua proxy; nếu không thì direct.
|
||||||
- Khi có email mới:
|
- Khi có email mới:
|
||||||
1. Anti-loop: nếu `sender` trùng email hệ thống -> `IGNORED`.
|
1. Anti-loop: nếu `sender` trùng email hệ thống -> `IGNORED`.
|
||||||
2. Whitelist: nếu không nằm trong danh sách cho phép -> `IGNORED`.
|
2. Idempotency: nếu `message_id` đã tồn tại -> bỏ qua.
|
||||||
3. Idempotency: nếu `message_id` đã tồn tại -> bỏ qua.
|
3. External rules: chạy pipeline các rule configurable (whitelist, blocklist, v.v.). Nếu reject -> `IGNORED`. Rule metadata được lưu vào `dispatch_context`.
|
||||||
4. Threading:
|
4. Threading:
|
||||||
- Nếu có `In-Reply-To`/`References`: tìm email cha trong DB, kế thừa `thread_id`.
|
- Nếu có `In-Reply-To`/`References`: tìm email cha trong DB, kế thừa `thread_id`.
|
||||||
- Nếu không có: `thread_id = message_id`.
|
- Nếu không có: `thread_id = message_id`.
|
||||||
5. Lưu DB với `status = RECEIVED`.
|
5. Lưu DB với `status = RECEIVED`, bao gồm `dispatch_context` từ rules.
|
||||||
|
|
||||||
4.2 Dispatch (OpenClaw)
|
4.2 Dispatch (OpenClaw)
|
||||||
- Lấy nội dung email hiện tại và history của cùng `thread_id` (chỉ các bản ghi `COMPLETED`).
|
- Lấy nội dung email hiện tại và history của cùng `thread_id` (chỉ các bản ghi `COMPLETED`).
|
||||||
@@ -84,6 +85,7 @@ Danh sách trạng thái hợp lệ:
|
|||||||
- `history` (optional)
|
- `history` (optional)
|
||||||
- `callback_url`
|
- `callback_url`
|
||||||
- `metadata.task_uuid`
|
- `metadata.task_uuid`
|
||||||
|
- `metadata.*` từ `dispatch_context` (rule-derived context)
|
||||||
- Cập nhật `status = AI_PROCESSING`.
|
- Cập nhật `status = AI_PROCESSING`.
|
||||||
|
|
||||||
Retry OpenClaw:
|
Retry OpenClaw:
|
||||||
@@ -146,6 +148,7 @@ Required:
|
|||||||
- `OPENCLAW_URL`, `OPENCLAW_API_KEY` (nếu dùng)
|
- `OPENCLAW_URL`, `OPENCLAW_API_KEY` (nếu dùng)
|
||||||
- `BRIDGE_CALLBACK_TOKEN`
|
- `BRIDGE_CALLBACK_TOKEN`
|
||||||
- `WHITELIST_EMAILS` (comma-separated)
|
- `WHITELIST_EMAILS` (comma-separated)
|
||||||
|
- `BLOCKLIST_EMAILS` (comma-separated, optional)
|
||||||
- `SYSTEM_EMAIL`
|
- `SYSTEM_EMAIL`
|
||||||
- `IMAP_PROXY_URL` (optional)
|
- `IMAP_PROXY_URL` (optional)
|
||||||
|
|
||||||
@@ -166,7 +169,7 @@ Ví dụ:
|
|||||||
3. OpenClaw/SMTP retry đúng thứ tự 1s -> 5s -> 15s.
|
3. OpenClaw/SMTP retry đúng thứ tự 1s -> 5s -> 15s.
|
||||||
4. Hết retry thì `FAILED` và có `last_error`.
|
4. Hết retry thì `FAILED` và có `last_error`.
|
||||||
5. Callback sai token trả 401 và không đổi state task.
|
5. Callback sai token trả 401 và không đổi state task.
|
||||||
6. Anti-loop + whitelist + idempotency hoạt động đúng.
|
6. Anti-loop + idempotency + external rules hoạt động đúng thứ tự.
|
||||||
|
|
||||||
10) Test matrix tối thiểu
|
10) Test matrix tối thiểu
|
||||||
|
|
||||||
@@ -175,4 +178,7 @@ Ví dụ:
|
|||||||
3. SMTP fail hết retry -> `FAILED`, có `last_error`.
|
3. SMTP fail hết retry -> `FAILED`, có `last_error`.
|
||||||
4. Callback sai token -> 401, state không đổi.
|
4. Callback sai token -> 401, state không đổi.
|
||||||
5. Duplicate `message_id` -> không tạo task mới.
|
5. Duplicate `message_id` -> không tạo task mới.
|
||||||
6. Sender là chính hệ thống hoặc ngoài whitelist -> `IGNORED`.
|
6. Sender là chính hệ thống -> `IGNORED`.
|
||||||
|
7. Sender ngoài whitelist -> `IGNORED` bởi external rules.
|
||||||
|
8. Sender trong blocklist -> `IGNORED` bởi external rules.
|
||||||
|
9. Rule metadata được truyền qua dispatch context đến OpenClaw.
|
||||||
|
|||||||
Reference in New Issue
Block a user