openvibers/openrestream Commits
Restart ICE, check If-Match and Bearer on WHIP PATCH and DELETE (#5)
Changes
@@@ -250,7 +250,7 @@ one slot. Status on 2026-09-23: OpenRe is deployed (`655b98a10aaa`) and DNS is d### WebRTC (worker ready)--`workers/webrtc.js` owns the whole WebRTC stack (decision 1: mediasoup is single-process, so ingest and consumption cannot be split): WHIP (RFC 9725) admission of `POST /whip/<key>` (Bearer optional) with `PATCH /whip/session/<id>` and `DELETE`, one mediasoup Router per session on UDP 10200–10300, the `/b/<key>` broadcaster and `/w/<id>` viewer signaling endpoints Live's broadcast/watch pages speak, a loopback RTP/SDP egress the restream worker and Media's RTP recorder consume, and live thumbnails. It needs `mediasoup` and `sdp-transform` installed (they are dependencies) and, in production, `MEDIASOUP_ANNOUNCED_IP`. docs/cutover.md has the WebRTC procedure (W-A once, then W-B1–W-B6 per slot). A webrtc session's playback descriptor is `webrtc: { signaling_url, announced_ip }`.++`workers/webrtc.js` owns the whole WebRTC stack (decision 1: mediasoup is single-process, so ingest and consumption cannot be split): WHIP (RFC 9725) admission of `POST /whip/<key>` (Bearer optional) with `PATCH /whip/session/<id>` (trickle ICE is acknowledged and discarded, since the transport is ICE-lite; new ICE credentials restart ICE: `200`, the server's new credentials and a new `ETag`; `If-Match` is checked) and `DELETE` (`404` for an unknown resource; both need the POST's Bearer again if it sent one), one mediasoup Router per session on UDP 10200–10300, the `/b/<key>` broadcaster and `/w/<id>` viewer signaling endpoints Live's broadcast/watch pages speak, a loopback RTP/SDP egress the restream worker and Media's RTP recorder consume, and live thumbnails. It needs `mediasoup` and `sdp-transform` installed (they are dependencies) and, in production, `MEDIASOUP_ANNOUNCED_IP`. docs/cutover.md has the WebRTC procedure (W-A once, then W-B1–W-B6 per slot). A webrtc session's playback descriptor is `webrtc: { signaling_url, announced_ip }`.## Acceptance@@@ -414,8 +414,9 @@ done **once** (the worker generation and its ports), then per slot like every pr`webrtc` is in `OPENRE_PROTOCOLS` (brief §4.5).- **B4 Broadcaster:** regenerate the key, then either- **WHIP:** POST an SDP offer to `https://ingest.openre.stream/whip/<key>` (Bearer optional, must-- match) → `201` + the answer + `Location: /whip/session/<id>`; trickle with `PATCH`, end with-- `DELETE`. The session goes live when ICE connects.++ match) → `201` + the answer + `Location: /whip/session/<id>` + an `ETag`; trickle or restart ICE++ with `PATCH` (`application/trickle-ice-sdpfrag`, `If-Match`), end with `DELETE` (both need the++ POST's Bearer again if it sent one). The session goes live when ICE connects.- **Browser:** connect `ws(s)://ingest.openre.stream/b/<key>` and run the mediasoup-client`sfu-get-capabilities → sfu-create-transport → sfu-connect-transport → sfu-produce` flow; thesession goes live on the first video producer.@@@ -126,15 +126,26 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives itconst offer = await pc.createOffer();await pc.setLocalDescription(offer);-- const res = await httpReq(port, 'POST', `/whip/${keyA}`, { headers: { 'content-type': 'application/sdp' }, body: pc.localDescription.sdp });++ // Trickle-only, as a browser behind NAT publishes: the offer carries no candidates, they follow++ // in a PATCH. The transport is ICE-lite, so ICE must connect from the client's checks alone.++ const fullOffer = pc.localDescription.sdp;++ const candidateLines = fullOffer.split('\r\n').filter((l) => l.startsWith('a=candidate:'));++ assert.ok(candidateLines.length > 0, 'the client gathered candidates to trickle');++ const trickleOffer = fullOffer.split('\r\n').filter((l) => !l.startsWith('a=candidate:') && l !== 'a=end-of-candidates').join('\r\n');++ const res = await httpReq(port, 'POST', `/whip/${keyA}`, { headers: { 'content-type': 'application/sdp' }, body: trickleOffer });assert.strictEqual(res.status, 201, `WHIP answer (${res.status}): ${res.body.slice(0, 200)}`);assert.match(res.body, /a=recvonly/, 'the answer is a recvonly SDP');assert.match(res.body, /a=ice-lite/);assert.match(res.headers.location, /\/whip\/session\/[0-9a-f]{16,}$/, 'Location points at the session resource');assert.match(String(res.headers.etag), /^"/, 'an ETag is returned');++ const resourceId = res.headers.location.split('/').pop();++ const ufrag = /a=ice-ufrag:(\S+)/.exec(fullOffer)[1];++ const trickle = await httpReq(port, 'PATCH', `/whip/session/${resourceId}`, { headers: { 'content-type': 'application/trickle-ice-sdpfrag', 'if-match': res.headers.etag }, body: `a=ice-ufrag:${ufrag}\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=mid:0\r\n${candidateLines.join('\r\n')}\r\na=end-of-candidates\r\n` });++ assert.strictEqual(trickle.status, 204, `trickled candidates are acknowledged (${trickle.status}): ${trickle.body}`);++await pc.setRemoteDescription({ type: 'answer', sdp: res.body });-- pub = { pc, track, feed: startDummyVideo(track), resourceId: res.headers.location.split('/').pop() };++ pub = { pc, track, feed: startDummyVideo(track), resourceId, answer: res.body, etag: res.headers.etag };await waitFor(() => ['connected', 'completed'].includes(pc.connectionState), { what: 'WHIP publisher ICE connected', timeoutMs: 20000 });const sess = await waitFor(async () => { const s = (await rt.store.sessions.ofWorker(worker.runtime.me.id)).find((x) => x.definition_id === defA.id); return s && s.state === 'live' ? s : null; }, { what: 'session live after ICE', timeoutMs: 15000 });@@@ -158,7 +169,45 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives itawait waitFor(() => udp.packets > 0, { what: 'RTP on the PlainRTP egress', timeoutMs: 20000 });const patch = await httpReq(port, 'PATCH', `/whip/session/${pub.resourceId}`, { headers: { 'content-type': 'application/trickle-ice-sdpfrag' }, body: 'a=ice-ufrag:x\r\na=candidate:1 1 udp 1 127.0.0.1 9999 typ host\r\n' });-- assert.strictEqual(patch.status, 204, 'trickle ICE is acknowledged');++ assert.strictEqual(patch.status, 204, 'a candidate of an unknown ICE generation is silently discarded');++});++++t('WHIP resource: PATCH/DELETE preconditions and auth, and an ICE restart that keeps media flowing', async () => {++ const at = `/whip/session/${pub.resourceId}`;++ const sdpfrag = { 'content-type': 'application/trickle-ice-sdpfrag' };++ const patch = (headers, body) => httpReq(port, 'PATCH', at, { headers: { ...sdpfrag, ...headers }, body });++ const unknown = `/whip/session/${crypto.randomBytes(16).toString('hex')}`;++ assert.strictEqual((await httpReq(port, 'PATCH', unknown, { headers: sdpfrag, body: 'a=end-of-candidates\r\n' })).status, 404);++ assert.strictEqual((await httpReq(port, 'DELETE', unknown)).status, 404, 'DELETE of an unknown resource is 404');++ assert.strictEqual((await httpReq(port, 'GET', at)).status, 405);++ assert.strictEqual((await httpReq(port, 'PATCH', at, { headers: { 'content-type': 'application/sdp' }, body: 'a=end-of-candidates\r\n' })).status, 415);++ assert.strictEqual((await patch({ 'if-match': '"stale"' }, 'a=end-of-candidates\r\n')).status, 412);++ assert.strictEqual((await patch({}, 'not an sdp fragment')).status, 400);++ assert.strictEqual((await patch({ authorization: 'Bearer ork_wrong' }, 'a=end-of-candidates\r\n')).status, 401, 'a Bearer that is not the key is refused');++ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: { authorization: 'Bearer ork_wrong' } })).status, 401);++++ // ICE restart (RFC 9725 §4.3.2): new client credentials → new server credentials and a new ETag.++ const before = udp.packets;++ pub.pc.restartIce();++ const offer = await pub.pc.createOffer({ iceRestart: true });++ await pub.pc.setLocalDescription(offer);++ const ufrag = /a=ice-ufrag:(\S+)/.exec(pub.pc.localDescription.sdp)[1];++ const pwd = /a=ice-pwd:(\S+)/.exec(pub.pc.localDescription.sdp)[1];++ const restart = await patch({ 'if-match': '*', authorization: `Bearer ${keyA}` }, `a=ice-ufrag:${ufrag}\r\na=ice-pwd:${pwd}\r\n`);++ assert.strictEqual(restart.status, 200, `ICE restart (${restart.status}): ${restart.body}`);++ assert.match(restart.headers['content-type'], /^application\/trickle-ice-sdpfrag/);++ assert.notStrictEqual(restart.headers.etag, pub.etag, 'a restart starts a new ICE session');++ const serverUfrag = /a=ice-ufrag:(\S+)/.exec(restart.body)[1];++ const serverPwd = /a=ice-pwd:(\S+)/.exec(restart.body)[1];++ assert.notStrictEqual(serverUfrag, /a=ice-ufrag:(\S+)/.exec(pub.answer)[1], 'the server has new credentials');++ assert.strictEqual((await patch({ 'if-match': pub.etag }, 'a=end-of-candidates\r\n')).status, 412, 'the old ICE session\'s ETag no longer matches');++ assert.strictEqual((await patch({ 'if-match': restart.headers.etag }, 'a=end-of-candidates\r\n')).status, 204);++ pub.etag = restart.headers.etag;++ pub.answer = pub.answer.replace(/a=ice-ufrag:\S+/g, `a=ice-ufrag:${serverUfrag}`).replace(/a=ice-pwd:\S+/g, `a=ice-pwd:${serverPwd}`);++ await pub.pc.setRemoteDescription({ type: 'answer', sdp: pub.answer });++ await waitFor(() => ['connected', 'completed'].includes(pub.pc.connectionState), { what: 'publisher ICE after restart', timeoutMs: 20000 });++ await waitFor(() => udp.packets > before + 20, { what: 'RTP still flowing after the ICE restart', timeoutMs: 20000 });++ assert.strictEqual((await rt.store.sessions.get(sessionA)).state, 'live', 'the session stays live across the restart');});t('a WS viewer consumes by playback id and is counted in the heartbeat', async () => {@@@ -242,6 +291,21 @@ t('admission refuses a bad key and a wrong-protocol key', async () => {await assert.rejects(() => wsOpen(`ws://127.0.0.1:${port}/b/${wrong.key.key}`), /http 403/, 'rtmp-only slot refused on the WebRTC listener');});++t('a WHIP resource created with a Bearer needs it again to be deleted', async () => {++ const created = await rt.store.definitions.create({ owner_subject: OWNER, protocols: ['webrtc'] });++ const pc = new werift.RTCPeerConnection({});++ pc.addTransceiver(new werift.MediaStreamTrack({ kind: 'video' }), { direction: 'sendonly' });++ await pc.setLocalDescription(await pc.createOffer());++ const auth = { authorization: `Bearer ${created.key.key}` };++ const res = await httpReq(port, 'POST', `/whip/${created.key.key}`, { headers: { 'content-type': 'application/sdp', ...auth }, body: pc.localDescription.sdp });++ assert.strictEqual(res.status, 201, `WHIP answer (${res.status}): ${res.body.slice(0, 200)}`);++ const at = new URL(res.headers.location).pathname;++ assert.strictEqual((await httpReq(port, 'DELETE', at)).status, 401, 'no Bearer');++ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: auth })).status, 200);++ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: auth })).status, 404, 'the resource is gone');++ pc.close();++});++t('DELETE ends the WHIP session', async () => {const res = await httpReq(port, 'DELETE', `/whip/session/${pub.resourceId}`);assert.strictEqual(res.status, 200);@@@ -66,6 +66,23 @@ t('WHIP SDP: parse, DTLS role, RTP parameters and the answer', () => {assert.match(answer, /a=rtpmap:96 VP8\/90000/);});++t('WHIP PATCH sdpfrag: credentials, candidates, end-of-candidates; malformed bodies throw', () => {++ const trickle = sdp.parseIceFragment('a=ice-ufrag:EsAw\r\na=ice-pwd:P2uYro0UCOQ4zxjKXaWCBui1\r\nm=audio 9 UDP/TLS/RTP/SAVPF 111\r\na=mid:0\r\n'++ + 'a=candidate:1387637174 1 udp 2122260223 192.0.2.1 61764 typ host generation 0 ufrag EsAw network-id 1\r\n'++ + 'a=candidate:3471623853 1 udp 2122194687 198.51.100.2 61765 typ host\r\na=end-of-candidates\r\n');++ assert.deepStrictEqual(trickle, { ufrag: 'EsAw', pwd: 'P2uYro0UCOQ4zxjKXaWCBui1', candidates: 2, endOfCandidates: true });++ // Media-level credentials count too; a bare candidate list has none.++ assert.strictEqual(sdp.parseIceFragment('m=video 9 UDP/TLS/RTP/SAVPF 96\na=ice-ufrag:abc\n').ufrag, 'abc');++ assert.deepStrictEqual(sdp.parseIceFragment('a=candidate:1 1 tcp 5 192.0.2.1 9 typ host tcptype active\n'), { ufrag: null, pwd: null, candidates: 1, endOfCandidates: false });++ assert.strictEqual(sdp.parseIceFragment('').candidates, 0);++ for (const bad of ['hello', '{"candidate":"x"}', 'a=candidate:garbage', undefined]) {++ assert.throws(() => sdp.parseIceFragment(bad), (err) => err.code === 'invalid_sdpfrag', `rejects ${bad}`);++ }++ const frag = sdp.buildIceFragment({ usernameFragment: 'u1', password: 'p1' });++ assert.deepStrictEqual(sdp.parseIceFragment(frag), { ufrag: 'u1', pwd: 'p1', candidates: 0, endOfCandidates: false });++ assert.match(frag, /^a=ice-lite\r\n/, 'the restart answer says the server is ICE-lite');++});++t('a webrtc session playback descriptor is { signaling_url, announced_ip } and carries thumbnail_url', async () => {const rt = await runtime({ dir: tmpDir(), env: { MEDIASOUP_ANNOUNCED_IP: '203.0.113.7' } });const { definition, key } = await rt.store.definitions.create({ owner_subject: OWNER, protocols: ['webrtc'] });@@@ -8,7 +8,8 @@* public 0.0.0.0:OPENRE_WEBRTC_PORT (9936 until the WebRTC cutover; Live's WHIP/SFU live on its* app port 3000), SO_REUSEPORT so two generations listen during a drain:* POST/OPTIONS /whip/<key> WHIP ingest (RFC 9725): offer → answer-- * PATCH/DELETE /whip/session/<id> trickle ICE / end++ * PATCH/DELETE /whip/session/<id> trickle ICE or ICE restart / end (ETag/If-Match,++ * the POST's Bearer, if any, is required again)* WS upgrade /b/<key> browser broadcaster signaling (mediasoup-client)* WS upgrade /w/<session id> viewer signaling, by playback id (never the key)* internal 127.0.0.1:<egressPort> the RTP egress API the restream worker, Media's RTP recorder and@@@ -181,23 +182,30 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),}// ── WHIP ────────────────────────────────────────────────────-- function whipHeaders(req, sessionId, resourceId) {++ function whipHeaders(req, resourceId, etag) {const host = req.headers.host || `${w.publicHost}:${w.publicPort}`;return {...WHIP_CORS,Location: `https://${host}/whip/session/${resourceId}`,-- ETag: `"${resourceId}"`,++ ETag: etag,};}++ /** A new entity-tag per ICE session (RFC 9725 §4.3: it changes on every ICE restart). */++ const newIceEtag = () => `"${crypto.randomBytes(8).toString('hex')}"`;++ const sha256 = (s) => crypto.createHash('sha256').update(String(s)).digest();++ const bearerOf = (req) => {++ const auth = req.headers.authorization;++ return auth && auth.startsWith('Bearer ') ? auth.slice(7).trim() : null;++ };++function sendWhipError(res, status, code, message) {res.writeHead(status, { ...WHIP_CORS, 'content-type': 'application/json', 'X-WHIP-ERROR': code });res.end(JSON.stringify({ error: message, error_code: code }));}async function handleWhipPost(req, res, key) {-- const auth = req.headers.authorization;-- const bearer = auth && auth.startsWith('Bearer ') ? auth.slice(7).trim() : null;++ const bearer = bearerOf(req);if (bearer && bearer !== key) return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');let offerText;try { offerText = await readBody(req); } catch { return sendWhipError(res, 413, 'body_too_large', 'Offer too large'); }@@@ -264,7 +272,14 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),return sendWhipError(res, 500, 'answer_generation_failed', 'Failed to generate SDP answer');}-- sess.whip.set(resourceId, { peerId, transportId: transportInfo.id, iceReady: false, iceGraceTimer: null, iceFailTimer: null });++ const remoteUfrag = offerSdp.iceUfrag || ((offerSdp.media || []).find((m) => m.iceUfrag) || {}).iceUfrag || null;++ sess.whip.set(resourceId, {++ peerId, transportId: transportInfo.id, iceReady: false, iceGraceTimer: null, iceFailTimer: null,++ // The resource's own state: the ICE session's entity-tag, the client's current ICE++ // username fragment, and the key's hash — a client that authenticated with a Bearer++ // must send it again on PATCH/DELETE; any Bearer sent must be the key.++ etag: newIceEtag(), remoteUfrag, keyHash: sha256(key), bearerRequired: Boolean(bearer),++ });// The session goes live when media flows: ICE connected (Live's whip-handler rule).transport.on('icestatechange', (state) => {@@@ -287,21 +302,69 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),});log.log(`[webrtc] WHIP session ${sess.sessionId} accepted for ${sess.definitionId} (key …${sess.hint}), ${Object.keys(producersByKind).length} producer(s)`);-- res.writeHead(201, { 'content-type': 'application/sdp', ...whipHeaders(req, sess.sessionId, resourceId) });++ res.writeHead(201, { 'content-type': 'application/sdp', ...whipHeaders(req, resourceId, sess.whip.get(resourceId).etag) });res.end(answer);}-- function handleWhipPatch(req, res, resourceId) {-- let found = null;-- for (const sess of sessions.values()) if (sess.whip.has(resourceId)) { found = sess; break; }++ function findWhip(resourceId) {++ for (const sess of sessions.values()) {++ const entry = sess.whip.get(resourceId);++ if (entry) return { sess, entry };++ }++ return null;++ }++++ /** RFC 9725 §4.5: the resource accepts the Bearer the endpoint did, and only that. */++ function whipResourceAuthorized(req, entry) {++ const bearer = bearerOf(req);++ if (!bearer) return !entry.bearerRequired;++ return crypto.timingSafeEqual(sha256(bearer), entry.keyHash);++ }++++ /**++ * Trickle ICE / ICE restart (RFC 9725 §4.3, RFC 8840 fragments). The transport is ICE-lite, so++ * trickled candidates are acknowledged and discarded: the client's connectivity checks reach++ * the transport's own candidates (sent in the answer) and mediasoup learns the client's address++ * from them. New ICE credentials in the fragment are an ICE restart: the transport gets new++ * credentials, answered 200 with them and a new ETag; the session stays up either way.++ */++ async function handleWhipPatch(req, res, resourceId) {++ const found = findWhip(resourceId);if (!found) return sendWhipError(res, 404, 'session_not_found', 'Session not found');-- readBody(req).then((body) => { sdpTool.countCandidates(body); res.writeHead(204, WHIP_CORS); res.end(); }).catch(() => { res.writeHead(204, WHIP_CORS); res.end(); });++ const { sess, entry } = found;++ if (!whipResourceAuthorized(req, entry)) return sendWhipError(res, 401, 'unauthorized', 'Bearer token required for this resource');++ if (!/^application\/trickle-ice-sdpfrag\s*(;|$)/i.test(req.headers['content-type'] || '')) {++ return sendWhipError(res, 415, 'unsupported_media_type', 'PATCH body must be application/trickle-ice-sdpfrag');++ }++ const ifMatch = (req.headers['if-match'] || '').trim();++ if (ifMatch && ifMatch !== '*' && ifMatch !== entry.etag) return sendWhipError(res, 412, 'etag_mismatch', 'If-Match does not name the current ICE session');++ let body;++ try { body = await readBody(req); } catch { return sendWhipError(res, 413, 'body_too_large', 'Fragment too large'); }++ let frag;++ try { frag = sdpTool.parseIceFragment(body); } catch { return sendWhipError(res, 400, 'invalid_sdpfrag', 'Invalid trickle-ice-sdpfrag'); }++++ if (frag.ufrag && frag.pwd && frag.ufrag !== entry.remoteUfrag) {++ let iceParameters;++ try { iceParameters = await sfu.restartIce(sess.sessionId, entry.peerId, entry.transportId); } catch (err) {++ log.warn(`[webrtc] WHIP ICE restart failed for ${sess.sessionId}: ${err.message}`);++ return sendWhipError(res, 500, 'ice_restart_failed', 'ICE restart failed');++ }++ if (!sess.whip.has(resourceId)) return sendWhipError(res, 404, 'session_not_found', 'Session not found');++ entry.remoteUfrag = frag.ufrag;++ entry.etag = newIceEtag();++ log.log(`[webrtc] WHIP ICE restart for ${sess.sessionId}`);++ res.writeHead(200, { ...WHIP_CORS, 'content-type': 'application/trickle-ice-sdpfrag', ETag: entry.etag });++ return res.end(sdpTool.buildIceFragment(iceParameters));++ }++ res.writeHead(204, WHIP_CORS);++ res.end();}async function handleWhipDelete(req, res, resourceId) {-- let found = null;-- for (const sess of sessions.values()) if (sess.whip.has(resourceId)) { found = sess; break; }-- if (found) await endSession(found.sessionId, 'whip_delete');++ const found = findWhip(resourceId);++ if (!found) return sendWhipError(res, 404, 'session_not_found', 'Session not found');++ if (!whipResourceAuthorized(req, found.entry)) return sendWhipError(res, 401, 'unauthorized', 'Bearer token required for this resource');++ await endSession(found.sess.sessionId, 'whip_delete');res.writeHead(200, WHIP_CORS);res.end();}@@@ -409,8 +472,9 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),const kMatch = WHIP_KEY_RE.exec(url.pathname);const sMatch = WHIP_SESSION_RE.exec(url.pathname);if (req.method === 'POST' && kMatch) return safe(handleWhipPost(req, res, decodeURIComponent(kMatch[1])), 'whip post');-- if (req.method === 'PATCH' && sMatch) return handleWhipPatch(req, res, sMatch[1]);++ if (req.method === 'PATCH' && sMatch) return safe(handleWhipPatch(req, res, sMatch[1]), 'whip patch');if (req.method === 'DELETE' && sMatch) return safe(handleWhipDelete(req, res, sMatch[1]), 'whip delete');++ if (sMatch || kMatch) { res.writeHead(405, { ...WHIP_CORS, Allow: sMatch ? 'PATCH, DELETE, OPTIONS' : 'POST, OPTIONS' }); return res.end(); }res.writeHead(404, WHIP_CORS);return res.end();}@@@ -261,15 +261,37 @@ function buildSdpAnswer(transport, offerSdp, producersByKind, { serverName = 'Opreturn write(sdpObj);}--/** How many ICE candidates a trickle-PATCH body carries (used only for logging). */--function countCandidates(sdpText) {-- if (typeof sdpText !== 'string' || !sdpText.trim()) return 0;-- try {-- const parsed = parse(sdpText);-- return Array.isArray(parsed.media) ? parsed.media.reduce((sum, m) => sum + (Array.isArray(m.candidates) ? m.candidates.length : 0), 0) : 0;-- } catch {-- return 0;++const FRAG_LINE_RE = /^[a-z]=\S/;++const CANDIDATE_RE = /^candidate:\S+ \d+ \S+ \d+ \S+ \d+ typ \S+/i;++++/**++ * Read a WHIP PATCH body, an RFC 8840 "application/trickle-ice-sdpfrag": the ICE credentials it++ * names (session or media level, the first wins) and how many candidates it carries. Throws++ * (code 'invalid_sdpfrag') on anything that is not an SDP fragment or on a malformed candidate.++ */++function parseIceFragment(text) {++ if (typeof text !== 'string') throw Object.assign(new Error('missing sdpfrag'), { code: 'invalid_sdpfrag' });++ const out = { ufrag: null, pwd: null, candidates: 0, endOfCandidates: false };++ for (const raw of text.split(/\r?\n/)) {++ const line = raw.trim();++ if (!line) continue;++ if (!FRAG_LINE_RE.test(line)) throw Object.assign(new Error(`not an SDP line: ${line.slice(0, 40)}`), { code: 'invalid_sdpfrag' });++ if (!line.startsWith('a=')) continue;++ const attr = line.slice(2);++ if (attr.startsWith('ice-ufrag:')) out.ufrag = out.ufrag || attr.slice(10).trim();++ else if (attr.startsWith('ice-pwd:')) out.pwd = out.pwd || attr.slice(8).trim();++ else if (attr === 'end-of-candidates') out.endOfCandidates = true;++ else if (attr.startsWith('candidate:')) {++ if (!CANDIDATE_RE.test(attr)) throw Object.assign(new Error('malformed candidate'), { code: 'invalid_sdpfrag' });++ out.candidates++;++ }}++ return out;++}++++/** The sdpfrag a WHIP resource answers an ICE restart with: the media server's new credentials. */++function buildIceFragment(iceParameters) {++ return ['a=ice-lite', `a=ice-ufrag:${iceParameters.usernameFragment}`, `a=ice-pwd:${iceParameters.password}`, ''].join('\r\n');}/**@@@ -297,5 +319,5 @@ function buildEgressSdp({ video, audio } = {}) {module.exports = {available, parse, write,-- getDtlsSetupAttribute, selectDtlsFingerprint, extractDtlsParameters, extractRtpParameters, buildSdpAnswer, countCandidates, buildEgressSdp,++ getDtlsSetupAttribute, selectDtlsFingerprint, extractDtlsParameters, extractRtpParameters, buildSdpAnswer, parseIceFragment, buildIceFragment, buildEgressSdp,};@@@ -112,6 +112,17 @@ function createSfu({ config, log = console }) {return transport;}++ /**++ * WHIP ICE restart (RFC 9725 §4.3.2). mediasoup's WebRtcTransport is ICE-lite and always the++ * controlled agent: it has no remote candidates to add (it learns the peer's address from the++ * peer's connectivity checks), so a restart only issues new local credentials.++ */++ async function restartIce(sessionId, peerId, transportId) {++ const transport = lookupTransport(sessionId, peerId, transportId);++ if (!transport) throw new Error('Transport not found');++ return await transport.restartIce();++ }++async function produce(sessionId, peerId, transportId, kind, rtpParameters) {const r = room(sessionId);if (!r) throw new Error('Router not found');@@@ -270,7 +281,7 @@ function createSfu({ config, log = console }) {off: (...args) => emitter.off(...args),init, setAnnouncedIp,getOrCreateRouter, hasRouter, getRouterCapabilities,-- createTransport, connectTransport, produce, consume,++ createTransport, connectTransport, restartIce, produce, consume,createPlainConsumer, closePlainConsumer,findProducerByKind, hasProducers, getProducers, getViewerCount,closePeer, closeSession, closeAll,
@@ -250,7 +250,7 @@ one slot. Status on 2026-09-23: OpenRe is deployed (`655b98a10aaa`) and DNS is d
@@ -250,7 +250,7 @@ one slot. Status on 2026-09-23: OpenRe is deployed (`655b98a10aaa`) and DNS is d
### WebRTC (worker ready)
### WebRTC (worker ready)
-`workers/webrtc.js` owns the whole WebRTC stack (decision 1: mediasoup is single-process, so ingest and consumption cannot be split): WHIP (RFC 9725) admission of `POST /whip/<key>` (Bearer optional) with `PATCH /whip/session/<id>` and `DELETE`, one mediasoup Router per session on UDP 10200–10300, the `/b/<key>` broadcaster and `/w/<id>` viewer signaling endpoints Live's broadcast/watch pages speak, a loopback RTP/SDP egress the restream worker and Media's RTP recorder consume, and live thumbnails. It needs `mediasoup` and `sdp-transform` installed (they are dependencies) and, in production, `MEDIASOUP_ANNOUNCED_IP`. docs/cutover.md has the WebRTC procedure (W-A once, then W-B1–W-B6 per slot). A webrtc session's playback descriptor is `webrtc: { signaling_url, announced_ip }`.
+`workers/webrtc.js` owns the whole WebRTC stack (decision 1: mediasoup is single-process, so ingest and consumption cannot be split): WHIP (RFC 9725) admission of `POST /whip/<key>` (Bearer optional) with `PATCH /whip/session/<id>` (trickle ICE is acknowledged and discarded, since the transport is ICE-lite; new ICE credentials restart ICE: `200`, the server's new credentials and a new `ETag`; `If-Match` is checked) and `DELETE` (`404` for an unknown resource; both need the POST's Bearer again if it sent one), one mediasoup Router per session on UDP 10200–10300, the `/b/<key>` broadcaster and `/w/<id>` viewer signaling endpoints Live's broadcast/watch pages speak, a loopback RTP/SDP egress the restream worker and Media's RTP recorder consume, and live thumbnails. It needs `mediasoup` and `sdp-transform` installed (they are dependencies) and, in production, `MEDIASOUP_ANNOUNCED_IP`. docs/cutover.md has the WebRTC procedure (W-A once, then W-B1–W-B6 per slot). A webrtc session's playback descriptor is `webrtc: { signaling_url, announced_ip }`.
## Acceptance
## Acceptance
@@ -414,8 +414,9 @@ done **once** (the worker generation and its ports), then per slot like every pr
@@ -414,8 +414,9 @@ done **once** (the worker generation and its ports), then per slot like every pr
`webrtc` is in `OPENRE_PROTOCOLS` (brief §4.5).
`webrtc` is in `OPENRE_PROTOCOLS` (brief §4.5).
- **B4 Broadcaster:** regenerate the key, then either
- **B4 Broadcaster:** regenerate the key, then either
- **WHIP:** POST an SDP offer to `https://ingest.openre.stream/whip/<key>` (Bearer optional, must
- **WHIP:** POST an SDP offer to `https://ingest.openre.stream/whip/<key>` (Bearer optional, must
- match) → `201` + the answer + `Location: /whip/session/<id>`; trickle with `PATCH`, end with
- `DELETE`. The session goes live when ICE connects.
+ match) → `201` + the answer + `Location: /whip/session/<id>` + an `ETag`; trickle or restart ICE
+ with `PATCH` (`application/trickle-ice-sdpfrag`, `If-Match`), end with `DELETE` (both need the
+ POST's Bearer again if it sent one). The session goes live when ICE connects.
- **Browser:** connect `ws(s)://ingest.openre.stream/b/<key>` and run the mediasoup-client
- **Browser:** connect `ws(s)://ingest.openre.stream/b/<key>` and run the mediasoup-client
`sfu-get-capabilities → sfu-create-transport → sfu-connect-transport → sfu-produce` flow; the
`sfu-get-capabilities → sfu-create-transport → sfu-connect-transport → sfu-produce` flow; the
session goes live on the first video producer.
session goes live on the first video producer.
@@ -126,15 +126,26 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives it
@@ -126,15 +126,26 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives it
const offer = await pc.createOffer();
const offer = await pc.createOffer();
await pc.setLocalDescription(offer);
await pc.setLocalDescription(offer);
- const res = await httpReq(port, 'POST', `/whip/${keyA}`, { headers: { 'content-type': 'application/sdp' }, body: pc.localDescription.sdp });
+ // Trickle-only, as a browser behind NAT publishes: the offer carries no candidates, they follow
+ // in a PATCH. The transport is ICE-lite, so ICE must connect from the client's checks alone.
+ const fullOffer = pc.localDescription.sdp;
+ const candidateLines = fullOffer.split('\r\n').filter((l) => l.startsWith('a=candidate:'));
+ assert.ok(candidateLines.length > 0, 'the client gathered candidates to trickle');
+ const trickleOffer = fullOffer.split('\r\n').filter((l) => !l.startsWith('a=candidate:') && l !== 'a=end-of-candidates').join('\r\n');
+ const res = await httpReq(port, 'POST', `/whip/${keyA}`, { headers: { 'content-type': 'application/sdp' }, body: trickleOffer });
assert.strictEqual(res.status, 201, `WHIP answer (${res.status}): ${res.body.slice(0, 200)}`);
assert.strictEqual(res.status, 201, `WHIP answer (${res.status}): ${res.body.slice(0, 200)}`);
assert.match(res.body, /a=recvonly/, 'the answer is a recvonly SDP');
assert.match(res.body, /a=recvonly/, 'the answer is a recvonly SDP');
assert.match(res.body, /a=ice-lite/);
assert.match(res.body, /a=ice-lite/);
assert.match(res.headers.location, /\/whip\/session\/[0-9a-f]{16,}$/, 'Location points at the session resource');
assert.match(res.headers.location, /\/whip\/session\/[0-9a-f]{16,}$/, 'Location points at the session resource');
assert.match(String(res.headers.etag), /^"/, 'an ETag is returned');
assert.match(String(res.headers.etag), /^"/, 'an ETag is returned');
+ const resourceId = res.headers.location.split('/').pop();
+ const ufrag = /a=ice-ufrag:(\S+)/.exec(fullOffer)[1];
+ const trickle = await httpReq(port, 'PATCH', `/whip/session/${resourceId}`, { headers: { 'content-type': 'application/trickle-ice-sdpfrag', 'if-match': res.headers.etag }, body: `a=ice-ufrag:${ufrag}\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=mid:0\r\n${candidateLines.join('\r\n')}\r\na=end-of-candidates\r\n` });
+ assert.strictEqual(trickle.status, 204, `trickled candidates are acknowledged (${trickle.status}): ${trickle.body}`);
+
await pc.setRemoteDescription({ type: 'answer', sdp: res.body });
await pc.setRemoteDescription({ type: 'answer', sdp: res.body });
- pub = { pc, track, feed: startDummyVideo(track), resourceId: res.headers.location.split('/').pop() };
+ pub = { pc, track, feed: startDummyVideo(track), resourceId, answer: res.body, etag: res.headers.etag };
await waitFor(() => ['connected', 'completed'].includes(pc.connectionState), { what: 'WHIP publisher ICE connected', timeoutMs: 20000 });
await waitFor(() => ['connected', 'completed'].includes(pc.connectionState), { what: 'WHIP publisher ICE connected', timeoutMs: 20000 });
const sess = await waitFor(async () => { const s = (await rt.store.sessions.ofWorker(worker.runtime.me.id)).find((x) => x.definition_id === defA.id); return s && s.state === 'live' ? s : null; }, { what: 'session live after ICE', timeoutMs: 15000 });
const sess = await waitFor(async () => { const s = (await rt.store.sessions.ofWorker(worker.runtime.me.id)).find((x) => x.definition_id === defA.id); return s && s.state === 'live' ? s : null; }, { what: 'session live after ICE', timeoutMs: 15000 });
@@ -158,7 +169,45 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives it
@@ -158,7 +169,45 @@ t('a WHIP offer is answered, goes live on ICE, and a PlainRTP egress receives it
await waitFor(() => udp.packets > 0, { what: 'RTP on the PlainRTP egress', timeoutMs: 20000 });
await waitFor(() => udp.packets > 0, { what: 'RTP on the PlainRTP egress', timeoutMs: 20000 });
const patch = await httpReq(port, 'PATCH', `/whip/session/${pub.resourceId}`, { headers: { 'content-type': 'application/trickle-ice-sdpfrag' }, body: 'a=ice-ufrag:x\r\na=candidate:1 1 udp 1 127.0.0.1 9999 typ host\r\n' });
const patch = await httpReq(port, 'PATCH', `/whip/session/${pub.resourceId}`, { headers: { 'content-type': 'application/trickle-ice-sdpfrag' }, body: 'a=ice-ufrag:x\r\na=candidate:1 1 udp 1 127.0.0.1 9999 typ host\r\n' });
- assert.strictEqual(patch.status, 204, 'trickle ICE is acknowledged');
+ assert.strictEqual(patch.status, 204, 'a candidate of an unknown ICE generation is silently discarded');
+});
+
+t('WHIP resource: PATCH/DELETE preconditions and auth, and an ICE restart that keeps media flowing', async () => {
+ const at = `/whip/session/${pub.resourceId}`;
+ const sdpfrag = { 'content-type': 'application/trickle-ice-sdpfrag' };
+ const patch = (headers, body) => httpReq(port, 'PATCH', at, { headers: { ...sdpfrag, ...headers }, body });
+ const unknown = `/whip/session/${crypto.randomBytes(16).toString('hex')}`;
+ assert.strictEqual((await httpReq(port, 'PATCH', unknown, { headers: sdpfrag, body: 'a=end-of-candidates\r\n' })).status, 404);
+ assert.strictEqual((await httpReq(port, 'DELETE', unknown)).status, 404, 'DELETE of an unknown resource is 404');
+ assert.strictEqual((await httpReq(port, 'GET', at)).status, 405);
+ assert.strictEqual((await httpReq(port, 'PATCH', at, { headers: { 'content-type': 'application/sdp' }, body: 'a=end-of-candidates\r\n' })).status, 415);
+ assert.strictEqual((await patch({ 'if-match': '"stale"' }, 'a=end-of-candidates\r\n')).status, 412);
+ assert.strictEqual((await patch({}, 'not an sdp fragment')).status, 400);
+ assert.strictEqual((await patch({ authorization: 'Bearer ork_wrong' }, 'a=end-of-candidates\r\n')).status, 401, 'a Bearer that is not the key is refused');
+ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: { authorization: 'Bearer ork_wrong' } })).status, 401);
+
+ // ICE restart (RFC 9725 §4.3.2): new client credentials → new server credentials and a new ETag.
+ const before = udp.packets;
+ pub.pc.restartIce();
+ const offer = await pub.pc.createOffer({ iceRestart: true });
+ await pub.pc.setLocalDescription(offer);
+ const ufrag = /a=ice-ufrag:(\S+)/.exec(pub.pc.localDescription.sdp)[1];
+ const pwd = /a=ice-pwd:(\S+)/.exec(pub.pc.localDescription.sdp)[1];
+ const restart = await patch({ 'if-match': '*', authorization: `Bearer ${keyA}` }, `a=ice-ufrag:${ufrag}\r\na=ice-pwd:${pwd}\r\n`);
+ assert.strictEqual(restart.status, 200, `ICE restart (${restart.status}): ${restart.body}`);
+ assert.match(restart.headers['content-type'], /^application\/trickle-ice-sdpfrag/);
+ assert.notStrictEqual(restart.headers.etag, pub.etag, 'a restart starts a new ICE session');
+ const serverUfrag = /a=ice-ufrag:(\S+)/.exec(restart.body)[1];
+ const serverPwd = /a=ice-pwd:(\S+)/.exec(restart.body)[1];
+ assert.notStrictEqual(serverUfrag, /a=ice-ufrag:(\S+)/.exec(pub.answer)[1], 'the server has new credentials');
+ assert.strictEqual((await patch({ 'if-match': pub.etag }, 'a=end-of-candidates\r\n')).status, 412, 'the old ICE session\'s ETag no longer matches');
+ assert.strictEqual((await patch({ 'if-match': restart.headers.etag }, 'a=end-of-candidates\r\n')).status, 204);
+ pub.etag = restart.headers.etag;
+ pub.answer = pub.answer.replace(/a=ice-ufrag:\S+/g, `a=ice-ufrag:${serverUfrag}`).replace(/a=ice-pwd:\S+/g, `a=ice-pwd:${serverPwd}`);
+ await pub.pc.setRemoteDescription({ type: 'answer', sdp: pub.answer });
+ await waitFor(() => ['connected', 'completed'].includes(pub.pc.connectionState), { what: 'publisher ICE after restart', timeoutMs: 20000 });
+ await waitFor(() => udp.packets > before + 20, { what: 'RTP still flowing after the ICE restart', timeoutMs: 20000 });
+ assert.strictEqual((await rt.store.sessions.get(sessionA)).state, 'live', 'the session stays live across the restart');
});
});
t('a WS viewer consumes by playback id and is counted in the heartbeat', async () => {
t('a WS viewer consumes by playback id and is counted in the heartbeat', async () => {
@@ -242,6 +291,21 @@ t('admission refuses a bad key and a wrong-protocol key', async () => {
@@ -242,6 +291,21 @@ t('admission refuses a bad key and a wrong-protocol key', async () => {
await assert.rejects(() => wsOpen(`ws://127.0.0.1:${port}/b/${wrong.key.key}`), /http 403/, 'rtmp-only slot refused on the WebRTC listener');
await assert.rejects(() => wsOpen(`ws://127.0.0.1:${port}/b/${wrong.key.key}`), /http 403/, 'rtmp-only slot refused on the WebRTC listener');
});
});
+t('a WHIP resource created with a Bearer needs it again to be deleted', async () => {
+ const created = await rt.store.definitions.create({ owner_subject: OWNER, protocols: ['webrtc'] });
+ const pc = new werift.RTCPeerConnection({});
+ pc.addTransceiver(new werift.MediaStreamTrack({ kind: 'video' }), { direction: 'sendonly' });
+ await pc.setLocalDescription(await pc.createOffer());
+ const auth = { authorization: `Bearer ${created.key.key}` };
+ const res = await httpReq(port, 'POST', `/whip/${created.key.key}`, { headers: { 'content-type': 'application/sdp', ...auth }, body: pc.localDescription.sdp });
+ assert.strictEqual(res.status, 201, `WHIP answer (${res.status}): ${res.body.slice(0, 200)}`);
+ const at = new URL(res.headers.location).pathname;
+ assert.strictEqual((await httpReq(port, 'DELETE', at)).status, 401, 'no Bearer');
+ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: auth })).status, 200);
+ assert.strictEqual((await httpReq(port, 'DELETE', at, { headers: auth })).status, 404, 'the resource is gone');
+ pc.close();
+});
+
t('DELETE ends the WHIP session', async () => {
t('DELETE ends the WHIP session', async () => {
const res = await httpReq(port, 'DELETE', `/whip/session/${pub.resourceId}`);
const res = await httpReq(port, 'DELETE', `/whip/session/${pub.resourceId}`);
assert.strictEqual(res.status, 200);
assert.strictEqual(res.status, 200);
@@ -66,6 +66,23 @@ t('WHIP SDP: parse, DTLS role, RTP parameters and the answer', () => {
@@ -66,6 +66,23 @@ t('WHIP SDP: parse, DTLS role, RTP parameters and the answer', () => {
assert.match(answer, /a=rtpmap:96 VP8\/90000/);
assert.match(answer, /a=rtpmap:96 VP8\/90000/);
});
});
+t('WHIP PATCH sdpfrag: credentials, candidates, end-of-candidates; malformed bodies throw', () => {
+ const trickle = sdp.parseIceFragment('a=ice-ufrag:EsAw\r\na=ice-pwd:P2uYro0UCOQ4zxjKXaWCBui1\r\nm=audio 9 UDP/TLS/RTP/SAVPF 111\r\na=mid:0\r\n'
+ + 'a=candidate:1387637174 1 udp 2122260223 192.0.2.1 61764 typ host generation 0 ufrag EsAw network-id 1\r\n'
+ + 'a=candidate:3471623853 1 udp 2122194687 198.51.100.2 61765 typ host\r\na=end-of-candidates\r\n');
+ assert.deepStrictEqual(trickle, { ufrag: 'EsAw', pwd: 'P2uYro0UCOQ4zxjKXaWCBui1', candidates: 2, endOfCandidates: true });
+ // Media-level credentials count too; a bare candidate list has none.
+ assert.strictEqual(sdp.parseIceFragment('m=video 9 UDP/TLS/RTP/SAVPF 96\na=ice-ufrag:abc\n').ufrag, 'abc');
+ assert.deepStrictEqual(sdp.parseIceFragment('a=candidate:1 1 tcp 5 192.0.2.1 9 typ host tcptype active\n'), { ufrag: null, pwd: null, candidates: 1, endOfCandidates: false });
+ assert.strictEqual(sdp.parseIceFragment('').candidates, 0);
+ for (const bad of ['hello', '{"candidate":"x"}', 'a=candidate:garbage', undefined]) {
+ assert.throws(() => sdp.parseIceFragment(bad), (err) => err.code === 'invalid_sdpfrag', `rejects ${bad}`);
+ }
+ const frag = sdp.buildIceFragment({ usernameFragment: 'u1', password: 'p1' });
+ assert.deepStrictEqual(sdp.parseIceFragment(frag), { ufrag: 'u1', pwd: 'p1', candidates: 0, endOfCandidates: false });
+ assert.match(frag, /^a=ice-lite\r\n/, 'the restart answer says the server is ICE-lite');
+});
+
t('a webrtc session playback descriptor is { signaling_url, announced_ip } and carries thumbnail_url', async () => {
t('a webrtc session playback descriptor is { signaling_url, announced_ip } and carries thumbnail_url', async () => {
const rt = await runtime({ dir: tmpDir(), env: { MEDIASOUP_ANNOUNCED_IP: '203.0.113.7' } });
const rt = await runtime({ dir: tmpDir(), env: { MEDIASOUP_ANNOUNCED_IP: '203.0.113.7' } });
const { definition, key } = await rt.store.definitions.create({ owner_subject: OWNER, protocols: ['webrtc'] });
const { definition, key } = await rt.store.definitions.create({ owner_subject: OWNER, protocols: ['webrtc'] });
@@ -8,7 +8,8 @@
@@ -8,7 +8,8 @@
* public 0.0.0.0:OPENRE_WEBRTC_PORT (9936 until the WebRTC cutover; Live's WHIP/SFU live on its
* public 0.0.0.0:OPENRE_WEBRTC_PORT (9936 until the WebRTC cutover; Live's WHIP/SFU live on its
* app port 3000), SO_REUSEPORT so two generations listen during a drain:
* app port 3000), SO_REUSEPORT so two generations listen during a drain:
* POST/OPTIONS /whip/<key> WHIP ingest (RFC 9725): offer → answer
* POST/OPTIONS /whip/<key> WHIP ingest (RFC 9725): offer → answer
- * PATCH/DELETE /whip/session/<id> trickle ICE / end
+ * PATCH/DELETE /whip/session/<id> trickle ICE or ICE restart / end (ETag/If-Match,
+ * the POST's Bearer, if any, is required again)
* WS upgrade /b/<key> browser broadcaster signaling (mediasoup-client)
* WS upgrade /b/<key> browser broadcaster signaling (mediasoup-client)
* WS upgrade /w/<session id> viewer signaling, by playback id (never the key)
* WS upgrade /w/<session id> viewer signaling, by playback id (never the key)
* internal 127.0.0.1:<egressPort> the RTP egress API the restream worker, Media's RTP recorder and
* internal 127.0.0.1:<egressPort> the RTP egress API the restream worker, Media's RTP recorder and
@@ -181,23 +182,30 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
@@ -181,23 +182,30 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
}
}
// ── WHIP ────────────────────────────────────────────────────
// ── WHIP ────────────────────────────────────────────────────
- function whipHeaders(req, sessionId, resourceId) {
+ function whipHeaders(req, resourceId, etag) {
const host = req.headers.host || `${w.publicHost}:${w.publicPort}`;
const host = req.headers.host || `${w.publicHost}:${w.publicPort}`;
return {
return {
...WHIP_CORS,
...WHIP_CORS,
Location: `https://${host}/whip/session/${resourceId}`,
Location: `https://${host}/whip/session/${resourceId}`,
- ETag: `"${resourceId}"`,
+ ETag: etag,
};
};
}
}
+ /** A new entity-tag per ICE session (RFC 9725 §4.3: it changes on every ICE restart). */
+ const newIceEtag = () => `"${crypto.randomBytes(8).toString('hex')}"`;
+ const sha256 = (s) => crypto.createHash('sha256').update(String(s)).digest();
+ const bearerOf = (req) => {
+ const auth = req.headers.authorization;
+ return auth && auth.startsWith('Bearer ') ? auth.slice(7).trim() : null;
+ };
+
function sendWhipError(res, status, code, message) {
function sendWhipError(res, status, code, message) {
res.writeHead(status, { ...WHIP_CORS, 'content-type': 'application/json', 'X-WHIP-ERROR': code });
res.writeHead(status, { ...WHIP_CORS, 'content-type': 'application/json', 'X-WHIP-ERROR': code });
res.end(JSON.stringify({ error: message, error_code: code }));
res.end(JSON.stringify({ error: message, error_code: code }));
}
}
async function handleWhipPost(req, res, key) {
async function handleWhipPost(req, res, key) {
- const auth = req.headers.authorization;
- const bearer = auth && auth.startsWith('Bearer ') ? auth.slice(7).trim() : null;
+ const bearer = bearerOf(req);
if (bearer && bearer !== key) return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');
if (bearer && bearer !== key) return sendWhipError(res, 401, 'bearer_mismatch', 'Bearer token does not match stream key');
let offerText;
let offerText;
try { offerText = await readBody(req); } catch { return sendWhipError(res, 413, 'body_too_large', 'Offer too large'); }
try { offerText = await readBody(req); } catch { return sendWhipError(res, 413, 'body_too_large', 'Offer too large'); }
@@ -264,7 +272,14 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
@@ -264,7 +272,14 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
return sendWhipError(res, 500, 'answer_generation_failed', 'Failed to generate SDP answer');
return sendWhipError(res, 500, 'answer_generation_failed', 'Failed to generate SDP answer');
}
}
- sess.whip.set(resourceId, { peerId, transportId: transportInfo.id, iceReady: false, iceGraceTimer: null, iceFailTimer: null });
+ const remoteUfrag = offerSdp.iceUfrag || ((offerSdp.media || []).find((m) => m.iceUfrag) || {}).iceUfrag || null;
+ sess.whip.set(resourceId, {
+ peerId, transportId: transportInfo.id, iceReady: false, iceGraceTimer: null, iceFailTimer: null,
+ // The resource's own state: the ICE session's entity-tag, the client's current ICE
+ // username fragment, and the key's hash — a client that authenticated with a Bearer
+ // must send it again on PATCH/DELETE; any Bearer sent must be the key.
+ etag: newIceEtag(), remoteUfrag, keyHash: sha256(key), bearerRequired: Boolean(bearer),
+ });
// The session goes live when media flows: ICE connected (Live's whip-handler rule).
// The session goes live when media flows: ICE connected (Live's whip-handler rule).
transport.on('icestatechange', (state) => {
transport.on('icestatechange', (state) => {
@@ -287,21 +302,69 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
@@ -287,21 +302,69 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
});
});
log.log(`[webrtc] WHIP session ${sess.sessionId} accepted for ${sess.definitionId} (key …${sess.hint}), ${Object.keys(producersByKind).length} producer(s)`);
log.log(`[webrtc] WHIP session ${sess.sessionId} accepted for ${sess.definitionId} (key …${sess.hint}), ${Object.keys(producersByKind).length} producer(s)`);
- res.writeHead(201, { 'content-type': 'application/sdp', ...whipHeaders(req, sess.sessionId, resourceId) });
+ res.writeHead(201, { 'content-type': 'application/sdp', ...whipHeaders(req, resourceId, sess.whip.get(resourceId).etag) });
res.end(answer);
res.end(answer);
}
}
- function handleWhipPatch(req, res, resourceId) {
- let found = null;
- for (const sess of sessions.values()) if (sess.whip.has(resourceId)) { found = sess; break; }
+ function findWhip(resourceId) {
+ for (const sess of sessions.values()) {
+ const entry = sess.whip.get(resourceId);
+ if (entry) return { sess, entry };
+ }
+ return null;
+ }
+
+ /** RFC 9725 §4.5: the resource accepts the Bearer the endpoint did, and only that. */
+ function whipResourceAuthorized(req, entry) {
+ const bearer = bearerOf(req);
+ if (!bearer) return !entry.bearerRequired;
+ return crypto.timingSafeEqual(sha256(bearer), entry.keyHash);
+ }
+
+ /**
+ * Trickle ICE / ICE restart (RFC 9725 §4.3, RFC 8840 fragments). The transport is ICE-lite, so
+ * trickled candidates are acknowledged and discarded: the client's connectivity checks reach
+ * the transport's own candidates (sent in the answer) and mediasoup learns the client's address
+ * from them. New ICE credentials in the fragment are an ICE restart: the transport gets new
+ * credentials, answered 200 with them and a new ETag; the session stays up either way.
+ */
+ async function handleWhipPatch(req, res, resourceId) {
+ const found = findWhip(resourceId);
if (!found) return sendWhipError(res, 404, 'session_not_found', 'Session not found');
if (!found) return sendWhipError(res, 404, 'session_not_found', 'Session not found');
- readBody(req).then((body) => { sdpTool.countCandidates(body); res.writeHead(204, WHIP_CORS); res.end(); }).catch(() => { res.writeHead(204, WHIP_CORS); res.end(); });
+ const { sess, entry } = found;
+ if (!whipResourceAuthorized(req, entry)) return sendWhipError(res, 401, 'unauthorized', 'Bearer token required for this resource');
+ if (!/^application\/trickle-ice-sdpfrag\s*(;|$)/i.test(req.headers['content-type'] || '')) {
+ return sendWhipError(res, 415, 'unsupported_media_type', 'PATCH body must be application/trickle-ice-sdpfrag');
+ }
+ const ifMatch = (req.headers['if-match'] || '').trim();
+ if (ifMatch && ifMatch !== '*' && ifMatch !== entry.etag) return sendWhipError(res, 412, 'etag_mismatch', 'If-Match does not name the current ICE session');
+ let body;
+ try { body = await readBody(req); } catch { return sendWhipError(res, 413, 'body_too_large', 'Fragment too large'); }
+ let frag;
+ try { frag = sdpTool.parseIceFragment(body); } catch { return sendWhipError(res, 400, 'invalid_sdpfrag', 'Invalid trickle-ice-sdpfrag'); }
+
+ if (frag.ufrag && frag.pwd && frag.ufrag !== entry.remoteUfrag) {
+ let iceParameters;
+ try { iceParameters = await sfu.restartIce(sess.sessionId, entry.peerId, entry.transportId); } catch (err) {
+ log.warn(`[webrtc] WHIP ICE restart failed for ${sess.sessionId}: ${err.message}`);
+ return sendWhipError(res, 500, 'ice_restart_failed', 'ICE restart failed');
+ }
+ if (!sess.whip.has(resourceId)) return sendWhipError(res, 404, 'session_not_found', 'Session not found');
+ entry.remoteUfrag = frag.ufrag;
+ entry.etag = newIceEtag();
+ log.log(`[webrtc] WHIP ICE restart for ${sess.sessionId}`);
+ res.writeHead(200, { ...WHIP_CORS, 'content-type': 'application/trickle-ice-sdpfrag', ETag: entry.etag });
+ return res.end(sdpTool.buildIceFragment(iceParameters));
+ }
+ res.writeHead(204, WHIP_CORS);
+ res.end();
}
}
async function handleWhipDelete(req, res, resourceId) {
async function handleWhipDelete(req, res, resourceId) {
- let found = null;
- for (const sess of sessions.values()) if (sess.whip.has(resourceId)) { found = sess; break; }
- if (found) await endSession(found.sessionId, 'whip_delete');
+ const found = findWhip(resourceId);
+ if (!found) return sendWhipError(res, 404, 'session_not_found', 'Session not found');
+ if (!whipResourceAuthorized(req, found.entry)) return sendWhipError(res, 401, 'unauthorized', 'Bearer token required for this resource');
+ await endSession(found.sess.sessionId, 'whip_delete');
res.writeHead(200, WHIP_CORS);
res.writeHead(200, WHIP_CORS);
res.end();
res.end();
}
}
@@ -409,8 +472,9 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
@@ -409,8 +472,9 @@ function createWebrtc({ rt, log = console, exit = (code) => process.exit(code),
const kMatch = WHIP_KEY_RE.exec(url.pathname);
const kMatch = WHIP_KEY_RE.exec(url.pathname);
const sMatch = WHIP_SESSION_RE.exec(url.pathname);
const sMatch = WHIP_SESSION_RE.exec(url.pathname);
if (req.method === 'POST' && kMatch) return safe(handleWhipPost(req, res, decodeURIComponent(kMatch[1])), 'whip post');
if (req.method === 'POST' && kMatch) return safe(handleWhipPost(req, res, decodeURIComponent(kMatch[1])), 'whip post');
- if (req.method === 'PATCH' && sMatch) return handleWhipPatch(req, res, sMatch[1]);
+ if (req.method === 'PATCH' && sMatch) return safe(handleWhipPatch(req, res, sMatch[1]), 'whip patch');
if (req.method === 'DELETE' && sMatch) return safe(handleWhipDelete(req, res, sMatch[1]), 'whip delete');
if (req.method === 'DELETE' && sMatch) return safe(handleWhipDelete(req, res, sMatch[1]), 'whip delete');
+ if (sMatch || kMatch) { res.writeHead(405, { ...WHIP_CORS, Allow: sMatch ? 'PATCH, DELETE, OPTIONS' : 'POST, OPTIONS' }); return res.end(); }
res.writeHead(404, WHIP_CORS);
res.writeHead(404, WHIP_CORS);
return res.end();
return res.end();
}
}
@@ -261,15 +261,37 @@ function buildSdpAnswer(transport, offerSdp, producersByKind, { serverName = 'Op
@@ -261,15 +261,37 @@ function buildSdpAnswer(transport, offerSdp, producersByKind, { serverName = 'Op
return write(sdpObj);
return write(sdpObj);
}
}
-/** How many ICE candidates a trickle-PATCH body carries (used only for logging). */
-function countCandidates(sdpText) {
- if (typeof sdpText !== 'string' || !sdpText.trim()) return 0;
- try {
- const parsed = parse(sdpText);
- return Array.isArray(parsed.media) ? parsed.media.reduce((sum, m) => sum + (Array.isArray(m.candidates) ? m.candidates.length : 0), 0) : 0;
- } catch {
- return 0;
+const FRAG_LINE_RE = /^[a-z]=\S/;
+const CANDIDATE_RE = /^candidate:\S+ \d+ \S+ \d+ \S+ \d+ typ \S+/i;
+
+/**
+ * Read a WHIP PATCH body, an RFC 8840 "application/trickle-ice-sdpfrag": the ICE credentials it
+ * names (session or media level, the first wins) and how many candidates it carries. Throws
+ * (code 'invalid_sdpfrag') on anything that is not an SDP fragment or on a malformed candidate.
+ */
+function parseIceFragment(text) {
+ if (typeof text !== 'string') throw Object.assign(new Error('missing sdpfrag'), { code: 'invalid_sdpfrag' });
+ const out = { ufrag: null, pwd: null, candidates: 0, endOfCandidates: false };
+ for (const raw of text.split(/\r?\n/)) {
+ const line = raw.trim();
+ if (!line) continue;
+ if (!FRAG_LINE_RE.test(line)) throw Object.assign(new Error(`not an SDP line: ${line.slice(0, 40)}`), { code: 'invalid_sdpfrag' });
+ if (!line.startsWith('a=')) continue;
+ const attr = line.slice(2);
+ if (attr.startsWith('ice-ufrag:')) out.ufrag = out.ufrag || attr.slice(10).trim();
+ else if (attr.startsWith('ice-pwd:')) out.pwd = out.pwd || attr.slice(8).trim();
+ else if (attr === 'end-of-candidates') out.endOfCandidates = true;
+ else if (attr.startsWith('candidate:')) {
+ if (!CANDIDATE_RE.test(attr)) throw Object.assign(new Error('malformed candidate'), { code: 'invalid_sdpfrag' });
+ out.candidates++;
+ }
}
}
+ return out;
+}
+
+/** The sdpfrag a WHIP resource answers an ICE restart with: the media server's new credentials. */
+function buildIceFragment(iceParameters) {
+ return ['a=ice-lite', `a=ice-ufrag:${iceParameters.usernameFragment}`, `a=ice-pwd:${iceParameters.password}`, ''].join('\r\n');
}
}
/**
/**
@@ -297,5 +319,5 @@ function buildEgressSdp({ video, audio } = {}) {
@@ -297,5 +319,5 @@ function buildEgressSdp({ video, audio } = {}) {
module.exports = {
module.exports = {
available, parse, write,
available, parse, write,
- getDtlsSetupAttribute, selectDtlsFingerprint, extractDtlsParameters, extractRtpParameters, buildSdpAnswer, countCandidates, buildEgressSdp,
+ getDtlsSetupAttribute, selectDtlsFingerprint, extractDtlsParameters, extractRtpParameters, buildSdpAnswer, parseIceFragment, buildIceFragment, buildEgressSdp,
};
};
@@ -112,6 +112,17 @@ function createSfu({ config, log = console }) {
@@ -112,6 +112,17 @@ function createSfu({ config, log = console }) {
return transport;
return transport;
}
}
+ /**
+ * WHIP ICE restart (RFC 9725 §4.3.2). mediasoup's WebRtcTransport is ICE-lite and always the
+ * controlled agent: it has no remote candidates to add (it learns the peer's address from the
+ * peer's connectivity checks), so a restart only issues new local credentials.
+ */
+ async function restartIce(sessionId, peerId, transportId) {
+ const transport = lookupTransport(sessionId, peerId, transportId);
+ if (!transport) throw new Error('Transport not found');
+ return await transport.restartIce();
+ }
+
async function produce(sessionId, peerId, transportId, kind, rtpParameters) {
async function produce(sessionId, peerId, transportId, kind, rtpParameters) {
const r = room(sessionId);
const r = room(sessionId);
if (!r) throw new Error('Router not found');
if (!r) throw new Error('Router not found');
@@ -270,7 +281,7 @@ function createSfu({ config, log = console }) {
@@ -270,7 +281,7 @@ function createSfu({ config, log = console }) {
off: (...args) => emitter.off(...args),
off: (...args) => emitter.off(...args),
init, setAnnouncedIp,
init, setAnnouncedIp,
getOrCreateRouter, hasRouter, getRouterCapabilities,
getOrCreateRouter, hasRouter, getRouterCapabilities,
- createTransport, connectTransport, produce, consume,
+ createTransport, connectTransport, restartIce, produce, consume,
createPlainConsumer, closePlainConsumer,
createPlainConsumer, closePlainConsumer,
findProducerByKind, hasProducers, getProducers, getViewerCount,
findProducerByKind, hasProducers, getProducers, getViewerCount,
closePeer, closeSession, closeAll,
closePeer, closeSession, closeAll,