Follow key provider key index on media frame cryptors

.changeset/sender-key-index-observer.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": patch
+---
+
+Media frame cryptors now follow the key provider's current key index, matching the JavaScript and Swift SDKs. `BaseKeyProvider` implements the new `KeyIndexObservable` contract: installing a shared key moves the current index, `setCurrentKeyIndex` re-points live senders, and `getLatestKeyIndex` returns the current index. Custom `KeyProvider` implementations can opt into live re-pointing by implementing `KeyIndexObservable`. Behavior change: in non-shared key mode `setKey` no longer moves the sender index — the key may belong to a remote participant — use `setCurrentKeyIndex` to switch the local one.

livekit-android-sdk/detekt-baseline-release.xml

@@ -80,6 +80,7 @@
     <ID>TooManyFunctions:CommunicationWorkaround.kt$CommunicationWorkaroundImpl : CommunicationWorkaround</ID>
     <ID>TooManyFunctions:E2EEManager.kt$E2EEManager</ID>
     <ID>TooManyFunctions:IncomingDataStreamManager.kt$IncomingDataStreamManagerImpl : IncomingDataStreamManager</ID>
+    <ID>TooManyFunctions:KeyProvider.kt$BaseKeyProvider : KeyProviderKeyIndexObservable</ID>
     <ID>TooManyFunctions:LKLog.kt$LKLog$Companion</ID>
     <ID>TooManyFunctions:LocalParticipant.kt$LocalParticipant : ParticipantOutgoingDataStreamManagerRpcManager</ID>
     <ID>TooManyFunctions:LocalVideoTrack.kt$LocalVideoTrack : VideoTrack</ID>

livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt

@@ -32,6 +32,8 @@ import io.livekit.android.room.track.RemoteVideoTrack
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.LKLog
+import io.livekit.android.webrtc.peerconnection.RTCThreadToken
+import io.livekit.android.webrtc.peerconnection.executeOnRTCThread
 import livekit.LivekitModels.Encryption
 import livekit.org.webrtc.FrameCryptor
 import livekit.org.webrtc.FrameCryptor.FrameCryptionState
@@ -40,16 +42,18 @@ import livekit.org.webrtc.FrameCryptorFactory
 import livekit.org.webrtc.PeerConnectionFactory
 import livekit.org.webrtc.RtpReceiver
 import livekit.org.webrtc.RtpSender
