From f6993ae50761790025af4eb9ad4323ba2b57f368 Mon Sep 17 00:00:00 2001 From: sudacode Date: Sun, 16 Aug 2026 01:46:16 -0700 Subject: [PATCH] fix(anime): harden stream proxy and session lifecycle - Reject non-origin-form proxy targets - Release stale anime browser sessions when senders move --- changes/anime-browser.md | 2 +- docs/architecture/domains.md | 7 +- src/anime-bridge/stream-strip-proxy.test.ts | 45 ++ src/anime-bridge/stream-strip-proxy.ts | 409 +++--------------- src/anime-bridge/stream-strip-response.ts | 212 +++++++++ src/anime-bridge/stream-strip-transport.ts | 113 +++++ .../anime-browser-application-runtime.ts | 15 +- .../runtime/anime-browser-sessions.test.ts | 28 ++ src/main/runtime/anime-browser-sessions.ts | 34 ++ 9 files changed, 495 insertions(+), 370 deletions(-) create mode 100644 src/anime-bridge/stream-strip-response.ts create mode 100644 src/anime-bridge/stream-strip-transport.ts create mode 100644 src/main/runtime/anime-browser-sessions.test.ts create mode 100644 src/main/runtime/anime-browser-sessions.ts diff --git a/changes/anime-browser.md b/changes/anime-browser.md index 278853ba..d3588ace 100644 --- a/changes/anime-browser.md +++ b/changes/anime-browser.md @@ -14,7 +14,7 @@ area: anime - Anime playback targets Japanese audio: dub-labelled entries are skipped when the source offers an alternative, `alang` prefers Japanese, and the source's own audio and subtitle tracks are loaded into mpv (Japanese selected) instead of being discarded, so all of them can be switched from mpv's track menu. - The primary subtitle slot stays reserved for Japanese: a source that only has, say, English subtitles gets them added with a normalized language tag (`English` → `en`) but not selected, so the regular `secondarySub` auto-load can route them to the secondary slot instead. - HLS streams pass through a local strip proxy that removes fake image headers some hosts glue onto their video segments and gives disguised segment URLs (`.image`, `.jpg`, `.css`, and other rotating fake extensions) a media-safe local alias, so those streams play in mpv and support Anki audio and image extraction with current ffmpeg releases. -- The strip proxy retries a failed segment fetch once after a short pause and logs upstream error statuses; a host that errors on the very first fetches right after an episode resolves no longer kills the whole playback, and disconnecting clients release their active upstream fetch instead of consuming the socket and bandwidth in the background. +- The strip proxy retries a failed segment fetch once after a short pause and logs upstream error statuses; a host that errors on the very first fetches right after an episode resolves no longer kills the whole playback, disconnecting clients release their active upstream fetch instead of consuming the socket and bandwidth in the background, and only origin-form requests can reach the configured local bridge. - The strip proxy no longer forwards `Range` headers to the bridge: ffmpeg opens every HLS segment with `Range: bytes=0-`, the bridge answers some of those with 206, and a partial response bypassed the disguise strip, so whether an episode played depended on the bridge's cache state. - A bridge that dies out from under the app (killed, crashed, or stopped mid-operation) no longer leaves the browser failing every request until an app restart: the exit is detected, surfaced in the status bar, and the bridge restarts on the next request. - "Playing" is only reported once mpv actually configures a video output; when a stream fails to decode, the browser shows mpv's error instead of claiming playback started while no window ever appeared. diff --git a/docs/architecture/domains.md b/docs/architecture/domains.md index acf38f77..b4ef5d9a 100644 --- a/docs/architecture/domains.md +++ b/docs/architecture/domains.md @@ -30,9 +30,12 @@ Read when: you need to find the owner module for a behavior or test surface - AniList tracking + character dictionary: `src/core/services/anilist/`, `src/main/runtime/composers/anilist-*`, `src/main/character-dictionary-runtime.ts`, `src/main/character-dictionary-runtime/` - Jellyfin integration: `src/core/services/jellyfin*.ts`, `src/main/runtime/composers/jellyfin-*` - Anime browser: extension bridge client, sidecar, and stream handling in `src/anime-bridge/`; + the loopback stream proxy separates request transport/retry from response transformation in + `stream-strip-transport.ts` and `stream-strip-response.ts`; browser window UI in `src/animeui/` (preload `src/preload-animeui.ts`); runtime wiring in - `src/main/runtime/anime-browser-runtime.ts`, `src/main/runtime/anime-browser-ipc-handlers.ts`, - `src/main/runtime/anime-bridge-installer.ts`, `src/main/runtime/stream-playback-metadata.ts`. + `src/main/runtime/anime-browser-application-runtime.ts`, `src/main/runtime/anime-browser-runtime.ts`, + `src/main/runtime/anime-browser-ipc-handlers.ts`, `src/main/runtime/anime-browser-sessions.ts`, + `src/main/runtime/anime-bridge-installer.ts`, and `src/main/runtime/stream-playback-metadata.ts`. The play queue resolves episodes on click and appends them to mpv's real playlist (`src/main/runtime/anime-browser-queue.ts`), then observes media-path changes to attach prepared external tracks and update the browser queue state. diff --git a/src/anime-bridge/stream-strip-proxy.test.ts b/src/anime-bridge/stream-strip-proxy.test.ts index 01e0a8d2..5ddbaf24 100644 --- a/src/anime-bridge/stream-strip-proxy.test.ts +++ b/src/anime-bridge/stream-strip-proxy.test.ts @@ -1,6 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import http from 'node:http'; +import net from 'node:net'; import type { AddressInfo } from 'node:net'; import { findTsSyncOffset, @@ -149,6 +150,25 @@ async function fetchBytes(url: string): Promise<{ status: number; body: Buffer } return { status: response.status, body: Buffer.from(await response.arrayBuffer()) }; } +async function requestRawTarget(proxyOrigin: string, target: string): Promise { + const proxy = new URL(proxyOrigin); + return await new Promise((resolve, reject) => { + const socket = net.createConnection(Number(proxy.port), proxy.hostname, () => { + socket.write(`GET ${target} HTTP/1.1\r\nHost: ${proxy.host}\r\nConnection: close\r\n\r\n`); + }); + let response = ''; + socket.setEncoding('utf8'); + socket.on('data', (chunk) => { + response += chunk; + const status = /^HTTP\/1\.1 (\d{3})/.exec(response)?.[1]; + if (!status) return; + socket.destroy(); + resolve(Number(status)); + }); + socket.once('error', reject); + }); +} + test('proxy strips the PNG disguise off a segment', async () => { const ts = makeTsPackets(8); const disguised = Buffer.concat([PNG_HEADER, ts]); @@ -184,6 +204,31 @@ test('proxy leaves non-TS bodies alone', async () => { ); }); +test('proxy accepts only origin-form targets for its configured HTTP upstream', async () => { + let upstreamHits = 0; + await withProxy( + (_req, res) => { + upstreamHits += 1; + res.writeHead(200, { 'content-type': 'text/plain' }).end('ok'); + }, + async (proxyOrigin, upstreamOrigin) => { + const upstream = new URL(upstreamOrigin); + const rejectedTargets = [ + `${upstreamOrigin}/absolute.ts`, + `//${upstream.host}/scheme-relative.ts`, + 'http://127.0.0.1:9/alternate-host.ts', + `https://${upstream.host}/https.ts`, + ]; + for (const target of rejectedTargets) { + assert.equal(await requestRawTarget(proxyOrigin, target), 502, target); + } + assert.equal(upstreamHits, 0); + assert.equal(await requestRawTarget(proxyOrigin, '/valid.ts?token=one'), 200); + assert.equal(upstreamHits, 1); + }, + ); +}); + test('proxy rewrites absolute upstream playlist entries to its own origin', async () => { await withProxy( (req, res) => { diff --git a/src/anime-bridge/stream-strip-proxy.ts b/src/anime-bridge/stream-strip-proxy.ts index 56dbb36f..59a92596 100644 --- a/src/anime-bridge/stream-strip-proxy.ts +++ b/src/anime-bridge/stream-strip-proxy.ts @@ -1,148 +1,21 @@ import http from 'node:http'; import type { AddressInfo } from 'node:net'; +import { handleUpstreamResponse, TS_SEGMENT_ALIAS_SUFFIX } from './stream-strip-response'; +import { forwardableRequestHeaders, requestUpstream } from './stream-strip-transport'; + +export { + DEFAULT_SCAN_LIMIT_BYTES, + findTsSyncOffset, + rewritePlaylistOrigins, + TS_PACKET_LENGTH, + TS_SEGMENT_ALIAS_SUFFIX, +} from './stream-strip-response'; /** - * Loopback proxy between mpv and the anime bridge that undoes segment - * disguises. Some hosts prepend a real image header (a 1x1 PNG in the wild) to - * every HLS segment so scrapers see "an image"; ffmpeg then probes the segment - * as a picture and playback dies with "no audio or video data played". Aniyomi - * strips this in its player; mpv needs the bytes fixed before it sees them. - * - * Only bridge-origin `.m3u8` streams are routed through here (see - * anime-browser-runtime). Playlist bodies get their absolute upstream origins - * rewritten so segment requests come back through the proxy; segment bodies are - * scanned for the first genuine MPEG-TS packet run and any junk before it is - * dropped. Anything that is not TS (fMP4, VTT, keys) passes through untouched. + * Loopback proxy between mpv and the anime bridge that removes disguised HLS + * segment prefixes before ffmpeg sees them. */ -export const TS_PACKET_LENGTH = 188; -const TS_SYNC_BYTE = 0x47; -/** - * Sync bytes that must repeat at exact packet spacing before an offset counts - * as TS data. One or two matches happen by chance in binary data; five in a - * row at 188-byte strides do not. - */ -const SYNC_RUN = 5; -/** - * FFmpeg 8.1 rejects HLS media whose URL suffix is not in its segment allowlist. - * Hosts disguise MPEG-TS segments behind rotating fake extensions (`.image`, - * `.jpg`, `.css`, ...), so the local playlist gives every proxied segment - * without a recognized media extension this safe alias and removes it again - * before forwarding. - */ -export const TS_SEGMENT_ALIAS_SUFFIX = '.subminer.ts'; -/** ffmpeg 8.1 hls demuxer `allowed_segment_extensions` defaults (minus `html`, - * which only newer builds accept and is a disguise whenever it shows up here). */ -const FFMPEG_SAFE_SEGMENT_EXTENSIONS = new Set([ - '3gp', - 'aac', - 'avi', - 'ac3', - 'eac3', - 'flac', - 'mkv', - 'm3u8', - 'm4a', - 'm4s', - 'm4v', - 'mpg', - 'mov', - 'mp2', - 'mp3', - 'mp4', - 'mpeg', - 'mpegts', - 'ogg', - 'ogv', - 'oga', - 'ts', - 'vob', - 'vtt', - 'wav', - 'webvtt', - 'cmfv', - 'cmfa', - 'ec3', - 'fmp4', -]); - -/** True when ffmpeg's picky segment-extension check would reject this path. */ -function needsTsSegmentAlias(pathname: string): boolean { - const name = pathname.slice(pathname.lastIndexOf('/') + 1).toLowerCase(); - const dot = name.lastIndexOf('.'); - if (dot === -1) return true; - return !FFMPEG_SAFE_SEGMENT_EXTENSIONS.has(name.slice(dot + 1)); -} -/** A disguise prefix is small; give up scanning after this much. */ -export const DEFAULT_SCAN_LIMIT_BYTES = 1024 * 1024; -/** Bytes needed to either find a run within the limit or rule one out. */ -const DECISION_BYTES = DEFAULT_SCAN_LIMIT_BYTES + (SYNC_RUN - 1) * TS_PACKET_LENGTH + 1; - -/** - * First offset at which a confirmed MPEG-TS packet run starts, or null when - * the data does not look like TS at all (within the scan limit). - */ -export function findTsSyncOffset( - data: Buffer, - scanLimit = DEFAULT_SCAN_LIMIT_BYTES, -): number | null { - const lastConfirmable = data.length - (SYNC_RUN - 1) * TS_PACKET_LENGTH - 1; - const end = Math.min(lastConfirmable, scanLimit); - for (let offset = 0; offset <= end; offset++) { - if (data[offset] !== TS_SYNC_BYTE) continue; - let confirmed = true; - for (let packet = 1; packet < SYNC_RUN; packet++) { - if (data[offset + packet * TS_PACKET_LENGTH] !== TS_SYNC_BYTE) { - confirmed = false; - break; - } - } - if (confirmed) return offset; - } - return null; -} - -/** - * Point absolute playlist entries at the proxy. Relative entries already - * resolve against whatever origin served the playlist, so they need no help. - */ -export function rewritePlaylistOrigins( - body: string, - upstreamOrigin: string, - proxyOrigin: string, -): string { - const rebased = body.split(upstreamOrigin).join(proxyOrigin); - return rebased - .split(/(\r?\n)/) - .map((line) => { - const uri = line.trim(); - if (!uri || uri.startsWith('#')) return line; - - let resolved: URL; - try { - resolved = new URL(uri, proxyOrigin); - } catch { - return line; - } - if (resolved.origin !== proxyOrigin || !needsTsSegmentAlias(resolved.pathname)) { - return line; - } - - const queryIndex = uri.search(/[?#]/); - const aliasIndex = queryIndex === -1 ? uri.length : queryIndex; - const leadingWhitespace = line.slice(0, line.indexOf(uri)); - const trailingWhitespace = line.slice(leadingWhitespace.length + uri.length); - return `${leadingWhitespace}${uri.slice(0, aliasIndex)}${TS_SEGMENT_ALIAS_SUFFIX}${uri.slice(aliasIndex)}${trailingWhitespace}`; - }) - .join(''); -} - -function removeTsSegmentAlias(url: URL): void { - if (url.pathname.endsWith(TS_SEGMENT_ALIAS_SUFFIX)) { - url.pathname = url.pathname.slice(0, -TS_SEGMENT_ALIAS_SUFFIX.length); - } -} - export interface StreamStripProxyOptions { /** Read per request so a bridge restart on a new port keeps working. */ upstreamOrigin: () => string; @@ -152,12 +25,6 @@ export interface StreamStripProxyOptions { } const DEFAULT_RETRY_DELAY_MS = 400; -/** - * Socket timeout on the upstream GET, cleared once its headers arrive. Node's - * http client has no deadline of its own, so a host that accepts the - * connection and then says nothing would hang mpv on that segment forever. - */ -const UPSTREAM_TIMEOUT_MS = 15_000; export interface StreamStripProxyHandle { origin: string; @@ -165,26 +32,27 @@ export interface StreamStripProxyHandle { close: () => Promise; } -interface ClientRequestLifecycle { - activeUpstreamRequest: http.ClientRequest | null; - closed: boolean; -} - -/** Response headers that must not be forwarded verbatim. */ -const DROPPED_HEADERS = new Set([ - 'connection', - 'keep-alive', - 'transfer-encoding', - 'content-length', -]); - -function forwardableHeaders(headers: http.IncomingHttpHeaders): http.OutgoingHttpHeaders { - const result: http.OutgoingHttpHeaders = {}; - for (const [name, value] of Object.entries(headers)) { - if (value === undefined || DROPPED_HEADERS.has(name.toLowerCase())) continue; - result[name] = value; +function resolveUpstreamUrl(requestTarget: string, configuredOrigin: string): URL { + if ( + !requestTarget.startsWith('/') || + requestTarget.startsWith('//') || + requestTarget.includes('#') + ) { + throw new Error('Stream proxy requests must use origin-form targets.'); } - return result; + + const upstream = new URL(configuredOrigin); + if (upstream.protocol !== 'http:') { + throw new Error('Stream proxy upstream must use HTTP.'); + } + const resolved = new URL(requestTarget, upstream.origin); + if (resolved.protocol !== 'http:' || resolved.origin !== upstream.origin) { + throw new Error('Stream proxy request escaped the configured upstream origin.'); + } + if (resolved.pathname.endsWith(TS_SEGMENT_ALIAS_SUFFIX)) { + resolved.pathname = resolved.pathname.slice(0, -TS_SEGMENT_ALIAS_SUFFIX.length); + } + return resolved; } export function startStreamStripProxy( @@ -192,6 +60,7 @@ export function startStreamStripProxy( ): Promise { const log = options.log ?? (() => {}); const retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS; + let origin = ''; const server = http.createServer((req, res) => { if (req.method !== 'GET' && req.method !== 'HEAD') { @@ -201,210 +70,36 @@ export function startStreamStripProxy( let upstreamUrl: URL; try { - upstreamUrl = new URL(req.url ?? '/', options.upstreamOrigin()); - removeTsSegmentAlias(upstreamUrl); + upstreamUrl = resolveUpstreamUrl(req.url ?? '/', options.upstreamOrigin()); } catch { res.writeHead(502).end(); return; } - const requestHeaders = forwardableHeaders(req.headers); + const requestHeaders = forwardableRequestHeaders(req.headers); delete requestHeaders.host; - // Never forward Range: ffmpeg opens every segment with `bytes=0-`, the - // bridge answers some of those 206, and a partial response cannot be - // stripped (only full 200 bodies are). Byte ranges into a resource whose - // bytes this proxy rewrites would be incoherent anyway. + // Rewritten bodies cannot honor byte ranges into the original representation. delete requestHeaders.range; res.on('error', () => {}); - const lifecycle: ClientRequestLifecycle = { activeUpstreamRequest: null, closed: false }; - const destroyUpstreamOnClientClose = (): void => { - lifecycle.closed = true; - lifecycle.activeUpstreamRequest?.destroy(); - }; - req.once('aborted', destroyUpstreamOnClientClose); - res.once('close', destroyUpstreamOnClientClose); - res.once('finish', () => { - req.off('aborted', destroyUpstreamOnClientClose); - res.off('close', destroyUpstreamOnClientClose); - lifecycle.activeUpstreamRequest = null; - }); - - requestUpstream(req, res, upstreamUrl, requestHeaders, 0, lifecycle); - }); - - /** - * One delayed retry on a failed GET: right after an episode resolve, the - * bridge (or the host behind it) can error on the very first segment - * fetches and be fine a moment later — mpv treats a playlist full of failed - * segments as a dead file and gives up for good. - */ - function requestUpstream( - req: http.IncomingMessage, - res: http.ServerResponse, - upstreamUrl: URL, - requestHeaders: http.OutgoingHttpHeaders, - attempt: number, - lifecycle: ClientRequestLifecycle, - ): void { - if (lifecycle.closed || res.destroyed) return; - const mayRetry = req.method === 'GET' && attempt === 0; - // Once the response is handed off, its headers (and often part of its body) - // are already on the wire: a later upstream error can only be reported by - // killing the connection, never by retrying or writing a 502. - let handedOff = false; - const retry = (): void => { - setTimeout(() => { - requestUpstream(req, res, upstreamUrl, requestHeaders, attempt + 1, lifecycle); - }, retryDelayMs); - }; - - const upstreamRequest = http.request( + requestUpstream({ + req, + res, upstreamUrl, - { method: req.method, headers: requestHeaders, timeout: UPSTREAM_TIMEOUT_MS }, - (upstream) => { - const clearActiveRequest = (): void => { - if (lifecycle.activeUpstreamRequest === upstreamRequest) { - lifecycle.activeUpstreamRequest = null; - } - }; - upstream.once('end', clearActiveRequest); - upstream.once('close', clearActiveRequest); - // Body streaming has its own pace; only the wait for headers is capped. - upstreamRequest.setTimeout(0); - const status = upstream.statusCode ?? 502; - if (status === 404 || status >= 500) { - if (mayRetry) { - log(`[stream-proxy] upstream ${status} for ${upstreamUrl.pathname}; retrying once`); - upstream.resume(); - retry(); - return; - } - log(`[stream-proxy] upstream ${status} for ${upstreamUrl.pathname}`); - } - handedOff = true; - handleUpstreamResponse(req, res, upstream); - }, - ); - lifecycle.activeUpstreamRequest = upstreamRequest; - // Destroying with an error routes the stall through the retry/502 path. - upstreamRequest.on('timeout', () => { - upstreamRequest.destroy(new Error(`upstream silent for ${UPSTREAM_TIMEOUT_MS}ms`)); + requestHeaders, + retryDelayMs, + log, + handleResponse: (upstream) => + handleUpstreamResponse({ + req, + res, + upstream, + upstreamOrigin: options.upstreamOrigin, + proxyOrigin: () => origin, + log, + }), }); - upstreamRequest.on('error', (error) => { - if (handedOff) { - log(`[stream-proxy] upstream failed mid-response: ${String(error)}`); - res.destroy(); - return; - } - if (mayRetry) { - log(`[stream-proxy] upstream request failed: ${String(error)}; retrying once`); - retry(); - return; - } - log(`[stream-proxy] upstream request failed: ${String(error)}`); - if (!res.headersSent) res.writeHead(502); - res.end(); - }); - upstreamRequest.end(); - } - - function handleUpstreamResponse( - req: http.IncomingMessage, - res: http.ServerResponse, - upstream: http.IncomingMessage, - ): void { - const status = upstream.statusCode ?? 502; - const pathname = (req.url ?? '').split('?', 1)[0] ?? ''; - const contentType = String(upstream.headers['content-type'] ?? ''); - const isPlaylist = pathname.endsWith('.m3u8') || contentType.includes('mpegurl'); - - upstream.on('error', () => res.destroy()); - - // Only a full 200 body is safe to modify; everything else (errors, range - // responses, HEAD) forwards untouched. - if (status !== 200 || req.method === 'HEAD') { - res.writeHead(status, forwardableHeaders(upstream.headers)); - upstream.pipe(res); - return; - } - - if (isPlaylist) { - const chunks: Buffer[] = []; - let buffered = 0; - upstream.on('data', (chunk: Buffer) => { - buffered += chunk.length; - // A playlist is text and small; anything this large is not one, and it - // has to be held whole in memory to be rewritten. - if (buffered > DECISION_BYTES) { - log(`[stream-proxy] playlist body over ${DECISION_BYTES} bytes; dropping`); - upstream.destroy(); - res.destroy(); - return; - } - chunks.push(chunk); - }); - upstream.on('end', () => { - const body = rewritePlaylistOrigins( - Buffer.concat(chunks).toString('utf8'), - options.upstreamOrigin(), - origin, - ); - res.writeHead(status, { - ...forwardableHeaders(upstream.headers), - 'content-length': Buffer.byteLength(body, 'utf8'), - }); - res.end(body); - }); - return; - } - - stripSegment(res, upstream); - } - - /** - * Buffer just enough of the body to find (or rule out) a TS packet run, - * drop everything before it, then stream the rest through untouched. - */ - function stripSegment(res: http.ServerResponse, upstream: http.IncomingMessage): void { - const chunks: Buffer[] = []; - let buffered = 0; - - const respond = (data: Buffer, remainderFollows: boolean): void => { - const offset = findTsSyncOffset(data) ?? 0; - if (offset > 0) log(`[stream-proxy] stripped ${offset} disguise bytes off a segment`); - const body = offset > 0 ? data.subarray(offset) : data; - - const headers = forwardableHeaders(upstream.headers); - const upstreamLength = Number(upstream.headers['content-length']); - if (remainderFollows) { - if (Number.isFinite(upstreamLength)) headers['content-length'] = upstreamLength - offset; - } else { - headers['content-length'] = body.length; - } - - res.writeHead(upstream.statusCode ?? 200, headers); - res.write(body); - }; - - const onData = (chunk: Buffer): void => { - chunks.push(chunk); - buffered += chunk.length; - if (buffered < DECISION_BYTES) return; - upstream.off('data', onData); - upstream.off('end', onEnd); - respond(Buffer.concat(chunks), true); - upstream.pipe(res); - }; - const onEnd = (): void => { - respond(Buffer.concat(chunks), false); - res.end(); - }; - upstream.on('data', onData); - upstream.on('end', onEnd); - } - - let origin = ''; + }); return new Promise((resolve, reject) => { server.once('error', reject); diff --git a/src/anime-bridge/stream-strip-response.ts b/src/anime-bridge/stream-strip-response.ts new file mode 100644 index 00000000..508ac515 --- /dev/null +++ b/src/anime-bridge/stream-strip-response.ts @@ -0,0 +1,212 @@ +import type http from 'node:http'; + +export const TS_PACKET_LENGTH = 188; +const TS_SYNC_BYTE = 0x47; +/** Five sync bytes at exact packet spacing make an accidental match implausible. */ +const SYNC_RUN = 5; + +export const TS_SEGMENT_ALIAS_SUFFIX = '.subminer.ts'; +const FFMPEG_SAFE_SEGMENT_EXTENSIONS = new Set([ + '3gp', + 'aac', + 'avi', + 'ac3', + 'eac3', + 'flac', + 'mkv', + 'm3u8', + 'm4a', + 'm4s', + 'm4v', + 'mpg', + 'mov', + 'mp2', + 'mp3', + 'mp4', + 'mpeg', + 'mpegts', + 'ogg', + 'ogv', + 'oga', + 'ts', + 'vob', + 'vtt', + 'wav', + 'webvtt', + 'cmfv', + 'cmfa', + 'ec3', + 'fmp4', +]); + +function needsTsSegmentAlias(pathname: string): boolean { + const name = pathname.slice(pathname.lastIndexOf('/') + 1).toLowerCase(); + const dot = name.lastIndexOf('.'); + if (dot === -1) return true; + return !FFMPEG_SAFE_SEGMENT_EXTENSIONS.has(name.slice(dot + 1)); +} + +export const DEFAULT_SCAN_LIMIT_BYTES = 1024 * 1024; +const DECISION_BYTES = DEFAULT_SCAN_LIMIT_BYTES + (SYNC_RUN - 1) * TS_PACKET_LENGTH + 1; + +export function findTsSyncOffset( + data: Buffer, + scanLimit = DEFAULT_SCAN_LIMIT_BYTES, +): number | null { + const lastConfirmable = data.length - (SYNC_RUN - 1) * TS_PACKET_LENGTH - 1; + const end = Math.min(lastConfirmable, scanLimit); + for (let offset = 0; offset <= end; offset++) { + if (data[offset] !== TS_SYNC_BYTE) continue; + let confirmed = true; + for (let packet = 1; packet < SYNC_RUN; packet++) { + if (data[offset + packet * TS_PACKET_LENGTH] !== TS_SYNC_BYTE) { + confirmed = false; + break; + } + } + if (confirmed) return offset; + } + return null; +} + +export function rewritePlaylistOrigins( + body: string, + upstreamOrigin: string, + proxyOrigin: string, +): string { + const rebased = body.split(upstreamOrigin).join(proxyOrigin); + return rebased + .split(/(\r?\n)/) + .map((line) => { + const uri = line.trim(); + if (!uri || uri.startsWith('#')) return line; + + let resolved: URL; + try { + resolved = new URL(uri, proxyOrigin); + } catch { + return line; + } + if (resolved.origin !== proxyOrigin || !needsTsSegmentAlias(resolved.pathname)) { + return line; + } + + const queryIndex = uri.search(/[?#]/); + const aliasIndex = queryIndex === -1 ? uri.length : queryIndex; + const leadingWhitespace = line.slice(0, line.indexOf(uri)); + const trailingWhitespace = line.slice(leadingWhitespace.length + uri.length); + return `${leadingWhitespace}${uri.slice(0, aliasIndex)}${TS_SEGMENT_ALIAS_SUFFIX}${uri.slice(aliasIndex)}${trailingWhitespace}`; + }) + .join(''); +} + +const DROPPED_RESPONSE_HEADERS = new Set([ + 'connection', + 'keep-alive', + 'transfer-encoding', + 'content-length', +]); + +function forwardableResponseHeaders(headers: http.IncomingHttpHeaders): http.OutgoingHttpHeaders { + const result: http.OutgoingHttpHeaders = {}; + for (const [name, value] of Object.entries(headers)) { + if (value === undefined || DROPPED_RESPONSE_HEADERS.has(name.toLowerCase())) continue; + result[name] = value; + } + return result; +} + +export function handleUpstreamResponse(options: { + req: http.IncomingMessage; + res: http.ServerResponse; + upstream: http.IncomingMessage; + upstreamOrigin: () => string; + proxyOrigin: () => string; + log: (message: string) => void; +}): void { + const { req, res, upstream, log } = options; + const status = upstream.statusCode ?? 502; + const pathname = (req.url ?? '').split('?', 1)[0] ?? ''; + const contentType = String(upstream.headers['content-type'] ?? ''); + const isPlaylist = pathname.endsWith('.m3u8') || contentType.includes('mpegurl'); + + upstream.on('error', () => res.destroy()); + + if (status !== 200 || req.method === 'HEAD') { + res.writeHead(status, forwardableResponseHeaders(upstream.headers)); + upstream.pipe(res); + return; + } + + if (isPlaylist) { + const chunks: Buffer[] = []; + let buffered = 0; + upstream.on('data', (chunk: Buffer) => { + buffered += chunk.length; + if (buffered > DECISION_BYTES) { + log(`[stream-proxy] playlist body over ${DECISION_BYTES} bytes; dropping`); + upstream.destroy(); + res.destroy(); + return; + } + chunks.push(chunk); + }); + upstream.on('end', () => { + const body = rewritePlaylistOrigins( + Buffer.concat(chunks).toString('utf8'), + options.upstreamOrigin(), + options.proxyOrigin(), + ); + res.writeHead(status, { + ...forwardableResponseHeaders(upstream.headers), + 'content-length': Buffer.byteLength(body, 'utf8'), + }); + res.end(body); + }); + return; + } + + stripSegment(res, upstream, log); +} + +function stripSegment( + res: http.ServerResponse, + upstream: http.IncomingMessage, + log: (message: string) => void, +): void { + const chunks: Buffer[] = []; + let buffered = 0; + + const respond = (data: Buffer, remainderFollows: boolean): void => { + const offset = findTsSyncOffset(data) ?? 0; + if (offset > 0) log(`[stream-proxy] stripped ${offset} disguise bytes off a segment`); + const body = offset > 0 ? data.subarray(offset) : data; + + const headers = forwardableResponseHeaders(upstream.headers); + const upstreamLength = Number(upstream.headers['content-length']); + if (remainderFollows) { + if (Number.isFinite(upstreamLength)) headers['content-length'] = upstreamLength - offset; + } else { + headers['content-length'] = body.length; + } + + res.writeHead(upstream.statusCode ?? 200, headers); + res.write(body); + }; + + const onData = (chunk: Buffer): void => { + chunks.push(chunk); + buffered += chunk.length; + if (buffered < DECISION_BYTES) return; + upstream.off('data', onData); + upstream.off('end', onEnd); + respond(Buffer.concat(chunks), true); + upstream.pipe(res); + }; + const onEnd = (): void => { + respond(Buffer.concat(chunks), false); + res.end(); + }; + upstream.on('data', onData); + upstream.on('end', onEnd); +} diff --git a/src/anime-bridge/stream-strip-transport.ts b/src/anime-bridge/stream-strip-transport.ts new file mode 100644 index 00000000..ca690cd8 --- /dev/null +++ b/src/anime-bridge/stream-strip-transport.ts @@ -0,0 +1,113 @@ +import http from 'node:http'; + +const UPSTREAM_TIMEOUT_MS = 15_000; + +interface ClientRequestLifecycle { + activeUpstreamRequest: http.ClientRequest | null; + closed: boolean; +} + +export function forwardableRequestHeaders( + headers: http.IncomingHttpHeaders, +): http.OutgoingHttpHeaders { + const result: http.OutgoingHttpHeaders = {}; + for (const [name, value] of Object.entries(headers)) { + if ( + value === undefined || + name.toLowerCase() === 'connection' || + name.toLowerCase() === 'keep-alive' || + name.toLowerCase() === 'transfer-encoding' || + name.toLowerCase() === 'content-length' + ) { + continue; + } + result[name] = value; + } + return result; +} + +export function requestUpstream(options: { + req: http.IncomingMessage; + res: http.ServerResponse; + upstreamUrl: URL; + requestHeaders: http.OutgoingHttpHeaders; + retryDelayMs: number; + log: (message: string) => void; + handleResponse: (upstream: http.IncomingMessage) => void; +}): void { + const lifecycle: ClientRequestLifecycle = { activeUpstreamRequest: null, closed: false }; + const destroyUpstreamOnClientClose = (): void => { + lifecycle.closed = true; + lifecycle.activeUpstreamRequest?.destroy(); + }; + options.req.once('aborted', destroyUpstreamOnClientClose); + options.res.once('close', destroyUpstreamOnClientClose); + options.res.once('finish', () => { + options.req.off('aborted', destroyUpstreamOnClientClose); + options.res.off('close', destroyUpstreamOnClientClose); + lifecycle.activeUpstreamRequest = null; + }); + + requestAttempt(options, lifecycle, 0); +} + +function requestAttempt( + options: Parameters[0], + lifecycle: ClientRequestLifecycle, + attempt: number, +): void { + const { req, res, upstreamUrl, requestHeaders, retryDelayMs, log } = options; + if (lifecycle.closed || res.destroyed) return; + const mayRetry = req.method === 'GET' && attempt === 0; + let handedOff = false; + const retry = (): void => { + setTimeout(() => requestAttempt(options, lifecycle, attempt + 1), retryDelayMs); + }; + + const upstreamRequest = http.request( + upstreamUrl, + { method: req.method, headers: requestHeaders, timeout: UPSTREAM_TIMEOUT_MS }, + (upstream) => { + const clearActiveRequest = (): void => { + if (lifecycle.activeUpstreamRequest === upstreamRequest) { + lifecycle.activeUpstreamRequest = null; + } + }; + upstream.once('end', clearActiveRequest); + upstream.once('close', clearActiveRequest); + upstreamRequest.setTimeout(0); + const status = upstream.statusCode ?? 502; + if (status === 404 || status >= 500) { + if (mayRetry) { + log(`[stream-proxy] upstream ${status} for ${upstreamUrl.pathname}; retrying once`); + upstream.resume(); + retry(); + return; + } + log(`[stream-proxy] upstream ${status} for ${upstreamUrl.pathname}`); + } + handedOff = true; + options.handleResponse(upstream); + }, + ); + lifecycle.activeUpstreamRequest = upstreamRequest; + upstreamRequest.on('timeout', () => { + upstreamRequest.destroy(new Error(`upstream silent for ${UPSTREAM_TIMEOUT_MS}ms`)); + }); + upstreamRequest.on('error', (error) => { + if (handedOff) { + log(`[stream-proxy] upstream failed mid-response: ${String(error)}`); + res.destroy(); + return; + } + if (mayRetry) { + log(`[stream-proxy] upstream request failed: ${String(error)}; retrying once`); + retry(); + return; + } + log(`[stream-proxy] upstream request failed: ${String(error)}`); + if (!res.headersSent) res.writeHead(502); + res.end(); + }); + upstreamRequest.end(); +} diff --git a/src/main/runtime/anime-browser-application-runtime.ts b/src/main/runtime/anime-browser-application-runtime.ts index 4a452033..d6dfce9e 100644 --- a/src/main/runtime/anime-browser-application-runtime.ts +++ b/src/main/runtime/anime-browser-application-runtime.ts @@ -9,6 +9,7 @@ import { type AnimeBrowserIpcSender, } from './anime-browser-ipc-handlers'; import { createAnimeBrowserRuntime, type AnimeBrowserRuntimeDeps } from './anime-browser-runtime'; +import { createAnimeBrowserSessionRegistry } from './anime-browser-sessions'; import { createOpenConfigSettingsWindowHandler } from './config-settings-window'; import { createCreateAnimeBrowserWindowHandler } from './setup-window-factory'; import { @@ -68,7 +69,6 @@ export interface AnimeBrowserApplicationRuntime { export function createAnimeBrowserApplicationRuntime( deps: AnimeBrowserApplicationRuntimeDeps, ): AnimeBrowserApplicationRuntime { - const sessions = new Map(); let playbackState: AnimeBrowserPlaybackState | null = toAnimeBrowserPlaybackState( deps.getInitialPlaybackMetadata(), ); @@ -114,20 +114,15 @@ export function createAnimeBrowserApplicationRuntime( }, onQueueState: (state) => broadcast(IPC_CHANNELS.event.animeBrowserQueueState, state), }); + const sessions = createAnimeBrowserSessionRegistry((sessionId) => + runtime.releaseSession(sessionId), + ); registerAnimeBrowserIpcHandlers({ ipcMain: deps.ipcMain, runtime, getPlaybackState: () => playbackState, - registerSession: (sessionId, sender) => { - if (sessions.get(sessionId) === sender) return; - sessions.set(sessionId, sender); - sender.once('destroyed', () => { - if (sessions.get(sessionId) !== sender) return; - sessions.delete(sessionId); - runtime.releaseSession(sessionId); - }); - }, + registerSession: sessions.register, }); let dockIconRetained = false; diff --git a/src/main/runtime/anime-browser-sessions.test.ts b/src/main/runtime/anime-browser-sessions.test.ts new file mode 100644 index 00000000..2a95e339 --- /dev/null +++ b/src/main/runtime/anime-browser-sessions.test.ts @@ -0,0 +1,28 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { AnimeBrowserIpcSender } from './anime-browser-ipc-handlers'; +import { createAnimeBrowserSessionRegistry } from './anime-browser-sessions'; + +test('moving a live sender to a new anime browser session releases its stale session', () => { + const destroyedListeners: Array<() => void> = []; + const sender: AnimeBrowserIpcSender = { + send: () => {}, + isDestroyed: () => false, + once: (_event, listener) => destroyedListeners.push(listener), + }; + const released: string[] = []; + const sessions = createAnimeBrowserSessionRegistry((sessionId) => released.push(sessionId)); + + sessions.register('old', sender); + sessions.register('old', sender); + assert.equal(destroyedListeners.length, 1, 'duplicate registration keeps its existing handler'); + + sessions.register('new', sender); + assert.deepEqual(released, ['old']); + assert.equal(sessions.get('old'), undefined); + assert.equal(sessions.get('new'), sender); + + for (const listener of destroyedListeners) listener(); + assert.deepEqual(released, ['old', 'new']); + assert.equal(sessions.get('new'), undefined); +}); diff --git a/src/main/runtime/anime-browser-sessions.ts b/src/main/runtime/anime-browser-sessions.ts new file mode 100644 index 00000000..7db77c6d --- /dev/null +++ b/src/main/runtime/anime-browser-sessions.ts @@ -0,0 +1,34 @@ +import type { AnimeBrowserIpcSender } from './anime-browser-ipc-handlers'; + +export interface AnimeBrowserSessionRegistry { + get: (sessionId: string) => AnimeBrowserIpcSender | undefined; + register: (sessionId: string, sender: AnimeBrowserIpcSender) => void; + values: () => IterableIterator; +} + +export function createAnimeBrowserSessionRegistry( + releaseSession: (sessionId: string) => void, +): AnimeBrowserSessionRegistry { + const sessions = new Map(); + + const register = (sessionId: string, sender: AnimeBrowserIpcSender): void => { + if (sessions.get(sessionId) === sender) return; + for (const [previousSessionId, registeredSender] of sessions) { + if (previousSessionId === sessionId || registeredSender !== sender) continue; + sessions.delete(previousSessionId); + releaseSession(previousSessionId); + } + sessions.set(sessionId, sender); + sender.once('destroyed', () => { + if (sessions.get(sessionId) !== sender) return; + sessions.delete(sessionId); + releaseSession(sessionId); + }); + }; + + return { + get: (sessionId) => sessions.get(sessionId), + register, + values: () => sessions.values(), + }; +}