diff --git a/internal/channel/mediafetch/inline.go b/internal/channel/mediafetch/inline.go new file mode 100644 index 0000000..0bfe96f --- /dev/null +++ b/internal/channel/mediafetch/inline.go @@ -0,0 +1,139 @@ +package mediafetch + +// 本文件承载**四个适配器共享的入站附件内联循环**(切片 IM-1)。四家的差异只在 +// 「怎么把一个平台媒体引用取回成字节」(telegram 要先 getFile 换路径、weixin 要多 +// 一层 AES),预筛、限额、超时、日志与诚实降级的口径完全相同 —— 因此只实现一次, +// 由各适配器实现 Fetcher 注入(三次复制必须抽取,AGENTS §10)。 + +import ( + "context" + "log/slog" + "net/url" + "path" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// nameMaxRunes 是用于预筛与日志的文件名展示上限:文件名来自渠道(发送方可任意 +// 构造),因此这里限长,且它只用于**扩展名判定与日志展示** —— 绝不进模型上下文 +// (模型可见的展示名由 channelbridge 的净化规则负责)。 +const nameMaxRunes = 128 + +// Fetcher 把一条平台媒体引用取回成内联字节。实现者(各适配器)负责把平台给出的 +// 文件 id / CDN 引用解析成一个**平台载荷给出**的资产地址,再交给 Fetch 取回。 +// +// 契约:返回的错误必须可安全进日志 —— **不得携带完整 URL 或凭据**(用 Fetch 就 +// 自动满足:它只回来源与原因)。 +type Fetcher interface { + Fetch(ctx context.Context, media adapter.Media) ([]byte, error) +} + +// Inline 把消息里**文件类**媒体在受支持格式与限额内的字节内联进 `Media.Data`, +// 使下游(channelbridge 的入站摄取)拿到字节。它是四个适配器共享的唯一实现: +// +// - 只处理**文件类**媒体(图片/语音走各自既有链路:图片进视觉链路,design/24 +// §4.2 末),本切片不改动它们; +// - 先按平台给的文件名**预筛扩展名**(`documents.ExtAllowed`,全仓唯一名单): +// 不在名单里 ⇒ **不发起任何下载**,只留一条 Debug 日志(下游会照 M1 既有口径 +// 给模型一行诚实提示); +// - 平台**声明的**体积超上限时同样不发起下载(省一次必然失败的大传输); +// - 只内联前 `documents.MaxPerMessage` 个文件附件(与 Web 侧、M1 摄取同一口径, +// 绝不另造一个数);超出者不下载并记一条 Warn; +// - 下载失败/超时/超限 ⇒ 该附件保持无字节(下游继续给诚实提示),记一条带来源的 +// Warn 日志(name/mime/size/host/err —— **绝不记正文、绝不记 URL 原串**), +// **消息照收、回合继续**; +// - 已经带字节的媒体(本地产出)原样跳过,绝不重复下载。 +// +// 它**不返回错误**:入站附件的失败由「留空字节 + 日志 + 下游诚实提示」表达,绝不 +// 让一条读不到的附件拖垮整个回合(失败不炸)。 +func Inline(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, f Fetcher) { + if msg == nil || f == nil || log == nil { + return + } + seen := 0 + for i := range msg.Media { + m := &msg.Media[i] + if !isFileMedia(msg, *m) { + continue + } + if seen >= documents.MaxPerMessage { + log.Warn("inbound file attachments beyond per-message limit not inlined", + "platform", msg.Platform, "files", len(msg.Media), "limit", documents.MaxPerMessage) + return + } + seen++ + if len(m.Data) > 0 { + continue + } + name := displayName(*m) + if !documents.ExtAllowed(documents.ExtOf(name)) { + // 预筛未过:**一个字节都不发出去**。这不是失败,是「这个格式本来就不读」。 + log.Debug("inbound attachment not inlined: extension not allowed", + "platform", msg.Platform, "name", name, "mime", m.MIMEType, "size", m.Size) + continue + } + if m.Size > documents.DefaultMaxBytes { + log.Warn("inbound attachment not inlined: declared size over limit", + "platform", msg.Platform, "name", name, "mime", m.MIMEType, "size", m.Size, + "limit", documents.DefaultMaxBytes) + continue + } + data, err := f.Fetch(ctx, *m) + if err != nil { + log.Warn("inbound attachment inline failed", + "platform", msg.Platform, "name", name, "mime", m.MIMEType, "size", m.Size, + "host", HostOf(m.URL), "err", err) + continue + } + if len(data) == 0 { + log.Warn("inbound attachment inline returned no bytes", + "platform", msg.Platform, "name", name, "mime", m.MIMEType, "host", HostOf(m.URL)) + continue + } + m.Data = data + log.Debug("inbound attachment inlined", + "platform", msg.Platform, "name", name, "mime", m.MIMEType, "size", len(data)) + } +} + +// isFileMedia 报告一条媒体是不是**文件类**。判定与 channelbridge 的入站摄取同形 +// (`MediaFile`,或「只给了内容类型、附件就在 Media 里」的 `ContentFile`):两边 +// 必须对「哪些附件该被内联」有一致的答案,否则会出现「内联了但没人消费」。 +func isFileMedia(msg *adapter.UnifiedMessage, m adapter.Media) bool { + if m.Kind == adapter.MediaFile { + return true + } + return m.Kind == "" && msg.ContentType == adapter.ContentFile +} + +// displayName 给出用于**预筛**的展示名:优先平台给的文件名;缺名时退回资产地址的 +// 路径末段(平台 CDN 的地址末段通常就是真实文件名;只取路径,**不含查询串**,因此 +// 不会把签名凭据带进日志)。两者都拿不到时返回空串 —— `ExtAllowed("")` 为 false, +// 于是**不下载**(deny-by-default:宁可诚实说读不到,也不去下一条我们连类型都不知道 +// 的资产)。文件名是渠道给的字符串,因此这里只取 basename 并限长。 +func displayName(m adapter.Media) string { + name := strings.TrimSpace(m.Name) + if name != "" { + return limitName(name) + } + u, err := url.Parse(strings.TrimSpace(m.URL)) + if err != nil { + return "" + } + base := path.Base(u.Path) + if base == "." || base == "/" { + return "" + } + return limitName(base) +} + +// limitName 按 rune 截断文件名:它只用于扩展名判定与日志展示。 +func limitName(name string) string { + runes := []rune(strings.TrimSpace(name)) + if len(runes) > nameMaxRunes { + return string(runes[:nameMaxRunes]) + } + return string(runes) +} diff --git a/internal/channel/mediafetch/inline_test.go b/internal/channel/mediafetch/inline_test.go new file mode 100644 index 0000000..8fe1170 --- /dev/null +++ b/internal/channel/mediafetch/inline_test.go @@ -0,0 +1,244 @@ +package mediafetch + +// 本文件锁定共享内联循环的口径(切片 IM-1):预筛、限额、条数上限、失败不炸。 +// 「正文里的任意 URL ⇒ 0 次命中」这条红线在各适配器的用例里断言(只有适配器才知道 +// 正文长什么样);这里锁的是「循环只会把**媒体引用**交给 Fetcher」这条结构性事实。 + +import ( + "context" + "errors" + "io" + "log/slog" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// countingFetcher 记录每次被要求取回的媒体引用(引用原串与次数),使「0 次命中」 +// 与「命中次数」都可断言。 +type countingFetcher struct { + mu sync.Mutex + seen []string + data []byte + err error + maxHit int +} + +func (f *countingFetcher) Fetch(_ context.Context, m adapter.Media) ([]byte, error) { + f.mu.Lock() + f.seen = append(f.seen, m.URL) + n := len(f.seen) + f.mu.Unlock() + if f.maxHit > 0 && n > f.maxHit { + return nil, errors.New("超过了本用例允许的取回次数") + } + if f.err != nil { + return nil, f.err + } + return f.data, nil +} + +func (f *countingFetcher) calls() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.seen) +} + +// testLogger 返回一个丢弃输出的带来源日志器:用例不关心内容,只保证路径不 panic。 +func testLogger(t *testing.T) *slog.Logger { + t.Helper() + return logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "internal/channel/mediafetch") +} + +// txtMessage 构造一条「带一个 .txt 附件」的入站消息。 +func txtMessage(url string) *adapter.UnifiedMessage { + return &adapter.UnifiedMessage{ + Platform: "test", + ContentType: adapter.ContentFile, + Content: "看看这个文件 " + url, // 正文里也带同一个 URL:正文**不参与**取回决策 + Media: []adapter.Media{{ + Kind: adapter.MediaFile, URL: url, Name: "notes.txt", MIMEType: "text/plain", Size: 12, + }}, + } +} + +// TestInlineDownloadsAllowedFile 是正向用例:白名单格式的文件附件被内联进 Data。 +func TestInlineDownloadsAllowedFile(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("hello\n")} + msg := txtMessage("https://cdn.example.com/notes.txt") + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 1 { + t.Fatalf("取回次数 = %d, want 1", f.calls()) + } + if string(msg.Media[0].Data) != "hello\n" { + t.Fatalf("Data = %q, want 取回的字节", msg.Media[0].Data) + } +} + +// TestInlineNeverFetchesFromBody 钉住结构性事实:循环只把 `Media[].URL` 交给 +// Fetcher —— 正文里那个 URL 永远不是取回对象(它是同一条文本,但只作为字符串存在)。 +func TestInlineNeverFetchesFromBody(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("ok")} + msg := txtMessage("https://cdn.example.com/notes.txt") + bodyURL := "https://evil.example.com/secret.txt" + msg.Content = "读一下 " + bodyURL + " 和 " + msg.Media[0].URL + msg.Media = nil // 只有正文带 URL、没有任何文件媒体 + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("正文里的 URL 被取回了(%v),want 0 次", f.seen) + } +} + +// TestInlineSkipsUnsupportedExtensionWithoutFetch 钉住预筛:非白名单格式**不下载**。 +func TestInlineSkipsUnsupportedExtensionWithoutFetch(t *testing.T) { + t.Parallel() + for _, name := range []string{"payload.exe", "script.js", "page.html", "archive.zip", ""} { + t.Run(name, func(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := txtMessage("https://cdn.example.com/whatever") + msg.Media[0].Name = name + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("格式 %q 不在白名单里却发起了取回", name) + } + if len(msg.Media[0].Data) != 0 { + t.Fatalf("格式 %q 不该被内联", name) + } + }) + } +} + +// TestInlineSkipsDeclaredOversizeWithoutFetch 钉住「平台声明的体积超上限时连请求都 +// 不发」(省一次必然失败的大传输)。 +func TestInlineSkipsDeclaredOversizeWithoutFetch(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := txtMessage("https://cdn.example.com/big.txt") + msg.Media[0].Size = documents.DefaultMaxBytes + 1 + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("声明超限的附件仍被取回 %d 次,want 0", f.calls()) + } +} + +// TestInlineFailureKeepsMessageAndContinues 钉住「失败不炸」:一份取不到不阻断其余 +// 附件,也不改动消息的其他字段。 +func TestInlineFailureKeepsMessageAndContinues(t *testing.T) { + t.Parallel() + f := &countingFetcher{err: ErrBadStatus} + msg := txtMessage("https://cdn.example.com/a.txt") + msg.Media = append(msg.Media, adapter.Media{ + Kind: adapter.MediaFile, URL: "https://cdn.example.com/b.txt", Name: "b.txt", + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 2 { + t.Fatalf("取回次数 = %d, want 2(一次失败不阻断第二次)", f.calls()) + } + for i, m := range msg.Media { + if len(m.Data) != 0 { + t.Fatalf("Media[%d] 在失败时被写入了字节", i) + } + } + if !strings.Contains(msg.Content, "https://cdn.example.com/a.txt") { + t.Fatalf("失败路径改动了正文:%q", msg.Content) + } +} + +// TestInlineCapsAttachmentsPerMessage 钉住条数上限与 Web 侧同源(documents.MaxPerMessage)。 +func TestInlineCapsAttachmentsPerMessage(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := txtMessage("https://cdn.example.com/0.txt") + for i := 1; i <= documents.MaxPerMessage+2; i++ { + msg.Media = append(msg.Media, adapter.Media{ + Kind: adapter.MediaFile, URL: "https://cdn.example.com/x", Name: "x.txt", + }) + } + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != documents.MaxPerMessage { + t.Fatalf("取回次数 = %d, want %d(与 Web 侧同一上限)", f.calls(), documents.MaxPerMessage) + } +} + +// TestInlineSkipsMediaThatAlreadyHasBytes 钉住「已有字节不重复下载」(本地产出的 +// 附件照原样通过)。 +func TestInlineSkipsMediaThatAlreadyHasBytes(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("new")} + msg := txtMessage("https://cdn.example.com/a.txt") + msg.Media[0].Data = []byte("local") + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("已有字节的附件被重复下载 %d 次", f.calls()) + } + if string(msg.Media[0].Data) != "local" { + t.Fatalf("本地字节被覆盖:%q", msg.Media[0].Data) + } +} + +// TestInlineIgnoresNonFileMedia 钉住边界:图片/语音媒体不走本循环(图片走视觉链路)。 +func TestInlineIgnoresNonFileMedia(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := &adapter.UnifiedMessage{ + Platform: "test", + ContentType: adapter.ContentImage, + Media: []adapter.Media{ + {Kind: adapter.MediaImage, URL: "https://cdn.example.com/a.txt", Name: "a.txt"}, + {Kind: adapter.MediaVoice, URL: "https://cdn.example.com/b.txt", Name: "b.txt"}, + }, + } + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("非文件类媒体被取回了 %d 次,want 0", f.calls()) + } +} + +// TestDisplayNameFallsBackToURLPath 钉住缺名时的兜底:从资产地址路径末段取扩展名。 +func TestDisplayNameFallsBackToURLPath(t *testing.T) { + t.Parallel() + cases := []struct { + name string + url string + want string + }{ + {"文件名优先", "https://cdn.example.com/download?id=1", "report.csv"}, + {"无名字取路径末段", "https://cdn.example.com/2024/report.csv", "report.csv"}, + {"路径无末段则空", "https://cdn.example.com/", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + m := adapter.Media{Name: "", URL: tc.url} + if tc.name == "文件名优先" { + m.Name = "report.csv" + } + if got := displayName(m); got != tc.want { + t.Fatalf("displayName = %q, want %q", got, tc.want) + } + }) + } +} diff --git a/internal/channel/mediafetch/mediafetch.go b/internal/channel/mediafetch/mediafetch.go new file mode 100644 index 0000000..cd87bb7 --- /dev/null +++ b/internal/channel/mediafetch/mediafetch.go @@ -0,0 +1,369 @@ +// Package mediafetch 实现 IM 适配器**入站附件字节内联**的共用件(切片 IM-1): +// 把「平台载荷给出的一个资产引用」取回成字节,交回给适配器写进 +// `adapter.Media.Data`,使 channelbridge 的入站摄取(M1)真的读到文件内容。 +// +// 三条安全口径(切片 IM-1 的硬约束,逐条都是红线): +// +// 1. **只取平台载荷给出的引用**:本包只接受调用方(各适配器)从平台原生载荷里 +// 解析出来的资产引用 —— 由各适配器的 Fetcher 负责构造;本包**从不**解析消息 +// 正文,也不从正文里找 URL(与 F3 的 SSRF 教训一致:服务端代取「用户指定的 +// 任意 URL」是另一回事,需要专门裁决)。 +// +// 2. **来源白名单、deny-by-default**:请求的 `scheme://host[:port]` 必须命中调用 +// 方声明的来源(平台 CDN / 平台 API 主机 / 部署自己声明的 OneBot 端点)。不在 +// 白名单里 ⇒ **一个字节都不发出去**;重定向的每一跳也按同一份白名单校验,因此 +// 白名单不能靠一次 302 被绕过。它比「先解析 IP 再判网段」更严:白名单之外的任何 +// 主机(云元数据、内网、任意公网)都不可达。 +// +// 与既有两处守卫的关系(**本包不是第三份 SSRF 实现**):`plugins/tools/io` +// 的 `validateRequestURL` 与 `internal/runtime/multimodal/vision` 的取图守卫 +// 都是「先放行、再按 IP 网段否决」的形状,且**刻意**允许环回/私网(同源自取 +// 需要它们);入站 IM 附件不需要那份宽容 —— 它只需要够到平台自己的资产主机, +// 所以这里用「按来源白名单短路」而非网段判定。视觉链路的守卫本切片**未改动**。 +// +// 3. **尺寸与时间双上限**:边读边限(`io.LimitReader`,超限即放弃),每份下载有 +// 具名超时(默认 `DefaultTimeout`)并接 ctx 取消。 +// +// 错误一律**不含完整 URL**(平台资产地址可能带签名凭据,例如 QQ CDN 的 `rkey`), +// 只带来源与原因:这让错误可以安全进日志与模型可见视图。 +package mediafetch + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +const ( + // DefaultTimeout 是单份附件下载的默认超时上限(具名常量,全仓唯一出处)。 + // 它只约束这一份下载:父 ctx 取消(例如渠道断开、回合结束)会立即一并取消。 + DefaultTimeout = 30 * time.Second + + // maxRedirects 是一次取回最多跟随的重定向跳数。比 net/http 默认的 10 更小: + // 取一份平台附件不需要长跳转链,少跟几跳就少几次对陌生主机的连接(每一跳都 + // 仍按白名单校验)。 + maxRedirects = 3 + + // wildcardPrefix / schemeSep 是白名单条目里通配与协议分隔的写法定界,避免魔法 + // 字符串散落在判定逻辑里。 + wildcardPrefix = "*." + schemeSep = "://" + + // unknownHost 是日志里 host 字段取不到来源时的占位值。 + unknownHost = "未知来源" +) + +// 取回路径的哨兵错误。调用方一律用 errors.Is 分类,**绝不按错误文本匹配**。 +var ( + // ErrOriginNotAllowed 报告请求的来源(或重定向目标)不在白名单内 —— 唯一一个 + // 「我们主动不发请求」的哨兵:它与「发了但失败」必须能分开(前者 0 次命中)。 + ErrOriginNotAllowed = errors.New("mediafetch: 资产来源不在白名单内") + + // ErrBadScheme 报告资产引用不是 http/https。 + ErrBadScheme = errors.New("mediafetch: 资产引用只支持 http/https") + + // ErrNoHost 报告资产引用没有主机。 + ErrNoHost = errors.New("mediafetch: 资产引用缺少主机") + + // ErrEmptyURL 报告资产引用为空。 + ErrEmptyURL = errors.New("mediafetch: 资产引用为空") + + // ErrTooLarge 报告附件超过体积上限(边读边限,超限即放弃)。 + ErrTooLarge = errors.New("mediafetch: 附件超过体积上限") + + // ErrBadStatus 报告平台资产响应的状态码不是 2xx。 + ErrBadStatus = errors.New("mediafetch: 平台资产响应非成功状态") + + // ErrTooManyRedirects 报告重定向超过上限。 + ErrTooManyRedirects = errors.New("mediafetch: 重定向超过上限") +) + +// Allowlist 是一组允许连接的来源(`scheme://host[:port]`,可写成 +// `https://*.example.com` 表示整棵子树)。**零值表示谁都不许**:deny-by-default。 +type Allowlist struct { + origins []string +} + +// NewAllowlist 构造白名单。条目按字面归一(小写、去尾斜杠);**无法解析的条目一律 +// 不匹配任何来源**(fail-closed:写错的条目不会意外放行任何主机)。 +func NewAllowlist(origins ...string) Allowlist { + out := make([]string, 0, len(origins)) + for _, o := range origins { + e := normalizeEntry(o) + if e != "" { + out = append(out, e) + } + } + return Allowlist{origins: out} +} + +// Empty 报告白名单是否没有任何条目(任何请求都会被拒)。 +func (a Allowlist) Empty() bool { return len(a.origins) == 0 } + +// Allowed 报告 rawURL 的来源是否在白名单内。 +func (a Allowlist) Allowed(rawURL string) bool { + origin, err := OriginOf(rawURL) + if err != nil { + return false + } + for _, e := range a.origins { + if matchOrigin(e, origin) { + return true + } + } + return false +} + +// OriginOf 返回绝对 URL 的来源字符串(`scheme://host[:port]`,协议默认端口被 +// 归一掉、主机小写)。它只用于**比对与日志**,绝不用来重建请求地址。 +func OriginOf(rawURL string) (string, error) { + u, err := url.Parse(strings.TrimSpace(rawURL)) + if err != nil { + return "", fmt.Errorf("mediafetch: 解析资产地址: %w", err) + } + return originOfURL(u) +} + +// Request 是一次取回的输入。URL 必须由调用方(适配器)从**平台载荷**构造;本包 +// 不提供任何「从消息正文里找 URL」的入口。 +type Request struct { + // URL 是平台载荷给出的资产地址(绝对 http/https URL)。 + URL string + // Allow 是允许连接的来源白名单。空名单 ⇒ 一律拒绝(deny-by-default)。 + Allow Allowlist + // Header 是随首次请求发送的附加头(例如平台凭据)。跟随重定向时由 net/http + // 按自身规则决定是否携带(跨主机时敏感头会被剥离),调用方无需处理。 + Header http.Header + // MaxBytes 是单份附件的字节上限;≤0 时取 documents.DefaultMaxBytes(20 MiB, + // 与 Web 上传同源,绝不在适配器里另写一个数)。 + MaxBytes int64 + // Timeout 是本次取回的超时;≤0 时取 DefaultTimeout。 + Timeout time.Duration + // Client 是可选的 HTTP 客户端(适配器已有的客户端)。取回时用它的**副本** + // (不改动调用方的客户端),只覆盖重定向校验。nil 时新建一个。 + Client *http.Client +} + +// Fetch 取回一份平台资产并返回其字节。失败一律返回错误(绝不返回「空字节 + nil」: +// 上游据此区分「拿到了空文件」与「没取到」)。 +func Fetch(ctx context.Context, req Request) ([]byte, error) { + if strings.TrimSpace(req.URL) == "" { + return nil, ErrEmptyURL + } + origin, err := OriginOf(req.URL) + if err != nil { + return nil, err + } + if !req.Allow.Allowed(req.URL) { + // 白名单未命中:**不发请求**。这条路径是可断言的「0 次命中」。 + return nil, fmt.Errorf("%w: %s", ErrOriginNotAllowed, origin) + } + + ctx, cancel := context.WithTimeout(ctx, positiveDuration(req.Timeout, DefaultTimeout)) + defer cancel() + client := fetchClient(req.Client, req.Allow) + maxBytes := positiveBytes(req.MaxBytes) + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, req.URL, nil) + if err != nil { + return nil, fmt.Errorf("mediafetch: 构造 %s 的取回请求: %w", origin, err) + } + applyHeader(httpReq.Header, req.Header) + + resp, err := client.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("mediafetch: 取回 %s 失败: %w", origin, sanitizeErr(err)) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("%w: %d(%s)", ErrBadStatus, resp.StatusCode, origin) + } + // 边读边限:多读一个字节用来判定「超限」,因此绝不会把一份超大文件读满内存。 + data, err := io.ReadAll(io.LimitReader(resp.Body, maxBytes+1)) + if err != nil { + return nil, fmt.Errorf("mediafetch: 读取 %s 失败: %w", origin, sanitizeErr(err)) + } + if int64(len(data)) > maxBytes { + return nil, fmt.Errorf("%w(上限 %d 字节,来源 %s)", ErrTooLarge, maxBytes, origin) + } + return data, nil +} + +// HostOf 返回 rawURL 的来源(用于日志),解析不出来时给占位值 —— 日志里**只有来源**, +// 绝不出现完整 URL(可能带凭据)。 +func HostOf(rawURL string) string { + origin, err := OriginOf(rawURL) + if err != nil { + return unknownHost + } + return origin +} + +// fetchClient 返回带「逐跳白名单校验」的客户端副本:绝不改动调用方的客户端(它可能 +// 正被上游请求等其他用途共享)。调用方原本的 CheckRedirect 会被覆盖 —— 入站附件 +// 的取回安全由本包的白名单独占负责,不叠加别处的重定向策略。 +func fetchClient(base *http.Client, allow Allowlist) *http.Client { + c := &http.Client{} + if base != nil { + *c = *base + } + c.CheckRedirect = func(r *http.Request, via []*http.Request) error { + if len(via) >= maxRedirects { + return ErrTooManyRedirects + } + if r == nil || r.URL == nil { + return ErrNoHost + } + if !allow.Allowed(r.URL.String()) { + return fmt.Errorf("%w: 重定向到 %s", ErrOriginNotAllowed, HostOf(r.URL.String())) + } + return nil + } + return c +} + +// applyHeader 把附加头装进请求。 +func applyHeader(dst, src http.Header) { + for k, vs := range src { + for _, v := range vs { + dst.Add(k, v) + } + } +} + +// sanitizeErr 剥掉 *url.Error 的外壳:它会把**完整 URL**(含查询串里的签名凭据) +// 拼进错误串。剥壳同时保留内层哨兵链(`context.DeadlineExceeded` 等仍可用 +// errors.Is 判定)。内层已是最终错误时**原样透传**(这里只做剥壳,不新增包装 —— +// 再包一层只会把同一个原因说两遍)。 +func sanitizeErr(err error) error { + var uerr *url.Error + if errors.As(err, &uerr) && uerr.Err != nil { + return uerr.Err + } + return err +} + +// originOfURL 把一个已解析的 URL 归一成来源字符串。 +func originOfURL(u *url.URL) (string, error) { + if u == nil { + return "", ErrEmptyURL + } + scheme := strings.ToLower(u.Scheme) + if scheme != "http" && scheme != "https" { + return "", fmt.Errorf("%w: %q", ErrBadScheme, scheme) + } + host := strings.ToLower(u.Hostname()) + if host == "" { + return "", ErrNoHost + } + port := u.Port() + if port == "" || port == defaultPort(scheme) { + return scheme + schemeSep + joinHostPort(host, ""), nil + } + return scheme + schemeSep + joinHostPort(host, port), nil +} + +// joinHostPort 拼主机与端口名,IPv6 字面量会被重新加回方括号。 +func joinHostPort(host, port string) string { + if strings.Contains(host, ":") { + if port == "" { + return "[" + host + "]" + } + return "[" + host + "]:" + port + } + if port == "" { + return host + } + return host + ":" + port +} + +// defaultPort 给出协议的默认端口(归一用)。 +func defaultPort(scheme string) string { + if scheme == "https" { + return "443" + } + return "80" +} + +// normalizeEntry 归一一条白名单条目:小写、去尾斜杠、去掉路径部分。无法解析的条目 +// 返回空串(随后被丢弃 ⇒ 永不匹配任何来源)。 +func normalizeEntry(entry string) string { + e := strings.ToLower(strings.TrimSpace(entry)) + if e == "" { + return "" + } + if !strings.Contains(e, schemeSep) { + return "" + } + // 只保留 scheme://host[:port],丢掉任何路径(条目写成 URL 时按来源处理)。 + trimmed := strings.TrimRight(e, "/") + if i := strings.Index(trimmed, schemeSep); i >= 0 { + rest := trimmed[i+len(schemeSep):] + if j := strings.IndexByte(rest, '/'); j >= 0 { + trimmed = trimmed[:i+len(schemeSep)+j] + } + } + return trimmed +} + +// matchOrigin 报告已归一的 origin 是否被条目 entry 覆盖:精确相等,或条目是 +// `scheme://*.suffix` 形态且 origin 的主机落在该子树内(`*.qq.com` 覆盖 +// `a.qq.com` 与 `a.b.qq.com`,但不覆盖 `qq.com` 自身)。 +func matchOrigin(entry, origin string) bool { + if entry == origin { + return true + } + i := strings.Index(entry, schemeSep) + if i < 0 { + return false + } + scheme, hostPat := entry[:i], entry[i+len(schemeSep):] + if !strings.HasPrefix(hostPat, wildcardPrefix) { + return false + } + suffix := hostPat[len(wildcardPrefix)-1:] // ".qq.com" + j := strings.Index(origin, schemeSep) + if j < 0 || origin[:j] != scheme { + return false + } + host := stripPort(origin[j+len(schemeSep):]) + return strings.HasSuffix(host, suffix) && len(host) > len(suffix) +} + +// stripPort 去掉来源主机里的端口(IPv6 字面量的方括号形态一并处理)。 +func stripPort(hostport string) string { + if strings.HasPrefix(hostport, "[") { + if k := strings.Index(hostport, "]"); k >= 0 { + return hostport[1:k] + } + return hostport + } + if k := strings.LastIndex(hostport, ":"); k >= 0 { + return hostport[:k] + } + return hostport +} + +// positiveDuration 返回 d(d ≤ 0 时用 fallback),使路径里不出现裸的「0 表示默认」。 +func positiveDuration(d, fallback time.Duration) time.Duration { + if d <= 0 { + return fallback + } + return d +} + +// positiveBytes 返回上限字节数(≤0 时取 documents.DefaultMaxBytes)。 +func positiveBytes(n int64) int64 { + if n <= 0 { + return documents.DefaultMaxBytes + } + return n +} diff --git a/internal/channel/mediafetch/mediafetch_test.go b/internal/channel/mediafetch/mediafetch_test.go new file mode 100644 index 0000000..419ab2e --- /dev/null +++ b/internal/channel/mediafetch/mediafetch_test.go @@ -0,0 +1,235 @@ +package mediafetch + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// hitServer 是一个「记录命中次数」的假资产服务端:所有断言「0 次命中」的反证用例 +// 都靠它的计数器说话(不是靠日志或猜测)。 +type hitServer struct { + srv *httptest.Server + hits atomic.Int32 + body []byte + delay time.Duration + code int + next string // 非空时回 302 到该地址 +} + +func newHitServer(t *testing.T, body []byte) *hitServer { + t.Helper() + hs := &hitServer{body: body} + hs.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hs.hits.Add(1) + if hs.delay > 0 { + time.Sleep(hs.delay) + } + if hs.next != "" { + http.Redirect(w, r, hs.next, http.StatusFound) + return + } + if hs.code != 0 { + w.WriteHeader(hs.code) + return + } + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write(hs.body) + })) + t.Cleanup(hs.srv.Close) + return hs +} + +func (h *hitServer) count() int { return int(h.hits.Load()) } + +// TestAllowlistMatchSemantics 钉住白名单的判定口径:deny-by-default、精确来源匹配、 +// `*.suffix` 覆盖整棵子树但**不**覆盖裸域名。 +func TestAllowlistMatchSemantics(t *testing.T) { + t.Parallel() + cases := []struct { + name string + entries []string + url string + want bool + }{ + {"空名单一律拒", nil, "https://cdn.example.com/a.txt", false}, + {"精确命中", []string{"https://cdn.example.com"}, "https://cdn.example.com/a.txt", true}, + {"主机大小写归一", []string{"https://cdn.example.com"}, "https://CDN.Example.COM/a.txt", true}, + {"协议不同即拒", []string{"https://cdn.example.com"}, "http://cdn.example.com/a.txt", false}, + {"端口不同即拒", []string{"https://cdn.example.com"}, "https://cdn.example.com:8443/a.txt", false}, + {"默认端口等价", []string{"https://cdn.example.com"}, "https://cdn.example.com:443/a.txt", true}, + {"通配覆盖子树", []string{"https://*.qq.com"}, "https://multimedia.nt.qq.com.cn/x", false}, + {"通配覆盖子域", []string{"https://*.qq.com"}, "https://a.qq.com/x", true}, + {"通配覆盖多级子域", []string{"https://*.qq.com"}, "https://a.b.qq.com/x", true}, + {"通配不覆盖裸域", []string{"https://*.qq.com"}, "https://qq.com/x", false}, + {"畸形条目不匹配任何东西", []string{"cdn.example.com"}, "https://cdn.example.com/a.txt", false}, + {"非 http 引用一律拒", []string{"https://cdn.example.com"}, "file:///etc/passwd", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := NewAllowlist(tc.entries...).Allowed(tc.url); got != tc.want { + t.Fatalf("Allowed(%q) = %v, want %v", tc.url, got, tc.want) + } + }) + } +} + +// TestFetchRejectsOffAllowlistWithoutRequest 是本包最要紧的反证:来源不在白名单里时 +// **一个请求都不发出去**(断言服务端 0 次命中)。 +func TestFetchRejectsOffAllowlistWithoutRequest(t *testing.T) { + t.Parallel() + hs := newHitServer(t, []byte("secret")) + // 白名单里放的是别的主机 —— 与真实资产主机同网段、同形态,只差来源不匹配。 + allow := NewAllowlist("https://assets.example.com") + + _, err := Fetch(context.Background(), Request{URL: hs.srv.URL + "/a.txt", Allow: allow, Client: hs.srv.Client()}) + if !errors.Is(err, ErrOriginNotAllowed) { + t.Fatalf("err = %v, want ErrOriginNotAllowed", err) + } + if hs.count() != 0 { + t.Fatalf("白名单外的来源被请求了 %d 次,want 0", hs.count()) + } +} + +// TestFetchReturnsBytes 是正向用例:来源在白名单里 ⇒ 原样取回字节。 +func TestFetchReturnsBytes(t *testing.T) { + t.Parallel() + hs := newHitServer(t, []byte("正文内容\n")) + allow := NewAllowlist(originOf(t, hs.srv.URL)) + + got, err := Fetch(context.Background(), Request{URL: hs.srv.URL + "/a.txt", Allow: allow, Client: hs.srv.Client()}) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + if string(got) != "正文内容\n" { + t.Fatalf("bytes = %q", got) + } + if hs.count() != 1 { + t.Fatalf("命中次数 = %d, want 1", hs.count()) + } +} + +// TestFetchRejectsOversizeWhileReading 钉住「边读边限」:服务端超过上限时放弃, +// 返回 ErrTooLarge(绝不把整份超大响应读满内存)。 +func TestFetchRejectsOversizeWhileReading(t *testing.T) { + t.Parallel() + hs := newHitServer(t, []byte(strings.Repeat("x", 4096))) + allow := NewAllowlist(originOf(t, hs.srv.URL)) + + _, err := Fetch(context.Background(), Request{ + URL: hs.srv.URL + "/big.txt", Allow: allow, Client: hs.srv.Client(), MaxBytes: 64, + }) + if !errors.Is(err, ErrTooLarge) { + t.Fatalf("err = %v, want ErrTooLarge", err) + } +} + +// TestFetchTimeoutHonorsContext 钉住超时:单份下载有具名超时,超时后放弃并保留 +// ctx 哨兵(调用方据此分类,不必解析错误文本)。 +func TestFetchTimeoutHonorsContext(t *testing.T) { + t.Parallel() + hs := newHitServer(t, []byte("slow")) + hs.delay = 500 * time.Millisecond + allow := NewAllowlist(originOf(t, hs.srv.URL)) + + start := time.Now() + _, err := Fetch(context.Background(), Request{ + URL: hs.srv.URL + "/slow.txt", Allow: allow, Client: hs.srv.Client(), Timeout: 50 * time.Millisecond, + }) + if err == nil { + t.Fatal("err = nil, want timeout") + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("取回耗时 %v,超时没有生效", elapsed) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want context.DeadlineExceeded(哨兵链必须保留)", err) + } +} + +// TestFetchBadStatus 钉住非 2xx 一律失败(绝不把错误页正文当附件收下)。 +func TestFetchBadStatus(t *testing.T) { + t.Parallel() + hs := newHitServer(t, nil) + hs.code = http.StatusNotFound + allow := NewAllowlist(originOf(t, hs.srv.URL)) + + _, err := Fetch(context.Background(), Request{URL: hs.srv.URL + "/missing.txt", Allow: allow, Client: hs.srv.Client()}) + if !errors.Is(err, ErrBadStatus) { + t.Fatalf("err = %v, want ErrBadStatus", err) + } +} + +// TestFetchBlocksRedirectOffAllowlist 钉住重定向逐跳校验:302 到白名单外的主机时 +// 拒绝,且**目标主机一次都没被请求**。 +func TestFetchBlocksRedirectOffAllowlist(t *testing.T) { + t.Parallel() + outside := newHitServer(t, []byte("evil")) + inside := newHitServer(t, []byte("ok")) + inside.next = outside.srv.URL + "/evil.txt" + allow := NewAllowlist(originOf(t, inside.srv.URL)) + + _, err := Fetch(context.Background(), Request{ + URL: inside.srv.URL + "/a.txt", Allow: allow, Client: inside.srv.Client(), + }) + if !errors.Is(err, ErrOriginNotAllowed) { + t.Fatalf("err = %v, want ErrOriginNotAllowed(重定向跨到白名单外)", err) + } + if outside.count() != 0 { + t.Fatalf("白名单外的重定向目标被请求了 %d 次,want 0", outside.count()) + } +} + +// TestFetchErrorNeverCarriesFullURL 钉住「错误可安全进日志」:平台资产地址可能带签名 +// 凭据(如 CDN 的 rkey),错误串里**绝不能**出现它。 +func TestFetchErrorNeverCarriesFullURL(t *testing.T) { + t.Parallel() + // 127.0.0.1:1 上不会有人监听;白名单放行它,于是请求真的会发出去但连不上。 + const secret = "rkey=SUPER_SECRET_VALUE" + allow := NewAllowlist("http://127.0.0.1:1") + + _, err := Fetch(context.Background(), Request{ + URL: "http://127.0.0.1:1/download?" + secret, Allow: allow, Timeout: 200 * time.Millisecond, + }) + if err == nil { + t.Fatal("err = nil, want dial failure") + } + if strings.Contains(err.Error(), secret) || strings.Contains(err.Error(), "/download") { + t.Fatalf("错误串里出现了 URL 或凭据:%v", err) + } + if !strings.Contains(err.Error(), "http://127.0.0.1:1") { + t.Fatalf("错误串里应保留来源(便于定位是哪个平台主机):%v", err) + } +} + +// TestFetchRejectsEmptyAndNonHTTP 钉住输入侧的基本拒绝面。 +func TestFetchRejectsEmptyAndNonHTTP(t *testing.T) { + t.Parallel() + allow := NewAllowlist("https://cdn.example.com") + if _, err := Fetch(context.Background(), Request{URL: " ", Allow: allow}); !errors.Is(err, ErrEmptyURL) { + t.Fatalf("空引用 err = %v, want ErrEmptyURL", err) + } + if _, err := Fetch(context.Background(), Request{URL: "ftp://cdn.example.com/a", Allow: allow}); !errors.Is(err, ErrBadScheme) { + t.Fatalf("非 http err = %v, want ErrBadScheme", err) + } + // HostOf 只回来源,绝不回完整 URL。 + if got := HostOf("https://cdn.example.com/a?token=SECRET"); strings.Contains(got, "SECRET") { + t.Fatalf("HostOf 泄露了查询串:%q", got) + } +} + +// originOf 返回测试服务端的来源串。 +func originOf(t *testing.T, rawURL string) string { + t.Helper() + origin, err := OriginOf(rawURL) + if err != nil { + t.Fatalf("OriginOf(%q): %v", rawURL, err) + } + return origin +} diff --git a/internal/services/channelbridge/inbound_documents.go b/internal/services/channelbridge/inbound_documents.go index 01d8de5..083d5b3 100644 --- a/internal/services/channelbridge/inbound_documents.go +++ b/internal/services/channelbridge/inbound_documents.go @@ -59,9 +59,13 @@ const ( inboundNoticeTooLarge = "[附件 %s · %s · %d 字节 · 无法读取:文件超过服务端体积上限]" inboundNoticeEmpty = "[附件 %s · %s · %d 字节 · 无法读取:文件为空]" inboundNoticeStoreError = "[附件 %s · %s · %d 字节 · 无法读取:服务端保存该附件失败]" - inboundNoticeNoBytes = "[附件 %s · %s · %d 字节 · 无法读取:渠道只给了远程引用,服务端未代取文件字节]" - inboundNoticeUnwired = "[附件 %s · %s · %d 字节 · 无法读取:服务端未接线文档存储]" - inboundNoticeTooMany = "[另有附件未处理:单条消息最多 %d 个文件]" + // inboundNoticeNoBytes 覆盖「服务端没拿到字节」这一类:渠道只给了远程引用、或 + // 代取失败(切片 IM-1 起,受支持格式的附件由**适配器**在入站时按平台资产白名单 + // 取回并内联;取不到时字节仍为空,走这一行)。文案对两种成因都成立 —— 把「代取 + // 失败」说成「我们根本不代取」也是一种不诚实。 + inboundNoticeNoBytes = "[附件 %s · %s · %d 字节 · 无法读取:服务端没有拿到文件字节(渠道只给了远程引用,或代取失败)]" + inboundNoticeUnwired = "[附件 %s · %s · %d 字节 · 无法读取:服务端未接线文档存储]" + inboundNoticeTooMany = "[另有附件未处理:单条消息最多 %d 个文件]" ) // inboundAttachments 是一次入站附件摄取的产物。 @@ -147,9 +151,11 @@ func (b *Bridge) ingestFiles(ctx context.Context, store *documents.Store, msg ad break } if len(media.Data) == 0 { - // 渠道只给了远程引用(CDN URL / 平台文件 id):本切片**不代取** —— - // 服务端代取任意 http(s) 正是 F3 的教训(design/24 §5),需要专门 - // 的裁决与地址守卫,不在这里顺手打开。绝不静默。 + // 渠道只给了远程引用(CDN URL / 平台文件 id)而字节仍为空:切片 IM-1 起, + // **适配器**在入站时会按平台资产白名单把受支持格式的附件取回并内联,因此 + // 走到这里意味着「渠道没给内容」或「代取失败」。服务端代取**任意 http(s) + // URL** 仍是 F3 的教训(需要专门裁决与地址守卫),而代取**平台自己的资产** + // 由适配器按来源白名单完成 —— 两者不是同一件事。绝不静默。 b.log.Warn("inbound attachment has no inlined bytes", "platform", msg.Platform, "channel", msg.ChannelID, "name", media.Name, "mime", media.MIMEType, "size", media.Size) diff --git a/internal/services/channelbridge/inbound_documents_honesty_test.go b/internal/services/channelbridge/inbound_documents_honesty_test.go index a69943b..a4191d1 100644 --- a/internal/services/channelbridge/inbound_documents_honesty_test.go +++ b/internal/services/channelbridge/inbound_documents_honesty_test.go @@ -41,8 +41,9 @@ func TestInboundOversizeFileIsHonestAboutSize(t *testing.T) { } } -// TestInboundRemoteOnlyMediaIsHonestWithoutFetch 钉住「渠道只给远程引用」这一现实形态 -// (适配器目前上报 URL/平台文件 id,而不是内联字节):本切片**不代取**,但绝不静默。 +// TestInboundRemoteOnlyMediaIsHonestWithoutFetch 钉住「字节为空」这一形态:桥接侧 +// **不代取**(受支持格式的代取由适配器在入站时按平台资产白名单完成,切片 IM-1), +// 字节空 ⇒ 一行诚实提示 + 一条带来源的日志,绝不静默。 func TestInboundRemoteOnlyMediaIsHonestWithoutFetch(t *testing.T) { t.Parallel() env := newIngestEnv(t, documents.StoreConfig{}, true) diff --git a/internal/services/channelbridge/inbound_documents_inline_test.go b/internal/services/channelbridge/inbound_documents_inline_test.go new file mode 100644 index 0000000..8679e37 --- /dev/null +++ b/internal/services/channelbridge/inbound_documents_inline_test.go @@ -0,0 +1,183 @@ +package channelbridge + +// 本文件承载 **IM-1 × M1 的跨层端到端用例**:让**真的 Telegram 适配器**在入站时把 +// 平台文件取回并内联,再把那条消息喂进**真的桥接入站流水线**(真 store + 真抽取器 + +// 真循环 + 假模型,夹具沿用 inbound_documents_test.go),断言模型可见视图里出现该附件 +// 的清单行与正文 —— 即任务书 IM-1 §4-1 要求的「与 M1 既有集成用例同形状」。 +// +// 另含本切片的红线:正文里点名的任意 URL **绝不**被取回(按路径分别断言命中次数)。 + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" + + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/telegram" +) + +// e2eAssetServer 是同时充当「Telegram Bot API」与「文件服务端」的假服务端:它按 +// 路径分别计数,使「附件取回 1 次、正文里的 URL 0 次」可以分别断言。 +type e2eAssetServer struct { + srv *httptest.Server + docHits atomic.Int32 + bodyHits atomic.Int32 + docBody string +} + +func newE2EAssetServer(t *testing.T, docBody string) *e2eAssetServer { + t.Helper() + s := &e2eAssetServer{docBody: docBody} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/getFile"): + okResponseJSON(w, map[string]any{ + "file_id": "doc-id", "file_path": "documents/report.txt", "file_size": len(docBody), + }) + case strings.Contains(r.URL.Path, "/file/bot") && strings.HasSuffix(r.URL.Path, "documents/report.txt"): + s.docHits.Add(1) + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write([]byte(docBody)) + case strings.Contains(r.URL.Path, "/file/bot"): + // 正文里点名的那个 URL 打到这里:它**必须** 0 次命中。 + s.bodyHits.Add(1) + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write([]byte("绝不该被取回")) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(s.srv.Close) + return s +} + +// okResponseJSON 写一个 Bot API 成功信封。 +func okResponseJSON(w http.ResponseWriter, result any) { + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": result}) +} + +// e2eUpdate 构造一条主人(chat/from = 100)的文档消息,正文里**额外点名**一个 URL。 +func e2eUpdate(t *testing.T, fileName, text string) []byte { + t.Helper() + raw, err := json.Marshal(map[string]any{ + "update_id": 1, + "message": map[string]any{ + "message_id": 42, + "date": 1700000000, + "chat": map[string]any{"id": 100, "type": "private"}, + "from": map[string]any{"id": 100, "first_name": "主人"}, + "text": text, + "document": map[string]any{ + "file_id": "doc-id", "file_name": fileName, "file_size": 64, "mime_type": "text/plain", + }, + }, + }) + if err != nil { + t.Fatalf("marshal update: %v", err) + } + return raw +} + +// TestInboundAdapterInlineReachesModelView 是切片 IM-1 的主验收用例:真适配器取回 +// 字节 → 桥接摄取 → 抽取 → **模型可见视图里出现清单行与正文**。 +func TestInboundAdapterInlineReachesModelView(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + body := "第一行:IM 附件正文(IM-1 端到端)\n第二行:必须真的进上下文\n" + s := newE2EAssetServer(t, body) + + // 真适配器:Token/BaseURL/Client 都指向假服务端,日志丢弃。 + ad := telegram.New( + telegram.Config{BotToken: "123456:TESTTOKEN"}, + telegram.WithBaseURL(s.srv.URL), + telegram.WithHTTPClient(s.srv.Client()), + telegram.WithLogger(slog.New(slog.NewTextHandler(discardWriter{}, nil))), + ) + var delivered []adapter.UnifiedMessage + ad.SetMessageHandler(func(_ context.Context, msg adapter.UnifiedMessage) { + delivered = append(delivered, msg) + }) + + // 正文里点名一个「附件地址」形态的 URL(假服务端确实能应答它)。 + bodyURL := s.srv.URL + "/file/bot123456:TESTTOKEN/documents/body_only.txt" + if _, err := ad.ProcessUpdate(context.Background(), e2eUpdate(t, "report.txt", "读一下 "+bodyURL)); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + if len(delivered) != 1 { + t.Fatalf("投递条数 = %d, want 1", len(delivered)) + } + if len(delivered[0].Media) != 1 || string(delivered[0].Media[0].Data) != body { + t.Fatalf("适配器没有把附件字节内联进来:%+v", delivered[0].Media) + } + + env.bridge.process(context.Background(), delivered[0]) + + names := docDirNames(t, env.uploads) + if len(names) != 1 { + t.Fatalf("落盘文件数 = %v, want 1(内联后的附件必须落库)", names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, "[附件 "+names[0]+" · ") { + t.Fatalf("模型可见视图里没有附件清单行:%q", visible) + } + if !strings.Contains(visible, "第二行:必须真的进上下文") { + t.Fatalf("模型可见视图里没有附件正文:%q", visible) + } + if !strings.Contains(visible, "读一下 ") { + t.Fatalf("模型可见视图里丢了用户文本:%q", visible) + } + // 红线:附件本身被取回 1 次,正文里点名的那个 URL 一次都没有。 + if got := s.docHits.Load(); got != 1 { + t.Fatalf("附件取回次数 = %d, want 1", got) + } + if got := s.bodyHits.Load(); got != 0 { + t.Fatalf("正文里的 URL 被取回 %d 次,want 0", got) + } +} + +// TestInboundAdapterUnsupportedFileStaysHonest 钉住「适配器不下载非白名单格式」与 +// 「桥接照既有口径给模型诚实提示」在同一条链路上一致。 +func TestInboundAdapterUnsupportedFileStaysHonest(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + s := newE2EAssetServer(t, "不该被取回的内容\n") + + ad := telegram.New( + telegram.Config{BotToken: "123456:TESTTOKEN"}, + telegram.WithBaseURL(s.srv.URL), + telegram.WithHTTPClient(s.srv.Client()), + telegram.WithLogger(slog.New(slog.NewTextHandler(discardWriter{}, nil))), + ) + var delivered []adapter.UnifiedMessage + ad.SetMessageHandler(func(_ context.Context, msg adapter.UnifiedMessage) { + delivered = append(delivered, msg) + }) + if _, err := ad.ProcessUpdate(context.Background(), e2eUpdate(t, "payload.exe", "跑一下这个")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + env.bridge.process(context.Background(), delivered[0]) + + if names := docDirNames(t, env.uploads); len(names) != 0 { + t.Fatalf("非白名单格式被落库了:%v", names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, "payload.exe") || !strings.Contains(visible, "无法读取") { + t.Fatalf("模型可见视图里没有诚实的拒绝提示:%q", visible) + } + if got := s.docHits.Load() + s.bodyHits.Load(); got != 0 { + t.Fatalf("非白名单格式发起了 %d 次资产请求,want 0", got) + } +} + +// discardWriter 是丢弃输出的 io.Writer(用例不关心适配器日志内容)。 +type discardWriter struct{} + +func (discardWriter) Write(p []byte) (int, error) { return len(p), nil } diff --git a/plugins/channels/kook/media.go b/plugins/channels/kook/media.go new file mode 100644 index 0000000..579be3e --- /dev/null +++ b/plugins/channels/kook/media.go @@ -0,0 +1,52 @@ +package kook + +// 本文件承载 KOOK 的**入站附件字节内联**(切片 IM-1):KOOK 的 s=0 消息把附件地址 +// 直接放在 `content` 字段里(图片/视频/文件/音频同理),因此文件类消息拿到的是一个 +// 绝对 CDN 地址,可以按来源白名单直接取回。 +// +// 两条边界: +// - **只认平台资产主机**:白名单 = KOOK 附件 CDN(`img.kookapp.cn`)+ 已配置的 +// API 根(`base_url`)。`content` 字段是发送方可影响的内容(一个伪造的文件消息 +// 完全可以填任意字符串),因此**不是**「平台保证的地址」——白名单在这里是硬门: +// 不在白名单里,一个字节都不发出去; +// - 尺寸/超时/预筛/日志口径全在 `internal/channel/mediafetch`(四个适配器共用)。 + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch" +) + +// kookAssetOrigin 是 KOOK 附件的 CDN 来源(平台只在它上面提供附件地址)。 +const kookAssetOrigin = "https://img.kookapp.cn" + +// inlineInboundMedia 把受支持格式的文件附件取回并内联到消息上。失败不抛出:该附件 +// 保持无字节,由 channelbridge 的入站摄取照既有口径给模型一行诚实提示。 +func (a *Adapter) inlineInboundMedia(ctx context.Context, msg *adapter.UnifiedMessage) { + mediafetch.Inline(ctx, a.log, msg, kookFetcher{a: a}) +} + +// kookFetcher 实现 mediafetch.Fetcher:KOOK 的媒体引用就是 `content` 里的绝对地址。 +type kookFetcher struct{ a *Adapter } + +// Fetch 按来源白名单取回一份 KOOK 附件。 +func (f kookFetcher) Fetch(ctx context.Context, m adapter.Media) ([]byte, error) { + return mediafetch.Fetch(ctx, mediafetch.Request{ + URL: m.URL, + Allow: f.a.assetAllowlist(), + Client: f.a.client, + Timeout: mediafetch.DefaultTimeout, + }) +} + +// assetAllowlist 给出 KOOK 附件的来源白名单:附件 CDN + 已配置的 API 根。第二条是 +// 为了让**自建/私有化部署**(把 `base_url` 指到自己的网关)也能走通,且它的可信度与 +// 其余 API 调用一致(同一条由部署声明的地址)。两条都读不出来时给空名单 ⇒ 谁都不许。 +func (a *Adapter) assetAllowlist() mediafetch.Allowlist { + origins := []string{kookAssetOrigin} + if origin, err := mediafetch.OriginOf(a.baseURL); err == nil { + origins = append(origins, origin) + } + return mediafetch.NewAllowlist(origins...) +} diff --git a/plugins/channels/kook/media_test.go b/plugins/channels/kook/media_test.go new file mode 100644 index 0000000..27c10d5 --- /dev/null +++ b/plugins/channels/kook/media_test.go @@ -0,0 +1,249 @@ +package kook + +// 本文件锁定 KOOK 的**入站附件字节内联**(切片 IM-1):KOOK 把附件地址放在消息的 +// `content` 字段里,因此这里用假 HTTP 服务端同时充当「资产 CDN」与「已配置的 API 根」, +// 逐条断言正向、超限、预筛、失败与红线(正文里的 URL 绝不被取回)。 + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// kookAssetServer 是假的「平台 + 资产」服务端:`/asset/...` 计文件下载命中次数, +// 其余路径一律 404。 +type kookAssetServer struct { + srv *httptest.Server + hits atomic.Int32 + body []byte + code int +} + +func newKookAssetServer(t *testing.T) *kookAssetServer { + t.Helper() + s := &kookAssetServer{body: []byte("文件的正文\n")} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, "/asset/") { + w.WriteHeader(http.StatusNotFound) + return + } + s.hits.Add(1) + if s.code != 0 { + w.WriteHeader(s.code) + return + } + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write(s.body) + })) + t.Cleanup(s.srv.Close) + return s +} + +// kookCapture 捕获投递给上层的最后一条消息。 +type kookCapture struct { + mu sync.Mutex + last adapter.UnifiedMessage + got int +} + +func (c *kookCapture) fn(_ context.Context, msg adapter.UnifiedMessage) { + c.mu.Lock() + c.last = msg + c.got++ + c.mu.Unlock() +} + +func (c *kookCapture) message(t *testing.T) adapter.UnifiedMessage { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + if c.got == 0 { + t.Fatal("入站消息没有被投递给上层") + } + return c.last +} + +// newInlineKookAdapter 装配一个接到假服务端的适配器(base_url 指到它,因此它同时 +// 是「平台 API 根」与「资产主机」)+ 捕获器 + 日志缓冲。 +func newInlineKookAdapter(t *testing.T, s *kookAssetServer) (*Adapter, *kookCapture, *bytes.Buffer) { + t.Helper() + logs := &bytes.Buffer{} + a := New( + Config{BotToken: "testtoken"}, + WithBaseURL(s.srv.URL), + WithHTTPClient(s.srv.Client()), + WithLogger(logger.WithSource(slog.New(slog.NewTextHandler(logs, nil)), "plugins/channels/kook")), + ) + c := &kookCapture{} + a.SetMessageHandler(c.fn) + return a, c, logs +} + +// kookEventJSON 构造一条 KOOK s=0 事件(type 与 content 由调用方给)。 +func kookEventJSON(kind int, content string) json.RawMessage { + ev := map[string]any{ + "channel_type": "GROUP", "type": kind, "target_id": "chan-1", "author_id": "u-1", + "content": content, "msg_id": "m-1", "msg_timestamp": 1700000000000, "nonce": "", + "extra": map[string]any{"type": kind, "guild_id": "g-1", "channel_name": "c", + "author": map[string]any{"id": "u-1", "username": "u", "nickname": "N"}}, + } + raw, err := json.Marshal(ev) + if err != nil { + panic(fmt.Sprintf("marshal kook event: %v", err)) + } + return raw +} + +// TestInboundFileIsInlined 正向:文件消息(type 4)的 CDN 地址被取回并内联。 +func TestInboundFileIsInlined(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + a, c, _ := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindFile, s.srv.URL+"/asset/report.txt"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + msg := c.message(t) + if len(msg.Media) != 1 || string(msg.Media[0].Data) != "文件的正文\n" { + t.Fatalf("Media.Data = %q, want 下载到的字节", msg.Media[0].Data) + } + if s.hits.Load() != 1 { + t.Fatalf("资产命中 = %d, want 1", s.hits.Load()) + } +} + +// TestInboundFileOffAllowlistIsNotDownloaded 钉住来源白名单:不在白名单里的主机 +// **一个字节都不发出去**(断言 0 次命中)。 +func TestInboundFileOffAllowlistIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + outside := newKookAssetServer(t) + a, c, logs := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindFile, outside.srv.URL+"/asset/report.txt"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + if outside.hits.Load() != 0 { + t.Fatalf("白名单外的主机被请求了 %d 次,want 0", outside.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("白名单外的附件被内联了") + } + if !strings.Contains(logs.String(), "inline failed") || !strings.Contains(logs.String(), "白名单") { + t.Fatalf("缺少带来源的失败日志:%q", logs.String()) + } +} + +// TestInboundUnsupportedExtensionIsNotDownloaded 钉住预筛:非白名单格式不下载。 +func TestInboundUnsupportedExtensionIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + a, c, _ := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindFile, s.srv.URL+"/asset/payload.exe"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + if s.hits.Load() != 0 { + t.Fatalf("非白名单格式发起了 %d 次请求,want 0", s.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("非白名单格式被内联了") + } +} + +// TestInboundOversizeBodyIsCutWhileReading 钉住边读边限(KOOK 事件不给体积,因此 +// 只能靠上限本身兜住)。 +func TestInboundOversizeBodyIsCutWhileReading(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + s.body = bytes.Repeat([]byte("x"), documents.DefaultMaxBytes+64) + a, c, logs := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindFile, s.srv.URL+"/asset/big.txt"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("超量响应被内联了") + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("超量没有留下带来源的失败日志:%q", logs.String()) + } +} + +// TestInboundFetchFailureIsHonest 钉住失败路径:5xx ⇒ 放弃内联,消息照收。 +func TestInboundFetchFailureIsHonest(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + s.code = http.StatusBadGateway + a, c, logs := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindFile, s.srv.URL+"/asset/report.txt"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + msg := c.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatal("失败时仍写入了字节") + } + if msg.ChannelID == "" { + t.Fatal("失败路径把消息弄丢了") + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("失败没有留下带来源的日志:%q", logs.String()) + } +} + +// TestInboundBodyURLIsNeverFetched 是红线用例:正文(text / kmarkdown 的 content) +// 里带任意 URL 时**绝不**对它发起请求(断言 0 次命中)。 +func TestInboundBodyURLIsNeverFetched(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + a, c, _ := newInlineKookAdapter(t, s) + + for _, kind := range []int{kindText, kindKMarkdown} { + if err := a.handleEvent(context.Background(), kookEventJSON(kind, "读一下 "+s.srv.URL+"/asset/secret.txt"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + } + + if s.hits.Load() != 0 { + t.Fatalf("正文里的 URL 被请求了 %d 次,want 0", s.hits.Load()) + } + if c.got == 0 { + t.Fatal("文本消息没有被投递") + } +} + +// TestInboundImageMediaKeepsExistingPath 钉住边界:图片类媒体不走文档内联。 +func TestInboundImageMediaKeepsExistingPath(t *testing.T) { + t.Parallel() + s := newKookAssetServer(t) + a, c, _ := newInlineKookAdapter(t, s) + + if err := a.handleEvent(context.Background(), kookEventJSON(kindImage, s.srv.URL+"/asset/photo.png"), nil); err != nil { + t.Fatalf("handleEvent: %v", err) + } + + if s.hits.Load() != 0 { + t.Fatalf("图片媒体被文档内联取回了 %d 次,want 0", s.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("图片媒体被写入了字节(应留给视觉链路)") + } +} diff --git a/plugins/channels/kook/websocket.go b/plugins/channels/kook/websocket.go index 7623f8c..9e3bc85 100644 --- a/plugins/channels/kook/websocket.go +++ b/plugins/channels/kook/websocket.go @@ -193,8 +193,10 @@ func (a *Adapter) handleEvent(ctx context.Context, data json.RawMessage, frame [ } // dispatch routes an inbound message to the push handler, falling back to the -// ReadMessages sink. +// ReadMessages sink. 附件字节在**路由之前**内联(切片 IM-1):网关 WS 与 webhook +// 两条入站路径共用这一个决策点,下游因此永远拿到「已带字节 / 明确没有字节」的消息。 func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { + a.inlineInboundMedia(ctx, &msg) a.handlerMu.RLock() fn := a.handler a.handlerMu.RUnlock() diff --git a/plugins/channels/qq_obv11/media.go b/plugins/channels/qq_obv11/media.go new file mode 100644 index 0000000..ab2c12d --- /dev/null +++ b/plugins/channels/qq_obv11/media.go @@ -0,0 +1,62 @@ +package qq_obv11 + +// 本文件承载 OneBot v11 的**入站附件字节内联**(切片 IM-1):文件类消息段(数组 +// 形态的 `{type:"file",data:{...}}` 与 CQ 码形态的 `[CQ:file,...]`)里,实现方通常 +// 会给出一个 `url`(离线文件/群文件的下载地址),字节就照它取回。 +// +// 三条边界: +// - **只认平台系资产主机**:白名单 = 腾讯系 CDN(`*.qq.com`、`*.qq.com.cn`、 +// `*.qpic.cn`,覆盖 `multimedia.nt.qq.com.cn` 与 `gchat.qpic.cn`)+ 部署自己 +// 声明的 OneBot API 端点(本地实现常自己下发文件 URL)。消息段里的 `url` 是 +// 对端上报的内容,白名单是硬门:不在名单里一个字节都不发出去; +// - **没有真 url 就不下载**:段里只有文件 id/文件名时(`url` 缺失,`file` 被当作 +// 引用写进了 URL 字段),它连绝对 http(s) 地址都不是 ⇒ 白名单直接否决,绝不 +// 拼凑地址去猜; +// - 尺寸/超时/预筛/日志口径全在 `internal/channel/mediafetch`(四个适配器共用)。 + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch" +) + +// 腾讯系资产主机白名单:OneBot 实现上报的文件地址落在这些子树里(`.qq.com` 家族 +// 覆盖 `multimedia.nt.qq.com.cn` 之外的多媒体域名,`.qpic.cn` 覆盖图片/文件 CDN)。 +// 通配条目**不含**裸的 `qq.com` 本身,因此不会顺带放行 `qq.com` 主页这类无关主机。 +const ( + qqAssetOriginQQ = "https://*.qq.com" + qqAssetOriginQQCN = "https://*.qq.com.cn" + qqAssetOriginQpic = "https://*.qpic.cn" +) + +// inlineInboundMedia 把受支持格式的文件附件取回并内联到消息上。失败不抛出:该附件 +// 保持无字节,由 channelbridge 的入站摄取照既有口径给模型一行诚实提示。 +func (a *Adapter) inlineInboundMedia(ctx context.Context, msg *adapter.UnifiedMessage) { + mediafetch.Inline(ctx, a.log, msg, qqFetcher{a: a}) +} + +// qqFetcher 实现 mediafetch.Fetcher:OneBot 上报的 `url` 就是资产地址。 +type qqFetcher struct{ a *Adapter } + +// Fetch 按来源白名单取回一份文件资产。下载**不复用**出站 HTTP API 的客户端:那个 +// 客户端带着 OneBot access_token 头,绝不能把它发到 CDN 主机上,因此这里交给 +// mediafetch 自建一个干净客户端。 +func (f qqFetcher) Fetch(ctx context.Context, m adapter.Media) ([]byte, error) { + return mediafetch.Fetch(ctx, mediafetch.Request{ + URL: m.URL, + Allow: f.a.assetAllowlist(), + Timeout: mediafetch.DefaultTimeout, + }) +} + +// assetAllowlist 给出 OneBot 文件资产的来源白名单:腾讯系 CDN + 已配置的 OneBot +// API 端点(`api_base_url`,本地部署常是 `http://127.0.0.1:5700` 这类地址 —— 它由 +// 部署显式声明,可信度与其余 API 调用一致)。`api_base_url` 为空时只用腾讯系条目。 +func (a *Adapter) assetAllowlist() mediafetch.Allowlist { + origins := []string{qqAssetOriginQQ, qqAssetOriginQQCN, qqAssetOriginQpic} + if origin, err := mediafetch.OriginOf(a.cfg.APIBaseURL); err == nil { + origins = append(origins, origin) + } + return mediafetch.NewAllowlist(origins...) +} diff --git a/plugins/channels/qq_obv11/media_test.go b/plugins/channels/qq_obv11/media_test.go new file mode 100644 index 0000000..3315ae5 --- /dev/null +++ b/plugins/channels/qq_obv11/media_test.go @@ -0,0 +1,266 @@ +package qq_obv11 + +// 本文件锁定 OneBot v11 的**入站附件字节内联**(切片 IM-1):文件段里实现方给出的 +// `url` 才是可取回的资产地址,白名单限死腾讯系主机 + 部署自己声明的 OneBot 端点。 +// +// 覆盖:正向、超限(边读边限)、非白名单格式不下载、来源不上白名单不下载、失败诚实 +// 放弃、以及红线(正文/文本段里的 URL 绝不被取回)。 + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// qqAssetServer 是假的「OneBot 实现 + 文件服务端」:`/download/...` 计命中次数。 +type qqAssetServer struct { + srv *httptest.Server + hits atomic.Int32 + body []byte + code int +} + +func newQQAssetServer(t *testing.T) *qqAssetServer { + t.Helper() + s := &qqAssetServer{body: []byte("文件正文\n")} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, "/download/") { + w.WriteHeader(http.StatusNotFound) + return + } + s.hits.Add(1) + if s.code != 0 { + w.WriteHeader(s.code) + return + } + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write(s.body) + })) + t.Cleanup(s.srv.Close) + return s +} + +// qqCapture 捕获投递给上层的最后一条消息。 +type qqCapture struct { + mu sync.Mutex + last adapter.UnifiedMessage + count int +} + +func (c *qqCapture) fn(_ context.Context, msg adapter.UnifiedMessage) { + c.mu.Lock() + c.last = msg + c.count++ + c.mu.Unlock() +} + +func (c *qqCapture) message(t *testing.T) adapter.UnifiedMessage { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + if c.count == 0 { + t.Fatal("入站消息没有被投递给上层") + } + return c.last +} + +// newInlineQQAdapter 装配一个把 `api_base_url` 指到假服务端的适配器(该端点因此同时 +// 进白名单,模拟本地 OneBot 实现自己下发文件 URL 的部署形态)+ 捕获器 + 日志缓冲。 +func newInlineQQAdapter(t *testing.T, s *qqAssetServer) (*Adapter, *qqCapture, *bytes.Buffer) { + t.Helper() + logs := &bytes.Buffer{} + a := NewAdapter(Config{ + Mode: "server", + APIBaseURL: s.srv.URL, + Logger: logger.WithSource(slog.New(slog.NewTextHandler(logs, nil)), "plugins/channels/qq_obv11"), + }) + c := &qqCapture{} + a.SetMessageHandler(c.fn) + return a, c, logs +} + +// qqFileEvent 构造一条私聊文件消息(数组形态的消息段)。 +func qqFileEvent(fileName, fileURL string, extraSegments ...any) []byte { + segs := []any{ + map[string]any{"type": "text", "data": map[string]any{"text": "看看这个文件"}}, + map[string]any{"type": "file", "data": map[string]any{"file": fileName, "url": fileURL}}, + } + segs = append(segs, extraSegments...) + ev := map[string]any{ + "post_type": "message", "message_type": "private", "time": 1700000000, + "self_id": 100, "message_id": 7, "user_id": 200, + "message": segs, + "sender": map[string]any{"user_id": 200, "nickname": "Alice", "role": ""}, + } + raw, err := json.Marshal(ev) + if err != nil { + panic(fmt.Sprintf("marshal ob event: %v", err)) + } + return raw +} + +// TestInboundFileIsInlined 正向:文件段的 `url` 被取回并内联。 +func TestInboundFileIsInlined(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + a, c, _ := newInlineQQAdapter(t, s) + + a.handlePayload(context.Background(), qqFileEvent("report.txt", s.srv.URL+"/download/report.txt")) + + msg := c.message(t) + if len(msg.Media) != 1 || string(msg.Media[0].Data) != "文件正文\n" { + t.Fatalf("Media.Data = %q, want 下载到的字节", msg.Media[0].Data) + } + if s.hits.Load() != 1 { + t.Fatalf("资产命中 = %d, want 1", s.hits.Load()) + } +} + +// TestInboundCQCodeFileIsInlined 正向(CQ 码形态):`[CQ:file,file=…,url=…]` 同样被内联。 +func TestInboundCQCodeFileIsInlined(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + a, c, _ := newInlineQQAdapter(t, s) + ev := map[string]any{ + "post_type": "message", "message_type": "private", "time": 1700000000, + "self_id": 100, "message_id": 8, "user_id": 200, + "message": "[CQ:file,file=report.txt,url=" + s.srv.URL + "/download/report.txt]", + "sender": map[string]any{"user_id": 200, "nickname": "Alice", "role": ""}, + } + raw, _ := json.Marshal(ev) + + a.handlePayload(context.Background(), raw) + + if string(c.message(t).Media[0].Data) != "文件正文\n" { + t.Fatalf("CQ 码形态的文件未被内联") + } +} + +// TestInboundFileOffAllowlistIsNotDownloaded 钉住来源白名单:不在白名单里的主机 +// **一个字节都不发出去**。 +func TestInboundFileOffAllowlistIsNotDownloaded(t *testing.T) { + t.Parallel() + inside := newQQAssetServer(t) + outside := newQQAssetServer(t) + a, c, logs := newInlineQQAdapter(t, inside) + + a.handlePayload(context.Background(), qqFileEvent("report.txt", outside.srv.URL+"/download/report.txt")) + + if outside.hits.Load() != 0 { + t.Fatalf("白名单外的主机被请求了 %d 次,want 0", outside.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("白名单外的附件被内联了") + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("缺少带来源的失败日志:%q", logs.String()) + } +} + +// TestInboundFileWithoutURLIsNotDownloaded 钉住「没有真 url 就不下载」:段里只有文件 +// id/文件名时(`url` 缺失,`file` 被当作引用),绝不拼凑地址去猜,也绝不发请求。 +func TestInboundFileWithoutURLIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + a, c, _ := newInlineQQAdapter(t, s) + + a.handlePayload(context.Background(), qqFileEvent("report.txt", "")) + + if s.hits.Load() != 0 { + t.Fatalf("无 url 的文件段发起了 %d 次请求,want 0", s.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("无 url 的文件段被内联了") + } +} + +// TestInboundUnsupportedExtensionIsNotDownloaded 钉住预筛:非白名单格式不下载。 +func TestInboundUnsupportedExtensionIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + a, c, _ := newInlineQQAdapter(t, s) + + a.handlePayload(context.Background(), qqFileEvent("payload.exe", s.srv.URL+"/download/payload.exe")) + + if s.hits.Load() != 0 { + t.Fatalf("非白名单格式发起了 %d 次请求,want 0", s.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("非白名单格式被内联了") + } +} + +// TestInboundOversizeBodyIsCutWhileReading 钉住边读边限:OneBot 段不给体积,只能靠 +// 上限本身兜住超量响应。 +func TestInboundOversizeBodyIsCutWhileReading(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + s.body = bytes.Repeat([]byte("x"), documents.DefaultMaxBytes+64) + a, c, logs := newInlineQQAdapter(t, s) + + a.handlePayload(context.Background(), qqFileEvent("big.txt", s.srv.URL+"/download/big.txt")) + + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("超量响应被内联了") + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("超量没有留下带来源的失败日志:%q", logs.String()) + } +} + +// TestInboundFetchFailureIsHonest 钉住失败路径:5xx ⇒ 放弃内联,消息照收。 +func TestInboundFetchFailureIsHonest(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + s.code = http.StatusInternalServerError + a, c, logs := newInlineQQAdapter(t, s) + + a.handlePayload(context.Background(), qqFileEvent("report.txt", s.srv.URL+"/download/report.txt")) + + msg := c.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatal("失败时仍写入了字节") + } + if msg.MessageID != "7" { + t.Fatalf("失败路径把消息弄丢了:%+v", msg.MessageID) + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("失败没有留下带来源的日志:%q", logs.String()) + } +} + +// TestInboundBodyURLIsNeverFetched 是红线用例:文本段(消息正文)里带任意 URL 时 +// **绝不**对它发起请求(断言 0 次命中)。 +func TestInboundBodyURLIsNeverFetched(t *testing.T) { + t.Parallel() + s := newQQAssetServer(t) + a, _, _ := newInlineQQAdapter(t, s) + bodyURL := s.srv.URL + "/download/secret.txt" + ev := map[string]any{ + "post_type": "message", "message_type": "private", "time": 1700000000, + "self_id": 100, "message_id": 9, "user_id": 200, + "message": []any{ + map[string]any{"type": "text", "data": map[string]any{"text": "读一下 " + bodyURL}}, + }, + "sender": map[string]any{"user_id": 200, "nickname": "Alice", "role": ""}, + } + raw, _ := json.Marshal(ev) + + a.handlePayload(context.Background(), raw) + + if s.hits.Load() != 0 { + t.Fatalf("正文里的 URL 被请求了 %d 次,want 0", s.hits.Load()) + } +} diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go index 2d6f74b..8097130 100644 --- a/plugins/channels/qq_obv11/session.go +++ b/plugins/channels/qq_obv11/session.go @@ -192,8 +192,11 @@ func (a *Adapter) handlePayload(ctx context.Context, payload []byte) { } // dispatch routes an inbound message to the registered push handler, falling -// back to the ReadMessages sink. +// back to the ReadMessages sink. 附件字节在**路由之前**内联(切片 IM-1):WS 两端 +// 与 HTTP 上报三条入站路径都经这里,下游因此永远拿到「已带字节 / 明确没有字节」的 +// 消息;失败只留日志,绝不挡住投递。 func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { + a.inlineInboundMedia(ctx, &msg) a.handlerMu.RLock() fn := a.handler a.handlerMu.RUnlock() diff --git a/plugins/channels/telegram/adapter.go b/plugins/channels/telegram/adapter.go index 63e0d4b..7939675 100644 --- a/plugins/channels/telegram/adapter.go +++ b/plugins/channels/telegram/adapter.go @@ -213,6 +213,9 @@ func (a *Adapter) ProcessUpdate(ctx context.Context, raw []byte) (int, error) { if err := msg.Validate(); err != nil { return 0, fmt.Errorf("telegram: validate webhook update: %w", err) } + // 附件字节在**投递之前**内联:webhook 与长轮询两条入站路径共用同一口径 + // (切片 IM-1)——下游因此拿到的永远是一份已经带字节(或明确没有字节)的消息。 + a.inlineInboundMedia(ctx, &msg) a.mu.RLock() handler := a.handler a.mu.RUnlock() diff --git a/plugins/channels/telegram/errors.go b/plugins/channels/telegram/errors.go index 0728abd..aea6e5a 100644 --- a/plugins/channels/telegram/errors.go +++ b/plugins/channels/telegram/errors.go @@ -14,4 +14,7 @@ var ( // ErrNoChatTarget is returned when an outbound message carries no usable // conversation/group identifier. ErrNoChatTarget = errors.New("telegram: message has no target chat") + // ErrNoFilePath is returned when getFile yields no usable file_path (empty + // or a path shape we refuse to splice into a download URL). + ErrNoFilePath = errors.New("telegram: file path unavailable") ) diff --git a/plugins/channels/telegram/media.go b/plugins/channels/telegram/media.go new file mode 100644 index 0000000..f0de588 --- /dev/null +++ b/plugins/channels/telegram/media.go @@ -0,0 +1,107 @@ +package telegram + +// 本文件承载 Telegram 的**入站附件字节内联**(切片 IM-1):Bot API 上报的文件类 +// 媒体只给 `file_id`,字节要分两步取回 —— 先 `getFile` 把平台文件 id 换成一个 +// **平台自己返回**的相对路径,再把该路径拼到已配置的 API 根之后下载。 +// +// 三条边界: +// - 地址**完全由配置与平台返回值构造**:调用方(发送方)无法让请求打到别的主机; +// 平台返回的路径还要过 `sanitizeFilePath`(拒绝对协议前缀、`..`、查询串); +// - 来源白名单 = 已配置的 API 根(默认 `https://api.telegram.org`):白名单之外的 +// 主机**一个字节都不发出去**; +// - 尺寸/超时/预筛/日志口径全在 `internal/channel/mediafetch`(四个适配器共用), +// 这里只负责「把平台引用换成一个平台资产地址」。 + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch" +) + +// getFileMethod 是 Bot API 里「用 file_id 换文件路径」的方法名。 +const getFileMethod = "getFile" + +// inlineInboundMedia 在消息投递给上层之前,把受支持格式的文件附件取回并内联。 +// 失败绝不抛出:该附件保持无字节,由 channelbridge 的入站摄取照既有口径给模型 +// 一行诚实提示(design/24 §4.4)。 +func (a *Adapter) inlineInboundMedia(ctx context.Context, msg *adapter.UnifiedMessage) { + mediafetch.Inline(ctx, a.log, msg, telegramFetcher{a: a}) +} + +// telegramFetcher 实现 mediafetch.Fetcher:Telegram 给的是 `file_id`,因此它必须 +// 先问平台「这个 id 对应哪个路径」,再下载。 +type telegramFetcher struct{ a *Adapter } + +// Fetch 取回一份 Telegram 附件的字节。体积上限与超时都不在这里写数:前者由 +// mediafetch 的默认值(`documents.DefaultMaxBytes`)兜底,后者是具名常量 +// `mediafetch.DefaultTimeout`。 +func (f telegramFetcher) Fetch(ctx context.Context, m adapter.Media) ([]byte, error) { + filePath, err := f.a.filePath(ctx, m.URL) + if err != nil { + return nil, err + } + return mediafetch.Fetch(ctx, mediafetch.Request{ + URL: f.a.fileURL(filePath), + Allow: f.a.assetAllowlist(), + Client: f.a.client, + Timeout: mediafetch.DefaultTimeout, + }) +} + +// assetAllowlist 给出文件下载的来源白名单:**只有已配置的 Bot API 根**。文件地址 +// 由「已配置的根 + 平台返回的相对路径」拼成,因此它不可能指向别的主机;白名单再把 +// 这条性质钉死在代码里(配置读不出根时给空名单 ⇒ 谁都不许,deny-by-default)。 +func (a *Adapter) assetAllowlist() mediafetch.Allowlist { + origin, err := mediafetch.OriginOf(a.baseURL) + if err != nil { + return mediafetch.NewAllowlist() + } + return mediafetch.NewAllowlist(origin) +} + +// fileURL 把平台给出的相对路径拼成下载地址。token 在路径里(与 endpoint 同口径), +// 绝不进查询串;这个地址**只用于请求**,绝不进日志(mediafetch 的错误只带来源)。 +func (a *Adapter) fileURL(filePath string) string { + return a.baseURL + "/file/bot" + a.cfg.BotToken + "/" + strings.TrimLeft(filePath, "/") +} + +// filePath 调 getFile 把平台文件 id 换成平台自己返回的相对路径,并做形态校验。 +func (a *Adapter) filePath(ctx context.Context, fileID string) (string, error) { + if strings.TrimSpace(fileID) == "" { + return "", ErrNoFilePath + } + out, err := a.doJSON(ctx, getFileMethod, map[string]any{"file_id": fileID}) + if err != nil { + return "", err + } + var res struct { + FilePath string `json:"file_path"` + } + if err := json.Unmarshal(out.Result, &res); err != nil { + return "", fmt.Errorf("telegram: getFile result: %w", err) + } + return sanitizeFilePath(res.FilePath) +} + +// sanitizeFilePath 校验平台返回的文件路径:非空、不含协议前缀 / `..` / 查询串 / +// 控制字符。校验失败时返回 ErrNoFilePath —— 绝不把可疑路径拼进下载地址。 +func sanitizeFilePath(filePath string) (string, error) { + path := strings.TrimSpace(filePath) + if path == "" { + return "", ErrNoFilePath + } + for _, r := range path { + if r < ' ' || r == 0x7f { + return "", fmt.Errorf("%w: 平台返回的路径含控制字符", ErrNoFilePath) + } + } + if strings.Contains(path, "://") || strings.Contains(path, "..") || + strings.ContainsAny(path, "\\?#") { + return "", fmt.Errorf("%w: 平台返回的路径形态非法", ErrNoFilePath) + } + return path, nil +} diff --git a/plugins/channels/telegram/media_test.go b/plugins/channels/telegram/media_test.go new file mode 100644 index 0000000..18314f6 --- /dev/null +++ b/plugins/channels/telegram/media_test.go @@ -0,0 +1,316 @@ +package telegram + +// 本文件锁定 Telegram 的**入站附件字节内联**(切片 IM-1): +// +// - 正向:平台给 file_id ⇒ getFile 换路径 ⇒ 下载字节内联(假 HTTP 服务端,不碰真平台); +// - 超限:声明超限不发请求;未声明而服务端超量 ⇒ 边读边限、放弃内联、日志与提示都不静默; +// - 非白名单格式 ⇒ **不发起下载**(连 getFile 都不调,断言 0 次命中); +// - 超时/失败 ⇒ 诚实放弃(消息照收); +// - **红线**:正文里的任意 URL 绝不被取回(断言服务端 0 次命中)。 + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// tgAssetServer 是假的 Bot API + 文件服务端:它区分 getFile 调用与真正的文件下载, +// 使「0 次命中」可以精确断言在**文件下载**这件事上。 +type tgAssetServer struct { + srv *httptest.Server + getFile atomic.Int32 + fileHits atomic.Int32 + allHits atomic.Int32 + fileBody []byte + fileCode int + fileDelay bool +} + +func newTGAssetServer(t *testing.T) *tgAssetServer { + t.Helper() + s := &tgAssetServer{fileBody: []byte("附件的正文\n")} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.allHits.Add(1) + switch { + case strings.HasSuffix(r.URL.Path, "/getFile"): + s.getFile.Add(1) + okResponse(w, map[string]any{"file_id": "doc-id", "file_path": "documents/file_1.txt"}) + case strings.Contains(r.URL.Path, "/file/bot"): + s.fileHits.Add(1) + if s.fileCode != 0 { + w.WriteHeader(s.fileCode) + return + } + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write(s.fileBody) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(s.srv.Close) + return s +} + +// captureHandler 捕获投递给上层的最后一条消息。 +type captureHandler struct { + mu sync.Mutex + last adapter.UnifiedMessage + got int +} + +func (h *captureHandler) fn(_ context.Context, msg adapter.UnifiedMessage) { + h.mu.Lock() + h.last = msg + h.got++ + h.mu.Unlock() +} + +func (h *captureHandler) message(t *testing.T) adapter.UnifiedMessage { + t.Helper() + h.mu.Lock() + defer h.mu.Unlock() + if h.got == 0 { + t.Fatal("附件消息没有被投递给上层") + } + return h.last +} + +// newInlineAdapter 装配一个接到假服务端的适配器 + 捕获器 + 日志缓冲。 +func newInlineAdapter(t *testing.T, s *tgAssetServer) (*Adapter, *captureHandler, *bytes.Buffer) { + t.Helper() + logs := &bytes.Buffer{} + a := New( + Config{BotToken: "123456:TESTTOKEN"}, + WithBaseURL(s.srv.URL), + WithHTTPClient(s.srv.Client()), + WithLogger(logger.WithSource(slog.New(slog.NewTextHandler(logs, nil)), "plugins/channels/telegram")), + ) + h := &captureHandler{} + a.SetMessageHandler(h.fn) + return a, h, logs +} + +// documentUpdate 构造一条 Telegram 文档消息的原始 update。 +func documentUpdate(name string, size int64, text string) []byte { + body := map[string]any{ + "update_id": 1, + "message": map[string]any{ + "message_id": 42, + "date": 1700000000, + "chat": map[string]any{"id": 7, "type": "private"}, + "from": map[string]any{"id": 9, "first_name": "主人"}, + "text": text, + "document": map[string]any{ + "file_id": "doc-id", "file_name": name, "file_size": size, "mime_type": "text/plain", + }, + }, + } + raw, err := json.Marshal(body) + if err != nil { + panic(fmt.Sprintf("marshal update: %v", err)) + } + return raw +} + +// TestInboundDocumentIsInlined 是正向用例:file_id ⇒ getFile ⇒ 下载 ⇒ Data 被填、 +// 且消息照常投递(Data 之外的字段一字不动)。 +func TestInboundDocumentIsInlined(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + a, h, _ := newInlineAdapter(t, s) + + if _, err := a.ProcessUpdate(context.Background(), documentUpdate("report.txt", 12, "看看这个")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media) != 1 || string(msg.Media[0].Data) != "附件的正文\n" { + t.Fatalf("Media.Data = %q, want 下载到的字节", msg.Media[0].Data) + } + if msg.Media[0].Name != "report.txt" || msg.Media[0].URL != "doc-id" { + t.Fatalf("内联改动了媒体元数据:%+v", msg.Media[0]) + } + if s.getFile.Load() != 1 || s.fileHits.Load() != 1 { + t.Fatalf("getFile=%d fileHits=%d, want 1/1", s.getFile.Load(), s.fileHits.Load()) + } +} + +// TestInboundDeclaredOversizeIsNotDownloaded 钉住「平台声明超限 ⇒ 连请求都不发」。 +func TestInboundDeclaredOversizeIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + a, h, logs := newInlineAdapter(t, s) + + _, err := a.ProcessUpdate(context.Background(), documentUpdate("big.txt", documents.DefaultMaxBytes+1, "超大文件")) + if err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatalf("超限附件被内联了(%d 字节)", len(msg.Media[0].Data)) + } + if s.getFile.Load() != 0 || s.allHits.Load() != 0 { + t.Fatalf("声明超限仍发起了请求:getFile=%d all=%d", s.getFile.Load(), s.allHits.Load()) + } + if !strings.Contains(logs.String(), "declared size over limit") { + t.Fatalf("超限没有留下带来源的日志:%q", logs.String()) + } +} + +// TestInboundOversizeBodyIsCutWhileReading 钉住「边读边限」:平台声明的体积不可信 +// (这里声明为 1 字节),服务端实际吐出超限字节时也要放弃。 +func TestInboundOversizeBodyIsCutWhileReading(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + s.fileBody = bytes.Repeat([]byte("x"), documents.DefaultMaxBytes+64) + a, h, logs := newInlineAdapter(t, s) + + if _, err := a.ProcessUpdate(context.Background(), documentUpdate("big.txt", 1, "超量")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatalf("超量响应被内联了(%d 字节)", len(msg.Media[0].Data)) + } + if s.fileHits.Load() != 1 { + t.Fatalf("fileHits = %d, want 1(这次请求本来就该发)", s.fileHits.Load()) + } + if !strings.Contains(logs.String(), "inline failed") || !strings.Contains(logs.String(), "体积上限") { + t.Fatalf("超量没有留下带来源的失败日志:%q", logs.String()) + } +} + +// TestInboundUnsupportedExtensionIsNotDownloaded 钉住预筛:非白名单格式**不发起下载** +// (连 getFile 都不调),消息照收。 +func TestInboundUnsupportedExtensionIsNotDownloaded(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + a, h, _ := newInlineAdapter(t, s) + + if _, err := a.ProcessUpdate(context.Background(), documentUpdate("payload.exe", 1024, "跑一下")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatalf("非白名单格式被内联了") + } + if s.allHits.Load() != 0 { + t.Fatalf("非白名单格式发起了 %d 次请求,want 0", s.allHits.Load()) + } +} + +// TestInboundFetchFailureIsHonest 钉住失败路径:服务端 5xx / 连接失败 ⇒ 放弃内联、 +// 消息照收、日志带来源(且不含 URL 原串与 token)。 +func TestInboundFetchFailureIsHonest(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + s.fileCode = http.StatusInternalServerError + a, h, logs := newInlineAdapter(t, s) + + if _, err := a.ProcessUpdate(context.Background(), documentUpdate("report.txt", 12, "读一下")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatalf("失败时仍写入了字节") + } + if !strings.Contains(msg.Content, "读一下") { + t.Fatalf("失败路径改动了正文:%q", msg.Content) + } + log := logs.String() + if !strings.Contains(log, "inline failed") { + t.Fatalf("失败没有留下带来源的日志:%q", log) + } + if strings.Contains(log, "TESTTOKEN") || strings.Contains(log, "TESTTOKEN") { + t.Fatalf("日志泄露了 token:%q", log) + } +} + +// TestInboundFetchHonorsContextCancel 钉住超时接线:父 ctx 取消时下载立刻放弃 +// (不必等 30s 的具名超时),消息照收。 +func TestInboundFetchHonorsContextCancel(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + a, h, logs := newInlineAdapter(t, s) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := a.ProcessUpdate(ctx, documentUpdate("report.txt", 12, "读一下")); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + msg := h.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatalf("ctx 已取消却仍内联了字节") + } + if !strings.Contains(logs.String(), "inline failed") { + t.Fatalf("ctx 取消没有留下带来源的日志:%q", logs.String()) + } +} + +// TestInboundBodyURLIsNeverFetched 是本切片的红线用例:正文里带任意 URL 的消息 +// **绝不**对那个 URL 发起请求(断言服务端 0 次命中)。 +func TestInboundBodyURLIsNeverFetched(t *testing.T) { + t.Parallel() + s := newTGAssetServer(t) + a, _, _ := newInlineAdapter(t, s) + // 正文直接点名一个「附件地址」形态的 URL(服务端确实能应答它)。 + body := "帮我读一下 " + s.srv.URL + "/file/bot123456:TESTTOKEN/documents/file_1.txt 这份文件" + + if _, err := a.ProcessUpdate(context.Background(), bodyOnlyUpdate(body)); err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + + if s.allHits.Load() != 0 { + t.Fatalf("正文里的 URL 被请求了 %d 次,want 0", s.allHits.Load()) + } +} + +// bodyOnlyUpdate 构造一条**只有正文、没有任何媒体**的消息。 +func bodyOnlyUpdate(text string) []byte { + body := map[string]any{ + "update_id": 2, + "message": map[string]any{ + "message_id": 43, + "date": 1700000001, + "chat": map[string]any{"id": 7, "type": "private"}, + "from": map[string]any{"id": 9, "first_name": "主人"}, + "text": text, + }, + } + raw, err := json.Marshal(body) + if err != nil { + panic(fmt.Sprintf("marshal update: %v", err)) + } + return raw +} + +// TestFilePathSanitization 钉住平台返回路径的形态校验:可疑路径一律拒绝,绝不拼进 +// 下载地址。 +func TestFilePathSanitization(t *testing.T) { + t.Parallel() + for _, bad := range []string{"", " ", "https://evil.example.com/x", "../../etc/passwd", "a\\b.txt", "x.txt?q=1", "x\n.txt"} { + if _, err := sanitizeFilePath(bad); err == nil { + t.Fatalf("sanitizeFilePath(%q) = nil, want error", bad) + } + } + if got, err := sanitizeFilePath("documents/file_1.txt"); err != nil || got != "documents/file_1.txt" { + t.Fatalf("sanitizeFilePath(合法路径) = %q, %v", got, err) + } +} diff --git a/plugins/channels/telegram/poll.go b/plugins/channels/telegram/poll.go index bc27d2e..c70b20e 100644 --- a/plugins/channels/telegram/poll.go +++ b/plugins/channels/telegram/poll.go @@ -71,6 +71,7 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes a.log.Warn("skip invalid message", "err", err) continue } + a.inlineInboundMedia(pollCtx, &msg) emit(msg) } a.mu.Lock() diff --git a/plugins/channels/weixin_ilink/adapter_http.go b/plugins/channels/weixin_ilink/adapter_http.go index 5b503d2..b93c0ab 100644 --- a/plugins/channels/weixin_ilink/adapter_http.go +++ b/plugins/channels/weixin_ilink/adapter_http.go @@ -18,20 +18,36 @@ import ( "strings" ) +// ilinkAuthType 是 iLink 请求头 `AuthorizationType` 的协议定值,仅此一处定义。 +const ilinkAuthType = "ilink_bot_token" + // authorizationHeader 为 req 装上 iLink 通用请求头:JSON 内容类型、bearer // token、认证类型、每次新生成的随机 X-WECHAT-UIN 防重放值,以及 App 标识头。 func (a *Adapter) authorizationHeader(req *http.Request) { - req.Header.Set("Content-Type", "application/json") - req.Header.Set("AuthorizationType", "ilink_bot_token") - req.Header.Set("X-WECHAT-UIN", randomUIN()) - req.Header.Set("iLink-App-Id", ilinkAppID) - req.Header.Set("iLink-App-ClientVersion", buildClientVersion(ilinkVersion)) + for k, vs := range a.inboundHeaders() { + for _, v := range vs { + req.Header.Set(k, v) + } + } +} + +// inboundHeaders 构造 iLink 通用请求头(JSON 内容类型、bearer token、认证类型、随机 +// 防重放值、App 标识头)。API 调用与**入站附件取回**共用同一份口径(切片 IM-1)—— +// CDN 与 API 同属 iLink 平台,凭据与标识没有任何区别。 +func (a *Adapter) inboundHeaders() http.Header { + h := http.Header{} + h.Set("Content-Type", "application/json") + h.Set("AuthorizationType", ilinkAuthType) + h.Set("X-WECHAT-UIN", randomUIN()) + h.Set("iLink-App-Id", ilinkAppID) + h.Set("iLink-App-ClientVersion", buildClientVersion(ilinkVersion)) a.mu.RLock() token := a.cfg.BotToken a.mu.RUnlock() if token != "" { - req.Header.Set("Authorization", "Bearer "+token) + h.Set("Authorization", "Bearer "+token) } + return h } // randomUIN 返回 base64 编码的十进制 uint32,每次请求都新生成,作为 iLink 的 diff --git a/plugins/channels/weixin_ilink/media.go b/plugins/channels/weixin_ilink/media.go index abdccaf..7981ceb 100644 --- a/plugins/channels/weixin_ilink/media.go +++ b/plugins/channels/weixin_ilink/media.go @@ -9,6 +9,9 @@ import ( "io" "net/http" "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch" ) // ErrKeyTooShort 在入站密钥推导无法给出所需分组长度的 AES 密钥时返回。 @@ -90,3 +93,53 @@ func (a *Adapter) fetchBody(ctx context.Context, rawURL string) ([]byte, error) } return data, nil } + +// inlineInboundMedia 把受支持格式的文件附件取回、解密并内联到消息上(切片 IM-1)。 +// 失败不抛出:该附件保持无字节,由 channelbridge 的入站摄取照既有口径给模型一行 +// 诚实提示(design/24 §4.4)——本切片绝不为了「让文件能读」而把失败吞掉。 +func (a *Adapter) inlineInboundMedia(ctx context.Context, msg *adapter.UnifiedMessage) { + mediafetch.Inline(ctx, a.log, msg, ilinkFetcher{a: a}) +} + +// ilinkFetcher 实现 mediafetch.Fetcher:平台给的是 CDN **相对引用**(少数时候是绝对 +// URL),两者都先经 `cdnURL` 归一,再按来源白名单取回,最后剥掉 AES-ECB 层 —— +// 与既有 `DownloadMedia` 同一条平台口径,只是多了一层白名单与独立的体积上限。 +type ilinkFetcher struct{ a *Adapter } + +// Fetch 取回并解密一份微信 iLink 附件。体积上限与超时都不在这里写数:前者由 +// mediafetch 的默认值(`documents.DefaultMaxBytes`)兜底,后者是具名常量 +// `mediafetch.DefaultTimeout`(**不是** `MaxMediaSize`:那是媒体通道自己的历史上限, +// 与本切片的文档上限刻意解耦)。 +func (f ilinkFetcher) Fetch(ctx context.Context, m adapter.Media) ([]byte, error) { + key, err := deriveInboundKey(f.a.cfg.BotToken) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: derive inbound key: %w", err) + } + ciphertext, err := mediafetch.Fetch(ctx, mediafetch.Request{ + URL: f.a.cdnURL(m.URL), + Allow: f.a.assetAllowlist(), + Header: f.a.inboundHeaders(), + Client: f.a.httpClient, + Timeout: mediafetch.DefaultTimeout, + }) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: fetch media: %w", err) + } + plaintext, err := aesECBDecrypt(key, ciphertext) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: decrypt media: %w", err) + } + return plaintext, nil +} + +// assetAllowlist 给出入站附件取回的来源白名单:**只有已配置的 CDN 根**(默认 +// `https://novac2c.cdn.weixin.qq.com/c2c` 的来源)。`cdnURL` 只把平台给的相对引用 +// 拼到它之后,但**绝对 URL 会被原样放行**(见 TestCDNURL_JoinsRelativeAndPassesAbsolute), +// 因此白名单是这里唯一能挡住「对端伪造绝对引用把我们引到别的主机」的门。 +func (a *Adapter) assetAllowlist() mediafetch.Allowlist { + origin, err := mediafetch.OriginOf(a.cfg.CDNBaseURL) + if err != nil { + return mediafetch.NewAllowlist() + } + return mediafetch.NewAllowlist(origin) +} diff --git a/plugins/channels/weixin_ilink/media_inline_test.go b/plugins/channels/weixin_ilink/media_inline_test.go new file mode 100644 index 0000000..1cb2e39 --- /dev/null +++ b/plugins/channels/weixin_ilink/media_inline_test.go @@ -0,0 +1,269 @@ +package weixin_ilink + +// 本文件锁定微信 iLink 的**入站附件字节内联**(切片 IM-1):CDN 引用 → 白名单取回 → +// 剥 AES-ECB 层 → 内联。覆盖正向、来源不上白名单不下载、非白名单格式不下载、超限、 +// 解密失败诚实放弃,以及红线(正文里的 URL 绝不被取回)。 + +import ( + "bytes" + "context" + "crypto/aes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// ilinkAssetServer 是假的 CDN:`/c2c/...` 计命中次数,并记录请求头(用于断言凭据与 +// 标识头确实随取回发出)。 +type ilinkAssetServer struct { + srv *httptest.Server + hits atomic.Int32 + body []byte + code int + auth atomic.Value +} + +func newIlinkAssetServer(t *testing.T, body []byte) *ilinkAssetServer { + t.Helper() + s := &ilinkAssetServer{body: body} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.hits.Add(1) + s.auth.Store(r.Header.Get("Authorization")) + if s.code != 0 { + w.WriteHeader(s.code) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(s.body) + })) + t.Cleanup(s.srv.Close) + return s +} + +// fileItemRaw 构造一条带文件 item 的原始 iLink 消息。 +func fileItemRaw(ref, name string) []byte { + raw, err := json.Marshal(map[string]any{ + "client_id": "c-1", "from_user_id": "u-1", "message_type": msgTypeUser, + "item_list": []any{ + map[string]any{"type": itemText, "text_item": map[string]any{"text": "看看这个文件"}}, + map[string]any{"type": itemFile, "file_item": map[string]any{ + "file_name": name, "url": ref, "size": 1024, "mime_type": "text/plain", + }}, + }, + }) + if err != nil { + panic(err.Error()) + } + return raw +} + +// textItemRaw 构造一条只有文本 item 的原始消息(正文里带 URL)。 +func textItemRaw(text string) []byte { + raw, err := json.Marshal(map[string]any{ + "client_id": "c-2", "from_user_id": "u-1", "message_type": msgTypeUser, + "item_list": []any{map[string]any{"type": itemText, "text_item": map[string]any{"text": text}}}, + }) + if err != nil { + panic(err.Error()) + } + return raw +} + +// ilinkInlineCapture 捕获 deliver 投递出来的消息。 +type ilinkInlineCapture struct { + mu sync.Mutex + last adapter.UnifiedMessage + got int +} + +func (c *ilinkInlineCapture) fn(_ context.Context, msg adapter.UnifiedMessage) { + c.mu.Lock() + c.last = msg + c.got++ + c.mu.Unlock() +} + +func (c *ilinkInlineCapture) message(t *testing.T) adapter.UnifiedMessage { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + if c.got == 0 { + t.Fatal("入站消息没有被投递出来") + } + return c.last +} + +// newInlineIlinkAdapter 装配一个「CDN 根指向假服务端」的适配器与捕获器。 +func newInlineIlinkAdapter(t *testing.T, s *ilinkAssetServer, key []byte) (*Adapter, *ilinkInlineCapture) { + t.Helper() + cfg := Config{BaseURL: s.srv.URL, CDNBaseURL: s.srv.URL, BotToken: encodeToken(key)} + a := newAdapter(t, cfg, s.srv) + c := &ilinkInlineCapture{} + a.SetMessageHandler(c.fn) + return a, c +} + +// keyOf 返回确定性的 AES 密钥(每个用例自己给种子)。 +func keyOf(seed byte) []byte { return bytes.Repeat([]byte{seed}, aes.BlockSize) } + +// encrypted 用 key 加密 plain,供假 CDN 下发。 +func encrypted(t *testing.T, key, plain []byte) []byte { + t.Helper() + out, err := aesECBEncrypt(key, plain) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + return out +} + +// TestInboundFileIsInlinedAndDecrypted 正向:CDN 引用 → 白名单取回 → 解密 → 内联。 +func TestInboundFileIsInlinedAndDecrypted(t *testing.T) { + t.Parallel() + key := keyOf(0x31) + plain := []byte("微信文件正文\n") + s := newIlinkAssetServer(t, encrypted(t, key, plain)) + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(fileItemRaw("media/1.txt", "report.txt")) + if m == nil { + t.Fatal("convertMessage 没有解析出消息") + } + a.deliver(context.Background(), *m, nil) + + msg := c.message(t) + if len(msg.Media) != 1 || string(msg.Media[0].Data) != string(plain) { + t.Fatalf("Media.Data = %q, want 解密后的正文", msg.Media[0].Data) + } + if s.hits.Load() != 1 { + t.Fatalf("CDN 命中 = %d, want 1", s.hits.Load()) + } + if got, _ := s.auth.Load().(string); !strings.HasPrefix(got, "Bearer ") { + t.Fatalf("CDN 取回没有带 bearer 凭据:%q", got) + } +} + +// TestInboundAbsoluteRefOffAllowlistIsNotDownloaded 钉住来源白名单:`cdnURL` 对**绝对 +// URL 原样放行**,因此对端伪造一个指向别处的主机时,唯一能挡住它的就是白名单 +// (断言那台主机 0 次命中)。 +func TestInboundAbsoluteRefOffAllowlistIsNotDownloaded(t *testing.T) { + t.Parallel() + key := keyOf(0x32) + inside := newIlinkAssetServer(t, encrypted(t, key, []byte("ok\n"))) + outside := newIlinkAssetServer(t, encrypted(t, key, []byte("evil\n"))) + a, c := newInlineIlinkAdapter(t, inside, key) + + _, m := a.convertMessage(fileItemRaw(outside.srv.URL+"/c2c/evil.txt", "report.txt")) + a.deliver(context.Background(), *m, nil) + + if outside.hits.Load() != 0 { + t.Fatalf("白名单外的绝对引用被请求了 %d 次,want 0", outside.hits.Load()) + } + if inside.hits.Load() != 0 { + t.Fatalf("白名单内的主机不该收到这次请求(引用指向别处):%d 次", inside.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("白名单外的附件被内联了") + } +} + +// TestInboundUnsupportedExtensionIsNotDownloaded 钉住预筛:非白名单格式不下载。 +func TestInboundUnsupportedExtensionIsNotDownloaded(t *testing.T) { + t.Parallel() + key := keyOf(0x33) + s := newIlinkAssetServer(t, encrypted(t, key, []byte("x"))) + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(fileItemRaw("media/1.exe", "payload.exe")) + a.deliver(context.Background(), *m, nil) + + if s.hits.Load() != 0 { + t.Fatalf("非白名单格式发起了 %d 次请求,want 0", s.hits.Load()) + } + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("非白名单格式被内联了") + } +} + +// TestInboundOversizeBodyIsCutWhileReading 钉住边读边限。这里刻意让假 CDN 下发一份 +// **合法 AES 密文**的超大附件:解密一定能成功,因此「Data 为空」只可能来自体积上限 +// (若用随机字节,解密失败会替上限背锅,用例就变成了假绿)。 +func TestInboundOversizeBodyIsCutWhileReading(t *testing.T) { + t.Parallel() + key := keyOf(0x34) + big := bytes.Repeat([]byte("x"), documents.DefaultMaxBytes+64) + s := newIlinkAssetServer(t, encrypted(t, key, big)) + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(fileItemRaw("media/big.txt", "big.txt")) + a.deliver(context.Background(), *m, nil) + + if len(c.message(t).Media[0].Data) != 0 { + t.Fatal("超量响应被内联了") + } +} + +// TestInboundUnreadableCiphertextIsHonest 钉住「解码失败 ⇒ 诚实放弃」:拿到的密文 +// 不是合法的 AES 分组序列时该附件留空,消息照收。 +func TestInboundUnreadableCiphertextIsHonest(t *testing.T) { + t.Parallel() + key := keyOf(0x35) + s := newIlinkAssetServer(t, []byte("not-a-cipher-block")) + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(fileItemRaw("media/bad.txt", "bad.txt")) + a.deliver(context.Background(), *m, nil) + + msg := c.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatal("解密失败时仍写入了字节") + } + if msg.SenderID != "u-1" { + t.Fatalf("失败路径把消息弄丢了:%+v", msg.SenderID) + } +} + +// TestInboundBodyURLIsNeverFetched 是红线用例:文本 item(正文)里带任意 URL 时 +// **绝不**对它发起请求(断言 0 次命中)。 +func TestInboundBodyURLIsNeverFetched(t *testing.T) { + t.Parallel() + key := keyOf(0x36) + s := newIlinkAssetServer(t, encrypted(t, key, []byte("secret\n"))) + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(textItemRaw("读一下 " + s.srv.URL + "/c2c/secret.txt")) + a.deliver(context.Background(), *m, nil) + + if s.hits.Load() != 0 { + t.Fatalf("正文里的 URL 被请求了 %d 次,want 0", s.hits.Load()) + } + if c.got == 0 { + t.Fatal("文本消息没有被投递") + } +} + +// TestInboundFetchFailureIsHonest 钉住失败路径:CDN 5xx ⇒ 放弃内联,消息照收。 +func TestInboundFetchFailureIsHonest(t *testing.T) { + t.Parallel() + key := keyOf(0x37) + s := newIlinkAssetServer(t, []byte("ignored")) + s.code = http.StatusBadGateway + a, c := newInlineIlinkAdapter(t, s, key) + + _, m := a.convertMessage(fileItemRaw("media/1.txt", "report.txt")) + a.deliver(context.Background(), *m, nil) + + msg := c.message(t) + if len(msg.Media[0].Data) != 0 { + t.Fatal("失败时仍写入了字节") + } + if msg.ChannelID != "u-1" { + t.Fatalf("失败路径把消息弄丢了:%+v", msg.ChannelID) + } +} diff --git a/plugins/channels/weixin_ilink/poll.go b/plugins/channels/weixin_ilink/poll.go index de4c71d..a97e903 100644 --- a/plugins/channels/weixin_ilink/poll.go +++ b/plugins/channels/weixin_ilink/poll.go @@ -124,8 +124,10 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes } // deliver 在已安装处理器时把归一化消息交给它,否则交给 emit 回调 -// (design/08 §1:推送式与拉取式投递)。 +// (design/08 §1:推送式与拉取式投递)。文件类附件的字节在**投递之前**内联 +// (切片 IM-1):轮询是这条适配器唯一的入站路径,因此这里是唯一的决策点。 func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) { + a.inlineInboundMedia(ctx, &m) if h := a.handlerLocked(); h != nil { h(ctx, m) return @@ -220,8 +222,9 @@ func (a *Adapter) convertMessage(raw json.RawMessage) (string, *adapter.UnifiedM } // convertItems 把 item 列表抽取成文本、一个主内容类型和媒体切片。文本 item 的文本 -// 会拼接起来;语音 item 贡献其 ASR 转写;图片/文件/视频 item 以 CDN 引用的形式惰性 -// 挂上(轮询期间绝不下载)。 +// 会拼接起来;语音 item 贡献其 ASR 转写;图片/文件/视频 item 先以 CDN 引用的形式 +// 挂上(转换期不下载);**文件类**附件的字节随后由 deliver → inlineInboundMedia 在 +// 受支持格式与限额内取回并内联(切片 IM-1),图片/视频仍走各自既有链路。 func (a *Adapter) convertItems(rawItems []json.RawMessage) (string, adapter.ContentType, []adapter.Media) { var text strings.Builder var media []adapter.Media