Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions .changeset/data-streams-v2.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
---
"client-sdk-android": minor
---

Data streams v2: Data streams are now backed by the shared Rust core (`livekit-ffi`), which adds the v2 wire format: a small, known-size payload is sent as a single packet with the content inline in the header, and payloads can be `deflate-raw` compressed. Both are negotiated per recipient, so rooms with older peers keep working.

- `ConnectOptions.clientProtocol` now defaults to `ClientProtocolVersion.DATA_STREAM_V2`.
- New `ClientCapability` enum and `Participant.capabilities`, mirroring the capabilities each peer's client advertises.
- New `RoomOptions.dataStreamOptions` / `Room.dataStreamOptions`, with `DataStreamOptions.maxPayloadByteLength` to cap the size of a reassembled incoming payload.
- New `compress` flag on `StreamTextOptions` and `StreamBytesOptions` (default `true`). Only applies to whole-payload sends such as `sendText`/`sendBytes`; incremental streams are never compressed.
- New `OutgoingDataStreamManager.sendBytes` convenience for sending a whole byte array.
- New `StreamException` types: `HeaderTooLargeException` and `PayloadTooLargeException` (both subclasses of `LengthExceededException`), and `InternalException`. `TerminatedException` gains a `reason`. An exhaustive `when` over `StreamException` needs a branch for the new cases.
- Incoming streams are held to the encryption type their header arrived with; a mismatching chunk or trailer fails the reader with `EncryptionTypeMismatch`.
5 changes: 5 additions & 0 deletions .changeset/validate-region-urls.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"client-sdk-android": patch
---

Validate region URLs before failing over to them. Regions returned by `/settings/regions` or reported by the server were previously used as-is. Each URL must now name a LiveKit Cloud host and use an encrypted scheme (`wss`/`https`); entries that do not are logged and dropped.
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import livekit.LivekitRtc
import okhttp3.OkHttpClient
import okhttp3.Request
import java.net.URI
import java.net.URISyntaxException

/**
* @suppress
Expand Down Expand Up @@ -94,19 +95,35 @@ constructor(
throw RoomException.ConnectException("Could not fetch region settings: empty response body!")
}

return json.decodeFromString<RegionSettings>(bodyString).also {
regionSettings = it
lastUpdateAt = SystemClock.elapsedRealtime()
}
return applyRegionSettings(json.decodeFromString<RegionSettings>(bodyString))
}

fun clearAttemptedRegions() {
attemptedRegions.clear()
}

fun setServerReportedRegions(regionSettings: RegionSettings) {
this.regionSettings = regionSettings
this.lastUpdateAt = SystemClock.elapsedRealtime()
applyRegionSettings(regionSettings)
}

/**
* Stores [settings], keeping only the regions this SDK may fail over to.
*
* Both sources of a region list are network input -- the `/settings/regions` response and the
* regions the server reports over signaling -- and every URL kept here is later connected to
* with the participant token. Each entry is therefore checked on its own merits rather than
* inheriting trust from the configured server URL. See [isTrustedRegionUrl].
*/
private fun applyRegionSettings(settings: RegionSettings): RegionSettings {
val trusted = settings.regions.filter { isTrustedRegionUrl(it.url) }
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
val rejected = settings.regions.size - trusted.size
if (rejected > 0) {
LKLog.w { "Ignoring $rejected of ${settings.regions.size} regions: not an encrypted LiveKit Cloud URL" }
}
return RegionSettings(trusted).also {
regionSettings = it
lastUpdateAt = SystemClock.elapsedRealtime()
}
}

