openvibers/openvibe.ai Commits

The loopback media reader rewrites HLS playlists, so Media's virtual clips (a playlist, no file) are transcribed instead of failing as unreadable (#140)

OP OpenVibers committed 21 days ago · 1218cacb9e · 3 files, +127 −22

Changes

ddiff --git a/server/fetcher.js b/server/fetcher.js
iindex f8eec62..c66b620 100644
---- a/server/fetcher.js
++++ b/server/fetcher.js
@@@ -181,7 +181,8 @@ function createFetcher(config, { transport = null } = {}) {
/**
* One upstream response for a media reader (server/media-proxy.js): the same URL rules as fetchUrl, redirects
* judged and followed here (the reader never sees them), a Range header passed through, the body left as a stream.
-- * → { res } (an http.IncomingMessage with status 200/206/416), or throws AiError.
++ * → { res, url } (an http.IncomingMessage with status 200/206/416, and the URL it came from after redirects), or
++ * throws AiError.
*/
async function openStream(raw, { method = 'GET', range = null, signal, timeoutMs = m.timeoutMs } = {}) {
let target = judge(raw);
@@@ -214,7 +215,7 @@ function createFetcher(config, { transport = null } = {}) {
err.upstreamStatus = res.statusCode; // the loopback reader passes a 4xx on, so ffmpeg reads "gone", not "down"
throw err;
}
-- return { res };
++ return { res, url: target.url.toString() }; // url: where the bytes came from, after redirects
}
throw new AiError(502, 'source.unavailable', 'too many redirects');
}
ddiff --git a/server/media-proxy.js b/server/media-proxy.js
iindex 973f119..0478dd7 100644
---- a/server/media-proxy.js
++++ b/server/media-proxy.js
@@@ -7,6 +7,11 @@
* signal pass streams it once, and each speech-to-text window seeks to its start with a range request. Nothing is
* written to disk.
*
++ * An HLS playlist (Media's virtual clips are one: `/c/<id>` redirects to `/o/<object>/master.m3u8`) is rewritten on
++ * the way through. Every URI in it, resolved against where the playlist really came from, becomes a part of the same
++ * token, `/m/<token>/<n>/<file name>` (the name keeps the extension ffmpeg's HLS reader checks), judged by the same
++ * rules when it is read. Without that, ffmpeg resolved `source/index.m3u8` against the loopback URL and got a 404.
++ *
* const handle = proxy.register(url) // judged now; valid until release() or 6 hours
* handle.url // what ffmpeg reads
* handle.release()
@@@ -18,57 +23,118 @@ const http = require('http');
const crypto = require('crypto');
const TTL_MS = 6 * 3600 * 1000;
++const MAX_PLAYLIST_BYTES = 8 * 1024 * 1024;
++const MAX_PARTS = 200000; // URIs one token's playlists may name (an 8-hour recording in 2 s segments is 14,400)
++const PLAYLIST_TYPE = /^\s*(application\/(vnd\.apple\.mpegurl|x-mpegurl)|audio\/(x-)?mpegurl)\b/i;
++
++function isPlaylist(contentType, url) {
++ if (PLAYLIST_TYPE.test(String(contentType || ''))) return true;
++ try { return /\.m3u8?$/i.test(new URL(url).pathname); } catch { return false; }
++}
++
++/** Each URI line and URI="…" attribute of an HLS playlist, resolved against `base` and replaced by `local(url)`. */
++function rewritePlaylist(text, base, local) {
++ return text.split('\n').map((line) => {
++ const t = line.trim();
++ if (!t) return line;
++ if (t.startsWith('#')) return line.replace(/URI="([^"]*)"/g, (all, u) => `URI="${local(new URL(u, base).toString())}"`);
++ return local(new URL(t, base).toString());
++ }).join('\n');
++}
function createMediaProxy({ fetcher, log = console, clock = { now: () => Date.now() } }) {
-- const entries = new Map(); // token -> { url, expires }
++ const entries = new Map(); // token -> { url, expires, parts: [url…], index: Map(url -> n) }
let server = null;
let ready = null;
++ let port = null;
++
++ /** The local URL of `url` as part of token `token`'s entry: judged now, numbered once per URL. As with a redirect,
++ * a playlist from a public host may not name the internal Media origin. */
++ function partUrl(token, e, url, fromInternal) {
++ if (fetcher.judge(url).internal && !fromInternal) {
++ throw Object.assign(new Error('a playlist from a public host may not name the internal Media origin'), { code: 'fetch.refused' });
++ }
++ let n = e.index.get(url);
++ if (n == null) {
++ if (e.parts.length >= MAX_PARTS) throw Object.assign(new Error(`a playlist names more than ${MAX_PARTS} URIs`), { code: 'playlist.too_large' });
++ n = e.parts.push(url) - 1;
++ e.index.set(url, n);
++ }
++ const name = (new URL(url).pathname.split('/').pop() || 'part').replace(/[^A-Za-z0-9._-]/g, '_').slice(0, 80);
++ return `http://127.0.0.1:${port}/m/${token}/${n}/${name}`;
++ }
++
++ function fail(req, res, ac, err) {
++ if (ac.signal.aborted) return;
++ log.warn && log.warn(`[media-proxy] ${err.code || 'error'}: ${String(err.detail || err.message).slice(0, 200)}`);
++ // Media's 4xx (the recording gone, not allowed) is passed on as it was: ffmpeg then says the media cannot be
++ // read, which is the caller's media and not whisper's health. Everything else (Media down) stays a 502.
++ const gone = err.upstreamStatus >= 400 && err.upstreamStatus < 500 ? err.upstreamStatus : null;
++ if (!res.headersSent) res.writeHead(gone || (err.status === 422 || err.code === 'fetch.refused' ? 403 : 502)).end();
++ else res.destroy();
++ }
++
++ /** Read a playlist (bounded), rewrite it to parts of this token and send it whole. */
++ function sendPlaylist(req, res, ac, token, e, up, from) {
++ if (req.method === 'HEAD') { up.resume(); res.writeHead(200, { 'content-type': 'application/vnd.apple.mpegurl' }).end(); return; }
++ const chunks = []; let size = 0;
++ up.on('data', (c) => {
++ size += c.length;
++ if (size > MAX_PLAYLIST_BYTES) { up.destroy(); fail(req, res, ac, Object.assign(new Error('playlist larger than 8 MB'), { code: 'playlist.too_large' })); return; }
++ chunks.push(c);
++ });
++ up.on('error', (err) => fail(req, res, ac, err));
++ up.on('end', () => {
++ if (res.headersSent || size > MAX_PLAYLIST_BYTES) return;
++ let fromInternal = false;
++ try { fromInternal = fetcher.judge(from).internal; } catch { /* object storage: public */ }
++ let body;
++ try { body = Buffer.from(rewritePlaylist(Buffer.concat(chunks).toString('utf8'), from, (u) => partUrl(token, e, u, fromInternal))); } catch (err) { fail(req, res, ac, err); return; }
++ res.writeHead(200, { 'content-type': 'application/vnd.apple.mpegurl', 'content-length': body.length });
++ res.end(body);
++ });
++ }
function start() {
if (ready) return ready;
server = http.createServer((req, res) => {
-- const m = /^\/m\/([0-9a-f]{32})$/.exec(String(req.url || ''));
++ const m = /^\/m\/([0-9a-f]{32})(?:\/(\d{1,6})(?:\/[A-Za-z0-9._-]*)?)?$/.exec(String(req.url || ''));
const e = m && entries.get(m[1]);
-- if (!e || e.expires < clock.now()) { res.writeHead(404).end(); return; }
++ const target = e && (m[2] == null ? e.url : e.parts[Number(m[2])]);
++ if (!e || e.expires < clock.now() || !target) { res.writeHead(404).end(); return; }
if (req.method !== 'GET' && req.method !== 'HEAD') { res.writeHead(405).end(); return; }
const range = typeof req.headers.range === 'string' && /^bytes=\d*-\d*$/.test(req.headers.range) ? req.headers.range : null;
const ac = new AbortController();
req.on('close', () => ac.abort());
-- fetcher.openStream(e.url, { method: req.method, range, signal: ac.signal, timeoutMs: 60000 }).then(({ res: up }) => {
++ fetcher.openStream(target, { method: req.method, range, signal: ac.signal, timeoutMs: 60000 }).then(({ res: up, url: from }) => {
++ // A playlist is small and rewritten whole; a range into one (only ever bytes=0- from ffmpeg) gets it all.
++ if (isPlaylist(up.headers['content-type'], from || target)) return sendPlaylist(req, res, ac, m[1], e, up, from || target);
const headers = {};
for (const k of ['content-type', 'content-length', 'content-range', 'accept-ranges', 'last-modified', 'etag']) if (up.headers[k]) headers[k] = up.headers[k];
res.writeHead(up.statusCode, headers);
if (req.method === 'HEAD') { up.resume(); res.end(); return; }
up.pipe(res);
up.on('error', () => res.destroy());
-- }, (err) => {
-- if (ac.signal.aborted) return;
-- log.warn && log.warn(`[media-proxy] ${err.code || 'error'}: ${String(err.detail || err.message).slice(0, 200)}`);
-- // Media's 4xx (the recording gone, not allowed) is passed on as it was: ffmpeg then says the media cannot be
-- // read, which is the caller's media and not whisper's health. Everything else (Media down) stays a 502.
-- const gone = err.upstreamStatus >= 400 && err.upstreamStatus < 500 ? err.upstreamStatus : null;
-- if (!res.headersSent) res.writeHead(gone || (err.status === 422 ? 403 : 502)).end();
-- else res.destroy();
-- });
++ }, (err) => fail(req, res, ac, err));
});
server.keepAliveTimeout = 5000;
-- ready = new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', () => { server.unref(); resolve(server.address().port); }); });
++ ready = new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', () => { server.unref(); port = server.address().port; resolve(port); }); });
return ready;
}
/** Judge `url` now and return a local URL for it. */
async function register(url) {
fetcher.judge(url);
-- const port = await start();
++ await start();
const token = crypto.randomBytes(16).toString('hex');
-- entries.set(token, { url, expires: clock.now() + TTL_MS });
++ entries.set(token, { url, expires: clock.now() + TTL_MS, parts: [], index: new Map() });
for (const [k, v] of entries) if (v.expires < clock.now()) entries.delete(k);
return { url: `http://127.0.0.1:${port}/m/${token}`, release: () => entries.delete(token) };
}
-- function close() { entries.clear(); if (server) server.close(); server = null; ready = null; }
++ function close() { entries.clear(); if (server) server.close(); server = null; ready = null; port = null; }
return { register, close, _entries: entries };
}
--module.exports = { createMediaProxy };
++module.exports = { createMediaProxy, rewritePlaylist };
ddiff --git a/test/media-proxy.test.js b/test/media-proxy.test.js
iindex 3f1b333..91e4dca 100644
---- a/test/media-proxy.test.js
++++ b/test/media-proxy.test.js
@@@ -2,21 +2,31 @@
// The loopback reader media.analyze streams through (server/media-proxy.js; roadmap WS-O task 2): ffmpeg reads a
// registered token and gets the real URL's bytes, Range requests pass through (206 with Content-Range), HEAD answers
// headers only, an unknown or released token is 404, and a redirect the fetcher's rules refuse (a public host
--// sending it to the internal Media origin) never reaches the reader: 403.
++// sending it to the internal Media origin) never reaches the reader: 403. An HLS playlist (Media's virtual clips) is
++// rewritten so each URI it names is read through the same token.
const assert = require('assert');
const nodeHttp = require('http');
const { suite } = require('./helpers');
const { load } = require('../server/config');
const { createFetcher } = require('../server/fetcher');
--const { createMediaProxy } = require('../server/media-proxy');
++const { createMediaProxy, rewritePlaylist } = require('../server/media-proxy');
const t = suite('media-proxy');
const BODY = Buffer.from('0123456789abcdefghijklmnopqrstuvwxyz');
let upstream; let base; let proxy;
++const HLS = {
++ '/o/med_1/master.m3u8': '#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1000\nsource/index.m3u8\n',
++ '/o/med_1/source/index.m3u8': '#EXTM3U\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-MAP:URI="init.mp4"\n#EXTINF:2.000,\n000001.m4s\n#EXTINF:2.000,\n000002.m4s?t=1\n#EXT-X-ENDLIST\n',
++ '/o/med_1/source/init.mp4': 'INIT', '/o/med_1/source/000001.m4s': 'SEG1', '/o/med_1/source/000002.m4s?t=1': 'SEG2',
++ '/o/med_2/index.m3u8': '#EXTM3U\n#EXTINF:2.000,\nhttps://evil.example/1.ts\n',
++};
t.test('boot an internal Media stand-in', async () => {
upstream = nodeHttp.createServer((req, res) => {
if (req.url === '/redirect-out') { res.writeHead(302, { Location: 'https://evil.example/x' }); return res.end(); }
++ // A virtual clip: /c/9 redirects to its master playlist, which names a media playlist relative to itself.
++ if (req.url === '/c/9') { res.writeHead(302, { Location: '/o/med_1/master.m3u8' }); return res.end(); }
++ if (HLS[req.url]) { res.writeHead(200, { 'Content-Type': req.url.endsWith('.m3u8') ? 'application/vnd.apple.mpegurl; charset=utf-8' : 'video/mp4' }); return res.end(HLS[req.url]); }
const m = /^bytes=(\d+)-(\d*)$/.exec(req.headers.range || '');
if (m) {
const start = Number(m[1]); const end = m[2] ? Number(m[2]) : BODY.length - 1;
@@@ -46,6 +56,34 @@ t.test('whole reads, ranges, HEAD and unknown tokens', async () => {
assert.strictEqual((await fetch(h.url)).status, 404, 'released');
});
++t.test('an HLS playlist is rewritten: every URI, resolved where the playlist came from, is a part of the same token', async () => {
++ const h = await proxy.register(`${base}/c/9`);
++ let r = await fetch(h.url);
++ assert.strictEqual(r.status, 200);
++ assert.match(r.headers.get('content-type'), /mpegurl/);
++ const master = await r.text();
++ const media = master.split('\n').find((l) => l.startsWith('http'));
++ assert.strictEqual(media, `${h.url}/0/index.m3u8`, 'source/index.m3u8 resolved against the redirect target, not the loopback URL');
++ const list = await (await fetch(media)).text();
++ assert.ok(list.includes(`#EXT-X-MAP:URI="${h.url}/1/init.mp4"`), list);
++ const segs = list.split('\n').filter((l) => l.startsWith('http'));
++ assert.deepStrictEqual(segs, [`${h.url}/2/000001.m4s`, `${h.url}/3/000002.m4s`], 'the file name keeps the extension ffmpeg checks');
++ assert.deepStrictEqual(await Promise.all([`${h.url}/1/init.mp4`, ...segs].map(async (u) => (await fetch(u)).text())), ['INIT', 'SEG1', 'SEG2']);
++ // Read again, the same URIs keep their numbers.
++ assert.strictEqual(await (await fetch(media)).text(), list);
++ assert.strictEqual((await fetch(`${h.url}/99/x.m4s`)).status, 404, 'a part no playlist named');
++ // A playlist naming a host the rules refuse is refused whole.
++ const bad = await proxy.register(`${base}/o/med_2/index.m3u8`);
++ assert.strictEqual((await fetch(bad.url)).status, 403);
++ h.release(); bad.release();
++ assert.strictEqual((await fetch(segs[0])).status, 404, 'parts go with their token');
++});
++
++t.test('rewritePlaylist: URI lines and URI attributes, relative and absolute; tags and blank lines kept', () => {
++ const out = rewritePlaylist('#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI="../k.bin"\n\na.ts\nhttps://cdn.example/b.ts\n', 'https://m.example/x/y/list.m3u8', (u) => `<${u}>`);
++ assert.strictEqual(out, '#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI="<https://m.example/x/k.bin>"\n\n<https://m.example/x/y/a.ts>\n<https://cdn.example/b.ts>\n');
++});
++
t.test('a URL the rules refuse is refused at registration; a refused redirect never reaches the reader', async () => {
await assert.rejects(proxy.register('http://10.0.0.1/v/1'), (e) => e.code === 'fetch.refused');
await assert.rejects(proxy.register('https://evil.example/v/1'), (e) => e.code === 'fetch.refused');