import { spawn, type ChildProcess } from 'node:child_process'; import { existsSync, mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { toBase64Url } from '../src/bytes.js'; import { ChannelSession, collectFrames, itemJson, itemText, parseLogRecords, type AuthRole, type ItemEvent, type ItemPlan, type ReceivedItem, } from '../src/index.js'; import { concatAll, ORIGIN, pseudoRandom, readAll, toHex } from './helpers.js'; /** * 端到端:用 @webdrop/core 生成的真实密文,对着真实的服务端(NativeAOT 可执行文件, * 没有就退回到 JIT 的 dll)走一遍完整流程。证明库产出的帧能通过服务端的全部校验, * 并且经由真实的 HTTP / WebSocket 往返后能被正确解密。 * * 用环境变量 WEBDROP_SERVER 指定可执行文件;WEBDROP_SERVER_DLL 指定 dll。 */ const root = join(import.meta.dirname, '..', '..', '..', 'server'); const exe = process.env.WEBDROP_SERVER ?? join(root, 'out', 'win-x64', 'webdrop.exe'); const linuxExe = join(root, 'out', 'linux-x64', 'webdrop'); const dll = process.env.WEBDROP_SERVER_DLL ?? join(root, 'src', 'WebDrop.Api', 'bin', 'Debug', 'net10.0', 'webdrop.dll'); const command: [string, string[]] | undefined = !process.env.WEBDROP_SERVER && existsSync(dll) ? ['dotnet', [dll]] : existsSync(exe) ? [exe, []] : existsSync(linuxExe) ? [linuxExe, []] : existsSync(dll) ? ['dotnet', [dll]] : undefined; let proc: ChildProcess | undefined; let dataDir = ''; let base = ''; let accountToken = ''; function headersFor(session: ChannelSession, role: AuthRole): Record { return { ...session.authHeaders(role), ...(session.role === 'owner' ? { 'X-WD-Account': accountToken } : {}) }; } beforeAll(async () => { if (!command) return; dataDir = mkdtempSync(join(tmpdir(), 'webdrop-e2e-')); proc = spawn(command[0], [...command[1], '--urls', 'http://127.0.0.1:0'], { env: { ...process.env, WebDrop__DataDir: dataDir, WebDrop__RequestsPerMinute: '1000000', WebDrop__ChannelCreatesPerMinute: '1000000', WebDrop__Captcha__Provider: 'none', }, stdio: ['ignore', 'pipe', 'pipe'], }); base = await new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error('服务端没有在 20 秒内启动')), 20_000); let buffer = ''; proc!.stdout!.on('data', (chunk: Buffer) => { buffer += chunk.toString(); const m = /Now listening on:\s*(http:\/\/\S+)/.exec(buffer); if (m) { clearTimeout(timer); resolve(m[1]!); } }); proc!.on('exit', (code) => reject(new Error(`服务端提前退出,代码 ${code}`))); }); const verifier = crypto.getRandomValues(new Uint8Array(32)); const keys = await crypto.subtle.generateKey({ name: 'ECDH', namedCurve: 'P-256' }, true, ['deriveBits']); const pubkey = await crypto.subtle.exportKey('jwk', keys.publicKey); const signingKeys = await crypto.subtle.generateKey({ name: 'ECDSA', namedCurve: 'P-256' }, true, ['sign', 'verify']); const signingPublicKey = await crypto.subtle.exportKey('jwk', signingKeys.publicKey); const encoded = (value: Uint8Array) => Buffer.from(value).toString('base64'); const registered = await fetch(`${base}/api/v1/accounts`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ username: 'core_e2e', salt: encoded(crypto.getRandomValues(new Uint8Array(16))), verifier: encoded(verifier), vault: encoded(new Uint8Array(28)), public_key: JSON.stringify(pubkey), signing_public_key: JSON.stringify(signingPublicKey), }) }); expect(registered.status).toBe(204); const session = await fetch(`${base}/api/v1/accounts/session`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ username: 'core_e2e', verifier: encoded(verifier) }) }); expect(session.status).toBe(200); accountToken = ((await session.json()) as { token: string }).token; }, 30_000); afterAll(async () => { if (proc && !proc.killed) proc.kill(); await new Promise((r) => setTimeout(r, 300)); if (dataDir) rmSync(dataDir, { recursive: true, force: true }); }); // ───────────────────────── 客户端辅助 ───────────────────────── const api = (path: string) => `${base}/api/v1${path}`; async function createChannel(opts: Parameters[0] = {}) { const created = await ChannelSession.create({ ...opts, settings: { name: 'Core E2E', ...opts.settings } }); const res = await fetch(api('/channels'), { method: 'POST', headers: { ...created.request.headers, 'X-WD-Account': accountToken, 'Content-Type': 'application/json' }, body: JSON.stringify(created.request.body), }); expect(res.status).toBe(201); const info = (await res.json()) as { id: string; mode: string }; created.session.verifyChannelInfo(info); // 核对服务端返回的模式与链接一致 return created.session; } async function joinChannel(session: ChannelSession, role: AuthRole = 'read') { const res = await fetch(api(`/channels/${session.id}/seats`), { method: 'POST', headers: headersFor(session, role) }); expect(res.status, await res.clone().text()).toBe(200); } async function send(session: ChannelSession, plan: ItemPlan, role: AuthRole) { const seqs: number[] = []; let first = true; for await (const frame of plan.frames) { const res = await fetch(api(`/channels/${session.id}/frames`), { method: 'POST', headers: { ...headersFor(session, role), 'Content-Type': 'application/octet-stream', ...(first ? plan.frame0Headers : {}) }, body: frame.bytes, }); expect(res.status, await res.clone().text()).toBeLessThan(300); seqs.push(((await res.json()) as { seq: number }).seq); first = false; } return seqs; } async function readLog(session: ChannelSession, role: AuthRole, from = 0) { const res = await fetch(api(`/channels/${session.id}/log?from=${from}`), { headers: headersFor(session, role) }); expect(res.status).toBe(200); return parseLogRecords(new Uint8Array(await res.arrayBuffer())); } /** 连接 WebSocket(凭据放在子协议里),收集事件直到满足 until。 */ function subscribe(session: ChannelSession, role: AuthRole, until: (events: ItemEvent[]) => boolean, from = 0) { const h = headersFor(session, role); const protocols = ['wd.v1', `wd.role.${role}`, `wd.ep.${h['X-WD-Endpoint']!}`]; if (h.Authorization) protocols.push(`wd.auth.${h.Authorization.replace('Bearer ', '')}`); if (session.role === 'owner') protocols.push(`wd.account.${accountToken.replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')}`); const ws = new WebSocket(`${base.replace('http', 'ws')}/api/v1/channels/${session.id}/ws?from=${from}`, protocols); ws.binaryType = 'arraybuffer'; const reader = session.createItemReader(); const events: ItemEvent[] = []; const done = new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error(`等待 WebSocket 事件超时,已收到 ${JSON.stringify(events.map((e) => e.type))}`)), 15_000); let chain = Promise.resolve(); ws.onmessage = (m) => { if (typeof m.data === 'string') return; // 控制消息(心跳等) chain = chain.then(async () => { events.push(...(await reader.push(new Uint8Array(m.data as ArrayBuffer)))); if (until(events)) { clearTimeout(timer); resolve(events); } }); }; ws.onerror = () => reject(new Error('WebSocket 出错')); }); const opened = new Promise((resolve) => (ws.onopen = () => resolve())); return { ws, done, opened, events }; } const items = (events: ItemEvent[]): ReceivedItem[] => events.flatMap((e) => (e.type === 'item' ? [e.item] : [])); const suite = command ? describe : describe.skip; // 全局 WebSocket 从 Node 22 起提供(本包要求 >=22);更旧的 Node 上明确跳过依赖它的用例,而不是误报失败 const wsIt = typeof WebSocket === 'undefined' ? it.skip : it; suite('端到端(真实服务端)', () => { it('文本:所有者发送,成员经日志拉取并解密', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); owner.deviceName = '桌面'; await joinChannel(member, 'read'); await send(owner, owner.encryptText('hello 端到端', { ts: 1 }), 'owner'); const reader = member.createItemReader(); const events: ItemEvent[] = []; for (const rec of await readLog(member, 'read')) events.push(...(await reader.push(rec))); const [item] = items(events); expect(itemText(item!)).toBe('hello 端到端'); expect(item!.meta).toMatchObject({ kind: 'text', dev: '桌面' }); }); wsIt('WebSocket 实时推送:成员在线时收到所有者新发的内容', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'read'); const sub = subscribe(member, 'read', (ev) => items(ev).length >= 2); await sub.opened; await send(owner, owner.encryptText('第一条'), 'owner'); await send(owner, owner.encryptJson({ n: 2 }), 'owner'); const got = items(await sub.done); expect(itemText(got[0]!)).toBe('第一条'); expect(itemJson(got[1]!)).toEqual({ n: 2 }); sub.ws.close(); }); wsIt('成员也能上传(写令牌由 K 派生),所有者实时收到', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'write'); const sub = subscribe(owner, 'owner', (ev) => items(ev).length >= 1); await sub.opened; await send(member, member.encryptText('来自成员'), 'write'); expect(itemText(items(await sub.done)[0]!)).toBe('来自成员'); sub.ws.close(); }); wsIt('快照:新快照取代旧快照,订阅者先后收到 item 和 removed(superseded)', async () => { const owner = await createChannel(); const sub = subscribe(owner, 'owner', (ev) => ev.some((e) => e.type === 'removed')); await sub.opened; await send(owner, owner.encryptJson({ v: 1 }, { snapshot: true }), 'owner'); await send(owner, owner.encryptJson({ v: 2 }, { snapshot: true }), 'owner'); const events = await sub.done; expect(events.map((e) => e.type)).toEqual(['item', 'item', 'removed']); expect(items(events).map((i) => itemJson(i))).toEqual([{ v: 1 }, { v: 2 }]); expect(events.find((e) => e.type === 'removed')).toMatchObject({ reason: 'superseded' }); sub.ws.close(); }); it('幂等:重发同一帧返回 200 和同一个 seq(确定性加密)', async () => { const owner = await createChannel(); const plan = owner.encryptText('retry', { ts: 5 }); const [first] = await send(owner, plan, 'owner'); const again = owner.encryptText('retry', { ts: 5, itemId: plan.idBytes }); const frame = (await collectFrames(again))[0]!; const res = await fetch(api(`/channels/${owner.id}/frames`), { method: 'POST', headers: { ...headersFor(owner, 'owner'), 'Content-Type': 'application/octet-stream' }, body: frame.bytes, }); expect(res.status).toBe(200); expect(((await res.json()) as { seq: number }).seq).toBe(first); }); it('服务端的存储里没有明文', async () => { const owner = await createChannel(); const marker = 'UNIQUE-PLAINTEXT-MARKER-7f3a9c'; await send(owner, owner.encryptText(marker), 'owner'); const res = await fetch(api(`/channels/${owner.id}/log`), { headers: headersFor(owner, 'owner') }); const wire = new Uint8Array(await res.arrayBuffer()); expect(new TextDecoder('latin1').decode(wire)).not.toContain(marker); expect(new TextDecoder('latin1').decode(wire)).not.toContain('"kind"'); }); it('大文件:多帧 GATED 上传,日志里只有 announce,下载流式解密后与原文一致', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'read'); const content = pseudoRandom(3_300_000, 99); const plan = owner.encryptFile(new Blob([content]), { name: 'big.bin', mime: 'application/octet-stream' }, { maxDownloads: 5 }); await send(owner, plan, 'owner'); expect(plan.gated).toBe(true); // 日志里只有一条 announce,没有任何文件字节 const log = await readLog(member, 'read'); expect(log.map((r) => r.kind)).toEqual(['announce']); // 流式下载 + 流式解密 const res = await fetch(api(`/channels/${owner.id}/items/${plan.itemId}/frames`), { headers: headersFor(member, 'read') }); expect(res.status).toBe(200); const { content: stream, result } = member.decryptItemStream(res.body!, { itemId: plan.idBytes }); const out = await readAll(stream); expect(await result).toMatchObject({ complete: true }); expect(out.length).toBe(content.length); expect(Buffer.compare(out, content)).toBe(0); expect((await result).meta).toMatchObject({ name: 'big.bin', size: content.length }); }, 60_000); it('下载上限:超过后从头下载返回 410,续传(from>0)在宽限期内仍可用', async () => { const owner = await createChannel(); const content = pseudoRandom(2_500_000, 5); const plan = owner.encryptFile(new Blob([content]), { name: 'once.bin' }, { maxDownloads: 1 }); await send(owner, plan, 'owner'); const url = api(`/channels/${owner.id}/items/${plan.itemId}/frames`); const first = await fetch(url, { headers: headersFor(owner, 'owner') }); const decoded = owner.decryptItemStream(first.body!); expect(Buffer.compare(await readAll(decoded.content), content)).toBe(0); const second = await fetch(url, { headers: headersFor(owner, 'owner') }); expect(second.status).toBe(410); expect(((await second.json()) as { error: string }).error).toBe('gone'); const resumed = await fetch(`${url}?from=1`, { headers: headersFor(owner, 'owner') }); expect(resumed.status).toBe(200); await resumed.arrayBuffer(); }, 60_000); it('上传断点续传:中途中断后用 fromFrame 重新加密剩余部分,服务端得到完整条目', async () => { const owner = await createChannel(); const content = pseudoRandom(3_300_000, 123); const blob = new Blob([content]); const itemId = owner.encryptFile(blob).idBytes; const full = await collectFrames(owner.encryptFile(blob, { name: 'resume.bin' }, { itemId, ts: 9 })); expect(full.length).toBe(4); // 第一轮只上传前两帧,然后「崩溃」 const first = owner.encryptFile(blob, { name: 'resume.bin' }, { itemId, ts: 9 }); let sent = 0; let isFirst = true; for await (const f of first.frames) { if (sent === 2) break; const res = await fetch(api(`/channels/${owner.id}/frames`), { method: 'POST', headers: { ...headersFor(owner, 'owner'), 'Content-Type': 'application/octet-stream', ...(isFirst ? first.frame0Headers : {}) }, body: f.bytes, }); expect(res.status).toBe(201); isFirst = false; sent++; } // 第二轮:从第 2 帧开始重新加密(字节与完整加密一致),并且重发第 1 帧也是幂等的 const rest = owner.encryptFile(blob, { name: 'resume.bin' }, { itemId, ts: 9, fromFrame: 1 }); const statuses: number[] = []; for await (const f of rest.frames) { const res = await fetch(api(`/channels/${owner.id}/frames`), { method: 'POST', headers: { ...headersFor(owner, 'owner'), 'Content-Type': 'application/octet-stream' }, body: f.bytes, }); statuses.push(res.status); } expect(statuses).toEqual([200, 201, 201]); // 第 1 帧已存在(幂等),第 2、3 帧是新的 const res = await fetch(api(`/channels/${owner.id}/items/${toBase64Url(itemId)}/frames`), { headers: headersFor(owner, 'owner') }); const { content: stream, result } = owner.decryptItemStream(res.body!); expect(Buffer.compare(await readAll(stream), content)).toBe(0); expect((await result).complete).toBe(true); }, 60_000); it('下载中途断线,用 nextIndex 续传,拼起来与原文一致', async () => { const owner = await createChannel(); const content = pseudoRandom(3_300_000, 7); const plan = owner.encryptFile(new Blob([content]), { name: 'cut.bin' }); await send(owner, plan, 'owner'); const url = api(`/channels/${owner.id}/items/${plan.itemId}/frames`); // 第一次:只读前约 1.5 帧就断开 const first = await fetch(url, { headers: headersFor(owner, 'owner') }); const reader = first.body!.getReader(); const chunks: Uint8Array[] = []; let got = 0; while (got < 1_600_000) { const { done, value } = await reader.read(); if (done) break; chunks.push(value); got += value.length; } await reader.cancel(); const part1 = await (async () => { const dec = owner.decryptItemStream(new ReadableStream({ start: (c) => { chunks.forEach((x) => c.enqueue(x)); c.close(); } })); return { bytes: await readAll(dec.content), result: await dec.result }; })(); expect(part1.result.complete).toBe(false); const second = await fetch(`${url}?from=${part1.result.nextIndex}`, { headers: headersFor(owner, 'owner') }); expect(second.status).toBe(200); const dec2 = owner.decryptItemStream(second.body!, { fromIndex: part1.result.nextIndex }); const part2 = await readAll(dec2.content); expect(toHex(concatAll([part1.bytes, part2]).subarray(0, 32))).toBe(toHex(content.subarray(0, 32))); expect(Buffer.compare(concatAll([part1.bytes, part2]), content)).toBe(0); }, 60_000); it('设备名:成员加密上传,所有者在席位列表里解开', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'read'); const blob = await member.sealLabel('我的 iPhone'); const put = await fetch(api(`/channels/${owner.id}/seats/me`), { method: 'PUT', headers: { ...headersFor(member, 'read'), 'Content-Type': 'application/octet-stream' }, body: blob, }); expect(put.status).toBe(204); const res = await fetch(api(`/channels/${owner.id}/seats`), { headers: headersFor(owner, 'owner') }); const seats = (await res.json()) as Array<{ endpoint_hash: string; label?: string }>; const mine = seats.find((s) => s.endpoint_hash === toBase64Url(member.endpointHash))!; expect(await owner.openLabel(mine.label!, mine.endpoint_hash)).toBe('我的 iPhone'); // 服务端存的是密文 expect(Buffer.from(mine.label!, 'base64url').toString('utf8')).not.toContain('iPhone'); }); it('最长的设备名(228 字节)也被服务端接受', async () => { const owner = await createChannel(); const blob = await owner.sealLabel('x'.repeat(228)); const put = await fetch(api(`/channels/${owner.id}/seats/me`), { method: 'PUT', headers: { ...headersFor(owner, 'owner'), 'Content-Type': 'application/octet-stream' }, body: blob, }); expect(put.status).toBe(204); }); it('raw 频道:匿名成员无需令牌即可读取,服务端看得到明文', async () => { const owner = await createChannel({ mode: 'raw' }); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'read'); await send(owner, owner.encryptText('raw 明文 visible'), 'owner'); const records = await readLog(member, 'read'); const [item] = items(await (async () => { const reader = member.createItemReader(); const out: ItemEvent[] = []; for (const r of records) out.push(...(await reader.push(r))); return out; })()); expect(item!.kind).toBe('text'); expect(itemText(item!)).toBe('raw 明文 visible'); // 对比 enc:raw 的明文确实出现在服务端返回的字节里 const res = await fetch(api(`/channels/${owner.id}/log`), { headers: headersFor(member, 'read') }); expect(new TextDecoder().decode(await res.arrayBuffer())).toContain('raw 明文 visible'); }); it('下发给成员的链接没有管理权:成员拿不到所有者能力', async () => { const owner = await createChannel(); const member = await ChannelSession.open(owner.link(ORIGIN)); await joinChannel(member, 'read'); const res = await fetch(api(`/channels/${owner.id}/seats`), { headers: { ...headersFor(member, 'read') } }); expect(res.status).toBe(403); expect(() => headersFor(member, 'owner')).toThrow(); }); });