working with livekit settings

This commit is contained in:
Thomas Ramé
2026-04-07 16:09:27 +02:00
parent 267b39a975
commit 7105c6bc70
2 changed files with 151 additions and 75 deletions
@@ -126,6 +126,19 @@ export class VaultE2EEManager extends EventEmitter {
case 'error': case 'error':
this.emit(EncryptionEvent.EncryptionError, data.error, data.participantIdentity) this.emit(EncryptionEvent.EncryptionError, data.error, data.participantIdentity)
break break
case 'pipeDead':
// Pipe ended (stream closed during disconnect). Clear E2EE_FLAG on all
// receivers so the next TrackSubscribed creates a fresh pipe.
this.room?.remoteParticipants.forEach((p) => {
p.trackPublications.forEach((pub) => {
if (pub.track?.receiver && E2EE_FLAG in pub.track.receiver) {
// @ts-expect-error
delete pub.track.receiver[E2EE_FLAG]
}
})
})
break
} }
} }
@@ -221,20 +234,19 @@ export class VaultE2EEManager extends EventEmitter {
return return
} }
// @ts-expect-error let writable: WritableStream
let writable: WritableStream = receiver.writableStream let readable: ReadableStream
// @ts-expect-error
let readable: ReadableStream = receiver.readableStream
if (!writable || !readable) { try {
// @ts-expect-error // @ts-expect-error
const receiverStreams = receiver.createEncodedStreams() const receiverStreams = receiver.createEncodedStreams()
// @ts-expect-error
receiver.writableStream = receiverStreams.writable
writable = receiverStreams.writable writable = receiverStreams.writable
// @ts-expect-error
receiver.readableStream = receiverStreams.readable
readable = receiverStreams.readable readable = receiverStreams.readable
} catch {
// createEncodedStreams() already called (receiver reuse after pipe death).
// Cannot re-create streams — this receiver is stuck.
console.warn(`[VaultE2EE] Cannot create encoded streams for ${participantIdentity} (receiver reuse). Pipe unrecoverable.`)
return
} }
this.worker.postMessage( this.worker.postMessage(
@@ -1,35 +1,42 @@
/** /**
* E2EE Worker using libsodium XChaCha20-Poly1305. * E2EE Worker — Step 2d: uses crypto.subtle AES-GCM with the SAME frame format
* Mirrors the structure of LiveKit's e2ee.worker but with libsodium crypto. * as LiveKit's built-in FrameCryptor, including preserved unencrypted header bytes.
* Receives encoded streams via transfer and pipes them with encrypt/decrypt transforms. *
* Frame format (same as LiveKit):
* [unencrypted header][ciphertext + GCM tag][IV (12B)][IV_LENGTH (1B)][key index (1B)]
*
* Unencrypted header sizes (VP8):
* - keyframe: 10 bytes
* - delta: 3 bytes
* - audio: 1 byte (Opus TOC)
*/ */
import _sodium from 'libsodium-wrappers-sumo'
let sodium: typeof _sodium let encryptionKey: CryptoKey | null = null
let symmetricKey: Uint8Array | null = null const IV_LENGTH = 12
const KEY_INDEX = 0
const transforms = new Map<string, { cancel: () => void }>() // Same constants as LiveKit's FrameCryptor
const UNENCRYPTED_BYTES = {
async function init() { key: 10, // VP8 keyframe
await _sodium.ready delta: 3, // VP8 delta frame
sodium = _sodium audio: 1, // Opus TOC byte
} }
const sodiumReady = init() function getUnencryptedBytes(frame: RTCEncodedVideoFrame | RTCEncodedAudioFrame): number {
// Audio frames don't have .type
function encrypt(plaintext: Uint8Array): Uint8Array { if (!('type' in frame)) {
const nonce = sodium.randombytes_buf(sodium.crypto_secretbox_NONCEBYTES) return UNENCRYPTED_BYTES.audio
const ciphertext = sodium.crypto_secretbox_easy(plaintext, nonce, symmetricKey!) }
const result = new Uint8Array(nonce.length + ciphertext.length) return frame.type === 'key' ? UNENCRYPTED_BYTES.key : UNENCRYPTED_BYTES.delta
result.set(nonce)
result.set(ciphertext, nonce.length)
return result
} }
function decrypt(data: Uint8Array): Uint8Array { function makeIV(ssrc: number, timestamp: number): Uint8Array {
const nonce = data.slice(0, sodium.crypto_secretbox_NONCEBYTES) const iv = new ArrayBuffer(IV_LENGTH)
const ciphertext = data.slice(sodium.crypto_secretbox_NONCEBYTES) const view = new DataView(iv)
return sodium.crypto_secretbox_open_easy(ciphertext, nonce, symmetricKey!) view.setUint32(0, ssrc, true)
view.setUint32(4, timestamp, true)
view.setUint32(8, ssrc ^ timestamp, true)
return new Uint8Array(iv)
} }
onmessage = async (ev: MessageEvent) => { onmessage = async (ev: MessageEvent) => {
@@ -37,86 +44,143 @@ onmessage = async (ev: MessageEvent) => {
switch (kind) { switch (kind) {
case 'init': case 'init':
await sodiumReady
postMessage({ kind: 'initAck', data: { enabled: true } }) postMessage({ kind: 'initAck', data: { enabled: true } })
break break
case 'setKey': case 'setKey': {
symmetricKey = data.key encryptionKey = await crypto.subtle.importKey(
// Echo back enable for the participant 'raw', data.key, { name: 'AES-GCM', length: 256 }, false, ['encrypt', 'decrypt'],
)
postMessage({ postMessage({
kind: 'enable', kind: 'enable',
data: { enabled: true, participantIdentity: data.participantIdentity }, data: { enabled: true, participantIdentity: data.participantIdentity },
}) })
break break
}
case 'encode': case 'encode':
case 'decode': { case 'decode': {
await sodiumReady
const { readableStream, writableStream, trackId, participantIdentity } = data const { readableStream, writableStream, trackId, participantIdentity } = data
const operation = kind === 'encode' ? 'encrypt' : 'decrypt' const operation = kind
let frameCount = 0
// Cancel existing transform for this track if any
const existing = transforms.get(trackId)
if (existing) existing.cancel()
const abortController = new AbortController()
const transformStream = new TransformStream({ const transformStream = new TransformStream({
transform: async (frame: RTCEncodedVideoFrame, controller: TransformStreamDefaultController) => { transform: async (frame: RTCEncodedVideoFrame | RTCEncodedAudioFrame, controller: TransformStreamDefaultController) => {
if (!symmetricKey || frame.data.byteLength === 0) {
return controller.enqueue(frame)
}
try { try {
const input = new Uint8Array(frame.data) if (!encryptionKey) return // drop — key not ready
const output = operation === 'encrypt' ? encrypt(input) : decrypt(input) if (!frame.data || frame.data.byteLength === 0) {
frame.data = output.buffer return controller.enqueue(frame)
controller.enqueue(frame) }
} catch (e) {
// Encrypt: drop frame (never send unencrypted) if (operation === 'encode') {
// Decrypt: emit error // ── Encrypt (same as LiveKit FrameCryptor.encodeFunction) ──
if (operation === 'decrypt') { const iv = makeIV(
postMessage({ (frame as any).getMetadata?.().synchronizationSource ?? 0,
kind: 'error', (frame as any).timestamp ?? 0,
data: { )
error: new Error(`Decryption failed for ${participantIdentity}`),
participantIdentity, const unencryptedBytes = getUnencryptedBytes(frame)
const frameHeader = new Uint8Array(frame.data, 0, unencryptedBytes)
const ciphertext = await crypto.subtle.encrypt(
{
name: 'AES-GCM',
iv,
additionalData: new Uint8Array(frame.data, 0, frameHeader.byteLength),
}, },
}) encryptionKey,
new Uint8Array(frame.data, unencryptedBytes),
)
// [header][ciphertext+tag][IV][IV_LENGTH][keyIndex]
const frameTrailer = new Uint8Array(2)
frameTrailer[0] = IV_LENGTH
frameTrailer[1] = KEY_INDEX
const newData = new Uint8Array(
frameHeader.byteLength + ciphertext.byteLength + iv.byteLength + frameTrailer.byteLength,
)
newData.set(frameHeader)
newData.set(new Uint8Array(ciphertext), frameHeader.byteLength)
newData.set(iv, frameHeader.byteLength + ciphertext.byteLength)
newData.set(frameTrailer, frameHeader.byteLength + ciphertext.byteLength + iv.byteLength)
frame.data = newData.buffer
controller.enqueue(frame)
} else {
// ── Decrypt (same as LiveKit FrameCryptor.decodeFunction) ──
const frameData = new Uint8Array(frame.data)
const unencryptedBytes = getUnencryptedBytes(frame)
const frameHeader = new Uint8Array(frame.data, 0, unencryptedBytes)
// Read trailer
const frameTrailer = new Uint8Array(frame.data, frame.data.byteLength - 2, 2)
const ivLength = frameTrailer[0]
// Extract IV
const iv = new Uint8Array(
frame.data,
frame.data.byteLength - ivLength - frameTrailer.byteLength,
ivLength,
)
// Extract ciphertext (between header and IV)
const ciphertextStart = frameHeader.byteLength
const ciphertextLength = frame.data.byteLength - frameHeader.byteLength - ivLength - frameTrailer.byteLength
const plaintext = await crypto.subtle.decrypt(
{
name: 'AES-GCM',
iv,
additionalData: new Uint8Array(frame.data, 0, frameHeader.byteLength),
},
encryptionKey,
new Uint8Array(frame.data, ciphertextStart, ciphertextLength),
)
// Reconstruct: [header][plaintext]
const newData = new Uint8Array(frameHeader.byteLength + plaintext.byteLength)
newData.set(frameHeader)
newData.set(new Uint8Array(plaintext), frameHeader.byteLength)
frame.data = newData.buffer
controller.enqueue(frame)
}
frameCount++
if (frameCount <= 5 || frameCount % 500 === 0) {
console.log(`[Worker] ${operation} frame #${frameCount} for ${participantIdentity}, ${frame.data.byteLength}B`)
}
} catch (e) {
// Swallow to keep pipe alive
if (frameCount < 10) {
console.error(`[Worker] ${operation} error for ${participantIdentity}:`, (e as Error)?.message)
} }
} }
}, },
}) })
readableStream readableStream
.pipeThrough(transformStream, { signal: abortController.signal }) .pipeThrough(transformStream)
.pipeTo(writableStream) .pipeTo(writableStream)
.catch(() => {}) .then(() => console.warn(`[Worker] pipe completed for ${participantIdentity}/${trackId}`))
.catch((err: Error) => console.warn(`[Worker] pipe error for ${participantIdentity}/${trackId}:`, err?.message))
transforms.set(trackId, { cancel: () => abortController.abort() })
break break
} }
case 'removeTransform':
// Do NOT cancel the pipe. The transferred streams stay open across track
// changes on reused receivers. The existing pipe will process frames from
// the new track. Cancelling would kill the pipe permanently since
// createEncodedStreams() can only be called once per receiver.
break
case 'enable': case 'enable':
// Echo back
postMessage({ postMessage({
kind: 'enable', kind: 'enable',
data: { enabled: data.enabled, participantIdentity: data.participantIdentity }, data: { enabled: data.enabled, participantIdentity: data.participantIdentity },
}) })
break break
case 'removeTransform':
case 'setRTPMap': case 'setRTPMap':
case 'setSifTrailer': case 'setSifTrailer':
case 'updateCodec': case 'updateCodec':
case 'ratchetRequest': case 'ratchetRequest':
// Not needed for libsodium — ignore
break break
} }
} }