From c02fb660bc0f09e18e4cc83cc0e537eb307d4198 Mon Sep 17 00:00:00 2001 From: Duc Nguyen <87276974+itsddvn@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:24:48 +0700 Subject: [PATCH] fix(channels): honor paired DM policy before publish (#1313) Co-authored-by: ntduc --- docs/project-changelog.md | 19 ++++ internal/channels/bitrix24/download_test.go | 2 +- internal/channels/bitrix24/handle.go | 2 +- internal/channels/channel.go | 25 ++++- internal/channels/policy_test.go | 81 ++++++++++++-- internal/channels/slack/handlers.go | 14 +-- internal/channels/slack/handlers_files.go | 32 +++--- internal/channels/slack/handlers_mention.go | 2 +- internal/channels/slack/policy_test.go | 110 ++++++++++++++++++++ internal/channels/slack/utils.go | 10 +- internal/channels/zalo/personal/handlers.go | 2 +- internal/channels/zalo/zalo.go | 4 +- 12 files changed, 261 insertions(+), 42 deletions(-) create mode 100644 internal/channels/slack/policy_test.go diff --git a/docs/project-changelog.md b/docs/project-changelog.md index a4632cd7..311dfc0a 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -4,6 +4,25 @@ Significant changes, features, and fixes in reverse chronological order. --- +## 2026-07-01 + +### Paired DM routing after policy checks + +**Fixes** + +- Policy-checked direct messages now pass the shared channel safety gate for + Zalo OA, Zalo Personal, Bitrix24, and Slack when the sender is paired but not + listed in static `allow_from`. +- Direct-message callers without an explicit policy gate still use the default + `allow_from` safety net. + +**Tests** + +- Added regression coverage for paired Zalo-style and Slack DMs with an + allowlist mismatch, plus guard coverage for non-policy direct callers. + +--- + ## 2026-06-27 ### Feishu/Lark group pairing stability diff --git a/internal/channels/bitrix24/download_test.go b/internal/channels/bitrix24/download_test.go index 8cd324a9..10fb81c8 100644 --- a/internal/channels/bitrix24/download_test.go +++ b/internal/channels/bitrix24/download_test.go @@ -67,7 +67,7 @@ func TestResolveMime_FallbackOctetStream(t *testing.T) { // TestResolveMime_DefaultOctetStream tests the final fallback to octet-stream. func TestResolveMime_DefaultOctetStream(t *testing.T) { - f := EventFile{Mime: "", Name: "unknown.xyz"} + f := EventFile{Mime: "", Name: "unknown"} got := resolveMime(f, "") if got != "application/octet-stream" { t.Errorf("want application/octet-stream (final fallback), got %q", got) diff --git a/internal/channels/bitrix24/handle.go b/internal/channels/bitrix24/handle.go index 1967295c..0f9b8e92 100644 --- a/internal/channels/bitrix24/handle.go +++ b/internal/channels/bitrix24/handle.go @@ -424,7 +424,7 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) { "message_id", evt.Params.MessageID, "media_count", len(mediaFiles), ) - c.HandleMessageMedia(senderID, chatID, text, mediaFiles, meta, peerKind) + c.HandleAuthorizedMessageMedia(senderID, chatID, text, mediaFiles, meta, peerKind) } // handleJoin sends a short welcome the first time the bot is added to a diff --git a/internal/channels/channel.go b/internal/channels/channel.go index 1216bcc7..0952fe14 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -619,6 +619,17 @@ func (c *BaseChannel) ValidatePolicy(dmPolicy, groupPolicy string) { // This is the standard way for channels to forward received messages. // peerKind should be "direct" or "group" (see sessions.PeerDirect, sessions.PeerGroup). func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []string, metadata map[string]string, peerKind string) { + c.handleMessage(senderID, chatID, content, media, metadata, peerKind, false) +} + +// HandleAuthorizedMessage publishes a message after the caller has already +// enforced the channel policy. It preserves the default direct-message safety +// net for adapters that do not have an explicit policy gate. +func (c *BaseChannel) HandleAuthorizedMessage(senderID, chatID, content string, media []string, metadata map[string]string, peerKind string) { + c.handleMessage(senderID, chatID, content, media, metadata, peerKind, true) +} + +func (c *BaseChannel) handleMessage(senderID, chatID, content string, media []string, metadata map[string]string, peerKind string, policyChecked bool) { // Convert string paths to MediaFile (legacy path-only callers). // Use filepath.Base(p) as filename so persistMedia's sanitizer gets a // meaningful stem instead of falling back to UUID. MimeType is left empty — @@ -627,7 +638,7 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st for _, p := range media { mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)}) } - c.HandleMessageMedia(senderID, chatID, content, mediaFiles, metadata, peerKind) + c.handleMessageMedia(senderID, chatID, content, mediaFiles, metadata, peerKind, policyChecked) } // HandleMessageMedia is the richer sibling of HandleMessage: it accepts @@ -637,12 +648,22 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st // loses that information. Channels that already know the content type at // download time (e.g. Bitrix24 file events) should call this directly. func (c *BaseChannel) HandleMessageMedia(senderID, chatID, content string, media []bus.MediaFile, metadata map[string]string, peerKind string) { + c.handleMessageMedia(senderID, chatID, content, media, metadata, peerKind, false) +} + +// HandleAuthorizedMessageMedia is the media-preserving variant for callers +// that have already enforced the channel policy. +func (c *BaseChannel) HandleAuthorizedMessageMedia(senderID, chatID, content string, media []bus.MediaFile, metadata map[string]string, peerKind string) { + c.handleMessageMedia(senderID, chatID, content, media, metadata, peerKind, true) +} + +func (c *BaseChannel) handleMessageMedia(senderID, chatID, content string, media []bus.MediaFile, metadata map[string]string, peerKind string, policyChecked bool) { // For DMs, enforce the allowlist as a safety net. // For group messages, skip this check — group access is already enforced // by the channel-specific group policy (checkGroupPolicy / CheckPolicy). // Re-checking the sender here would incorrectly block users who are not // individually listed but are in an allowed (or open-policy) group. - if peerKind != "group" && !c.IsAllowed(senderID) { + if peerKind != "group" && !policyChecked && !c.IsAllowed(senderID) { return } diff --git a/internal/channels/policy_test.go b/internal/channels/policy_test.go index 398163c0..676884dc 100644 --- a/internal/channels/policy_test.go +++ b/internal/channels/policy_test.go @@ -6,13 +6,14 @@ import ( "testing" "time" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/store" ) // mockPairingStore is a test implementation of store.PairingStore. type mockPairingStore struct { pairedDevices map[string]map[string]bool // senderID -> channel -> paired - failIsPaired bool // force IsPaired to return error + failIsPaired bool // force IsPaired to return error } func newMockPairingStore() *mockPairingStore { @@ -142,12 +143,12 @@ func TestCheckDMPolicy_PolicyAllowlist(t *testing.T) { // TestCheckDMPolicy_PolicyPairing checks pairing status. func TestCheckDMPolicy_PolicyPairing(t *testing.T) { tests := []struct { - name string - senderID string - allowList []string - paired bool - failPairingCheck bool - wantResult PolicyResult + name string + senderID string + allowList []string + paired bool + failPairingCheck bool + wantResult PolicyResult }{ { name: "Paired sender is allowed", @@ -219,6 +220,72 @@ func TestCheckDMPolicy_DefaultToPairing(t *testing.T) { } } +func TestHandleAuthorizedMessage_PairedDirectMessageBypassesAllowlistSafetyNet(t *testing.T) { + msgBus := bus.New() + bc := NewBaseChannel(TypeZaloPersonal, msgBus, []string{"195835936795841454"}) + bc.SetName("zalo-cppai-pm") + bc.SetAgentID("cppai-pm") + + ps := newMockPairingStore() + ps.setPaired("648444320145379814", "zalo-cppai-pm") + bc.SetPairingService(ps) + + if got := bc.CheckDMPolicy(context.Background(), "648444320145379814", "pairing"); got != PolicyAllow { + t.Fatalf("CheckDMPolicy(pairing) = %v; want PolicyAllow", got) + } + + bc.HandleAuthorizedMessage("648444320145379814", "648444320145379814", "xin chao", nil, nil, "direct") + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected paired direct message to publish inbound") + } + if msg.Channel != "zalo-cppai-pm" { + t.Fatalf("Channel = %q; want zalo-cppai-pm", msg.Channel) + } + if msg.SenderID != "648444320145379814" { + t.Fatalf("SenderID = %q; want 648444320145379814", msg.SenderID) + } + if msg.PeerKind != "direct" { + t.Fatalf("PeerKind = %q; want direct", msg.PeerKind) + } +} + +func TestHandleMessage_PairedDirectMessageWithoutPolicyCheckStillDrops(t *testing.T) { + msgBus := bus.New() + bc := NewBaseChannel(TypeZaloPersonal, msgBus, []string{"195835936795841454"}) + bc.SetName("zalo-cppai-pm") + + ps := newMockPairingStore() + ps.setPaired("648444320145379814", "zalo-cppai-pm") + bc.SetPairingService(ps) + + bc.HandleMessage("648444320145379814", "648444320145379814", "xin chao", nil, nil, "direct") + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if msg, ok := msgBus.ConsumeInbound(ctx); ok { + t.Fatalf("expected direct message without explicit policy check to drop, got %+v", msg) + } +} + +func TestHandleMessageMedia_UnpairedDirectMessageOutsideAllowlistStillDrops(t *testing.T) { + msgBus := bus.New() + bc := NewBaseChannel(TypeZaloPersonal, msgBus, []string{"195835936795841454"}) + bc.SetName("zalo-cppai-pm") + bc.SetPairingService(newMockPairingStore()) + + bc.HandleMessage("648444320145379814", "648444320145379814", "xin chao", nil, nil, "direct") + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if msg, ok := msgBus.ConsumeInbound(ctx); ok { + t.Fatalf("expected unpaired direct message outside allowlist to drop, got %+v", msg) + } +} + // TestCheckGroupPolicy_PolicyDisabled rejects group messages. func TestCheckGroupPolicy_PolicyDisabled(t *testing.T) { bc := NewBaseChannel("test", nil, nil) diff --git a/internal/channels/slack/handlers.go b/internal/channels/slack/handlers.go index fa9e2999..8a8ca26b 100644 --- a/internal/channels/slack/handlers.go +++ b/internal/channels/slack/handlers.go @@ -99,14 +99,6 @@ func (c *Channel) handleMessage(ev *slackevents.MessageEvent) { } } - // For DMs, apply global allowlist filter (allow_from contains user IDs). - // For groups, skip — group policy already handles channel/user filtering. - if isDM && !c.IsAllowed(senderID) { - slog.Debug("slack message rejected by allowlist", - "user_id", senderID, "display_name", displayName) - return - } - // Process file attachments from Slack message var mediaPaths []string var allItems []mediaItem @@ -263,7 +255,7 @@ func (c *Channel) handleMessage(ev *slackevents.MessageEvent) { // Message debounce: batch rapid messages per-thread if c.debounceDelay > 0 { - if c.debounceMessage(localKey, senderID, channelID, finalContent, mediaPaths, metadata, peerKind) { + if c.debounceMessage(localKey, senderID, channelID, finalContent, mediaPaths, metadata, peerKind, true) { // Record thread participation even when debounced if peerKind == "group" && replyThreadTS != "" { participKey := channelID + ":particip:" + replyThreadTS @@ -273,7 +265,7 @@ func (c *Channel) handleMessage(ev *slackevents.MessageEvent) { } } - c.HandleMessage(senderID, channelID, finalContent, mediaPaths, metadata, peerKind) + c.HandleAuthorizedMessage(senderID, channelID, finalContent, mediaPaths, metadata, peerKind) // Record thread participation for auto-reply cache if peerKind == "group" { @@ -292,7 +284,7 @@ func (c *Channel) fetchThreadParentContext(ctx context.Context, channelID, threa ChannelID: channelID, Latest: threadTS, Limit: 1, - Inclusive: true, + Inclusive: true, } history, err := c.api.GetConversationHistoryContext(ctx, params) if err != nil || len(history.Messages) == 0 { diff --git a/internal/channels/slack/handlers_files.go b/internal/channels/slack/handlers_files.go index 96259adc..128e1ec2 100644 --- a/internal/channels/slack/handlers_files.go +++ b/internal/channels/slack/handlers_files.go @@ -20,27 +20,29 @@ import ( // --- Message debounce/batching --- type debounceEntry struct { - timer *time.Timer - messages []string - mu sync.Mutex - senderID string - channelID string - media []string - metadata map[string]string - peerKind string + timer *time.Timer + messages []string + mu sync.Mutex + senderID string + channelID string + media []string + metadata map[string]string + peerKind string + authorized bool } // debounceMessage batches rapid messages. Returns true if message was debounced. -func (c *Channel) debounceMessage(localKey, senderID, channelID, content string, media []string, metadata map[string]string, peerKind string) bool { +func (c *Channel) debounceMessage(localKey, senderID, channelID, content string, media []string, metadata map[string]string, peerKind string, authorized bool) bool { c.debounceMu.Lock() entry, loaded := c.debounceTimers[localKey] if !loaded { entry = &debounceEntry{ - senderID: senderID, - channelID: channelID, - media: media, - metadata: metadata, - peerKind: peerKind, + senderID: senderID, + channelID: channelID, + media: media, + metadata: metadata, + peerKind: peerKind, + authorized: authorized, } c.debounceTimers[localKey] = entry } @@ -85,7 +87,7 @@ func (c *Channel) flushDebounce(localKey string) { combined := strings.Join(entry.messages, "\n") entry.mu.Unlock() - c.HandleMessage(entry.senderID, entry.channelID, combined, entry.media, entry.metadata, entry.peerKind) + c.publishMessage(entry.senderID, entry.channelID, combined, entry.media, entry.metadata, entry.peerKind, entry.authorized) if entry.peerKind == "group" { c.GroupHistory().Clear(localKey) diff --git a/internal/channels/slack/handlers_mention.go b/internal/channels/slack/handlers_mention.go index 1db6fcd6..57e5a0a9 100644 --- a/internal/channels/slack/handlers_mention.go +++ b/internal/channels/slack/handlers_mention.go @@ -98,7 +98,7 @@ func (c *Channel) handleAppMention(ev *slackevents.AppMentionEvent) { metadata["message_thread_id"] = replyThreadTS } - c.HandleMessage(senderID, channelID, finalContent, nil, metadata, "group") + c.HandleAuthorizedMessage(senderID, channelID, finalContent, nil, metadata, "group") // Record thread participation if replyThreadTS != "" { diff --git a/internal/channels/slack/policy_test.go b/internal/channels/slack/policy_test.go new file mode 100644 index 00000000..1c4ecbb2 --- /dev/null +++ b/internal/channels/slack/policy_test.go @@ -0,0 +1,110 @@ +package slack + +import ( + "context" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type slackPolicyPairingStore struct { + paired map[string]map[string]bool +} + +func newSlackPolicyPairingStore() *slackPolicyPairingStore { + return &slackPolicyPairingStore{paired: make(map[string]map[string]bool)} +} + +func (s *slackPolicyPairingStore) RequestPairing(context.Context, string, string, string, string, map[string]string) (string, error) { + return "code123", nil +} + +func (s *slackPolicyPairingStore) ApprovePairing(context.Context, string, string) (*store.PairedDeviceData, error) { + return nil, nil +} + +func (s *slackPolicyPairingStore) DenyPairing(context.Context, string) error { return nil } + +func (s *slackPolicyPairingStore) RevokePairing(context.Context, string, string) error { return nil } + +func (s *slackPolicyPairingStore) IsPaired(_ context.Context, senderID, channel string) (bool, error) { + if s.paired[senderID] == nil { + return false, nil + } + return s.paired[senderID][channel], nil +} + +func (s *slackPolicyPairingStore) ListPending(context.Context) []store.PairingRequestData { + return nil +} + +func (s *slackPolicyPairingStore) ListPaired(context.Context) []store.PairedDeviceData { + return nil +} + +func (s *slackPolicyPairingStore) MigrateGroupChatID(context.Context, string, string, string) error { + return nil +} + +func (s *slackPolicyPairingStore) setPaired(senderID, channel string) { + if s.paired[senderID] == nil { + s.paired[senderID] = make(map[string]bool) + } + s.paired[senderID][channel] = true +} + +func TestSlackPairedDirectMessageBypassesAllowlistAfterPolicyCheck(t *testing.T) { + msgBus := bus.New() + pairingStore := newSlackPolicyPairingStore() + pairingStore.setPaired("U_PAIRED", "slack-test") + + ch, err := New(config.SlackConfig{ + BotToken: "xoxb-test", + AppToken: "xapp-test", + AllowFrom: []string{"U_ALLOWED"}, + DMPolicy: "pairing", + }, msgBus, pairingStore, nil) + if err != nil { + t.Fatalf("New() error = %v", err) + } + ch.SetName("slack-test") + + if !ch.checkDMPolicy(context.Background(), "U_PAIRED", "D123") { + t.Fatal("expected paired Slack sender to pass DM policy") + } + ch.HandleAuthorizedMessage("U_PAIRED", "D123", "hello", nil, map[string]string{"username": "paired"}, "direct") + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected paired Slack DM to publish inbound") + } + if msg.SenderID != "U_PAIRED" || msg.ChatID != "D123" || msg.PeerKind != "direct" { + t.Fatalf("unexpected inbound message: %+v", msg) + } +} + +func TestSlackHandleMessageWithoutPolicyCheckStillUsesAllowlistSafetyNet(t *testing.T) { + msgBus := bus.New() + ch, err := New(config.SlackConfig{ + BotToken: "xoxb-test", + AppToken: "xapp-test", + AllowFrom: []string{"U_ALLOWED"}, + DMPolicy: "pairing", + }, msgBus, newSlackPolicyPairingStore(), nil) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + ch.HandleMessage("U_PAIRED", "D123", "hello", nil, map[string]string{"username": "paired"}, "direct") + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if msg, ok := msgBus.ConsumeInbound(ctx); ok { + t.Fatalf("expected unverified Slack DM to drop, got %+v", msg) + } +} diff --git a/internal/channels/slack/utils.go b/internal/channels/slack/utils.go index f61320b0..a7d7d26a 100644 --- a/internal/channels/slack/utils.go +++ b/internal/channels/slack/utils.go @@ -17,8 +17,16 @@ import ( // is in the allowlist, enabling group-level allowlisting without requiring individual user IDs. // This is Slack-specific: other channels only check senderID in BaseChannel.HandleMessage. func (c *Channel) HandleMessage(senderID, chatID, content string, mediaPaths []string, metadata map[string]string, peerKind string) { + c.publishMessage(senderID, chatID, content, mediaPaths, metadata, peerKind, false) +} + +func (c *Channel) HandleAuthorizedMessage(senderID, chatID, content string, mediaPaths []string, metadata map[string]string, peerKind string) { + c.publishMessage(senderID, chatID, content, mediaPaths, metadata, peerKind, true) +} + +func (c *Channel) publishMessage(senderID, chatID, content string, mediaPaths []string, metadata map[string]string, peerKind string, policyChecked bool) { // Allow if either the sender or the Slack channel ID is in the allowlist. - if !c.IsAllowed(senderID) && !c.IsAllowed(chatID) { + if !policyChecked && !c.IsAllowed(senderID) && !c.IsAllowed(chatID) { return } diff --git a/internal/channels/zalo/personal/handlers.go b/internal/channels/zalo/personal/handlers.go index 0546b7b1..cff69d8a 100644 --- a/internal/channels/zalo/personal/handlers.go +++ b/internal/channels/zalo/personal/handlers.go @@ -72,7 +72,7 @@ func (c *Channel) handleDM(msg protocol.UserMessage) { "platform": channels.TypeZaloPersonal, "display_name": channels.SanitizeDisplayName(senderName), } - c.HandleMessage(senderID, threadID, content, media, metadata, "direct") + c.HandleAuthorizedMessage(senderID, threadID, content, media, metadata, "direct") } func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { diff --git a/internal/channels/zalo/zalo.go b/internal/channels/zalo/zalo.go index 3368e394..b50460de 100644 --- a/internal/channels/zalo/zalo.go +++ b/internal/channels/zalo/zalo.go @@ -229,7 +229,7 @@ func (c *Channel) handleTextMessage(msg *zaloMessage) { "platform": "zalo", } - c.HandleMessage(senderID, chatID, content, nil, metadata, "direct") + c.HandleAuthorizedMessage(senderID, chatID, content, nil, metadata, "direct") } func (c *Channel) handleImageMessage(msg *zaloMessage) { @@ -287,7 +287,7 @@ func (c *Channel) handleImageMessage(msg *zaloMessage) { "platform": "zalo", } - c.HandleMessage(senderID, chatID, content, media, metadata, "direct") + c.HandleAuthorizedMessage(senderID, chatID, content, media, metadata, "direct") } // --- DM Policy ---