diff --git a/docs/guides/media_resolver.md b/docs/guides/media_resolver.md index 5d92058..70010d8 100644 --- a/docs/guides/media_resolver.md +++ b/docs/guides/media_resolver.md @@ -10,7 +10,7 @@ - 把易碎的 TikHub 解析逻辑集中到专用服务，本仓库退化为「下载 + 转录 + LLM」； - 抖音改为下载**完整 mp4 再由 CapsWriter 提取音轨**（而非旧版直接抓 `music.play_url` 的 mp3）—— 对套用热门 BGM 模板的口播视频，提取的是**视频自带人声**而非背景乐，转录更准； -- 支持微信视频号（`https://weixin.qq.com/sph/`）链接转录，通过 MediaResolverAPI 的流式代理端点边解密边拉取。 +- 支持微信视频号（`https://weixin.qq.com/sph/`）链接转录：resolver 只负责解析并下发解密文件头 + 微信 CDN 直链（`GET /api/stream/wechat_channels/{sph_code}/direct`），由本服务直连 CDN 按 Range 拼接成完整 mp4（不经 resolver 流式中转，避免跨机房 DERP 慢速）。 > ⚠️ **行为变更**：开启后抖音下载体积由 mp3 增大为 mp4。长视频可能撞 `storage.max_download_size_mb` > 上限，或 CapsWriter 一次性入内存的限制。短视频无影响。 @@ -33,8 +33,8 @@ 2. 微信视频号转录前提： - 必须设置 `downloaders.use_media_resolver: true`（无原生下载器兜底）； - 若配置了 `security.download_url_allowlist` 安全下载白名单，MediaResolverAPI 服务域名/IP 须在允许列表中； - - 视频号 `video_url` 指向 resolver 自己的流式代理端点（`/api/stream/wechat_channels/{sph_code}`），下载时下载器会自动携带 `X-API-Key` 鉴权头（第三方 CDN 直链则不会携带）； - - resolver 流式端点受上游并发限制，若单进程并发超限会返回 429（Too Many Requests）。 + - 视频号 `video_url` 仍指向 resolver 的流式端点路径（`/api/stream/wechat_channels/{sph_code}`），下载器识别后改调 `/direct` 拿到解密头与 CDN 直链，再由本机 Range 续传拼接；只有调 `/direct` 时携带 `X-API-Key`，请求微信 CDN 时不携带； + - CDN 直链含时效 token，慢读可能被 CDN 掐断，下载器会重新调 `/direct` 换链并从已下载偏移续传（最多 5 次）。 3. 服务健康检查： ```bash @@ -93,7 +93,6 @@ curl -X POST http://localhost:8000/api/transcribe \ | 图文/已删除/私密等无视频内容 | 该内容无可转录视频 | 否 | | 全部解析源失败 | 解析失败，稍后再试 | 否 | | 服务端错误（HTTP 5xx） | 解析服务异常 | 是 | -| 流式端点并发超限（HTTP 429） | 解析服务繁忙，稍后再试 | 是 | > 注：当前 MediaResolverAPI 的 `error` 仅返回文案、无结构化 `error.code`，因此「图文/删除」类终态 > 可能被笼统归为「解析失败，稍后再试」。若你维护该服务，建议为终态返回 `error.code` 以便精确区分。 @@ -114,8 +113,9 @@ MediaResolverAPI 返回的视频直链在下载前会经过 **SSRF 校验**（`u |------|------| | 提示「鉴权失败」 | 检查 `media_resolver.api_key`；用 `curl -H "X-API-Key: "` 直接打 `/api/resolve` 验证 | | 提示「解析服务暂不可用」 | 检查 `base_url` 可达性、`/health`、Docker 内是否误用 localhost | -| 视频号下载失败（401） | 确认下载请求发往 resolver 域名并携带了 `X-API-Key` | -| 视频号下载失败（429） | MediaResolverAPI 流式并发超限，稍后重试或调整 resolver 服务并发能力 | +| 视频号下载失败（401） | 确认 `/direct` 请求发往 resolver 域名并携带了 `X-API-Key` | +| 视频号下载失败（502） | resolver 拉微信元数据/CDN 失败，稍后重试 | +| resolver 返回的 `video_url`（流式端点）的 netloc 须与 `media_resolver.base_url` 一致，否则 X-API-Key 不会携带且可能触发 SSRF 拦截 | 核对 `media_resolver.base_url` 与解析结果里的主机名（含端口）是否完全一致（小写比较） | | 抖音下载撞大小上限 | 调高 `storage.max_download_size_mb`，或对长视频暂时关闭开关 | | 想确认走了哪个下载器 | 看日志 `为URL创建下载器: ..., 类型: MediaResolverDownloader` | diff --git a/docs/sessions/wechat-direct/reviews/pr68-verdict-r1.md b/docs/sessions/wechat-direct/reviews/pr68-verdict-r1.md new file mode 100644 index 0000000..4fdfd9b --- /dev/null +++ b/docs/sessions/wechat-direct/reviews/pr68-verdict-r1.md @@ -0,0 +1,66 @@ +# PR #68 独立审查 verdict（R1） + +- **审查对象**：`git diff 4e0a5ab0e8841b07577a2d9e984a078f18a6f68c..b2d67c3`（H0 冻结；之后的新提交不属于本轮） +- **风险等级**：internal（P1 红线：数据丢失、静默出错、崩溃、越权访问、损坏他人数据） +- **真实使用方式**：单机单用户部署，resolver 为自有服务，视频号 URL 来自用户提交 +- **Verdict 文件**：`docs/sessions/wechat-direct/reviews/pr68-verdict-r1.md`（本文件） +- **审查人**：Kimi（独立 review 卡 dlg-20260902-122616-64c9fb，全新会话，未见实现方报告） + +## 本轮新证据 + +1. 在 b2d67c3 临时 worktree 实跑三个受影响测试文件：`test_media_resolver_wechat_direct.py` + `test_media_resolver_client.py` + `test_media_resolver_downloader.py`，**64 passed**。 +2. 降层实测探针：按 `test_final_size_mismatch_raises` 的场景（content_length = head+50、CDN 恒 410）实跑 `_download_wechat_direct`，实际抛出的是 `wechat direct no progress`（零进展分支），**不是**测试名声称的 size-mismatch 分支（见 P2-2）。 +3. OCR 前置扫描：`ocr-review --from 4e0a5ab --to b2d67c3`，status=**reviewed**（primary_selected，非 skipped），13 条 finding 逐条分诊见下表。 + +## 不变式逐条核验 + +| # | 不变式 | 结论 | 证据 | +|---|--------|------|------| +| 1 | token 保密 | ✅ | 全部日志/异常只含 sph_code/offset/字节数/status（`media_resolver.py:343-470`、`media_resolver_client.py:209-264`）；cdn_url 仅存在局部变量 `payload`，不入 `_resolve_cache`、不落盘；连接失败只记 `type(e).__name__`；测试 `test_returns_flat_json_and_omits_cdn_from_logs` 与 `_no_secret` 断言锁死 | +| 2 | 密钥隔离 | ✅ | CDN 请求头只有 `Range`（`media_resolver.py:445-448`），X-API-Key 仅由 client 发向 `self.base_url` 拼出的 `/direct`（`media_resolver_client.py:214-215`）；测试断言 `X-API-Key not in CDN headers` 且 /direct 携带该头 | +| 3 | 完整性 | ✅ 实现 / ⚠️ 防线 | ftyp bytes[4:8] 校验（`media_resolver.py:318-322`）；Content-Range 起点 != offset 即换链（`:449-456`）；收尾 `os.path.getsize(local_path) == content_length` 断言打在终态文件本身（`:373-378`）。但收尾校验无有效测试锁死（P2-2） | +| 4 | 有界失败 | ✅ | 换链上限 5（`:418-422`）；连续零进展 2 次放弃（`:414-417`）；超限/畸形/no-ftyp 全部 fail fast 抛错；溢出 chunk 不写入、不截断凑数（`:462-468`）；任何异常 `_discard_temp` 销毁临时文件（`:379-381`） | +| 5 | 路由隔离 | ✅ | `_wechat_stream_sph_code`（`:296-307`）：netloc 小写精确相等 + path 正则 `^/api/stream/wechat_channels/([A-Za-z0-9]{1,64})$`；其他 URL 走既有 `_prepare_download_headers` + 基类逻辑；`test_non_wechat_url_does_not_call_direct` 锁死 | +| 6 | SSRF | ⚠️ 一处缺口 | 每次 CDN GET 前过 `validate_url_safe`（`:393-398`），但 requests 默认跟随 30x 重定向，**redirect 目标未过校验**（P2-1） | + +## Findings + +### P2-1 CDN 请求跟随重定向，redirect 目标绕过 validate_url_safe（违反不变式 6） + +- **证据**：`media_resolver.py:445` `requests.get(cdn_url, headers={"Range": ...}, stream=True, timeout=...)` 未传 `allow_redirects=False`，requests 对 GET 默认跟随最多 30 次重定向；`validate_url_safe` 只在 `:394` 校验初始 cdn_url，30x 的 Location 目标（可为内网地址）不再过 SSRF 校验。 +- **两问**：①会被触发吗——cdn_url 来自自有 resolver、指向 finder.video.qq.com，微信 CDN 正常直接回 206/4xx，30x 到内网需 CDN 行为异常或中间设备介入，概率低（无真实 token 无法实测线上 CDN 是否 30x）；②后果——请求不携带任何凭据（无 X-API-Key、无 cookie），响应体写入本地临时文件、不回显不记日志，最坏是内网内容拼进 mp4 导致文件损坏/下载失败，无数据外泄通道。后果可接受但有明确防线缺口 → **P2**，不阻塞合并。 +- **建议**：CDN GET 加 `allow_redirects=False`（30x 即 status != 206，自动落入既有换链逻辑，与有界失败协议天然兼容）。 + +### P2-2 `test_final_size_mismatch_raises` 名不符实：收尾 size 校验无测试锁死（不变式 3 的防线缺口） + +- **证据**：`tests/unit/test_media_resolver_wechat_direct.py` 中该测试用 8 份相同 payload + 恒 410 的 getter。实测（本审查探针，输出见上）：第二次零进展即在 `media_resolver.py:414-417` 抛 `wechat direct no progress`，**永远到不了** `:373-378` 的 `getsize != content_length` 分支。全 diff 无其他测试覆盖收尾 size 校验。 +- **两问**：①实现本身正确（探针与代码阅读双重确认），缺的是回归防护——收尾断言是「产出正确文件」的最后一道防线，改坏它没有任何测试变红；②后果是未来的回归静默通过 → **P2**，建议补一条真正打到 size-mismatch 分支的测试（如 CDN 返回短 body 后以 206 正常结束流），并给现有测试改名或加错误消息断言锁定目标分支。 +- 不阻塞合并（实现无缺陷），但必须记 backlog。 + +### P3（记 backlog，不阻塞） + +1. **换链后未复核新 payload 的 content_length/encrypted_head_bytes 与首次一致**（不变式 3 边缘）：`_append_wechat_cdn` 换链后只取 `cdn_url`（`media_resolver.py:423-433`），若 resolver 对同一 sph_code 返回变化的总长，最终校验仍按首次 content_length，理论上可产出「长度正确、内容混合」的文件。真实使用下同一 sph_code 内容在下载窗口内变化的可能性极低（自有 resolver）。 +2. **两处不可达死代码**：`_append_wechat_cdn` 循环末尾的 `raise DownloadFailedError("incomplete")`（`media_resolver.py:434-437`，循环内 ==/> 两种出口已覆盖）；`fetch_wechat_direct` for 循环后的 `raise NetworkError`（`media_resolver_client.py:262-264`，`max_retries = max(1, ...)` 保证循环内必 return/raise）。 +3. **SSLError 被当瞬时断流处理**（OCR [8]，工具标 medium）：`_stream_wechat_cdn_range` 的 `except requests.RequestException` 涵盖 SSLError，证书校验失败会走换链而非 fail fast。两问：换链有上界（5 次后抛错），不静默产坏文件、不泄密；单机直连部署无透明代理 → P3。建议单独 catch SSLError 直接抛 DownloadFailedError。 +4. **SSRF 校验失败重抛缺 `from e`**（OCR [12]）：`media_resolver.py:396-398` 丢原始异常链，诊断性小事。 +5. **测试加固类**（OCR [0][1][3][4][5]）：secret 断言用子串且排在严格相等断言之后（相等先失败会跳过泄露断言）；`patch_get`/`patch_get` 饱和重放最后一条响应；缺 `encrypted_head_bytes=0`、非法 base64 边界用例。均为测试健壮性，不影响生产行为。 + +### 驳回的 OCR finding + +- **OCR [10]**（head-only 分支 `encrypted_head_bytes > content_length`「静默成功」）：误读。该分支写入 head 后 `len(head) != content_length` 必抛 `DownloadFailedError` 且临时文件被销毁（`media_resolver.py:364-369, 379-381`），不存在静默成功。不成立。 +- **OCR [11]**（validate_url_safe 应提升到换链边界而非每次请求）：与 spec 不变式 6 明文「每次 CDN 请求前过 validate_url_safe」直接冲突，且循环每轮恰好一次 GET，当前粒度即 spec 要求。反着 spec，不成立。 + +## 降层三问 + +**① 最终文件返回给转录流程之前，哪些已发生的动作不可逆？** +没有不可逆动作。写入目标是 `temp_manager.create_temp_file` 的独立临时文件，任何异常经 `except Exception: _discard_temp(local_path); raise`（`media_resolver.py:379-381`）销毁；网络侧全是 GET/Range 只读请求，不改 resolver、不改 CDN 状态。唯一的「交付」是 return 路径那一刻，此前全部可回滚。 + +**② 换链续传的 offset 单一事实源是什么、并发场景下唯一吗？** +单一事实源是 `_append_wechat_cdn` 的栈帧局部变量 `offset`：初值 `len(head)`（client 已校验 `len(head) == encrypted_head_bytes`），只由 `_stream_wechat_cdn_range` 返回的**实际写入字节数** gained 累加（`media_resolver.py:402-403`），不从 Content-Range、不从 resolver 响应反推。并发：每次 `download_file` 调用持有独立临时文件与独立栈帧 offset，downloader 实例上没有跨调用的下载进度状态（`_resolve_cache`/`_video_url_to_page` 与本路径无关）；单机转录流程串行调用。唯一且安全。 + +**③ 保护覆盖的是「写入字节数」还是「产出正确文件」这一行为？** +覆盖到终态交付物本身：收尾断言是 `os.path.getsize(local_path) == content_length`（对落盘文件的实测，不是中间计数器）+ head 的 ftyp magic 校验，量纲与「产出正确文件」一致。不在覆盖范围内的是「内容字节正确性」（CDN 返回同长度错误内容不可发现）——spec 只要求字节数 + ftyp，保护与 spec 一致。唯一缺口是这道终态断言没有回归测试锁死（P2-2）。 + +## 总结论 + +**pass**。无 P1；必修清单为空。六条领域不变式实现层面全部成立（实测 64 项单测全绿 + 降层探针佐证）。两条 P2（redirect 过 SSRF、收尾校验缺测试）与五条 P3 记 backlog，不阻塞合并，建议下一轮或后续卡处理。 diff --git a/docs/sessions/wechat-direct/reviews/pr68-verdict-r2.md b/docs/sessions/wechat-direct/reviews/pr68-verdict-r2.md new file mode 100644 index 0000000..fb2d80f --- /dev/null +++ b/docs/sessions/wechat-direct/reviews/pr68-verdict-r2.md @@ -0,0 +1,255 @@ +# PR #68 独立审查 verdict（R2：对抗性运行时实测） + +- **审查对象**：`git diff 4e0a5ab0e8841b07577a2d9e984a078f18a6f68c..b2d67c3`（H0 冻结；之后的新提交不属于本轮） +- **H0 SHA**：`b2d67c3d39a7788a0245d6523ff3271d95b4439a` +- **风险等级**：internal（P1 红线：数据丢失、静默出错、崩溃、越权访问、损坏他人数据） +- **真实使用方式**：单机部署，resolver 为自有服务，视频号 URL 来自用户提交。本轮在审查 worktree 所在机器上实测：`169.254.169.254` 的 AWS IMDS v1 / v2 token / GCP metadata 三个 URL 均 `curl -m 2` 连接超时（http=000），本机无云元数据服务。 +- **Verdict 文件**：`docs/sessions/wechat-direct/reviews/pr68-verdict-r2.md`（本文件） +- **审查人**：Grok（独立 review 卡 dlg-20260902-125054-3011bf，全新会话，未见实现方报告） +- **与 R1 的关系**：R1 是静态 diff + 单测 + OCR，verdict **pass / 无 P1**，P2-1 为「CDN GET 跟随 30x、redirect 目标未过 `validate_url_safe`」。本轮**不重复读 diff**，只用真实本地 HTTP stub 驱动 `MediaResolverDownloader.download_file` 的真实 `requests` 路径，确认或推翻该判定，并覆盖其余不变式。 + +## 本轮新证据是什么 + +1. 在 `b2d67c3` 上起真实 `ThreadingHTTPServer`（`127.0.0.1` 高端口），同时扮演 resolver（`GET /api/stream/wechat_channels/{sph}/direct`）与 CDN（Range / 302 / TCP 掐断）。**未使用 `unittest.mock`**。 +2. 仅把 `VTAPI_CONFIG` 里的 `media_resolver.base_url` 指向 stub、`api_key` 设为测试钥；CDN 要走真实 `validate_url_safe`，故把 `127.0.0.1` 写入 `security.download_url_allowlist`（生产已有的白名单机制，不是 mock）。`169.254.169.254` **不在**白名单。 +3. 用真实 `MediaResolverDownloader().download_file(stream_url, filename)` 跑 6 个对抗场景；stub 侧记录每个请求的 path / headers / 状态；loguru sink 捕获真实日志。 +4. 跨文档：`docs/guides/media_resolver.md`（本 diff）对照上游 `/home/zlx/projects/work/MediaResolverAPI/README.md` 的 `/direct` 与「客户端拼接协议」节。 +5. 本轮未重跑 OCR：同一份 H0 diff 已在 R1 `ocr-review` 得到 `status=reviewed`；对同一 diff 再扫不算新证据。 + +Harness 与原始结果在 `/tmp/pr68-r2-run/`（不入库）：`results.json`、`scenario1_fixed.json`、`routing.json`。 + +## 不变式速览（本轮实测） + +| # | 不变式 | 本轮结论 | 关键实测 | +|---|--------|----------|----------| +| 1 | token 保密 | ✅ INFO/异常干净；⚠️ DEBUG 存量泄露 | 场景 6 | +| 2 | 密钥隔离 | ✅ | 场景 5：resolver 必带 `X-API-Key`，CDN 从不带 | +| 3 | 完整性 | ✅ | 场景 2：掐断后续传，落盘字节 == 期望全文 | +| 4 | 有界失败 | ✅ | 场景 2/3/4：失败抛错并销毁临时文件，不截断凑数 | +| 5 | 路由隔离 | ✅ | 匹配 stream path 才进新路径；`/direct` 后缀与错误 netloc 不进 | +| 6 | SSRF | ⚠️ 初始 URL 有效 / 302 目标无效 | 场景 3 拦截元数据直链；场景 1/1b **跟随** 302 且不对 Location 再校验。确认 R1 P2-1，不升级 P1 | + +--- + +## 场景 1：CDN 响应 302 到另一个 stub 路径 + +**注入是否生效** + +- stub `/direct` 返回 `cdn_url=http://127.0.0.1:/cdn/token-R2CDNTOKENAAAA1111-redir.mp4`（含时效 token 字面量）。 +- stub 对该 path 回 **302** `Location: http://127.0.0.1:/cdn/redirect-target`。 +- 第一次跑时 harness 误把 `/cdn/redirect-target` 也匹配成 302（路径含子串 `redir`），出现 30 次自环后 `TooManyRedirects` → `CDN connect failed`。这只能证明「跟随了」，不能证明「跟随后会不会把目标正文写进文件」。 +- 修正 sink 后复测（`/tmp/pr68-r2-run/scenario1_fixed.json`）：stub 记录 + + | path | status | 次数 | + |------|--------|------| + | `.../SPHREDIR01/direct` | 200 | 1 | + | `/cdn/token-R2CDNTOKENAAAA1111-redir.mp4` | 302 | 1 | + | `/cdn/redirect-target` | 206 / 32768 bytes | 1 | + + 第二次 CDN GET 的 `Range: bytes=20-` 被转发到 redirect-target。注入生效，路径完整走到跟随后的 206。 + +**观察到的行为** + +- `download_file` **跟随** 302（`requests.get` 默认 `allow_redirects=True`，代码未关）。 +- 跟随后的 206 正文被写入临时文件；下载成功返回路径；落盘 32788 字节且 `equals_full=true`。 +- 全程 **1 次** `/direct`（没有因为 302 本身去换链——跟随成功后视作一次合法 206）。 +- redirect-target **没有**再过 `validate_url_safe`（日志里只有对原始 token URL 的 `URL safety check passed`）。 + +**判定（不变式 6）** + +确认 R1 P2-1，**不推翻**。行为是「跟随并采用目标正文」，不是「拒绝 30x」。详见 Findings P2-1。 + +--- + +## 场景 1b（加强）：CDN 302 到 `http://169.254.169.254/latest/meta-data/` + +**注入是否生效** + +- stub 对 `/cdn/token-...-redir-meta.mp4` 回 302，`Location` 为元数据 URL。 +- stub 记录该 302 发出 2 次（首次 + 一次换链重试）。CDN 侧没有出现对 stub 其它 path 的 206。 + +**观察到的行为** + +- `elapsed_s = 20.0365`（CDN connect timeout 为 `(10, 60)` 的 10 秒连接超时 × 2 次零进展）。 +- 日志：`wechat direct CDN connect failed` → refresh `/direct` → 再次 GET 原 token URL → 再 302 → 再 10s 超时 → `DownloadFailedError: wechat direct no progress ... offset=20`。 +- 对比场景 3（直链就是 169.254.169.254）仅 **0.0039s** 即 SSRF 拒绝：证明 1b **确实对 Location 发了 TCP 连接**，而不是在校验层拦住。 +- 最终无落盘文件（`file: null`），临时文件被丢弃。 + +**判定** + +运行时确认：对已通过 SSRF 的 `cdn_url`，其 302 Location 可以是链路本地元数据地址，代码会尝试连接。本机 IMDS 不可达，后果是失败而非读到元数据。不升级 P1，理由见 Findings。 + +--- + +## 场景 2：CDN 流在 50% 处 TCP 掐断，换链后续传 + +**注入是否生效** + +- 第 1 次 `/direct` 下发 `token-R2CDNTOKENAAAA1111`。 +- stub 对该 URL 回 206，`Content-Range: bytes 20-32787/32788`，`Content-Length` 广告 32768（剩余全文），但只写出 **16384** 字节后 `shutdown(SHUT_RDWR)` + `close()`。 +- stub `sent_status`：`abort_after=16384, advertised=32768`。 +- 第 2 次 `/direct` 下发 `token-R2CDNTOKENBBBB2222`；对该 URL 完整 206。 + +**观察到的行为** + +- 日志：`CDN stream interrupted sph_code=SPHABORT50 offset=20 gained=16384`。 +- 换链：`refresh CDN link ... offset=16404 refreshes=1`（20+16384=16404）。 +- 第二次 CDN 请求头 `Range: bytes=16404-`（stub 实测）。 +- `download complete ... bytes=32788`；落盘文件 `equals_full=true`，`head_ok=true`。 +- `/direct` 调用 2 次，CDN 两个 token 各 1 次。无截断凑数。 + +**判定（不变式 3、4）** + +通过。TCP 层掐断被当成可恢复中断，换链后续传，终态字节与 stub 下发的明文头+身子完全一致。 + +--- + +## 场景 3：`cdn_url` 换成 `http://169.254.169.254/latest/meta-data/` + +**注入是否生效** + +- stub `/direct` 的 JSON 里 `payload_cdn_host=169.254.169.254`（stub 自己记了一份 host，响应体给了客户端）。 +- `direct` 调用 1 次；**CDN 请求计数 = 0**（没有任何 Range GET 离开本机去元数据，也没有打到 stub 的 `/cdn/`）。 +- 预检：`validate_url_safe("http://169.254.169.254/latest/meta-data/")` 抛 `Access to cloud metadata endpoint is blocked`；`127.0.0.1` 因白名单放行。两相对照，说明白名单没有把元数据一起放掉。 + +**观察到的行为** + +- `elapsed_s = 0.0039`（远小于 10s 连接超时）。 +- `DownloadFailedError: wechat direct CDN failed SSRF check sph_code=SPHSSRF001 offset=20`。 +- 无落盘文件。日志 ERROR 行只含 sph_code/offset，不含元数据 URL 字面量。 + +**判定（不变式 6 的「初始 cdn_url」部分）** + +通过。直链指向元数据时，校验在 `requests.get` 之前生效，fail fast。与场景 1b 的「先连再失败」可区分：这里是路径走到了校验且拦住，不是「没走到 CDN 分支」。 + +--- + +## 场景 4：`head_b64` 解码后无 `ftyp` magic + +**注入是否生效** + +- stub 返回 `head_b64` = 20 字节且 `[4:8] != b"ftyp"`（`XXXX`），`encrypted_head_bytes=20` 与解码长度一致（通过 client 的长度校验，专门打 downloader 的 magic 校验）。 +- `/direct` 1 次且 HTTP 200；**CDN 计数 = 0**。 + +**观察到的行为** + +- `DownloadFailedError: wechat direct head lacks ftyp magic sph_code=SPHNOFTYP1 head_bytes=20`。 +- 无落盘文件（进入 `_decode_wechat_head` 失败发生在 `create_temp_file` 之后、写 CDN 之前；异常路径 `_discard_temp`）。 +- 没有对 CDN 的 Range 请求，也没有「先写坏头再继续」。 + +**判定（不变式 3、4）** + +通过。无 ftyp 时 fail fast，不静默产出坏文件。 + +--- + +## 场景 5：密钥隔离（全程 stub 记 headers） + +**注入是否生效** + +- resolver 与 CDN 都是同一 HTTP 进程，每个请求的 header 名与值都记下来。场景 2（完整成功路径，含换链）是主证据；1/3/4 作对照。 + +**观察到的行为（场景 2）** + +| 角色 | path | `X-API-Key` | `Range` | +|------|------|-------------|---------| +| resolver | `/api/stream/wechat_channels/SPHABORT50/direct` ×2 | **有**，值等于配置钥 | 无 | +| CDN | `/cdn/token-R2CDNTOKENAAAA1111.mp4` | **无**；头集合 = Host, User-Agent, Accept-Encoding, Accept, Connection, Range | `bytes=20-` | +| CDN | `/cdn/token-R2CDNTOKENBBBB2222.mp4` | **无**（同上） | `bytes=16404-` | + +场景 1 修复后：302 与 redirect-target 两次 CDN GET 同样无 `X-API-Key`，Range 被跟随转发。场景 3/4 无 CDN 请求；其唯一的 `/direct` 带钥。 + +**判定（不变式 2）** + +通过。`all_resolver_have_key=true`，`any_cdn_has_key=false`。 + +--- + +## 场景 6：日志不含 stub 下发的 `cdn_url` 字面量 + +**注入是否生效** + +- 故意使用不会在正常日志模板里出现的 token：`R2CDNTOKENAAAA1111` / `R2CDNTOKENBBBB2222`。 +- loguru sink 开到 DEBUG，捕获 downloader / client / `url_validator` 的真实输出；异常 `str(e)` 一并扫描。 + +**观察到的行为** + +- **INFO / WARNING / ERROR**（downloader 与 client 自己打的行）只含 `sph_code` / `offset` / `content_length` / `status` / `gained` / 异常类名。场景 2 成功路径与场景 3/4 失败路径的异常消息均不含 token。 +- **DEBUG**：`url_validator._validate_and_resolve` 存量代码 `URL safety check passed: {url[:100]}` 会把完整 `cdn_url`（含 token）打出来。场景 2 捕获到两条 DEBUG，分别含 A/B 两个 token。生产 `config.jsonc` 与 example 的 `log.level` 均为 **INFO**，默认不会落这条。 +- 这是新代码把 `cdn_url` 传进存量 `validate_url_safe` 之后的副作用，不是 downloader 自己把直链写进 INFO。 + +**判定（不变式 1）** + +INFO/异常层通过。DEBUG 层记 P3（见 Findings），不阻塞。不把存量 `url_validator` 本身当本 diff 的缺陷。 + +--- + +## 路由隔离（补充探针，非卡面六场景之一） + +对真实构造出的 `MediaResolverDownloader._wechat_stream_sph_code`： + +| URL | 是否走新路径 | +|-----|----------------| +| `http://127.0.0.1:/api/stream/wechat_channels/SPHROUTE01`（netloc 匹配 + path 正则命中） | 是，`sph_code=SPHROUTE01` | +| 同上但 path 多 `/direct` 后缀 | 否 | +| `http://example.invalid/api/stream/wechat_channels/SPHROUTE01`（netloc 不同） | 否 | + +与不变式 5 一致。 + +--- + +## 跨文档一致性核对 + +对照对象： + +- 本 diff：`docs/guides/media_resolver.md`（相对 `4e0a5ab` 的 12 行改动） +- 上游：`/home/zlx/projects/work/MediaResolverAPI/README.md`「GET /api/stream/wechat_channels/{sph_code}/direct」与「客户端拼接协议」 + +| 上游条款 | 下游文档 | 实现（本轮或 R1） | 出入 | +|----------|----------|-------------------|------| +| `GET .../direct` 扁平 JSON：`sph_code, cdn_url, content_length, encrypted_head_bytes, head_b64, content_type` | 写了端点与「解密文件头 + CDN 直链」，未列字段表，未提 `content_type` | client 校验前 5 项；`content_type` 忽略 | 省略，不构成协议冲突 | +| `cdn_url` 含时效 token，**不得写入日志或持久化** | 只写「CDN 直链含时效 token」，未复述禁日志 | INFO 层做到；DEBUG 见 P3 | 下游文档弱于上游契约 | +| `encrypted_head_bytes >= content_length` 则不要再发 Range | 未写 head-only 分支 | 实现有该分支 | 文档省略 | +| Range 起点 = `encrypted_head_bytes`（示例 `bytes=131072-`） | 只写「从已下载偏移续传（最多 5 次）」 | 实测首次 `Range: bytes=20-`（本 stub 头长 20），续传 `bytes=16404-` | 实现与上游一致；下游未写初始偏移 | +| 连接被掐或 401/403/404/410 → 换链续传 | 只写「慢读可能被 CDN 掐断」 | 掐断实测通过；4xx 集合在代码里，本轮未再打 4xx | 下游未提 4xx 换链 | +| 收尾字节数 == `content_length`，不等则续传或重拉 | 未写收尾校验 | 实现是校验失败 **抛错**（有界失败），不是无限重拉 | 与本仓不变式 4 一致，比上游「重拉」更严；文档未写 | +| `/direct` 不占流式并发槽；429 是流式端点的事 | 本 diff **删掉**了 429 用户提示行，401 排查改为「确认 `/direct` 带钥」 | 与上游分工一致 | 一致 | +| 流式 `video_url` 仍指向 `/api/stream/wechat_channels/{sph}` | 明确写「仍指向流式端点路径，下载器识别后改调 `/direct`」 | 路由探针确认 | 一致 | +| 只有调 `/direct` 带 `X-API-Key`，CDN 不带 | 写了 | 场景 5 实测 | 一致 | +| 「安全」节：SSRF 作用于 MediaResolverAPI 返回的「视频直链」 | 沿用旧表述（历史上指 `video_url`） | 新路径 SSRF 打在 `cdn_url`；`video_url` 是 resolver 流式端点 | 文档未点名 `cdn_url`，也未提 302 目标不复核 | + +**文档 finding**：无协议写反。有若干省略（head-only、4xx 换链、禁日志、cdn_url SSRF）。记 P3，不阻塞。排查表里「netloc 不一致则 X-API-Key 不携带」描述的是**未命中新路径时**的旧流式下载行为，与 `/direct` 路径不矛盾。 + +--- + +## Findings + +### P2-1（确认 R1，不升级）CDN GET 跟随 30x，Location 未过 `validate_url_safe`（不变式 6） + +- **本轮新证据**：场景 1 修复后，stub 对 token URL 回 302、对 `/cdn/redirect-target` 回 206，客户端跟随并把目标正文拼成完整 mp4（`equals_full=true`）。场景 1b 把 Location 设为 `http://169.254.169.254/latest/meta-data/`，耗时 20.0s（两次 10s connect timeout），证明对元数据 IP **发起了真实 TCP**；对比场景 3 直链元数据 0.0039s 即被拦。 +- **工具标注 / 本仓判定 / 两问**：R1 标 P2。本仓仍 P2。 + 1. 真实使用下会被触发吗？本机三次 curl 元数据均超时，无 IMDS。正常微信 CDN 对带 Range 的 GET 回 206/4xx，不 302 到内网。要走到 1b 这种洞，需要「已通过 SSRF 的 URL」（例如攻击者控制的公网主机，或异常 CDN）再 302 到内网。自有 resolver + 微信 CDN 的主路径不会这样。第一问在本机量过：元数据不可达，正常 CDN 行为未在本轮观察（无真实 token，按卡面用 stub 注入）。 + 2. 触发了后果能否接受？跟随后的响应体写入临时 mp4，成功则进转录；CDN 请求不带 `X-API-Key`（场景 5）。本机无 IMDS，1b 的后果是 `no progress` 失败、不落盘。云上若开放 IMDS，理论上可把元数据拼进文件——这是防线缺口，但当前真实部署测不到该后果。internal 的「越权访问」两问未同时成立 → **不升 P1**。 +- **建议（与 R1 相同，不重复开新意见）**：CDN GET 加 `allow_redirects=False`；30x 落入既有 `status != 206` 换链逻辑。 + +### P3-1 DEBUG 日志可含 `cdn_url`（不变式 1 边缘；不审存量实现） + +- 新代码把带 token 的 `cdn_url` 传给存量 `validate_url_safe`。后者 DEBUG 行 `URL safety check passed: {url[:100]}` 会打出 token。生产 `log.level=INFO`，INFO/异常层实测干净。记 backlog：若有人把级别调到 DEBUG，时效 token 会进日志文件。修复应在新代码侧避免把完整 `cdn_url` 交给会记日志的 API，或对 validator 的 debug 做截断/脱敏——那是后续卡，本轮不改被审代码。 + +### P3-2 下游文档省略上游协议若干条款（跨文档） + +- 未复述「cdn_url 不得进日志」、head-only 不再 Range、401/403/404/410 换链、收尾长度校验、SSRF 对象是 `cdn_url`。没有写反。不阻塞。 + +R1 的 P2-2（`test_final_size_mismatch_raises` 名不符实）与其余 P3 本轮未再测，不重复、不撤销。 + +## 降层三问（本轮运行时答案） + +1. **终态交付前哪些动作不可逆？** 场景 3/4/1b 失败均无残留文件。场景 2 成功才返回路径。网络侧全是 GET。无通知、无对外写入。 +2. **续传 offset 的事实源？** 场景 2：stub 广告 32768、只给 16384 → 客户端 `gained=16384` → 下一跳 `Range: bytes=16404-`。offset 来自实际写入字节，不是 Content-Range 反推。 +3. **保护的是写入计数还是产出文件？** 场景 2 断言打在落盘字节与期望 `FULL` 逐字节相等，不是只比长度计数器。 + +## 总结论 + +**pass**。无新增 P1；必修清单为空。 + +本轮用真实 HTTP 对抗跑完卡面 6 场景：密钥隔离、掐断续传完整性、直链 SSRF、无 ftyp fail fast、INFO 层 token 保密均成立。R1 P2-1（302 跟随绕过 SSRF）被运行时**确认**（含跟随后写文件、以及 302 到 169.254.169.254 的真实 TCP），两问后仍为 P2，不阻塞合并。P3 两条（DEBUG 泄露、文档省略）记 backlog。 diff --git a/src/video_transcript_api/downloaders/media_resolver.py b/src/video_transcript_api/downloaders/media_resolver.py index 81a6e62..3fc2eb0 100644 --- a/src/video_transcript_api/downloaders/media_resolver.py +++ b/src/video_transcript_api/downloaders/media_resolver.py @@ -12,11 +12,15 @@ 详见 docs/designs/media-resolver-integration.md。 """ +import base64 import math import os +import re from typing import Dict, Optional from urllib.parse import urlparse, urlunparse +import requests + from .base import BaseDownloader from .models import VideoMetadata, DownloadInfo from .media_resolver_client import MediaResolverClient @@ -35,6 +39,16 @@ "weixin.qq.com", ) +# 视频号流式端点 path（不含 /direct）；命中则走 CDN 直连拼接 +_WECHAT_STREAM_PATH_RE = re.compile( + r"^/api/stream/wechat_channels/([A-Za-z0-9]{1,64})$" +) +# /direct 换新链上限（不含首次拉取） +_WECHAT_DIRECT_MAX_REFRESHES = 5 +_WECHAT_CDN_TIMEOUT = (10, 60) +_WECHAT_CDN_RETRY_STATUSES = frozenset({401, 403, 404, 410}) +_MP4_FTYP_MAGIC = b"ftyp" + class MediaResolverDownloader(BaseDownloader): """通过 MediaResolverAPI 解析抖音/小红书/微信视频号的下载器。""" @@ -235,6 +249,10 @@ def download_file(self, url, filename, max_retries: int = 3): 反查该 video_url 对应的页面 url，force_refresh 重解析得到新直链再下一次； 仍失败则抛 DownloadFailedError（爆炸半径仅限本类）。 """ + sph_code = self._wechat_stream_sph_code(url) + if sph_code: + return self._download_wechat_direct(sph_code, filename) + self._prepare_download_headers(url) local_file = super().download_file(url, filename, max_retries=max_retries) if local_file: @@ -271,3 +289,188 @@ def download_file(self, url, filename, max_retries: int = 3): if local_file: return local_file raise DownloadFailedError(f"重解析后仍下载失败: {fresh_url[:100]}") + + # ------------------------------------------------------------------ # + # 视频号：/direct 解密头 + CDN Range 直连拼接（不经 resolver 流式中转） + # ------------------------------------------------------------------ # + def _wechat_stream_sph_code(self, url: str) -> Optional[str]: + """命中 resolver 视频号流式端点时返回 sph_code，否则 None。""" + if not url or not self._resolver_netloc: + return None + try: + parsed = urlparse(url) + except Exception: + return None + if (parsed.netloc or "").lower() != self._resolver_netloc: + return None + match = _WECHAT_STREAM_PATH_RE.match(parsed.path or "") + return match.group(1) if match else None + + @staticmethod + def _decode_wechat_head(payload: dict, sph_code: str) -> bytes: + """解码 head_b64 并校验 mp4 ftyp magic（bytes[4:8]）。""" + head_b64 = payload.get("head_b64") + if not isinstance(head_b64, str) or not head_b64: + raise ResolverResponseError(f"wechat direct missing decrypted head sph_code={sph_code}") + try: + head = base64.b64decode(head_b64, validate=True) + except ValueError as e: + raise ResolverResponseError(f"wechat direct head decode failed sph_code={sph_code}") from e + if len(head) != payload.get("encrypted_head_bytes"): + raise ResolverResponseError(f"wechat direct head size mismatch sph_code={sph_code} decoded={len(head)}") + if len(head) < 8 or head[4:8] != _MP4_FTYP_MAGIC: + raise DownloadFailedError(f"wechat direct head lacks ftyp magic sph_code={sph_code} head_bytes={len(head)}") + return head + + @staticmethod + def _parse_content_range(header: Optional[str]): + if not isinstance(header, str): + return None + m = re.fullmatch(r"(?i)bytes\s+(\d+)-(\d+)/(\d+)", header.strip()) + if not m: + return None + start, end, total = map(int, m.groups()) + return None if end < start or total <= 0 else (start, end, total) + + def _download_wechat_direct(self, sph_code: str, filename: str) -> str: + """调 /direct 取解密头与 CDN 直链，Range 拼接成完整 mp4。""" + logger.info(f"wechat direct download start sph_code={sph_code}") + payload = self.client.fetch_wechat_direct(sph_code) + content_length = payload.get("content_length") + encrypted_head_bytes = payload.get("encrypted_head_bytes") + if not isinstance(content_length, int) or isinstance(content_length, bool): + raise ResolverResponseError(f"wechat direct invalid content_length sph_code={sph_code}") + if self.max_download_bytes and content_length > self.max_download_bytes: + raise DownloadFailedError( + f"wechat direct file exceeds size limit sph_code={sph_code} " + f"content_length={content_length} limit={self.max_download_bytes}" + ) + head = self._decode_wechat_head(payload, sph_code) + local_path = self.temp_manager.create_temp_file(suffix=".mp4") + try: + with open(local_path, "wb") as fh: + fh.write(head) + if encrypted_head_bytes >= content_length: + if len(head) != content_length: + raise DownloadFailedError( + f"wechat direct head-only size mismatch sph_code={sph_code} " + f"head_bytes={len(head)} content_length={content_length}" + ) + else: + self._append_wechat_cdn(fh, sph_code, payload, len(head), content_length) + actual_size = os.path.getsize(local_path) + if actual_size != content_length: + raise DownloadFailedError( + f"wechat direct size mismatch sph_code={sph_code} " + f"actual={actual_size} content_length={content_length}" + ) + logger.info(f"wechat direct download complete sph_code={sph_code} bytes={actual_size}") + return str(local_path) + except Exception: + self._discard_temp(local_path) + raise + + def _append_wechat_cdn(self, fh, sph_code, payload, offset, content_length): + """CDN Range 追加；中断/4xx/Range 不符则 /direct 换链续传。""" + origin = payload + cdn_url = payload.get("cdn_url") + if not isinstance(cdn_url, str) or not cdn_url.strip(): + raise ResolverResponseError(f"wechat direct missing CDN link sph_code={sph_code}") + refreshes = 0 + zero_streak = 0 + while offset < content_length: + try: + validate_url_safe(cdn_url) + except URLValidationError: + raise DownloadFailedError(f"wechat direct CDN failed SSRF check sph_code={sph_code} offset={offset}") + logger.info( + f"wechat direct CDN GET sph_code={sph_code} offset={offset} " + f"content_length={content_length} refreshes={refreshes}" + ) + gained = self._stream_wechat_cdn_range(cdn_url, fh, sph_code, offset, content_length) + offset += gained + zero_streak = (zero_streak + 1) if gained == 0 else 0 + if offset == content_length: + return + if offset > content_length: + raise DownloadFailedError( + f"wechat direct wrote past content_length sph_code={sph_code} " + f"offset={offset} content_length={content_length}" + ) + if zero_streak >= 2: + raise DownloadFailedError(f"wechat direct no progress sph_code={sph_code} offset={offset}") + if refreshes >= _WECHAT_DIRECT_MAX_REFRESHES: + raise DownloadFailedError( + f"wechat direct refresh limit sph_code={sph_code} offset={offset} refreshes={refreshes}" + ) + logger.info( + f"wechat direct refresh CDN link sph_code={sph_code} " + f"offset={offset} refreshes={refreshes + 1}" + ) + fresh = self.client.fetch_wechat_direct(sph_code) + for field in ("content_length", "encrypted_head_bytes", "head_b64"): + if origin.get(field) != fresh.get(field): + raise DownloadFailedError( + f"wechat direct object identity mismatch sph_code={sph_code} field={field}" + ) + cdn_url = fresh.get("cdn_url") + if not isinstance(cdn_url, str) or not cdn_url.strip(): + raise ResolverResponseError(f"wechat direct missing CDN link after refresh sph_code={sph_code}") + refreshes += 1 + raise DownloadFailedError( + f"wechat direct incomplete sph_code={sph_code} offset={offset} content_length={content_length}" + ) + + def _stream_wechat_cdn_range(self, cdn_url, fh, sph_code, offset, content_length) -> int: + """Range GET，返回本次写入字节。直链不进日志/异常；CDN 不带 X-API-Key。""" + response = None + gained = 0 + try: + try: + response = requests.get( + cdn_url, headers={"Range": f"bytes={offset}-"}, + stream=True, timeout=_WECHAT_CDN_TIMEOUT, allow_redirects=False, + ) + except requests.RequestException: + logger.warning(f"wechat direct CDN connect failed sph_code={sph_code} offset={offset}") + return 0 + parsed = self._parse_content_range(response.headers.get("Content-Range")) + clen = response.headers.get("Content-Length") + try: + clen_ok = clen in (None, "") or int(clen) == parsed[1] - parsed[0] + 1 + except (TypeError, ValueError): + clen_ok = False + range_ok = parsed is not None and parsed[0] == offset and parsed[2] == content_length and clen_ok + status = response.status_code + if status in _WECHAT_CDN_RETRY_STATUSES or status != 206 or not range_ok: + logger.warning( + f"wechat direct CDN unexpected response status={status} " + f"range={parsed} offset={offset} sph_code={sph_code}" + ) + return 0 + try: + for chunk in response.iter_content(chunk_size=8192): + if not chunk: + continue + if offset + gained + len(chunk) > content_length: + logger.error( + f"wechat direct CDN overflow sph_code={sph_code} offset={offset} " + f"gained={gained} chunk={len(chunk)} content_length={content_length}" + ) + return gained + fh.write(chunk) + gained += len(chunk) + if offset + gained == content_length: + break + except requests.RequestException: + logger.warning( + f"wechat direct CDN stream interrupted sph_code={sph_code} offset={offset} gained={gained}" + ) + return gained + finally: + if response is not None: + try: + response.close() + except Exception: + pass + diff --git a/src/video_transcript_api/downloaders/media_resolver_client.py b/src/video_transcript_api/downloaders/media_resolver_client.py index a957116..612768d 100644 --- a/src/video_transcript_api/downloaders/media_resolver_client.py +++ b/src/video_transcript_api/downloaders/media_resolver_client.py @@ -15,6 +15,8 @@ 异常映射见 Error & Rescue Registry（设计文档）。 """ +import base64 +import re import time from typing import Optional @@ -51,6 +53,9 @@ "RESOLVE_FAILED", "PROVIDER_ERROR", } +# 视频号 sph_code：与 resolver 路径约束一致（1-64 位字母数字） +_WECHAT_SPH_CODE_RE = re.compile(r"^[A-Za-z0-9]{1,64}$") + # 文案兜底关键词（服务只回 message、无 code 时使用） _NON_VIDEO_KEYWORDS = ( "图文", "图片", "无视频", "已删除", "删除", "私密", "不存在", "下架", @@ -201,6 +206,98 @@ def resolve( f"解析服务多次失败: {last_network_error}" ) + def fetch_wechat_direct(self, sph_code: str) -> dict: + """GET /direct，返回扁平 JSON（不解码文件头；日志/异常不含 CDN 直链）。""" + if not isinstance(sph_code, str) or not _WECHAT_SPH_CODE_RE.fullmatch(sph_code): + raise InvalidURLError(f"illegal wechat sph_code: {sph_code!r}") + + endpoint = f"{self.base_url}/api/stream/wechat_channels/{sph_code}/direct" + headers = {"X-API-Key": self.api_key, "Accept": "application/json"} + last_network_error: Optional[Exception] = None + + for attempt in range(1, self.max_retries + 1): + logger.info(f"wechat direct request attempt={attempt} sph_code={sph_code}") + try: + response = requests.get(endpoint, headers=headers, timeout=self.timeout) + except requests.RequestException as e: + last_network_error = e + logger.warning( + f"wechat direct unreachable attempt={attempt} sph_code={sph_code} err={type(e).__name__}" + ) + if attempt < self.max_retries: + time.sleep(self.retry_delay * attempt) + continue + raise NetworkError(f"wechat direct service unavailable sph_code={sph_code}") + + status = response.status_code + logger.info(f"wechat direct response status={status} sph_code={sph_code}") + if status == 401: + raise ResolverAuthError( + f"wechat direct auth failed(401) sph_code={sph_code}: {self._safe_text(response)}" + ) + if status == 400: + raise InvalidURLError( + f"wechat direct illegal sph_code(400) sph_code={sph_code}: {self._safe_text(response)}" + ) + if status >= 500: + logger.warning(f"wechat direct server error {status} attempt={attempt} sph_code={sph_code}") + if attempt < self.max_retries: + time.sleep(self.retry_delay * attempt) + continue + raise ResolverServerError( + f"wechat direct server error({status}) sph_code={sph_code}: {self._safe_text(response)}" + ) + if status != 200: + raise ResolverResponseError( + f"wechat direct unexpected status({status}) sph_code={sph_code}: {self._safe_text(response)}" + ) + try: + body = response.json() + except ValueError as e: + raise ResolverResponseError( + f"wechat direct response is not JSON sph_code={sph_code}: {e}" + ) + return self._validate_wechat_direct_body(sph_code, body) + + raise NetworkError( + f"wechat direct failed after retries sph_code={sph_code}: " + f"{type(last_network_error).__name__ if last_network_error else 'unknown'}" + ) + + def _validate_wechat_direct_body(self, sph_code: str, body: dict) -> dict: + """校验 /direct 扁平 JSON；异常消息不含 CDN 直链。""" + if not isinstance(body, dict) or body.get("sph_code") != sph_code: + raise ResolverResponseError(f"wechat direct invalid body sph_code={sph_code}") + content_length = self._require_positive_int(body.get("content_length"), "content_length", sph_code) + encrypted_head_bytes = self._require_positive_int( + body.get("encrypted_head_bytes"), "encrypted_head_bytes", sph_code + ) + head_b64 = body.get("head_b64") + if not isinstance(head_b64, str) or not head_b64: + raise ResolverResponseError(f"wechat direct missing head_b64 sph_code={sph_code}") + try: + head = base64.b64decode(head_b64, validate=True) + except ValueError as e: + raise ResolverResponseError( + f"wechat direct head_b64 is not standard base64 sph_code={sph_code}" + ) from e + if len(head) != encrypted_head_bytes: + raise ResolverResponseError( + f"wechat direct head_b64 length mismatch sph_code={sph_code} " + f"decoded={len(head)} encrypted_head_bytes={encrypted_head_bytes}" + ) + logger.info( + f"wechat direct fetch ok sph_code={sph_code} " + f"content_length={content_length} encrypted_head_bytes={encrypted_head_bytes}" + ) + return body + + @staticmethod + def _require_positive_int(value, field: str, sph_code: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ResolverResponseError(f"wechat direct invalid {field} sph_code={sph_code}") + return value + def _classify_failure(self, body: dict) -> None: """根据 error.code / 文案把 success=false 映射为终态异常（必抛）。""" error = body.get("error") or {} diff --git a/tests/unit/test_media_resolver_client.py b/tests/unit/test_media_resolver_client.py index 391c905..0658420 100644 --- a/tests/unit/test_media_resolver_client.py +++ b/tests/unit/test_media_resolver_client.py @@ -4,6 +4,9 @@ All console output is English only. """ +import base64 +import logging + import pytest import requests @@ -216,3 +219,73 @@ def test_top_level_not_object(self, monkeypatch): patch_post(monkeypatch, FakeResponse(200, ["a", "b"])) with pytest.raises(ResolverResponseError): make_client().resolve("http://page") + + +# --------------------------------------------------------------------------- # +# GET /direct (wechat channels plaintext head + CDN URL) +# --------------------------------------------------------------------------- # + + +_SECRET_CDN = "https://finder.video.qq.com/secret-token-ABCDEF123456.mp4" +_SPH = "AHaM8SrlXX" +_DIRECT_HEAD = b"\x00\x00\x00\x20" + b"ftyp" + b"\x00" * 24 + + +def _direct_payload(**overrides): + body = { + "sph_code": _SPH, + "cdn_url": _SECRET_CDN, + "content_length": 1000, + "encrypted_head_bytes": len(_DIRECT_HEAD), + "head_b64": base64.b64encode(_DIRECT_HEAD).decode("ascii"), + } + body.update(overrides) + return body + + +def patch_get(monkeypatch, *responses_or_exc): + seq = list(responses_or_exc) + calls = {"n": 0, "urls": [], "headers": []} + + def fake_get(url, headers=None, timeout=None, **kwargs): + calls["urls"].append(url) + calls["headers"].append(headers) + idx = min(calls["n"], len(seq) - 1) + calls["n"] += 1 + item = seq[idx] + if isinstance(item, Exception): + raise item + return item + + monkeypatch.setattr(requests, "get", fake_get) + return calls + + +class TestFetchWechatDirect: + def test_returns_flat_json_and_omits_cdn_from_logs(self, monkeypatch, caplog): + caplog.set_level(logging.DEBUG) + calls = patch_get(monkeypatch, FakeResponse(200, _direct_payload())) + out = make_client().fetch_wechat_direct(_SPH) + assert out["cdn_url"] == _SECRET_CDN + assert calls["urls"][0].endswith(f"/api/stream/wechat_channels/{_SPH}/direct") + assert calls["headers"][0]["X-API-Key"] == "k" + joined = " ".join(r.getMessage() for r in caplog.records) + assert _SECRET_CDN not in joined + + def test_401_auth(self, monkeypatch): + patch_get(monkeypatch, FakeResponse(401, text="unauthorized")) + with pytest.raises(ResolverAuthError) as ei: + make_client().fetch_wechat_direct(_SPH) + assert _SECRET_CDN not in str(ei.value) + + def test_502_retries_then_server_error(self, monkeypatch): + calls = patch_get(monkeypatch, FakeResponse(502, text="up"), FakeResponse(502, text="up")) + with pytest.raises(ResolverServerError): + make_client(max_retries=2).fetch_wechat_direct(_SPH) + assert calls["n"] == 2 + + def test_head_b64_length_mismatch(self, monkeypatch): + patch_get(monkeypatch, FakeResponse(200, _direct_payload(encrypted_head_bytes=8))) + with pytest.raises(ResolverResponseError) as ei: + make_client().fetch_wechat_direct(_SPH) + assert _SECRET_CDN not in str(ei.value) diff --git a/tests/unit/test_media_resolver_downloader.py b/tests/unit/test_media_resolver_downloader.py index 152794c..4f21a9a 100644 --- a/tests/unit/test_media_resolver_downloader.py +++ b/tests/unit/test_media_resolver_downloader.py @@ -26,6 +26,9 @@ def resolve(self, url, translate=False, force_refresh=False): raise item return item + def fetch_wechat_direct(self, sph_code): + raise AssertionError("fetch_wechat_direct should not be called in this test") + def make_downloader(responses, base_url="http://resolver.local:8000", api_key="secret-key"): dl = MediaResolverDownloader() @@ -249,30 +252,6 @@ def test_extract_video_id_wechat_channels(self): # --------------------------------------------------------------------------- # class TestConditionalDownloadHeaders: - def test_resolver_domain_includes_api_key_header(self, monkeypatch): - monkeypatch.setattr(BaseDownloader, "_validate_media_file", lambda self, path: True) - dl = make_downloader([WECHAT_DATA], api_key="my-secret-key") - di = dl.get_download_info("https://weixin.qq.com/sph/AOzokRxWHz") - - captured_headers = [] - - def fake_get(url, headers=None, stream=True, timeout=60): - captured_headers.append((url, headers)) - mock_resp = Mock() - mock_resp.raise_for_status.return_value = None - mock_resp.headers = {"Content-Length": "100"} - mock_resp.iter_content.return_value = [b"fake_mp4_bytes"] - return mock_resp - - import requests - monkeypatch.setattr(requests, "get", fake_get) - out = dl.download_file(di.download_url, di.filename) - assert out is not None - assert len(captured_headers) == 1 - url, headers = captured_headers[0] - assert url == "http://resolver.local:8000/api/stream/wechat_channels/AOzokRxWHz" - assert headers == {"X-API-Key": "my-secret-key"} - def test_cdn_domain_does_not_include_api_key_header(self, monkeypatch): monkeypatch.setattr(BaseDownloader, "_validate_media_file", lambda self, path: True) dl = make_downloader([DOUYIN_DATA], api_key="my-secret-key") @@ -299,12 +278,10 @@ def fake_get(url, headers=None, stream=True, timeout=60): def test_reresolve_applies_correct_headers_on_retry(self, monkeypatch): monkeypatch.setattr(BaseDownloader, "_validate_media_file", lambda self, path: True) - fresh_stream = dict( - WECHAT_DATA, - video_url="http://resolver.local:8000/api/stream/wechat_channels/AOzokRxWHz?retry=1", - ) - dl = make_downloader([WECHAT_DATA, fresh_stream], api_key="my-secret-key") - di = dl.get_download_info("https://weixin.qq.com/sph/AOzokRxWHz") + stale = dict(DOUYIN_DATA, video_url="https://cdn.example.com/v/7123-stale.mp4") + fresh = dict(DOUYIN_DATA, video_url="https://cdn.example.com/v/7123-fresh.mp4") + dl = make_downloader([stale, fresh], api_key="my-secret-key") + di = dl.get_download_info("https://www.douyin.com/video/7123") captured_headers = [] attempt = {"count": 0} @@ -312,7 +289,7 @@ def test_reresolve_applies_correct_headers_on_retry(self, monkeypatch): def fake_get(url, headers=None, stream=True, timeout=60): attempt["count"] += 1 captured_headers.append((url, headers)) - if attempt["count"] == 1: + if "stale" in url: import requests raise requests.exceptions.HTTPError("403 Forbidden") mock_resp = Mock() @@ -326,6 +303,6 @@ def fake_get(url, headers=None, stream=True, timeout=60): out = dl.download_file(di.download_url, di.filename, max_retries=1) assert out is not None assert len(captured_headers) >= 2 - fresh_call_headers = [h for u, h in captured_headers if "retry=1" in u] + fresh_call_headers = [h for u, h in captured_headers if "fresh" in u] assert fresh_call_headers - assert fresh_call_headers[0] == {"X-API-Key": "my-secret-key"} + assert fresh_call_headers[0] is None or "X-API-Key" not in (fresh_call_headers[0] or {}) diff --git a/tests/unit/test_media_resolver_wechat_direct.py b/tests/unit/test_media_resolver_wechat_direct.py new file mode 100644 index 0000000..89433a3 --- /dev/null +++ b/tests/unit/test_media_resolver_wechat_direct.py @@ -0,0 +1,172 @@ +"""Wechat /direct + CDN Range splice tests. HTTP mocked. English only.""" +import base64, logging +from unittest.mock import Mock +import pytest, requests +from video_transcript_api.downloaders.media_resolver import MediaResolverDownloader +from video_transcript_api.errors import DownloadFailedError, ResolverAuthError + +SPH = "AHaM8SrlXX" +STREAM = f"http://resolver.local:8000/api/stream/wechat_channels/{SPH}" +CDN_A = "https://finder.video.qq.com/secret-token-AAAA1111.mp4" +CDN_B = "https://finder.video.qq.com/secret-token-BBBB2222.mp4" +HEAD = b"\x00\x00\x00\x18ftypisom" + b"\x00" * 8 +SECRETS = (CDN_A, CDN_B, "secret-token-AAAA", "secret-token-BBBB") +def _payload(head, rest, cdn=CDN_A): + body = head + rest + return { + "sph_code": SPH, "cdn_url": cdn, "content_length": len(body), + "encrypted_head_bytes": len(head), + "head_b64": base64.b64encode(head).decode("ascii"), + }, body +def _resp(status, start, total, chunks, error=None): + cr = f"bytes {start}-{start}/*" if total == "*" else f"bytes {start}-{total - 1}/{total}" + r = Mock(status_code=status, headers={"Content-Range": cr}) + r.close.return_value = None + if error: + def _iter(chunk_size=8192): + yield from chunks + raise error + r.iter_content.side_effect = _iter + else: + r.iter_content.return_value = list(chunks) + return r +def _status(code): + r = Mock(status_code=code, headers={}) + r.iter_content.return_value = []; r.close.return_value = None; return r +def _dl(tmp_path, payloads, monkeypatch, getter=None, max_bytes=0): + class TM: + def create_temp_file(self, suffix=".mp4"): + p = tmp_path / f"w{suffix}"; p.write_bytes(b""); return str(p) + def untrack_file(self, path): + return None + class Client: + def __init__(self): + self.calls = [] + def fetch_wechat_direct(self, sph_code): + self.calls.append(sph_code) + item = payloads[min(len(self.calls) - 1, len(payloads) - 1)] + if isinstance(item, Exception): + raise item + return item + dl = MediaResolverDownloader() + dl._resolver_netloc = "resolver.local:8000" + dl.client = Client() + dl.temp_manager = TM() + if max_bytes: + dl.max_download_bytes = max_bytes + monkeypatch.setattr("video_transcript_api.downloaders.media_resolver.validate_url_safe", lambda u: u) + if getter is not None: + monkeypatch.setattr(requests, "get", getter) + return dl + +def _no_secret(caplog, exc=None): + text = " ".join(r.getMessage() for r in caplog.records) + for s in SECRETS: + assert s not in text + if exc is not None: + assert s not in str(exc) + +class TestWechatDirectSplice: + def test_happy_path_head_plus_one_206(self, monkeypatch, tmp_path, caplog): + caplog.set_level(logging.DEBUG) + rest = b"REST-BYTES-OK" + payload, expected = _payload(HEAD, rest) + calls = [] + def getter(url, headers=None, stream=True, timeout=None, **kw): + calls.append((url, dict(headers or {}), timeout, kw.get("allow_redirects"))) + return _resp(206, len(HEAD), len(expected), [rest]) + out = _dl(tmp_path, [payload], monkeypatch, getter).download_file(STREAM, "x.mp4") + assert open(out, "rb").read() == expected + assert calls == [(CDN_A, {"Range": f"bytes={len(HEAD)}-"}, (10, 60), False)] + assert "X-API-Key" not in calls[0][1] + _no_secret(caplog) + + def test_head_only_does_not_hit_cdn(self, monkeypatch, tmp_path): + payload, expected = _payload(HEAD, b"") + getter = lambda *a, **k: pytest.fail("CDN must not be called") + out = _dl(tmp_path, [payload], monkeypatch, getter).download_file(STREAM, "x.mp4") + assert open(out, "rb").read() == expected + + def test_interrupt_then_refresh_resumes_range(self, monkeypatch, tmp_path, caplog): + caplog.set_level(logging.DEBUG) + rest = b"ABCDEFGHIJKLMNOP" + first, expected = _payload(HEAD, rest, CDN_A) + second, _ = _payload(HEAD, rest, CDN_B) + calls, n = [], {"i": 0} + def getter(url, headers=None, stream=True, timeout=None, **kw): + n["i"] += 1 + calls.append((url, (headers or {}).get("Range"), kw.get("allow_redirects"))) + if n["i"] == 1: + return _resp(206, len(HEAD), len(expected), [rest[:6]], + error=requests.exceptions.ConnectionError("cut")) + return _resp(206, len(HEAD) + 6, len(expected), [rest[6:]]) + dl = _dl(tmp_path, [first, second], monkeypatch, getter) + assert open(dl.download_file(STREAM, "x.mp4"), "rb").read() == expected + assert calls[1] == (CDN_B, f"bytes={len(HEAD) + 6}-", False) + _no_secret(caplog) + + @pytest.mark.parametrize("mode", ["410", "bad_range", "star_total", "wrong_total", "302"]) + def test_cdn_error_then_refresh_succeeds(self, monkeypatch, tmp_path, mode): + rest = b"BODY" + first, expected = _payload(HEAD, rest, CDN_A) + second, _ = _payload(HEAD, rest, CDN_B) + n = {"i": 0} + def getter(url, headers=None, stream=True, timeout=None, **kw): + n["i"] += 1 + n["redir"] = kw.get("allow_redirects") + if n["i"] > 1: + return _resp(206, len(HEAD), len(expected), [rest]) + return { + "410": _status(410), "302": _status(302), + "star_total": _resp(206, len(HEAD), "*", [b"NO"]), + "wrong_total": _resp(206, len(HEAD), len(expected) + 99, [b"NO"]), + "bad_range": _resp(206, 0, len(expected), [b"NO"]), + }[mode] + dl = _dl(tmp_path, [first, second], monkeypatch, getter) + assert open(dl.download_file(STREAM, "x.mp4"), "rb").read() == expected + assert dl.client.calls == [SPH, SPH] and n["redir"] is False + + @pytest.mark.parametrize("field", ["content_length", "head_b64"]) + def test_refresh_identity_mismatch(self, monkeypatch, tmp_path, field): + first, _ = _payload(HEAD, b"BODY", CDN_A) + second, _ = _payload(HEAD, b"BODY", CDN_B) + second[field] = first[field] + 1 if field == "content_length" else "AAAA" + with pytest.raises(DownloadFailedError) as ei: + _dl(tmp_path, [first, second], monkeypatch, + lambda *a, **k: _status(410)).download_file(STREAM, "x.mp4") + assert SPH in str(ei.value) and field in str(ei.value) + assert not (tmp_path / "w.mp4").exists() + assert all(s not in str(ei.value) for s in SECRETS) + + def test_final_size_mismatch_raises(self, monkeypatch, tmp_path, caplog): + caplog.set_level(logging.DEBUG) + payload, _ = _payload(HEAD, b"PARTIAL") + payload["content_length"] = payload["encrypted_head_bytes"] + 50 + with pytest.raises(DownloadFailedError) as ei: + _dl(tmp_path, [payload] * 8, monkeypatch, lambda *a, **k: _status(410)).download_file(STREAM, "x.mp4") + _no_secret(caplog, ei.value) + + @pytest.mark.parametrize("kind", ["oversize", "noftyp", "auth"]) + def test_fail_before_cdn(self, monkeypatch, tmp_path, caplog, kind): + caplog.set_level(logging.DEBUG) + getter = lambda *a, **k: pytest.fail("CDN must not be called") + if kind == "oversize": + payload, expected = _payload(HEAD, b"X" * 50) + with pytest.raises(DownloadFailedError): + _dl(tmp_path, [payload], monkeypatch, getter, max_bytes=len(expected) - 1).download_file(STREAM, "x.mp4") + elif kind == "noftyp": + payload, _ = _payload(b"\x00\x00\x00\x18XXXX" + b"\x00" * 16, b"REST") + with pytest.raises(DownloadFailedError) as ei: + _dl(tmp_path, [payload], monkeypatch, getter).download_file(STREAM, "x.mp4") + assert "ftyp" in str(ei.value).lower() + _no_secret(caplog, ei.value) + else: + with pytest.raises(ResolverAuthError) as ei: + _dl(tmp_path, [ResolverAuthError("no key")], monkeypatch).download_file(STREAM, "x.mp4") + _no_secret(caplog, ei.value) + + def test_non_wechat_url_does_not_call_direct(self, monkeypatch, tmp_path): + dl = _dl(tmp_path, [], monkeypatch) + monkeypatch.setattr("video_transcript_api.downloaders.base.BaseDownloader.download_file", lambda *a, **k: "/tmp/cdn-ok.mp4") + assert dl.download_file("https://cdn.example.com/v/7123.mp4", "x.mp4") == "/tmp/cdn-ok.mp4" + assert dl.client.calls == []