openvibers/openvibe.live Commits
Generate OpenRe WHIP test keys at runtime (#13)
Changes
@@@ -5,8 +5,7 @@* 'openre' means OpenRe.Stream ingests this slot's RTMP: Live's RTMP server refuses the slot's* key (so a stream can never be ingested twice), the Go Live UI shows OpenRe's ingest URL and* rotates OpenRe's key, and the slot's sessions reach `streams` through the mirror (mirror.js).-- * Only RTMP moves (OPENRE_PROTOCOLS); WHIP, browser/WebRTC and JSMPEG stay on Live until-- * OpenRe carries them.++ * Only RTMP moves (OPENRE_PROTOCOLS). WHIP and JSMPEG still publish through Live.** Switch off = zero behaviour change: every helper here returns what Live did before unless the* slot says 'openre' AND OpenRe is configured (openre-client enabled()).@@@ -33,9 +32,9 @@ function slotIsOpenre(id) {}/**-- * Live's RTMP publish handler asks this before accepting a key. A slot key is refused when its slot-- * is on OpenRe; a personal key (users.stream_key, no slot) is refused when any of the user's slots is,-- * because Live would attach it to the user's live rows and push the user's destinations twice.++ * Live's publish handlers ask this before accepting a key. Only RTMP is refused when its slot++ * is on OpenRe; an RTMP personal key (users.stream_key, no slot) is refused when any of the user's++ * slots is on OpenRe, because Live would attach it to the user's live rows and push destinations twice.*/function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {try {@@@ -22,6 +22,24 @@ function matchesStreamPath(url, streamKey) {}}++/**++ * Consult the per-protocol ingest authority for each publish. OpenRe currently ingests RTMP++ * only, so JSMPEG continues to publish through Live after an RTMP switch.++ */++function refusedByOpenre(streamKey) {++ try {++ const db = require('../db/database');++ const user = db.getUserByStreamKey(streamKey);++ const managedStream = user ? null : db.getManagedStreamByStreamKey(streamKey);++ if (!user && !managedStream) return false;++ if (!require('../openre/authority').refusesLiveIngest({ managedStream, user, protocol: 'jsmpeg' })) return false;++ console.log(`[JSMPEG] Rejected: ${managedStream ? `slot ${managedStream.id}` : `personal key of ${user.username}`} is ingested by OpenRe`);++ return true;++ } catch {++ return false;++ }++}++class JSMPEGRelay {constructor() {/** @type {Map<string, { videoWss: WebSocket.Server, audioWss: WebSocket.Server, videoServer: http.Server, audioServer: http.Server }>} */@@@ -54,7 +72,7 @@ class JSMPEGRelay {const videoServer = http.createServer((req, res) => {// FFmpeg sends MPEG1 data via HTTP POSTreq.socket.setNoDelay(true);-- if (req.method === 'POST' && matchesStreamPath(req.url, streamKey)) {++ if (req.method === 'POST' && matchesStreamPath(req.url, streamKey) && !refusedByOpenre(streamKey)) {req.on('data', (chunk) => {// Broadcast to WebSocket viewersif (videoWss.clients.size > 0) {@@@ -99,7 +117,7 @@ class JSMPEGRelay {const audioWss = new WebSocket.Server({ noServer: true, perMessageDeflate: false, maxPayload: 64 * 1024 });const audioServer = http.createServer((req, res) => {req.socket.setNoDelay(true);-- if (req.method === 'POST' && matchesStreamPath(req.url, streamKey)) {++ if (req.method === 'POST' && matchesStreamPath(req.url, streamKey) && !refusedByOpenre(streamKey)) {req.on('data', (chunk) => {// Broadcast to WebSocket viewersif (audioWss.clients.size > 0) {@@@ -119,6 +119,18 @@ function sendWhipError(res, status, code, message) {.json({ error: message, error_code: code });}++/**++ * Consult the per-protocol ingest authority before accepting a publisher. OpenRe currently++ * ingests RTMP only, so WHIP continues to publish through Live after an RTMP switch.++ */++function refusedByOpenre(slotId, userId) {++ const managedStream = slotId ? db.getManagedStreamById(slotId) : null;++ const user = managedStream || !userId ? null : db.getUserById(userId);++ if (!require('../openre/authority').refusesLiveIngest({ managedStream, user, protocol: 'webrtc' })) return false;++ console.log(`[WHIP] Rejected: ${managedStream ? `slot ${managedStream.id}` : `personal key of user ${userId}`} is ingested by OpenRe`);++ return true;++}++function logWhipStage(stage, streamId, info = {}) {const details = Object.entries(info).filter(([, value]) => value !== undefined && value !== null)@@@ -670,6 +682,11 @@ async function handleWhipPost(req, res) {return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');}++ if (refusedByOpenre(managedStream.id, managedStream.user_id)) {++ logWhipStage('auth_key_fail', pathParam, { reason: 'stream_key_not_found' });++ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');++ }++userId = managedStream.user_id;logWhipStage('auth_key', pathParam, { user_id: userId, managed_stream_id: managedStream.id });@@@ -736,6 +753,9 @@ async function handleWhipPost(req, res) {logWhipStage('auth_slot_fail', pathParam, { reason: 'not_slot_owner', user_id: user.id, slot_id: slotId });return sendWhipError(res, 403, 'not_slot_owner', 'This stream slot belongs to another account');}++ if (refusedByOpenre(managedStream.id, user.id)) {++ return sendWhipError(res, 401, 'invalid_key', 'Stream key or token does not match');++ }userId = user.id;// Find live session for this slot owned by this userconst liveSessions = db.getLiveStreamsByUserId(user.id) || [];@@@ -758,6 +778,9 @@ async function handleWhipPost(req, res) {}if (!stream) {++ if (refusedByOpenre(managedStream.id, managedStream.user_id)) {++ return sendWhipError(res, 401, 'invalid_key', 'Stream key does not match this slot');++ }userId = managedStream.user_id;logWhipStage('auth_slot_key', pathParam, { user_id: userId, slot_id: slotId });@@@ -813,6 +836,9 @@ async function handleWhipPost(req, res) {}if (!stream.is_live) return sendWhipError(res, 409, 'stream_not_live', 'Stream is not live — go live first');if (stream.protocol !== 'webrtc') return sendWhipError(res, 409, 'wrong_protocol', 'Stream protocol must be webrtc for WHIP');++ if (refusedByOpenre(stream.managed_stream_id, user.id)) {++ return sendWhipError(res, 401, 'invalid_token', 'Invalid or expired token');++ }logWhipStage('auth_jwt', pathParam, { user_id: userId });}@@@ -844,6 +870,10 @@ async function handleWhipPost(req, res) {const roomId = `stream-${streamId}`;const room = await webrtcSFU.getOrCreateRoom(roomId);++ // Check authority again after room creation in case supported ingest protocols change.++ if (refusedByOpenre(stream.managed_stream_id, userId)) {++ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');++ }logWhipStage('room_creation', streamId, { roomId });const resourceId = generateResourceId();@@@ -897,6 +927,10 @@ async function handleWhipPost(req, res) {return sendWhipError(res, 502, 'dtls_negotiation_failed', 'DTLS negotiation failed');}logWhipStage('dtls_connect', streamId, { transportId: transportInfo.id });++ if (refusedByOpenre(stream.managed_stream_id, userId)) {++ cleanupSession(resourceId);++ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');++ }const routerCaps = room.router.rtpCapabilities;const producersByKind = {};@@@ -6,6 +6,7 @@// Events deliveries, applied once, in revision order) become ordinary `streams` rows.const assert = require('assert');++const crypto = require('crypto');const fs = require('fs');const os = require('os');const path = require('path');@@@ -149,7 +150,15 @@ function sessionEvent(type, sessionId, revision, payload = {}) {assert.strictEqual(switched.ingest_authority, 'openre');assert.strictEqual(db.getManagedStreamByStreamKey('livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), undefined, "Live's old key is rotated away");assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched }), true, "Live's RTMP refuses the slot");-- assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only RTMP moved');++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only the protocols OpenRe carries are refused');++ assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp']);++ for (const protocol of ['rtmp', 'webrtc', 'jsmpeg']) {++ const refused = protocol === 'rtmp';++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol }), refused, `${protocol}: slot refusal matches OpenRe support`);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 }, protocol }), refused, `${protocol}: personal key refusal matches OpenRe support`);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(703), protocol }), false, `${protocol}: a live slot is admitted`);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 }, protocol }), false, `${protocol}: a personal key without an OpenRe slot is admitted`);++ }assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 } }), false, 'personal keys of users without an OpenRe slot stay on Live');assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 } }), true, 'a personal key cannot double-ingest a switched channel');const shown = authority.serializeSlot(switched);@@@ -217,6 +226,84 @@ function sessionEvent(type, sessionId, revision, payload = {}) {delete noId.event_id;assert.strictEqual(await deliver(base, noId), 204);++ // ── WHIP and JSMPEG publishers ───────────────────────────++ d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (603, 'browsercaster', 'BrowserCaster', 'x', 'personalkey603')").run();++ const whipKey = () => {++ let key;++ do { key = crypto.randomBytes(16).toString('hex'); } while (!/[a-f]/.test(key));++ return key;++ };++ const openReWhipKey = whipKey();++ const liveWhipKey = whipKey();++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, ingest_authority) VALUES (704, 603, 'WHIP on OpenRe', 'webrtc', ?, 'openre')").run(openReWhipKey);++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (705, 602, 'WHIP on Live', 'webrtc', ?)").run(liveWhipKey);++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, ingest_authority) VALUES (706, 603, 'JSMPEG on OpenRe', 'jsmpeg', 'jsmpegopenre706', 'openre')").run();++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (707, 602, 'JSMPEG on Live', 'jsmpeg', 'jsmpeglive707')").run();++ process.env.OPENRE_URL = stubUrl;++ require('../server/openre/openre-client')._reset();++ const whip = require('../server/streaming/whip-handler');++ const publish = async (id, { bearer, key } = {}) => {++ const out = {};++ const res = { status(c) { out.status = c; return this; }, set(k, v) { out[String(k).toLowerCase()] = v; return this; }, json(b) { out.body = b; return this; } };++ const req = { params: { streamId: String(id) }, headers: bearer ? { authorization: `Bearer ${bearer}` } : {}, query: key ? { key } : {}, body: 'v=0\r\n' };++ await whip.handleWhipPost(req, res);++ return out;++ };++ for (const [label, call] of [++ ['key in the path', () => publish(openReWhipKey)],++ ['slot id + key', () => publish(704, { key: openReWhipKey })],++ ]) {++ const r = await call();++ assert.ok([200, 201, 503].includes(r.status), `WHIP ${label} stays on Live: ${r.status}`);++ }++ const admitted = await publish(liveWhipKey);++ assert.notStrictEqual(admitted.status, 401, 'a Live slot is still admitted (here it reaches the SFU check)');++ for (const r of [admitted, await publish(705, { key: liveWhipKey })]) assert.ok([200, 201, 503].includes(r.status), `live slot: ${r.status}`);++ // Switching RTMP authority leaves an existing WHIP slot on Live.++ d.prepare("UPDATE managed_streams SET ingest_authority = 'openre' WHERE id = 705").run();++ assert.ok([200, 201, 503].includes((await publish(liveWhipKey)).status), 'WHIP stays on Live after the switch');++ d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 705").run();++++ const relay = require('../server/streaming/jsmpeg-relay');++ relay.nextVideoPort = 29710 + (process.pid % 1000) * 4;++ relay.nextAudioPort = relay.nextVideoPort + 1;++ const post = (port, key) => new Promise((resolve, reject) => {++ const r = http.request({ host: '127.0.0.1', port, method: 'POST', path: `/${key}`, agent: false }, (res) => { res.resume(); res.on('end', () => resolve(res.statusCode)); });++ r.on('error', reject);++ r.end(Buffer.from('mpeg'));++ });++ const openreCh = relay.createChannel('jsmpegopenre706');++ const liveCh = relay.createChannel('jsmpeglive707');++ const personalCh = relay.createChannel('personalkey601');++ await new Promise(r => setTimeout(r, 200));++ assert.strictEqual(await post(openreCh.videoPort, 'jsmpegopenre706'), 200, 'JSMPEG video stays on Live');++ assert.strictEqual(await post(openreCh.audioPort, 'jsmpegopenre706'), 200, 'JSMPEG audio stays on Live');++ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 200, "JSMPEG: the owner's personal key stays on Live");++ assert.strictEqual(await post(liveCh.videoPort, 'jsmpeglive707'), 200, 'JSMPEG: a Live slot is admitted');++ assert.strictEqual(await post(liveCh.audioPort, 'jsmpeglive707'), 200);++ relay.closeAll();++++ // OPENRE_URL unset: nothing is refused, for any protocol.++ delete process.env.OPENRE_URL;++ for (const protocol of ['rtmp', 'webrtc', 'jsmpeg']) {++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(704), protocol }), false);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 603 }, protocol }), false);++ }++ assert.notStrictEqual((await publish(openReWhipKey)).status, 401, 'without OPENRE_URL the WHIP slot is admitted again');++ process.env.OPENRE_URL = stubUrl;++ require('../server/openre/openre-client')._reset();++++ // The admin status lists the protocols OpenRe carries.++ const statusApp = express();++ // routes.js takes requireAdmin at load: admit the test caller before the router is first required.++ require('../server/auth/auth').requireAdmin = (req, res, next) => { req.user = { id: 1, username: 'admin', role: 'admin' }; next(); };++ statusApp.use('/api/admin/openre', require('../server/openre/routes'));++ const statusServer = await new Promise(r => { const s = statusApp.listen(0, '127.0.0.1', () => r(s)); });++ const statusRes = await fetch(`http://127.0.0.1:${statusServer.address().port}/api/admin/openre/status`);++ if (statusRes.status === 200) assert.deepStrictEqual((await statusRes.json()).protocols, ['rtmp']);++ else assert.fail(`status route answered ${statusRes.status}`);++ statusServer.close();++// ── Rollback ─────────────────────────────────────────────assert.strictEqual((await authority.setAuthority(701, 'live')).status, 200);const back = db.getManagedStreamById(701);@@@ -230,5 +317,5 @@ function sessionEvent(type, sessionId, revision, payload = {}) {mirror._reset();server.close();stub.close();-- console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, signed mirror (once, ordered), reconcile, rollback');++ console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, WHIP/JSMPEG stay on Live, signed mirror (once, ordered), reconcile, rollback');})().catch((err) => { console.error(err); process.exit(1); });
@@ -5,8 +5,7 @@
@@ -5,8 +5,7 @@
* 'openre' means OpenRe.Stream ingests this slot's RTMP: Live's RTMP server refuses the slot's
* 'openre' means OpenRe.Stream ingests this slot's RTMP: Live's RTMP server refuses the slot's
* key (so a stream can never be ingested twice), the Go Live UI shows OpenRe's ingest URL and
* key (so a stream can never be ingested twice), the Go Live UI shows OpenRe's ingest URL and
* rotates OpenRe's key, and the slot's sessions reach `streams` through the mirror (mirror.js).
* rotates OpenRe's key, and the slot's sessions reach `streams` through the mirror (mirror.js).
- * Only RTMP moves (OPENRE_PROTOCOLS); WHIP, browser/WebRTC and JSMPEG stay on Live until
- * OpenRe carries them.
+ * Only RTMP moves (OPENRE_PROTOCOLS). WHIP and JSMPEG still publish through Live.
*
*
* Switch off = zero behaviour change: every helper here returns what Live did before unless the
* Switch off = zero behaviour change: every helper here returns what Live did before unless the
* slot says 'openre' AND OpenRe is configured (openre-client enabled()).
* slot says 'openre' AND OpenRe is configured (openre-client enabled()).
@@ -33,9 +32,9 @@ function slotIsOpenre(id) {
@@ -33,9 +32,9 @@ function slotIsOpenre(id) {
}
}
/**
/**
- * Live's RTMP publish handler asks this before accepting a key. A slot key is refused when its slot
- * is on OpenRe; a personal key (users.stream_key, no slot) is refused when any of the user's slots is,
- * because Live would attach it to the user's live rows and push the user's destinations twice.
+ * Live's publish handlers ask this before accepting a key. Only RTMP is refused when its slot
+ * is on OpenRe; an RTMP personal key (users.stream_key, no slot) is refused when any of the user's
+ * slots is on OpenRe, because Live would attach it to the user's live rows and push destinations twice.
*/
*/
function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {
function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {
try {
try {
@@ -22,6 +22,24 @@ function matchesStreamPath(url, streamKey) {
@@ -22,6 +22,24 @@ function matchesStreamPath(url, streamKey) {
}
}
}
}
+/**
+ * Consult the per-protocol ingest authority for each publish. OpenRe currently ingests RTMP
+ * only, so JSMPEG continues to publish through Live after an RTMP switch.
+ */
+function refusedByOpenre(streamKey) {
+ try {
+ const db = require('../db/database');
+ const user = db.getUserByStreamKey(streamKey);
+ const managedStream = user ? null : db.getManagedStreamByStreamKey(streamKey);
+ if (!user && !managedStream) return false;
+ if (!require('../openre/authority').refusesLiveIngest({ managedStream, user, protocol: 'jsmpeg' })) return false;
+ console.log(`[JSMPEG] Rejected: ${managedStream ? `slot ${managedStream.id}` : `personal key of ${user.username}`} is ingested by OpenRe`);
+ return true;
+ } catch {
+ return false;
+ }
+}
+
class JSMPEGRelay {
class JSMPEGRelay {
constructor() {
constructor() {
/** @type {Map<string, { videoWss: WebSocket.Server, audioWss: WebSocket.Server, videoServer: http.Server, audioServer: http.Server }>} */
/** @type {Map<string, { videoWss: WebSocket.Server, audioWss: WebSocket.Server, videoServer: http.Server, audioServer: http.Server }>} */
@@ -54,7 +72,7 @@ class JSMPEGRelay {
@@ -54,7 +72,7 @@ class JSMPEGRelay {
const videoServer = http.createServer((req, res) => {
const videoServer = http.createServer((req, res) => {
// FFmpeg sends MPEG1 data via HTTP POST
// FFmpeg sends MPEG1 data via HTTP POST
req.socket.setNoDelay(true);
req.socket.setNoDelay(true);
- if (req.method === 'POST' && matchesStreamPath(req.url, streamKey)) {
+ if (req.method === 'POST' && matchesStreamPath(req.url, streamKey) && !refusedByOpenre(streamKey)) {
req.on('data', (chunk) => {
req.on('data', (chunk) => {
// Broadcast to WebSocket viewers
// Broadcast to WebSocket viewers
if (videoWss.clients.size > 0) {
if (videoWss.clients.size > 0) {
@@ -99,7 +117,7 @@ class JSMPEGRelay {
@@ -99,7 +117,7 @@ class JSMPEGRelay {
const audioWss = new WebSocket.Server({ noServer: true, perMessageDeflate: false, maxPayload: 64 * 1024 });
const audioWss = new WebSocket.Server({ noServer: true, perMessageDeflate: false, maxPayload: 64 * 1024 });
const audioServer = http.createServer((req, res) => {
const audioServer = http.createServer((req, res) => {
req.socket.setNoDelay(true);
req.socket.setNoDelay(true);
- if (req.method === 'POST' && matchesStreamPath(req.url, streamKey)) {
+ if (req.method === 'POST' && matchesStreamPath(req.url, streamKey) && !refusedByOpenre(streamKey)) {
req.on('data', (chunk) => {
req.on('data', (chunk) => {
// Broadcast to WebSocket viewers
// Broadcast to WebSocket viewers
if (audioWss.clients.size > 0) {
if (audioWss.clients.size > 0) {
@@ -119,6 +119,18 @@ function sendWhipError(res, status, code, message) {
@@ -119,6 +119,18 @@ function sendWhipError(res, status, code, message) {
.json({ error: message, error_code: code });
.json({ error: message, error_code: code });
}
}
+/**
+ * Consult the per-protocol ingest authority before accepting a publisher. OpenRe currently
+ * ingests RTMP only, so WHIP continues to publish through Live after an RTMP switch.
+ */
+function refusedByOpenre(slotId, userId) {
+ const managedStream = slotId ? db.getManagedStreamById(slotId) : null;
+ const user = managedStream || !userId ? null : db.getUserById(userId);
+ if (!require('../openre/authority').refusesLiveIngest({ managedStream, user, protocol: 'webrtc' })) return false;
+ console.log(`[WHIP] Rejected: ${managedStream ? `slot ${managedStream.id}` : `personal key of user ${userId}`} is ingested by OpenRe`);
+ return true;
+}
+
function logWhipStage(stage, streamId, info = {}) {
function logWhipStage(stage, streamId, info = {}) {
const details = Object.entries(info)
const details = Object.entries(info)
.filter(([, value]) => value !== undefined && value !== null)
.filter(([, value]) => value !== undefined && value !== null)
@@ -670,6 +682,11 @@ async function handleWhipPost(req, res) {
@@ -670,6 +682,11 @@ async function handleWhipPost(req, res) {
return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');
return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');
}
}
+ if (refusedByOpenre(managedStream.id, managedStream.user_id)) {
+ logWhipStage('auth_key_fail', pathParam, { reason: 'stream_key_not_found' });
+ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');
+ }
+
userId = managedStream.user_id;
userId = managedStream.user_id;
logWhipStage('auth_key', pathParam, { user_id: userId, managed_stream_id: managedStream.id });
logWhipStage('auth_key', pathParam, { user_id: userId, managed_stream_id: managedStream.id });
@@ -736,6 +753,9 @@ async function handleWhipPost(req, res) {
@@ -736,6 +753,9 @@ async function handleWhipPost(req, res) {
logWhipStage('auth_slot_fail', pathParam, { reason: 'not_slot_owner', user_id: user.id, slot_id: slotId });
logWhipStage('auth_slot_fail', pathParam, { reason: 'not_slot_owner', user_id: user.id, slot_id: slotId });
return sendWhipError(res, 403, 'not_slot_owner', 'This stream slot belongs to another account');
return sendWhipError(res, 403, 'not_slot_owner', 'This stream slot belongs to another account');
}
}
+ if (refusedByOpenre(managedStream.id, user.id)) {
+ return sendWhipError(res, 401, 'invalid_key', 'Stream key or token does not match');
+ }
userId = user.id;
userId = user.id;
// Find live session for this slot owned by this user
// Find live session for this slot owned by this user
const liveSessions = db.getLiveStreamsByUserId(user.id) || [];
const liveSessions = db.getLiveStreamsByUserId(user.id) || [];
@@ -758,6 +778,9 @@ async function handleWhipPost(req, res) {
@@ -758,6 +778,9 @@ async function handleWhipPost(req, res) {
}
}
if (!stream) {
if (!stream) {
+ if (refusedByOpenre(managedStream.id, managedStream.user_id)) {
+ return sendWhipError(res, 401, 'invalid_key', 'Stream key does not match this slot');
+ }
userId = managedStream.user_id;
userId = managedStream.user_id;
logWhipStage('auth_slot_key', pathParam, { user_id: userId, slot_id: slotId });
logWhipStage('auth_slot_key', pathParam, { user_id: userId, slot_id: slotId });
@@ -813,6 +836,9 @@ async function handleWhipPost(req, res) {
@@ -813,6 +836,9 @@ async function handleWhipPost(req, res) {
}
}
if (!stream.is_live) return sendWhipError(res, 409, 'stream_not_live', 'Stream is not live — go live first');
if (!stream.is_live) return sendWhipError(res, 409, 'stream_not_live', 'Stream is not live — go live first');
if (stream.protocol !== 'webrtc') return sendWhipError(res, 409, 'wrong_protocol', 'Stream protocol must be webrtc for WHIP');
if (stream.protocol !== 'webrtc') return sendWhipError(res, 409, 'wrong_protocol', 'Stream protocol must be webrtc for WHIP');
+ if (refusedByOpenre(stream.managed_stream_id, user.id)) {
+ return sendWhipError(res, 401, 'invalid_token', 'Invalid or expired token');
+ }
logWhipStage('auth_jwt', pathParam, { user_id: userId });
logWhipStage('auth_jwt', pathParam, { user_id: userId });
}
}
@@ -844,6 +870,10 @@ async function handleWhipPost(req, res) {
@@ -844,6 +870,10 @@ async function handleWhipPost(req, res) {
const roomId = `stream-${streamId}`;
const roomId = `stream-${streamId}`;
const room = await webrtcSFU.getOrCreateRoom(roomId);
const room = await webrtcSFU.getOrCreateRoom(roomId);
+ // Check authority again after room creation in case supported ingest protocols change.
+ if (refusedByOpenre(stream.managed_stream_id, userId)) {
+ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');
+ }
logWhipStage('room_creation', streamId, { roomId });
logWhipStage('room_creation', streamId, { roomId });
const resourceId = generateResourceId();
const resourceId = generateResourceId();
@@ -897,6 +927,10 @@ async function handleWhipPost(req, res) {
@@ -897,6 +927,10 @@ async function handleWhipPost(req, res) {
return sendWhipError(res, 502, 'dtls_negotiation_failed', 'DTLS negotiation failed');
return sendWhipError(res, 502, 'dtls_negotiation_failed', 'DTLS negotiation failed');
}
}
logWhipStage('dtls_connect', streamId, { transportId: transportInfo.id });
logWhipStage('dtls_connect', streamId, { transportId: transportInfo.id });
+ if (refusedByOpenre(stream.managed_stream_id, userId)) {
+ cleanupSession(resourceId);
+ return sendWhipError(res, 401, 'invalid_stream_key', 'Stream key not recognized');
+ }
const routerCaps = room.router.rtpCapabilities;
const routerCaps = room.router.rtpCapabilities;
const producersByKind = {};
const producersByKind = {};
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
// Events deliveries, applied once, in revision order) become ordinary `streams` rows.
// Events deliveries, applied once, in revision order) become ordinary `streams` rows.
const assert = require('assert');
const assert = require('assert');
+const crypto = require('crypto');
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
const path = require('path');
const path = require('path');
@@ -149,7 +150,15 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -149,7 +150,15 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
assert.strictEqual(switched.ingest_authority, 'openre');
assert.strictEqual(switched.ingest_authority, 'openre');
assert.strictEqual(db.getManagedStreamByStreamKey('livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), undefined, "Live's old key is rotated away");
assert.strictEqual(db.getManagedStreamByStreamKey('livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), undefined, "Live's old key is rotated away");
assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched }), true, "Live's RTMP refuses the slot");
assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched }), true, "Live's RTMP refuses the slot");
- assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only RTMP moved');
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only the protocols OpenRe carries are refused');
+ assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp']);
+ for (const protocol of ['rtmp', 'webrtc', 'jsmpeg']) {
+ const refused = protocol === 'rtmp';
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol }), refused, `${protocol}: slot refusal matches OpenRe support`);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 }, protocol }), refused, `${protocol}: personal key refusal matches OpenRe support`);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(703), protocol }), false, `${protocol}: a live slot is admitted`);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 }, protocol }), false, `${protocol}: a personal key without an OpenRe slot is admitted`);
+ }
assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 } }), false, 'personal keys of users without an OpenRe slot stay on Live');
assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 } }), false, 'personal keys of users without an OpenRe slot stay on Live');
assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 } }), true, 'a personal key cannot double-ingest a switched channel');
assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 } }), true, 'a personal key cannot double-ingest a switched channel');
const shown = authority.serializeSlot(switched);
const shown = authority.serializeSlot(switched);
@@ -217,6 +226,84 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -217,6 +226,84 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
delete noId.event_id;
delete noId.event_id;
assert.strictEqual(await deliver(base, noId), 204);
assert.strictEqual(await deliver(base, noId), 204);
+ // ── WHIP and JSMPEG publishers ───────────────────────────
+ d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (603, 'browsercaster', 'BrowserCaster', 'x', 'personalkey603')").run();
+ const whipKey = () => {
+ let key;
+ do { key = crypto.randomBytes(16).toString('hex'); } while (!/[a-f]/.test(key));
+ return key;
+ };
+ const openReWhipKey = whipKey();
+ const liveWhipKey = whipKey();
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, ingest_authority) VALUES (704, 603, 'WHIP on OpenRe', 'webrtc', ?, 'openre')").run(openReWhipKey);
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (705, 602, 'WHIP on Live', 'webrtc', ?)").run(liveWhipKey);
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, ingest_authority) VALUES (706, 603, 'JSMPEG on OpenRe', 'jsmpeg', 'jsmpegopenre706', 'openre')").run();
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (707, 602, 'JSMPEG on Live', 'jsmpeg', 'jsmpeglive707')").run();
+ process.env.OPENRE_URL = stubUrl;
+ require('../server/openre/openre-client')._reset();
+ const whip = require('../server/streaming/whip-handler');
+ const publish = async (id, { bearer, key } = {}) => {
+ const out = {};
+ const res = { status(c) { out.status = c; return this; }, set(k, v) { out[String(k).toLowerCase()] = v; return this; }, json(b) { out.body = b; return this; } };
+ const req = { params: { streamId: String(id) }, headers: bearer ? { authorization: `Bearer ${bearer}` } : {}, query: key ? { key } : {}, body: 'v=0\r\n' };
+ await whip.handleWhipPost(req, res);
+ return out;
+ };
+ for (const [label, call] of [
+ ['key in the path', () => publish(openReWhipKey)],
+ ['slot id + key', () => publish(704, { key: openReWhipKey })],
+ ]) {
+ const r = await call();
+ assert.ok([200, 201, 503].includes(r.status), `WHIP ${label} stays on Live: ${r.status}`);
+ }
+ const admitted = await publish(liveWhipKey);
+ assert.notStrictEqual(admitted.status, 401, 'a Live slot is still admitted (here it reaches the SFU check)');
+ for (const r of [admitted, await publish(705, { key: liveWhipKey })]) assert.ok([200, 201, 503].includes(r.status), `live slot: ${r.status}`);
+ // Switching RTMP authority leaves an existing WHIP slot on Live.
+ d.prepare("UPDATE managed_streams SET ingest_authority = 'openre' WHERE id = 705").run();
+ assert.ok([200, 201, 503].includes((await publish(liveWhipKey)).status), 'WHIP stays on Live after the switch');
+ d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 705").run();
+
+ const relay = require('../server/streaming/jsmpeg-relay');
+ relay.nextVideoPort = 29710 + (process.pid % 1000) * 4;
+ relay.nextAudioPort = relay.nextVideoPort + 1;
+ const post = (port, key) => new Promise((resolve, reject) => {
+ const r = http.request({ host: '127.0.0.1', port, method: 'POST', path: `/${key}`, agent: false }, (res) => { res.resume(); res.on('end', () => resolve(res.statusCode)); });
+ r.on('error', reject);
+ r.end(Buffer.from('mpeg'));
+ });
+ const openreCh = relay.createChannel('jsmpegopenre706');
+ const liveCh = relay.createChannel('jsmpeglive707');
+ const personalCh = relay.createChannel('personalkey601');
+ await new Promise(r => setTimeout(r, 200));
+ assert.strictEqual(await post(openreCh.videoPort, 'jsmpegopenre706'), 200, 'JSMPEG video stays on Live');
+ assert.strictEqual(await post(openreCh.audioPort, 'jsmpegopenre706'), 200, 'JSMPEG audio stays on Live');
+ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 200, "JSMPEG: the owner's personal key stays on Live");
+ assert.strictEqual(await post(liveCh.videoPort, 'jsmpeglive707'), 200, 'JSMPEG: a Live slot is admitted');
+ assert.strictEqual(await post(liveCh.audioPort, 'jsmpeglive707'), 200);
+ relay.closeAll();
+
+ // OPENRE_URL unset: nothing is refused, for any protocol.
+ delete process.env.OPENRE_URL;
+ for (const protocol of ['rtmp', 'webrtc', 'jsmpeg']) {
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(704), protocol }), false);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 603 }, protocol }), false);
+ }
+ assert.notStrictEqual((await publish(openReWhipKey)).status, 401, 'without OPENRE_URL the WHIP slot is admitted again');
+ process.env.OPENRE_URL = stubUrl;
+ require('../server/openre/openre-client')._reset();
+
+ // The admin status lists the protocols OpenRe carries.
+ const statusApp = express();
+ // routes.js takes requireAdmin at load: admit the test caller before the router is first required.
+ require('../server/auth/auth').requireAdmin = (req, res, next) => { req.user = { id: 1, username: 'admin', role: 'admin' }; next(); };
+ statusApp.use('/api/admin/openre', require('../server/openre/routes'));
+ const statusServer = await new Promise(r => { const s = statusApp.listen(0, '127.0.0.1', () => r(s)); });
+ const statusRes = await fetch(`http://127.0.0.1:${statusServer.address().port}/api/admin/openre/status`);
+ if (statusRes.status === 200) assert.deepStrictEqual((await statusRes.json()).protocols, ['rtmp']);
+ else assert.fail(`status route answered ${statusRes.status}`);
+ statusServer.close();
+
// ── Rollback ─────────────────────────────────────────────
// ── Rollback ─────────────────────────────────────────────
assert.strictEqual((await authority.setAuthority(701, 'live')).status, 200);
assert.strictEqual((await authority.setAuthority(701, 'live')).status, 200);
const back = db.getManagedStreamById(701);
const back = db.getManagedStreamById(701);
@@ -230,5 +317,5 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -230,5 +317,5 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
mirror._reset();
mirror._reset();
server.close();
server.close();
stub.close();
stub.close();
- console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, signed mirror (once, ordered), reconcile, rollback');
+ console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, WHIP/JSMPEG stay on Live, signed mirror (once, ordered), reconcile, rollback');
})().catch((err) => { console.error(err); process.exit(1); });
})().catch((err) => { console.error(err); process.exit(1); });