From 9063a6a25f98f3cb87b143079480951c3b1d6fa6 Mon Sep 17 00:00:00 2001 From: Sakurapainting Date: Fri, 27 Mar 2026 18:29:39 +0800 Subject: [PATCH] fix(qq): use openid routing and allowlist compatibility --- pkg/channels/qq/qq.go | 244 +++++++++++++++++++++++++++++++------ pkg/channels/qq/qq_test.go | 184 +++++++++++++++++++++++++++- 2 files changed, 390 insertions(+), 38 deletions(-) diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index f2b70aec9..905fcaf55 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -41,6 +41,7 @@ const ( typingResend = 8 * time.Second typingSeconds = 10 bytesPerMiB = 1024 * 1024 + qqStartupProbe = 15 * time.Second ) type qqAPI interface { @@ -82,6 +83,60 @@ type QQChannel struct { stopOnce sync.Once } +type qqRawAuthor struct { + UserOpenID string `json:"user_openid"` + MemberOpenID string `json:"member_openid"` +} + +type qqRawEnvelope struct { + D struct { + Author qqRawAuthor `json:"author"` + GroupOpenID string `json:"group_openid"` + } `json:"d"` +} + +func parseQQOpenIDs(raw []byte) (userOpenID, memberOpenID, groupOpenID string) { + if len(raw) == 0 { + return "", "", "" + } + + var env qqRawEnvelope + if err := json.Unmarshal(raw, &env); err != nil { + return "", "", "" + } + + return strings.TrimSpace(env.D.Author.UserOpenID), + strings.TrimSpace(env.D.Author.MemberOpenID), + strings.TrimSpace(env.D.GroupOpenID) +} + +// resolveAllowedSender validates sender against allow_from using openid first, +// then falls back to legacy author/member id for backward compatibility. +func (c *QQChannel) resolveAllowedSender(primaryOpenID, legacyID string) (bus.SenderInfo, bool, bool) { + primary := bus.SenderInfo{ + Platform: "qq", + PlatformID: primaryOpenID, + CanonicalID: identity.BuildCanonicalID("qq", primaryOpenID), + } + + if c.IsAllowedSender(primary) { + return primary, true, false + } + + legacyID = strings.TrimSpace(legacyID) + if legacyID != "" && legacyID != primaryOpenID { + legacy := bus.SenderInfo{ + Platform: "qq", + PlatformID: legacyID, + } + if c.IsAllowedSender(legacy) { + return legacy, true, true + } + } + + return primary, false, false +} + func NewQQChannel(cfg config.QQConfig, messageBus *bus.MessageBus) (*QQChannel, error) { base := channels.NewBaseChannel("qq", cfg, messageBus, cfg.AllowFrom, channels.WithMaxMessageLength(cfg.MaxMessageLength), @@ -127,8 +182,24 @@ func (c *QQChannel) Start(ctx context.Context) error { // initialize OpenAPI client c.api = botgo.NewOpenAPI(c.config.AppID, c.tokenSource).WithTimeout(5 * time.Second) + readyCh := make(chan struct{}, 1) + sessionErrCh := make(chan error, 1) + // register event handlers intent := event.RegisterHandlers( + event.ReadyHandler(func(_ *dto.WSPayload, _ *dto.WSReadyData) { + select { + case readyCh <- struct{}{}: + default: + } + }), + event.ErrorNotifyHandler(func(err error) { + fmt.Printf("QQ gateway error: %v\n", err) + select { + case sessionErrCh <- err: + default: + } + }), c.handleC2CMessage(), c.handleGroupATMessage(), ) @@ -145,17 +216,50 @@ func (c *QQChannel) Start(ctx context.Context) error { // create and save sessionManager c.sessionManager = botgo.NewSessionManager() + startupErr := make(chan error, 1) // start WebSocket connection in goroutine to avoid blocking go func() { - if err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent); err != nil { + err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent) + if err != nil { logger.ErrorCF("qq", "WebSocket session error", map[string]any{ "error": err.Error(), }) + fmt.Printf("QQ WebSocket session error: %v\n", err) c.SetRunning(false) } + select { + case startupErr <- err: + default: + } }() + select { + case <-readyCh: + fmt.Println("QQ WebSocket ready") + case err := <-sessionErrCh: + if c.cancel != nil { + c.cancel() + } + if err != nil { + return fmt.Errorf("QQ websocket failed before ready: %w", err) + } + return fmt.Errorf("QQ websocket failed before ready") + case err := <-startupErr: + if c.cancel != nil { + c.cancel() + } + if err != nil { + return fmt.Errorf("failed to establish QQ websocket session: %w", err) + } + return fmt.Errorf("QQ websocket session exited unexpectedly during startup") + case <-time.After(qqStartupProbe): + if c.cancel != nil { + c.cancel() + } + return fmt.Errorf("timeout waiting for QQ websocket READY event") + } + // start dedup janitor goroutine go c.dedupJanitor() @@ -597,29 +701,53 @@ func (c *QQChannel) maxBase64FileSizeBytes() int64 { // handleC2CMessage handles QQ private messages. func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler { return func(event *dto.WSPayload, data *dto.WSC2CMessageData) error { + if data == nil { + return nil + } + // deduplication check if c.isDuplicate(data.ID) { return nil } - // extract user info - var senderID string - if data.Author != nil && data.Author.ID != "" { - senderID = data.Author.ID - } else { + var raw []byte + if event != nil { + raw = event.RawMessage + } + userOpenID, _, _ := parseQQOpenIDs(raw) + legacyAuthorID := "" + if data.Author != nil { + legacyAuthorID = strings.TrimSpace(data.Author.ID) + } + + // QQ C2C endpoint requires user_openid; fallback to author.id for compatibility. + senderID := userOpenID + if senderID == "" { + senderID = legacyAuthorID + } + if senderID == "" { logger.WarnC("qq", "Received message with no sender ID") return nil } - sender := bus.SenderInfo{ - Platform: "qq", - PlatformID: data.Author.ID, - CanonicalID: identity.BuildCanonicalID("qq", data.Author.ID), - } - - if !c.IsAllowedSender(sender) { + sender, allowed, usedLegacyFallback := c.resolveAllowedSender(senderID, legacyAuthorID) + if !allowed { + logger.WarnCF("qq", "Dropped C2C message by allow_from", map[string]any{ + "sender_openid": senderID, + "legacy_id": legacyAuthorID, + "message_id": data.ID, + }) + fmt.Println("QQ inbound C2C dropped by allow_from") return nil } + if usedLegacyFallback { + logger.WarnCF("qq", "allow_from matched legacy QQ sender id", map[string]any{ + "sender_openid": senderID, + "legacy_id": legacyAuthorID, + "message_id": data.ID, + }) + fmt.Println("QQ inbound C2C allow_from matched legacy id") + } content := strings.TrimSpace(data.Content) mediaPaths, attachmentNotes := c.extractInboundAttachments(senderID, data.ID, data.Attachments) @@ -638,21 +766,25 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler { }) // Store chat routing context. - c.chatType.Store(senderID, "direct") - c.lastMsgID.Store(senderID, data.ID) + chatID := senderID + c.chatType.Store(chatID, "direct") + c.lastMsgID.Store(chatID, data.ID) // Reset msg_seq counter for new inbound message. - c.msgSeqCounters.Store(senderID, new(atomic.Uint64)) + c.msgSeqCounters.Store(chatID, new(atomic.Uint64)) metadata := map[string]string{ "account_id": senderID, } + if legacyAuthorID != "" && legacyAuthorID != senderID { + metadata["legacy_account_id"] = legacyAuthorID + } c.HandleMessage(c.ctx, - bus.Peer{Kind: "direct", ID: senderID}, + bus.Peer{Kind: "direct", ID: chatID}, data.ID, senderID, - senderID, + chatID, content, mediaPaths, metadata, @@ -666,32 +798,67 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler { // handleGroupATMessage handles QQ group @ messages. func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler { return func(event *dto.WSPayload, data *dto.WSGroupATMessageData) error { + if data == nil { + return nil + } + // deduplication check if c.isDuplicate(data.ID) { return nil } - // extract user info - var senderID string - if data.Author != nil && data.Author.ID != "" { - senderID = data.Author.ID - } else { + var raw []byte + if event != nil { + raw = event.RawMessage + } + _, memberOpenID, groupOpenID := parseQQOpenIDs(raw) + legacyMemberID := "" + if data.Author != nil { + legacyMemberID = strings.TrimSpace(data.Author.ID) + } + + // For QQ group callbacks, member_openid/group_openid are preferred identifiers. + senderID := memberOpenID + if senderID == "" { + senderID = legacyMemberID + } + if senderID == "" { logger.WarnC("qq", "Received group message with no sender ID") return nil } - sender := bus.SenderInfo{ - Platform: "qq", - PlatformID: data.Author.ID, - CanonicalID: identity.BuildCanonicalID("qq", data.Author.ID), + chatID := groupOpenID + if chatID == "" { + chatID = strings.TrimSpace(data.GroupID) } - - if !c.IsAllowedSender(sender) { + if chatID == "" { + logger.WarnC("qq", "Received group message with no group ID") return nil } + sender, allowed, usedLegacyFallback := c.resolveAllowedSender(senderID, legacyMemberID) + if !allowed { + logger.WarnCF("qq", "Dropped group message by allow_from", map[string]any{ + "sender_openid": senderID, + "legacy_id": legacyMemberID, + "group_id": chatID, + "message_id": data.ID, + }) + fmt.Println("QQ inbound group dropped by allow_from") + return nil + } + if usedLegacyFallback { + logger.WarnCF("qq", "allow_from matched legacy QQ member id", map[string]any{ + "sender_openid": senderID, + "legacy_id": legacyMemberID, + "group_id": chatID, + "message_id": data.ID, + }) + fmt.Println("QQ inbound group allow_from matched legacy id") + } + content := strings.TrimSpace(data.Content) - mediaPaths, attachmentNotes := c.extractInboundAttachments(data.GroupID, data.ID, data.Attachments) + mediaPaths, attachmentNotes := c.extractInboundAttachments(chatID, data.ID, data.Attachments) for _, note := range attachmentNotes { content = appendContent(content, note) } @@ -709,28 +876,31 @@ func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler { logger.InfoCF("qq", "Received group AT message", map[string]any{ "sender": senderID, - "group": data.GroupID, + "group": chatID, "length": len(content), "media_count": len(mediaPaths), }) // Store chat routing context using GroupID as chatID. - c.chatType.Store(data.GroupID, "group") - c.lastMsgID.Store(data.GroupID, data.ID) + c.chatType.Store(chatID, "group") + c.lastMsgID.Store(chatID, data.ID) // Reset msg_seq counter for new inbound message. - c.msgSeqCounters.Store(data.GroupID, new(atomic.Uint64)) + c.msgSeqCounters.Store(chatID, new(atomic.Uint64)) metadata := map[string]string{ "account_id": senderID, - "group_id": data.GroupID, + "group_id": chatID, + } + if legacyMemberID != "" && legacyMemberID != senderID { + metadata["legacy_account_id"] = legacyMemberID } c.HandleMessage(c.ctx, - bus.Peer{Kind: "group", ID: data.GroupID}, + bus.Peer{Kind: "group", ID: chatID}, data.ID, senderID, - data.GroupID, + chatID, content, mediaPaths, metadata, diff --git a/pkg/channels/qq/qq_test.go b/pkg/channels/qq/qq_test.go index 83a912cd7..5ef2b403b 100644 --- a/pkg/channels/qq/qq_test.go +++ b/pkg/channels/qq/qq_test.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "os" + "path/filepath" "strings" "sync/atomic" "testing" @@ -62,6 +63,187 @@ func TestHandleC2CMessage_IncludesAccountIDMetadata(t *testing.T) { } } +func TestHandleC2CMessage_PrefersUserOpenIDFromRawPayload(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &QQChannel{ + BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, nil), + dedup: make(map[string]time.Time), + done: make(chan struct{}), + ctx: context.Background(), + } + + raw := []byte(`{"d":{"author":{"id":"legacy-id","user_openid":"user-openid-123"}}}`) + err := ch.handleC2CMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSC2CMessageData{ + ID: "msg-openid", + Content: "hello", + Author: &dto.User{ + ID: "legacy-id", + }, + }) + if err != nil { + t.Fatalf("handleC2CMessage() error = %v", err) + } + + inbound := waitInboundMessage(t, messageBus) + if inbound.Metadata["account_id"] != "user-openid-123" { + t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "user-openid-123") + } + if inbound.ChatID != "user-openid-123" { + t.Fatalf("inbound.ChatID = %q, want %q", inbound.ChatID, "user-openid-123") + } +} + +func TestHandleC2CMessage_AllowListFallsBackToLegacyID(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &QQChannel{ + BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, []string{"legacy-id"}), + dedup: make(map[string]time.Time), + done: make(chan struct{}), + ctx: context.Background(), + } + + raw := []byte(`{"d":{"author":{"id":"legacy-id","user_openid":"user-openid-123"}}}`) + err := ch.handleC2CMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSC2CMessageData{ + ID: "msg-allow-fallback", + Content: "hello", + Author: &dto.User{ + ID: "legacy-id", + }, + }) + if err != nil { + t.Fatalf("handleC2CMessage() error = %v", err) + } + + inbound := waitInboundMessage(t, messageBus) + if inbound.SenderID != "user-openid-123" { + t.Fatalf("inbound.SenderID = %q, want %q", inbound.SenderID, "user-openid-123") + } + if inbound.Metadata["account_id"] != "user-openid-123" { + t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "user-openid-123") + } + if inbound.Metadata["legacy_account_id"] != "legacy-id" { + t.Fatalf("legacy_account_id metadata = %q, want %q", inbound.Metadata["legacy_account_id"], "legacy-id") + } +} + +func TestHandleGroupATMessage_PrefersGroupOpenIDFromRawPayload(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &QQChannel{ + BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, nil), + dedup: make(map[string]time.Time), + done: make(chan struct{}), + ctx: context.Background(), + } + + raw := []byte(`{"d":{"group_openid":"group-openid-abc","author":{"id":"legacy-member-id","member_openid":"member-openid-xyz"}}}`) + err := ch.handleGroupATMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSGroupATMessageData{ + ID: "group-openid-msg", + GroupID: "legacy-group-id", + Content: "@bot hello", + Author: &dto.User{ + ID: "legacy-member-id", + }, + }) + if err != nil { + t.Fatalf("handleGroupATMessage() error = %v", err) + } + + inbound := waitInboundMessage(t, messageBus) + if inbound.Peer.ID != "group-openid-abc" { + t.Fatalf("inbound.Peer.ID = %q, want %q", inbound.Peer.ID, "group-openid-abc") + } + if inbound.Metadata["group_id"] != "group-openid-abc" { + t.Fatalf("group_id metadata = %q, want %q", inbound.Metadata["group_id"], "group-openid-abc") + } + if inbound.Metadata["account_id"] != "member-openid-xyz" { + t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "member-openid-xyz") + } +} + +func TestHandleGroupATMessage_AllowListFallsBackToLegacyMemberID(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &QQChannel{ + BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, []string{"legacy-member-id"}), + dedup: make(map[string]time.Time), + done: make(chan struct{}), + ctx: context.Background(), + } + + raw := []byte(`{"d":{"group_openid":"group-openid-abc","author":{"id":"legacy-member-id","member_openid":"member-openid-xyz"}}}`) + err := ch.handleGroupATMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSGroupATMessageData{ + ID: "group-allow-fallback", + GroupID: "legacy-group-id", + Content: "@bot hello", + Author: &dto.User{ + ID: "legacy-member-id", + }, + }) + if err != nil { + t.Fatalf("handleGroupATMessage() error = %v", err) + } + + inbound := waitInboundMessage(t, messageBus) + if inbound.SenderID != "member-openid-xyz" { + t.Fatalf("inbound.SenderID = %q, want %q", inbound.SenderID, "member-openid-xyz") + } + if inbound.Metadata["account_id"] != "member-openid-xyz" { + t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "member-openid-xyz") + } + if inbound.Metadata["legacy_account_id"] != "legacy-member-id" { + t.Fatalf("legacy_account_id metadata = %q, want %q", inbound.Metadata["legacy_account_id"], "legacy-member-id") + } +} + +func TestParseQQOpenIDs(t *testing.T) { + tests := []struct { + name string + raw []byte + wantUser string + wantMember string + wantGroup string + }{ + { + name: "empty", + raw: nil, + wantUser: "", + wantMember: "", + wantGroup: "", + }, + { + name: "invalid json", + raw: []byte("{not-json}"), + wantUser: "", + wantMember: "", + wantGroup: "", + }, + { + name: "all openids", + raw: []byte(`{"d":{"group_openid":"group-1","author":{"user_openid":"user-1","member_openid":"member-1"}}}`), + wantUser: "user-1", + wantMember: "member-1", + wantGroup: "group-1", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + userOpenID, memberOpenID, groupOpenID := parseQQOpenIDs(tc.raw) + if userOpenID != tc.wantUser || memberOpenID != tc.wantMember || groupOpenID != tc.wantGroup { + t.Fatalf( + "parseQQOpenIDs(%s) = (%q, %q, %q), want (%q, %q, %q)", + tc.name, + userOpenID, + memberOpenID, + groupOpenID, + tc.wantUser, + tc.wantMember, + tc.wantGroup, + ) + } + }) + } +} + func TestHandleC2CMessage_AttachmentOnlyPublishesMedia(t *testing.T) { messageBus := bus.NewMessageBus() store := media.NewFileMediaStore() @@ -682,7 +864,7 @@ func waitInboundMessage(t *testing.T, messageBus *bus.MessageBus) bus.InboundMes func writeTempFile(t *testing.T, dir, name string, content []byte) string { t.Helper() - path := dir + "/" + name + path := filepath.Join(dir, name) if err := os.WriteFile(path, content, 0o600); err != nil { t.Fatalf("WriteFile() error = %v", err) }