feat(channels): 适配器入站内联附件字节(telegram/kook/qq_obv11/weixin_ilink)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 新增 internal/channel/mediafetch:来源白名单(deny-by-default,支持 *.suffix)
  + 逐跳重定向校验 + 边读边限 + 超时;只回来源主机,绝不回完整 URL
- 四渠道各自 media.go 接通入站内联:只从平台载荷给出的引用取回字节
  (telegram file_id→getFile→file_path 形态校验;kook CDN;qq 文件段 url;
   weixin cdnURL + AES-ECB 解密)
- 常量全部复用既有来源:documents.ExtAllowed / DefaultMaxBytes / MaxPerMessage
- channelbridge 诚实提示文案改为对「渠道只给远程引用」与「代取失败」两种成因都成立
- 18 个测试函数(含四渠道「正文里的 URL ⇒ 0 次命中」红线反证、端到端进模型可见视图)
- 变异验证:去掉尺寸上限 ⇒ 四渠道红(含修正后不再是假绿的 weixin 超量用例)
This commit is contained in:
AskaEth committed 2026-10-04 22:13:17 +08:00
1 parent 46caa724f2
commit 3be70f8232
22 files changed
+2601 -19

No files matched your search

+139
View File
@@ -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)
}
+244
View File
@@ -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)
}
})
}
}
+369
View File
@@ -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
}
@@ -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
}
@@ -59,9 +59,13 @@ const (
inboundNoticeTooLarge = "[附件 %s · %s · %d 字节 · 无法读取:文件超过服务端体积上限]" inboundNoticeTooLarge = "[附件 %s · %s · %d 字节 · 无法读取:文件超过服务端体积上限]"
inboundNoticeEmpty = "[附件 %s · %s · %d 字节 · 无法读取:文件为空]" inboundNoticeEmpty = "[附件 %s · %s · %d 字节 · 无法读取:文件为空]"
inboundNoticeStoreError = "[附件 %s · %s · %d 字节 · 无法读取:服务端保存该附件失败]" inboundNoticeStoreError = "[附件 %s · %s · %d 字节 · 无法读取:服务端保存该附件失败]"
inboundNoticeNoBytes = "[附件 %s · %s · %d 字节 · 无法读取:渠道只给了远程引用,服务端未代取文件字节]" // inboundNoticeNoBytes 覆盖「服务端没拿到字节」这一类:渠道只给了远程引用、或
inboundNoticeUnwired = "[附件 %s · %s · %d 字节 · 无法读取:服务端未接线文档存储]" // 代取失败(切片 IM-1 起,受支持格式的附件由**适配器**在入站时按平台资产白名单
inboundNoticeTooMany = "[另有附件未处理:单条消息最多 %d 个文件]" // 取回并内联;取不到时字节仍为空,走这一行)。文案对两种成因都成立 —— 把「代取
// 失败」说成「我们根本不代取」也是一种不诚实。
inboundNoticeNoBytes = "[附件 %s · %s · %d 字节 · 无法读取:服务端没有拿到文件字节(渠道只给了远程引用,或代取失败)]"
inboundNoticeUnwired = "[附件 %s · %s · %d 字节 · 无法读取:服务端未接线文档存储]"
inboundNoticeTooMany = "[另有附件未处理:单条消息最多 %d 个文件]"
) )
// inboundAttachments 是一次入站附件摄取的产物。 // inboundAttachments 是一次入站附件摄取的产物。
@@ -147,9 +151,11 @@ func (b *Bridge) ingestFiles(ctx context.Context, store *documents.Store, msg ad
break break
} }
if len(media.Data) == 0 { if len(media.Data) == 0 {
// 渠道只给了远程引用(CDN URL / 平台文件 id):本切片**不代取** —— // 渠道只给了远程引用(CDN URL / 平台文件 id)而字节仍为空:切片 IM-1 起,
// 服务端代取任意 http(s) 正是 F3 的教训(design/24 §5),需要专门 // **适配器**在入站时会按平台资产白名单把受支持格式的附件取回并内联,因此
// 的裁决与地址守卫,不在这里顺手打开。绝不静默。 // 走到这里意味着「渠道没给内容」或「代取失败」。服务端代取**任意 http(s)
// URL** 仍是 F3 的教训(需要专门裁决与地址守卫),而代取**平台自己的资产**
// 由适配器按来源白名单完成 —— 两者不是同一件事。绝不静默。
b.log.Warn("inbound attachment has no inlined bytes", b.log.Warn("inbound attachment has no inlined bytes",
"platform", msg.Platform, "channel", msg.ChannelID, "platform", msg.Platform, "channel", msg.ChannelID,
"name", media.Name, "mime", media.MIMEType, "size", media.Size) "name", media.Name, "mime", media.MIMEType, "size", media.Size)
@@ -41,8 +41,9 @@ func TestInboundOversizeFileIsHonestAboutSize(t *testing.T) {
} }
} }
// TestInboundRemoteOnlyMediaIsHonestWithoutFetch 钉住「渠道只给远程引用」这一现实形态 // TestInboundRemoteOnlyMediaIsHonestWithoutFetch 钉住「字节为空」这一形态:桥接侧
// (适配器目前上报 URL/平台文件 id,而不是内联字节):本切片**不代取**,但绝不静默。 // **不代取**(受支持格式的代取由适配器在入站时按平台资产白名单完成,切片 IM-1),
// 字节空 ⇒ 一行诚实提示 + 一条带来源的日志,绝不静默。
func TestInboundRemoteOnlyMediaIsHonestWithoutFetch(t *testing.T) { func TestInboundRemoteOnlyMediaIsHonestWithoutFetch(t *testing.T) {
t.Parallel() t.Parallel()
env := newIngestEnv(t, documents.StoreConfig{}, true) env := newIngestEnv(t, documents.StoreConfig{}, true)
@@ -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 }
+52
View File
@@ -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...)
}
+249
View File
@@ -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("图片媒体被写入了字节(应留给视觉链路)")
}
}
+3 -1
View File
@@ -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 // 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) { func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) {
a.inlineInboundMedia(ctx, &msg)
a.handlerMu.RLock() a.handlerMu.RLock()
fn := a.handler fn := a.handler
a.handlerMu.RUnlock() a.handlerMu.RUnlock()
+62
View File
@@ -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...)
}
+266
View File
@@ -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())
}
}
+4 -1
View File
@@ -192,8 +192,11 @@ func (a *Adapter) handlePayload(ctx context.Context, payload []byte) {
} }
// dispatch routes an inbound message to the registered push handler, falling // 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) { func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) {
a.inlineInboundMedia(ctx, &msg)
a.handlerMu.RLock() a.handlerMu.RLock()
fn := a.handler fn := a.handler
a.handlerMu.RUnlock() a.handlerMu.RUnlock()
+3
View File
@@ -213,6 +213,9 @@ func (a *Adapter) ProcessUpdate(ctx context.Context, raw []byte) (int, error) {
if err := msg.Validate(); err != nil { if err := msg.Validate(); err != nil {
return 0, fmt.Errorf("telegram: validate webhook update: %w", err) return 0, fmt.Errorf("telegram: validate webhook update: %w", err)
} }
// 附件字节在**投递之前**内联:webhook 与长轮询两条入站路径共用同一口径
// (切片 IM-1)——下游因此拿到的永远是一份已经带字节(或明确没有字节)的消息。
a.inlineInboundMedia(ctx, &msg)
a.mu.RLock() a.mu.RLock()
handler := a.handler handler := a.handler
a.mu.RUnlock() a.mu.RUnlock()
+3
View File
@@ -14,4 +14,7 @@ var (
// ErrNoChatTarget is returned when an outbound message carries no usable // ErrNoChatTarget is returned when an outbound message carries no usable
// conversation/group identifier. // conversation/group identifier.
ErrNoChatTarget = errors.New("telegram: message has no target chat") 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")
) )
+107
View File
@@ -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
}
+316
View File
@@ -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)
}
}
+1
View File
@@ -71,6 +71,7 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes
a.log.Warn("skip invalid message", "err", err) a.log.Warn("skip invalid message", "err", err)
continue continue
} }
a.inlineInboundMedia(pollCtx, &msg)
emit(msg) emit(msg)
} }
a.mu.Lock() a.mu.Lock()
+22 -6
View File
@@ -18,20 +18,36 @@ import (
"strings" "strings"
) )
// ilinkAuthType 是 iLink 请求头 `AuthorizationType` 的协议定值,仅此一处定义。
const ilinkAuthType = "ilink_bot_token"
// authorizationHeader 为 req 装上 iLink 通用请求头:JSON 内容类型、bearer // authorizationHeader 为 req 装上 iLink 通用请求头:JSON 内容类型、bearer
// token、认证类型、每次新生成的随机 X-WECHAT-UIN 防重放值,以及 App 标识头。 // token、认证类型、每次新生成的随机 X-WECHAT-UIN 防重放值,以及 App 标识头。
func (a *Adapter) authorizationHeader(req *http.Request) { func (a *Adapter) authorizationHeader(req *http.Request) {
req.Header.Set("Content-Type", "application/json") for k, vs := range a.inboundHeaders() {
req.Header.Set("AuthorizationType", "ilink_bot_token") for _, v := range vs {
req.Header.Set("X-WECHAT-UIN", randomUIN()) req.Header.Set(k, v)
req.Header.Set("iLink-App-Id", ilinkAppID) }
req.Header.Set("iLink-App-ClientVersion", buildClientVersion(ilinkVersion)) }
}
// 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() a.mu.RLock()
token := a.cfg.BotToken token := a.cfg.BotToken
a.mu.RUnlock() a.mu.RUnlock()
if token != "" { if token != "" {
req.Header.Set("Authorization", "Bearer "+token) h.Set("Authorization", "Bearer "+token)
} }
return h
} }
// randomUIN 返回 base64 编码的十进制 uint32,每次请求都新生成,作为 iLink 的 // randomUIN 返回 base64 编码的十进制 uint32,每次请求都新生成,作为 iLink 的
+53
View File
@@ -9,6 +9,9 @@ import (
"io" "io"
"net/http" "net/http"
"strings" "strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch"
) )
// ErrKeyTooShort 在入站密钥推导无法给出所需分组长度的 AES 密钥时返回。 // ErrKeyTooShort 在入站密钥推导无法给出所需分组长度的 AES 密钥时返回。
@@ -90,3 +93,53 @@ func (a *Adapter) fetchBody(ctx context.Context, rawURL string) ([]byte, error)
} }
return data, nil 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)
}
@@ -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)
}
}
+6 -3
View File
@@ -124,8 +124,10 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes
} }
// deliver 在已安装处理器时把归一化消息交给它,否则交给 emit 回调 // deliver 在已安装处理器时把归一化消息交给它,否则交给 emit 回调
// (design/08 §1:推送式与拉取式投递)。 // (design/08 §1:推送式与拉取式投递)。文件类附件的字节在**投递之前**内联
// (切片 IM-1):轮询是这条适配器唯一的入站路径,因此这里是唯一的决策点。
func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) { func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) {
a.inlineInboundMedia(ctx, &m)
if h := a.handlerLocked(); h != nil { if h := a.handlerLocked(); h != nil {
h(ctx, m) h(ctx, m)
return return
@@ -220,8 +222,9 @@ func (a *Adapter) convertMessage(raw json.RawMessage) (string, *adapter.UnifiedM
} }
// convertItems 把 item 列表抽取成文本、一个主内容类型和媒体切片。文本 item 的文本 // 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) { func (a *Adapter) convertItems(rawItems []json.RawMessage) (string, adapter.ContentType, []adapter.Media) {
var text strings.Builder var text strings.Builder
var media []adapter.Media var media []adapter.Media