From 369dea582cc53956b23233fce92d41fb46905a67 Mon Sep 17 00:00:00 2001
From: DrSmoothl <1787882683@qq.com>
Date: Tue, 30 Jun 2026 04:04:05 +0800
Subject: [PATCH] feat(onebot): download stream actions (#163 phase 3)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

NapCat-compatible server-push download: one request answered by a file_info
header, N base64 file_chunk frames, then a terminal file_complete. Bytes come
from a local file under the stream temp root, an http(s) URL, or a cached
image/voice id resolved to its QQ CDN URL. test_download_stream pushes 10
canned frames so a client can validate the transport without touching QQ.

Actions: download_file_stream / download_file_image_stream /
download_file_record_stream / test_download_stream.

Hardened beyond the NapCat original (which streams any local path and follows
redirects to anywhere), per three adversarial review passes:
- arbitrary-file-read: local reads are fenced inside the stream temp root,
  realpath-resolved so a symlink can't point out.
- SSRF: every URL (incl. each redirect hop, followed manually) is guarded —
  http(s) only, and neither the literal host nor any resolved address may be
  private/loopback/link-local; IPv6 is byte-parsed so normalized v4-mapped
  literals (::ffff:a9fe:a9fe) can't slip past. (DNS rebinding remains a
  documented residual — needs connect-time peer-IP validation.)
