From 6fe6c5f354049a0870d805bde601d443b27011af Mon Sep 17 00:00:00 2001 From: Bram Date: Thu, 9 Apr 2026 21:50:57 +0200 Subject: [PATCH] some fixes --- Dockers/xtream-proxy/Dockerfile | 2 +- Dockers/xtream-proxy/package.json | 2 +- Dockers/xtream-proxy/server.js | 274 +++++++++++++++++++++++++----- 3 files changed, 231 insertions(+), 47 deletions(-) diff --git a/Dockers/xtream-proxy/Dockerfile b/Dockers/xtream-proxy/Dockerfile index 8f4a510..a39f8e0 100644 --- a/Dockers/xtream-proxy/Dockerfile +++ b/Dockers/xtream-proxy/Dockerfile @@ -24,4 +24,4 @@ ENV HEALTH_PATH=/ HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \ CMD node -e "require('http').get('http://127.0.0.1:'+(process.env.LISTEN_PORT||8080)+'/health',(r)=>{let d='';r.on('data',c=>d+=c);r.on('end',()=>{try{process.exit(JSON.parse(d).ok?0:1)}catch{process.exit(1)}})}).on('error',()=>process.exit(1))" -CMD ["npm", "start"] +CMD ["node", "server.js"] diff --git a/Dockers/xtream-proxy/package.json b/Dockers/xtream-proxy/package.json index 62050f4..e7bc5a8 100644 --- a/Dockers/xtream-proxy/package.json +++ b/Dockers/xtream-proxy/package.json @@ -7,6 +7,6 @@ }, "dependencies": { "express": "^4.21.0", - "http-proxy": "^1.18.1" + "ws": "^8.18.0" } } diff --git a/Dockers/xtream-proxy/server.js b/Dockers/xtream-proxy/server.js index ff94a15..8b25bca 100644 --- a/Dockers/xtream-proxy/server.js +++ b/Dockers/xtream-proxy/server.js @@ -4,30 +4,32 @@ const fs = require('fs'); const http = require('http'); const https = require('https'); const express = require('express'); -const httpProxy = require('http-proxy'); +const WebSocket = require('ws'); const LISTEN_PORT = Number(process.env.LISTEN_PORT || 8080); const HOSTS_FILE = process.env.HOSTS_FILE || '/config/hosts.txt'; -const CHECK_INTERVAL_SEC = Math.max(5, Number(process.env.CHECK_INTERVAL_SEC || 30)); +const CHECK_INTERVAL_SEC = Math.max( + 5, + Number(process.env.CHECK_INTERVAL_SEC || 30) || 30 +); const PING_TIMEOUT_MS = Math.max(1000, Number(process.env.PING_TIMEOUT_MS || 8000)); const HEALTH_PATH = process.env.HEALTH_PATH || '/'; +const REJECT_UNAUTHORIZED = process.env.PROXY_INSECURE_TLS !== '1'; + +const HOP_BY_HOP = new Set([ + 'connection', + 'keep-alive', + 'proxy-authenticate', + 'proxy-authorization', + 'te', + 'trailers', + 'transfer-encoding', + 'upgrade', +]); let currentTarget = null; let lastStats = { checkedAt: null, results: [], selected: null }; -const proxy = httpProxy.createProxyServer({ - ws: true, - xfwd: true, - proxyTimeout: 0, -}); - -proxy.on('error', (err, req, res) => { - if (res && !res.headersSent) { - res.writeHead(502, { 'Content-Type': 'text/plain' }); - res.end(`Upstream error: ${err.message}`); - } -}); - function readHostsFile() { if (!fs.existsSync(HOSTS_FILE)) { console.error(`Hosts file not found: ${HOSTS_FILE}`); @@ -52,6 +54,8 @@ function readHostsFile() { } function pingBase(base, path) { + const HARD_CAP_MS = PING_TIMEOUT_MS + 2000; + return new Promise((resolve) => { let url; try { @@ -62,9 +66,29 @@ function pingBase(base, path) { return; } + let settled = false; + let clientReq = null; + + const finish = (result) => { + if (settled) return; + settled = true; + clearTimeout(hardTimer); + try { + clientReq?.destroy(); + } catch (_) {} + resolve(result); + }; + + const hardTimer = setTimeout(() => { + try { + clientReq?.destroy(); + } catch (_) {} + finish({ base, ok: false, ms: null, error: 'hard_timeout' }); + }, HARD_CAP_MS); + const lib = url.protocol === 'https:' ? https : http; const start = Date.now(); - const req = lib.request( + clientReq = lib.request( { protocol: url.protocol, hostname: url.hostname, @@ -72,6 +96,7 @@ function pingBase(base, path) { path: url.pathname + url.search, method: 'GET', timeout: PING_TIMEOUT_MS, + rejectUnauthorized: REJECT_UNAUTHORIZED, headers: { 'User-Agent': 'xtream-proxy-health/1.0', Connection: 'close', @@ -82,21 +107,21 @@ function pingBase(base, path) { res.on('end', () => { const ms = Date.now() - start; const ok = res.statusCode >= 200 && res.statusCode < 500; - resolve({ base, ok, ms, status: res.statusCode }); + finish({ base, ok, ms, status: res.statusCode }); }); res.on('error', () => { - resolve({ base, ok: false, ms: null, error: 'response_error' }); + finish({ base, ok: false, ms: null, error: 'response_error' }); }); } ); - req.on('error', (e) => { - resolve({ base, ok: false, ms: null, error: e.code || e.message }); + clientReq.on('error', (e) => { + finish({ base, ok: false, ms: null, error: e.code || e.message }); }); - req.on('timeout', () => { - req.destroy(); - resolve({ base, ok: false, ms: null, error: 'timeout' }); + clientReq.on('timeout', () => { + clientReq.destroy(); + finish({ base, ok: false, ms: null, error: 'timeout' }); }); - req.end(); + clientReq.end(); }); } @@ -123,15 +148,14 @@ async function pickFastestHost() { }; if (ok[0]) { - if (currentTarget !== ok[0].base) { - console.log( - `Selected upstream ${ok[0].base} (${ok[0].ms} ms). Others: ${ok - .slice(1) - .map((x) => `${x.base}=${x.ms}ms`) - .join(', ') || 'none'}` - ); - } + const prev = currentTarget; currentTarget = ok[0].base; + const others = ok.slice(1).map((x) => `${x.base}=${x.ms}ms`).join(', ') || 'none'; + if (prev !== currentTarget) { + console.log(`Upstream switched → ${currentTarget} (${ok[0].ms} ms). Also: ${others}`); + } else { + console.log(`Health check: keeping ${currentTarget} (${ok[0].ms} ms). Others: ${others}`); + } } else { console.warn( 'No host responded successfully; keeping previous target:', @@ -140,10 +164,108 @@ async function pickFastestHost() { } } +let checkTimer = null; +let checkRunning = false; + function scheduleChecks() { - setInterval(() => { - pickFastestHost().catch((e) => console.error('Health check error:', e)); - }, CHECK_INTERVAL_SEC * 1000); + const delayMs = CHECK_INTERVAL_SEC * 1000; + const tick = async () => { + if (checkRunning) { + console.warn('Health check still running; skipping overlapping run (next in ' + CHECK_INTERVAL_SEC + 's)'); + checkTimer = setTimeout(tick, delayMs); + return; + } + checkRunning = true; + try { + await pickFastestHost(); + } catch (e) { + console.error('Health check error:', e); + } finally { + checkRunning = false; + checkTimer = setTimeout(tick, delayMs); + } + }; + checkTimer = setTimeout(tick, delayMs); +} + +function filterHopByHop(headers) { + const out = Object.create(null); + if (!headers) return out; + for (const [k, v] of Object.entries(headers)) { + if (v === undefined) continue; + if (HOP_BY_HOP.has(k.toLowerCase())) continue; + out[k] = v; + } + return out; +} + +function buildRequestHeaders(req, targetUrl) { + const out = filterHopByHop(req.headers); + const existing = req.headers['x-forwarded-for']; + const client = req.socket.remoteAddress || ''; + out['x-forwarded-for'] = existing ? `${existing}, ${client}` : client; + const xfProto = req.headers['x-forwarded-proto']; + out['x-forwarded-proto'] = typeof xfProto === 'string' && xfProto ? xfProto : 'http'; + const xfHost = req.headers['x-forwarded-host'] || req.headers.host; + if (xfHost) out['x-forwarded-host'] = xfHost; + out.host = targetUrl.host; + return out; +} + +function proxyHttp(req, res) { + const base = currentTarget.replace(/\/$/, ''); + let targetUrl; + try { + targetUrl = new URL(req.url, `${base}/`); + } catch { + res.status(400).type('text/plain').send('Bad request URL'); + return; + } + + const lib = targetUrl.protocol === 'https:' ? https : http; + const opts = { + protocol: targetUrl.protocol, + hostname: targetUrl.hostname, + port: targetUrl.port || (targetUrl.protocol === 'https:' ? 443 : 80), + path: targetUrl.pathname + targetUrl.search, + method: req.method, + headers: buildRequestHeaders(req, targetUrl), + rejectUnauthorized: REJECT_UNAUTHORIZED, + }; + + const pReq = lib.request(opts, (pRes) => { + res.writeHead(pRes.statusCode, filterHopByHop(pRes.headers)); + pRes.pipe(res); + }); + + pReq.on('error', (err) => { + if (!res.headersSent) { + res.writeHead(502, { 'Content-Type': 'text/plain' }); + } + res.end(`Upstream error: ${err.message}`); + }); + + req.pipe(pReq); +} + +function toWsUrl(httpBase, reqPath) { + const base = httpBase.replace(/\/$/, ''); + const u = new URL(reqPath, `${base}/`); + const proto = u.protocol === 'https:' ? 'wss:' : 'ws:'; + return `${proto}//${u.host}${u.pathname}${u.search}`; +} + +function buildWsClientHeaders(req, targetWsUrl) { + const u = new URL(targetWsUrl); + const out = filterHopByHop(req.headers); + for (const key of Object.keys(out)) { + if (key.toLowerCase().startsWith('sec-websocket')) { + delete out[key]; + } + } + delete out.host; + out.host = u.host; + return out; } const app = express(); @@ -162,24 +284,73 @@ app.use((req, res) => { res.status(503).type('text/plain').send('No upstream available yet (hosts file empty or all pings failed).'); return; } - proxy.web(req, res, { - target: currentTarget, - changeOrigin: true, - secure: process.env.PROXY_INSECURE_TLS !== '1', - }); + proxyHttp(req, res); }); const server = http.createServer(app); +const wss = new WebSocket.Server({ noServer: true }); server.on('upgrade', (req, socket, head) => { if (!currentTarget) { + socket.write('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n'); socket.destroy(); return; } - proxy.ws(req, socket, head, { - target: currentTarget, - changeOrigin: true, - secure: process.env.PROXY_INSECURE_TLS !== '1', + + let upstreamUrl; + try { + upstreamUrl = toWsUrl(currentTarget, req.url); + } catch { + socket.destroy(); + return; + } + + wss.handleUpgrade(req, socket, head, (clientWs) => { + const upstream = new WebSocket(upstreamUrl, { + rejectUnauthorized: REJECT_UNAUTHORIZED, + headers: buildWsClientHeaders(req, upstreamUrl), + }); + + const pending = []; + let upstreamReady = false; + + clientWs.on('message', (data, isBinary) => { + if (upstreamReady && upstream.readyState === WebSocket.OPEN) { + upstream.send(data, { binary: isBinary }); + } else { + pending.push({ data, isBinary }); + } + }); + + upstream.on('open', () => { + upstreamReady = true; + for (const m of pending) { + if (upstream.readyState === WebSocket.OPEN) upstream.send(m.data, { binary: m.isBinary }); + } + pending.length = 0; + + upstream.on('message', (data, isBinary) => { + if (clientWs.readyState === WebSocket.OPEN) clientWs.send(data, { binary: isBinary }); + }); + clientWs.on('ping', (buf) => upstream.ping(buf)); + upstream.on('ping', (buf) => clientWs.ping(buf)); + clientWs.on('pong', (buf) => upstream.pong(buf)); + upstream.on('pong', (buf) => clientWs.pong(buf)); + }); + + const shutdown = () => { + try { + clientWs.terminate(); + } catch (_) {} + try { + upstream.terminate(); + } catch (_) {} + }; + + upstream.on('error', shutdown); + clientWs.on('error', shutdown); + upstream.on('close', shutdown); + clientWs.on('close', shutdown); }); }); @@ -187,11 +358,24 @@ async function main() { await pickFastestHost(); server.listen(LISTEN_PORT, () => { console.log(`Xtream proxy listening on :${LISTEN_PORT}`); - console.log(`Hosts file: ${HOSTS_FILE}, interval: ${CHECK_INTERVAL_SEC}s, health path: ${HEALTH_PATH}`); + console.log( + `Hosts file: ${HOSTS_FILE}, interval: ${CHECK_INTERVAL_SEC}s (from end of each run), health path: ${HEALTH_PATH}` + ); scheduleChecks(); }); } +function shutdown() { + if (checkTimer) { + clearTimeout(checkTimer); + checkTimer = null; + } + server.close(() => process.exit(0)); +} + +process.on('SIGTERM', shutdown); +process.on('SIGINT', shutdown); + main().catch((e) => { console.error(e); process.exit(1);