+import java.util.Collections
 
 class E2EEManager
 @AssistedInject
 constructor(
     @Assisted val keyProvider: KeyProvider,
     val peerConnectionFactory: PeerConnectionFactory,
     dataPacketCryptorManagerFactory: DataPacketCryptorManager.Factory,
+    private val rtcThreadToken: RTCThreadToken,
 ) {
     private var room: Room? = null
-    private val frameCryptors = mutableMapOf<Pair<String, Participant.Identity>, FrameCryptor>()
+    private val frameCryptors = Collections.synchronizedMap(mutableMapOf<Pair<String, Participant.Identity>, FrameCryptor>())
     private var algorithm: FrameCryptorAlgorithm = FrameCryptorAlgorithm.AES_GCM
     private lateinit var emitEvent: (roomEvent: RoomEvent) -> Unit?
 
@@ -58,9 +62,11 @@ constructor(
     var enabled: Boolean = false
         set(value) {
             field = value
-            for (item in frameCryptors.entries) {
-                val frameCryptor = item.value
-                frameCryptor.isEnabled = enabled
+            synchronized(frameCryptors) {
+                for (item in frameCryptors.entries) {
+                    val frameCryptor = item.value
+                    frameCryptor.isEnabled = enabled
+                }
             }
         }
 
@@ -96,6 +102,7 @@ constructor(
         this.enabled = true
         this.room = room
         this.emitEvent = emitEvent
+        observeKeyIndex()
         this.room?.localParticipant?.trackPublications?.forEach { item ->
             val participant = this.room!!.localParticipant
             val publication = item.value
@@ -242,14 +249,42 @@ constructor(
         LKLog.d { "ratchetSharedKey: newKey: $newKey" }
     }
 
+    /**
+     * Re-stamps the frame cryptors of every track the local participant has
+     * already published with the provider's latest key index. Runs on the
+     * RTC thread like all other frame cryptor calls.
+     */
+    internal fun setSenderKeyIndex(keyIndex: Int) {
+        val localIdentity = room?.localParticipant?.identity ?: return
+        executeOnRTCThread(rtcThreadToken) {
+            synchronized(frameCryptors) {
+                for ((key, frameCryptor) in frameCryptors) {
+                    if (key.second == localIdentity) {
+                        frameCryptor.keyIndex = keyIndex
+                    }
+                }
+            }
+        }
+    }
+
+    private fun observeKeyIndex() {
+        (keyProvider as? KeyIndexObservable)?.addKeyIndexObserver(this) { index ->
+            setSenderKeyIndex(index)
+        }
+    }
+
     internal fun cleanup() {
-        for (frameCryptor in frameCryptors.values) {
-            frameCryptor.dispose()
+        (keyProvider as? KeyIndexObservable)?.removeKeyIndexObserver(this)
+        synchronized(frameCryptors) {
+            for (frameCryptor in frameCryptors.values) {
+                frameCryptor.dispose()
+            }
+            frameCryptors.clear()
         }
-        frameCryptors.clear()
     }
 
     internal fun dispose() {
+        (keyProvider as? KeyIndexObservable)?.removeKeyIndexObserver(this)
         dataPacketCryptorManager.dispose()
     }
 

livekit-android-sdk/src/main/java/io/livekit/android/e2ee/KeyProvider.kt

@@ -42,6 +42,34 @@ interface KeyProvider {
     var enableSharedKey: Boolean
 }
 
+/**
+ * Implemented by [KeyProvider]s that track the key index the local
+ * participant encrypts with and can push changes to observers.
+ *
+ * [E2EEManager] subscribes so the frame cryptors of media tracks that are
+ * already published get re-stamped whenever the index changes. Receivers
+ * are unaffected: they read the index from each incoming frame.
+ */
+interface KeyIndexObservable {
+
+    /**
+     * The key index the local participant encrypts with.
+     */
+    val currentKeyIndex: Int
+
+    /**
+     * Registers [observer] under [owner]; [removeKeyIndexObserver] with the
+     * same owner detaches it. The observer is invoked with the latest index
+     * on every change and may be called from any thread.
+     */
+    fun addKeyIndexObserver(owner: Any, observer: (keyIndex: Int) -> Unit)
+
+    /**
+     * Detaches the observer previously registered under [owner].
+     */
+    fun removeKeyIndexObserver(owner: Any)
+}
+
 class BaseKeyProvider(
     ratchetSalt: String = defaultRatchetSalt,
     uncryptedMagicBytes: String = defaultMagicBytes,
@@ -51,9 +79,18 @@ class BaseKeyProvider(
     keyRingSize: Int = defaultKeyRingSize,
     discardFrameWhenCryptorNotReady: Boolean = defaultDiscardFrameWhenCryptorNotReady,
     keyDerivationAlgorithm: FrameCryptorKeyDerivationAlgorithm = defaultKeyDerivationAlgorithm,
-) : KeyProvider {
+) : KeyProvider, KeyIndexObservable {
+
+    private val keyRingSize: Int = keyRingSize
+    private val keyIndexStateLock = Any()
+    private val keyIndexDispatchLock = Any()
+    private val keyIndexObservers = mutableMapOf<Any, (Int) -> Unit>()
+
+    @Volatile
+    private var _currentKeyIndex: Int = 0
 
-    private val latestSetIndex = mutableMapOf<String, Int>()
+    override val currentKeyIndex: Int
+        get() = _currentKeyIndex
 
     override val rtcKeyProvider: FrameCryptorKeyProvider = FrameCryptorFactory.createFrameCryptorKeyProvider(
         enableSharedKey,
@@ -67,7 +104,11 @@ class BaseKeyProvider(
     )
 
     override fun setSharedKey(key: String, keyIndex: Int?): Boolean {
-        return rtcKeyProvider.setSharedKey(keyIndex ?: 0, key.toByteArray())
+        return rtcKeyProvider.setSharedKey(keyIndex ?: 0, key.toByteArray()).also { succeeded ->
+            if (succeeded) {
+                setCurrentKeyIndex(keyIndex ?: 0)
+            }
+        }
     }
 
     override fun ratchetSharedKey(keyIndex: Int?): ByteArray {
@@ -94,10 +135,7 @@ class BaseKeyProvider(
             return
         }
 
-        val keyIndex = keyIndex ?: 0
-        latestSetIndex[participantId] = keyIndex
-
-        rtcKeyProvider.setKey(participantId, keyIndex, key.toByteArray())
+        rtcKeyProvider.setKey(participantId, keyIndex ?: 0, key.toByteArray())
     }
 
     override fun ratchetKey(participantId: String, keyIndex: Int?): ByteArray {
@@ -112,7 +150,42 @@ class BaseKeyProvider(
         rtcKeyProvider.setSifTrailer(trailer)
     }
 
+    /**
+     * Switches the local participant's encryption to the key at [keyIndex]
+     * (modulo the key ring size).
+     *
+     * Takes effect immediately for the media tracks already published and
+     * for every data packet sent afterwards. Receivers are unaffected: they
+     * read the index from each frame.
+     */
+    fun setCurrentKeyIndex(keyIndex: Int) {
+        val observers = synchronized(keyIndexStateLock) {
+            _currentKeyIndex = Math.floorMod(keyIndex, keyRingSize)
+            keyIndexObservers.values.toList()
+        }
+        // Dispatch is serialized and each pass re-reads the newest index,
+        // so overlapping rotations can't leave senders on a stale index.
+        synchronized(keyIndexDispatchLock) {
+            val latest = _currentKeyIndex
+            observers.forEach { it(latest) }
+        }
+    }
+
+    /**
+     * The key index outgoing media frames and data packets are stamped with.
+     * In non-shared key mode [setKey] does not move it — the key may belong
+     * to a remote participant — so switch the local one with
+     * [setCurrentKeyIndex].
+     */
     override fun getLatestKeyIndex(participantId: String): Int {
-        return latestSetIndex[participantId] ?: 0
+        return currentKeyIndex
+    }
+
+    override fun addKeyIndexObserver(owner: Any, observer: (keyIndex: Int) -> Unit) {
+        synchronized(keyIndexStateLock) { keyIndexObservers[owner] = observer }
+    }
+
+    override fun removeKeyIndexObserver(owner: Any) {
+        synchronized(keyIndexStateLock) { keyIndexObservers.remove(owner) }
     }
 }