- robustness: TTFB-only connect timeout (doesn't cut a large body mid-stream)
  plus a per-chunk idle watchdog that aborts a stalled source so a hung
  upstream can't pin a socket; total size capped; non-2xx bodies cancelled;
  client disconnect aborts via the sink.

501 onebot tests green.
---
 .../onebot/src/actions/stream-download.ts     | 379 ++++++++++++++++++
 packages/onebot/src/actions/stream-file.ts    |   2 +-
 packages/onebot/src/api-handler.ts            |   2 +
 packages/onebot/tests/stream-download.test.ts | 116 ++++++
 4 files changed, 498 insertions(+), 1 deletion(-)
 create mode 100644 packages/onebot/src/actions/stream-download.ts
 create mode 100644 packages/onebot/tests/stream-download.test.ts

diff --git a/packages/onebot/src/actions/stream-download.ts b/packages/onebot/src/actions/stream-download.ts
new file mode 100644
index 00000000..7e2ca27d
--- /dev/null
+++ b/packages/onebot/src/actions/stream-download.ts
@@ -0,0 +1,379 @@
+// Stream API download actions (#163) — NapCat-compatible server-push streaming.
+//
+// download_file_stream / _image_stream / _record_stream answer one request with
+// many frames: a `file_info` header, then N `file_chunk` frames (base64), then a
+// terminal `file_complete`. The bytes come from one of three sources, resolved
+// from the `file` arg:
+//   • a local file UNDER the stream temp root (e.g. something a client just
+//     pushed via upload_file_stream) — streamed straight off disk;
+//   • an http(s) URL — fetched and re-streamed;
+//   • a cached image/voice id from a received message — resolved to its QQ CDN
+//     URL, then fetched.
+// test_download_stream pushes 10 canned frames so a client can verify the
+// streaming transport without touching QQ.
+//
+// Hardening beyond the NapCat original (which streams ANY local path and
+// fetches ANY URL): local reads are fenced inside the stream temp root (no
+// arbitrary-file read via a leaked token), and URL fetches are SSRF-guarded
+// (no private/loopback/link-local targets) + size-capped.
+
+import fs from 'fs';
+import path from 'path';
+import net from 'net';
+import dns from 'dns/promises';
+import { resolveLocalFilePath } from '@snowluma/protocol/highway/utils';
+import { defineStreamAction, f } from '../action-kit';
+import type { ApiActionContext, ApiHandler } from '../api-handler';
+import { okResponse, type ApiResponse, type JsonObject, type JsonValue } from '../types';
+import { type StreamSink, StreamStatus } from '../streaming';
+import { STREAM_ROOT } from './stream-file';
+
+const DEFAULT_CHUNK_BYTES = 64 * 1024;
+const MAX_CHUNK_REQUEST_BYTES = 16 * 1024 * 1024;     // cap the client-requested chunk size
+const MAX_DOWNLOAD_BYTES = 4 * 1024 * 1024 * 1024;    // 4 GiB total
+const FETCH_CONNECT_TIMEOUT_MS = 30_000;              // TTFB only — cleared once headers arrive (not the whole transfer)
+const IDLE_TIMEOUT_MS = 60_000;                       // abort if no chunk arrives for this long (anti-stall)
+const MAX_REDIRECTS = 5;
+const DOWNLOAD_USER_AGENT =
+  'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36';
+
+class DownloadError extends Error {}
+
+// ─────────────────────────── source resolution ───────────────────────────
+
+function isWithinStreamRoot(p: string): boolean {
+  // Resolve the root through any symlinks too (e.g. macOS /tmp → /private/tmp),
+  // so a realpath'd target compares against a realpath'd root — otherwise a
+  // legitimate file reads as "outside" purely from the tmpdir symlink.
+  let root = path.resolve(STREAM_ROOT);
+  try { root = fs.realpathSync(STREAM_ROOT); } catch { /* root may not exist yet */ }
+  const rel = path.relative(root, path.resolve(p));
+  return rel !== '' && !rel.startsWith('..') && !path.isAbsolute(rel);
+}
+
+function isPrivateV4(o: number[]): boolean {
+  return (
+    o[0] === 0 || o[0] === 127 || o[0] === 10 ||
+    (o[0] === 192 && o[1] === 168) ||
+    (o[0] === 172 && o[1] >= 16 && o[1] <= 31) ||
+    (o[0] === 169 && o[1] === 254) ||        // link-local
+    (o[0] === 100 && o[1] >= 64 && o[1] <= 127) // CGNAT
+  );
+}
+
+/** Parse an IPv6 string (any form, incl. `::` compression and embedded IPv4)
+ *  to its 16 bytes. `new URL().hostname` normalises v4-mapped addresses to hex
+ *  (`::ffff:7f00:1`), so string matching is unreliable — we work on bytes. */
+function ipv6ToBytes(input: string): Uint8Array | null {
+  let s = input.replace(/^\[|\]$/g, '');
+  if (net.isIP(s) !== 6) return null;
+  const v4m = /:(\d{1,3}(?:\.\d{1,3}){3})$/.exec(s); // embedded IPv4 tail
+  if (v4m) {
+    const o = v4m[1].split('.').map(Number);
+    if (o.some((n) => n > 255)) return null;
+    s = s.slice(0, v4m.index) + ':' + `${((o[0] << 8) | o[1]).toString(16)}:${((o[2] << 8) | o[3]).toString(16)}`;
+  }
+  const halves = s.split('::');
+  if (halves.length > 2) return null;
+  const head = halves[0] ? halves[0].split(':') : [];
+  const tail = halves.length === 2 && halves[1] ? halves[1].split(':') : [];
+  let groups: string[];
+  if (halves.length === 2) {
+    const missing = 8 - head.length - tail.length;
+    if (missing < 0) return null;
+    groups = [...head, ...new Array(missing).fill('0'), ...tail];
+  } else {
+    groups = head;
+  }
+  if (groups.length !== 8) return null;
+  const bytes = new Uint8Array(16);
+  for (let i = 0; i < 8; i++) {
+    const g = parseInt(groups[i] || '0', 16);
+    if (Number.isNaN(g) || g < 0 || g > 0xffff) return null;
+    bytes[i * 2] = g >> 8;
+    bytes[i * 2 + 1] = g & 0xff;
+  }
+  return bytes;
+}
+
+function isPrivateIp(ip: string): boolean {
+  const host = ip.replace(/^\[|\]$/g, '');
+  const fam = net.isIP(host);
+  if (fam === 4) return isPrivateV4(host.split('.').map(Number));
+  if (fam === 6) {
+    const b = ipv6ToBytes(host);
+    if (!b) return true; // unparseable IPv6 → treat as unsafe
+    if (b.every((x) => x === 0)) return true;                                  // ::
+    if (b.slice(0, 15).every((x) => x === 0) && b[15] === 1) return true;       // ::1
+    if ((b[0] & 0xfe) === 0xfc) return true;                                    // fc00::/7 unique-local
+    if (b[0] === 0xfe && (b[1] & 0xc0) === 0x80) return true;                   // fe80::/10 link-local
+    // v4-mapped ::ffff:a.b.c.d / v4-compatible ::a.b.c.d
+    if (b.slice(0, 10).every((x) => x === 0) && ((b[10] === 0xff && b[11] === 0xff) || (b[10] === 0 && b[11] === 0))) {
+      return isPrivateV4([b[12], b[13], b[14], b[15]]);
+    }
+    // NAT64 64:ff9b::/96
+    if (b[0] === 0x00 && b[1] === 0x64 && b[2] === 0xff && b[3] === 0x9b && b.slice(4, 12).every((x) => x === 0)) {
+      return isPrivateV4([b[12], b[13], b[14], b[15]]);
+    }
+    return false;
+  }
+  return false;
+}
+
+/** Parse + SSRF-guard a URL: http(s) only, and neither the literal host nor any
+ *  DNS-resolved address may be private/loopback/link-local. Not bulletproof
+ *  against DNS rebinding, but blocks the obvious metadata-endpoint / internal
+ *  service reach. */
+async function assertSafeUrl(raw: string): Promise<URL> {
+  let u: URL;
+  try { u = new URL(raw); } catch { throw new DownloadError('invalid URL'); }
+  if (u.protocol !== 'http:' && u.protocol !== 'https:') throw new DownloadError('only http(s) URLs are allowed');
+  const host = u.hostname.replace(/^\[|\]$/g, '');
+  if (net.isIP(host)) {
+    if (isPrivateIp(host)) throw new DownloadError('refusing to fetch a private/loopback address');
+    return u;
+  }
+  let addrs: Array<{ address: string }>;
+  try { addrs = await dns.lookup(host, { all: true }); } catch { throw new DownloadError(`cannot resolve host: ${host}`); }
+  for (const a of addrs) {
+    if (isPrivateIp(a.address)) throw new DownloadError('refusing to fetch a host that resolves to a private address');
+  }
+  return u;
+}
+
+interface DownloadSource {
+  name: string;
+  size: number; // 0 when unknown
+  iterable: AsyncIterable<Uint8Array>;
+  /** Force the underlying source closed (idle-timeout watchdog / abort). */
+  abort?: () => void;
+}
+
+const asStr = (v: JsonValue | undefined): string => (typeof v === 'string' ? v : '');
+const asNum = (v: JsonValue | undefined): number => {
+  const n = Number(v);
+  return Number.isFinite(n) && n > 0 ? n : 0;
+};
+
+/** Follow redirects MANUALLY, SSRF-guarding every hop. Default fetch follows
+ *  redirects automatically and would NOT re-check the target, so a public URL
+ *  could 302 to a private/metadata address — re-validating each Location closes
+ *  that. Each hop gets a TTFB-only timeout (cleared once headers arrive, so a
+ *  legitimate slow/large body isn't cut mid-transfer); the final hop's
+ *  AbortController is returned so the caller's idle watchdog can kill a stalled
+ *  body. (DNS rebinding between our check and undici's own connect remains a
+ *  known residual — it needs connect-time peer-IP validation to fully close.) */
+async function fetchGuarded(rawUrl: string): Promise<{ resp: Response; abort: () => void }> {
+  let current = rawUrl;
+  for (let hop = 0; hop <= MAX_REDIRECTS; hop++) {
+    const u = await assertSafeUrl(current);
+    const ac = new AbortController();
+    const t = setTimeout(() => ac.abort(), FETCH_CONNECT_TIMEOUT_MS);
+    let resp: Response;
+    try {
+      resp = await fetch(u, { headers: { 'User-Agent': DOWNLOAD_USER_AGENT }, redirect: 'manual', signal: ac.signal });
+    } finally {
+      clearTimeout(t);
+    }
+    if (resp.status >= 300 && resp.status < 400) {
+      const loc = resp.headers.get('location');
+      await resp.body?.cancel().catch(() => { /* ignore */ });
+      if (!loc) throw new DownloadError(`download failed: ${resp.status} redirect without Location`);
+      current = new URL(loc, u).toString();
+      continue;
+    }
+    return { resp, abort: () => ac.abort() };
+  }
+  throw new DownloadError('download failed: too many redirects');
+}
+
+async function openUrl(rawUrl: string, name: string, knownSize: number): Promise<DownloadSource> {
+  const { resp, abort } = await fetchGuarded(rawUrl);
+  if (!resp.ok) {
+    await resp.body?.cancel().catch(() => { /* ignore */ });
+    throw new DownloadError(`download failed: HTTP ${resp.status}`);
+  }
+  const declared = Number(resp.headers.get('content-length') ?? '');
+  const size = (Number.isFinite(declared) && declared > 0 ? declared : 0) || knownSize;
+  if (!resp.body) throw new DownloadError('download failed: empty response body');
+  return { name: name || urlBasename(rawUrl), size, iterable: resp.body as unknown as AsyncIterable<Uint8Array>, abort };
+}
+
+function urlBasename(u: string): string {
+  try {
+    const name = path.basename(new URL(u).pathname);
+    return name && name !== '/' ? decodeURIComponent(name) : 'download.bin';
+  } catch { return 'download.bin'; }
+}
+
+async function resolveDownload(
+  target: string,
+  ctx: ApiActionContext,
+  prefer: 'auto' | 'image' | 'record',
+  chunkSize: number,
+): Promise<DownloadSource> {
+  // 1. local file — but ONLY a real existing file, fenced inside the stream
+  //    temp root. `resolveLocalFilePath` returns the input verbatim for any
+  //    non-URL string (including cached-media ids), so we must NOT treat a
+  //    non-existent "path" as local — that would shadow the cached-media branch
+  //    below (the whole point of the *_image/_record_stream actions).
+  const local = resolveLocalFilePath(target);
+  if (local) {
+    let st: fs.Stats | null = null;
+    try { st = fs.statSync(local); } catch { st = null; }
+    if (st?.isFile()) {
+      // Fence against the *real* path so a symlink inside the root can't point
+      // out (lexical resolve alone would follow it).
+      let real = local;
+      try { real = fs.realpathSync(local); } catch { /* keep lexical */ }
+      if (!isWithinStreamRoot(real)) throw new DownloadError('local downloads are restricted to the stream temp directory');
+      const rs = fs.createReadStream(real, { highWaterMark: chunkSize });
+      return { name: path.basename(real), size: st.size, iterable: rs, abort: () => rs.destroy() };
+    }
+  }
+
+  // 2. explicit http(s) URL.
+  if (/^https?:\/\//i.test(target)) return openUrl(target, '', 0);
+
+  // 3. cached image / voice id → QQ CDN URL.
+  let info: JsonObject | null = null;
+  if (prefer !== 'record') info = await ctx.getImageInfo(target);
+  if (!info && prefer !== 'image') info = await ctx.getRecordInfo(target);
+  if (info) {
+    const url = asStr(info.url) || asStr(info.file);
+    if (/^https?:\/\//i.test(url)) return openUrl(url, asStr(info.file_name), asNum(info.file_size));
+  }
+
+  throw new DownloadError('file not found (expected a stream-dir path, an http(s) URL, or a cached image/voice id)');
+}
+
+// ─────────────────────────── push engine ───────────────────────────
+
+async function streamChunks(
+  sink: StreamSink,
+  iterable: AsyncIterable<Uint8Array>,
+  totalSize: number,
+  dataType: string,
+  abort?: () => void,
+): Promise<{ totalChunks: number; totalBytes: number }> {
+  let index = 0;
+  let bytesRead = 0;
+  // Anti-stall: if no chunk arrives for IDLE_TIMEOUT_MS, force the source closed
+  // so a server that sends headers then hangs can't pin a socket forever (the
+  // sink's own liveness check only fires when we actually send a frame).
+  let idle: NodeJS.Timeout | null = null;
+  const armIdle = (): void => {
+    if (idle) clearTimeout(idle);
+    idle = setTimeout(() => { abort?.(); }, IDLE_TIMEOUT_MS);
+    if (idle.unref) idle.unref();
+  };
+  armIdle();
+  try {
+    for await (const part of iterable) {
+      armIdle();
+      const buf = Buffer.isBuffer(part) ? part : Buffer.from(part);
+      bytesRead += buf.length;
+      if (bytesRead > MAX_DOWNLOAD_BYTES) throw new DownloadError(`download exceeds size limit (${MAX_DOWNLOAD_BYTES} bytes)`);
+      const b64 = buf.toString('base64');
+      await sink.send({
+        type: StreamStatus.Stream,
+        data_type: dataType,
+        index,
+        data: b64,
+        size: buf.length,
+        progress: totalSize > 0 ? Math.min(100, Math.round((bytesRead / totalSize) * 100)) : 0,
+        base64_size: b64.length,
+      });
+      index++;
+    }
+  } finally {
+    if (idle) clearTimeout(idle);
+  }
+  return { totalChunks: index, totalBytes: bytesRead };
+}
+
+async function runDownload(
+  target: string,
+  chunkSizeReq: number | undefined,
+  ctx: ApiActionContext,
+  sink: StreamSink,
+  prefer: 'auto' | 'image' | 'record',
+): Promise<ApiResponse> {
+  if (!target) throw new DownloadError('file is required');
+  let chunkSize = chunkSizeReq && chunkSizeReq > 0 ? chunkSizeReq : DEFAULT_CHUNK_BYTES;
+  if (chunkSize > MAX_CHUNK_REQUEST_BYTES) chunkSize = MAX_CHUNK_REQUEST_BYTES;
+
+  const src = await resolveDownload(target, ctx, prefer, chunkSize);
+  await sink.send({
+    type: StreamStatus.Stream,
+    data_type: 'file_info',
+    file_name: src.name,
+    file_size: src.size,
+    chunk_size: chunkSize,
+  });
+  const { totalChunks, totalBytes } = await streamChunks(sink, src.iterable, src.size, 'file_chunk', src.abort);
+  return okResponse({
+    type: StreamStatus.Response,
+    data_type: 'file_complete',
+    file_name: src.name,
+    total_chunks: totalChunks,
+    total_bytes: totalBytes,
+    message: 'Download completed',
+  });
+}
+
+// ─────────────────────────── action definitions ───────────────────────────
+
+const downloadParams = {
+  file: f.string().optional().describe('文件路径(限 stream 临时目录)/ http(s) URL'),
+  file_id: f.string().optional().describe('文件 ID(缓存的图片/语音 id)'),
+  chunk_size: f.int({ min: 1 }).optional().describe('分块大小(字节,默认 64KB)'),
+};
+
+const downloadFileStream = defineStreamAction({
+  name: 'download_file_stream',
+  summary: '以流式方式下载文件(stream 目录本地文件 / URL / 缓存媒体)',
+  returns: '流式帧:file_info → file_chunk* → file_complete',
+  params: downloadParams,
+  run: (p, ctx, _raw, sink) => runDownload(p.file || p.file_id || '', p.chunk_size, ctx, sink, 'auto'),
+});
+
+const downloadFileImageStream = defineStreamAction({
+  name: 'download_file_image_stream',
+  summary: '以流式方式下载图片(缓存图片 id / URL / stream 目录本地文件)',
+  returns: '流式帧:file_info → file_chunk* → file_complete',
+  params: downloadParams,
+  run: (p, ctx, _raw, sink) => runDownload(p.file || p.file_id || '', p.chunk_size, ctx, sink, 'image'),
+});
+
+const downloadFileRecordStream = defineStreamAction({
+  name: 'download_file_record_stream',
+  summary: '以流式方式下载语音(缓存语音 id / URL / stream 目录本地文件)',
+  returns: '流式帧:file_info → file_chunk* → file_complete',
+  params: downloadParams,
+  run: (p, ctx, _raw, sink) => runDownload(p.file || p.file_id || '', p.chunk_size, ctx, sink, 'record'),
+});
+
+const testDownloadStream = defineStreamAction({
+  name: 'test_download_stream',
+  summary: '测试下载流(推送 10 个数据帧,验证流式传输,不触达 QQ)',
+  returns: '流式帧:data_chunk*10 → data_complete(error=true 时以 error 帧结束)',
+  params: { error: f.bool().default(false).describe('是否触发测试错误') },
+  run: async (p, _ctx, _raw, sink) => {
+    for (let i = 0; i < 10; i++) {
+      await sink.send({ type: StreamStatus.Stream, data: `Index-> ${i + 1}`, data_type: 'data_chunk' });
+    }
+    if (p.error) throw new DownloadError('This is a test error');
+    return okResponse({ type: StreamStatus.Response, data_type: 'data_complete', data: 'Stream transmission complete' });
+  },
+});
+
+export function register(h: ApiHandler, ctx: ApiActionContext): void {
+  downloadFileStream.register(h, ctx);
+  downloadFileImageStream.register(h, ctx);
+  downloadFileRecordStream.register(h, ctx);
+  testDownloadStream.register(h, ctx);
+}
+
+// Exposed for tests.
+export const __test = { resolveDownload, isWithinStreamRoot, isPrivateIp, assertSafeUrl, streamChunks, runDownload };
diff --git a/packages/onebot/src/actions/stream-file.ts b/packages/onebot/src/actions/stream-file.ts
index 5716e7b0..7e5b6a8f 100644
--- a/packages/onebot/src/actions/stream-file.ts
+++ b/packages/onebot/src/actions/stream-file.ts
@@ -29,7 +29,7 @@ import { StreamStatus } from '../streaming';
 
 const log = createLogger('OneBot.Stream');
 
-const STREAM_ROOT = path.join(os.tmpdir(), 'snowluma-stream');
+export const STREAM_ROOT = path.join(os.tmpdir(), 'snowluma-stream');
 export const STREAM_UPLOAD_DIR = path.join(STREAM_ROOT, 'upload');
 export const STREAM_DOWNLOAD_DIR = path.join(STREAM_ROOT, 'download');
 
diff --git a/packages/onebot/src/api-handler.ts b/packages/onebot/src/api-handler.ts
index 1cc40c85..743c393b 100644
--- a/packages/onebot/src/api-handler.ts
+++ b/packages/onebot/src/api-handler.ts
@@ -12,6 +12,7 @@ import { register as registerMessage } from './actions/message';
 import { register as registerQzone } from './actions/qzone';
 import { register as registerRequest } from './actions/request';
 import { register as registerStreamFile } from './actions/stream-file';
+import { register as registerStreamDownload } from './actions/stream-download';
 import type { ForwardPreviewMeta } from './modules/message-actions';
 import type { JsonObject, JsonValue, MessageMeta } from './types';
 import { RETCODE, failedResponse } from './types';
@@ -122,6 +123,7 @@ export class ApiHandler {
     registerGroupAlbum(this, context);
     registerQzone(this, context);
     registerStreamFile(this, context);
+    registerStreamDownload(this, context);
   }
 
   registerAction(action: string, handler: ActionHandler): void {
diff --git a/packages/onebot/tests/stream-download.test.ts b/packages/onebot/tests/stream-download.test.ts
new file mode 100644
index 00000000..0fff50e1
--- /dev/null
+++ b/packages/onebot/tests/stream-download.test.ts
@@ -0,0 +1,116 @@
+// Phase 3 — download stream actions (#163): server-push multi-frame.
+import { describe, it, expect, afterEach } from 'vitest';
+import fs from 'fs';
+import path from 'path';
+import { ApiHandler, type ApiActionContext } from '../src/api-handler';
+import { __test } from '../src/actions/stream-download';
+import { STREAM_DOWNLOAD_DIR, __test as fileTest } from '../src/actions/stream-file';
+
+const parse = (frames: string[]) => frames.map((f) => JSON.parse(f) as Record<string, any>);
+
+afterEach(() => {
+  fileTest.cleanDir(STREAM_DOWNLOAD_DIR);
+});
+
+describe('test_download_stream', () => {
+  it('pushes 10 data frames then a data_complete terminal', async () => {
+    const h = new ApiHandler({} as ApiActionContext);
+    expect(h.isStreamAction('test_download_stream')).toBe(true);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'test_download_stream', params: {}, echo: 'T' }), (j) => frames.push(j));
+
+    const got = parse(frames);
+    expect(got).toHaveLength(11);
+    expect(got[0]).toMatchObject({ stream: 'stream-action', echo: 'T', data: { type: 'stream', data_type: 'data_chunk', data: 'Index-> 1' } });
+    expect(got[9]).toMatchObject({ data: { data: 'Index-> 10' } });
+    expect(got[10]).toMatchObject({ status: 'ok', data: { type: 'response', data_type: 'data_complete', data: 'Stream transmission complete' } });
+  });
+
+  it('error=true streams the 10 frames then terminates with an error frame', async () => {
+    const h = new ApiHandler({} as ApiActionContext);
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'test_download_stream', params: { error: true } }), (j) => frames.push(j));
+    const got = parse(frames);
+    expect(got).toHaveLength(11);
+    expect(got[9]).toMatchObject({ data: { data_type: 'data_chunk' } });
+    expect(got[10]).toMatchObject({ status: 'failed', data: { type: 'error' } });
+  });
+});
+
+describe('download_file_stream (local, fenced)', () => {
+  it('streams file_info → file_chunk* → file_complete and reassembles exactly', async () => {
+    fs.mkdirSync(STREAM_DOWNLOAD_DIR, { recursive: true });
+    const fp = path.join(STREAM_DOWNLOAD_DIR, 'dl.bin');
+    const content = Buffer.from('snowluma stream download payload '.repeat(4000)); // ~128 KB → multiple 64 KB chunks
+    fs.writeFileSync(fp, content);
+
+    const h = new ApiHandler({} as ApiActionContext);
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'download_file_stream', params: { file: fp, chunk_size: 65536 }, echo: 'D' }), (j) => frames.push(j));
+
+    const got = parse(frames);
+    expect(got[0]).toMatchObject({ stream: 'stream-action', echo: 'D', data: { data_type: 'file_info', file_name: 'dl.bin', file_size: content.length, chunk_size: 65536 } });
+
+    const chunks = got.filter((p) => p.data?.data_type === 'file_chunk');
+    expect(chunks.length).toBeGreaterThan(1);
+    // indices are sequential and progress is monotonic
+    chunks.forEach((c, i) => expect(c.data.index).toBe(i));
+
+    const last = got[got.length - 1];
+    expect(last).toMatchObject({ status: 'ok', data: { type: 'response', data_type: 'file_complete', total_bytes: content.length, total_chunks: chunks.length } });
+
+    const reassembled = Buffer.concat(chunks.map((c) => Buffer.from(c.data.data, 'base64')));
+    expect(reassembled.equals(content)).toBe(true);
+  });
+
+  it('refuses an arbitrary local path outside the stream root [arbitrary-read guard]', async () => {
+    await expect(__test.resolveDownload('/etc/passwd', {} as ApiActionContext, 'auto', 65536)).rejects.toThrow(/restricted to the stream temp/i);
+    await expect(__test.resolveDownload('file:///etc/hosts', {} as ApiActionContext, 'auto', 65536)).rejects.toThrow(/restricted to the stream temp/i);
+
+    // and end-to-end the dispatch surfaces it as an error frame, not a leak
+    const h = new ApiHandler({} as ApiActionContext);
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'download_file_stream', params: { file: '/etc/passwd' } }), (j) => frames.push(j));
+    expect(parse(frames).at(-1)).toMatchObject({ status: 'failed', data: { type: 'error' } });
+  });
+});
+
+describe('SSRF guard', () => {
+  it('classifies private/loopback/link-local addresses (incl. normalized v4-mapped IPv6) [C2]', () => {
+    const priv = [
+      '127.0.0.1', '10.0.0.5', '192.168.1.1', '172.16.0.1', '169.254.1.1', '0.0.0.0', '100.64.0.1',
+      '::1', 'fc00::1', 'fd12::1', 'fe80::1',
+      '::ffff:127.0.0.1', '::ffff:169.254.169.254',  // dotted v4-mapped
+      '::ffff:7f00:1', '::ffff:a9fe:a9fe',            // hex v4-mapped (what `new URL` produces)
+      '::127.0.0.1', '64:ff9b::7f00:1',               // v4-compatible, NAT64-mapped loopback
+    ];
+    for (const ip of priv) expect(__test.isPrivateIp(ip), ip).toBe(true);
+    for (const ip of ['8.8.8.8', '1.1.1.1', '203.0.113.4', '2606:4700:4700::1111', '::ffff:8.8.8.8']) {
+      expect(__test.isPrivateIp(ip), ip).toBe(false);
+    }
+  });
+
+  it('rejects non-http(s) schemes and private/loopback URL targets — incl. v4-mapped literals [C1/C2]', async () => {
+    await expect(__test.assertSafeUrl('ftp://example.com/x')).rejects.toThrow(/http\(s\)/i);
+    await expect(__test.assertSafeUrl('file:///etc/passwd')).rejects.toThrow(/http\(s\)/i);
+    await expect(__test.assertSafeUrl('http://127.0.0.1/x')).rejects.toThrow(/private\/loopback/i);
+    await expect(__test.assertSafeUrl('http://[::1]/x')).rejects.toThrow(/private\/loopback/i);
+    // v4-mapped IPv6 literal → new URL normalizes hostname to hex; must still block
+    await expect(__test.assertSafeUrl('http://[::ffff:127.0.0.1]/x')).rejects.toThrow(/private\/loopback/i);
+    await expect(__test.assertSafeUrl('http://[::ffff:169.254.169.254]/meta')).rejects.toThrow(/private\/loopback/i);
+    await expect(__test.assertSafeUrl('http://localhost/x')).rejects.toThrow(/private/i);
+  });
+});
+
+describe('cached image/voice id resolution reaches the media branch [M1]', () => {
+  it('a cache id flows past the local check to getImageInfo (then SSRF-guards its URL)', async () => {
+    const ctx = {
+      getImageInfo: async () => ({ url: 'http://10.0.0.7/cached.png', file: '', file_name: 'cached.png', file_size: '9' }),
+      getRecordInfo: async () => null,
+    } as unknown as ApiActionContext;
+    // 'abc123.image' is not a real file → must NOT be rejected as "restricted";
+    // it reaches getImageInfo, whose private URL is then blocked by the SSRF guard.
+    await expect(__test.resolveDownload('abc123.image', ctx, 'image', 65536)).rejects.toThrow(/private/i);
+  });
+});