diff --git a/cmd/cyrene/channels.go b/cmd/cyrene/channels.go index a2ca419..beeeb40 100644 --- a/cmd/cyrene/channels.go +++ b/cmd/cyrene/channels.go @@ -136,10 +136,16 @@ func channelAccountsFromEnv() map[string][]config.ChannelAccount { "webhook_url": os.Getenv("CHANNEL_TELEGRAM_WEBHOOK_URL"), }}) } + // qq-obv11 的四个端（裁决 1.4）：transport(ws|http) × mode(client|server)。 + // 默认 transport=ws、mode=server，与改造前的行为完全一致（保守：不改变已部署 + // 现状）；listen_addr 是 server 端的显式监听地址，留空即不开端口（AGENTS §9）。 out["qq-obv11"] = append(out["qq-obv11"], config.ChannelAccount{Config: map[string]any{ + "transport": envString("CHANNEL_QQ_OBV11_TRANSPORT", "ws"), "mode": envString("CHANNEL_QQ_OBV11_MODE", "server"), "listen_addr": os.Getenv("CHANNEL_QQ_OBV11_LISTEN_ADDR"), "remote_url": os.Getenv("CHANNEL_QQ_OBV11_REMOTE_URL"), + "api_base_url": os.Getenv("CHANNEL_QQ_OBV11_API_BASE_URL"), + "report_path": os.Getenv("CHANNEL_QQ_OBV11_REPORT_PATH"), "access_token": os.Getenv("CHANNEL_QQ_OBV11_ACCESS_TOKEN"), "self_id": os.Getenv("CHANNEL_QQ_OBV11_SELF_ID"), }}) diff --git a/cmd/cyrene/channels_obv11_transport_test.go b/cmd/cyrene/channels_obv11_transport_test.go new file mode 100644 index 0000000..9073df4 --- /dev/null +++ b/cmd/cyrene/channels_obv11_transport_test.go @@ -0,0 +1,61 @@ +package main + +// 覆盖裁决 1.4 在组合根侧的落点：qq-obv11 的四个端所需配置项由 .env 兜底进适配器配置， +// 且默认值保持现状（transport=ws、mode=server），不改变已部署行为。 + +import "testing" + +// TestChannelAccountsFromEnv_QQObv11FourEnds 验证 env 兜底把 transport / mode / +// listen_addr / remote_url / api_base_url / report_path 带进账号配置，并且默认口径不变。 +func TestChannelAccountsFromEnv_QQObv11FourEnds(t *testing.T) { + for _, k := range []string{ + "CHANNEL_QQ_OBV11_TRANSPORT", "CHANNEL_QQ_OBV11_MODE", "CHANNEL_QQ_OBV11_LISTEN_ADDR", + "CHANNEL_QQ_OBV11_REMOTE_URL", "CHANNEL_QQ_OBV11_API_BASE_URL", "CHANNEL_QQ_OBV11_REPORT_PATH", + } { + t.Setenv(k, "") + } + acct := qqObv11EnvAccount(t) + if acct["transport"] != "ws" || acct["mode"] != "server" { + t.Fatalf("defaults = transport:%v mode:%v, want ws/server", acct["transport"], acct["mode"]) + } + + t.Setenv("CHANNEL_QQ_OBV11_TRANSPORT", "http") + t.Setenv("CHANNEL_QQ_OBV11_MODE", "client") + t.Setenv("CHANNEL_QQ_OBV11_API_BASE_URL", "http://127.0.0.1:3000") + t.Setenv("CHANNEL_QQ_OBV11_REPORT_PATH", "/onebot/report") + + list := channelAccountsFromEnv()["qq-obv11"] + if len(list) != 1 { + t.Fatalf("qq-obv11 env accounts = %d, want 1", len(list)) + } + acct = list[0].Config + if acct["transport"] != "http" || acct["mode"] != "client" { + t.Errorf("transport/mode = %v/%v, want http/client", acct["transport"], acct["mode"]) + } + if acct["api_base_url"] != "http://127.0.0.1:3000" || acct["report_path"] != "/onebot/report" { + t.Errorf("api_base_url/report_path = %v/%v, want the env values", acct["api_base_url"], acct["report_path"]) + } + + // 端到端一次：env 配置经 DecodeConfig 进适配器后，四个端的开关仍然在。 + built, err := buildAccountAdapter("qq-obv11", list[0], testLog()) + if err != nil { + t.Fatalf("buildAccountAdapter error: %v", err) + } + cfg := built.Config() + if cfg["transport"] != "http" || cfg["mode"] != "client" { + t.Errorf("adapter transport/mode = %v/%v, want http/client", cfg["transport"], cfg["mode"]) + } + if cfg["apiBaseURL"] != "http://127.0.0.1:3000" || cfg["reportPath"] != "/onebot/report" { + t.Errorf("adapter apiBaseURL/reportPath = %v/%v, want the env values", cfg["apiBaseURL"], cfg["reportPath"]) + } +} + +// qqObv11EnvAccount 取出 env 兜底里合成的 qq-obv11 账号配置。 +func qqObv11EnvAccount(t *testing.T) map[string]any { + t.Helper() + list := channelAccountsFromEnv()["qq-obv11"] + if len(list) != 1 { + t.Fatalf("qq-obv11 env accounts = %d, want 1", len(list)) + } + return list[0].Config +} diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go index 78bfcc2..140cba5 100644 --- a/plugins/channels/qq_obv11/adapter.go +++ b/plugins/channels/qq_obv11/adapter.go @@ -1,14 +1,15 @@ -// 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. +// Package qq_obv11 实现内置的 QQ OneBot v11 渠道适配器。它按 `transport`(ws|http) +// × `mode`(client|server) 的**四个端**（裁决 1.4）与 OneBot 实现通信： // -// 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. +// - ws + client：Cyrene 主动拨号 OneBot 的 WS 端点，收发共用一个长连接； +// - ws + server：OneBot 反连 Cyrene 的 `listen_addr`，收发共用一个长连接； +// - http + client：Cyrene 作为 HTTP 客户端调用 OneBot 的 HTTP API（出站）； +// - http + server：Cyrene 监听 `listen_addr`，接收 OneBot POST 上来的事件上报（入站）。 +// +// 原生 CQ 码 / 数组段消息统一归一成 adapter.UnifiedMessage，回复走同一个端并受限频 +// 约束。设计依据是 DevDocs/design/08-渠道接入.md §1/§3/§4；本包只依赖 +// internal/channel/adapter、internal/logger 与标准库，WS 帧编解码自带（模块不引入 +// 第三方 websocket 依赖）。 package qq_obv11 import ( @@ -18,6 +19,7 @@ import ( "log/slog" "net" "net/http" + "strings" "sync" "time" @@ -34,18 +36,39 @@ const ( defaultMaxProactivePerWindow = 5 // defaultProactiveWindow is the sliding window for the send rate limit. defaultProactiveWindow = 10 * time.Second + + // transportWS / transportHTTP 是 OneBot v11 的两种通信介质（裁决 1.4）。 + transportWS = "ws" + transportHTTP = "http" + // modeServer / modeClient 是每种介质下的两个角色：谁向谁发数据（裁决 1.4）。 + modeServer = "server" + modeClient = "client" + // defaultReportPath 是 http+server 接收上报的默认路径。 + defaultReportPath = "/" ) // 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". + // Transport 选择通信介质："ws"（默认，保持既有部署行为）或 "http"（裁决 1.4）。 + // 空值与未知值一律回落 "ws"，绝不静默改走 HTTP。 + Transport string `json:"transport"` + // Mode 选择该介质下的角色（谁向谁发数据），"server"（默认）或 "client"： + // ws —— server 等 OneBot 反连本适配器，client 由本适配器拨号 OneBot； + // http —— server 收 OneBot 的 HTTP 上报，client 调用 OneBot 的 HTTP API。 Mode string `json:"mode"` // ListenAddr is the TCP address to bind in server mode (host:port). + // **必须由配置显式给出**：OneBot 反连端口是 AGENTS §9 登记的唯一豁免， + // 空值让 Connect 直接失败，绝不默认开监听。 ListenAddr string `json:"listen_addr"` - // RemoteURL is the ws:// endpoint to dial in client mode. + // RemoteURL 是 ws+client 拨号的 ws://（或 wss://）端点。 RemoteURL string `json:"remote_url"` - // AccessToken is the optional OneBot access token for both modes. + // APIBaseURL 是 OneBot HTTP API 基址（如 http://127.0.0.1:3000）：http+client + // 必填；http+server 可选，给了才有出站（回消息）能力。必须显式配置——不从 + // RemoteURL 推推导（v1「剥掉最后一段路径」的推导是有损的，见 design/08 对照）。 + APIBaseURL string `json:"api_base_url"` + // ReportPath 是 http+server 接收上报的路径。空值默认 "/"。 + ReportPath string `json:"report_path"` + // AccessToken is the optional OneBot access token for all four ends. AccessToken string `json:"access_token"` // SelfID is the bot's own QQ number when known. SelfID string `json:"self_id"` @@ -66,7 +89,8 @@ type Config struct { // // 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). +// limited). HTTP 传输没有长连接：入站由上报 handler 直接派发，出站走 httpAPI 的 +// 请求-响应路径。 type Adapter struct { cfg Config log *slog.Logger @@ -75,6 +99,7 @@ type Adapter struct { conn *wsConn ln net.Listener srv *http.Server + httpAPI *obHTTPClient started bool connected bool cancel context.CancelFunc @@ -97,11 +122,12 @@ type Adapter struct { // Compile-time conformance with the channel contract. var _ adapter.PlatformAdapter = (*Adapter)(nil) -// NewAdapter builds a qq_obv11 Adapter from cfg, applying safe defaults. +// NewAdapter builds a qq_obv11 Adapter from cfg, applying safe defaults. 默认值口径 +// （裁决 1.4）：transport=ws、mode=server，即**不给出任何新配置时行为与改造前一致**。 func NewAdapter(cfg Config) *Adapter { - if cfg.Mode == "" { - cfg.Mode = "server" - } + cfg.Transport = normalizeTransport(cfg.Transport) + cfg.Mode = normalizeMode(cfg.Mode) + cfg.ReportPath = normalizeReportPath(cfg.ReportPath) if cfg.MaxProactivePerWindow <= 0 { cfg.MaxProactivePerWindow = defaultMaxProactivePerWindow } @@ -127,9 +153,12 @@ 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{ + "transport": a.cfg.Transport, "mode": a.cfg.Mode, "listenAddr": a.cfg.ListenAddr, - "remoteURL": a.cfg.RemoteURL, + "remoteURL": configRemoteURL(a.cfg), + "apiBaseURL": a.cfg.APIBaseURL, + "reportPath": a.cfg.ReportPath, "accessToken": a.cfg.AccessToken, "selfID": a.cfg.SelfID, "maxProactivePerWindow": a.cfg.MaxProactivePerWindow, @@ -138,7 +167,7 @@ func (a *Adapter) Config() map[string]any { } } -// Connect establishes the session according to the configured mode. It is +// Connect establishes the session for the configured transport × mode. It is // idempotent. func (a *Adapter) Connect(ctx context.Context) error { a.mu.Lock() @@ -149,21 +178,33 @@ func (a *Adapter) Connect(ctx context.Context) error { a.mu.Unlock() var err error - if a.cfg.Mode == "client" { + switch { + case a.cfg.Transport == transportHTTP: + err = a.connectHTTP(ctx) + case a.cfg.Mode == modeClient: err = a.connectClient(ctx) - } else { + default: err = a.connectServer(ctx) } if err != nil { return fmt.Errorf("qq-obv11: connect: %w", err) } + httpTransport := a.cfg.Transport == transportHTTP a.mu.Lock() a.started = true + if httpTransport { + // HTTP 没有持久连接可观察：监听已绑定 / 出站客户端已装配即视为可服务， + // 真实可达性交给 HealthCheck 的 get_login_info 探活。 + a.connected = true + } a.lifeCtx, a.cancel = context.WithCancel(context.Background()) a.mu.Unlock() - go a.readLoop(a.lifeCtx) + // WS 两端共用一个长连接读循环；HTTP 的入站由上报 handler 直接进 handlePayload。 + if !httpTransport { + go a.readLoop(a.lifeCtx) + } return nil } @@ -178,10 +219,12 @@ func (a *Adapter) Disconnect(ctx context.Context) error { srv := a.srv ln := a.ln conn := a.conn + api := a.httpAPI a.conn = nil a.connected = false a.ln = nil a.srv = nil + a.httpAPI = nil a.mu.Unlock() if !started && cancel == nil { @@ -193,6 +236,10 @@ func (a *Adapter) Disconnect(ctx context.Context) error { if conn != nil { _ = conn.Close() } + // 出站 HTTP 客户端没有长连接，只需释放连接池里的 keep-alive 连接。 + if api != nil { + api.Close() + } if srv != nil { shutdownCtx, c := context.WithTimeout(context.Background(), 3*time.Second) defer c() @@ -206,17 +253,107 @@ func (a *Adapter) Disconnect(ctx context.Context) error { return nil } -// IsConnected reports whether a live OneBot session exists. +// IsConnected reports whether a live OneBot session exists. HTTP 传输没有长连接， +// 它反映的是「适配器已就绪」（监听已绑定 / 出站客户端已装配）。 func (a *Adapter) IsConnected() bool { a.mu.RLock() defer a.mu.RUnlock() return a.connected } -// HealthCheck verifies the session is usable. +// HealthCheck verifies the session is usable. HTTP 传输额外做一次 get_login_info +// 探活：没有长连接可观察，只有真实请求才能说明 OneBot 实现可达。 func (a *Adapter) HealthCheck(ctx context.Context) error { if !a.IsConnected() { return fmt.Errorf("qq-obv11: health check: %w", adapter.ErrNotConnected) } + if api := a.currentAPI(); api != nil { + return api.probe(ctx) + } return nil } + +// currentAPI 返回已装配的 OneBot HTTP API 出站客户端；未装配时（ws 两端，或 +// http+server 只收上报、没配 api_base_url）为 nil。 +func (a *Adapter) currentAPI() *obHTTPClient { + a.mu.RLock() + defer a.mu.RUnlock() + return a.httpAPI +} + +// connectHTTP 装配 HTTP 传输（裁决 1.4）。OneBot v11 的 HTTP 通信天然分两半：动作 +// （Cyrene→OneBot，client 端，走 OneBot 的 HTTP API）与上报（OneBot→Cyrene，server +// 端，走本适配器的监听端口）。`mode` 决定哪一半是**必需**的——server 必须有显式 +// listen_addr（AGENTS §9 豁免口径：绝不默认开端口），client 必须有 api_base_url； +// 另一半只要对应字段也给齐就一并启用，于是单个适配器即可完成「收上报 + 回消息」闭环。 +func (a *Adapter) connectHTTP(ctx context.Context) error { + // 先做纯配置校验并构造出站客户端（无副作用），再绑监听端口：任何一端配置有误都 + // 不能留下半开的端口——桥接器对 Connect 失败不会回头调用 Disconnect。 + var api *obHTTPClient + if a.cfg.APIBaseURL != "" { + client, err := newOBHTTPClient(a.cfg.APIBaseURL, a.cfg.AccessToken) + if err != nil { + return err + } + api = client + } else if a.cfg.Mode == modeClient { + return errors.New("qq-obv11: http client mode requires apiBaseUrl") + } + + if a.cfg.ListenAddr != "" { + if err := a.connectHTTPServer(ctx); err != nil { + return err + } + } else if a.cfg.Mode == modeServer { + return errors.New("qq-obv11: http server mode requires listenAddr") + } + + if api != nil { + a.mu.Lock() + a.httpAPI = api + a.mu.Unlock() + } + return nil +} + +// configRemoteURL 是 Config() 上报的「远端地址」：channelbridge 的凭据门按 +// remoteURL（client 模式）/ listenAddr（server 模式）判断该渠道是否值得连接 +// （internal/services/channelbridge/connect.go:143-147）。HTTP client 端的远端就是 +// OneBot 的 HTTP API 基址，因此 RemoteURL 为空时回退到 APIBaseURL——否则 +// transport=http + mode=client 会被误判成「无凭据」而永不连接。 +func configRemoteURL(cfg Config) string { + if cfg.RemoteURL != "" { + return cfg.RemoteURL + } + return cfg.APIBaseURL +} + +// normalizeTransport 归一化通信介质：空值保持现状 ws，未知值同样回落 ws， +// 不静默按 http 处理。 +func normalizeTransport(raw string) string { + if strings.EqualFold(strings.TrimSpace(raw), transportHTTP) { + return transportHTTP + } + return transportWS +} + +// normalizeMode 归一化角色：空值与未知值回落 server（与既有默认一致）。 +func normalizeMode(raw string) string { + if strings.EqualFold(strings.TrimSpace(raw), modeClient) { + return modeClient + } + return modeServer +} + +// normalizeReportPath 归一化上报路径：空值默认 "/"，并保证以 "/" 开头。 +func normalizeReportPath(raw string) string { + p := strings.TrimSpace(raw) + switch { + case p == "": + return defaultReportPath + case !strings.HasPrefix(p, "/"): + return "/" + p + default: + return p + } +} diff --git a/plugins/channels/qq_obv11/http_client.go b/plugins/channels/qq_obv11/http_client.go new file mode 100644 index 0000000..3ef4cdf --- /dev/null +++ b/plugins/channels/qq_obv11/http_client.go @@ -0,0 +1,172 @@ +package qq_obv11 + +// 本文件实现 OneBot v11 的 HTTP **client 端**（裁决 1.4 的四个端之一）：Cyrene 作为 +// HTTP 客户端调用 OneBot 实现（NapCat / go-cqhttp / LLOneBot 等）的 HTTP API，把 action +// 请求 POST 到 `{api_base_url}/{action}`，并*等待*带 echo 的响应，按 retcode/status 判成败。 +// +// 对 v1 老代码的批判性取舍（AGENTS §9.5）：v1 的发送是「写完就报成功」不等待响应 +// （qq/adapter.go:612），出站 token 还拼在 URL query 里（会进日志与代理）。这里改为 +// ①请求-响应 + 超时、②出站 token 一律走 `Authorization: Bearer` 头、③错误与日志都不 +// 含 token；v1 那种「从 ws:// 地址猜 http:// 基址」的推导也不再沿用——HTTP 基址必须由 +// `api_base_url` 显式给出，避免路径剥离猜错。 + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +const ( + // defaultHTTPTimeout 是单次 OneBot HTTP API 调用的超时上限。HTTP 传输没有连接可 + // 观察，一次调用挂死必须自己收敛，不能拖住消息流水线。 + defaultHTTPTimeout = 15 * time.Second + // maxHTTPRespBytes 限制单次响应体大小，防御异常对端把进程内存吃满。 + maxHTTPRespBytes = 4 << 20 +) + +// obAPIResponse 是 OneBot v11 HTTP API 的响应信封。 +type obAPIResponse struct { + Status string `json:"status"` + RetCode int `json:"retcode"` + Data json.RawMessage `json:"data"` + Msg string `json:"msg"` + Wording string `json:"wording"` + Echo string `json:"echo"` +} + +// obActionReq 是通用的 OneBot v11 action 请求体（send_msg 之外的轻量 action 复用， +// 例如探活用的 get_login_info）。 +type obActionReq struct { + Action string `json:"action"` + Params any `json:"params"` + Echo string `json:"echo,omitempty"` +} + +// obHTTPClient 是 OneBot HTTP API 的出站客户端。它无可变状态，http.Client 自身可并发 +// 复用，因此多个发送协程可以同时调用。 +type obHTTPClient struct { + baseURL string + token string + http *http.Client +} + +// newOBHTTPClient 校验并构造 HTTP 客户端。baseURL 必须是 http/https 且带 host，并且 +// **不得内嵌账号密码**（userinfo）——凭据只走 Authorization 头，避免它出现在 URL 与 +// 错误信息里。 +func newOBHTTPClient(baseURL, token string) (*obHTTPClient, error) { + trimmed := strings.TrimSpace(baseURL) + if trimmed == "" { + return nil, errors.New("qq-obv11: empty api base url") + } + u, err := url.Parse(trimmed) + if err != nil { + return nil, fmt.Errorf("qq-obv11: parse api base url: %w", err) + } + if u.Scheme != "http" && u.Scheme != "https" { + return nil, fmt.Errorf("qq-obv11: api base url scheme %q (want http/https)", u.Scheme) + } + if u.Host == "" { + return nil, errors.New("qq-obv11: api base url has no host") + } + if u.User != nil { + return nil, errors.New("qq-obv11: api base url must not embed credentials") + } + return &obHTTPClient{ + baseURL: strings.TrimRight(trimmed, "/"), + token: token, + http: &http.Client{Timeout: defaultHTTPTimeout}, + }, nil +} + +// Close 释放连接池中的空闲 keep-alive 连接。HTTP 没有持久会话，重复调用安全。 +func (c *obHTTPClient) Close() { + if c == nil { + return + } + c.http.CloseIdleConnections() +} + +// send 把一条已构造好的 action 请求（payload 已含 action/params/echo）POST 给 OneBot +// 并等待结果。它替代 WS 端的「写一帧就返回」：HTTP 有明确响应，成败以 retcode 为准。 +func (c *obHTTPClient) send(ctx context.Context, payload []byte) error { + return c.post(ctx, payload) +} + +// probe 用最轻量的标准 action `get_login_info` 探活 OneBot 实现，供 HealthCheck 使用。 +// 它不打印 token，也不记录返回的机器人账号信息。 +func (c *obHTTPClient) probe(ctx context.Context) error { + payload, err := buildActionReq("get_login_info", map[string]any{}, "") + if err != nil { + return err + } + return c.post(ctx, payload) +} + +// post 发送一条 action 请求并解析响应。任何失败都返回带上下文（action、retcode）的 +// wrapped error；错误信息与日志都不含 token。 +func (c *obHTTPClient) post(ctx context.Context, payload []byte) error { + if c == nil { + return errors.New("qq-obv11: http api client not configured") + } + action := actionOf(payload) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/"+action, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("qq-obv11: build http %s request: %w", action, err) + } + req.Header.Set("Content-Type", "application/json") + if c.token != "" { + // 出站 token 只走 Authorization 头：不进 URL query、不进日志。 + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.http.Do(req) + if err != nil { + return fmt.Errorf("qq-obv11: http %s request: %w", action, err) + } + defer func() { _ = resp.Body.Close() }() + + body, err := io.ReadAll(io.LimitReader(resp.Body, maxHTTPRespBytes)) + if err != nil { + return fmt.Errorf("qq-obv11: read http %s response: %w", action, err) + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("qq-obv11: http %s status %d", action, resp.StatusCode) + } + var env obAPIResponse + if err := json.Unmarshal(body, &env); err != nil { + return fmt.Errorf("qq-obv11: decode http %s response: %w", action, err) + } + if env.RetCode != 0 || (env.Status != "" && !strings.EqualFold(env.Status, "ok")) { + return fmt.Errorf("qq-obv11: http %s failed: retcode=%d status=%q msg=%q", + action, env.RetCode, env.Status, env.Msg) + } + return nil +} + +// actionOf 从已编码的 action 请求里取出 action 名，仅用于日志与错误上下文。 +func actionOf(payload []byte) string { + var head struct { + Action string `json:"action"` + } + if err := json.Unmarshal(payload, &head); err != nil || head.Action == "" { + return "action" + } + return head.Action +} + +// buildActionReq 编码一条通用 OneBot action 请求。 +func buildActionReq(action string, params any, echo string) ([]byte, error) { + b, err := json.Marshal(obActionReq{Action: action, Params: params, Echo: echo}) + if err != nil { + return nil, fmt.Errorf("qq-obv11: encode %s request: %w", action, err) + } + return b, nil +} diff --git a/plugins/channels/qq_obv11/http_server.go b/plugins/channels/qq_obv11/http_server.go new file mode 100644 index 0000000..830e758 --- /dev/null +++ b/plugins/channels/qq_obv11/http_server.go @@ -0,0 +1,87 @@ +package qq_obv11 + +// 本文件实现 OneBot v11 的 HTTP **server 端**（裁决 1.4 的四个端之一）：Cyrene 监听一个 +// HTTP 端口，接收 OneBot 实现主动 POST 上来的事件上报（OneBot→Cyrene 方向）。这是四端里 +// 唯一新增的监听端口，受 AGENTS §9 的登记豁免约束：**必须**由配置 `listen_addr` 显式给出， +// 空值直接让 Connect 失败、绝不默认开（design/13 §0）。鉴权 fail-closed：配置了 +// access_token 时，Authorization: Bearer 或 ?access_token= 必须匹配，否则 403 直接拒绝。 + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" +) + +// maxReportBodyBytes 限制单条上报体大小：事件 JSON 很小，超过上限的一律拒收，避免 +// 恶意对端把上报路径当内存放大器（与 WS 帧上限同一口径）。 +const maxReportBodyBytes = 10 << 20 // 10 MiB + +// connectHTTPServer 在 listen_addr 上启动 OneBot HTTP 上报接收端。返回时监听已绑定； +// 真正的事件在 OneBot POST 上来时经 handlePayload 进入消息流。 +func (a *Adapter) connectHTTPServer(_ context.Context) error { + if a.cfg.ListenAddr == "" { + return errors.New("qq-obv11: http server mode requires listenAddr") + } + mux := http.NewServeMux() + mux.HandleFunc(a.cfg.ReportPath, a.httpReportHandler) + + ln, err := net.Listen("tcp", a.cfg.ListenAddr) + if err != nil { + return fmt.Errorf("qq-obv11: listen %s: %w", a.cfg.ListenAddr, err) + } + + srv := &http.Server{Handler: mux} + a.mu.Lock() + a.ln = ln + a.srv = srv + a.mu.Unlock() + + go func() { + if serveErr := srv.Serve(ln); serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { + a.log.Warn("onebot http report server stopped", "err", serveErr) + } + }() + a.log.Info("onebot http report server listening (http server mode)", "addr", ln.Addr().String()) + return nil +} + +// httpReportHandler 接收 OneBot v11 的 HTTP POST 上报。处理是同步的——与 WS 读循环 +// 一致，保证同一账号的事件顺序；桥接器的入站只是入队，所以这里能立刻回 204，OneBot +// 侧不会因超时而重发。 +func (a *Adapter) httpReportHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if a.cfg.AccessToken != "" && !a.checkAuth(r) { + http.Error(w, "Forbidden: invalid access_token", http.StatusForbidden) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, maxReportBodyBytes+1)) + if err != nil { + http.Error(w, "cannot read report body", http.StatusBadRequest) + return + } + if len(body) > maxReportBodyBytes { + http.Error(w, "report body too large", http.StatusRequestEntityTooLarge) + return + } + // 必须用适配器生命周期 context：r.Context() 在响应写完时就被取消，会连带掐断 + // 入站流水线（桥接器的处理是异步派发的）。 + a.handlePayload(a.reportContext(), body) + w.WriteHeader(http.StatusNoContent) +} + +// reportContext 返回适配器生命周期 context（Connect 时建立）；适配器尚未就绪时退回 +// 后台 context，保证事件仍能派发而不是被静默丢弃。 +func (a *Adapter) reportContext() context.Context { + a.mu.RLock() + defer a.mu.RUnlock() + if a.lifeCtx != nil { + return a.lifeCtx + } + return context.Background() +} diff --git a/plugins/channels/qq_obv11/http_transport_test.go b/plugins/channels/qq_obv11/http_transport_test.go new file mode 100644 index 0000000..8333362 --- /dev/null +++ b/plugins/channels/qq_obv11/http_transport_test.go @@ -0,0 +1,446 @@ +package qq_obv11 + +// 本文件覆盖裁决 1.4 的**四个端**（http/ws × client/server）里的 HTTP 两端，以及 +// transport/mode 配置项的默认值与归一化。WS 两端已有既有用例（TestAdapterServerModeRoundTrip / +// TestAdapterClientModeReceiveAndSend），这里用本地 httptest 假 OneBot 实现驱动，绝不连真实 QQ。 + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// groupEventJSON 是一条标准的 OneBot v11 群消息上报。 +const groupEventJSON = `{"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"}}` + +// recordedAPIRequest 是假 OneBot HTTP API 收到的一条 action 请求。 +type recordedAPIRequest struct { + action string + auth string + body []byte +} + +// fakeOneBotAPI 是本地假的 OneBot HTTP API：记录收到的 action 请求，并按 retCode / +// status 回一个标准响应信封，用来验证 HTTP client 端的「请求-响应 + 判成败」语义。 +type fakeOneBotAPI struct { + mu sync.Mutex + requests []recordedAPIRequest + retCode int + status string +} + +func (f *fakeOneBotAPI) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.requests = append(f.requests, recordedAPIRequest{ + action: strings.TrimPrefix(r.URL.Path, "/"), + auth: r.Header.Get("Authorization"), + body: body, + }) + retCode, status := f.retCode, f.status + f.mu.Unlock() + if status == "" { + status = "ok" + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "status": status, + "retcode": retCode, + "data": map[string]any{}, + "msg": "fake", + }) + }) +} + +func (f *fakeOneBotAPI) snapshot() []recordedAPIRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]recordedAPIRequest(nil), f.requests...) +} + +// reportResult 是一次上报 POST 的结果。 +type reportResult struct { + status int + body string +} + +// postReport 模拟 OneBot 实现向 Cyrene 的上报端 POST 一条事件。 +func postReport(t *testing.T, url, token, body string) reportResult { + t.Helper() + req, err := http.NewRequest(http.MethodPost, url, strings.NewReader(body)) + if err != nil { + t.Fatalf("build report request: %v", err) + } + req.Header.Set("Content-Type", "application/json") + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("post report to %s: %v", url, err) + } + defer func() { _ = resp.Body.Close() }() + raw, _ := io.ReadAll(resp.Body) + return reportResult{status: resp.StatusCode, body: string(raw)} +} + +// TestHTTPClientModeSendsViaAPI 覆盖 **http + client**：Cyrene→OneBot（调 OneBot 的 +// HTTP API）。断言 send_msg 真的发出去、出站 token 只在 Authorization 头里、该模式 +// 不开任何监听端口，并且探活走的是 get_login_info。 +func TestHTTPClientModeSendsViaAPI(t *testing.T) { + t.Parallel() + + api := &fakeOneBotAPI{} + srv := httptest.NewServer(api.handler()) + defer srv.Close() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeClient, + APIBaseURL: srv.URL, + AccessToken: "secret-token", + }) + 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("IsConnected() = false after Connect, want true") + } + if addr := a.Addr(); addr != "" { + t.Errorf("http client mode must not listen, Addr() = %q", addr) + } + if err := a.HealthCheck(context.Background()); err != nil { + t.Errorf("HealthCheck() error = %v, want nil", err) + } + + 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) + } + + reqs := api.snapshot() + if len(reqs) != 2 { + t.Fatalf("api requests = %+v, want health probe + send_msg", reqs) + } + if reqs[0].action != "get_login_info" || reqs[1].action != "send_msg" { + t.Errorf("actions = %q,%q want get_login_info,send_msg", reqs[0].action, reqs[1].action) + } + for _, r := range reqs { + if r.auth != "Bearer secret-token" { + t.Errorf("%s Authorization = %q, want Bearer header", r.action, r.auth) + } + if strings.Contains(string(r.body), "access_token") { + t.Errorf("%s body carries access_token: %s", r.action, r.body) + } + } + assertSendMsg(t, reqs[1].body) +} + +// TestHTTPServerModeReceivesReport 覆盖 **http + server**：OneBot→Cyrene（OneBot 把 +// 事件上报 POST 到 Cyrene 的监听端口）。断言事件被归一化派发、成功回 204；只收不发 +// 的端没有出站通道，SendMessage 必须明确报未连接而不是静默成功。 +func TestHTTPServerModeReceivesReport(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + AccessToken: "secret-token", + ReportPath: "/onebot/report", + }) + got := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(_ context.Context, m adapter.UnifiedMessage) { got <- m }) + + 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, want the report listener") + } + + if res := postReport(t, "http://"+addr+"/onebot/report", "secret-token", groupEventJSON); res.status != http.StatusNoContent { + t.Fatalf("report status = %d (body %q), want 204", res.status, res.body) + } + + select { + case m := <-got: + if m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" || m.Content != "hello" { + t.Errorf("report message = %+v, want group/123/hello", m) + } + case <-time.After(3 * time.Second): + t.Fatal("http report never reached the message handler") + } + + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}}, + }) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Errorf("SendMessage error = %v, want ErrNotConnected (no api_base_url)", err) + } +} + +// TestHTTPServerModeRejectsBadToken 验证上报端 fail-closed：配置了 access_token 时， +// 缺失或错误的 token 一律 403，事件绝不进消息流。 +func TestHTTPServerModeRejectsBadToken(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + AccessToken: "secret-token", + }) + // 缓冲足够大：即便鉴权门失效、两条上报都进来，handler 也不会阻塞在发送上。 + got := make(chan adapter.UnifiedMessage, 4) + a.SetMessageHandler(func(_ context.Context, m adapter.UnifiedMessage) { got <- m }) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + url := "http://" + a.Addr() + "/" + for _, token := range []string{"", "wrong-token"} { + if res := postReport(t, url, token, groupEventJSON); res.status != http.StatusForbidden { + t.Errorf("token %q status = %d, want 403", token, res.status) + } + } + select { + case m := <-got: + t.Fatalf("rejected report still reached the handler: %+v", m) + case <-time.After(150 * time.Millisecond): + } +} + +// TestHTTPBothEndsFullLoop 覆盖 HTTP 双端闭环：同一个适配器既收 OneBot 上报，又能经 +// OneBot HTTP API 回消息——OneBot v11 的 HTTP 通信本来就是这两半合起来才闭环。 +func TestHTTPBothEndsFullLoop(t *testing.T) { + t.Parallel() + + api := &fakeOneBotAPI{} + srv := httptest.NewServer(api.handler()) + defer srv.Close() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + APIBaseURL: srv.URL, + }) + got := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(_ context.Context, m adapter.UnifiedMessage) { got <- m }) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + if res := postReport(t, "http://"+a.Addr()+"/", "", groupEventJSON); res.status != http.StatusNoContent { + t.Fatalf("report status = %d, want 204", res.status) + } + select { + case <-got: + case <-time.After(3 * time.Second): + t.Fatal("report never reached the message handler") + } + + 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) + } + reqs := api.snapshot() + if len(reqs) != 1 || reqs[0].action != "send_msg" { + t.Fatalf("api requests = %+v, want exactly one send_msg", reqs) + } + assertSendMsg(t, reqs[0].body) +} + +// TestHTTPClientSurfacesAPIFailure 验证 HTTP client 端等响应并判成败：OneBot 回 +// retcode!=0 时发送必须报错。v1「写完就报成功」的实现在这里会静默失败，这正是本次 +// 改造要修的反例。 +func TestHTTPClientSurfacesAPIFailure(t *testing.T) { + t.Parallel() + + api := &fakeOneBotAPI{retCode: 100, status: "failed"} + srv := httptest.NewServer(api.handler()) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelDirect, ChannelID: "42", Content: "hi"}}, + }) + if err == nil { + t.Fatal("SendMessage returned nil, want the OneBot retcode failure surfaced") + } + if !strings.Contains(err.Error(), "retcode=100") { + t.Errorf("error = %v, want it to name retcode=100", err) + } +} + +// TestHTTPModeRequiresItsOwnEndpoint 验证 mode 决定哪一端必需：配置不全时 Connect +// 失败，且**绝不留半开的监听端口**（桥接器对 Connect 失败不会回头调用 Disconnect）。 +func TestHTTPModeRequiresItsOwnEndpoint(t *testing.T) { + t.Parallel() + + // http+client 缺 api_base_url。 + client := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient}) + if err := client.Connect(context.Background()); err == nil { + t.Fatal("http client mode without apiBaseUrl: Connect() = nil, want error") + } + _ = client.Disconnect(context.Background()) + + // http+server 缺 listen_addr：默认绝不监听。 + server := NewAdapter(Config{Transport: transportHTTP}) + if err := server.Connect(context.Background()); err == nil { + t.Fatal("http server mode without listenAddr: Connect() = nil, want error") + } + if addr := server.Addr(); addr != "" { + t.Errorf("failed Connect must not leave a listener, Addr() = %q", addr) + } + _ = server.Disconnect(context.Background()) + + // 出站基址非法时先失败、后绑端口：端口不能漏。 + leak := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + APIBaseURL: "ws://not-http.example:3000", + }) + if err := leak.Connect(context.Background()); err == nil { + t.Fatal("invalid apiBaseUrl: Connect() = nil, want error") + } + if addr := leak.Addr(); addr != "" { + t.Errorf("invalid apiBaseUrl leaked a listener at %q", addr) + } + _ = leak.Disconnect(context.Background()) + + // server 端给齐 listen_addr、不给 api_base_url 时仍可启动（只收入站）。 + recvOnly := NewAdapter(Config{Transport: transportHTTP, Mode: modeServer, ListenAddr: "127.0.0.1:0"}) + if err := recvOnly.Connect(context.Background()); err != nil { + t.Fatalf("http server mode with listenAddr: Connect error = %v", err) + } + if recvOnly.Addr() == "" { + t.Error("http server mode: Addr() empty, want the report listener") + } + _ = recvOnly.Disconnect(context.Background()) +} + +// TestFourEndsConfigDefaultsAndNormalization 验证四端配置项的默认值口径（裁决 1.4）： +// 不给任何新配置时 transport=ws、mode=server，与改造前完全一致；未知值也不静默改道。 +func TestFourEndsConfigDefaultsAndNormalization(t *testing.T) { + t.Parallel() + + base := NewAdapter(Config{}).Config() + if base["transport"] != transportWS || base["mode"] != modeServer { + t.Errorf("defaults = transport:%v mode:%v, want ws/server", base["transport"], base["mode"]) + } + if base["reportPath"] != defaultReportPath { + t.Errorf("default reportPath = %v, want %q", base["reportPath"], defaultReportPath) + } + + cases := []struct{ in, want string }{ + {"", transportWS}, + {"WS", transportWS}, + {" http ", transportHTTP}, + {"tcp", transportWS}, + } + for _, c := range cases { + if got := NewAdapter(Config{Transport: c.in}).Config()["transport"]; got != c.want { + t.Errorf("transport %q → %v, want %q", c.in, got, c.want) + } + } + + // http+client 的 Config() 必须把 api_base_url 作为远端地址上报，否则 channelbridge + // 的凭据门（connect.go:143-147 按 remoteURL 判 client 模式）会把该账号当「无凭据」 + // 跳过，HTTP client 端在生产组合根里永远不会连接。 + cfg := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeClient, + APIBaseURL: "http://127.0.0.1:3000", + }).Config() + if cfg["remoteURL"] != "http://127.0.0.1:3000" { + t.Errorf("http client remoteURL = %v, want the api base url", cfg["remoteURL"]) + } +} + +// TestNewOBHTTPClientValidatesBaseURL 验证出站基址显式校验：必须 http/https 且带 host， +// 不得内嵌凭据（凭据只走 Authorization 头，避免进 URL / 日志 / 错误信息）。 +func TestNewOBHTTPClientValidatesBaseURL(t *testing.T) { + t.Parallel() + + for _, bad := range []string{"", "ws://host:6700", "ftp://host", "http://", "http://user:pass@host:3000"} { + if _, err := newOBHTTPClient(bad, "tok"); err == nil { + t.Errorf("newOBHTTPClient(%q) = nil error, want rejection", bad) + } + } + + c, err := newOBHTTPClient("http://127.0.0.1:3000/", "tok") + if err != nil { + t.Fatalf("newOBHTTPClient valid base error: %v", err) + } + if c.baseURL != "http://127.0.0.1:3000" { + t.Errorf("baseURL = %q, want the trailing slash trimmed", c.baseURL) + } +} + +// TestHTTPReportPathAndMethodGuards 验证上报路径归一化、非 POST 拒绝，以及未声明的 +// 路径不会被上报端受理。 +func TestHTTPReportPathAndMethodGuards(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + ReportPath: "hook", + }) + if got := a.Config()["reportPath"]; got != "/hook" { + t.Fatalf("reportPath = %v, want /hook", got) + } + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + getResp, err := http.Get("http://" + a.Addr() + "/hook") + if err != nil { + t.Fatalf("GET /hook: %v", err) + } + defer func() { _ = getResp.Body.Close() }() + if getResp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("GET status = %d, want 405", getResp.StatusCode) + } + + other, err := http.Post("http://"+a.Addr()+"/elsewhere", "application/json", strings.NewReader(groupEventJSON)) + if err != nil { + t.Fatalf("POST /elsewhere: %v", err) + } + defer func() { _ = other.Body.Close() }() + if other.StatusCode != http.StatusNotFound { + t.Errorf("POST /elsewhere status = %d, want 404", other.StatusCode) + } +} diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go index 9b65476..2d6f74b 100644 --- a/plugins/channels/qq_obv11/session.go +++ b/plugins/channels/qq_obv11/session.go @@ -34,8 +34,10 @@ func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) if err := ctx.Err(); err != nil { return fmt.Errorf("qq-obv11: send message: %w", err) } + // HTTP 传输没有 ws 连接：只要出站 HTTP API 客户端已装配（http+client 或 + // http+server 配了 api_base_url）就照样能回消息。 conn := a.currentConn() - if conn == nil { + if conn == nil && a.currentAPI() == nil { return fmt.Errorf("qq-obv11: send message: %w", adapter.ErrNotConnected) } for i, msg := range resp.Messages { @@ -54,7 +56,7 @@ func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) return fmt.Errorf("qq-obv11: send proactive: %w", err) } conn := a.currentConn() - if conn == nil { + if conn == nil && a.currentAPI() == nil { return fmt.Errorf("qq-obv11: send proactive: %w", adapter.ErrNotConnected) } return a.sendMessage(ctx, conn, msg) @@ -81,9 +83,10 @@ func (a *Adapter) Capabilities() adapter.Capabilities { }.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. +// ListChannels returns the groups the adapter currently knows about (learned from +// inbound events). 全量好友/群列表要靠 OneBot 的 get_group_list / get_friend_list +// 动作，HTTP 出站客户端（api_base_url）现已具备，但把它接进本方法属后续工作， +// 本次四端改造不扩大范围。 func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) { a.groupMu.RLock() defer a.groupMu.RUnlock() @@ -206,8 +209,9 @@ func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { } } -// sendMessage builds and writes a single send_msg request, enforcing the rate -// limit and writing the payload to the connection. +// sendMessage builds and dispatches a single send_msg request, enforcing the +// rate limit first. 出站通道按 transport 分流：WS 两端把 JSON 帧写到长连接上； +// HTTP 端把同一条 action 请求 POST 给 OneBot 的 HTTP API 并等响应。 func (a *Adapter) sendMessage(ctx context.Context, conn *wsConn, msg adapter.UnifiedMessage) error { segs, err := outboundSegments(msg) if err != nil { @@ -230,9 +234,18 @@ func (a *Adapter) sendMessage(ctx context.Context, conn *wsConn, msg adapter.Uni if err := a.tryAcquireSend(msg); err != nil { return err } - if err := conn.WriteMessage(opText, payload); err != nil { - a.setDisconnected() - return fmt.Errorf("qq-obv11: write send_msg: %w", err) + switch api := a.currentAPI(); { + case api != nil: + // HTTP 端：Cyrene→OneBot 走 OneBot 的 HTTP API，等响应判成败（裁决 1.4）。 + // 失败不把适配器标记为断开——HTTP 没有会话可重连，下次发送会重试。 + if err := api.send(ctx, payload); err != nil { + return err + } + default: + 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() diff --git a/plugins/channels/qq_obv11/ws_client.go b/plugins/channels/qq_obv11/ws_client.go index dbf0ad3..34c99cb 100644 --- a/plugins/channels/qq_obv11/ws_client.go +++ b/plugins/channels/qq_obv11/ws_client.go @@ -1,8 +1,9 @@ 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. +// 本文件实现四个端里的 **ws + client**（裁决 1.4）：连接由 Cyrene 发起 +// （Cyrene→OneBot），主动拨号 OneBot 实现的「正向 WS」端点（remote_url），建立后 +// 收发都走这条长连接。access_token 同时以 Authorization: Bearer 头与 +// ?access_token= 查询参数发送，兼容只认其中一种的 OneBot 实现。 import ( "context" diff --git a/plugins/channels/qq_obv11/ws_server.go b/plugins/channels/qq_obv11/ws_server.go index 0a8cbd2..63e8db8 100644 --- a/plugins/channels/qq_obv11/ws_server.go +++ b/plugins/channels/qq_obv11/ws_server.go @@ -1,9 +1,10 @@ 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. +// 本文件实现四个端里的 **ws + server**（裁决 1.4）：Cyrene 监听 `listen_addr`，等 +// OneBot 实现（NapCat / go-cqhttp 的「反向 WS」）反连上来——连接由 OneBot 发起 +// （OneBot→Cyrene），建立后收发都走这条长连接。端口必须由配置显式给出，默认不开 +// （AGENTS §9 的登记豁免）。鉴权与 HTTP server 端一致：配置了 access_token 时， +// Authorization: Bearer 或 ?access_token= 必须匹配，否则 403。 import ( "context" @@ -28,13 +29,16 @@ func (a *Adapter) connectServer(ctx context.Context) error { return fmt.Errorf("qq-obv11: listen %s: %w", a.cfg.ListenAddr, err) } + // 先把 *http.Server 落在局部变量上再交给 goroutine：Disconnect 会把 a.srv 置 nil， + // 直接读 a.srv 会与它构成数据竞争。 + srv := &http.Server{Handler: mux} a.mu.Lock() a.ln = ln - a.srv = &http.Server{Handler: mux} + a.srv = srv a.mu.Unlock() go func() { - if serveErr := a.srv.Serve(ln); serveErr != nil && serveErr != http.ErrServerClosed { + if serveErr := srv.Serve(ln); serveErr != nil && serveErr != http.ErrServerClosed { a.log.Warn("websocket server stopped", "err", serveErr) } }()