feat(onebot): download stream actions (#163 phase 3)
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.
This commit is contained in:
DrSmoothl
@@ -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 };
|
||||
@@ -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');
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user