diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go new file mode 100644 index 0000000..622e6e2 --- /dev/null +++ b/plugins/channels/qq_obv11/adapter.go @@ -0,0 +1,222 @@ +// Package qq_obv11 implements the built-in QQ OneBot v11 channel adapter for +// Cyrene v2. It speaks the OneBot v11 WebSocket protocol (both server mode, +// where the adapter accepts inbound OneBot connections, and client mode, where +// it dials out to a OneBot endpoint), converts native CQ-code / segment payloads +// into the normalized adapter.UnifiedMessage model, and answers via send_msg API +// calls subject to a send rate limit. +// +// The design authority is DevDocs/design/08-渠道接入.md (sections 1, 3, and 4). +// It depends only on internal/channel/adapter, internal/logger, and the standard +// library; the WebSocket frame codec is implemented locally because the module +// pins no third-party websocket dependency. +package qq_obv11 + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // defaultMaxMessageLength bounds a single outgoing QQ message length. + defaultMaxMessageLength = 4000 + // defaultRecommendBurstMax is the safe burst size for one reply. + defaultRecommendBurstMax = 4 + // defaultMaxProactivePerWindow caps proactive sends within the window. + defaultMaxProactivePerWindow = 5 + // defaultProactiveWindow is the sliding window for the send rate limit. + defaultProactiveWindow = 10 * time.Second +) + +// Config configures a qq_obv11 Adapter. +type Config struct { + // Mode selects the WS role: "server" accepts inbound OneBot connections, + // "client" dials the OneBot endpoint. Empty defaults to "server". + Mode string + // ListenAddr is the TCP address to bind in server mode (host:port). + ListenAddr string + // RemoteURL is the ws:// endpoint to dial in client mode. + RemoteURL string + // AccessToken is the optional OneBot access token for both modes. + AccessToken string + // SelfID is the bot's own QQ number when known. + SelfID string + // MaxProactivePerWindow caps sends within ProactiveWindow. Zero uses the + // default. + MaxProactivePerWindow int + // ProactiveWindow is the rate-limit sliding window. Zero uses the default. + ProactiveWindow time.Duration + // SendInterval is a minimum delay between consecutive sends, when set. + SendInterval time.Duration + // DialContext dials the WS endpoint in client mode. Nil uses net.Dialer. + DialContext DialContextFunc + // Logger is the base logger. Nil builds one from internal/logger. + Logger *slog.Logger +} + +// Adapter is the OneBot v11 implementation of adapter.PlatformAdapter. +// +// It is concurrency-safe: one goroutine runs the read loop while multiple +// goroutines may send (writes are serialized on the wsConn and sends are rate +// limited). +type Adapter struct { + cfg Config + log *slog.Logger + + mu sync.RWMutex + conn *wsConn + ln net.Listener + srv *http.Server + started bool + connected bool + cancel context.CancelFunc + lifeCtx context.Context + newConn chan struct{} + + handlerMu sync.RWMutex + handler adapter.MessageHandler + emitMu sync.Mutex + emit func(adapter.UnifiedMessage) + + limiter *rateLimiter + + groupMu sync.RWMutex + groups map[string]string + + lastSend time.Time +} + +// Compile-time conformance with the channel contract. +var _ adapter.PlatformAdapter = (*Adapter)(nil) + +// NewAdapter builds a qq_obv11 Adapter from cfg, applying safe defaults. +func NewAdapter(cfg Config) *Adapter { + if cfg.Mode == "" { + cfg.Mode = "server" + } + if cfg.MaxProactivePerWindow <= 0 { + cfg.MaxProactivePerWindow = defaultMaxProactivePerWindow + } + if cfg.ProactiveWindow <= 0 { + cfg.ProactiveWindow = defaultProactiveWindow + } + base := cfg.Logger + if base == nil { + base = logger.New(logger.Options{}) + } + return &Adapter{ + cfg: cfg, + log: logger.WithSource(base, "plugins/channels/qq_obv11"), + groups: map[string]string{}, + newConn: make(chan struct{}, 1), + limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), + } +} + +// Name returns the platform identifier "qq-obv11". +func (a *Adapter) Name() string { return platformName } + +// Config returns a snapshot of the adapter configuration as a plain map. +func (a *Adapter) Config() map[string]any { + return map[string]any{ + "mode": a.cfg.Mode, + "listenAddr": a.cfg.ListenAddr, + "remoteURL": a.cfg.RemoteURL, + "accessToken": a.cfg.AccessToken, + "selfID": a.cfg.SelfID, + "maxProactivePerWindow": a.cfg.MaxProactivePerWindow, + "proactiveWindowSeconds": a.cfg.ProactiveWindow.Seconds(), + "sendIntervalSeconds": a.cfg.SendInterval.Seconds(), + } +} + +// Connect establishes the session according to the configured mode. It is +// idempotent. +func (a *Adapter) Connect(ctx context.Context) error { + a.mu.Lock() + if a.started { + a.mu.Unlock() + return nil + } + a.mu.Unlock() + + var err error + if a.cfg.Mode == "client" { + err = a.connectClient(ctx) + } else { + err = a.connectServer(ctx) + } + if err != nil { + return fmt.Errorf("qq-obv11: connect: %w", err) + } + + a.mu.Lock() + a.started = true + a.lifeCtx, a.cancel = context.WithCancel(context.Background()) + a.mu.Unlock() + + go a.readLoop(a.lifeCtx) + return nil +} + +// Disconnect tears down the session. It is idempotent. +func (a *Adapter) Disconnect(ctx context.Context) error { + a.mu.Lock() + started := a.started + a.started = false + cancel := a.cancel + a.cancel = nil + a.lifeCtx = nil + srv := a.srv + ln := a.ln + conn := a.conn + a.conn = nil + a.connected = false + a.ln = nil + a.srv = nil + a.mu.Unlock() + + if !started && cancel == nil { + return nil + } + if cancel != nil { + cancel() + } + if conn != nil { + _ = conn.Close() + } + if srv != nil { + shutdownCtx, c := context.WithTimeout(context.Background(), 3*time.Second) + defer c() + if serr := srv.Shutdown(shutdownCtx); serr != nil && !errors.Is(serr, context.DeadlineExceeded) { + return fmt.Errorf("qq-obv11: server shutdown: %w", serr) + } + } + if ln != nil { + _ = ln.Close() + } + return nil +} + +// IsConnected reports whether a live OneBot session exists. +func (a *Adapter) IsConnected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +// HealthCheck verifies the session is usable. +func (a *Adapter) HealthCheck(ctx context.Context) error { + if !a.IsConnected() { + return fmt.Errorf("qq-obv11: health check: %w", adapter.ErrNotConnected) + } + return nil +} diff --git a/plugins/channels/qq_obv11/adapter_test.go b/plugins/channels/qq_obv11/adapter_test.go new file mode 100644 index 0000000..65b0fe8 --- /dev/null +++ b/plugins/channels/qq_obv11/adapter_test.go @@ -0,0 +1,417 @@ +package qq_obv11 + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestAdapterNameAndConfig(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{ + Mode: "client", + RemoteURL: "ws://onebot.example:6700", + AccessToken: "secret", + MaxProactivePerWindow: 3, + ProactiveWindow: 5 * time.Second, + }) + if a.Name() != platformName { + t.Errorf("Name() = %q, want %q", a.Name(), platformName) + } + cfg := a.Config() + if cfg["mode"] != "client" { + t.Errorf("Config() mode = %v, want client", cfg["mode"]) + } + if cfg["remoteURL"] != "ws://onebot.example:6700" { + t.Errorf("Config() remoteURL = %v", cfg["remoteURL"]) + } + if cfg["accessToken"] != "secret" { + t.Errorf("Config() accessToken = %v", cfg["accessToken"]) + } + if cfg["maxProactivePerWindow"] != 3 { + t.Errorf("Config() maxProactivePerWindow = %v, want 3", cfg["maxProactivePerWindow"]) + } + if got := cfg["proactiveWindowSeconds"].(float64); got != 5 { + t.Errorf("Config() proactiveWindowSeconds = %v, want 5", got) + } +} + +func TestAdapterDefaultModeIsServer(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + if a.Config()["mode"] != "server" { + t.Errorf("default mode = %v, want server", a.Config()["mode"]) + } +} + +func TestAdapterCapabilities(t *testing.T) { + t.Parallel() + + caps := NewAdapter(Config{}).Capabilities().Normalize() + if caps.MaxMessageLength != defaultMaxMessageLength { + t.Errorf("MaxMessageLength = %d, want %d", caps.MaxMessageLength, defaultMaxMessageLength) + } + if caps.RecommendBurstMax != defaultRecommendBurstMax { + t.Errorf("RecommendBurstMax = %d, want %d", caps.RecommendBurstMax, defaultRecommendBurstMax) + } + if !caps.Markdown || !caps.Image || !caps.Voice || !caps.Proactive { + t.Errorf("expected markdown/image/voice/proactive on, got %+v", caps) + } + if caps.Reaction || caps.TypingHint { + t.Errorf("expected reaction/typing off, got %+v", caps) + } +} + +func TestAdapterSetMessageHandlerAndDispatch(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + got := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { + got <- m + }) + + want := adapter.UnifiedMessage{Platform: platformName, ChannelType: adapter.ChannelDirect, ContentType: adapter.ContentText, Content: "hi"} + a.dispatch(context.Background(), want) + + select { + case m := <-got: + if m.Content != "hi" { + t.Errorf("handler got content = %q, want hi", m.Content) + } + case <-time.After(3 * time.Second): + t.Fatal("handler not invoked") + } +} + +func TestAdapterHealthCheckDisconnected(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{Mode: "server", ListenAddr: "127.0.0.1:0"}) + if a.IsConnected() { + t.Error("IsConnected() = true before Connect, want false") + } + if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) { + t.Errorf("HealthCheck() error = %v, want ErrNotConnected", err) + } +} + +func TestAdapterSendWhenDisconnected(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelDirect, ChannelID: "100", Content: "x"}}, + }) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Errorf("SendMessage() error = %v, want ErrNotConnected", err) + } +} + +func TestAdapterDisconnectIdempotent(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + ctx := context.Background() + if err := a.Disconnect(ctx); err != nil { + t.Fatalf("first Disconnect error: %v", err) + } + if err := a.Disconnect(ctx); err != nil { + t.Fatalf("second Disconnect error: %v", err) + } +} + +// TestAdapterServerModeRoundTrip spins up the adapter as a server, connects a +// fake OneBot client, and verifies the message event is normalized and that a +// reply is emitted as a send_msg frame back to the client. +func TestAdapterServerModeRoundTrip(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{Mode: "server", ListenAddr: "127.0.0.1:0"}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + addr := a.Addr() + if addr == "" { + t.Fatal("Addr() empty after Connect") + } + if a.IsConnected() { + t.Error("server mode should not be connected before a client connects") + } + + msgCh := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m }) + + client, err := dialWS(context.Background(), nil, "ws://"+addr+"/", "") + if err != nil { + t.Fatalf("client dial error: %v", err) + } + defer client.Close() + + if !waitUntil(3*time.Second, a.IsConnected) { + t.Fatal("server never reported connected after client dialed") + } + + evt := `{"post_type":"message","message_type":"group","time":1700000000,"self_id":100,"message_id":9,"user_id":42,"group_id":123,"group_name":"Team","message":[{"type":"text","data":{"text":"hello"}}],"sender":{"user_id":42,"nickname":"Bob","card":"Bobby","role":"member"}}` + if err := client.WriteMessage(opText, []byte(evt)); err != nil { + t.Fatalf("client write error: %v", err) + } + + select { + case m := <-msgCh: + if m.Platform != platformName || m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" { + t.Errorf("unexpected message: platform=%q type=%q group=%q", m.Platform, m.ChannelType, m.GroupID) + } + if m.GroupName != "Team" || m.Content != "hello" { + t.Errorf("group message = %+v, want Team/hello", m) + } + if m.SenderName != "Bobby" || m.Sender.Role != adapter.RoleMember { + t.Errorf("sender = %+v, want Bobby/member", m.Sender) + } + case <-time.After(3 * time.Second): + t.Fatal("server handler never received a message") + } + + // Reply. + reply := adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{ + {ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}, + }} + if err := a.SendMessage(context.Background(), reply); err != nil { + t.Fatalf("SendMessage error: %v", err) + } + + seen := make(chan []byte, 1) + go func() { + _, payload, err := client.ReadMessage() + if err != nil { + seen <- nil + return + } + seen <- payload + }() + select { + case payload := <-seen: + if payload == nil { + t.Fatal("client never received a send_msg") + } + assertSendMsg(t, payload) + case <-time.After(3 * time.Second): + t.Fatal("client never received a send_msg frame") + } +} + +// TestAdapterClientModeReceiveAndSend spins up a fake OneBot server that pushes +// a group event, then verifies the adapter (client) normalizes it and replies. +func TestAdapterClientModeReceiveAndSend(t *testing.T) { + t.Parallel() + + gotCh := make(chan []byte, 4) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + t.Errorf("server upgrade error: %v", err) + return + } + defer conn.Close() + evt := `{"post_type":"message","message_type":"group","time":1700000001,"self_id":100,"message_id":10,"user_id":42,"group_id":123,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":42,"nickname":"Bob","role":"member"}}` + _ = conn.WriteMessage(opText, []byte(evt)) + for { + _, payload, err := conn.ReadMessage() + if err != nil { + return + } + gotCh <- payload + } + })) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: "client", RemoteURL: url}) + msgCh := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m }) + + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + if !a.IsConnected() { + t.Error("client mode IsConnected() = false after Connect, want true") + } + if err := a.HealthCheck(context.Background()); err != nil { + t.Errorf("HealthCheck() error = %v, want nil", err) + } + + select { + case m := <-msgCh: + if m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" || m.Content != "hi" { + t.Errorf("received message = %+v, want group/123/hi", m) + } + case <-time.After(3 * time.Second): + t.Fatal("adapter never received the pushed event") + } + + if err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}}, + }); err != nil { + t.Fatalf("SendMessage error: %v", err) + } + + select { + case payload := <-gotCh: + assertSendMsg(t, payload) + case <-time.After(3 * time.Second): + t.Fatal("fake OneBot server never received a send_msg") + } +} + +// TestSendProactiveRateLimited verifies proactive sends are rejected after the +// configured per-window quota is exceeded. +func TestSendProactiveRateLimited(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + return + } + defer conn.Close() + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + })) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{ + Mode: "client", + RemoteURL: url, + MaxProactivePerWindow: 2, + ProactiveWindow: 10 * time.Second, + }) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + msg := adapter.UnifiedMessage{ChannelType: adapter.ChannelDirect, ChannelID: "100", Content: "x"} + if err := a.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("first SendProactive error: %v", err) + } + if err := a.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("second SendProactive error: %v", err) + } + if err := a.SendProactive(context.Background(), msg); !errors.Is(err, ErrRateLimited) { + t.Errorf("third SendProactive error = %v, want ErrRateLimited", err) + } +} + +// TestReadMessagesPullMode verifies the pull-based ReadMessages sink receives +// normalized messages when no push handler is installed. +func TestReadMessagesPullMode(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + t.Errorf("server upgrade error: %v", err) + return + } + defer conn.Close() + evt := `{"post_type":"message","message_type":"private","time":1700000000,"self_id":100,"message_id":3,"user_id":200,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":200,"nickname":"Alice","role":""}}` + _ = conn.WriteMessage(opText, []byte(evt)) + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + })) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: "client", RemoteURL: url}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + emitCh := make(chan adapter.UnifiedMessage, 1) + go a.ReadMessages(ctx, func(m adapter.UnifiedMessage) { emitCh <- m }) + // Wait until the sink is registered before connecting, so the pushed event + // is not dropped before ReadMessages installs its emit callback. + if !waitUntil(3*time.Second, func() bool { + a.emitMu.Lock() + defer a.emitMu.Unlock() + return a.emit != nil + }) { + t.Fatal("ReadMessages never registered its emit sink") + } + + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + select { + case m := <-emitCh: + if m.ChannelType != adapter.ChannelDirect || m.ChannelID != "200" || m.Content != "hi" { + t.Errorf("pull-mode message = %+v, want direct/200/hi", m) + } + case <-time.After(3 * time.Second): + t.Fatal("ReadMessages never received the pushed event") + } +} + +func assertSendMsg(t *testing.T, payload []byte) { + t.Helper() + var req struct { + Action string `json:"action"` + Params struct { + MessageType string `json:"message_type"` + UserID int64 `json:"user_id"` + GroupID int64 `json:"group_id"` + Message []map[string]any `json:"message"` + } `json:"params"` + } + if err := json.Unmarshal(payload, &req); err != nil { + t.Fatalf("send_msg unmarshal error: %v (payload=%s)", err, payload) + } + if req.Action != "send_msg" { + t.Errorf("action = %q, want send_msg", req.Action) + } + if req.Params.MessageType != "group" { + t.Errorf("message_type = %q, want group", req.Params.MessageType) + } + if req.Params.GroupID != 123 { + t.Errorf("group_id = %d, want 123", req.Params.GroupID) + } + if len(req.Params.Message) != 1 { + t.Fatalf("message segments len = %d, want 1", len(req.Params.Message)) + } + first := req.Params.Message[0] + if first["type"] != "text" || first["data"].(map[string]any)["text"] != "pong" { + t.Errorf("first segment = %+v, want text pong", first) + } +} + +func waitUntil(timeout time.Duration, cond func() bool) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(20 * time.Millisecond) + } + return false +} diff --git a/plugins/channels/qq_obv11/message.go b/plugins/channels/qq_obv11/message.go new file mode 100644 index 0000000..17549cf --- /dev/null +++ b/plugins/channels/qq_obv11/message.go @@ -0,0 +1,470 @@ +package qq_obv11 + +// This file carries OneBot v11 protocol types and the conversions between the +// native event/segment model and the normalized adapter.UnifiedMessage / +// UnifiedResponse model. It understands both array-format messages +// ([]{"type","data"}) and the legacy CQ-code string format +// ([CQ:image,file=...]text), and maps QQ media (image/record/video/file) into +// adapter.Media and QQ mentions into adapter.Mention. + +import ( + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +const platformName = "qq-obv11" + +// OneBot v11 event as delivered over WebSocket. +type obEvent struct { + PostType string `json:"post_type"` + MessageType string `json:"message_type"` + Time int64 `json:"time"` + SelfID int64 `json:"self_id"` + MessageID int64 `json:"message_id"` + UserID int64 `json:"user_id"` + GroupID int64 `json:"group_id"` + GroupName string `json:"group_name"` + SubType string `json:"sub_type"` + RawMessage string `json:"raw_message"` + Message json.RawMessage `json:"message"` + Sender *obSender `json:"sender"` +} + +// obSender is the sender identity carried by a message event. +type obSender struct { + UserID int64 `json:"user_id"` + Nickname string `json:"nickname"` + Card string `json:"card"` + Role string `json:"role"` + Sex string `json:"sex"` +} + +// obSegment is one element of an array-format message. +type obSegment struct { + Type string `json:"type"` + Data map[string]any `json:"data"` +} + +// obSendMsg is the outbound OneBot v11 API request (action + params). +type obSendMsg struct { + Action string `json:"action"` + Params obSendParams `json:"params"` + Echo string `json:"echo,omitempty"` +} + +// obSendParams carries the send_msg parameters. +type obSendParams struct { + MessageType string `json:"message_type"` + UserID int64 `json:"user_id,omitempty"` + GroupID int64 `json:"group_id,omitempty"` + Message []obSegment `json:"message"` + AutoEscape bool `json:"auto_escape"` +} + +// parseEvent decodes a raw WebSocket text frame into a OneBot event. If the +// payload is not a message-style event the returned error is non-nil and the +// caller should attempt to fall back to another interpretation. +func parseEvent(payload []byte) (*obEvent, error) { + var ev obEvent + if err := json.Unmarshal(payload, &ev); err != nil { + return nil, fmt.Errorf("qq-obv11: decode event: %w", err) + } + return &ev, nil +} + +// toUnified converts a parsed message event into a normalized UnifiedMessage. +// raw preserves the original event payload for downstream tracing. +func (a *Adapter) toUnified(ev *obEvent, raw []byte) adapter.UnifiedMessage { + text, media, mentions, replyID, replyText, hasText := a.parseMessage(ev) + + senderID := formatID(ev.UserID) + senderName := "unknown" + role := adapter.RoleUnknown + if ev.Sender != nil { + senderName = firstNonEmpty(ev.Sender.Card, ev.Sender.Nickname, senderID) + role = roleFromOneBot(ev.Sender.Role) + } + + msg := adapter.UnifiedMessage{ + Platform: platformName, + SenderID: senderID, + SenderName: senderName, + Content: text, + ContentType: contentOf(media, hasText), + Media: media, + Mentions: mentions, + ReplyToID: replyID, + ReplyToText: replyText, + MessageID: formatID(ev.MessageID), + RawData: raw, + Timestamp: time.Unix(ev.Time, 0), + Sender: adapter.Actor{ + UserID: senderID, + DisplayName: senderName, + Role: role, + }, + } + + switch ev.MessageType { + case "group": + gid := formatID(ev.GroupID) + msg.ChannelType = adapter.ChannelGroup + msg.ChannelID = gid + msg.GroupID = gid + name := firstNonEmpty(ev.GroupName, a.groupName(gid)) + msg.GroupName = name + msg.Sender.GroupID = gid + msg.Sender.IsGroup = true + if name != "" { + a.rememberGroup(gid, name) + } + case "private": + msg.ChannelType = adapter.ChannelDirect + msg.ChannelID = senderID + } + return msg +} + +// parseMessage extracts the normalized parts of an event's message payload, +// handling both array-format segments and the CQ-code string format. hasText +// reports whether the message carries any real text (as opposed to only media / +// placeholder markers), which drives ContentType selection. +func (a *Adapter) parseMessage(ev *obEvent) (text string, media []adapter.Media, mentions []adapter.Mention, replyID, replyText string, hasText bool) { + raw := ev.Message + if len(raw) == 0 { + raw = []byte(ev.RawMessage) + } + if len(raw) > 0 { + switch raw[0] { + case '[': + return parseSegmentArray(raw) + case '"': + var s string + if err := json.Unmarshal(raw, &s); err == nil { + return parseCQMessage(s) + } + } + } + return parseCQMessage(ev.RawMessage) +} + +// parseSegmentArray parses an array-format message. +func parseSegmentArray(raw []byte) (string, []adapter.Media, []adapter.Mention, string, string, bool) { + var segs []obSegment + _ = json.Unmarshal(raw, &segs) + + var b strings.Builder + var media []adapter.Media + var mentions []adapter.Mention + var replyID, replyText string + hasText := false + + for _, s := range segs { + switch s.Type { + case "text": + t := segStr(s.Data, "text") + b.WriteString(t) + if t != "" { + hasText = true + } + case "at": + qq := segStr(s.Data, "qq") + if qq != "" { + mentions = append(mentions, adapter.Mention{UserID: qq}) + b.WriteString("@" + qq) + } + case "image": + media = append(media, mediaFromSeg(s.Data, adapter.MediaImage)) + b.WriteString("[图片]") + case "record": + media = append(media, mediaFromSeg(s.Data, adapter.MediaVoice)) + b.WriteString("[语音]") + case "video": + media = append(media, mediaFromSeg(s.Data, adapter.MediaVideo)) + b.WriteString("[视频]") + case "file": + media = append(media, mediaFromSeg(s.Data, adapter.MediaFile)) + b.WriteString("[文件]") + case "face": + b.WriteString("[表情]") + case "reply": + if id := segStr(s.Data, "id"); id != "" { + replyID = id + } + if t := segStr(s.Data, "text"); t != "" { + replyText = t + } + b.WriteString("[回复]") + case "json", "card": + b.WriteString("[卡片消息]") + default: + b.WriteString("[未知消息]") + } + } + return b.String(), media, mentions, replyID, replyText, hasText +} + +// parseCQMessage parses a CQ-code string, yielding text, media, mentions, and a +// hasText flag indicating real (non-placeholder) content. It scans for +// [CQ:type,...] codes manually (no package-level regexp, per the no-global-var +// constraint). +func parseCQMessage(s string) (string, []adapter.Media, []adapter.Mention, string, string, bool) { + var b strings.Builder + var media []adapter.Media + var mentions []adapter.Mention + var replyID, replyText string + hasText := false + + last := 0 + for i := 0; i < len(s); { + if !strings.HasPrefix(s[i:], "[CQ:") { + i++ + continue + } + end := strings.IndexByte(s[i:], ']') + if end < 0 { + break + } + fragment := s[last:i] + b.WriteString(fragment) + if fragment != "" { + hasText = true + } + code := s[i : i+end+1] + typ, params := decodeCQCode(code) + switch typ { + case "at": + if qq := params["qq"]; qq != "" { + mentions = append(mentions, adapter.Mention{UserID: qq}) + b.WriteString("@" + qq) + } + case "image": + media = append(media, mediaFromParams(params, adapter.MediaImage)) + b.WriteString("[图片]") + case "record": + media = append(media, mediaFromParams(params, adapter.MediaVoice)) + b.WriteString("[语音]") + case "video": + media = append(media, mediaFromParams(params, adapter.MediaVideo)) + b.WriteString("[视频]") + case "file": + media = append(media, mediaFromParams(params, adapter.MediaFile)) + b.WriteString("[文件]") + case "face": + b.WriteString("[表情]") + case "reply": + if id := params["id"]; id != "" { + replyID = id + } + replyText = params["text"] + b.WriteString("[回复]") + case "json", "card": + b.WriteString("[卡片消息]") + default: + b.WriteString(code) + } + last = i + end + 1 + i = last + } + tail := s[last:] + b.WriteString(tail) + if tail != "" { + hasText = true + } + return b.String(), media, mentions, replyID, replyText, hasText +} + +func decodeCQCode(code string) (string, map[string]string) { + inner := strings.TrimPrefix(code, "[CQ:") + inner = strings.TrimSuffix(inner, "]") + parts := strings.SplitN(inner, ",", 2) + typ := parts[0] + params := map[string]string{} + if len(parts) == 2 && parts[1] != "" { + for _, kv := range strings.Split(parts[1], ",") { + kvParts := strings.SplitN(kv, "=", 2) + if len(kvParts) != 2 { + continue + } + params[kvParts[0]] = unescapeCQ(kvParts[1]) + } + } + return typ, params +} + +func unescapeCQ(s string) string { + s = strings.ReplaceAll(s, ",", ",") + s = strings.ReplaceAll(s, "[", "[") + s = strings.ReplaceAll(s, "]", "]") + s = strings.ReplaceAll(s, "&", "&") + s = strings.ReplaceAll(s, "(", "(") + s = strings.ReplaceAll(s, ")", ")") + return s +} + +// mediaFromSeg builds a Media from an array-format segment's data map. +func mediaFromSeg(d map[string]any, kind adapter.MediaKind) adapter.Media { + url := segStr(d, "url") + file := segStr(d, "file") + if url == "" { + url = file + } + return adapter.Media{Kind: kind, URL: url, Name: file} +} + +// mediaFromParams builds a Media from parsed CQ params. +func mediaFromParams(p map[string]string, kind adapter.MediaKind) adapter.Media { + url := p["url"] + file := p["file"] + if url == "" { + url = file + } + return adapter.Media{Kind: kind, URL: url, Name: file} +} + +// outboundSegments converts a unified message into OneBot message segments. +func outboundSegments(msg adapter.UnifiedMessage) ([]obSegment, error) { + var segs []obSegment + if msg.Content != "" { + segs = append(segs, obSegment{Type: "text", Data: map[string]any{"text": msg.Content}}) + } + for _, m := range msg.Media { + file := firstNonEmpty(m.URL, m.Name) + switch m.Kind { + case adapter.MediaImage: + segs = append(segs, obSegment{Type: "image", Data: map[string]any{"file": file}}) + case adapter.MediaVoice: + segs = append(segs, obSegment{Type: "record", Data: map[string]any{"file": file}}) + case adapter.MediaVideo: + segs = append(segs, obSegment{Type: "video", Data: map[string]any{"file": file}}) + case adapter.MediaFile: + segs = append(segs, obSegment{Type: "file", Data: map[string]any{"file": file}}) + } + } + if len(segs) == 0 { + if msg.Content == "" { + return nil, fmt.Errorf("qq-obv11: message has no content or media: %w", adapter.ErrUnsupported) + } + segs = append(segs, obSegment{Type: "text", Data: map[string]any{"text": msg.Content}}) + } + return segs, nil +} + +// resolveTarget determines where a reply should be delivered. +func resolveTarget(msg adapter.UnifiedMessage) (messageType string, userID, groupID int64, err error) { + if msg.ChannelType == adapter.ChannelGroup { + gid := firstNonEmpty(msg.GroupID, msg.ChannelID, msg.Sender.GroupID) + if gid == "" { + return "", 0, 0, fmt.Errorf("qq-obv11: group reply missing target: %w", adapter.ErrUnsupported) + } + g, convErr := parseInt64(gid) + if convErr != nil { + return "", 0, 0, fmt.Errorf("qq-obv11: group target %q: %w", gid, convErr) + } + return "group", 0, g, nil + } + uid := firstNonEmpty(msg.ChannelID, msg.SenderID, msg.Sender.UserID) + if uid == "" { + return "", 0, 0, fmt.Errorf("qq-obv11: private reply missing target: %w", adapter.ErrUnsupported) + } + u, convErr := parseInt64(uid) + if convErr != nil { + return "", 0, 0, fmt.Errorf("qq-obv11: user target %q: %w", uid, convErr) + } + return "private", u, 0, nil +} + +func buildSendReq(messageType string, userID, groupID int64, segs []obSegment, echo string) ([]byte, error) { + req := obSendMsg{ + Action: "send_msg", + Params: obSendParams{ + MessageType: messageType, + UserID: userID, + GroupID: groupID, + Message: segs, + AutoEscape: false, + }, + Echo: echo, + } + b, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("qq-obv11: encode send_msg: %w", err) + } + return b, nil +} + +// contentOf picks the primary content type. When the message has no real text +// (only media / placeholder markers), it reflects the first media kind; +// otherwise text is the default. +func contentOf(media []adapter.Media, hasText bool) adapter.ContentType { + if !hasText && len(media) > 0 { + switch media[0].Kind { + case adapter.MediaImage: + return adapter.ContentImage + case adapter.MediaVoice: + return adapter.ContentVoice + default: + return adapter.ContentFile + } + } + return adapter.ContentText +} + +// roleFromOneBot maps the OneBot sender role field onto the actor Role enum. +// It surfaces the platform role verbatim as structured actor metadata; the +// channel-layer policy engine still decides real access from the bound owner +// identity, not from this field alone. +func roleFromOneBot(raw string) adapter.Role { + switch raw { + case "owner": + return adapter.RoleOwner + case "admin", "member": + return adapter.RoleMember + default: + return adapter.RoleUnknown + } +} + +func segStr(d map[string]any, key string) string { + if d == nil { + return "" + } + switch v := d[key].(type) { + case string: + return v + case float64: + return strconv.FormatInt(int64(v), 10) + case json.Number: + return v.String() + default: + return "" + } +} + +func formatID(id int64) string { + if id == 0 { + return "" + } + return strconv.FormatInt(id, 10) +} + +func parseInt64(s string) (int64, error) { + if s == "" { + return 0, fmt.Errorf("empty id") + } + return strconv.ParseInt(s, 10, 64) +} + +func firstNonEmpty(ss ...string) string { + for _, s := range ss { + if s != "" { + return s + } + } + return "" +} diff --git a/plugins/channels/qq_obv11/message_test.go b/plugins/channels/qq_obv11/message_test.go new file mode 100644 index 0000000..11ef1d0 --- /dev/null +++ b/plugins/channels/qq_obv11/message_test.go @@ -0,0 +1,285 @@ +package qq_obv11 + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestToUnifiedPrivateTextAndFace(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + raw := []byte(`{"post_type":"message","message_type":"private","time":1700000000,"self_id":100,"message_id":5,"user_id":200,"message":[{"type":"text","data":{"text":"hi "}},{"type":"face","data":{"id":1}}],"sender":{"user_id":200,"nickname":"Alice","role":"owner"}}`) + ev, err := parseEvent(raw) + if err != nil { + t.Fatalf("parseEvent error: %v", err) + } + msg := a.toUnified(ev, raw) + + if msg.Platform != platformName { + t.Errorf("Platform = %q, want %q", msg.Platform, platformName) + } + if msg.ChannelType != adapter.ChannelDirect { + t.Errorf("ChannelType = %q, want %q", msg.ChannelType, adapter.ChannelDirect) + } + if msg.ChannelID != "200" || msg.SenderID != "200" { + t.Errorf("ChannelID/SenderID = %q/%q, want 200/200", msg.ChannelID, msg.SenderID) + } + if msg.SenderName != "Alice" { + t.Errorf("SenderName = %q, want Alice", msg.SenderName) + } + if msg.Content != "hi [表情]" { + t.Errorf("Content = %q, want %q", msg.Content, "hi [表情]") + } + if msg.ContentType != adapter.ContentText { + t.Errorf("ContentType = %q, want text", msg.ContentType) + } + if msg.MessageID != "5" { + t.Errorf("MessageID = %q, want 5", msg.MessageID) + } + if !msg.Timestamp.Equal(time.Unix(1700000000, 0)) { + t.Errorf("Timestamp = %v, want %v", msg.Timestamp, time.Unix(1700000000, 0)) + } + if msg.Sender.Role != adapter.RoleOwner || !msg.Sender.IsOwner() { + t.Errorf("Sender.Role = %q, want owner", msg.Sender.Role) + } +} + +func TestToUnifiedGroupWithMediaAndMention(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{Mode: "server"}) + raw := []byte(`{"post_type":"message","message_type":"group","time":1700000001,"self_id":100,"message_id":9,"user_id":42,"group_id":123,"group_name":"Team","message":[{"type":"at","data":{"qq":"300"}},{"type":"text","data":{"text":" look "}},{"type":"image","data":{"file":"p.jpg","url":"http://cdn/p.jpg"}},{"type":"record","data":{"file":"v.amr"}}],"sender":{"user_id":42,"nickname":"Bob","card":"Bobby","role":"member"}}`) + ev, err := parseEvent(raw) + if err != nil { + t.Fatalf("parseEvent error: %v", err) + } + msg := a.toUnified(ev, raw) + + if !msg.IsGroup() { + t.Fatal("expected group message") + } + if msg.ChannelType != adapter.ChannelGroup || msg.ChannelID != "123" || msg.GroupID != "123" { + t.Errorf("group fields = %q/%q/%q, want group/123/123", msg.ChannelType, msg.ChannelID, msg.GroupID) + } + if msg.GroupName != "Team" { + t.Errorf("GroupName = %q, want Team", msg.GroupName) + } + if msg.SenderName != "Bobby" { + t.Errorf("SenderName = %q, want Bobby (card)", msg.SenderName) + } + if msg.Sender.Role != adapter.RoleMember { + t.Errorf("Sender.Role = %q, want member", msg.Sender.Role) + } + if len(msg.Mentions) != 1 || msg.Mentions[0].UserID != "300" { + t.Errorf("Mentions = %+v, want one @300", msg.Mentions) + } + if len(msg.Media) != 2 { + t.Fatalf("Media len = %d, want 2", len(msg.Media)) + } + if msg.Media[0].Kind != adapter.MediaImage || msg.Media[0].URL != "http://cdn/p.jpg" || msg.Media[0].Name != "p.jpg" { + t.Errorf("Media[0] = %+v, want image p.jpg", msg.Media[0]) + } + if msg.Media[1].Kind != adapter.MediaVoice { + t.Errorf("Media[1].Kind = %q, want voice", msg.Media[1].Kind) + } + if msg.Content != "@300 look [图片][语音]" { + t.Errorf("Content = %q, want %q", msg.Content, "@300 look [图片][语音]") + } +} + +func TestToUnifiedCQString(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + raw := []byte(`{"post_type":"message","message_type":"private","time":1700000002,"self_id":100,"message_id":7,"user_id":200,"message":"[CQ:at,qq=300]hello[CQ:image,file=a.jpg,url=http://cdn/a.jpg]","sender":{"user_id":200,"nickname":"Alice","role":""}}`) + ev, err := parseEvent(raw) + if err != nil { + t.Fatalf("parseEvent error: %v", err) + } + msg := a.toUnified(ev, raw) + + if msg.Content != "@300hello[图片]" { + t.Errorf("Content = %q, want %q", msg.Content, "@300hello[图片]") + } + if len(msg.Mentions) != 1 || msg.Mentions[0].UserID != "300" { + t.Errorf("Mentions = %+v, want @300", msg.Mentions) + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaImage || msg.Media[0].URL != "http://cdn/a.jpg" { + t.Errorf("Media = %+v, want image a.jpg", msg.Media) + } + if msg.Sender.Role != adapter.RoleUnknown { + t.Errorf("Sender.Role = %q, want unknown for empty role", msg.Sender.Role) + } +} + +func TestContentTypePureImage(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + raw := []byte(`{"post_type":"message","message_type":"private","time":1700000003,"self_id":100,"message_id":8,"user_id":200,"message":[{"type":"image","data":{"file":"a.jpg"}}],"sender":{"user_id":200,"nickname":"Alice"}}`) + ev, err := parseEvent(raw) + if err != nil { + t.Fatalf("parseEvent error: %v", err) + } + msg := a.toUnified(ev, raw) + + if msg.ContentType != adapter.ContentImage { + t.Errorf("ContentType = %q, want image", msg.ContentType) + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaImage { + t.Errorf("Media = %+v, want one image", msg.Media) + } +} + +func TestOutboundSegmentsTextAndImage(t *testing.T) { + t.Parallel() + + segs, err := outboundSegments(adapter.UnifiedMessage{ + Content: "pong", + Media: []adapter.Media{{Kind: adapter.MediaImage, URL: "http://cdn/x.jpg"}}, + }) + if err != nil { + t.Fatalf("outboundSegments error: %v", err) + } + if len(segs) != 2 { + t.Fatalf("segs len = %d, want 2", len(segs)) + } + if segs[0].Type != "text" || segs[0].Data["text"] != "pong" { + t.Errorf("segs[0] = %+v, want text pong", segs[0]) + } + if segs[1].Type != "image" || segs[1].Data["file"] != "http://cdn/x.jpg" { + t.Errorf("segs[1] = %+v, want image", segs[1]) + } +} + +func TestOutboundSegmentsEmptyErrors(t *testing.T) { + t.Parallel() + + _, err := outboundSegments(adapter.UnifiedMessage{}) + if err == nil { + t.Error("empty message: expected error, got nil") + } +} + +func TestResolveTargetGroupAndPrivate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + msg adapter.UnifiedMessage + wantType string + wantUser int64 + wantGroup int64 + wantErr bool + }{ + { + name: "group", + msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "x"}, + wantType: "group", + wantGroup: 123, + }, + { + name: "private by channel id", + msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelDirect, ChannelID: "200", Content: "x"}, + wantType: "private", + wantUser: 200, + }, + { + name: "private by sender", + msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelDirect, SenderID: "300", Content: "x"}, + wantType: "private", + wantUser: 300, + }, + { + name: "group missing target", + msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelGroup, Content: "x"}, + wantErr: true, + }, + { + name: "private missing target", + msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelDirect, Content: "x"}, + wantErr: true, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + gotType, gotUser, gotGroup, err := resolveTarget(tt.msg) + if (err != nil) != tt.wantErr { + t.Fatalf("resolveTarget error = %v, wantErr %v", err, tt.wantErr) + } + if err != nil { + return + } + if gotType != tt.wantType || gotUser != tt.wantUser || gotGroup != tt.wantGroup { + t.Errorf("resolveTarget = %q/%d/%d, want %q/%d/%d", gotType, gotUser, gotGroup, tt.wantType, tt.wantUser, tt.wantGroup) + } + }) + } +} + +func TestRoleFromOneBot(t *testing.T) { + t.Parallel() + + tests := []struct { + raw string + want adapter.Role + }{ + {"owner", adapter.RoleOwner}, + {"admin", adapter.RoleMember}, + {"member", adapter.RoleMember}, + {"", adapter.RoleUnknown}, + {"bogus", adapter.RoleUnknown}, + } + for _, tt := range tests { + tt := tt + t.Run(tt.raw, func(t *testing.T) { + t.Parallel() + if got := roleFromOneBot(tt.raw); got != tt.want { + t.Errorf("roleFromOneBot(%q) = %q, want %q", tt.raw, got, tt.want) + } + }) + } +} + +func TestBuildSendReq(t *testing.T) { + t.Parallel() + + segs := []obSegment{{Type: "text", Data: map[string]any{"text": "hi"}}} + payload, err := buildSendReq("group", 0, 123, segs, "echo_1") + if err != nil { + t.Fatalf("buildSendReq error: %v", err) + } + if string(payload) == "" { + t.Fatal("empty payload") + } +} + +func TestListChannelsCachesGroups(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{Mode: "server"}) + a.rememberGroup("123", "Team") + a.rememberGroup("456", "Gamer") + + channels, err := a.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels error: %v", err) + } + if len(channels) != 2 { + t.Fatalf("channels len = %d, want 2", len(channels)) + } + byID := map[string]adapter.ChannelInfo{} + for _, c := range channels { + byID[c.ChannelID] = c + } + if c := byID["123"]; c.GroupName != "Team" || c.ChannelType != adapter.ChannelGroup || c.GroupID != "123" { + t.Errorf("channel 123 = %+v", c) + } +} diff --git a/plugins/channels/qq_obv11/ratelimit.go b/plugins/channels/qq_obv11/ratelimit.go new file mode 100644 index 0000000..f05c712 --- /dev/null +++ b/plugins/channels/qq_obv11/ratelimit.go @@ -0,0 +1,52 @@ +package qq_obv11 + +// This file implements the sliding-window send rate limiter that keeps QQ +// OneBot proactive/reply traffic from bursting (see DevDocs/design/08 section +// 3, "注意风控"). + +import ( + "errors" + "sync" + "time" +) + +// ErrRateLimited is returned by SendMessage/SendProactive when the configured +// per-window send quota is exhausted. +var ErrRateLimited = errors.New("qq-obv11: send rate limit exceeded") + +// rateLimiter enforces a maximum number of events within a sliding window. It +// is disabled when max is non-positive. +type rateLimiter struct { + max int + window time.Duration + mu sync.Mutex + times []time.Time +} + +func newRateLimiter(max int, window time.Duration) *rateLimiter { + return &rateLimiter{max: max, window: window} +} + +// allow records a send at now and reports whether it fits within the window +// quota. A disabled limiter (max <= 0) always allows. The timestamp is only +// recorded when the send is accepted. +func (r *rateLimiter) allow(now time.Time) bool { + if r == nil || r.max <= 0 { + return true + } + r.mu.Lock() + defer r.mu.Unlock() + + cutoff := now.Add(-r.window) + i := 0 + for i < len(r.times) && !r.times[i].After(cutoff) { + i++ + } + r.times = r.times[i:] + + if len(r.times) >= r.max { + return false + } + r.times = append(r.times, now) + return true +} diff --git a/plugins/channels/qq_obv11/ratelimit_test.go b/plugins/channels/qq_obv11/ratelimit_test.go new file mode 100644 index 0000000..0d6dc39 --- /dev/null +++ b/plugins/channels/qq_obv11/ratelimit_test.go @@ -0,0 +1,67 @@ +package qq_obv11 + +import ( + "errors" + "testing" + "time" +) + +func TestRateLimiterAllowWithinWindow(t *testing.T) { + t.Parallel() + + rl := newRateLimiter(3, 10*time.Second) + now := time.Now() + for i := 0; i < 3; i++ { + if !rl.allow(now) { + t.Fatalf("allow #%d denied, want allowed", i) + } + } + if rl.allow(now) { + t.Error("4th allow in window accepted, want denied") + } +} + +func TestRateLimiterDeniedOverWindow(t *testing.T) { + t.Parallel() + + rl := newRateLimiter(2, 10*time.Second) + base := time.Now() + if !rl.allow(base) || !rl.allow(base) { + t.Fatal("first two sends should be allowed") + } + if rl.allow(base) { + t.Error("third send within window allowed, want denied") + } + // After the window elapses, the budget frees up. + if !rl.allow(base.Add(11 * time.Second)) { + t.Error("send after window elapsed denied, want allowed") + } +} + +func TestRateLimiterDisabledAllowsAll(t *testing.T) { + t.Parallel() + + rl := newRateLimiter(0, 10*time.Second) + for i := 0; i < 100; i++ { + if !rl.allow(time.Now()) { + t.Fatalf("disabled limiter denied send #%d", i) + } + } +} + +func TestRateLimiterNilAllows(t *testing.T) { + t.Parallel() + + var rl *rateLimiter + if !rl.allow(time.Now()) { + t.Error("nil limiter denied send, want allowed") + } +} + +func TestErrRateLimitedSentinel(t *testing.T) { + t.Parallel() + + if !errors.Is(ErrRateLimited, ErrRateLimited) { + t.Error("ErrRateLimited should be its own sentinel") + } +} diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go new file mode 100644 index 0000000..8474d2f --- /dev/null +++ b/plugins/channels/qq_obv11/session.go @@ -0,0 +1,346 @@ +package qq_obv11 + +// This file holds the message-flow half of the Adapter: the inbound read loop +// (server/client mode sessions, reconnect, dispatch) and the outbound send path +// (rate limit, chunk to the connection). It is split out of adapter.go to keep +// each file under the project's 500-line limit. + +import ( + "context" + "fmt" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// ReadMessages registers emit as the pull-based sink and blocks until ctx is +// canceled. Inbound events arrive via the read loop and are dispatched to +// SetMessageHandler (push) when set, otherwise to emit. +func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) { + a.emitMu.Lock() + a.emit = emit + a.emitMu.Unlock() + defer func() { + a.emitMu.Lock() + a.emit = nil + a.emitMu.Unlock() + }() + <-ctx.Done() +} + +// SendMessage delivers a reply (possibly several messages) to the channel, +// subject to the send rate limit. +func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { + if err := ctx.Err(); err != nil { + return fmt.Errorf("qq-obv11: send message: %w", err) + } + conn := a.currentConn() + if conn == nil { + return fmt.Errorf("qq-obv11: send message: %w", adapter.ErrNotConnected) + } + for i, msg := range resp.Messages { + if err := a.sendMessage(ctx, conn, msg); err != nil { + return fmt.Errorf("qq-obv11: send message #%d: %w", i, err) + } + } + return nil +} + +// SendProactive pushes a message not triggered by a recent inbound. It is +// rate limited (see design/08 section 3: OneBot needs a rate limit to avoid +// unusual burst patterns). +func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { + if err := ctx.Err(); err != nil { + return fmt.Errorf("qq-obv11: send proactive: %w", err) + } + conn := a.currentConn() + if conn == nil { + return fmt.Errorf("qq-obv11: send proactive: %w", adapter.ErrNotConnected) + } + return a.sendMessage(ctx, conn, msg) +} + +// SetMessageHandler installs the push-style inbound callback. A nil fn clears it. +func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) { + a.handlerMu.Lock() + a.handler = fn + a.handlerMu.Unlock() +} + +// Capabilities declares what the QQ OneBot v11 adapter supports. +func (a *Adapter) Capabilities() adapter.Capabilities { + return adapter.Capabilities{ + MaxMessageLength: defaultMaxMessageLength, + Markdown: true, + Image: true, + Voice: true, + Reaction: false, + TypingHint: false, + RecommendBurstMax: defaultRecommendBurstMax, + Proactive: true, + }.Normalize() +} + +// ListChannels returns the groups the adapter currently knows about. Enumerating +// friends / full group list requires the OneBot HTTP API, which is out of scope +// for the WebSocket-only M6 adapter; it is reported as a follow-up. +func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) { + a.groupMu.RLock() + defer a.groupMu.RUnlock() + out := make([]adapter.ChannelInfo, 0, len(a.groups)) + for id, name := range a.groups { + out = append(out, adapter.ChannelInfo{ + ChannelID: id, + ChannelType: adapter.ChannelGroup, + ChannelName: name, + GroupID: id, + GroupName: name, + }) + } + return out, nil +} + +// Addr returns the bound listen address in server mode ("" when not serving). +func (a *Adapter) Addr() string { + a.mu.RLock() + defer a.mu.RUnlock() + if a.ln == nil { + return "" + } + return a.ln.Addr().String() +} + +// readLoop is the single inbound-reader goroutine. It reads frames from the +// active connection, dispatching message events; in client mode it reconnects +// after drops, in server mode it waits for a new OneBot client. +func (a *Adapter) readLoop(ctx context.Context) { + for { + conn := a.currentConn() + if conn == nil { + if a.cfg.Mode == "client" { + if err := a.connectClient(ctx); err != nil { + a.log.Warn("onebot reconnect failed", "err", err) + if !sleepCtx(ctx, 3*time.Second) { + return + } + continue + } + } else if !a.waitConn(ctx) { + return + } + conn = a.currentConn() + } + a.readConn(ctx, conn) + if ctx.Err() != nil { + return + } + } +} + +// readConn reads frames from conn until it closes, the context is canceled, or +// a close frame is received, marking the adapter disconnected on exit. +func (a *Adapter) readConn(ctx context.Context, conn *wsConn) { + watchDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + _ = conn.Close() + case <-watchDone: + } + }() + defer close(watchDone) + + for { + opcode, payload, err := conn.ReadMessage() + if err != nil { + a.setDisconnected() + return + } + switch opcode { + case opPing: + _ = conn.WriteMessage(opPong, payload) + case opText, opBinary: + a.handlePayload(ctx, payload) + case opClose: + _ = conn.WriteMessage(opClose, nil) + a.setDisconnected() + return + } + } +} + +// handlePayload parses a text frame, converting message events into normalized +// UnifiedMessage and dispatching them. +func (a *Adapter) handlePayload(ctx context.Context, payload []byte) { + ev, err := parseEvent(payload) + if err != nil { + a.log.Debug("non-event payload ignored", "err", err) + return + } + if ev.PostType != "message" { + return + } + msg := a.toUnified(ev, payload) + if err := msg.Validate(); err != nil { + a.log.Warn("invalid unified message dropped", "err", err) + return + } + a.dispatch(ctx, msg) +} + +// dispatch routes an inbound message to the registered push handler, falling +// back to the ReadMessages sink. +func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { + a.handlerMu.RLock() + fn := a.handler + a.handlerMu.RUnlock() + if fn != nil { + fn(ctx, msg) + return + } + a.emitMu.Lock() + emit := a.emit + a.emitMu.Unlock() + if emit != nil { + emit(msg) + } +} + +// sendMessage builds and writes a single send_msg request, enforcing the rate +// limit and writing the payload to the connection. +func (a *Adapter) sendMessage(ctx context.Context, conn *wsConn, msg adapter.UnifiedMessage) error { + segs, err := outboundSegments(msg) + if err != nil { + return err + } + messageType, userID, groupID, err := resolveTarget(msg) + if err != nil { + return err + } + echo := fmt.Sprintf("echo_%d", time.Now().UnixNano()) + payload, err := buildSendReq(messageType, userID, groupID, segs, echo) + if err != nil { + return err + } + if a.cfg.SendInterval > 0 { + if err := a.waitSendInterval(ctx); err != nil { + return err + } + } + if err := a.tryAcquireSend(); err != nil { + return err + } + if err := conn.WriteMessage(opText, payload); err != nil { + a.setDisconnected() + return fmt.Errorf("qq-obv11: write send_msg: %w", err) + } + a.mu.Lock() + a.lastSend = time.Now() + a.mu.Unlock() + return nil +} + +// waitSendInterval sleeps so consecutive sends are at least SendInterval apart. +func (a *Adapter) waitSendInterval(ctx context.Context) error { + for { + a.mu.RLock() + last := a.lastSend + a.mu.RUnlock() + if last.IsZero() { + return nil + } + wait := a.cfg.SendInterval - time.Since(last) + if wait <= 0 { + return nil + } + select { + case <-ctx.Done(): + return fmt.Errorf("qq-obv11: send interval: %w", ctx.Err()) + case <-time.After(wait): + } + } +} + +// tryAcquireSend checks the send rate limiter, returning ErrRateLimited when the +// configured window quota is exhausted. +func (a *Adapter) tryAcquireSend() error { + if !a.limiter.allow(time.Now()) { + return fmt.Errorf("qq-obv11: %w", ErrRateLimited) + } + return nil +} + +// setConn installs a new active connection, closing any previous one and waking +// the read loop (server mode). +func (a *Adapter) setConn(conn *wsConn) { + a.mu.Lock() + old := a.conn + a.conn = conn + a.connected = true + a.mu.Unlock() + if old != nil && old != conn { + _ = old.Close() + } + select { + case a.newConn <- struct{}{}: + default: + } +} + +// setDisconnected marks the session down, dropping the current connection. +func (a *Adapter) setDisconnected() { + a.mu.Lock() + if a.conn != nil { + _ = a.conn.Close() + a.conn = nil + } + a.connected = false + a.mu.Unlock() +} + +// currentConn returns the active wsConn, or nil. +func (a *Adapter) currentConn() *wsConn { + a.mu.RLock() + defer a.mu.RUnlock() + return a.conn +} + +// waitConn blocks in server mode until a OneBot client connects or ctx ends. +func (a *Adapter) waitConn(ctx context.Context) bool { + for a.currentConn() == nil { + select { + case <-a.newConn: + case <-ctx.Done(): + return false + case <-time.After(time.Second): + } + } + return true +} + +func (a *Adapter) rememberGroup(id, name string) { + if id == "" { + return + } + a.groupMu.Lock() + a.groups[id] = name + a.groupMu.Unlock() +} + +func (a *Adapter) groupName(id string) string { + if id == "" { + return "" + } + a.groupMu.RLock() + defer a.groupMu.RUnlock() + return a.groups[id] +} + +func sleepCtx(ctx context.Context, d time.Duration) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } +} diff --git a/plugins/channels/qq_obv11/websocket.go b/plugins/channels/qq_obv11/websocket.go new file mode 100644 index 0000000..9bf111b --- /dev/null +++ b/plugins/channels/qq_obv11/websocket.go @@ -0,0 +1,406 @@ +package qq_obv11 + +// This file implements a minimal RFC 6455 WebSocket (text/binary frames, +// ping/pong, close, per-message client masking) using only the standard +// library. The project pins no third-party WebSocket dependency (go.mod must +// not change), so the OneBot v11 adapter carries its own tiny frame codec. It +// is deliberately narrow: it supports what OneBot v11 needs — UTF-8 JSON text +// frames carried over a single upgraded TCP connection — and nothing more. + +import ( + "bufio" + "context" + "crypto/rand" + "crypto/sha1" + "crypto/tls" + "encoding/base64" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "sync" + "time" +) + +// WebSocket control/data opcodes (RFC 6455 section 5.2). +const ( + opContinuation = 0x0 + opText = 0x1 + opBinary = 0x2 + opClose = 0x8 + opPing = 0x9 + opPong = 0xA +) + +// wsGUID is the fixed handshake GUID from RFC 6455 section 1.3. +const wsGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +// maxFramePayload caps a single WebSocket frame so a malicious peer cannot make +// the adapter buffer an unbounded payload. Chat/JSON payloads are tiny; this is +// generous. +const maxFramePayload = 10 << 20 // 10 MiB + +// DialContextFunc dials a raw TCP (or TLS) connection. It is injectable so +// tests can route the handshake through a fake server; the default is a +// net.Dialer. +type DialContextFunc func(ctx context.Context, network, address string) (net.Conn, error) + +// wsConn is a single WebSocket connection over an upgraded TCP stream. It is +// safe for one concurrent reader plus concurrent writers (writes are mutex +// serialized); exactly one goroutine may call ReadMessage at a time. +type wsConn struct { + conn net.Conn + br *bufio.Reader + bw *bufio.Writer + isClient bool + writeMu sync.Mutex + closeOnce sync.Once +} + +// newWSConn wraps an upgraded connection. br and bw are the buffered reader and +// writer from the handshake; passing nil causes them to be created from conn. +// isClient reports whether this side is the WebSocket client (which must mask +// its outgoing frames). +func newWSConn(conn net.Conn, br *bufio.Reader, bw *bufio.Writer, isClient bool) *wsConn { + if br == nil { + br = bufio.NewReader(conn) + } + if bw == nil { + bw = bufio.NewWriter(conn) + } + return &wsConn{conn: conn, br: br, bw: bw, isClient: isClient} +} + +// Close closes the underlying connection exactly once. +func (c *wsConn) Close() error { + var err error + c.closeOnce.Do(func() { err = c.conn.Close() }) + return err +} + +// WriteMessage writes a single unmasked/known-masked data or control frame. +// The caller must hold no locks. +func (c *wsConn) WriteMessage(opcode byte, payload []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + + b0 := byte(0x80) | opcode // FIN set on the single frame + if err := c.bw.WriteByte(b0); err != nil { + return err + } + + maskBit := byte(0) + if c.isClient { + maskBit = 0x80 + } + if err := writeFrameHeader(c.bw, maskBit, len(payload)); err != nil { + return err + } + + var maskKey [4]byte + if c.isClient { + if _, err := rand.Read(maskKey[:]); err != nil { + return err + } + if _, err := c.bw.Write(maskKey[:]); err != nil { + return err + } + } + + if !c.isClient { + if _, err := c.bw.Write(payload); err != nil { + return err + } + return c.bw.Flush() + } + if err := writeMaskedPayload(c.bw, payload, maskKey); err != nil { + return err + } + return c.bw.Flush() +} + +func writeFrameHeader(w *bufio.Writer, maskBit byte, n int) error { + switch { + case n < 126: + return w.WriteByte(maskBit | byte(n)) + case n <= 0xFFFF: + if err := w.WriteByte(maskBit | 126); err != nil { + return err + } + var ext [2]byte + binary.BigEndian.PutUint16(ext[:], uint16(n)) + _, err := w.Write(ext[:]) + return err + default: + if err := w.WriteByte(maskBit | 127); err != nil { + return err + } + var ext [8]byte + binary.BigEndian.PutUint64(ext[:], uint64(n)) + _, err := w.Write(ext[:]) + return err + } +} + +func writeMaskedPayload(w *bufio.Writer, payload []byte, maskKey [4]byte) error { + var buf [4096]byte + for off := 0; off < len(payload); off += len(buf) { + end := off + len(buf) + if end > len(payload) { + end = len(payload) + } + for i, b := range payload[off:end] { + buf[i] = b ^ maskKey[(off+i)&3] + } + if _, err := w.Write(buf[:end-off]); err != nil { + return err + } + } + return nil +} + +// ReadMessage reads the next complete data message, reassembling fragmented +// frames, replying to pings, and terminating (io.EOF) on a close frame or +// network error. Only one goroutine may call it. +func (c *wsConn) ReadMessage() (byte, []byte, error) { + var ( + payload []byte + dataOp byte + started bool + ) + + for { + fin, opcode, data, err := c.readFrame() + if err != nil { + return 0, nil, err + } + + switch opcode { + case opPing: + if err := c.WriteMessage(opPong, data); err != nil { + return 0, nil, err + } + continue + case opPong: + continue + case opClose: + _ = c.WriteMessage(opClose, nil) + return opClose, nil, io.EOF + case opContinuation: + if !started { + return 0, nil, errors.New("websocket: unexpected continuation frame") + } + payload = append(payload, data...) + if fin { + return dataOp, payload, nil + } + case opText, opBinary: + if started { + return 0, nil, errors.New("websocket: new data frame during fragmented message") + } + dataOp = opcode + if fin { + return opcode, data, nil + } + started = true + payload = append([]byte(nil), data...) + default: + return 0, nil, fmt.Errorf("websocket: unsupported opcode %d", opcode) + } + } +} + +// readFrame reads a single frame header and payload, unmasking when required. +func (c *wsConn) readFrame() (fin bool, opcode byte, payload []byte, err error) { + var hdr [2]byte + if _, err = io.ReadFull(c.br, hdr[:]); err != nil { + return + } + fin = hdr[0]&0x80 != 0 + opcode = hdr[0] & 0x0F + masked := hdr[1]&0x80 != 0 + + length := uint64(hdr[1] & 0x7F) + switch length { + case 126: + var ext [2]byte + if _, err = io.ReadFull(c.br, ext[:]); err != nil { + return + } + length = uint64(binary.BigEndian.Uint16(ext[:])) + case 127: + var ext [8]byte + if _, err = io.ReadFull(c.br, ext[:]); err != nil { + return + } + length = binary.BigEndian.Uint64(ext[:]) + } + if length > maxFramePayload { + err = fmt.Errorf("websocket: frame too large (%d bytes)", length) + return + } + + var maskKey [4]byte + if masked { + if _, err = io.ReadFull(c.br, maskKey[:]); err != nil { + return + } + } + + payload = make([]byte, length) + if _, err = io.ReadFull(c.br, payload); err != nil { + return + } + if masked { + for i := range payload { + payload[i] ^= maskKey[i&3] + } + } + return fin, opcode, payload, nil +} + +// computeAccept derives the Sec-WebSocket-Accept value for a handshake key. +func computeAccept(key string) string { + h := sha1.New() + _, _ = io.WriteString(h, key+wsGUID) + return base64.StdEncoding.EncodeToString(h.Sum(nil)) +} + +// randomWSKey returns a random WebSocket handshake key. +func randomWSKey() string { + b := make([]byte, 16) + _, _ = rand.Read(b) + return base64.StdEncoding.EncodeToString(b) +} + +// upgradeWS performs the server side of the WebSocket handshake by hijacking +// the HTTP connection and returning the upgraded wsConn. +func upgradeWS(w http.ResponseWriter, r *http.Request) (*wsConn, error) { + if !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") || + !headerHasToken(r.Header, "Connection", "upgrade") { + http.Error(w, "not a websocket upgrade", http.StatusBadRequest) + return nil, errors.New("websocket: not an upgrade request") + } + key := r.Header.Get("Sec-WebSocket-Key") + if key == "" { + http.Error(w, "missing Sec-WebSocket-Key", http.StatusBadRequest) + return nil, errors.New("websocket: missing handshake key") + } + + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "hijacking not supported", http.StatusInternalServerError) + return nil, errors.New("websocket: hijacking not supported") + } + netConn, brw, err := hj.Hijack() + if err != nil { + return nil, fmt.Errorf("websocket: hijack: %w", err) + } + + resp := "HTTP/1.1 101 Switching Protocols\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Accept: " + computeAccept(key) + "\r\n\r\n" + if _, err := brw.WriteString(resp); err != nil { + _ = netConn.Close() + return nil, err + } + if err := brw.Flush(); err != nil { + _ = netConn.Close() + return nil, err + } + return newWSConn(netConn, brw.Reader, brw.Writer, false), nil +} + +// headerHasToken reports whether a comma-separated header value contains the +// given token case-insensitively. +func headerHasToken(h http.Header, name, token string) bool { + for _, val := range h.Values(name) { + for _, part := range strings.Split(val, ",") { + if strings.EqualFold(strings.TrimSpace(part), token) { + return true + } + } + } + return false +} + +// dialWS performs the client side of the WebSocket handshake, returning the +// upgraded wsConn. accessToken, when non-empty, is sent both as an +// Authorization: Bearer header and as an access_token query parameter for +// OneBot implementations that expect either. +func dialWS(ctx context.Context, dial DialContextFunc, rawURL, accessToken string) (*wsConn, error) { + if dial == nil { + dial = (&net.Dialer{Timeout: 15 * time.Second}).DialContext + } + u, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("websocket: parse url %q: %w", rawURL, err) + } + if u.Scheme != "ws" && u.Scheme != "wss" { + return nil, fmt.Errorf("websocket: unsupported scheme %q (want ws/wss)", u.Scheme) + } + + host := u.Host + if u.Port() == "" { + if u.Scheme == "wss" { + host += ":443" + } else { + host += ":80" + } + } + conn, err := dial(ctx, "tcp", host) + if err != nil { + return nil, fmt.Errorf("websocket: dial %s: %w", host, err) + } + + if u.Scheme == "wss" { + tlsConn := tls.Client(conn, &tls.Config{ServerName: u.Hostname()}) + if err := tlsConn.HandshakeContext(ctx); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("websocket: tls handshake: %w", err) + } + conn = tlsConn + } + + brw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)) + key := randomWSKey() + var b strings.Builder + b.WriteString("GET " + u.RequestURI() + " HTTP/1.1\r\n") + b.WriteString("Host: " + u.Host + "\r\n") + b.WriteString("Upgrade: websocket\r\n") + b.WriteString("Connection: Upgrade\r\n") + b.WriteString("Sec-WebSocket-Key: " + key + "\r\n") + b.WriteString("Sec-WebSocket-Version: 13\r\n") + if accessToken != "" { + b.WriteString("Authorization: Bearer " + accessToken + "\r\n") + } + b.WriteString("\r\n") + if _, err := brw.WriteString(b.String()); err != nil { + _ = conn.Close() + return nil, err + } + if err := brw.Flush(); err != nil { + _ = conn.Close() + return nil, err + } + + resp, err := http.ReadResponse(brw.Reader, &http.Request{Method: "GET"}) + if err != nil { + _ = conn.Close() + return nil, fmt.Errorf("websocket: read handshake response: %w", err) + } + if resp.StatusCode != http.StatusSwitchingProtocols { + _ = conn.Close() + return nil, fmt.Errorf("websocket: unexpected handshake status %d", resp.StatusCode) + } + if got := resp.Header.Get("Sec-WebSocket-Accept"); got != computeAccept(key) { + _ = conn.Close() + return nil, errors.New("websocket: bad Sec-WebSocket-Accept in response") + } + return newWSConn(conn, brw.Reader, brw.Writer, true), nil +} diff --git a/plugins/channels/qq_obv11/websocket_test.go b/plugins/channels/qq_obv11/websocket_test.go new file mode 100644 index 0000000..3dba42c --- /dev/null +++ b/plugins/channels/qq_obv11/websocket_test.go @@ -0,0 +1,146 @@ +package qq_obv11 + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// TestWSClientServerRoundTrip exercises a full handshake plus bidirectional +// text frames over a real loopback server, verifying that a client's masked +// frames are unmasked by the server and the server's unmasked frames are read +// correctly by the client. +func TestWSClientServerRoundTrip(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + t.Errorf("upgrade error: %v", err) + return + } + defer conn.Close() + op, payload, err := conn.ReadMessage() + if err != nil { + return + } + if op != opText { + t.Errorf("server read opcode = %d, want text", op) + return + } + _ = conn.WriteMessage(opText, payload) // echo + })) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + client, err := dialWS(context.Background(), nil, url, "") + if err != nil { + t.Fatalf("dialWS error: %v", err) + } + defer client.Close() + + if err := client.WriteMessage(opText, []byte("hello")); err != nil { + t.Fatalf("client write error: %v", err) + } + _, echo, err := client.ReadMessage() + if err != nil { + t.Fatalf("client read error: %v", err) + } + if string(echo) != "hello" { + t.Errorf("echo = %q, want hello", echo) + } +} + +// rawFrame builds an unmasked frame for tests that drive reassembly directly. +func rawFrame(fin bool, opcode byte, payload []byte) []byte { + first := opcode + if fin { + first |= 0x80 + } + out := []byte{first} + n := len(payload) + switch { + case n < 126: + out = append(out, byte(n)) + case n <= 0xFFFF: + out = append(out, 126, byte(n>>8), byte(n)) + default: + out = append(out, 127, 0, 0, 0, 0, byte(n>>24), byte(n>>16), byte(n>>8), byte(n)) + } + return append(out, payload...) +} + +func TestWSFragmentedReadAssembles(t *testing.T) { + t.Parallel() + + c1, c2 := net.Pipe() + defer c1.Close() + defer c2.Close() + + client := newWSConn(c2, nil, nil, true) + go func() { + _, _ = c1.Write(rawFrame(false, opText, []byte("hel"))) + _, _ = c1.Write(rawFrame(true, opContinuation, []byte("lo"))) + }() + + op, payload, err := client.ReadMessage() + if err != nil { + t.Fatalf("fragmented read error: %v", err) + } + if op != opText { + t.Errorf("opcode = %d, want text", op) + } + if string(payload) != "hello" { + t.Errorf("payload = %q, want hello", payload) + } +} + +func TestWSPingRepliesPong(t *testing.T) { + t.Parallel() + + c1, c2 := net.Pipe() + defer c1.Close() + defer c2.Close() + + server := newWSConn(c1, nil, nil, false) + client := newWSConn(c2, nil, nil, true) + + pong := make(chan string, 1) + go func() { + if err := server.WriteMessage(opPing, []byte("hb")); err != nil { + pong <- "err" + return + } + for { + _, op, payload, err := server.readFrame() + if err != nil { + pong <- "err" + return + } + if op == opPong { + _ = server.Close() // unblocks the client's pending read + pong <- string(payload) + return + } + } + }() + + // The client's ReadMessage processes the ping (auto-replying a pong) and + // then blocks until the server closes, which unblocks it with EOF. + if _, _, err := client.ReadMessage(); err == nil { + t.Error("client.ReadMessage returned nil error, want EOF after close") + } + + select { + case got := <-pong: + if got != "hb" { + t.Errorf("pong payload = %q, want hb", got) + } + case <-time.After(3 * time.Second): + t.Fatal("server never saw a pong") + } +} diff --git a/plugins/channels/qq_obv11/ws_client.go b/plugins/channels/qq_obv11/ws_client.go new file mode 100644 index 0000000..dbf0ad3 --- /dev/null +++ b/plugins/channels/qq_obv11/ws_client.go @@ -0,0 +1,42 @@ +package qq_obv11 + +// This file implements the client role of the OneBot v11 WebSocket channel: the +// adapter dials out to a OneBot (正向) WebSocket endpoint, authenticating with +// an access token when configured. + +import ( + "context" + "errors" + "fmt" + "net/url" + "strings" +) + +// connectClient dials the configured OneBot WebSocket endpoint and installs the +// resulting connection as the active session. +func (a *Adapter) connectClient(ctx context.Context) error { + if a.cfg.RemoteURL == "" { + return errors.New("qq-obv11: client mode requires remoteURL") + } + target := appendQueryToken(a.cfg.RemoteURL, a.cfg.AccessToken) + conn, err := dialWS(ctx, a.cfg.DialContext, target, a.cfg.AccessToken) + if err != nil { + return fmt.Errorf("qq-obv11: dial onebot ws %s: %w", a.cfg.RemoteURL, err) + } + a.setConn(conn) + a.log.Info("connected to onebot ws (client mode)", "url", a.cfg.RemoteURL) + return nil +} + +// appendQueryToken adds an access_token query parameter to a URL, preserving any +// existing query string. +func appendQueryToken(rawURL, token string) string { + if token == "" { + return rawURL + } + sep := "?" + if strings.Contains(rawURL, "?") { + sep = "&" + } + return rawURL + sep + "access_token=" + url.QueryEscape(token) +} diff --git a/plugins/channels/qq_obv11/ws_server.go b/plugins/channels/qq_obv11/ws_server.go new file mode 100644 index 0000000..0a8cbd2 --- /dev/null +++ b/plugins/channels/qq_obv11/ws_server.go @@ -0,0 +1,73 @@ +package qq_obv11 + +// This file implements the server role of the OneBot v11 WebSocket channel: +// the adapter listens on a TCP port and accepts inbound OneBot (反向) client +// connections, validates any configured access token, and upgrades each HTTP +// request to a WebSocket session. + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "strings" +) + +// connectServer starts the HTTP listener in server mode. It returns once the +// listener is bound; the session becomes live when a OneBot client connects. +func (a *Adapter) connectServer(ctx context.Context) error { + if a.cfg.ListenAddr == "" { + return errors.New("qq-obv11: server mode requires listenAddr") + } + mux := http.NewServeMux() + mux.HandleFunc("/", a.wsHandler) + + ln, err := net.Listen("tcp", a.cfg.ListenAddr) + if err != nil { + return fmt.Errorf("qq-obv11: listen %s: %w", a.cfg.ListenAddr, err) + } + + a.mu.Lock() + a.ln = ln + a.srv = &http.Server{Handler: mux} + a.mu.Unlock() + + go func() { + if serveErr := a.srv.Serve(ln); serveErr != nil && serveErr != http.ErrServerClosed { + a.log.Warn("websocket server stopped", "err", serveErr) + } + }() + a.log.Info("websocket server listening (server mode)", "addr", ln.Addr().String()) + return nil +} + +// wsHandler is the HTTP handler that upgrades one inbound OneBot client +// connection to a WebSocket session. +func (a *Adapter) wsHandler(w http.ResponseWriter, r *http.Request) { + if a.cfg.AccessToken != "" && !a.checkAuth(r) { + http.Error(w, "Forbidden: invalid access_token", http.StatusForbidden) + return + } + conn, err := upgradeWS(w, r) + if err != nil { + a.log.Warn("websocket upgrade failed", "err", err) + return + } + a.setConn(conn) + a.log.Info("onebot client connected (server mode)", "remote", r.RemoteAddr) +} + +// checkAuth validates the access token carried by an upgrade request. OneBot +// implementations pass it either as an access_token query parameter or as an +// Authorization: Bearer header. +func (a *Adapter) checkAuth(r *http.Request) bool { + if got := r.URL.Query().Get("access_token"); got != "" { + return got == a.cfg.AccessToken + } + auth := r.Header.Get("Authorization") + if strings.HasPrefix(auth, "Bearer ") { + return strings.TrimPrefix(auth, "Bearer ") == a.cfg.AccessToken + } + return false +}