From 41fd62490de6d5e0ae9dc859ffde1c50684d4375 Mon Sep 17 00:00:00 2001 From: UN Date: Thu, 22 Oct 2026 20:53:47 +0500 Subject: [PATCH] Follow key provider key index on media frame cryptors Ports the semantics merged in livekit/client-sdk-swift#1157 to Android, superseding the approach in #1039. BaseKeyProvider now implements the new KeyIndexObservable contract: the current key index is the single source of truth for what the local participant encrypts with. Installing a shared key moves the index and setCurrentKeyIndex() exposes it directly; getLatestKeyIndex() returns it for both media senders and data packets. E2EEManager subscribes to the provider and re-stamps the frame cryptors of already-published local tracks on the RTC thread whenever the index changes, and stamps new senders at creation. Receivers are unaffected: they read the index from each incoming frame. Cryptor map iteration is synchronized, and a serialized dispatch that re-reads the latest index prevents overlapping rotations from leaving senders stale. Behavior change: in non-shared key mode setKey() no longer moves the sender index, since the key may belong to a remote participant. Use setCurrentKeyIndex() to switch the local one. Custom KeyProvider implementations opt into live re-pointing by implementing KeyIndexObservable. --- .changeset/sender-key-index-observer.md | 5 + .../detekt-baseline-release.xml | 1 + .../io/livekit/android/e2ee/E2EEManager.kt | 49 +++- .../io/livekit/android/e2ee/KeyProvider.kt | 89 +++++++- .../android/e2ee/FrameCryptorKeyIndexTest.kt | 210 ++++++++++++++++++ 5 files changed, 339 insertions(+), 15 deletions(-) create mode 100644 .changeset/sender-key-index-observer.md create mode 100644 livekit-android-test/src/test/java/io/livekit/android/e2ee/FrameCryptorKeyIndexTest.kt diff --git a/.changeset/sender-key-index-observer.md b/.changeset/sender-key-index-observer.md new file mode 100644 index 000000000..fc750d7c1 --- /dev/null +++ b/.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. diff --git a/livekit-android-sdk/detekt-baseline-release.xml b/livekit-android-sdk/detekt-baseline-release.xml index 6bd3cb94e..950f68845 100644 --- a/livekit-android-sdk/detekt-baseline-release.xml +++ b/livekit-android-sdk/detekt-baseline-release.xml @@ -80,6 +80,7 @@ TooManyFunctions:CommunicationWorkaround.kt$CommunicationWorkaroundImpl : CommunicationWorkaround TooManyFunctions:E2EEManager.kt$E2EEManager TooManyFunctions:IncomingDataStreamManager.kt$IncomingDataStreamManagerImpl : IncomingDataStreamManager + TooManyFunctions:KeyProvider.kt$BaseKeyProvider : KeyProviderKeyIndexObservable TooManyFunctions:LKLog.kt$LKLog$Companion TooManyFunctions:LocalParticipant.kt$LocalParticipant : ParticipantOutgoingDataStreamManagerRpcManager TooManyFunctions:LocalVideoTrack.kt$LocalVideoTrack : VideoTrack diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt b/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt index 40913db63..f4cc3abed 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt +++ b/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,6 +42,7 @@ 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 @@ -47,9 +50,10 @@ constructor( @Assisted val keyProvider: KeyProvider, val peerConnectionFactory: PeerConnectionFactory, dataPacketCryptorManagerFactory: DataPacketCryptorManager.Factory, + private val rtcThreadToken: RTCThreadToken, ) { private var room: Room? = null - private val frameCryptors = mutableMapOf, FrameCryptor>() + private val frameCryptors = Collections.synchronizedMap(mutableMapOf, 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() } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/KeyProvider.kt b/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/KeyProvider.kt index 55dc57abd..5abc00900 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/KeyProvider.kt +++ b/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 Unit>() + + @Volatile + private var _currentKeyIndex: Int = 0 - private val latestSetIndex = mutableMapOf() + 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) } } } diff --git a/livekit-android-test/src/test/java/io/livekit/android/e2ee/FrameCryptorKeyIndexTest.kt b/livekit-android-test/src/test/java/io/livekit/android/e2ee/FrameCryptorKeyIndexTest.kt new file mode 100644 index 000000000..a4f1b0cf6 --- /dev/null +++ b/livekit-android-test/src/test/java/io/livekit/android/e2ee/FrameCryptorKeyIndexTest.kt @@ -0,0 +1,210 @@ +/* + * Copyright 2026 LiveKit, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.livekit.android.e2ee + +import io.livekit.android.room.Room +import io.livekit.android.room.participant.LocalParticipant +import io.livekit.android.room.participant.Participant +import io.livekit.android.room.participant.RemoteParticipant +import io.livekit.android.room.track.LocalAudioTrack +import io.livekit.android.room.track.RemoteAudioTrack +import io.livekit.android.room.track.Track +import io.livekit.android.room.track.TrackPublication +import io.livekit.android.test.BaseTest +import io.livekit.android.webrtc.peerconnection.RTCThreadToken +import livekit.org.webrtc.FrameCryptor +import livekit.org.webrtc.FrameCryptorFactory +import livekit.org.webrtc.FrameCryptorKeyProvider +import livekit.org.webrtc.PeerConnectionFactory +import livekit.org.webrtc.RtpReceiver +import livekit.org.webrtc.RtpSender +import org.junit.Assert.assertEquals +import org.junit.Test +import org.mockito.Mockito.mockStatic +import org.mockito.Mockito.never +import org.mockito.kotlin.any +import org.mockito.kotlin.mock +import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever + +class FrameCryptorKeyIndexTest : BaseTest() { + + private val rtcKeyProvider = mock() + private val senderCryptor = mock() + private val receiverCryptor = mock() + private val rtcThreadToken = object : RTCThreadToken { + override val isDisposed: Boolean = false + } + + private fun mockedFactory() = mockStatic(FrameCryptorFactory::class.java).apply { + `when` { + FrameCryptorFactory.createFrameCryptorKeyProvider( + any(), + any(), + any(), + any(), + any(), + any(), + any(), + any(), + ) + }.thenReturn(rtcKeyProvider) + `when` { + FrameCryptorFactory.createFrameCryptorForRtpSender(any(), any(), any(), any(), any()) + }.thenReturn(senderCryptor) + `when` { + FrameCryptorFactory.createFrameCryptorForRtpReceiver(any(), any(), any(), any(), any()) + }.thenReturn(receiverCryptor) + } + + private fun createProvider(sharedKey: Boolean = true): BaseKeyProvider { + whenever(rtcKeyProvider.setSharedKey(any(), any())).thenReturn(true) + return BaseKeyProvider(enableSharedKey = sharedKey) + } + + private fun createManager(provider: KeyProvider): Triple { + val room = mock() + val localParticipant = mock() + whenever(room.localParticipant).thenReturn(localParticipant) + whenever(room.remoteParticipants).thenReturn(emptyMap()) + whenever(localParticipant.identity).thenReturn(Participant.Identity("local")) + whenever(localParticipant.trackPublications).thenReturn(emptyMap()) + + val manager = E2EEManager( + provider, + mock(), + object : DataPacketCryptorManager.Factory { + override fun create(keyProvider: KeyProvider) = mock() + }, + rtcThreadToken, + ) + manager.setup(room) { } + return Triple(manager, room, localParticipant) + } + + private fun addLocalSender(manager: E2EEManager, room: Room, participant: LocalParticipant): FrameCryptor { + val track = mock() + whenever(track.sender).thenReturn(mock()) + whenever(track.kind).thenReturn(Track.Kind.AUDIO) + val publication = mock() + whenever(publication.track).thenReturn(track) + whenever(publication.sid).thenReturn("sender-sid") + + manager.addPublishedTrack(track, publication, participant, room) + return senderCryptor + } + + private fun addRemoteReceiver(manager: E2EEManager, room: Room): FrameCryptor { + val participant = mock() + whenever(participant.identity).thenReturn(Participant.Identity("remote")) + val track = mock() + whenever(track.receiver).thenReturn(mock()) + whenever(track.kind).thenReturn(Track.Kind.AUDIO) + val publication = mock() + whenever(publication.track).thenReturn(track) + whenever(publication.sid).thenReturn("receiver-sid") + + manager.addSubscribedTrack(track, publication, participant, room) + return receiverCryptor + } + + @Test + fun sharedKeyInstallMovesCurrentKeyIndex() { + mockedFactory().use { + val provider = createProvider() + provider.setSharedKey("key", 3) + assertEquals(3, provider.currentKeyIndex) + assertEquals(3, provider.getLatestKeyIndex("anyone")) + } + } + + @Test + fun setCurrentKeyIndexWrapsModuloKeyRingSize() { + mockedFactory().use { + val provider = createProvider() + provider.setCurrentKeyIndex(20) + assertEquals(20 % 16, provider.currentKeyIndex) + } + } + + @Test + fun nonSharedSetKeyDoesNotMoveCurrentKeyIndex() { + mockedFactory().use { + val provider = createProvider(sharedKey = false) + // A remote participant's key install cannot redirect the local sender index. + provider.setKey("key", "remote", 7) + assertEquals(0, provider.currentKeyIndex) + assertEquals(0, provider.getLatestKeyIndex("remote")) + } + } + + @Test + fun observerSeesLatestIndexAndRemovalStopsCallbacks() { + mockedFactory().use { + val provider = createProvider() + val seen = mutableListOf() + provider.addKeyIndexObserver(this) { seen.add(it) } + + provider.setCurrentKeyIndex(3) + provider.setCurrentKeyIndex(5) + assertEquals(listOf(3, 5), seen) + + provider.removeKeyIndexObserver(this) + provider.setCurrentKeyIndex(7) + assertEquals(listOf(3, 5), seen) + } + } + + @Test + fun newSendersAreStampedWithCurrentKeyIndex() { + mockedFactory().use { + val provider = createProvider() + provider.setSharedKey("key", 4) + + val (manager, room, localParticipant) = createManager(provider) + addLocalSender(manager, room, localParticipant) + + verify(senderCryptor).keyIndex = 4 + } + } + + @Test + fun existingSendersFollowKeyIndexChanges() { + mockedFactory().use { + val provider = createProvider() + val (manager, room, localParticipant) = createManager(provider) + addLocalSender(manager, room, localParticipant) + + provider.setSharedKey("rotated", 9) + verify(senderCryptor).keyIndex = 9 + } + } + + @Test + fun receiversAreNotRestamped() { + mockedFactory().use { + val provider = createProvider() + val (manager, room, localParticipant) = createManager(provider) + addLocalSender(manager, room, localParticipant) + addRemoteReceiver(manager, room) + + provider.setSharedKey("rotated", 9) + verify(senderCryptor).keyIndex = 9 + verify(receiverCryptor, never()).keyIndex = 9 + } + } +}