fix(channels): honor paired DM policy before publish (#1313)

Co-authored-by: ntduc <ntduc@cpp.ai.vn>

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

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)

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

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
 	}
 

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)

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 {

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)

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 != "" {

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)
+	}
+}

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
 	}
 

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) {

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 ---