zfc/sub2api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 重写 proxy.js — 预收集 body + H1/H2 自适应，本地测试 4/4 通过
Some checks failed
CI / golangci-lint (push) Has been cancelled
Details
Security Scan / backend-security (push) Has been cancelled
Details
CI / test (push) Has been cancelled
Details
Security Scan / frontend-security (push) Has been cancelled
Details

Browse Source
This commit is contained in:
win 2026-03-22 02:19:38 +08:00
parent 88432f9438
commit 5d476fbc09
1 changed files with 106 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

282
tools/node-tls-proxy/proxy.js
View File

@ -19,222 +19,172 @@ const log = (level, msg, extra = {}) => {
}; };
const HEALTH_PATH = '/__health'; const HEALTH_PATH = '/__health';
const h2Hosts = new Set(); // 已知需要 H2 的主机
const h2Sessions = new Map(); // H2 session 缓存
// ─── 协议缓存：记录哪些主机需要 H2 ────────────────────── // ─── H2 session 管理 ────────────────────────────────────
// 首次请求用 H1，如果秒挂（socket hang up < 2s）自动切 H2 并缓存 function getOrCreateH2Session(host) {
const h2Hosts = new Set();
// ─── H2 会话池 ──────────────────────────────────────────
const h2Sessions = new Map();
function getH2Session(host) {
const existing = h2Sessions.get(host); const existing = h2Sessions.get(host);
if (existing && !existing.closed && !existing.destroyed) return existing; if (existing && !existing.closed && !existing.destroyed) return existing;
if (existing) { try { existing.close(); } catch (_) {} }
const session = http2.connect(`https://${host}`); const session = http2.connect(`https://${host}`);
session.on('error', (err) => { session.on('error', (err) => {
log('warn', 'h2_session_error', { host, error: err.message }); log('warn', 'h2_session_error', { host, error: err.message });
h2Sessions.delete(host); h2Sessions.delete(host);
try { session.close(); } catch (_) {}
}); });
session.on('close', () => h2Sessions.delete(host)); session.on('close', () => h2Sessions.delete(host));
session.setTimeout(IDLE_TIMEOUT, () => { session.on('goaway', () => { h2Sessions.delete(host); try { session.close(); } catch (_) {} });
session.close(); session.setTimeout(IDLE_TIMEOUT, () => { session.close(); h2Sessions.delete(host); });
h2Sessions.delete(host);
});
h2Sessions.set(host, session); h2Sessions.set(host, session);
return session; return session;
} }
function waitForConnect(session) {
if (session.connected) return Promise.resolve();
return new Promise((resolve, reject) => {
session.once('connect', resolve);
session.once('error', reject);
const t = setTimeout(() => reject(new Error('h2 connect timeout')), CONNECT_TIMEOUT);
session.once('connect', () => clearTimeout(t));
});
}
// ─── CONNECT 隧道 ──────────────────────────────────────── // ─── CONNECT 隧道 ────────────────────────────────────────
function connectViaProxy(proxyUrl, targetHost, targetPort) { function connectViaProxy(proxyUrl, targetHost, targetPort) {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
const proxy = new URL(proxyUrl); const proxy = new URL(proxyUrl);
const conn = net.connect(parseInt(proxy.port || '80', 10), proxy.hostname, () => { const conn = net.connect(parseInt(proxy.port || '80', 10), proxy.hostname, () => {
const auth = proxy.username const auth = proxy.username
? `Proxy-Authorization: Basic ${Buffer.from( ? `Proxy-Authorization: Basic ${Buffer.from(`${decodeURIComponent(proxy.username)}:${decodeURIComponent(proxy.password || '')}`).toString('base64')}\r\n`
`${decodeURIComponent(proxy.username)}:${decodeURIComponent(proxy.password || '')}`
).toString('base64')}\r\n`
: ''; : '';
conn.write( conn.write(`CONNECT ${targetHost}:${targetPort} HTTP/1.1\r\nHost: ${targetHost}:${targetPort}\r\n${auth}\r\n`);
`CONNECT ${targetHost}:${targetPort} HTTP/1.1\r\n` +
`Host: ${targetHost}:${targetPort}\r\n` + auth + '\r\n'
);
}); });
conn.once('error', reject); conn.once('error', reject);
conn.setTimeout(CONNECT_TIMEOUT, () => conn.destroy(new Error('proxy CONNECT timeout'))); conn.setTimeout(CONNECT_TIMEOUT, () => conn.destroy(new Error('CONNECT timeout')));
let buf = ''; let buf = '';
const onData = (chunk) => { conn.on('data', function onData(chunk) {
buf += chunk.toString(); buf += chunk.toString();
const idx = buf.indexOf('\r\n\r\n'); const idx = buf.indexOf('\r\n\r\n');
if (idx === -1) return; if (idx === -1) return;
conn.removeListener('data', onData); conn.removeListener('data', onData);
const code = parseInt(buf.split(' ')[1], 10); const code = parseInt(buf.split(' ')[1], 10);
if (code === 200) { if (code === 200) { conn.setTimeout(0); resolve(conn); }
conn.setTimeout(0); else { conn.destroy(); reject(new Error(`CONNECT ${code}`)); }
const rest = buf.slice(idx + 4); });
if (rest.length > 0) conn.unshift(Buffer.from(rest)); });
resolve(conn); }
} else {
conn.destroy(); // ─── 收集请求体 ──────────────────────────────────────────
reject(new Error(`CONNECT failed: ${code}`)); function collectBody(req) {
} return new Promise((resolve) => {
}; const chunks = [];
conn.on('data', onData); req.on('data', (c) => chunks.push(c));
req.on('end', () => resolve(Buffer.concat(chunks)));
req.on('error', () => resolve(Buffer.concat(chunks)));
}); });
} }
// ─── H1 代理 ───────────────────────────────────────────── // ─── H1 代理 ─────────────────────────────────────────────
function proxyViaH1(targetHost, req, res) { function sendViaH1(targetHost, method, path, reqHeaders, body, res) {
return new Promise((resolve) => { return new Promise((resolve) => {
const headers = { ...req.headers }; const headers = { ...reqHeaders, host: targetHost };
headers.host = targetHost; ['x-forwarded-host', 'connection', 'keep-alive', 'proxy-connection', 'transfer-encoding'].forEach(h => delete headers[h]);
delete headers['x-forwarded-host']; if (body.length > 0) headers['content-length'] = String(body.length);
delete headers['connection'];
delete headers['keep-alive'];
delete headers['proxy-connection'];
delete headers['transfer-encoding'];
const opts = {
hostname: targetHost, port: 443, path: req.url,
method: req.method, headers, servername: targetHost,
timeout: CONNECT_TIMEOUT,
};
const opts = { hostname: targetHost, port: 443, path, method, headers, servername: targetHost, timeout: CONNECT_TIMEOUT };
const startTime = Date.now(); const startTime = Date.now();
let proxyReq;
const doRequest = (requestOpts) => {
proxyReq = https.request(requestOpts);
const finish = (requestOpts) => {
const proxyReq = https.request(requestOpts);
proxyReq.on('response', (proxyRes) => { proxyReq.on('response', (proxyRes) => {
log('info', 'proxy_response', { host: targetHost, status: proxyRes.statusCode, path: req.url, proto: 'h1' }); log('info', 'proxy_response', { host: targetHost, status: proxyRes.statusCode, path, proto: 'h1' });
const rh = { ...proxyRes.headers }; const rh = { ...proxyRes.headers };
delete rh['connection']; delete rh['connection']; delete rh['keep-alive'];
delete rh['keep-alive'];
res.writeHead(proxyRes.statusCode, rh); res.writeHead(proxyRes.statusCode, rh);
proxyRes.pipe(res, { end: true }); proxyRes.pipe(res, { end: true });
proxyRes.on('error', (e) => { log('error', 'h1_response_error', { error: e.message }); res.end(); });
resolve('ok'); resolve('ok');
}); });
proxyReq.on('error', (err) => { proxyReq.on('error', (err) => {
const elapsed = Date.now() - startTime; if (err.message === 'socket hang up' && (Date.now() - startTime) < 2000) {
// socket hang up < 2 秒 = 服务器拒绝 H1，切换到 H2 log('info', 'h1_rejected_switching_to_h2', { host: targetHost });
if (err.message === 'socket hang up' && elapsed < 2000) {
log('info', 'h1_rejected_switching_to_h2', { host: targetHost, elapsed });
h2Hosts.add(targetHost); h2Hosts.add(targetHost);
proxyViaH2(targetHost, req, res); sendViaH2(targetHost, method, path, reqHeaders, body, res).then(() => resolve('h2'));
resolve('h2_fallback');
return; return;
} }
log('error', 'h1_upstream_error', { error: err.message, host: targetHost, path: req.url }); log('error', 'h1_error', { error: err.message, host: targetHost, path });
if (!res.headersSent) { if (!res.headersSent) { res.writeHead(502); res.end(JSON.stringify({ error: err.message })); }
res.writeHead(502, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: 'upstream_error', message: err.message }));
}
resolve('error'); resolve('error');
}); });
proxyReq.on('timeout', () => proxyReq.destroy(new Error('timeout')));
proxyReq.on('timeout', () => proxyReq.destroy(new Error('upstream timeout'))); proxyReq.end(body);
req.on('close', () => { if (!proxyReq.destroyed) proxyReq.destroy(); });
req.pipe(proxyReq, { end: true });
}; };
if (UPSTREAM_PROXY) { if (UPSTREAM_PROXY) {
connectViaProxy(UPSTREAM_PROXY, targetHost, 443) connectViaProxy(UPSTREAM_PROXY, targetHost, 443)
.then((socket) => { opts.socket = socket; opts.agent = false; doRequest(opts); }) .then((socket) => { opts.socket = socket; opts.agent = false; finish(opts); })
.catch((err) => { .catch((err) => { log('error', 'tunnel_failed', { error: err.message }); if (!res.headersSent) { res.writeHead(502); res.end('tunnel error'); } resolve('error'); });
log('error', 'proxy_tunnel_failed', { error: err.message, host: targetHost });
if (!res.headersSent) {
res.writeHead(502, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: 'proxy_tunnel_error' }));
}
resolve('error');
});
} else { } else {
doRequest(opts); finish(opts);
} }
}); });
} }
// ─── H2 代理 ───────────────────────────────────────────── // ─── H2 代理 ─────────────────────────────────────────────
function proxyViaH2(targetHost, req, res) { async function sendViaH2(targetHost, method, path, reqHeaders, body, res) {
try { try {
const session = getH2Session(targetHost); const session = getOrCreateH2Session(targetHost);
await waitForConnect(session);
const headers = {}; const headers = {};
// 只拷贝合法的 H2 头（跳过 H1 专用头和连接头） const skip = new Set(['host','connection','keep-alive','proxy-connection','transfer-encoding','upgrade','x-forwarded-host','http2-settings']);
const skipHeaders = new Set([ for (const [k, v] of Object.entries(reqHeaders)) {
'host', 'connection', 'keep-alive', 'proxy-connection', if (!skip.has(k.toLowerCase())) headers[k] = v;
'transfer-encoding', 'upgrade', 'x-forwarded-host',
'http2-settings',
]);
for (const [k, v] of Object.entries(req.headers)) {
if (!skipHeaders.has(k.toLowerCase())) {
headers[k] = v;
}
} }
headers[':method'] = method;
// H2 伪头 headers[':path'] = path;
headers[':method'] = req.method;
headers[':path'] = req.url;
headers[':authority'] = targetHost; headers[':authority'] = targetHost;
headers[':scheme'] = 'https'; headers[':scheme'] = 'https';
if (body.length > 0) headers['content-length'] = String(body.length);
const h2Stream = session.request(headers); const stream = session.request(headers);
let responded = false; let responded = false;
h2Stream.on('response', (h2Headers) => { stream.on('response', (h2h) => {
responded = true; responded = true;
const status = h2Headers[':status'] || 502; const status = h2h[':status'] || 502;
const respHeaders = {}; const rh = {};
for (const [k, v] of Object.entries(h2Headers)) { for (const [k, v] of Object.entries(h2h)) { if (!k.startsWith(':')) rh[k] = v; }
if (!k.startsWith(':')) respHeaders[k] = v; log('info', 'proxy_response', { host: targetHost, status, path, proto: 'h2' });
} res.writeHead(status, rh);
log('info', 'proxy_response', { host: targetHost, status, path: req.url, proto: 'h2' }); stream.on('data', (c) => res.write(c));
res.writeHead(status, respHeaders); stream.on('end', () => res.end());
h2Stream.pipe(res, { end: true });
}); });
h2Stream.on('error', (err) => { stream.on('error', (err) => {
log('error', 'h2_stream_error', { error: err.message, host: targetHost, path: req.url }); if (err.message && err.message.includes('NGHTTP2')) {
h2Sessions.delete(targetHost); // 清理坏 session h2Sessions.delete(targetHost);
try { session.close(); } catch (_) {}
}
if (responded) { if (!res.writableEnded) res.end(); return; }
log('error', 'h2_error', { error: err.message, host: targetHost, path });
if (!res.headersSent) { res.writeHead(502); res.end(JSON.stringify({ error: err.message })); }
});
stream.on('close', () => {
if (!responded && !res.headersSent) { if (!responded && !res.headersSent) {
res.writeHead(502, { 'content-type': 'application/json' }); log('warn', 'h2_no_response', { host: targetHost, path });
res.end(JSON.stringify({ error: 'h2_error', message: err.message })); res.writeHead(502); res.end('{"error":"h2_no_response"}');
} } else if (!res.writableEnded) { res.end(); }
}); });
h2Stream.on('close', () => { stream.setTimeout(CONNECT_TIMEOUT, () => stream.close());
if (!responded && !res.headersSent) { stream.end(body);
log('warn', 'h2_stream_closed_no_response', { host: targetHost, path: req.url });
res.writeHead(502, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: 'h2_no_response' }));
}
});
// 超时
h2Stream.setTimeout(CONNECT_TIMEOUT, () => {
log('warn', 'h2_timeout', { host: targetHost, path: req.url });
h2Stream.close();
});
req.on('close', () => {
if (!h2Stream.destroyed) h2Stream.close();
});
// pipe 请求体
req.pipe(h2Stream, { end: true });
} catch (err) { } catch (err) {
log('error', 'h2_proxy_exception', { error: err.message, host: targetHost }); log('error', 'h2_exception', { error: err.message, host: targetHost });
h2Sessions.delete(targetHost); h2Sessions.delete(targetHost);
if (!res.headersSent) { if (!res.headersSent) { res.writeHead(502); res.end(JSON.stringify({ error: err.message })); }
res.writeHead(502, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: 'h2_exception', message: err.message }));
}
} }
} }
@ -243,64 +193,44 @@ async function proxyRequest(req, res) {
const targetHost = req.headers['x-forwarded-host'] || UPSTREAM_HOST; const targetHost = req.headers['x-forwarded-host'] || UPSTREAM_HOST;
log('info', 'proxy_request', { host: targetHost, method: req.method, path: req.url }); log('info', 'proxy_request', { host: targetHost, method: req.method, path: req.url });
// 已知需要 H2 的主机直接走 H2 const body = await collectBody(req);
if (h2Hosts.has(targetHost)) {
proxyViaH2(targetHost, req, res);
return;
}
// 首次请求走 H1，如果秒挂自动切 H2 if (h2Hosts.has(targetHost)) {
await proxyViaH1(targetHost, req, res); await sendViaH2(targetHost, req.method, req.url, req.headers, body, res);
} else {
await sendViaH1(targetHost, req.method, req.url, req.headers, body, res);
}
} }
// ─── HTTP 服务器 ───────────────────────────────────────── // ─── HTTP 服务器 ─────────────────────────────────────────
const server = http.createServer((req, res) => { const server = http.createServer((req, res) => {
if (req.url === HEALTH_PATH) { if (req.url === HEALTH_PATH) {
res.writeHead(200, { 'content-type': 'application/json' }); res.writeHead(200, { 'content-type': 'application/json' });
res.end(JSON.stringify({ res.end(JSON.stringify({ status: 'ok', node: process.version, openssl: process.versions.openssl, uptime: process.uptime(), h2Hosts: [...h2Hosts] }));
status: 'ok',
upstream: UPSTREAM_HOST,
node: process.version,
openssl: process.versions.openssl,
uptime: process.uptime(),
h2Hosts: [...h2Hosts],
}));
return; return;
} }
proxyRequest(req, res).catch((err) => { proxyRequest(req, res).catch((err) => {
log('error', 'unhandled_error', { error: err.message }); log('error', 'unhandled', { error: err.message });
if (!res.headersSent) { if (!res.headersSent) { res.writeHead(500); res.end('internal error'); }
res.writeHead(500, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: 'internal_error' }));
}
}); });
}); });
server.timeout = 0; server.timeout = 0;
server.keepAliveTimeout = IDLE_TIMEOUT; server.keepAliveTimeout = IDLE_TIMEOUT;
server.headersTimeout = 60000; server.headersTimeout = 60000;
server.listen(LISTEN_PORT, LISTEN_HOST, () => { server.listen(LISTEN_PORT, LISTEN_HOST, () => {
log('info', 'node-tls-proxy started', { log('info', 'node-tls-proxy started', { listen: `${LISTEN_HOST}:${LISTEN_PORT}`, node: process.version, openssl: process.versions.openssl });
listen: `${LISTEN_HOST}:${LISTEN_PORT}`,
upstream: `${UPSTREAM_HOST}:443`,
proxy: UPSTREAM_PROXY || '(direct)',
node: process.version,
openssl: process.versions.openssl,
});
}); });
let shuttingDown = false; let stopping = false;
function shutdown(signal) { function shutdown(sig) {
if (shuttingDown) return; if (stopping) return; stopping = true;
shuttingDown = true; for (const s of h2Sessions.values()) try { s.close(); } catch (_) {}
log('info', `received ${signal}, shutting down`);
for (const s of h2Sessions.values()) s.close();
h2Sessions.clear(); h2Sessions.clear();
server.close(() => process.exit(0)); server.close(() => process.exit(0));
setTimeout(() => process.exit(1), 5000); setTimeout(() => process.exit(1), 5000);
} }
process.on('SIGTERM', () => shutdown('SIGTERM')); process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT')); process.on('SIGINT', () => shutdown('SIGINT'));
process.on('uncaughtException', (err) => log('error', 'uncaught', { error: err.message, stack: err.stack })); process.on('uncaughtException', (e) => log('error', 'uncaught', { error: e.message }));
process.on('unhandledRejection', (r) => log('error', 'unhandled_rejection', { error: String(r) })); process.on('unhandledRejection', (r) => log('error', 'rejection', { error: String(r) }));