@AssistedFactory
Expand All @@ -117,6 +134,33 @@ constructor(

internal fun URI.isLKCloud() = regionUrlProviderTesting || host != null && (host.endsWith(".livekit.cloud") || host.endsWith(".livekit.run"))

/**
* Whether [url] is a region URL this SDK may fail over to.
*
* A region list arrives over the network and the SDK authenticates to whatever it names, sending
* the participant token as a bearer header, so an entry is accepted only when it parses, names a
* LiveKit Cloud host, and uses an encrypted scheme. The `ws`/`http` schemes are refused so a
* region list cannot move that exchange onto cleartext.
*
* Distinct from [isLKCloud], which asks whether to attempt region discovery at all for the URL
* the application supplied, and which carries a testing override. This check has no override: a
* release build cannot be made to accept an untrusted region.
*/
internal fun isTrustedRegionUrl(url: String): Boolean {
val uri = try {
URI(url)
} catch (e: URISyntaxException) {
LKLog.w(e) { "Ignoring region url that could not be parsed: $url" }
return false
}
val scheme = uri.scheme?.lowercase() ?: return false
if (scheme != "wss" && scheme != "https") {
return false
}
val host = uri.host?.lowercase() ?: return false
return host.endsWith(".livekit.cloud") || host.endsWith(".livekit.run")
Comment thread
davidliu marked this conversation as resolved.
Resolved
}

internal fun URI.getCloudConfigUrl(appendPath: String = ""): URI {
val scheme = if (this.scheme.startsWith("ws")) {
this.scheme.replaceFirst("ws", "http")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@ import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
Expand Down Expand Up @@ -102,8 +104,98 @@ class RegionUrlProviderTest : BaseTest() {
// Still served from the cached settings.
assertEquals(1, server.requestCount)
}

@Test
fun trustedRegionUrlAcceptsEncryptedCloudUrls() {
assertTrue(isTrustedRegionUrl("wss://eu.livekit.cloud"))
assertTrue(isTrustedRegionUrl("https://eu.livekit.cloud"))
assertTrue(isTrustedRegionUrl("https://eu.livekit.cloud:443/rtc"))
assertTrue(isTrustedRegionUrl("wss://staging.livekit.run"))
assertTrue(isTrustedRegionUrl("https://EU.LIVEKIT.CLOUD"))
}

@Test
fun trustedRegionUrlRejectsCleartextSchemes() {
// The transport carrying the token must stay encrypted, even to a real cloud host.
assertFalse(isTrustedRegionUrl("ws://eu.livekit.cloud"))
assertFalse(isTrustedRegionUrl("http://eu.livekit.cloud"))
}

@Test
fun trustedRegionUrlRejectsForeignHosts() {
assertFalse(isTrustedRegionUrl("https://other.example.com"))
assertFalse(isTrustedRegionUrl("wss://other.example.com/rtc"))
// Near-misses must not match on suffix alone.
assertFalse(isTrustedRegionUrl("https://not-livekit.cloud"))
assertFalse(isTrustedRegionUrl("https://livekit.cloud.example.com"))
}

@Test
fun trustedRegionUrlRejectsMalformedAndHostlessUrls() {
assertFalse(isTrustedRegionUrl(""))
assertFalse(isTrustedRegionUrl("not a url"))
assertFalse(isTrustedRegionUrl("eu.livekit.cloud"))
assertFalse(isTrustedRegionUrl("file:///etc/passwd"))
}

/** A fetched region the SDK must not fail over to is dropped, not connected to with the token. */
@Test
fun rejectsUntrustedFetchedRegions() = runTest {
server = MockWebServer()
server.enqueue(MockResponse().setBody(untrustedRegionResponse))
val regionUrlProvider = RegionUrlProvider(server.url("").toUri(), "token", OkHttpClient.Builder().build(), Json { ignoreUnknownKeys = true })

val settings = regionUrlProvider.fetchRegionSettings()
assertNotNull(settings)
assertEquals(listOf("https://good.livekit.cloud"), settings!!.regions.map { it.url })

assertEquals("https://good.livekit.cloud", regionUrlProvider.getNextBestRegionUrl())
assertNull(regionUrlProvider.getNextBestRegionUrl())
}

/** Server-reported regions arrive over signaling and are the second ingestion path. */
@Test
fun rejectsUntrustedServerReportedRegions() = runTest {
server = MockWebServer()
val regionUrlProvider = RegionUrlProvider(server.url("").toUri(), "token", OkHttpClient.Builder().build(), Json { ignoreUnknownKeys = true })

regionUrlProvider.setServerReportedRegions(
RegionSettings(
listOf(
RegionInfo("foreign", "https://other.example.com", 1L),
RegionInfo("cleartext", "ws://cleartext.livekit.cloud", 2L),
RegionInfo("good", "https://good.livekit.cloud", 3L),
),
),
)

assertEquals("https://good.livekit.cloud", regionUrlProvider.getNextBestRegionUrl())
assertNull(regionUrlProvider.getNextBestRegionUrl())
// No fetch was needed: the server-reported list was used directly.
assertEquals(0, server.requestCount)
}
}

private val untrustedRegionResponse = """{
"regions": [
{
"region": "foreign",
"url": "https://other.example.com",
"distance": "1"
},
{
"region": "cleartext",
"url": "ws://cleartext.livekit.cloud",
"distance": "2"
},
{
"region": "good",
"url": "https://good.livekit.cloud",
"distance": "3"
}
]
}"""

private val regionResponse = """{
"regions": [
{
Expand Down
Loading