openvibers/openvibe.live Commits
Refuse WHIP, browser and JSMPEG ingest for slots on OpenRe
Changes
@@@ -2,10 +2,11 @@/*** The per-slot ingest switch: managed_streams.ingest_authority = 'live' (default) | 'openre'.*-- * '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 and JSMPEG still publish through Live.++ * 'openre' means OpenRe.Stream ingests this slot: Live's RTMP server, WHIP handler, browser++ * broadcaster socket and JSMPEG relay refuse the slot's key (so a stream can never be ingested++ * twice), the Go Live UI shows OpenRe's ingest URLs and rotates OpenRe's key, and the slot's++ * sessions reach `streams` through the mirror (mirror.js). RTMP, WebRTC (WHIP) and JSMPEG move++ * (OPENRE_PROTOCOLS): OpenRe serves WHIP on ingest.openre.stream and runs the JSMPEG worker.** 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()).@@@ -14,7 +15,15 @@ const crypto = require('crypto');const db = require('../db/database');const client = require('./openre-client');--const OPENRE_PROTOCOLS = new Set(['rtmp']);++const OPENRE_PROTOCOLS = new Set(['rtmp', 'webrtc', 'jsmpeg']);++++/** The OpenRe definition protocol a slot publishes with (OpenRe's own migrate-live mapping). */++function openreProtocolOf(slot) {++ const p = String((slot && slot.protocol) || '').toLowerCase();++ if (p === 'webrtc' || p === 'whip') return 'webrtc';++ if (p === 'jsmpeg') return 'jsmpeg';++ return 'rtmp';++}function authorityOf(slot) {if (!slot || slot.ingest_authority !== 'openre') return 'live';@@@ -32,9 +41,10 @@ function slotIsOpenre(id) {}/**-- * 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.++ * Live's publish handlers (RTMP, WHIP, browser broadcaster, JSMPEG) ask 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.*/function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {try {@@@ -58,20 +68,33 @@ function serializeSlot(slot) {return { ...slot, stream_key: null, ingest_authority: 'openre', stream_key_managed_by: 'openre', openre_manage_url: client.manageUrl(slot.openre_stream_id) };}++/** OpenRe's ingest URLs from a stream or rotation answer (`ingest.rtmp.url`, `ingest.webrtc.whip_url`,++ * `ingest.jsmpeg.url`; each only for the protocols the definition allows). The key goes after each:++ * rtmp://…/live + key, <whip_url>/<key>, <jsmpeg_url>/<key>/<width>/<height>/. */++function ingestUrls(ingest) {++ const i = ingest || {};++ return {++ rtmp_url: i.rtmp ? i.rtmp.url : null,++ whip_url: i.webrtc ? i.webrtc.whip_url || null : null,++ jsmpeg_url: i.jsmpeg ? i.jsmpeg.url || null : null,++ };++}++/**-- * What the Go Live UI shows for an OpenRe slot: OpenRe's RTMP URL and the hint of its current key-- * (the key itself is only ever shown once, by a rotation).++ * What the Go Live UI shows for an OpenRe slot: OpenRe's ingest URLs and the hint of its current key++ * (one key for every protocol; the key itself is only ever shown once, by a rotation).*/async function ingestFor(slot, subject) {const stream = slot.openre_stream_id ? await client.getStream(slot.openre_stream_id, { subject }) : await client.streamForSlot(slot.id);-- const rtmp = stream && stream.ingest && stream.ingest.rtmp;++ const ingest = (stream && stream.ingest) || {};++ const keyHint = ['rtmp', 'webrtc', 'jsmpeg'].map(p => ingest[p] && ingest[p].key_hint).find(Boolean);return {ingest_authority: 'openre',stream_key_managed_by: 'openre',openre_stream_id: stream ? stream.id : null,-- rtmp_url: rtmp ? rtmp.url : null,-- stream_key_hint: rtmp && rtmp.key_hint-- ? `Shown once: press Regenerate for a new key (current key ends in …${rtmp.key_hint})`++ ...ingestUrls(ingest),++ stream_key_hint: keyHint++ ? `Shown once: press Regenerate for a new key (current key ends in …${keyHint})`: 'Press Regenerate to get your OpenRe stream key',openre_manage_url: client.manageUrl(stream ? stream.id : slot.openre_stream_id),};@@@ -82,7 +105,7 @@ async function rotateFor(slot, subject) {const streamId = slot.openre_stream_id || (await client.streamForSlot(slot.id) || {}).id;if (!streamId) throw new client.OpenReError('this slot has no OpenRe stream yet', 409);const r = await client.rotateKey(streamId, { subject });-- return { stream_key: r.key.key, stream_key_managed_by: 'openre', rtmp_url: r.ingest && r.ingest.rtmp ? r.ingest.rtmp.url : null };++ return { stream_key: r.key.key, stream_key_managed_by: 'openre', ...ingestUrls(r.ingest) };}function recordingModeFor(slot) {@@@ -94,7 +117,8 @@ function recordingModeFor(slot) {/*** The admin switch (PUT /api/admin/openre/managed/:id/ingest-authority).* 'openre': needs OpenRe configured, the slot offline on Live, the owner's canonical subject;-- * finds or creates the OpenRe definition, then (one transaction) flips the slot and++ * finds or creates the OpenRe definition (allowing the slot's protocol on it), then++ * (one transaction) flips the slot and* rotates Live's own key for it, so no key that existed before works anywhere.* 'live': flips back (rollback). Live's key was rotated at the switch: the broadcaster* regenerates it on the Go Live page.@@@ -111,9 +135,12 @@ async function setAuthority(slotId, authority, { force = false } = {}) {if (slot.ingest_authority === 'openre') return { status: 200, body: { managed_stream_id: slot.id, ingest_authority: 'openre', openre_stream_id: slot.openre_stream_id, unchanged: true } };const live = db.get('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1', [slot.id]);if (live) return { status: 409, error: `Slot ${slot.id} is live on Live (stream ${live.id}); switch it in a maintenance window, while it is offline` };++ const protocol = openreProtocolOf(slot);const method = String(slot.streaming_method || '').toLowerCase();-- if (!force && slot.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(method)) {-- return { status: 409, error: `Slot ${slot.id} is a ${slot.protocol}${method ? `/${method}` : ''} slot; OpenRe only ingests RTMP today (pass force to switch anyway)` };++ // WHIP encoders take OpenRe's WHIP URL and key; the Go Live page's in-browser broadcaster only++ // publishes to Live's own SFU, so a browser slot moved now could not go live from the page.++ if (!force && protocol === 'webrtc' && !['whip', 'obs'].includes(method)) {++ return { status: 409, error: `Slot ${slot.id} is a ${slot.protocol}${method ? `/${method}` : ''} slot; the Go Live page cannot publish to OpenRe yet, only WHIP encoders can (pass force to switch anyway)` };}const subject = require('../auth/identity-sync').subjectOf(slot.user_id);if (!subject) return { status: 409, error: 'The owner has no canonical subject yet (they need to sign in to Live once)' };@@@ -121,7 +148,10 @@ async function setAuthority(slotId, authority, { force = false } = {}) {if (!stream) {let visibility = 'public';try { visibility = db.resolveStreamVodVisibility({ user_id: slot.user_id, managed_stream_id: slot.id }); } catch { /* public */ }-- stream = await client.createStreamForSlot(slot, { subject, recordingMode: recordingModeFor(slot), recordingVisibility: visibility });++ stream = await client.createStreamForSlot(slot, { subject, protocols: [protocol], recordingMode: recordingModeFor(slot), recordingVisibility: visibility });++ } else if (!(stream.protocols || ['rtmp']).includes(protocol)) {++ // An existing definition (OpenRe's own default is RTMP only) would refuse the slot's encoder.++ stream = await client.updateStream(stream.id, { protocols: [...(stream.protocols || ['rtmp']), protocol] }, { subject });}const newLiveKey = crypto.randomBytes(20).toString('hex');const flipped = db.getDb().transaction(() => {@@@ -137,7 +167,7 @@ async function setAuthority(slotId, authority, { force = false } = {}) {managed_stream_id: slot.id,ingest_authority: 'openre',openre_stream_id: stream.id,-- rtmp_url: stream.ingest && stream.ingest.rtmp ? stream.ingest.rtmp.url : null,++ ...ingestUrls(stream.ingest),live_key_rotated: true,next: 'The broadcaster presses Regenerate on the Go Live page (OpenRe issues the key) and pastes the new server and key into OBS.',},@@@ -80,9 +80,10 @@ const streamForSlot = (managedStreamId) => wrap('stream lookup', () => openre()./** Create the definition for a slot, owned by the streamer's canonical subject. The key it* returns is dropped unseen: the streamer gets a usable key by rotating. */--async function createStreamForSlot(slot, { subject, recordingMode, recordingVisibility }) {++async function createStreamForSlot(slot, { subject, protocols, recordingMode, recordingVisibility }) {const r = await wrap('stream create', () => openre().streams.create({title: slot.title || 'Stream',++ ...(protocols ? { protocols } : {}),recording_mode: recordingMode,recording_visibility: recordingVisibility,mirror_to_live: true,@@@ -95,6 +96,7 @@ async function createStreamForSlot(slot, { subject, recordingMode, recordingVisi}const rotateKey = (streamId, { subject, graceSeconds = 0 } = {}) => wrap('key rotate', () => openre().streams.rotateKey(streamId, { subject, graceSeconds }));++const updateStream = (streamId, patch, { subject } = {}) => wrap('stream update', () => openre().streams.update(streamId, patch, { subject }));const getStream = (streamId, { subject } = {}) => wrap('stream read', () => openre().streams.get(streamId, { subject }));const getSession = (sessionId) => wrap('session read', () => openre().sessions.get(sessionId));/** Playback descriptor for a session (the SDK caches it 10 s: the FLV proxy asks on every viewer connect). */@@@ -106,4 +108,4 @@ function manageUrl(streamId) {function _reset() { sdk = null; }--module.exports = { enabled, settings, request, streamForSlot, createStreamForSlot, rotateKey, getStream, getSession, playback, manageUrl, OpenReError, _reset };++module.exports = { enabled, settings, request, streamForSlot, createStreamForSlot, updateStream, rotateKey, getStream, getSession, playback, manageUrl, OpenReError, _reset };@@@ -235,6 +235,15 @@ class BroadcastServer extends EventEmitter {ws.close(4003, 'Not your stream');return;}++ // OpenRe.Stream ingests this slot (or, for a slot-less stream, one of the owner's slots):++ // Live's SFU never takes a second publisher for it. See server/openre/authority.js.++ const openre = require('../openre/authority');++ const managedStream = openre.slotById(stream.managed_stream_id);++ if (openre.refusesLiveIngest({ managedStream, user: managedStream ? null : user, protocol: 'webrtc' })) {++ console.warn(`[Broadcast] Broadcaster refused for stream ${streamId}: ${managedStream ? `slot ${managedStream.id}` : `personal stream of ${user.username}`} is ingested by OpenRe`);++ ws.close(4003, 'Stream key not recognized');++ return;++ }}// Viewers: only for a stream that exists and is live, and only a handful of sockets per address.@@@ -2270,6 +2270,24 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {const hostname = config.host === '0.0.0.0' ? req.hostname : config.host;++ if (openreAuthority.OPENRE_PROTOCOLS.has(stream.protocol) && openreAuthority.slotIsOpenre(stream.managed_stream_id)) {++ // OpenRe ingests this slot: its server URL for the slot's protocol, the hint of its key++ // (never the key). No Live relay channel, room or recorder is set up for it.++ const slot = openreAuthority.slotById(stream.managed_stream_id);++ const ingest = await openreAuthority.ingestFor(slot, require('../auth/identity-sync').subjectOf(slot.user_id)).catch(() => ({}));++ const serverUrl = stream.protocol === 'webrtc' ? { whipUrl: ingest.whip_url || null }++ : stream.protocol === 'jsmpeg' ? { jsmpegUrl: ingest.jsmpeg_url || null }++ : { rtmpUrl: ingest.rtmp_url || null };++ endpoint = {++ ...serverUrl,++ streamKey: null,++ streamKeyHint: ingest.stream_key_hint || 'Press Regenerate to get your OpenRe stream key',++ keyManagedBy: 'openre',++ ...(stream.protocol === 'rtmp' ? { flvUrl: `/api/streams/rtmp-proxy/${stream.id}.flv` } : {}),++ };++ return res.json({ endpoint, stream_key: null });++ }++if (stream.protocol === 'jsmpeg') {endpoint = jsmpegRelay.getChannelInfo(msKey) || jsmpegRelay.createChannel(msKey);@@@ -2305,18 +2323,6 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {whipUrlSource,...(whipUrlWarning ? { whipUrlWarning } : {}),};-- } else if (stream.protocol === 'rtmp' && openreAuthority.slotIsOpenre(stream.managed_stream_id)) {-- // OpenRe ingests this slot: its server URL, the hint of its key (never the key).-- const slot = openreAuthority.slotById(stream.managed_stream_id);-- const ingest = await openreAuthority.ingestFor(slot, require('../auth/identity-sync').subjectOf(slot.user_id)).catch(() => ({}));-- endpoint = {-- rtmpUrl: ingest.rtmp_url || null,-- streamKey: null,-- streamKeyHint: ingest.stream_key_hint || 'Press Regenerate to get your OpenRe stream key',-- keyManagedBy: 'openre',-- flvUrl: `/api/streams/rtmp-proxy/${stream.id}.flv`,-- };-- return res.json({ endpoint, stream_key: null });} else if (stream.protocol === 'rtmp') {const rtmpHost = config.rtmp.host || (() => {try { return new URL(config.baseUrl).hostname; } catch { return hostname; }@@@ -1,9 +1,10 @@'use strict';// OpenRe ingest switch (roadmap Wave 7, ADR-009). With OPENRE_URL unset, or a slot on 'live' (the--// default), nothing changes. With a slot switched to 'openre': Live's RTMP refuses its key, the--// Go Live helpers show OpenRe's URL and rotate OpenRe's key, and OpenRe session events (signed--// Events deliveries, applied once, in revision order) become ordinary `streams` rows.++// default), nothing changes. With a slot switched to 'openre': Live's RTMP, WHIP, browser++// broadcaster and JSMPEG ingest refuse its key, the Go Live helpers show OpenRe's URLs and rotate++// OpenRe's key, and OpenRe session events (signed Events deliveries, applied once, in revision++// order) become ordinary `streams` rows.const assert = require('assert');const crypto = require('crypto');@@@ -21,7 +22,14 @@ const SUBJECT = 'usr_01J0000000000000000000000Q';const SES = (n) => `ses_01J00000000000000000000${String(n).padStart(3, '0')}`.slice(0, 30);const openreCalls = [];let openreSessions = {};--let openreStreams = [];++/** OpenRe definitions by id: { id, ref, protocols }. */++const openreDefs = {};++const ingestOf = (protocols) => ({++ ...(protocols.includes('rtmp') ? { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live', key_hint: 'WXYZ' } } : {}),++ ...(protocols.includes('webrtc') ? { webrtc: { whip_url: 'https://ingest.openre.stream/whip', signaling_url: 'wss://ingest.openre.stream/b', key_hint: 'WXYZ' } } : {}),++ ...(protocols.includes('jsmpeg') ? { jsmpeg: { url: 'http://ingest.openre.stream:8081', key_hint: 'WXYZ' } } : {}),++});++const view = (def) => ({ id: def.id, protocols: def.protocols, ingest: ingestOf(def.protocols) });const stub = http.createServer((req, res) => {let body = '';@@@ -37,11 +45,25 @@ const stub = http.createServer((req, res) => {if (!req.url.startsWith('/api/v1/')) return send(404, {});assert.strictEqual(req.headers.authorization, 'Bearer svc-live');openreCalls.push({ method: req.method, url: req.url, subject: req.headers['x-ov-subject'] || null, body: body ? JSON.parse(body) : null });-- const stream = (id) => ({ id, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live', key_hint: 'WXYZ' } } });-- if (req.method === 'GET' && req.url.startsWith('/api/v1/streams?external_ref=')) return send(200, { streams: openreStreams });-- if (req.method === 'POST' && req.url === '/api/v1/streams') { openreStreams = [stream('std_01J0000000000000000000000S')]; return send(201, { stream: openreStreams[0], key: { key: 'ork_dropped' } }); }-- if (req.method === 'GET' && req.url.startsWith('/api/v1/streams/std_')) return send(200, { stream: stream(req.url.split('/').pop()) });-- if (req.method === 'POST' && req.url.endsWith('/keys/rotate')) return send(200, { key: { key: 'ork_newkey_shown_once' }, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live' } } });++ const defOf = (u) => openreDefs[u.split('/')[4]] || { id: u.split('/')[4], protocols: ['rtmp'] };++ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams?external_ref=')) {++ const ref = new URL(req.url, 'http://x').searchParams.get('external_ref');++ return send(200, { streams: Object.values(openreDefs).filter(x => x.ref === ref).map(view) });++ }++ if (req.method === 'POST' && req.url === '/api/v1/streams') {++ const b = JSON.parse(body);++ const slotId = b.external_refs[0].id;++ const def = { id: slotId === '701' ? 'std_01J0000000000000000000000S' : `std_slot${slotId}`, ref: `live:managed_stream:${slotId}`, protocols: b.protocols || ['rtmp'] };++ openreDefs[def.id] = def;++ return send(201, { stream: view(def), key: { key: 'ork_dropped' } });++ }++ if (req.method === 'PATCH' && req.url.startsWith('/api/v1/streams/std_')) {++ const def = defOf(req.url);++ def.protocols = JSON.parse(body).protocols;++ return send(200, { stream: view(def) });++ }++ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams/std_')) return send(200, { stream: view(defOf(req.url)) });++ if (req.method === 'POST' && req.url.endsWith('/keys/rotate')) return send(200, { key: { key: 'ork_newkey_shown_once' }, ingest: ingestOf(defOf(req.url).protocols) });const m = /^\/api\/v1\/sessions\/(ses_[0-9A-Z]+)(\/playback)?$/.exec(req.url);if (m && openreSessions[m[1]]) {if (m[2]) return send(200, { playback: { flv: { internal_url: `http://127.0.0.1:19999/live/${m[1]}.flv` } } });@@@ -82,6 +104,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {(async () => {const db = require('../server/db/database');db.initDb();++ // broadcast-server takes authenticateWs at load (the mirror may load it first): 'tok-<user id>' signs in.++ require('../server/auth/auth').authenticateWs = (token) => (/^tok-\d+$/.test(token || '') ? db.getUserById(Number(token.slice(4))) || null : null);const d = db.getDb();d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (601, 'caster', 'Caster', 'x', 'personalkey601')").run();d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (602, 'nosub', 'NoSub', 'x')").run();@@@ -89,6 +113,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (701, 601, 'OBS slot', 'rtmp', 'livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'obs')").run();d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (702, 601, 'Browser slot', 'webrtc', 'livekey702aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (703, 602, 'No subject', 'rtmp', 'livekey703aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (708, 601, 'WHIP encoder slot', 'webrtc', 'livekey708aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'whip')").run();++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (709, 601, 'JSMPEG slot', 'jsmpeg', 'livekey709aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'cli')").run();const authority = require('../server/openre/authority');const mirror = require('../server/openre/mirror');@@@ -134,7 +160,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams').get().n, 0, 'a slot on live is never mirrored');// ── The switch ───────────────────────────────────────────-- assert.strictEqual((await authority.setAuthority(702, 'openre')).status, 409, 'a browser slot needs force');++ assert.match((await authority.setAuthority(702, 'openre')).error || '', /Go Live page cannot publish to OpenRe/, 'a browser slot needs force');assert.strictEqual((await authority.setAuthority(703, 'openre')).status, 409, 'no canonical subject');const liveRow = db.createStream({ user_id: 601, managed_stream_id: 701, title: 'on Live', protocol: 'rtmp' });assert.strictEqual((await authority.setAuthority(701, 'openre')).status, 409, 'refused while live on Live');@@@ -146,16 +172,17 @@ function sessionEvent(type, sessionId, revision, payload = {}) {assert.strictEqual(created.subject, SUBJECT, 'created for the owner subject');assert.deepStrictEqual(created.body.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:701', 'live:user:601']);assert.strictEqual(created.body.mirror_to_live, true);++ assert.deepStrictEqual(created.body.protocols, ['rtmp'], "the definition allows the slot's protocol");++ assert.strictEqual(flip.body.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');const switched = db.getManagedStreamById(701);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 the protocols OpenRe carries are refused');-- assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp']);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'srt' }), false, 'only the protocols OpenRe carries are refused');++ assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp', 'webrtc', 'jsmpeg']);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: switched, protocol }), true, `${protocol}: the slot key is refused`);++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 }, protocol }), true, `${protocol}: the owner's personal key is refused`);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`);}@@@ -166,11 +193,34 @@ function sessionEvent(type, sessionId, revision, payload = {}) {assert.strictEqual(shown.stream_key_managed_by, 'openre');const ingest = await authority.ingestFor(switched, SUBJECT);assert.strictEqual(ingest.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');++ assert.strictEqual(ingest.whip_url, null);++ assert.strictEqual(ingest.jsmpeg_url, null);assert.match(ingest.stream_key_hint, /…WXYZ/);const rotated = await authority.rotateFor(switched, SUBJECT);assert.strictEqual(rotated.stream_key, 'ork_newkey_shown_once');++ assert.strictEqual(rotated.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');assert.strictEqual(openreCalls.filter(c => c.url.endsWith('/keys/rotate')).pop().subject, SUBJECT);++ // A WHIP encoder slot: its definition is created for WebRTC, the Go Live helpers give OpenRe's WHIP URL.++ const whipFlip = await authority.setAuthority(708, 'openre');++ assert.strictEqual(whipFlip.status, 200, JSON.stringify(whipFlip));++ assert.deepStrictEqual(openreCalls.filter(c => c.method === 'POST' && c.url === '/api/v1/streams').pop().body.protocols, ['webrtc']);++ assert.strictEqual(whipFlip.body.whip_url, 'https://ingest.openre.stream/whip');++ const whipIngest = await authority.ingestFor(db.getManagedStreamById(708), SUBJECT);++ assert.deepStrictEqual([whipIngest.rtmp_url, whipIngest.whip_url, whipIngest.jsmpeg_url], [null, 'https://ingest.openre.stream/whip', null]);++ assert.match(whipIngest.stream_key_hint, /…WXYZ/);++ assert.strictEqual((await authority.rotateFor(db.getManagedStreamById(708), SUBJECT)).whip_url, 'https://ingest.openre.stream/whip');++ // A JSMPEG slot whose OpenRe definition already exists (RTMP only, OpenRe's default): JSMPEG is added to it.++ openreDefs.std_slot709 = { id: 'std_slot709', ref: 'live:managed_stream:709', protocols: ['rtmp'] };++ const jsmpegFlip = await authority.setAuthority(709, 'openre');++ assert.strictEqual(jsmpegFlip.status, 200, JSON.stringify(jsmpegFlip));++ const patched = openreCalls.filter(c => c.method === 'PATCH').pop();++ assert.strictEqual(patched.url, '/api/v1/streams/std_slot709');++ assert.strictEqual(patched.subject, SUBJECT);++ assert.deepStrictEqual(patched.body, { protocols: ['rtmp', 'jsmpeg'] });++ assert.strictEqual(jsmpegFlip.body.jsmpeg_url, 'http://ingest.openre.stream:8081');++ assert.strictEqual((await authority.ingestFor(db.getManagedStreamById(709), SUBJECT)).jsmpeg_url, 'http://ingest.openre.stream:8081');++// ── Mirror ────────────────────────────────────────────────assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { headers: { 'X-OpenVibe-Signature': 'sha256=00', 'X-OpenVibe-Timestamp': '1', 'X-OpenVibe-Signature-V2': 't=1,v2=00' } }), 401, 'bad signature');assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { v1Only: true }), 401, 'v1 only (no v2 header): refused');@@@ -249,21 +299,44 @@ function sessionEvent(type, sessionId, revision, payload = {}) {await whip.handleWhipPost(req, res);return out;};++ const streamCount = () => d.prepare('SELECT COUNT(*) AS n FROM streams').get().n;++ const before = streamCount();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}`);++ assert.strictEqual(r.status, 401, `WHIP ${label} on an OpenRe slot is refused`);++ assert.ok(!/openre/i.test(JSON.stringify(r.body)), 'the refusal does not say OpenRe exists');}++ assert.strictEqual(streamCount(), before, 'a refused WHIP publish creates no stream row');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.++ // The switch landing between publishes: the next check refuses.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');++ assert.strictEqual((await publish(liveWhipKey)).status, 401, 'refused as soon as the slot is on OpenRe');d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 705").run();++ // The Go Live page's in-browser broadcaster (/ws/broadcast, role=broadcaster) is WebRTC ingest too.++ const broadcastServer = require('../server/streaming/broadcast-server');++ const connectBroadcaster = (streamId, token) => {++ const closed = [];++ const ws = { readyState: 1, on() {}, send() {}, ping() {}, close(code, reason) { closed.push([code, reason]); } };++ try {++ broadcastServer.handleConnection(ws, { url: `/ws/broadcast?streamId=${streamId}&role=broadcaster&token=${token}`, headers: {}, socket: {} });++ } catch { /* past the ingest checks: the SFU is not running here */ }++ broadcastServer.clients.delete(ws);++ broadcastServer.rooms.clear();++ return closed;++ };++ const browserOnOpenre = db.createStream({ user_id: 603, managed_stream_id: 704, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;++ const browserOnLive = db.createStream({ user_id: 602, managed_stream_id: 705, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;++ const browserPersonal = db.createStream({ user_id: 603, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;++ assert.deepStrictEqual(connectBroadcaster(browserOnOpenre, 'tok-603'), [[4003, 'Stream key not recognized']], 'browser broadcaster on an OpenRe slot is refused');++ assert.deepStrictEqual(connectBroadcaster(browserPersonal, 'tok-603'), [[4003, 'Stream key not recognized']], 'a slot-less stream of an owner with an OpenRe slot is refused');++ assert.deepStrictEqual(connectBroadcaster(browserOnLive, 'tok-602'), [], 'browser broadcaster on a Live slot is admitted');++const relay = require('../server/streaming/jsmpeg-relay');relay.nextVideoPort = 29710 + (process.pid % 1000) * 4;relay.nextAudioPort = relay.nextVideoPort + 1;@@@ -276,12 +349,11 @@ function sessionEvent(type, sessionId, revision, payload = {}) {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(openreCh.videoPort, 'jsmpegopenre706'), 404, 'JSMPEG video: an OpenRe slot key is refused');++ assert.strictEqual(await post(openreCh.audioPort, 'jsmpegopenre706'), 404, 'JSMPEG audio: an OpenRe slot key is refused');++ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 404, "JSMPEG: the owner's personal key is refused while a slot is on OpenRe");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;@@@ -290,6 +362,10 @@ function sessionEvent(type, sessionId, revision, payload = {}) {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');++ assert.deepStrictEqual(connectBroadcaster(browserOnOpenre, 'tok-603'), [], 'without OPENRE_URL the browser broadcaster is admitted again');++ assert.strictEqual(await post(openreCh.videoPort, 'jsmpegopenre706'), 200, 'without OPENRE_URL the JSMPEG slot is admitted again');++ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 200, 'without OPENRE_URL the personal JSMPEG key is admitted again');++ relay.closeAll();process.env.OPENRE_URL = stubUrl;require('../server/openre/openre-client')._reset();@@@ -300,7 +376,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {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']);++ if (statusRes.status === 200) assert.deepStrictEqual((await statusRes.json()).protocols, ['rtmp', 'webrtc', 'jsmpeg']);else assert.fail(`status route answered ${statusRes.status}`);statusServer.close();@@@ -317,5 +393,5 @@ function sessionEvent(type, sessionId, revision, payload = {}) {mirror._reset();server.close();stub.close();-- console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, WHIP/JSMPEG stay on Live, signed mirror (once, ordered), reconcile, rollback');++ console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP/WHIP/browser/JSMPEG refusal, OpenRe ingest URLs, signed mirror (once, ordered), reconcile, rollback');})().catch((err) => { console.error(err); process.exit(1); });
@@ -2,10 +2,11 @@
@@ -2,10 +2,11 @@
/**
/**
* The per-slot ingest switch: managed_streams.ingest_authority = 'live' (default) | 'openre'.
* The per-slot ingest switch: managed_streams.ingest_authority = 'live' (default) | 'openre'.
*
*
- * '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 and JSMPEG still publish through Live.
+ * 'openre' means OpenRe.Stream ingests this slot: Live's RTMP server, WHIP handler, browser
+ * broadcaster socket and JSMPEG relay refuse the slot's key (so a stream can never be ingested
+ * twice), the Go Live UI shows OpenRe's ingest URLs and rotates OpenRe's key, and the slot's
+ * sessions reach `streams` through the mirror (mirror.js). RTMP, WebRTC (WHIP) and JSMPEG move
+ * (OPENRE_PROTOCOLS): OpenRe serves WHIP on ingest.openre.stream and runs the JSMPEG worker.
*
*
* 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()).
@@ -14,7 +15,15 @@ const crypto = require('crypto');
@@ -14,7 +15,15 @@ const crypto = require('crypto');
const db = require('../db/database');
const db = require('../db/database');
const client = require('./openre-client');
const client = require('./openre-client');
-const OPENRE_PROTOCOLS = new Set(['rtmp']);
+const OPENRE_PROTOCOLS = new Set(['rtmp', 'webrtc', 'jsmpeg']);
+
+/** The OpenRe definition protocol a slot publishes with (OpenRe's own migrate-live mapping). */
+function openreProtocolOf(slot) {
+ const p = String((slot && slot.protocol) || '').toLowerCase();
+ if (p === 'webrtc' || p === 'whip') return 'webrtc';
+ if (p === 'jsmpeg') return 'jsmpeg';
+ return 'rtmp';
+}
function authorityOf(slot) {
function authorityOf(slot) {
if (!slot || slot.ingest_authority !== 'openre') return 'live';
if (!slot || slot.ingest_authority !== 'openre') return 'live';
@@ -32,9 +41,10 @@ function slotIsOpenre(id) {
@@ -32,9 +41,10 @@ function slotIsOpenre(id) {
}
}
/**
/**
- * 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.
+ * Live's publish handlers (RTMP, WHIP, browser broadcaster, JSMPEG) ask 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.
*/
*/
function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {
function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {
try {
try {
@@ -58,20 +68,33 @@ function serializeSlot(slot) {
@@ -58,20 +68,33 @@ function serializeSlot(slot) {
return { ...slot, stream_key: null, ingest_authority: 'openre', stream_key_managed_by: 'openre', openre_manage_url: client.manageUrl(slot.openre_stream_id) };
return { ...slot, stream_key: null, ingest_authority: 'openre', stream_key_managed_by: 'openre', openre_manage_url: client.manageUrl(slot.openre_stream_id) };
}
}
+/** OpenRe's ingest URLs from a stream or rotation answer (`ingest.rtmp.url`, `ingest.webrtc.whip_url`,
+ * `ingest.jsmpeg.url`; each only for the protocols the definition allows). The key goes after each:
+ * rtmp://…/live + key, <whip_url>/<key>, <jsmpeg_url>/<key>/<width>/<height>/. */
+function ingestUrls(ingest) {
+ const i = ingest || {};
+ return {
+ rtmp_url: i.rtmp ? i.rtmp.url : null,
+ whip_url: i.webrtc ? i.webrtc.whip_url || null : null,
+ jsmpeg_url: i.jsmpeg ? i.jsmpeg.url || null : null,
+ };
+}
+
/**
/**
- * What the Go Live UI shows for an OpenRe slot: OpenRe's RTMP URL and the hint of its current key
- * (the key itself is only ever shown once, by a rotation).
+ * What the Go Live UI shows for an OpenRe slot: OpenRe's ingest URLs and the hint of its current key
+ * (one key for every protocol; the key itself is only ever shown once, by a rotation).
*/
*/
async function ingestFor(slot, subject) {
async function ingestFor(slot, subject) {
const stream = slot.openre_stream_id ? await client.getStream(slot.openre_stream_id, { subject }) : await client.streamForSlot(slot.id);
const stream = slot.openre_stream_id ? await client.getStream(slot.openre_stream_id, { subject }) : await client.streamForSlot(slot.id);
- const rtmp = stream && stream.ingest && stream.ingest.rtmp;
+ const ingest = (stream && stream.ingest) || {};
+ const keyHint = ['rtmp', 'webrtc', 'jsmpeg'].map(p => ingest[p] && ingest[p].key_hint).find(Boolean);
return {
return {
ingest_authority: 'openre',
ingest_authority: 'openre',
stream_key_managed_by: 'openre',
stream_key_managed_by: 'openre',
openre_stream_id: stream ? stream.id : null,
openre_stream_id: stream ? stream.id : null,
- rtmp_url: rtmp ? rtmp.url : null,
- stream_key_hint: rtmp && rtmp.key_hint
- ? `Shown once: press Regenerate for a new key (current key ends in …${rtmp.key_hint})`
+ ...ingestUrls(ingest),
+ stream_key_hint: keyHint
+ ? `Shown once: press Regenerate for a new key (current key ends in …${keyHint})`
: 'Press Regenerate to get your OpenRe stream key',
: 'Press Regenerate to get your OpenRe stream key',
openre_manage_url: client.manageUrl(stream ? stream.id : slot.openre_stream_id),
openre_manage_url: client.manageUrl(stream ? stream.id : slot.openre_stream_id),
};
};
@@ -82,7 +105,7 @@ async function rotateFor(slot, subject) {
@@ -82,7 +105,7 @@ async function rotateFor(slot, subject) {
const streamId = slot.openre_stream_id || (await client.streamForSlot(slot.id) || {}).id;
const streamId = slot.openre_stream_id || (await client.streamForSlot(slot.id) || {}).id;
if (!streamId) throw new client.OpenReError('this slot has no OpenRe stream yet', 409);
if (!streamId) throw new client.OpenReError('this slot has no OpenRe stream yet', 409);
const r = await client.rotateKey(streamId, { subject });
const r = await client.rotateKey(streamId, { subject });
- return { stream_key: r.key.key, stream_key_managed_by: 'openre', rtmp_url: r.ingest && r.ingest.rtmp ? r.ingest.rtmp.url : null };
+ return { stream_key: r.key.key, stream_key_managed_by: 'openre', ...ingestUrls(r.ingest) };
}
}
function recordingModeFor(slot) {
function recordingModeFor(slot) {
@@ -94,7 +117,8 @@ function recordingModeFor(slot) {
@@ -94,7 +117,8 @@ function recordingModeFor(slot) {
/**
/**
* The admin switch (PUT /api/admin/openre/managed/:id/ingest-authority).
* The admin switch (PUT /api/admin/openre/managed/:id/ingest-authority).
* 'openre': needs OpenRe configured, the slot offline on Live, the owner's canonical subject;
* 'openre': needs OpenRe configured, the slot offline on Live, the owner's canonical subject;
- * finds or creates the OpenRe definition, then (one transaction) flips the slot and
+ * finds or creates the OpenRe definition (allowing the slot's protocol on it), then
+ * (one transaction) flips the slot and
* rotates Live's own key for it, so no key that existed before works anywhere.
* rotates Live's own key for it, so no key that existed before works anywhere.
* 'live': flips back (rollback). Live's key was rotated at the switch: the broadcaster
* 'live': flips back (rollback). Live's key was rotated at the switch: the broadcaster
* regenerates it on the Go Live page.
* regenerates it on the Go Live page.
@@ -111,9 +135,12 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
@@ -111,9 +135,12 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
if (slot.ingest_authority === 'openre') return { status: 200, body: { managed_stream_id: slot.id, ingest_authority: 'openre', openre_stream_id: slot.openre_stream_id, unchanged: true } };
if (slot.ingest_authority === 'openre') return { status: 200, body: { managed_stream_id: slot.id, ingest_authority: 'openre', openre_stream_id: slot.openre_stream_id, unchanged: true } };
const live = db.get('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1', [slot.id]);
const live = db.get('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1', [slot.id]);
if (live) return { status: 409, error: `Slot ${slot.id} is live on Live (stream ${live.id}); switch it in a maintenance window, while it is offline` };
if (live) return { status: 409, error: `Slot ${slot.id} is live on Live (stream ${live.id}); switch it in a maintenance window, while it is offline` };
+ const protocol = openreProtocolOf(slot);
const method = String(slot.streaming_method || '').toLowerCase();
const method = String(slot.streaming_method || '').toLowerCase();
- if (!force && slot.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(method)) {
- return { status: 409, error: `Slot ${slot.id} is a ${slot.protocol}${method ? `/${method}` : ''} slot; OpenRe only ingests RTMP today (pass force to switch anyway)` };
+ // WHIP encoders take OpenRe's WHIP URL and key; the Go Live page's in-browser broadcaster only
+ // publishes to Live's own SFU, so a browser slot moved now could not go live from the page.
+ if (!force && protocol === 'webrtc' && !['whip', 'obs'].includes(method)) {
+ return { status: 409, error: `Slot ${slot.id} is a ${slot.protocol}${method ? `/${method}` : ''} slot; the Go Live page cannot publish to OpenRe yet, only WHIP encoders can (pass force to switch anyway)` };
}
}
const subject = require('../auth/identity-sync').subjectOf(slot.user_id);
const subject = require('../auth/identity-sync').subjectOf(slot.user_id);
if (!subject) return { status: 409, error: 'The owner has no canonical subject yet (they need to sign in to Live once)' };
if (!subject) return { status: 409, error: 'The owner has no canonical subject yet (they need to sign in to Live once)' };
@@ -121,7 +148,10 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
@@ -121,7 +148,10 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
if (!stream) {
if (!stream) {
let visibility = 'public';
let visibility = 'public';
try { visibility = db.resolveStreamVodVisibility({ user_id: slot.user_id, managed_stream_id: slot.id }); } catch { /* public */ }
try { visibility = db.resolveStreamVodVisibility({ user_id: slot.user_id, managed_stream_id: slot.id }); } catch { /* public */ }
- stream = await client.createStreamForSlot(slot, { subject, recordingMode: recordingModeFor(slot), recordingVisibility: visibility });
+ stream = await client.createStreamForSlot(slot, { subject, protocols: [protocol], recordingMode: recordingModeFor(slot), recordingVisibility: visibility });
+ } else if (!(stream.protocols || ['rtmp']).includes(protocol)) {
+ // An existing definition (OpenRe's own default is RTMP only) would refuse the slot's encoder.
+ stream = await client.updateStream(stream.id, { protocols: [...(stream.protocols || ['rtmp']), protocol] }, { subject });
}
}
const newLiveKey = crypto.randomBytes(20).toString('hex');
const newLiveKey = crypto.randomBytes(20).toString('hex');
const flipped = db.getDb().transaction(() => {
const flipped = db.getDb().transaction(() => {
@@ -137,7 +167,7 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
@@ -137,7 +167,7 @@ async function setAuthority(slotId, authority, { force = false } = {}) {
managed_stream_id: slot.id,
managed_stream_id: slot.id,
ingest_authority: 'openre',
ingest_authority: 'openre',
openre_stream_id: stream.id,
openre_stream_id: stream.id,
- rtmp_url: stream.ingest && stream.ingest.rtmp ? stream.ingest.rtmp.url : null,
+ ...ingestUrls(stream.ingest),
live_key_rotated: true,
live_key_rotated: true,
next: 'The broadcaster presses Regenerate on the Go Live page (OpenRe issues the key) and pastes the new server and key into OBS.',
next: 'The broadcaster presses Regenerate on the Go Live page (OpenRe issues the key) and pastes the new server and key into OBS.',
},
},
@@ -80,9 +80,10 @@ const streamForSlot = (managedStreamId) => wrap('stream lookup', () => openre().
@@ -80,9 +80,10 @@ const streamForSlot = (managedStreamId) => wrap('stream lookup', () => openre().
/** Create the definition for a slot, owned by the streamer's canonical subject. The key it
/** Create the definition for a slot, owned by the streamer's canonical subject. The key it
* returns is dropped unseen: the streamer gets a usable key by rotating. */
* returns is dropped unseen: the streamer gets a usable key by rotating. */
-async function createStreamForSlot(slot, { subject, recordingMode, recordingVisibility }) {
+async function createStreamForSlot(slot, { subject, protocols, recordingMode, recordingVisibility }) {
const r = await wrap('stream create', () => openre().streams.create({
const r = await wrap('stream create', () => openre().streams.create({
title: slot.title || 'Stream',
title: slot.title || 'Stream',
+ ...(protocols ? { protocols } : {}),
recording_mode: recordingMode,
recording_mode: recordingMode,
recording_visibility: recordingVisibility,
recording_visibility: recordingVisibility,
mirror_to_live: true,
mirror_to_live: true,
@@ -95,6 +96,7 @@ async function createStreamForSlot(slot, { subject, recordingMode, recordingVisi
@@ -95,6 +96,7 @@ async function createStreamForSlot(slot, { subject, recordingMode, recordingVisi
}
}
const rotateKey = (streamId, { subject, graceSeconds = 0 } = {}) => wrap('key rotate', () => openre().streams.rotateKey(streamId, { subject, graceSeconds }));
const rotateKey = (streamId, { subject, graceSeconds = 0 } = {}) => wrap('key rotate', () => openre().streams.rotateKey(streamId, { subject, graceSeconds }));
+const updateStream = (streamId, patch, { subject } = {}) => wrap('stream update', () => openre().streams.update(streamId, patch, { subject }));
const getStream = (streamId, { subject } = {}) => wrap('stream read', () => openre().streams.get(streamId, { subject }));
const getStream = (streamId, { subject } = {}) => wrap('stream read', () => openre().streams.get(streamId, { subject }));
const getSession = (sessionId) => wrap('session read', () => openre().sessions.get(sessionId));
const getSession = (sessionId) => wrap('session read', () => openre().sessions.get(sessionId));
/** Playback descriptor for a session (the SDK caches it 10 s: the FLV proxy asks on every viewer connect). */
/** Playback descriptor for a session (the SDK caches it 10 s: the FLV proxy asks on every viewer connect). */
@@ -106,4 +108,4 @@ function manageUrl(streamId) {
@@ -106,4 +108,4 @@ function manageUrl(streamId) {
function _reset() { sdk = null; }
function _reset() { sdk = null; }
-module.exports = { enabled, settings, request, streamForSlot, createStreamForSlot, rotateKey, getStream, getSession, playback, manageUrl, OpenReError, _reset };
+module.exports = { enabled, settings, request, streamForSlot, createStreamForSlot, updateStream, rotateKey, getStream, getSession, playback, manageUrl, OpenReError, _reset };
@@ -235,6 +235,15 @@ class BroadcastServer extends EventEmitter {
@@ -235,6 +235,15 @@ class BroadcastServer extends EventEmitter {
ws.close(4003, 'Not your stream');
ws.close(4003, 'Not your stream');
return;
return;
}
}
+ // OpenRe.Stream ingests this slot (or, for a slot-less stream, one of the owner's slots):
+ // Live's SFU never takes a second publisher for it. See server/openre/authority.js.
+ const openre = require('../openre/authority');
+ const managedStream = openre.slotById(stream.managed_stream_id);
+ if (openre.refusesLiveIngest({ managedStream, user: managedStream ? null : user, protocol: 'webrtc' })) {
+ console.warn(`[Broadcast] Broadcaster refused for stream ${streamId}: ${managedStream ? `slot ${managedStream.id}` : `personal stream of ${user.username}`} is ingested by OpenRe`);
+ ws.close(4003, 'Stream key not recognized');
+ return;
+ }
}
}
// Viewers: only for a stream that exists and is live, and only a handful of sockets per address.
// Viewers: only for a stream that exists and is live, and only a handful of sockets per address.
@@ -2270,6 +2270,24 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {
@@ -2270,6 +2270,24 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {
const hostname = config.host === '0.0.0.0' ? req.hostname : config.host;
const hostname = config.host === '0.0.0.0' ? req.hostname : config.host;
+ if (openreAuthority.OPENRE_PROTOCOLS.has(stream.protocol) && openreAuthority.slotIsOpenre(stream.managed_stream_id)) {
+ // OpenRe ingests this slot: its server URL for the slot's protocol, the hint of its key
+ // (never the key). No Live relay channel, room or recorder is set up for it.
+ const slot = openreAuthority.slotById(stream.managed_stream_id);
+ const ingest = await openreAuthority.ingestFor(slot, require('../auth/identity-sync').subjectOf(slot.user_id)).catch(() => ({}));
+ const serverUrl = stream.protocol === 'webrtc' ? { whipUrl: ingest.whip_url || null }
+ : stream.protocol === 'jsmpeg' ? { jsmpegUrl: ingest.jsmpeg_url || null }
+ : { rtmpUrl: ingest.rtmp_url || null };
+ endpoint = {
+ ...serverUrl,
+ streamKey: null,
+ streamKeyHint: ingest.stream_key_hint || 'Press Regenerate to get your OpenRe stream key',
+ keyManagedBy: 'openre',
+ ...(stream.protocol === 'rtmp' ? { flvUrl: `/api/streams/rtmp-proxy/${stream.id}.flv` } : {}),
+ };
+ return res.json({ endpoint, stream_key: null });
+ }
+
if (stream.protocol === 'jsmpeg') {
if (stream.protocol === 'jsmpeg') {
endpoint = jsmpegRelay.getChannelInfo(msKey) || jsmpegRelay.createChannel(msKey);
endpoint = jsmpegRelay.getChannelInfo(msKey) || jsmpegRelay.createChannel(msKey);
@@ -2305,18 +2323,6 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {
@@ -2305,18 +2323,6 @@ router.get('/:id/endpoint', requireAuth, async (req, res) => {
whipUrlSource,
whipUrlSource,
...(whipUrlWarning ? { whipUrlWarning } : {}),
...(whipUrlWarning ? { whipUrlWarning } : {}),
};
};
- } else if (stream.protocol === 'rtmp' && openreAuthority.slotIsOpenre(stream.managed_stream_id)) {
- // OpenRe ingests this slot: its server URL, the hint of its key (never the key).
- const slot = openreAuthority.slotById(stream.managed_stream_id);
- const ingest = await openreAuthority.ingestFor(slot, require('../auth/identity-sync').subjectOf(slot.user_id)).catch(() => ({}));
- endpoint = {
- rtmpUrl: ingest.rtmp_url || null,
- streamKey: null,
- streamKeyHint: ingest.stream_key_hint || 'Press Regenerate to get your OpenRe stream key',
- keyManagedBy: 'openre',
- flvUrl: `/api/streams/rtmp-proxy/${stream.id}.flv`,
- };
- return res.json({ endpoint, stream_key: null });
} else if (stream.protocol === 'rtmp') {
} else if (stream.protocol === 'rtmp') {
const rtmpHost = config.rtmp.host || (() => {
const rtmpHost = config.rtmp.host || (() => {
try { return new URL(config.baseUrl).hostname; } catch { return hostname; }
try { return new URL(config.baseUrl).hostname; } catch { return hostname; }
@@ -1,9 +1,10 @@
@@ -1,9 +1,10 @@
'use strict';
'use strict';
// OpenRe ingest switch (roadmap Wave 7, ADR-009). With OPENRE_URL unset, or a slot on 'live' (the
// OpenRe ingest switch (roadmap Wave 7, ADR-009). With OPENRE_URL unset, or a slot on 'live' (the
-// default), nothing changes. With a slot switched to 'openre': Live's RTMP refuses its key, the
-// Go Live helpers show OpenRe's URL and rotate OpenRe's key, and OpenRe session events (signed
-// Events deliveries, applied once, in revision order) become ordinary `streams` rows.
+// default), nothing changes. With a slot switched to 'openre': Live's RTMP, WHIP, browser
+// broadcaster and JSMPEG ingest refuse its key, the Go Live helpers show OpenRe's URLs and rotate
+// OpenRe's key, and OpenRe session events (signed Events deliveries, applied once, in revision
+// order) become ordinary `streams` rows.
const assert = require('assert');
const assert = require('assert');
const crypto = require('crypto');
const crypto = require('crypto');
@@ -21,7 +22,14 @@ const SUBJECT = 'usr_01J0000000000000000000000Q';
@@ -21,7 +22,14 @@ const SUBJECT = 'usr_01J0000000000000000000000Q';
const SES = (n) => `ses_01J00000000000000000000${String(n).padStart(3, '0')}`.slice(0, 30);
const SES = (n) => `ses_01J00000000000000000000${String(n).padStart(3, '0')}`.slice(0, 30);
const openreCalls = [];
const openreCalls = [];
let openreSessions = {};
let openreSessions = {};
-let openreStreams = [];
+/** OpenRe definitions by id: { id, ref, protocols }. */
+const openreDefs = {};
+const ingestOf = (protocols) => ({
+ ...(protocols.includes('rtmp') ? { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live', key_hint: 'WXYZ' } } : {}),
+ ...(protocols.includes('webrtc') ? { webrtc: { whip_url: 'https://ingest.openre.stream/whip', signaling_url: 'wss://ingest.openre.stream/b', key_hint: 'WXYZ' } } : {}),
+ ...(protocols.includes('jsmpeg') ? { jsmpeg: { url: 'http://ingest.openre.stream:8081', key_hint: 'WXYZ' } } : {}),
+});
+const view = (def) => ({ id: def.id, protocols: def.protocols, ingest: ingestOf(def.protocols) });
const stub = http.createServer((req, res) => {
const stub = http.createServer((req, res) => {
let body = '';
let body = '';
@@ -37,11 +45,25 @@ const stub = http.createServer((req, res) => {
@@ -37,11 +45,25 @@ const stub = http.createServer((req, res) => {
if (!req.url.startsWith('/api/v1/')) return send(404, {});
if (!req.url.startsWith('/api/v1/')) return send(404, {});
assert.strictEqual(req.headers.authorization, 'Bearer svc-live');
assert.strictEqual(req.headers.authorization, 'Bearer svc-live');
openreCalls.push({ method: req.method, url: req.url, subject: req.headers['x-ov-subject'] || null, body: body ? JSON.parse(body) : null });
openreCalls.push({ method: req.method, url: req.url, subject: req.headers['x-ov-subject'] || null, body: body ? JSON.parse(body) : null });
- const stream = (id) => ({ id, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live', key_hint: 'WXYZ' } } });
- if (req.method === 'GET' && req.url.startsWith('/api/v1/streams?external_ref=')) return send(200, { streams: openreStreams });
- if (req.method === 'POST' && req.url === '/api/v1/streams') { openreStreams = [stream('std_01J0000000000000000000000S')]; return send(201, { stream: openreStreams[0], key: { key: 'ork_dropped' } }); }
- if (req.method === 'GET' && req.url.startsWith('/api/v1/streams/std_')) return send(200, { stream: stream(req.url.split('/').pop()) });
- if (req.method === 'POST' && req.url.endsWith('/keys/rotate')) return send(200, { key: { key: 'ork_newkey_shown_once' }, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live' } } });
+ const defOf = (u) => openreDefs[u.split('/')[4]] || { id: u.split('/')[4], protocols: ['rtmp'] };
+ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams?external_ref=')) {
+ const ref = new URL(req.url, 'http://x').searchParams.get('external_ref');
+ return send(200, { streams: Object.values(openreDefs).filter(x => x.ref === ref).map(view) });
+ }
+ if (req.method === 'POST' && req.url === '/api/v1/streams') {
+ const b = JSON.parse(body);
+ const slotId = b.external_refs[0].id;
+ const def = { id: slotId === '701' ? 'std_01J0000000000000000000000S' : `std_slot${slotId}`, ref: `live:managed_stream:${slotId}`, protocols: b.protocols || ['rtmp'] };
+ openreDefs[def.id] = def;
+ return send(201, { stream: view(def), key: { key: 'ork_dropped' } });
+ }
+ if (req.method === 'PATCH' && req.url.startsWith('/api/v1/streams/std_')) {
+ const def = defOf(req.url);
+ def.protocols = JSON.parse(body).protocols;
+ return send(200, { stream: view(def) });
+ }
+ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams/std_')) return send(200, { stream: view(defOf(req.url)) });
+ if (req.method === 'POST' && req.url.endsWith('/keys/rotate')) return send(200, { key: { key: 'ork_newkey_shown_once' }, ingest: ingestOf(defOf(req.url).protocols) });
const m = /^\/api\/v1\/sessions\/(ses_[0-9A-Z]+)(\/playback)?$/.exec(req.url);
const m = /^\/api\/v1\/sessions\/(ses_[0-9A-Z]+)(\/playback)?$/.exec(req.url);
if (m && openreSessions[m[1]]) {
if (m && openreSessions[m[1]]) {
if (m[2]) return send(200, { playback: { flv: { internal_url: `http://127.0.0.1:19999/live/${m[1]}.flv` } } });
if (m[2]) return send(200, { playback: { flv: { internal_url: `http://127.0.0.1:19999/live/${m[1]}.flv` } } });
@@ -82,6 +104,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -82,6 +104,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
(async () => {
(async () => {
const db = require('../server/db/database');
const db = require('../server/db/database');
db.initDb();
db.initDb();
+ // broadcast-server takes authenticateWs at load (the mirror may load it first): 'tok-<user id>' signs in.
+ require('../server/auth/auth').authenticateWs = (token) => (/^tok-\d+$/.test(token || '') ? db.getUserById(Number(token.slice(4))) || null : null);
const d = db.getDb();
const d = db.getDb();
d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (601, 'caster', 'Caster', 'x', 'personalkey601')").run();
d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (601, 'caster', 'Caster', 'x', 'personalkey601')").run();
d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (602, 'nosub', 'NoSub', 'x')").run();
d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (602, 'nosub', 'NoSub', 'x')").run();
@@ -89,6 +113,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -89,6 +113,8 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (701, 601, 'OBS slot', 'rtmp', 'livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'obs')").run();
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (701, 601, 'OBS slot', 'rtmp', 'livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'obs')").run();
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (702, 601, 'Browser slot', 'webrtc', 'livekey702aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (702, 601, 'Browser slot', 'webrtc', 'livekey702aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (703, 602, 'No subject', 'rtmp', 'livekey703aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (703, 602, 'No subject', 'rtmp', 'livekey703aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (708, 601, 'WHIP encoder slot', 'webrtc', 'livekey708aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'whip')").run();
+ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (709, 601, 'JSMPEG slot', 'jsmpeg', 'livekey709aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'cli')").run();
const authority = require('../server/openre/authority');
const authority = require('../server/openre/authority');
const mirror = require('../server/openre/mirror');
const mirror = require('../server/openre/mirror');
@@ -134,7 +160,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -134,7 +160,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams').get().n, 0, 'a slot on live is never mirrored');
assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams').get().n, 0, 'a slot on live is never mirrored');
// ── The switch ───────────────────────────────────────────
// ── The switch ───────────────────────────────────────────
- assert.strictEqual((await authority.setAuthority(702, 'openre')).status, 409, 'a browser slot needs force');
+ assert.match((await authority.setAuthority(702, 'openre')).error || '', /Go Live page cannot publish to OpenRe/, 'a browser slot needs force');
assert.strictEqual((await authority.setAuthority(703, 'openre')).status, 409, 'no canonical subject');
assert.strictEqual((await authority.setAuthority(703, 'openre')).status, 409, 'no canonical subject');
const liveRow = db.createStream({ user_id: 601, managed_stream_id: 701, title: 'on Live', protocol: 'rtmp' });
const liveRow = db.createStream({ user_id: 601, managed_stream_id: 701, title: 'on Live', protocol: 'rtmp' });
assert.strictEqual((await authority.setAuthority(701, 'openre')).status, 409, 'refused while live on Live');
assert.strictEqual((await authority.setAuthority(701, 'openre')).status, 409, 'refused while live on Live');
@@ -146,16 +172,17 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -146,16 +172,17 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
assert.strictEqual(created.subject, SUBJECT, 'created for the owner subject');
assert.strictEqual(created.subject, SUBJECT, 'created for the owner subject');
assert.deepStrictEqual(created.body.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:701', 'live:user:601']);
assert.deepStrictEqual(created.body.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:701', 'live:user:601']);
assert.strictEqual(created.body.mirror_to_live, true);
assert.strictEqual(created.body.mirror_to_live, true);
+ assert.deepStrictEqual(created.body.protocols, ['rtmp'], "the definition allows the slot's protocol");
+ assert.strictEqual(flip.body.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');
const switched = db.getManagedStreamById(701);
const switched = db.getManagedStreamById(701);
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 the protocols OpenRe carries are refused');
- assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp']);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'srt' }), false, 'only the protocols OpenRe carries are refused');
+ assert.deepStrictEqual([...authority.OPENRE_PROTOCOLS], ['rtmp', 'webrtc', 'jsmpeg']);
for (const protocol of ['rtmp', 'webrtc', 'jsmpeg']) {
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: switched, protocol }), true, `${protocol}: the slot key is refused`);
+ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 }, protocol }), true, `${protocol}: the owner's personal key is refused`);
assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(703), protocol }), false, `${protocol}: a live slot is admitted`);
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 }, protocol }), false, `${protocol}: a personal key without an OpenRe slot is admitted`);
}
}
@@ -166,11 +193,34 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -166,11 +193,34 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
assert.strictEqual(shown.stream_key_managed_by, 'openre');
assert.strictEqual(shown.stream_key_managed_by, 'openre');
const ingest = await authority.ingestFor(switched, SUBJECT);
const ingest = await authority.ingestFor(switched, SUBJECT);
assert.strictEqual(ingest.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');
assert.strictEqual(ingest.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');
+ assert.strictEqual(ingest.whip_url, null);
+ assert.strictEqual(ingest.jsmpeg_url, null);
assert.match(ingest.stream_key_hint, /…WXYZ/);
assert.match(ingest.stream_key_hint, /…WXYZ/);
const rotated = await authority.rotateFor(switched, SUBJECT);
const rotated = await authority.rotateFor(switched, SUBJECT);
assert.strictEqual(rotated.stream_key, 'ork_newkey_shown_once');
assert.strictEqual(rotated.stream_key, 'ork_newkey_shown_once');
+ assert.strictEqual(rotated.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');
assert.strictEqual(openreCalls.filter(c => c.url.endsWith('/keys/rotate')).pop().subject, SUBJECT);
assert.strictEqual(openreCalls.filter(c => c.url.endsWith('/keys/rotate')).pop().subject, SUBJECT);
+ // A WHIP encoder slot: its definition is created for WebRTC, the Go Live helpers give OpenRe's WHIP URL.
+ const whipFlip = await authority.setAuthority(708, 'openre');
+ assert.strictEqual(whipFlip.status, 200, JSON.stringify(whipFlip));
+ assert.deepStrictEqual(openreCalls.filter(c => c.method === 'POST' && c.url === '/api/v1/streams').pop().body.protocols, ['webrtc']);
+ assert.strictEqual(whipFlip.body.whip_url, 'https://ingest.openre.stream/whip');
+ const whipIngest = await authority.ingestFor(db.getManagedStreamById(708), SUBJECT);
+ assert.deepStrictEqual([whipIngest.rtmp_url, whipIngest.whip_url, whipIngest.jsmpeg_url], [null, 'https://ingest.openre.stream/whip', null]);
+ assert.match(whipIngest.stream_key_hint, /…WXYZ/);
+ assert.strictEqual((await authority.rotateFor(db.getManagedStreamById(708), SUBJECT)).whip_url, 'https://ingest.openre.stream/whip');
+ // A JSMPEG slot whose OpenRe definition already exists (RTMP only, OpenRe's default): JSMPEG is added to it.
+ openreDefs.std_slot709 = { id: 'std_slot709', ref: 'live:managed_stream:709', protocols: ['rtmp'] };
+ const jsmpegFlip = await authority.setAuthority(709, 'openre');
+ assert.strictEqual(jsmpegFlip.status, 200, JSON.stringify(jsmpegFlip));
+ const patched = openreCalls.filter(c => c.method === 'PATCH').pop();
+ assert.strictEqual(patched.url, '/api/v1/streams/std_slot709');
+ assert.strictEqual(patched.subject, SUBJECT);
+ assert.deepStrictEqual(patched.body, { protocols: ['rtmp', 'jsmpeg'] });
+ assert.strictEqual(jsmpegFlip.body.jsmpeg_url, 'http://ingest.openre.stream:8081');
+ assert.strictEqual((await authority.ingestFor(db.getManagedStreamById(709), SUBJECT)).jsmpeg_url, 'http://ingest.openre.stream:8081');
+
// ── Mirror ────────────────────────────────────────────────
// ── Mirror ────────────────────────────────────────────────
assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { headers: { 'X-OpenVibe-Signature': 'sha256=00', 'X-OpenVibe-Timestamp': '1', 'X-OpenVibe-Signature-V2': 't=1,v2=00' } }), 401, 'bad signature');
assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { headers: { 'X-OpenVibe-Signature': 'sha256=00', 'X-OpenVibe-Timestamp': '1', 'X-OpenVibe-Signature-V2': 't=1,v2=00' } }), 401, 'bad signature');
assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { v1Only: true }), 401, 'v1 only (no v2 header): refused');
assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { v1Only: true }), 401, 'v1 only (no v2 header): refused');
@@ -249,21 +299,44 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -249,21 +299,44 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
await whip.handleWhipPost(req, res);
await whip.handleWhipPost(req, res);
return out;
return out;
};
};
+ const streamCount = () => d.prepare('SELECT COUNT(*) AS n FROM streams').get().n;
+ const before = streamCount();
for (const [label, call] of [
for (const [label, call] of [
['key in the path', () => publish(openReWhipKey)],
['key in the path', () => publish(openReWhipKey)],
['slot id + key', () => publish(704, { key: openReWhipKey })],
['slot id + key', () => publish(704, { key: openReWhipKey })],
]) {
]) {
const r = await call();
const r = await call();
- assert.ok([200, 201, 503].includes(r.status), `WHIP ${label} stays on Live: ${r.status}`);
+ assert.strictEqual(r.status, 401, `WHIP ${label} on an OpenRe slot is refused`);
+ assert.ok(!/openre/i.test(JSON.stringify(r.body)), 'the refusal does not say OpenRe exists');
}
}
+ assert.strictEqual(streamCount(), before, 'a refused WHIP publish creates no stream row');
const admitted = await publish(liveWhipKey);
const admitted = await publish(liveWhipKey);
assert.notStrictEqual(admitted.status, 401, 'a Live slot is still admitted (here it reaches the SFU check)');
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}`);
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.
+ // The switch landing between publishes: the next check refuses.
d.prepare("UPDATE managed_streams SET ingest_authority = 'openre' WHERE id = 705").run();
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');
+ assert.strictEqual((await publish(liveWhipKey)).status, 401, 'refused as soon as the slot is on OpenRe');
d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 705").run();
d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 705").run();
+ // The Go Live page's in-browser broadcaster (/ws/broadcast, role=broadcaster) is WebRTC ingest too.
+ const broadcastServer = require('../server/streaming/broadcast-server');
+ const connectBroadcaster = (streamId, token) => {
+ const closed = [];
+ const ws = { readyState: 1, on() {}, send() {}, ping() {}, close(code, reason) { closed.push([code, reason]); } };
+ try {
+ broadcastServer.handleConnection(ws, { url: `/ws/broadcast?streamId=${streamId}&role=broadcaster&token=${token}`, headers: {}, socket: {} });
+ } catch { /* past the ingest checks: the SFU is not running here */ }
+ broadcastServer.clients.delete(ws);
+ broadcastServer.rooms.clear();
+ return closed;
+ };
+ const browserOnOpenre = db.createStream({ user_id: 603, managed_stream_id: 704, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;
+ const browserOnLive = db.createStream({ user_id: 602, managed_stream_id: 705, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;
+ const browserPersonal = db.createStream({ user_id: 603, title: 'browser', protocol: 'webrtc' }).lastInsertRowid;
+ assert.deepStrictEqual(connectBroadcaster(browserOnOpenre, 'tok-603'), [[4003, 'Stream key not recognized']], 'browser broadcaster on an OpenRe slot is refused');
+ assert.deepStrictEqual(connectBroadcaster(browserPersonal, 'tok-603'), [[4003, 'Stream key not recognized']], 'a slot-less stream of an owner with an OpenRe slot is refused');
+ assert.deepStrictEqual(connectBroadcaster(browserOnLive, 'tok-602'), [], 'browser broadcaster on a Live slot is admitted');
+
const relay = require('../server/streaming/jsmpeg-relay');
const relay = require('../server/streaming/jsmpeg-relay');
relay.nextVideoPort = 29710 + (process.pid % 1000) * 4;
relay.nextVideoPort = 29710 + (process.pid % 1000) * 4;
relay.nextAudioPort = relay.nextVideoPort + 1;
relay.nextAudioPort = relay.nextVideoPort + 1;
@@ -276,12 +349,11 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -276,12 +349,11 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
const liveCh = relay.createChannel('jsmpeglive707');
const liveCh = relay.createChannel('jsmpeglive707');
const personalCh = relay.createChannel('personalkey601');
const personalCh = relay.createChannel('personalkey601');
await new Promise(r => setTimeout(r, 200));
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(openreCh.videoPort, 'jsmpegopenre706'), 404, 'JSMPEG video: an OpenRe slot key is refused');
+ assert.strictEqual(await post(openreCh.audioPort, 'jsmpegopenre706'), 404, 'JSMPEG audio: an OpenRe slot key is refused');
+ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 404, "JSMPEG: the owner's personal key is refused while a slot is on OpenRe");
assert.strictEqual(await post(liveCh.videoPort, 'jsmpeglive707'), 200, 'JSMPEG: a Live slot is admitted');
assert.strictEqual(await post(liveCh.videoPort, 'jsmpeglive707'), 200, 'JSMPEG: a Live slot is admitted');
assert.strictEqual(await post(liveCh.audioPort, 'jsmpeglive707'), 200);
assert.strictEqual(await post(liveCh.audioPort, 'jsmpeglive707'), 200);
- relay.closeAll();
// OPENRE_URL unset: nothing is refused, for any protocol.
// OPENRE_URL unset: nothing is refused, for any protocol.
delete process.env.OPENRE_URL;
delete process.env.OPENRE_URL;
@@ -290,6 +362,10 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -290,6 +362,10 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 603 }, 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');
assert.notStrictEqual((await publish(openReWhipKey)).status, 401, 'without OPENRE_URL the WHIP slot is admitted again');
+ assert.deepStrictEqual(connectBroadcaster(browserOnOpenre, 'tok-603'), [], 'without OPENRE_URL the browser broadcaster is admitted again');
+ assert.strictEqual(await post(openreCh.videoPort, 'jsmpegopenre706'), 200, 'without OPENRE_URL the JSMPEG slot is admitted again');
+ assert.strictEqual(await post(personalCh.videoPort, 'personalkey601'), 200, 'without OPENRE_URL the personal JSMPEG key is admitted again');
+ relay.closeAll();
process.env.OPENRE_URL = stubUrl;
process.env.OPENRE_URL = stubUrl;
require('../server/openre/openre-client')._reset();
require('../server/openre/openre-client')._reset();
@@ -300,7 +376,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -300,7 +376,7 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
statusApp.use('/api/admin/openre', require('../server/openre/routes'));
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 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`);
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']);
+ if (statusRes.status === 200) assert.deepStrictEqual((await statusRes.json()).protocols, ['rtmp', 'webrtc', 'jsmpeg']);
else assert.fail(`status route answered ${statusRes.status}`);
else assert.fail(`status route answered ${statusRes.status}`);
statusServer.close();
statusServer.close();
@@ -317,5 +393,5 @@ function sessionEvent(type, sessionId, revision, payload = {}) {
@@ -317,5 +393,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, WHIP/JSMPEG stay on Live, signed mirror (once, ordered), reconcile, rollback');
+ console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP/WHIP/browser/JSMPEG refusal, OpenRe ingest URLs, signed mirror (once, ordered), reconcile, rollback');
})().catch((err) => { console.error(err); process.exit(1); });
})().catch((err) => { console.error(err); process.exit(1); });