From e646e368696b6d97c5a4b3df6bfb5d5c8be5c922 Mon Sep 17 00:00:00 2001 From: Technomancer702 Date: Mon, 12 Oct 2026 16:06:52 -0700 Subject: [PATCH] fix(proxy): stream binary media with cancellation and bounded HLS buffers Adapted from MikaelKW/nodecast-tv-plus commit 011b634417b2ecd2a2844df9808b42de831ba194. Use awaited pipeline, abort upstream on disconnect, avoid retries after headers, preserve real range support, and rewrite bounded manifests using relative proxy paths. Cover interrupted streams, first-byte delivery, ranges, redirects, split signatures, and invalid inputs. Co-authored-by: Mikael --- server/routes/proxy.js | 179 +-------------------------------- server/services/proxyStream.js | 138 +++++++++++++++++++++++++ test/proxy-stream.test.js | 121 ++++++++++++++++++++++ 3 files changed, 260 insertions(+), 178 deletions(-) create mode 100644 server/services/proxyStream.js create mode 100644 test/proxy-stream.test.js diff --git a/server/routes/proxy.js b/server/routes/proxy.js index 9ad3a26..b5d0b0d 100644 --- a/server/routes/proxy.js +++ b/server/routes/proxy.js @@ -597,184 +597,7 @@ router.post('/epg/:sourceId/channels', async (req, res) => { * This handles CORS for streams that don't allow cross-origin * Supports HTTP Range requests for video seeking */ -router.get('/stream', async (req, res) => { - const maxRetries = 2; - let lastError = null; - - for (let attempt = 1; attempt <= maxRetries; attempt++) { - try { - let { url } = req.query; - if (!url) { - return res.status(400).json({ error: 'URL required' }); - } - - // Forward some headers to be more "transparent" back to the origin - // Pluto TV uses multiple domains for content delivery - const plutoDomains = ['pluto.tv', 'pluto.io', 'plutotv.net', 'siloh.pluto.tv', 'service-stitcher']; - const isPluto = plutoDomains.some(domain => url.includes(domain)); - - const headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', - 'Accept': '*/*', - 'Accept-Language': 'en-US,en;q=0.9', - // Using https and matching the origin of the request - 'Origin': isPluto ? 'https://pluto.tv' : new URL(url).origin, - 'Referer': isPluto ? 'https://pluto.tv/' : new URL(url).origin + '/' - }; - - // Forward Range header for video seeking support - const rangeHeader = req.get('range'); - if (rangeHeader) { - headers['Range'] = rangeHeader; - } - - const response = await fetch(url, { headers }); - - // Retry on 5xx errors (transient upstream issues) - if (response.status >= 500 && attempt < maxRetries) { - console.log(`[Proxy] Upstream 5xx error (attempt ${attempt}/${maxRetries}), retrying in 500ms...`); - await new Promise(r => setTimeout(r, 500)); - continue; - } - - if (!response.ok) { - console.error(`Upstream error for ${url.substring(0, 80)}...: ${response.status} ${response.statusText}`); - if (response.status === 403) { - const errorBody = await response.text().catch(() => 'N/A'); - console.error(`403 Response body: ${errorBody.substring(0, 200)}`); - } - return res.status(response.status).send(`Failed to fetch stream: ${response.statusText}`); - } - - const contentType = response.headers.get('content-type') || ''; - res.set('Access-Control-Allow-Origin', '*'); - - // Forward range-related headers for video seeking support - const contentLength = response.headers.get('content-length'); - const contentRange = response.headers.get('content-range'); - const acceptRanges = response.headers.get('accept-ranges'); - - if (contentLength) { - res.set('Content-Length', contentLength); - } - if (contentRange) { - res.set('Content-Range', contentRange); - } - if (acceptRanges) { - res.set('Accept-Ranges', acceptRanges); - } else if (contentLength && !contentRange) { - // If server supports content-length but didn't explicitly state accept-ranges, - // we can safely assume it supports byte ranges - res.set('Accept-Ranges', 'bytes'); - } - - // Set status code (206 for partial content when range request was made) - res.status(response.status); - - // Create an async iterator for the response body - const iterator = response.body[Symbol.asyncIterator](); - const first = await iterator.next(); - - if (first.done) { - res.set('Content-Type', contentType || 'application/octet-stream'); - return res.end(); - } - - const firstChunk = Buffer.from(first.value); - - // Peek at first bytes to check for HLS manifest ({ #EXTM3U }) - const textPrefix = firstChunk.subarray(0, 7).toString('utf8'); - const contentLooksLikeHls = textPrefix === '#EXTM3U'; - - if (contentLooksLikeHls) { - // HLS Manifest: We must read the WHOLE manifest to rewrite it - const chunks = [firstChunk]; - - // Consume the rest of the stream - let result = await iterator.next(); - while (!result.done) { - chunks.push(Buffer.from(result.value)); - result = await iterator.next(); - } - - const buffer = Buffer.concat(chunks); - const finalUrl = response.url || url; - console.log(`[Proxy] Processing HLS manifest from: ${finalUrl.substring(0, 80)}...`); - res.set('Content-Type', 'application/vnd.apple.mpegurl'); - - let manifest = buffer.toString('utf-8'); - - const finalUrlObj = new URL(finalUrl); - const baseUrl = finalUrlObj.origin + finalUrlObj.pathname.substring(0, finalUrlObj.pathname.lastIndexOf('/') + 1); - - manifest = manifest.split('\n').map(line => { - const trimmed = line.trim(); - if (trimmed === '' || trimmed.startsWith('#')) { - // Handle both URI="..." and URI='...' formats - if (trimmed.includes('URI=')) { - // Replace both double and single quoted URIs - return line.replace(/URI=["']([^"']+)["']/g, (match, p1) => { - try { - const absoluteUrl = new URL(p1, baseUrl).href; - return `URI="${req.protocol}://${req.get('host')}${req.baseUrl}/stream?url=${encodeURIComponent(absoluteUrl)}"`; - } catch (e) { - return match; - } - }); - } - return line; - } - - // Stream URL handling - try { - let absoluteUrl; - if (trimmed.startsWith('http://') || trimmed.startsWith('https://')) { - absoluteUrl = trimmed; - } else { - absoluteUrl = new URL(trimmed, baseUrl).href; - } - return `${req.protocol}://${req.get('host')}${req.baseUrl}/stream?url=${encodeURIComponent(absoluteUrl)}`; - } catch (e) { return line; } - }).join('\n'); - - return res.send(manifest); - } - - // Binary content (Video Segment or Key): Collect and send - console.log(`[Proxy] Serving binary content (${contentType})`); - res.set('Content-Type', contentType || 'application/octet-stream'); - - // For small files (like encryption keys), collect all data and send at once - // This ensures proper Content-Length and response completion - const chunks = [firstChunk]; - let result = await iterator.next(); - while (!result.done) { - chunks.push(Buffer.from(result.value)); - result = await iterator.next(); - } - const fullContent = Buffer.concat(chunks); - - // Set Content-Length for proper client handling - res.set('Content-Length', fullContent.length); - res.send(fullContent); - return; // Success - exit the retry loop - - } catch (err) { - lastError = err; - console.error(`Stream proxy error (attempt ${attempt}/${maxRetries}):`, err.message); - if (attempt < maxRetries) { - console.log('[Proxy] Retrying after error...'); - await new Promise(r => setTimeout(r, 500)); - continue; - } - } - } - - // All retries failed - if (!res.headersSent) { - res.status(500).json({ error: lastError?.message || 'Stream proxy failed after retries' }); - } -}); +router.get('/stream', require('../services/proxyStream')); /** * Proxy images (channel logos, posters) diff --git a/server/services/proxyStream.js b/server/services/proxyStream.js new file mode 100644 index 0000000..cc92f57 --- /dev/null +++ b/server/services/proxyStream.js @@ -0,0 +1,138 @@ +const { Readable } = require('node:stream'); +const { pipeline } = require('node:stream/promises'); + +const MAX_HLS_MANIFEST_BYTES = 5 * 1024 * 1024; + +async function* prepend(first, iterator) { + try { + yield first; + for (let next = await iterator.next(); !next.done; next = await iterator.next()) { + yield Buffer.from(next.value); + } + } finally { + try { await iterator.return?.(); } catch { /* Upstream may already be aborted. */ } + } +} + +async function collectManifest(first, iterator) { + const chunks = []; + let size = 0; + for await (const chunk of prepend(first, iterator)) { + size += chunk.length; + if (size > MAX_HLS_MANIFEST_BYTES) { + const error = new Error('HLS manifest exceeds the proxy size limit'); + error.statusCode = 502; + throw error; + } + chunks.push(chunk); + } + return Buffer.concat(chunks, size).toString('utf8'); +} + +function rewriteManifest(manifest, sourceUrl, proxyPath) { + const proxied = value => `${proxyPath}?url=${encodeURIComponent(new URL(value, sourceUrl).href)}`; + return manifest.split('\n').map(line => { + const trimmed = line.trim(); + if (!trimmed) return line; + if (trimmed.startsWith('#')) { + return line.replace(/URI=["']([^"']+)["']/g, (match, value) => { + try { return `URI="${proxied(value)}"`; } catch { return match; } + }); + } + try { return proxied(trimmed); } catch { return line; } + }).join('\n'); +} + +// Adapted from NodeCast TV Plus's streaming proxy. Binary media is never +// buffered in full; only manifests are collected, under a fixed size limit. +async function proxyStream(req, res) { + let url; + try { + const raw = req.query.url; + if (typeof raw !== 'string' || /[\x00-\x1f\x7f]/.test(raw)) throw new Error(); + const parsed = new URL(raw); + if (!['http:', 'https:'].includes(parsed.protocol)) throw new Error(); + url = parsed.href; + } catch { + return res.status(400).json({ error: 'A valid HTTP(S) URL is required' }); + } + + let lastError; + for (let attempt = 1; attempt <= 2; attempt++) { + if (res.destroyed) return; + const controller = new AbortController(); + const abort = () => { if (!res.writableEnded) controller.abort(); }; + res.once('close', abort); + try { + const origin = new URL(url).origin; + const pluto = ['pluto.tv', 'pluto.io', 'plutotv.net', 'service-stitcher'].some(domain => url.includes(domain)); + const headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', + Accept: '*/*', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'identity', + Origin: pluto ? 'https://pluto.tv' : origin, + Referer: pluto ? 'https://pluto.tv/' : `${origin}/` + }; + const range = req.get('range'); + if (range) headers.Range = range; + const upstream = await fetch(url, { headers, signal: controller.signal }); + if (upstream.status >= 500 && attempt < 2) { + await upstream.body?.cancel(); + await new Promise(resolve => setTimeout(resolve, 500)); + continue; + } + if (!upstream.ok) return res.status(upstream.status).send(`Failed to fetch stream: ${upstream.statusText}`); + + const type = upstream.headers.get('content-type') || 'application/octet-stream'; + const length = upstream.headers.get('content-length'); + const encoding = upstream.headers.get('content-encoding'); + const contentRange = upstream.headers.get('content-range'); + const accepts = upstream.headers.get('accept-ranges'); + res.set('Access-Control-Allow-Origin', '*'); + res.status(upstream.status); + if (length && (!encoding || encoding === 'identity')) res.set('Content-Length', length); + if (contentRange) res.set('Content-Range', contentRange); + if (range && !(upstream.status === 206 && contentRange)) res.set('Accept-Ranges', 'none'); + else if (accepts) res.set('Accept-Ranges', accepts); + + if (!upstream.body) return res.end(); + const iterator = upstream.body[Symbol.asyncIterator](); + const prefix = []; + let size = 0; + // The HLS signature can be split across network chunks. + while (size < 7) { + const next = await iterator.next(); + if (next.done) break; + const chunk = Buffer.from(next.value); + prefix.push(chunk); + size += chunk.length; + } + const first = Buffer.concat(prefix, size); + const isManifest = first.subarray(0, 7).toString() === '#EXTM3U' + || /(?:mpegurl|m3u8)/i.test(type); + if (isManifest) { + const manifest = await collectManifest(first, iterator); + res.removeHeader('Content-Length'); + res.removeHeader('Content-Range'); + res.set('Content-Type', 'application/vnd.apple.mpegurl'); + return res.send(rewriteManifest(manifest, upstream.url || url, `${req.baseUrl}/stream`)); + } + res.set('Content-Type', type); + await pipeline(Readable.from(prepend(first, iterator), { objectMode: false }), res); + return; + } catch (error) { + if (res.destroyed || controller.signal.aborted) return; + lastError = error; + if (res.headersSent) { res.destroy(error); return; } + res.removeHeader('Content-Length'); + res.removeHeader('Content-Range'); + if (error.statusCode) return res.status(error.statusCode).json({ error: error.message }); + if (attempt < 2) await new Promise(resolve => setTimeout(resolve, 500)); + } finally { + res.off('close', abort); + controller.abort(); + } + } + if (!res.headersSent && !res.destroyed) res.status(502).json({ error: lastError?.message || 'Stream proxy failed' }); +} + +module.exports = proxyStream; diff --git a/test/proxy-stream.test.js b/test/proxy-stream.test.js new file mode 100644 index 0000000..9098d4c --- /dev/null +++ b/test/proxy-stream.test.js @@ -0,0 +1,121 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const http = require('node:http'); +const express = require('express'); +const proxyStream = require('../server/services/proxyStream'); + +async function server(t, handler) { + const instance = http.createServer(handler); + await new Promise(resolve => instance.listen(0, '127.0.0.1', resolve)); + t.after(() => new Promise(resolve => { + instance.close(resolve); + instance.closeAllConnections(); + })); + return `http://127.0.0.1:${instance.address().port}`; +} + +async function fixture(t, upstreamHandler) { + const upstream = await server(t, upstreamHandler); + const app = express(), router = express.Router(); + router.get('/stream', proxyStream); + app.use('/api/proxy', router); + const base = await server(t, app); + return { upstream, base, proxy: route => `${base}/api/proxy/stream?url=${encodeURIComponent(upstream + route)}` }; +} + +test('binary proxy delivers the first chunk before upstream finishes', async t => { + let finish; + const { proxy } = await fixture(t, (req, res) => { + res.writeHead(200, { 'Content-Type': 'video/mp4' }); + res.write('first-media-chunk'); + finish = () => res.end('last-media-chunk'); + }); + const response = await fetch(proxy('/video'), { signal: AbortSignal.timeout(3000) }); + const reader = response.body.getReader(); + assert.equal(Buffer.from((await reader.read()).value).toString(), 'first-media-chunk'); + finish(); + assert.equal(Buffer.from((await reader.read()).value).toString(), 'last-media-chunk'); + assert.equal((await reader.read()).done, true); +}); + +test('proxy forwards ranges and reports when the origin ignores them', async t => { + const { proxy } = await fixture(t, (req, res) => { + assert.equal(req.headers.range, 'bytes=2-5'); + if (req.url === '/honored') { + res.writeHead(206, { 'Content-Length': 4, 'Content-Range': 'bytes 2-5/10', 'Accept-Ranges': 'bytes' }); + res.end('2345'); + } else { res.writeHead(200, { 'Content-Length': 10 }); res.end('0123456789'); } + }); + const honored = await fetch(proxy('/honored'), { headers: { Range: 'bytes=2-5' } }); + assert.equal(honored.status, 206); + assert.equal(honored.headers.get('content-range'), 'bytes 2-5/10'); + assert.equal(await honored.text(), '2345'); + const ignored = await fetch(proxy('/ignored'), { headers: { Range: 'bytes=2-5' } }); + assert.equal(ignored.status, 200); + assert.equal(ignored.headers.get('accept-ranges'), 'none'); + assert.equal(await ignored.text(), '0123456789'); +}); + +test('HLS rewrite handles split signatures, encryption keys, and the final redirected URL', async t => { + const { proxy, upstream } = await fixture(t, (req, res) => { + if (req.url === '/start') { res.writeHead(302, { Location: '/nested/guide.m3u8' }); return res.end(); } + res.writeHead(200, { 'Content-Type': 'application/octet-stream' }); + res.write('#EXT'); + setImmediate(() => res.end('M3U\n#EXT-X-KEY:METHOD=AES-128,URI="key.bin"\nsegment.ts\n')); + }); + const response = await fetch(proxy('/start')); + const manifest = await response.text(); + assert.equal(response.headers.get('content-type'), 'application/vnd.apple.mpegurl; charset=utf-8'); + assert.ok(manifest.includes(`/api/proxy/stream?url=${encodeURIComponent(upstream + '/nested/key.bin')}`)); + assert.ok(manifest.includes(`/api/proxy/stream?url=${encodeURIComponent(upstream + '/nested/segment.ts')}`)); + assert.ok(!manifest.includes('http://localhost')); +}); + +test('oversized HLS manifests are rejected with a bounded buffer', async t => { + const { proxy } = await fixture(t, (req, res) => { + res.writeHead(200, { 'Content-Type': 'application/vnd.apple.mpegurl' }); + res.end('#EXTM3U\n' + '#'.repeat(5 * 1024 * 1024)); + }); + const response = await fetch(proxy('/oversized')); + assert.equal(response.status, 502); + assert.match((await response.json()).error, /size limit/); +}); + +test('client cancellation closes upstream and upstream disconnects do not crash the process', async t => { + let upstreamClosed; + const closed = new Promise(resolve => { upstreamClosed = resolve; }); + const { proxy } = await fixture(t, (req, res) => { + if (req.url === '/healthy') return res.end('healthy-media'); + res.writeHead(200, { 'Content-Type': 'video/mp4' }); + res.write('first-media-chunk'); + if (req.url === '/disconnect') setTimeout(() => res.destroy(), 25); + else res.once('close', upstreamClosed); + }); + const response = await fetch(proxy('/cancel')); + const reader = response.body.getReader(); + await reader.read(); + await reader.cancel(); + await Promise.race([closed, new Promise((_, reject) => { + const timer = setTimeout(() => reject(new Error('Upstream not cancelled')), 2000); + timer.unref(); + })]); + const broken = await fetch(proxy('/disconnect')); + await assert.rejects(broken.arrayBuffer()); + assert.equal(await (await fetch(proxy('/healthy'))).text(), 'healthy-media'); +}); + +test('one transient upstream error is retried; malformed query values return 400', async t => { + let calls = 0; + const { proxy, base } = await fixture(t, (req, res) => { + calls++; + if (calls === 1) { res.writeHead(503); return res.end('retry'); } + res.end('recovered-media'); + }); + assert.equal(await (await fetch(proxy('/retry'))).text(), 'recovered-media'); + assert.equal(calls, 2); + for (const query of ['url=file%3A%2F%2F%2Fetc%2Fpasswd', 'url[x]=http%3A%2F%2Fexample.invalid', + 'url=http%3A%2F%2Fexample.invalid&url=http%3A%2F%2Fexample.invalid']) { + assert.equal((await fetch(`${base}/api/proxy/stream?${query}`)).status, 400); + } + assert.equal(calls, 2); +});