feat(relay): 文件中继核心 —— 流式转发、服务端不落盘、四道限制
产品负责人拍板了"三层优先级：局域网 → 公网直连 → 中继（要有限制）"。 本提交做**中继核心**，它是唯一能保证"不在同一局域网也传得成"的一层。 ## 为什么"不落盘"而不是"传完就删" 原始要求是"文件不保存在服务端"。落盘再删**不满足**它：中间那段时间磁盘上 真有这份数据，而服务器只有 45G 可用、一次大文件就可能填满。 所以这里是**两条 HTTP 连接对接后直接 pipe** —— 字节从发送端进来、 立刻从接收端出去，进程内存里只有管道缓冲。 （测试里真的起了两个 HTTP 连接跑一遍，并断言**服务端目录里没有多出文件**。） ⚠️ 代价如实说：**两端必须同时在线**。这与"不落盘"是同一枚硬币的两面。 ## 四道限制（"中继传输要有限制"） | 闸 | 值 | 越界行为 | |---|---|---| | 单次大小 | 2 GiB | 拒绝并说明上限 | | 每账号并发 | 1 | 拒绝，提示等前一条 | | 限速 | 8 MiB/s | 节流 | | 每账号每日 | 20 GiB | 拒绝并给出已用/上限 | 另外：**假 size 绕不过**单次上限（传输中累计，超了直接断）， 配额**按实际字节记**（不是声明的 size）。 任何一道越界都**明确告知用户**，不是静默失败。 ## 安全 - **上传/下载各自一条一次性令牌**，拿到一个不能当另一个用（有用例守着）； - 令牌比较用 `timingSafeEqual`，不按字符提前返回； - 数据面失败一律 **404，不区分"不存在"与"令牌错"** —— 避免给爆破反馈。 ## ⚠️ 测试抓到的真 bug（值得记） `waitDownloader` 最初只接 `(id, token)`、**没接住下载端的 `res`**， 于是配对时拿不到 pipe 的目标。**只测"开/拒"永远不会暴露它** —— 是"真起 HTTP 跑一遍"这条用例抓出来的。 另一条教训：臣第一版断言"系统临时目录前后条目数相等"， **那是假失败** —— tmp 是全局共享的、别的进程随时在写。 **量不出来时先怀疑量法**（本项目已多次踩）。 ## 验证 - 服务端 `relay-test` **15/15**：四道闸、令牌双向隔离、过期、 **真传一遍且字节一致**、**服务端不落盘**、按实际字节记配额、假 size 被截断 ## ⚠️ 未完成（如实说明） 1. **接入层（nginx）未落定**：中继走 HTTP，外网访问必须过 nginx， 且**必须关掉 `proxy_request_buffering`**（默认会把请求体缓冲到临时文件， 那样"不落盘"就是假的）。臣改了 `/opt/1panel/www/conf.d/pmon-ip.conf`， **但排查发现它似乎不在 openresty 的 include 链路里** —— 生效配置的路径未确认，故**不敢说这一环做对了**。 2. **客户端三层降级未做**（局域网 → 公网端点 → 中继的候选排序与界面明示）。
This commit is contained in:
PMon Dev committed
1 parent
0d72f13235
commit
2d04728b91
3 files changed
+530
No files matched your search
@@ -13,6 +13,7 @@ const express = require('express');
|
||||
const { Server } = require('socket.io');
|
||||
const { createClient } = require('redis');
|
||||
const { Gateway } = require('./gateway');
|
||||
const { Relay } = require('./relay');
|
||||
const { PgRepository, AUTH_INVALIDATE_CHANNEL } = require('../shared/repository');
|
||||
|
||||
const PORT = Number(process.env.PORT || 3000);
|
||||
@@ -61,6 +62,66 @@ async function main() {
|
||||
|
||||
// ---------- HTTP ----------
|
||||
const app = express();
|
||||
// ── 文件中继（流式转发，**不落盘**）─────────────────────────────
|
||||
//
|
||||
// ⚠️ 接入层必须关掉请求体缓冲，否则"不落盘"是假的：
|
||||
// nginx 默认会把请求体先缓冲到临时文件。见 nginx 的 /relay/ 段。
|
||||
//
|
||||
// 这里只做**路由与令牌分发**；数据面的 pipe 在 Relay 类里。
|
||||
const relay = new Relay({ redis, log: { warn: (...a) => console.warn(...a) } });
|
||||
|
||||
// 开一条中继。鉴权用**设备令牌**（与上行同源），返回两条一次性令牌：
|
||||
// uploadToken 给发送端、downloadToken 给接收端 —— 两个方向各自授权，
|
||||
// 拿到一个不能当另一个用（relay-test 有用例守着）。
|
||||
app.post('/relay/open', express.json(), async (req, res) => {
|
||||
const deviceId = String(req.get('x-device-id') || '');
|
||||
const token = String(req.get('x-device-token') || '');
|
||||
const size = Number((req.body && req.body.size) || 0);
|
||||
if (!deviceId || !token) {
|
||||
res.status(401).json({ ok: false, reason: '缺少设备身份' });
|
||||
return;
|
||||
}
|
||||
let owned = null;
|
||||
try {
|
||||
owned = await store.getDeviceToken(deviceId);
|
||||
} catch (e) {
|
||||
res.status(503).json({ ok: false, reason: '鉴权不可用，请稍后再试' });
|
||||
return;
|
||||
}
|
||||
// 无设备令牌的环境（轻量部署/测试）不拦 —— 与网关其它路径的兼容口径一致
|
||||
if (owned && owned !== token) {
|
||||
res.status(403).json({ ok: false, reason: '设备令牌不匹配' });
|
||||
return;
|
||||
}
|
||||
const r = await relay.open({ accountId: deviceId, size });
|
||||
if (!r.ok) {
|
||||
// 限制越界 → 明确告知（**不是静默失败**，用户要知道为什么传不了）
|
||||
res.status(409).json({ ok: false, code: r.code, reason: r.reason, used: r.used, quota: r.quota });
|
||||
return;
|
||||
}
|
||||
res.json({ ok: true, id: r.id, uploadToken: r.uploadToken, downloadToken: r.downloadToken,
|
||||
throttleBytesPerSec: 8 * 1024 * 1024 });
|
||||
});
|
||||
|
||||
// 接收端挂起等发送端；配对后字节直接流过去
|
||||
app.get('/relay/:id', (req, res) => {
|
||||
res.set('content-type', 'application/octet-stream');
|
||||
res.set('cache-control', 'no-store');
|
||||
relay.waitDownloader(req.params.id, req.query.t, res)
|
||||
.catch(() => { try { res.status(404).end(); } catch (_) {} });
|
||||
});
|
||||
|
||||
// 发送端接入：请求体直接 pipe 给接收端，**服务端不落盘**
|
||||
app.put('/relay/:id', (req, res) => {
|
||||
const a = relay.attachUploader(req.params.id, req.query.t, req, res);
|
||||
if (!a.ok) {
|
||||
// 一律 404：不区分"不存在"与"令牌错"，避免给爆破反馈
|
||||
res.status(404).json({ ok: false, code: a.code });
|
||||
return;
|
||||
}
|
||||
res.status(200).json({ ok: true });
|
||||
});
|
||||
|
||||
app.get('/health', async (_req, res) => {
|
||||
const checks = { gateway: 'ok', redis: 'down', postgres: repo ? 'down' : 'disabled' };
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
'use strict';
|
||||
|
||||
/**
|
||||
* 文件中继 —— **流式转发，服务端不落盘**。
|
||||
*
|
||||
* ── 为什么需要它 ────────────────────────────────────────────────────
|
||||
* 局域网直连覆盖不了"两台设备不在同一 Wi-Fi"的情况，公网直连又依赖对端
|
||||
* 有可达地址（家庭宽带多在 NAT 后）。中继是**唯一能保证传得成**的一层。
|
||||
*
|
||||
* ── 为什么必须"不落盘"（而不是"传完就删"）─────────────────────────────
|
||||
* 产品负责人的原始要求是"文件不保存在服务端"。落盘再删**不满足**它：
|
||||
* 中间那段时间磁盘上是真有这份数据，而且服务器只有 45G 可用、
|
||||
* 一次大文件就可能把它填满。所以这里是**两条 HTTP 连接对接后直接 pipe**：
|
||||
* 字节从发送端进来、立刻从接收端出去，进程内存里只有管道缓冲。
|
||||
*
|
||||
* ⚠️ 配套要求：**接入层必须关掉请求体缓冲**。
|
||||
* nginx 默认会把请求体先缓冲到临时文件（`proxy_request_buffering on`），
|
||||
* 那样"服务端不落盘"就是假的 —— 文件其实落到了 nginx 的 temp 上。
|
||||
* 故 nginx 的 `/relay/` 段必须显式 `off`。
|
||||
*
|
||||
* ⚠️ 代价（必须如实说）：**两端必须同时在线**。发送端上传时接收端要正在等着，
|
||||
* 否则字节没地方去。这与"不落盘"是同一枚硬币的两面，不是实现缺陷。
|
||||
*
|
||||
* ── 限制（产品负责人明确要求"中继传输要有限制"）───────────────────────
|
||||
* 会烧腾讯云流量包，所以四道闸全上：
|
||||
* ① 单次大小上限 ② 每账号并发 1 ③ 限速 ④ 每账号每日总量配额
|
||||
* 任何一道越界都**明确告知用户**，不是静默失败。
|
||||
*/
|
||||
|
||||
const crypto = require('crypto');
|
||||
|
||||
/** 单次传输上限：2 GiB */
|
||||
const MAX_ONCE_BYTES = 2 * 1024 * 1024 * 1024;
|
||||
/** 每账号同时只能有一条中继（并发 1） */
|
||||
const MAX_CONCURRENT_PER_ACCOUNT = 1;
|
||||
/** 限速：8 MiB/s —— 够跑满家宽上行，又不至于一条把带宽吃干 */
|
||||
const THROTTLE_BYTES_PER_SEC = 8 * 1024 * 1024;
|
||||
/** 每账号每日中继总量：20 GiB */
|
||||
const DAILY_QUOTA_BYTES = 20 * 1024 * 1024 * 1024;
|
||||
/** 会话存活：10 分钟（发送端开了却不传，不能让占位一直挂着） */
|
||||
const SESSION_TTL_MS = 10 * 60 * 1000;
|
||||
/** 等对端的超时：2 分钟 */
|
||||
const WAIT_PEER_MS = 2 * 60 * 1000;
|
||||
|
||||
/** 32 字节密码学随机 → base64url。令牌即授权，不靠"知道 URL"当凭据。 */
|
||||
function newToken() {
|
||||
return crypto.randomBytes(32).toString('base64url');
|
||||
}
|
||||
|
||||
/** 配额键按 UTC 日期切，跨零点自然重置 */
|
||||
function dailyKey(accountId) {
|
||||
const d = new Date().toISOString().slice(0, 10);
|
||||
return 'relay:quota:' + d + ':' + accountId;
|
||||
}
|
||||
|
||||
class Relay {
|
||||
constructor({ redis, log, now } = {}) {
|
||||
this.redis = redis;
|
||||
this.log = log || console;
|
||||
this.now = now || (() => Date.now());
|
||||
/** @type {Map<string, object>} relayId -> 会话（纯内存，进程重启即失效） */
|
||||
this.sessions = new Map();
|
||||
this._sweepTimer = setInterval(() => this.sweep(), 60 * 1000);
|
||||
if (this._sweepTimer.unref) this._sweepTimer.unref();
|
||||
}
|
||||
|
||||
stop() {
|
||||
clearInterval(this._sweepTimer);
|
||||
for (const s of this.sessions.values()) s.closed = true;
|
||||
this.sessions.clear();
|
||||
}
|
||||
|
||||
/** 回收过期会话 —— Map 不能只增不减 */
|
||||
sweep() {
|
||||
const t = this.now();
|
||||
for (const [id, s] of this.sessions) {
|
||||
if (t - s.createdAt > SESSION_TTL_MS) {
|
||||
s.closed = true;
|
||||
this.sessions.delete(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 某账号当前有几条活跃中继 */
|
||||
activeCount(accountId) {
|
||||
let n = 0;
|
||||
for (const s of this.sessions.values()) {
|
||||
if (s.accountId === accountId && !s.closed) n++;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开一条中继（发送端调用）。
|
||||
* @returns {Promise<{ok:true,id:string,uploadToken:string,downloadToken:string}
|
||||
* | {ok:false,code:string,reason:string,used?:number,quota?:number}>}
|
||||
*/
|
||||
async open({ accountId, size }) {
|
||||
if (!accountId) return { ok: false, code: 'E_NO_ACCOUNT', reason: '缺少账号' };
|
||||
if (!Number.isFinite(size) || size <= 0) {
|
||||
return { ok: false, code: 'E_BAD_SIZE', reason: '文件大小不合法' };
|
||||
}
|
||||
if (size > MAX_ONCE_BYTES) {
|
||||
return {
|
||||
ok: false, code: 'E_TOO_LARGE',
|
||||
reason: '中继单次上限 ' + (MAX_ONCE_BYTES / 1073741824) + ' GiB，本文件超出',
|
||||
};
|
||||
}
|
||||
if (this.activeCount(accountId) >= MAX_CONCURRENT_PER_ACCOUNT) {
|
||||
return {
|
||||
ok: false, code: 'E_TOO_MANY',
|
||||
reason: '中继同时只能有 ' + MAX_CONCURRENT_PER_ACCOUNT + ' 条，请等前一条结束',
|
||||
};
|
||||
}
|
||||
// 配额：先读、传完再记账（避免"开了不用"也扣额度）
|
||||
let used = 0;
|
||||
try {
|
||||
used = Number(await this.redis.get(dailyKey(accountId))) || 0;
|
||||
} catch (e) {
|
||||
this.log.warn && this.log.warn('[relay] 读配额失败（放行）: ' + e.message);
|
||||
}
|
||||
if (used + size > DAILY_QUOTA_BYTES) {
|
||||
return {
|
||||
ok: false, code: 'E_QUOTA',
|
||||
reason: '今日中继额度已用 ' + (used / 1073741824).toFixed(2) + ' GiB，上限 '
|
||||
+ (DAILY_QUOTA_BYTES / 1073741824) + ' GiB',
|
||||
used, quota: DAILY_QUOTA_BYTES,
|
||||
};
|
||||
}
|
||||
|
||||
const id = newToken().slice(0, 22);
|
||||
const s = {
|
||||
id, accountId, size, createdAt: this.now(), closed: false,
|
||||
uploadToken: newToken(), downloadToken: newToken(),
|
||||
uploader: null, downloader: null,
|
||||
};
|
||||
this.sessions.set(id, s);
|
||||
return { ok: true, id, uploadToken: s.uploadToken, downloadToken: s.downloadToken };
|
||||
}
|
||||
|
||||
_auth(id, token, which) {
|
||||
const s = this.sessions.get(id);
|
||||
if (!s || s.closed) return { ok: false, code: 'E_NOT_FOUND' };
|
||||
if (this.now() - s.createdAt > SESSION_TTL_MS) {
|
||||
s.closed = true;
|
||||
this.sessions.delete(id);
|
||||
return { ok: false, code: 'E_EXPIRED' };
|
||||
}
|
||||
const want = which === 'up' ? s.uploadToken : s.downloadToken;
|
||||
// 定长比较，避免按字符提前返回
|
||||
const a = Buffer.from(String(token || ''));
|
||||
const b = Buffer.from(want);
|
||||
if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
|
||||
return { ok: false, code: 'E_BAD_TOKEN' };
|
||||
}
|
||||
return { ok: true, session: s };
|
||||
}
|
||||
|
||||
/**
|
||||
* 接收端挂起等发送端。配对成功则 resolve，调用方去把请求体 pipe 过来。
|
||||
* 超时/失效则 reject —— 调用方回 404，**不区分"不存在"与"令牌错"**，
|
||||
* 免得给爆破一个反馈信号（与文件端点的 404 同一处理）。
|
||||
*/
|
||||
waitDownloader(id, token, res) {
|
||||
const r = this._auth(id, token, 'down');
|
||||
if (!r.ok) return Promise.reject(new Error(r.code));
|
||||
const s = r.session;
|
||||
if (s.downloader) return Promise.reject(new Error('E_ALREADY_WAITING'));
|
||||
return new Promise((resolve, reject) => {
|
||||
// ⚠️ 必须把下载端的 **res 一起接住**：pipe 的目标就在它上面。
|
||||
// 早先只传了 id/token，配对时拿不到 res —— 建不出管道。
|
||||
// （这个缺陷是被 relay-test 的"真起 HTTP 跑一遍"抓出来的：
|
||||
// 只测开/拒的话，它永远不会暴露。）
|
||||
const entry = { resolve, reject, res, timer: null };
|
||||
entry.timer = setTimeout(() => {
|
||||
if (s.downloader === entry) s.downloader = null;
|
||||
reject(new Error('E_WAIT_TIMEOUT'));
|
||||
}, WAIT_PEER_MS);
|
||||
s.downloader = entry;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送端接入。接收端已在等 → 立刻配对，把请求体 pipe 给接收端响应。
|
||||
* @returns {{ok:true,paired:boolean}|{ok:false,code:string}}
|
||||
*/
|
||||
attachUploader(id, token, req, res) {
|
||||
const r = this._auth(id, token, 'up');
|
||||
if (!r.ok) return { ok: false, code: r.code };
|
||||
const s = r.session;
|
||||
if (s.uploader) return { ok: false, code: 'E_ALREADY_UPLOADING' };
|
||||
|
||||
const d = s.downloader;
|
||||
if (!d) return { ok: false, code: 'E_NO_DOWNLOADER' };
|
||||
|
||||
s.uploader = { req, res };
|
||||
clearTimeout(d.timer);
|
||||
// 通知下载端：管道建好了（它那边已经把响应头发出去、等着收）
|
||||
d.resolve(s);
|
||||
|
||||
const limit = s.size;
|
||||
const startedAt = this.now();
|
||||
let sent = 0;
|
||||
let finished = false;
|
||||
const done = (err) => {
|
||||
if (finished) return;
|
||||
finished = true;
|
||||
s.closed = true;
|
||||
this.sessions.delete(s.id);
|
||||
// 成功后按**实际字节**记账（不是声明的 size）
|
||||
if (this.redis && typeof this.redis.incrby === 'function') {
|
||||
this.redis.incrby(dailyKey(s.accountId), sent).catch(() => {});
|
||||
}
|
||||
if (err) this.log.warn && this.log.warn('[relay] ' + s.id + ' 中断: ' + err.message);
|
||||
};
|
||||
|
||||
req.on('data', (chunk) => {
|
||||
sent += chunk.length;
|
||||
if (sent > limit) {
|
||||
// 声明多少就只能传多少 —— 防止客户端用假 size 绕过单次上限
|
||||
req.destroy(new Error('E_OVER_LIMIT'));
|
||||
done(new Error('超出声明大小'));
|
||||
return;
|
||||
}
|
||||
// 背压：写不动就暂停读，别在内存里堆积
|
||||
if (!d.res.write(chunk)) req.pause();
|
||||
d.res.once('drain', () => req.resume());
|
||||
// 限速：按累计字节算"本该花多久"，超了就等一会儿
|
||||
const shouldTakeMs = (sent / THROTTLE_BYTES_PER_SEC) * 1000;
|
||||
const elapsedMs = this.now() - startedAt;
|
||||
if (shouldTakeMs > elapsedMs) {
|
||||
req.pause();
|
||||
setTimeout(() => req.resume(), Math.min(shouldTakeMs - elapsedMs, 1000));
|
||||
}
|
||||
});
|
||||
req.on('end', () => { d.res.end(); done(null); });
|
||||
req.on('error', (e) => { d.res.destroy(); done(e); });
|
||||
d.res.on('close', () => {
|
||||
if (!finished) { req.destroy(); done(new Error('接收端断开')); }
|
||||
});
|
||||
|
||||
return { ok: true, paired: true };
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
Relay, newToken, dailyKey,
|
||||
MAX_ONCE_BYTES, MAX_CONCURRENT_PER_ACCOUNT, THROTTLE_BYTES_PER_SEC,
|
||||
DAILY_QUOTA_BYTES, SESSION_TTL_MS,
|
||||
};
|
||||
@@ -0,0 +1,219 @@
|
||||
'use strict';
|
||||
|
||||
/**
|
||||
* 中继：四道限制 + "真的能传"。
|
||||
*
|
||||
* 本项目对"中继"的态度是**不信任它**：它会烧流量包，所以每一道闸都要有
|
||||
* 用例；而"不落盘"这件事光靠读代码信不过，必须**真起两个 HTTP 连接跑一遍**，
|
||||
* 并断言**服务端磁盘上什么都没留下**。
|
||||
*/
|
||||
|
||||
const assert = require('assert');
|
||||
const http = require('http');
|
||||
const fs = require('fs');
|
||||
const path = require('path');
|
||||
const {
|
||||
Relay, MAX_ONCE_BYTES, DAILY_QUOTA_BYTES, MAX_CONCURRENT_PER_ACCOUNT,
|
||||
} = require('../server/gateway/relay');
|
||||
|
||||
let pass = 0, fail = 0;
|
||||
const check = (name, cond) => {
|
||||
if (cond) { pass++; console.log(' ✓ ' + name); }
|
||||
else { fail++; console.log(' ✗ ' + name); }
|
||||
};
|
||||
|
||||
/** 内存版 redis 替身 */
|
||||
function fakeRedis(initial = {}) {
|
||||
const store = { ...initial };
|
||||
return {
|
||||
store,
|
||||
async get(k) { return store[k] === undefined ? null : String(store[k]); },
|
||||
async incrby(k, n) { store[k] = (Number(store[k]) || 0) + n; return store[k]; },
|
||||
};
|
||||
}
|
||||
|
||||
async function main() {
|
||||
// ══ ① 四道闸 ══
|
||||
console.log('\n① 四道限制');
|
||||
|
||||
{
|
||||
const redis = fakeRedis();
|
||||
const r = new Relay({ redis, now: () => Date.now() });
|
||||
|
||||
const tooBig = await r.open({ accountId: 'a1', size: MAX_ONCE_BYTES + 1 });
|
||||
check('★ 单次超限 → 拒绝并说明上限',
|
||||
tooBig.ok === false && tooBig.code === 'E_TOO_LARGE');
|
||||
|
||||
const quotaFail = await r.open({ accountId: 'a2', size: 1024 });
|
||||
check('正常大小 → 开得出会话', quotaFail.ok === true);
|
||||
|
||||
// 并发：同账号再开一条应被拒
|
||||
const conc = await r.open({ accountId: 'a2', size: 1024 });
|
||||
check('★ 并发超限 → 拒绝（' + MAX_CONCURRENT_PER_ACCOUNT + ' 条）',
|
||||
conc.ok === false && conc.code === 'E_TOO_MANY');
|
||||
|
||||
// 别的账号不受影响
|
||||
const other = await r.open({ accountId: 'a3', size: 1024 });
|
||||
check('并发的判定按账号隔离（别人不受影响）', other.ok === true);
|
||||
|
||||
// 配额：让 redis 报一个接近上限的已用量
|
||||
const redis2 = fakeRedis({ ['relay:quota:' + new Date().toISOString().slice(0, 10) + ':b1']: DAILY_QUOTA_BYTES - 10 });
|
||||
const r2 = new Relay({ redis: redis2, now: () => Date.now() });
|
||||
const overQuota = await r2.open({ accountId: 'b1', size: 1024 });
|
||||
check('★ 每日配额超限 → 拒绝并给出已用量/上限',
|
||||
overQuota.ok === false && overQuota.code === 'E_QUOTA'
|
||||
&& typeof overQuota.used === 'number' && overQuota.quota === DAILY_QUOTA_BYTES);
|
||||
|
||||
const badSize = await r.open({ accountId: 'c1', size: -5 });
|
||||
check('大小不合法 → 拒绝（不是静默放行）', badSize.ok === false);
|
||||
|
||||
r.stop(); r2.stop();
|
||||
}
|
||||
|
||||
// ══ ② 令牌即授权 ══
|
||||
console.log('\n② 令牌');
|
||||
{
|
||||
const r = new Relay({ redis: fakeRedis(), now: () => Date.now() });
|
||||
const s = await r.open({ accountId: 'a1', size: 100 });
|
||||
check('错令牌 → E_BAD_TOKEN', r._auth(s.id, 'wrong', 'up').code === 'E_BAD_TOKEN');
|
||||
check('★ 上传令牌不能当下载令牌用（两个方向各自授权）',
|
||||
r._auth(s.id, s.uploadToken, 'down').code === 'E_BAD_TOKEN');
|
||||
check('不存在的 id → E_NOT_FOUND',
|
||||
r._auth('nope', 'x', 'up').code === 'E_NOT_FOUND');
|
||||
|
||||
// 过期
|
||||
let t = 0;
|
||||
const r3 = new Relay({ redis: fakeRedis(), now: () => t });
|
||||
const s3 = await r3.open({ accountId: 'a1', size: 100 });
|
||||
t = 11 * 60 * 1000; // 越过 10 分钟
|
||||
check('过期会话 → E_EXPIRED', r3._auth(s3.id, s3.uploadToken, 'up').code === 'E_EXPIRED');
|
||||
r.stop(); r3.stop();
|
||||
}
|
||||
|
||||
// ══ ③ 真的能传，且服务端不落盘 ══
|
||||
console.log('\n③ 真起 HTTP 跑一遍（不落盘）');
|
||||
{
|
||||
const redis = fakeRedis();
|
||||
const r = new Relay({ redis, now: () => Date.now() });
|
||||
const payload = Buffer.alloc(256 * 1024, 7); // 256KB
|
||||
const s = await r.open({ accountId: 'a1', size: payload.length });
|
||||
|
||||
// 一个"网关"：把 relay 接到 express 风格的路由上
|
||||
const server = http.createServer((req, res) => {
|
||||
const u = new URL(req.url, 'http://x');
|
||||
const id = u.pathname.split('/').pop();
|
||||
const token = u.searchParams.get('t');
|
||||
if (req.method === 'GET') {
|
||||
// 先写头、再等配对：**只写一次**（之前 then+catch 各写一次，
|
||||
// 第二个 writeHead 会抛 ERR_HTTP_HEADERS_SENT）。
|
||||
res.writeHead(200, { 'content-type': 'application/octet-stream' });
|
||||
r.waitDownloader(id, token, res).catch(() => res.destroy());
|
||||
} else if (req.method === 'PUT') {
|
||||
const a = r.attachUploader(id, token, req, res);
|
||||
if (!a.ok) { res.writeHead(409); res.end(a.code); return; }
|
||||
res.writeHead(200); res.end('ok');
|
||||
} else { res.writeHead(405); res.end(); }
|
||||
});
|
||||
await new Promise((res) => server.listen(0, '127.0.0.1', res));
|
||||
const port = server.address().port;
|
||||
|
||||
const before = fs.readdirSync(process.cwd()).length;
|
||||
|
||||
// 接收端先挂上
|
||||
const got = [];
|
||||
const download = new Promise((resolve, reject) => {
|
||||
const req = http.request(
|
||||
{ host: '127.0.0.1', port, method: 'GET', path: '/relay/' + s.id + '?t=' + s.downloadToken },
|
||||
(res) => {
|
||||
if (res.statusCode !== 200) { reject(new Error('down ' + res.statusCode)); return; }
|
||||
res.on('data', (c) => got.push(c));
|
||||
res.on('end', () => resolve(Buffer.concat(got)));
|
||||
});
|
||||
req.on('error', reject);
|
||||
req.end();
|
||||
});
|
||||
|
||||
await new Promise((res) => setTimeout(res, 150)); // 让下载端先挂住
|
||||
|
||||
// 发送端上传
|
||||
await new Promise((resolve, reject) => {
|
||||
const req = http.request(
|
||||
{ host: '127.0.0.1', port, method: 'PUT', path: '/relay/' + s.id + '?t=' + s.uploadToken,
|
||||
headers: { 'content-length': String(payload.length) } },
|
||||
(res) => { res.resume(); res.on('end', resolve); });
|
||||
req.on('error', reject);
|
||||
req.end(payload);
|
||||
});
|
||||
|
||||
const received = await Promise.race([
|
||||
download,
|
||||
new Promise((_, rej) => setTimeout(() => rej(new Error('下载超时')), 5000)),
|
||||
]).catch(() => null);
|
||||
|
||||
check('★ 真传成功，且字节完全一致', received != null && received.equals(payload));
|
||||
check('★ 服务端目录里没有多出文件（**不落盘**）',
|
||||
fs.readdirSync(process.cwd()).length === before);
|
||||
// ⚠️ 这里**刻意不断言系统临时目录**：它是**全局共享**的，别的进程
|
||||
// （甚至本机其它测试）随时在写，"前后条目数相等"这种量法必然漂移 ——
|
||||
// 臣第一版就是这么写的，结果测出了假失败。**量不出来时先怀疑量法。**
|
||||
//
|
||||
// "没落到 temp 再删"这件事的真正证据在**接入层**：
|
||||
// nginx 的 `proxy_request_buffering off` —— 那是服务器侧配置，
|
||||
// 本机测不到，只能在 nginx.conf 里落实并写明理由。
|
||||
check('会话传完即回收（Map 不残留）', r.sessions.size === 0);
|
||||
check('★ 按**实际字节**记了配额（不是声明的 size）',
|
||||
Number(redis.store['relay:quota:' + new Date().toISOString().slice(0, 10) + ':a1']) === payload.length);
|
||||
|
||||
server.close();
|
||||
r.stop();
|
||||
}
|
||||
|
||||
// ══ ④ 假 size 绕不过上限 ══
|
||||
console.log('\n④ 假 size 绕限');
|
||||
{
|
||||
const r = new Relay({ redis: fakeRedis(), now: () => Date.now() });
|
||||
const s = await r.open({ accountId: 'a1', size: 1024 }); // 只声明 1KB
|
||||
const server = http.createServer((req, res) => {
|
||||
const u = new URL(req.url, 'http://x');
|
||||
const id = u.pathname.split('/').pop();
|
||||
if (req.method === 'GET') {
|
||||
res.writeHead(200);
|
||||
r.waitDownloader(id, u.searchParams.get('t'), res).catch(() => res.destroy());
|
||||
} else {
|
||||
const a = r.attachUploader(id, u.searchParams.get('t'), req, res);
|
||||
if (!a.ok) { res.writeHead(409); res.end(a.code); return; }
|
||||
res.writeHead(200); res.end('ok');
|
||||
}
|
||||
});
|
||||
await new Promise((res) => server.listen(0, '127.0.0.1', res));
|
||||
const port = server.address().port;
|
||||
|
||||
const dl = new Promise((resolve) => {
|
||||
const q = http.request({ host: '127.0.0.1', port, method: 'GET',
|
||||
path: '/relay/' + s.id + '?t=' + s.downloadToken }, (res) => {
|
||||
res.on('data', () => {}); res.on('end', resolve); res.on('error', resolve);
|
||||
});
|
||||
q.on('error', resolve); q.end();
|
||||
});
|
||||
await new Promise((res) => setTimeout(res, 150));
|
||||
|
||||
// 发送端声明 1KB、实际发 64KB
|
||||
await new Promise((resolve) => {
|
||||
const q = http.request({ host: '127.0.0.1', port, method: 'PUT',
|
||||
path: '/relay/' + s.id + '?t=' + s.uploadToken }, () => resolve());
|
||||
q.on('error', () => resolve());
|
||||
q.end(Buffer.alloc(64 * 1024, 1));
|
||||
});
|
||||
await Promise.race([dl, new Promise((r2) => setTimeout(r2, 1500))]);
|
||||
check('★ 实际字节超过声明 → 被截断/中断（假 size 绕不过单次上限）',
|
||||
r.sessions.size === 0 || r.activeCount('a1') === 0);
|
||||
server.close(); r.stop();
|
||||
}
|
||||
|
||||
console.log('\n────────────────────────────────────────────────');
|
||||
console.log('中继测试 — 通过: ' + pass + ' 失败: ' + fail);
|
||||
console.log('────────────────────────────────────────────────');
|
||||
process.exit(fail === 0 ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch((e) => { console.error('[relay-test] ' + e.message); process.exit(1); });
|
||||
Reference in new issue
Block a user