feat(qq_obv11): OneBot v11 四端齐备(裁决 1.4 / 12.4.1)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 四端=ws×client / ws×server(原有,补语义注释)+ **http×client**(新增 http_client.go:
  调 OneBot HTTP API、请求-响应、retcode 判成败、token 只走 Authorization 头)+
  **http×server**(新增 http_server.go:监听 listen_addr 收上报、fail-closed 鉴权、10MiB 体上限)
- 配置面:适配器新增 transport / api_base_url / report_path(json tag 沿用既有口径);
  env 兜底 CHANNEL_QQ_OBV11_{TRANSPORT,API_BASE_URL,REPORT_PATH};**默认值=现状**(ws+server)
- 出站分流:session.go 发送路径按 transport 走 WS 写帧或 HTTP POST(不改它则 HTTP 端无法出站)
- 端口豁免边界:http+server 沿用 AGENTS §9 已登记的同一豁免(design/13 §0 原文即含 HTTP server),
  且 listen_addr 为空时报错、**绝不开端口**(与 ws 侧同名校验成对)
- 关键取舍(已上报):OneBot 的 HTTP 是「动作 + 上报」两半,mode 决定**哪一半必需**、
  另一半给齐字段就一并启用 ⇒ 单个适配器即可「收上报 + 回消息」,四端仍各自可独立选择与测试
- 复核:build/vet 绿 + qq_obv11/cmd 包测试绿 + `-race` 干净;10 条新用例(HTTP 两端全部用本地
  httptest 假 OneBot,未连真实 QQ);变异 3 条(伪造 retcode 成功 / 去掉上报鉴权门 / 默认值翻 http)已还原
- 未接:get_group_list(后续)、无 token 时放行(与既有 ws_server 一致,如需 fail-closed 请裁决)
This commit is contained in:
AskaEth committed 2026-10-04 13:58:18 +08:00
1 parent 0e1f163ef9
commit 9253fef92a
9 files changed
+972 -45

No files matched your search

+6
View File
@@ -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"),
}})
@@ -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
}
+163 -26
View File
@@ -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
}
}
+172
View File
@@ -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
}
+87
View File
@@ -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()
}
@@ -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)
}
}
+23 -10
View File
@@ -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()
+4 -3
View File
@@ -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"
+10 -6
View File
@@ -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)
}
}()