diff --git a/server/internal/integrations/lark/channel_store.go b/server/internal/integrations/lark/channel_store.go index 3e8c333cec..2a8495e738 100644 --- a/server/internal/integrations/lark/channel_store.go +++ b/server/internal/integrations/lark/channel_store.go @@ -83,7 +83,10 @@ func (s *ChannelStore) GetLarkInstallationByAppID(ctx context.Context, appID str } func (s *ChannelStore) GetLarkInstallation(ctx context.Context, id pgtype.UUID) (db.LarkInstallation, error) { - row, err := s.Queries.GetChannelInstallation(ctx, id) + row, err := s.Queries.GetChannelInstallation(ctx, db.GetChannelInstallationParams{ + ID: id, + ChannelType: channelTypeFeishu, + }) if err != nil { return db.LarkInstallation{}, err } @@ -94,6 +97,7 @@ func (s *ChannelStore) GetLarkInstallationInWorkspace(ctx context.Context, arg d row, err := s.Queries.GetChannelInstallationInWorkspace(ctx, db.GetChannelInstallationInWorkspaceParams{ ID: arg.ID, WorkspaceID: arg.WorkspaceID, + ChannelType: channelTypeFeishu, }) if err != nil { return db.LarkInstallation{}, err @@ -102,7 +106,10 @@ func (s *ChannelStore) GetLarkInstallationInWorkspace(ctx context.Context, arg d } func (s *ChannelStore) ListLarkInstallationsByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]db.LarkInstallation, error) { - rows, err := s.Queries.ListChannelInstallationsByWorkspace(ctx, workspaceID) + rows, err := s.Queries.ListChannelInstallationsByWorkspace(ctx, db.ListChannelInstallationsByWorkspaceParams{ + WorkspaceID: workspaceID, + ChannelType: channelTypeFeishu, + }) if err != nil { return nil, err } @@ -110,7 +117,7 @@ func (s *ChannelStore) ListLarkInstallationsByWorkspace(ctx context.Context, wor } func (s *ChannelStore) ListActiveLarkInstallations(ctx context.Context) ([]db.LarkInstallation, error) { - rows, err := s.Queries.ListActiveChannelInstallations(ctx) + rows, err := s.Queries.ListActiveChannelInstallations(ctx, channelTypeFeishu) if err != nil { return nil, err } @@ -155,7 +162,10 @@ func (s *ChannelStore) SetLarkInstallationStatus(ctx context.Context, arg db.Set // keyed by id and effectively single-writer, so the non-atomic RMW is safe — // the same shape the channel.sql comment documents for this query. func (s *ChannelStore) SetLarkInstallationBotUnionID(ctx context.Context, arg db.SetLarkInstallationBotUnionIDParams) error { - row, err := s.Queries.GetChannelInstallation(ctx, arg.ID) + row, err := s.Queries.GetChannelInstallation(ctx, db.GetChannelInstallationParams{ + ID: arg.ID, + ChannelType: channelTypeFeishu, + }) if err != nil { return err } @@ -245,7 +255,10 @@ func (s *ChannelStore) GetLarkChatSessionBinding(ctx context.Context, arg db.Get } func (s *ChannelStore) GetLarkChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) (db.LarkChatSessionBinding, error) { - row, err := s.Queries.GetChannelChatSessionBindingBySession(ctx, chatSessionID) + row, err := s.Queries.GetChannelChatSessionBindingBySession(ctx, db.GetChannelChatSessionBindingBySessionParams{ + ChatSessionID: chatSessionID, + ChannelType: channelTypeFeishu, + }) if err != nil { return db.LarkChatSessionBinding{}, err } @@ -345,7 +358,10 @@ func (s *ChannelStore) ConsumeLarkBindingToken(ctx context.Context, tokenHash st // ---- outbound card ---- func (s *ChannelStore) GetLarkOutboundCardByTask(ctx context.Context, taskID pgtype.UUID) (db.LarkOutboundCardMessage, error) { - row, err := s.Queries.GetChannelOutboundCardByTask(ctx, taskID) + row, err := s.Queries.GetChannelOutboundCardByTask(ctx, db.GetChannelOutboundCardByTaskParams{ + TaskID: taskID, + ChannelType: channelTypeFeishu, + }) if err != nil { return db.LarkOutboundCardMessage{}, err } diff --git a/server/internal/integrations/lark/channel_store_scope_test.go b/server/internal/integrations/lark/channel_store_scope_test.go new file mode 100644 index 0000000000..7d034aeb6c --- /dev/null +++ b/server/internal/integrations/lark/channel_store_scope_test.go @@ -0,0 +1,171 @@ +package lark + +import ( + "context" + "errors" + "os" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" +) + +// channelScopeTestDB connects to the test Postgres (DATABASE_URL or the default +// local DSN, same as the handler suite) and returns a pool, or skips when no +// migrated database is reachable. Kept local to this file so the rest of the +// lark package stays DB-free. +func channelScopeTestDB(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("DATABASE_URL") + if dsn == "" { + dsn = "postgres://multica:multica@localhost:5432/multica?sslmode=disable" + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Skipf("no database: %v", err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Skipf("database not reachable: %v", err) + } + var present bool + if err := pool.QueryRow(ctx, "SELECT to_regclass('public.channel_installation') IS NOT NULL").Scan(&present); err != nil || !present { + pool.Close() + t.Skip("channel_installation not present (database not migrated)") + } + t.Cleanup(pool.Close) + return pool +} + +// TestChannelStore_ScopesToFeishu is the MUL-3515 regression guard: the +// Lark/Feishu wrappers on ChannelStore must never read another channel_type's +// rows, even when a non-Feishu installation / chat-session binding / outbound +// card shares the same workspace, chat_session, or task. (Member-removal and +// chat-session cleanup deliberately stay all-channel; that is covered by the +// handler tests.) +func TestChannelStore_ScopesToFeishu(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + // Synthetic, distinctive identifiers. channel_* has no foreign keys, so + // these rows need no parent records and nothing cascades — which is also + // why the test cleans up explicitly, by deterministic key, before and + // after (a killed prior run must not leave colliding rows behind). + const ( + feishuApp = "cli_scope_feishu" + slackApp = "cli_scope_slack" + wsID = "5c09e000-0000-4000-8000-000000000001" + agentID = "5c09e000-0000-4000-8000-000000000002" + chatSessionID = "5c09e000-0000-4000-8000-000000000003" + taskID = "5c09e000-0000-4000-8000-000000000004" + installerID = "5c09e000-0000-4000-8000-000000000005" + ) + clean := func() { + _, _ = pool.Exec(context.Background(), + `DELETE FROM channel_installation WHERE config->>'app_id' = ANY($1)`, + []string{feishuApp, slackApp}) + _, _ = pool.Exec(context.Background(), + `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, chatSessionID) + _, _ = pool.Exec(context.Background(), + `DELETE FROM channel_outbound_card_message WHERE task_id = $1`, taskID) + } + clean() + t.Cleanup(clean) + + insertInstallation := func(channelType, app string) pgtype.UUID { + var id string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id) +VALUES ($1, $2, $3, jsonb_build_object('app_id', $4::text), $5) +RETURNING id +`, wsID, agentID, channelType, app, installerID).Scan(&id); err != nil { + t.Fatalf("insert %s installation: %v", channelType, err) + } + return util.MustParseUUID(id) + } + feishuID := insertInstallation("feishu", feishuApp) + slackID := insertInstallation("slack", slackApp) + + // A non-Feishu binding/card sharing this test's chat_session and task. + if _, err := pool.Exec(ctx, ` +INSERT INTO channel_chat_session_binding (chat_session_id, installation_id, channel_type, channel_chat_id, chat_type) +VALUES ($1, $2, 'slack', 'oc_scope_slack', 'p2p') +`, chatSessionID, slackID); err != nil { + t.Fatalf("insert slack chat binding: %v", err) + } + if _, err := pool.Exec(ctx, ` +INSERT INTO channel_outbound_card_message (chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status) +VALUES ($1, $2, 'slack', 'oc_scope_slack', 'om_scope_slack', 'pending') +`, chatSessionID, taskID); err != nil { + t.Fatalf("insert slack outbound card: %v", err) + } + + wsUUID := util.MustParseUUID(wsID) + sessionUUID := util.MustParseUUID(chatSessionID) + taskUUID := util.MustParseUUID(taskID) + + // --- installation reads: Feishu visible, Slack invisible --- + + if got, err := store.GetLarkInstallation(ctx, feishuID); err != nil || got.AppID != feishuApp { + t.Fatalf("GetLarkInstallation(feishu): got app=%q err=%v, want app=%q nil", got.AppID, err, feishuApp) + } + if _, err := store.GetLarkInstallation(ctx, slackID); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("GetLarkInstallation(slack): err=%v, want pgx.ErrNoRows (scoped out)", err) + } + + if got, err := store.GetLarkInstallationInWorkspace(ctx, db.GetLarkInstallationInWorkspaceParams{ID: feishuID, WorkspaceID: wsUUID}); err != nil || got.AppID != feishuApp { + t.Fatalf("GetLarkInstallationInWorkspace(feishu): got app=%q err=%v, want app=%q nil", got.AppID, err, feishuApp) + } + if _, err := store.GetLarkInstallationInWorkspace(ctx, db.GetLarkInstallationInWorkspaceParams{ID: slackID, WorkspaceID: wsUUID}); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("GetLarkInstallationInWorkspace(slack): err=%v, want pgx.ErrNoRows (scoped out)", err) + } + + // list-by-workspace: only the Feishu installation in this workspace + byWs, err := store.ListLarkInstallationsByWorkspace(ctx, wsUUID) + if err != nil { + t.Fatalf("ListLarkInstallationsByWorkspace: %v", err) + } + if len(byWs) != 1 || byWs[0].AppID != feishuApp { + apps := make([]string, len(byWs)) + for i, r := range byWs { + apps[i] = r.AppID + } + t.Fatalf("ListLarkInstallationsByWorkspace: got apps=%v, want exactly [%s]", apps, feishuApp) + } + + // list-active (hub boot): includes the Feishu install, never the Slack one + active, err := store.ListActiveLarkInstallations(ctx) + if err != nil { + t.Fatalf("ListActiveLarkInstallations: %v", err) + } + var sawFeishu, sawSlack bool + for _, r := range active { + switch r.AppID { + case feishuApp: + sawFeishu = true + case slackApp: + sawSlack = true + } + } + if !sawFeishu { + t.Fatal("ListActiveLarkInstallations: missing the active Feishu installation") + } + if sawSlack { + t.Fatal("ListActiveLarkInstallations: returned a Slack installation (hub would supervise another channel)") + } + + // --- outbound reads: a Slack binding/card must not be seen as Feishu --- + + if _, err := store.GetLarkChatSessionBindingBySession(ctx, sessionUUID); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("GetLarkChatSessionBindingBySession(slack-bound session): err=%v, want pgx.ErrNoRows (scoped out)", err) + } + if _, err := store.GetLarkOutboundCardByTask(ctx, taskUUID); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("GetLarkOutboundCardByTask(slack card): err=%v, want pgx.ErrNoRows (scoped out)", err) + } +} diff --git a/server/pkg/db/generated/channel.sql.go b/server/pkg/db/generated/channel.sql.go index c3f7937633..fac27bee4d 100644 --- a/server/pkg/db/generated/channel.sql.go +++ b/server/pkg/db/generated/channel.sql.go @@ -405,12 +405,20 @@ func (q *Queries) GetChannelChatSessionBinding(ctx context.Context, arg GetChann const getChannelChatSessionBindingBySession = `-- name: GetChannelChatSessionBindingBySession :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at FROM channel_chat_session_binding WHERE chat_session_id = $1 + AND channel_type = $2 ` +type GetChannelChatSessionBindingBySessionParams struct { + ChatSessionID pgtype.UUID `json:"chat_session_id"` + ChannelType string `json:"channel_type"` +} + // Reverse lookup for the outbound patcher: given a chat_session_id, find // its channel binding to know which (installation, chat_id) to send to. -func (q *Queries) GetChannelChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) (ChannelChatSessionBinding, error) { - row := q.db.QueryRow(ctx, getChannelChatSessionBindingBySession, chatSessionID) +// Scoped by channel_type so a future non-Feishu binding on the same +// chat_session is never treated as a Feishu reply target. +func (q *Queries) GetChannelChatSessionBindingBySession(ctx context.Context, arg GetChannelChatSessionBindingBySessionParams) (ChannelChatSessionBinding, error) { + row := q.db.QueryRow(ctx, getChannelChatSessionBindingBySession, arg.ChatSessionID, arg.ChannelType) var i ChannelChatSessionBinding err := row.Scan( &i.ID, @@ -428,11 +436,19 @@ func (q *Queries) GetChannelChatSessionBindingBySession(ctx context.Context, cha } const getChannelInstallation = `-- name: GetChannelInstallation :one -SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE id = $1 +SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation +WHERE id = $1 AND channel_type = $2 ` -func (q *Queries) GetChannelInstallation(ctx context.Context, id pgtype.UUID) (ChannelInstallation, error) { - row := q.db.QueryRow(ctx, getChannelInstallation, id) +type GetChannelInstallationParams struct { + ID pgtype.UUID `json:"id"` + ChannelType string `json:"channel_type"` +} + +// Scoped by channel_type: a per-channel caller (e.g. the Feishu store) +// must never resolve another channel's installation by guessing its UUID. +func (q *Queries) GetChannelInstallation(ctx context.Context, arg GetChannelInstallationParams) (ChannelInstallation, error) { + row := q.db.QueryRow(ctx, getChannelInstallation, arg.ID, arg.ChannelType) var i ChannelInstallation err := row.Scan( &i.ID, @@ -492,16 +508,19 @@ func (q *Queries) GetChannelInstallationByAppID(ctx context.Context, arg GetChan const getChannelInstallationInWorkspace = `-- name: GetChannelInstallationInWorkspace :one SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation -WHERE id = $1 AND workspace_id = $2 +WHERE id = $1 + AND workspace_id = $2 + AND channel_type = $3 ` type GetChannelInstallationInWorkspaceParams struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` + ChannelType string `json:"channel_type"` } func (q *Queries) GetChannelInstallationInWorkspace(ctx context.Context, arg GetChannelInstallationInWorkspaceParams) (ChannelInstallation, error) { - row := q.db.QueryRow(ctx, getChannelInstallationInWorkspace, arg.ID, arg.WorkspaceID) + row := q.db.QueryRow(ctx, getChannelInstallationInWorkspace, arg.ID, arg.WorkspaceID, arg.ChannelType) var i ChannelInstallation err := row.Scan( &i.ID, @@ -523,12 +542,19 @@ func (q *Queries) GetChannelInstallationInWorkspace(ctx context.Context, arg Get const getChannelOutboundCardByTask = `-- name: GetChannelOutboundCardByTask :one SELECT id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at FROM channel_outbound_card_message WHERE task_id = $1 + AND channel_type = $2 ` +type GetChannelOutboundCardByTaskParams struct { + TaskID pgtype.UUID `json:"task_id"` + ChannelType string `json:"channel_type"` +} + // The partial unique index on (task_id) WHERE task_id IS NOT NULL -// guarantees at most one row. -func (q *Queries) GetChannelOutboundCardByTask(ctx context.Context, taskID pgtype.UUID) (ChannelOutboundCardMessage, error) { - row := q.db.QueryRow(ctx, getChannelOutboundCardByTask, taskID) +// guarantees at most one row. Scoped by channel_type so a future non-Feishu +// card for the same task is not patched as a Feishu card. +func (q *Queries) GetChannelOutboundCardByTask(ctx context.Context, arg GetChannelOutboundCardByTaskParams) (ChannelOutboundCardMessage, error) { + row := q.db.QueryRow(ctx, getChannelOutboundCardByTask, arg.TaskID, arg.ChannelType) var i ChannelOutboundCardMessage err := row.Scan( &i.ID, @@ -577,13 +603,15 @@ func (q *Queries) GetChannelUserBindingByUserID(ctx context.Context, arg GetChan const listActiveChannelInstallations = `-- name: ListActiveChannelInstallations :many SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE status = 'active' + AND channel_type = $1 ORDER BY created_at ASC ` -// Boot path for the inbound hub: every active installation, any channel -// type, so the hub can claim leases and open connections. -func (q *Queries) ListActiveChannelInstallations(ctx context.Context) ([]ChannelInstallation, error) { - rows, err := q.db.Query(ctx, listActiveChannelInstallations) +// Boot path for a per-channel-type inbound hub: every active installation of +// the given channel_type, so a hub claims leases and opens connections only +// for its own platform and never supervises another channel's installation. +func (q *Queries) ListActiveChannelInstallations(ctx context.Context, channelType string) ([]ChannelInstallation, error) { + rows, err := q.db.Query(ctx, listActiveChannelInstallations, channelType) if err != nil { return nil, err } @@ -661,11 +689,19 @@ func (q *Queries) ListChannelInboundAuditByInstallation(ctx context.Context, arg const listChannelInstallationsByWorkspace = `-- name: ListChannelInstallationsByWorkspace :many SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE workspace_id = $1 + AND channel_type = $2 ORDER BY created_at ASC ` -func (q *Queries) ListChannelInstallationsByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]ChannelInstallation, error) { - rows, err := q.db.Query(ctx, listChannelInstallationsByWorkspace, workspaceID) +type ListChannelInstallationsByWorkspaceParams struct { + WorkspaceID pgtype.UUID `json:"workspace_id"` + ChannelType string `json:"channel_type"` +} + +// Scoped by channel_type so a per-channel management surface (e.g. the Lark +// installation list) only ever sees its own platform's installations. +func (q *Queries) ListChannelInstallationsByWorkspace(ctx context.Context, arg ListChannelInstallationsByWorkspaceParams) ([]ChannelInstallation, error) { + rows, err := q.db.Query(ctx, listChannelInstallationsByWorkspace, arg.WorkspaceID, arg.ChannelType) if err != nil { return nil, err } diff --git a/server/pkg/db/queries/channel.sql b/server/pkg/db/queries/channel.sql index 3b34f58ae2..4510a5f737 100644 --- a/server/pkg/db/queries/channel.sql +++ b/server/pkg/db/queries/channel.sql @@ -39,11 +39,16 @@ ON CONFLICT (workspace_id, agent_id, channel_type) DO UPDATE SET RETURNING *; -- name: GetChannelInstallation :one -SELECT * FROM channel_installation WHERE id = $1; +-- Scoped by channel_type: a per-channel caller (e.g. the Feishu store) +-- must never resolve another channel's installation by guessing its UUID. +SELECT * FROM channel_installation +WHERE id = sqlc.arg('id') AND channel_type = sqlc.arg('channel_type'); -- name: GetChannelInstallationInWorkspace :one SELECT * FROM channel_installation -WHERE id = $1 AND workspace_id = $2; +WHERE id = sqlc.arg('id') + AND workspace_id = sqlc.arg('workspace_id') + AND channel_type = sqlc.arg('channel_type'); -- name: GetChannelInstallationByAppID :one -- Inbound routing. The platform event carries only the channel's app @@ -59,15 +64,20 @@ WHERE channel_type = sqlc.arg('channel_type') AND config ->> 'app_id' = sqlc.arg('app_id')::text; -- name: ListChannelInstallationsByWorkspace :many +-- Scoped by channel_type so a per-channel management surface (e.g. the Lark +-- installation list) only ever sees its own platform's installations. SELECT * FROM channel_installation -WHERE workspace_id = $1 +WHERE workspace_id = sqlc.arg('workspace_id') + AND channel_type = sqlc.arg('channel_type') ORDER BY created_at ASC; -- name: ListActiveChannelInstallations :many --- Boot path for the inbound hub: every active installation, any channel --- type, so the hub can claim leases and open connections. +-- Boot path for a per-channel-type inbound hub: every active installation of +-- the given channel_type, so a hub claims leases and opens connections only +-- for its own platform and never supervises another channel's installation. SELECT * FROM channel_installation WHERE status = 'active' + AND channel_type = sqlc.arg('channel_type') ORDER BY created_at ASC; -- name: SetChannelInstallationStatus :exec @@ -183,8 +193,11 @@ WHERE installation_id = $1 AND channel_chat_id = $2; -- name: GetChannelChatSessionBindingBySession :one -- Reverse lookup for the outbound patcher: given a chat_session_id, find -- its channel binding to know which (installation, chat_id) to send to. +-- Scoped by channel_type so a future non-Feishu binding on the same +-- chat_session is never treated as a Feishu reply target. SELECT * FROM channel_chat_session_binding -WHERE chat_session_id = $1; +WHERE chat_session_id = sqlc.arg('chat_session_id') + AND channel_type = sqlc.arg('channel_type'); -- name: UpdateChannelChatSessionBindingReplyTarget :exec -- Records the most recent inbound trigger message + thread so the decoupled @@ -288,9 +301,11 @@ RETURNING *; -- name: GetChannelOutboundCardByTask :one -- The partial unique index on (task_id) WHERE task_id IS NOT NULL --- guarantees at most one row. +-- guarantees at most one row. Scoped by channel_type so a future non-Feishu +-- card for the same task is not patched as a Feishu card. SELECT * FROM channel_outbound_card_message -WHERE task_id = $1; +WHERE task_id = sqlc.arg('task_id') + AND channel_type = sqlc.arg('channel_type'); -- name: UpdateChannelOutboundCardStatus :exec UPDATE channel_outbound_card_message