diff --git a/crates/collab/tests/integration/channel_tests.rs b/crates/collab/tests/integration/channel_tests.rs index 329478819e..7b20620082 100644 --- a/crates/collab/tests/integration/channel_tests.rs +++ b/crates/collab/tests/integration/channel_tests.rs @@ -589,6 +589,95 @@ async fn test_channel_room( ); } +#[gpui::test] +async fn test_rejoining_channel_after_stale_connection_cleanup_connects_livekit( + executor: BackgroundExecutor, + cx_a: &mut TestAppContext, + cx_a2: &mut TestAppContext, + cx_b: &mut TestAppContext, +) { + let mut server = TestServer::start(executor.clone()).await; + let client_a = server.create_client(cx_a, "user_a").await; + let client_b = server.create_client(cx_b, "user_b").await; + + let channel_id = server + .make_channel("zed", None, (&client_a, cx_a), &mut [(&client_b, cx_b)]) + .await; + + let active_call_a = cx_a.read(ActiveCall::global); + active_call_a + .update(cx_a, |active_call, cx| { + active_call.join_channel(channel_id, cx) + }) + .await + .unwrap(); + + let active_call_b = cx_b.read(ActiveCall::global); + active_call_b + .update(cx_b, |active_call, cx| { + active_call.join_channel(channel_id, cx) + }) + .await + .unwrap(); + + executor.run_until_parked(); + + let old_room_a = + cx_a.read(|cx| active_call_a.read_with(cx, |call, _| call.room().unwrap().clone())); + cx_a.read(|cx| old_room_a.read_with(cx, |room, cx| assert!(room.is_connected(cx)))); + + server.disconnect_client(client_a.peer_id().unwrap()); + executor.run_until_parked(); + server.advance_livekit_timestamp(); + + let client_a2 = server.create_client(cx_a2, "user_a").await; + let active_call_a2 = cx_a2.read(ActiveCall::global); + active_call_a2 + .update(cx_a2, |active_call, cx| { + active_call.join_channel(channel_id, cx) + }) + .await + .unwrap(); + + executor.run_until_parked(); + + let room_a2 = + cx_a2.read(|cx| active_call_a2.read_with(cx, |call, _| call.room().unwrap().clone())); + cx_a2.read(|cx| room_a2.read_with(cx, |room, cx| assert!(room.is_connected(cx)))); + assert_eq!( + room_participants(&room_a2, cx_a2), + RoomParticipants { + remote: vec!["user_b".to_string()], + pending: vec![] + } + ); + + let room_b = + cx_b.read(|cx| active_call_b.read_with(cx, |call, _| call.room().unwrap().clone())); + cx_b.read(|cx| room_b.read_with(cx, |room, cx| assert!(room.is_connected(cx)))); + assert_eq!( + room_participants(&room_b, cx_b), + RoomParticipants { + remote: vec!["user_a".to_string()], + pending: vec![] + } + ); + + cx_a2.read(|cx| { + client_a2.channel_store().read_with(cx, |channels, _| { + let mut participant_ids = channels + .channel_participants(channel_id) + .iter() + .map(|participant| participant.legacy_id) + .collect::>(); + participant_ids.sort_unstable(); + let mut expected_ids = vec![client_a2.user_id().unwrap(), client_b.user_id().unwrap()]; + expected_ids.sort_unstable(); + assert_eq!(participant_ids, expected_ids); + }) + }); +} + #[gpui::test] async fn test_channel_jumping(executor: BackgroundExecutor, cx_a: &mut TestAppContext) { let mut server = TestServer::start(executor.clone()).await; diff --git a/crates/collab/tests/integration/test_server.rs b/crates/collab/tests/integration/test_server.rs index 62169c9a93..3434674e63 100644 --- a/crates/collab/tests/integration/test_server.rs +++ b/crates/collab/tests/integration/test_server.rs @@ -48,7 +48,7 @@ use std::{ use util::path; use workspace::{MultiWorkspace, Workspace, WorkspaceStore}; -use livekit_client::test::TestServer as LivekitTestServer; +use livekit_client::test::{ManualUnixTimestampSource, TestServer as LivekitTestServer}; use crate::db_tests::TestDb; @@ -56,6 +56,7 @@ pub struct TestServer { pub app_state: Arc, pub test_livekit_server: Arc, pub test_db: TestDb, + livekit_timestamp_source: Arc, server: Arc, next_github_user_id: i32, connection_killers: Arc>>>, @@ -96,11 +97,13 @@ impl TestServer { TestDb::sqlite(deterministic.clone()) }; let livekit_server_id = NEXT_LIVEKIT_SERVER_ID.fetch_add(1, SeqCst); - let livekit_server = LivekitTestServer::create( + let livekit_timestamp_source = Arc::new(ManualUnixTimestampSource::new(1_234_567)); + let livekit_server = LivekitTestServer::create_with_timestamp_source( format!("http://livekit.{}.test", livekit_server_id), format!("devkey-{}", livekit_server_id), format!("secret-{}", livekit_server_id), deterministic.clone(), + livekit_timestamp_source.clone(), ) .unwrap(); let executor = Executor::Deterministic(deterministic.clone()); @@ -121,10 +124,15 @@ impl TestServer { forbid_connections: Default::default(), next_github_user_id: 0, test_db, + livekit_timestamp_source, test_livekit_server: livekit_server, } } + pub fn advance_livekit_timestamp(&self) { + self.livekit_timestamp_source.advance(); + } + pub async fn start2( cx_a: &mut TestAppContext, cx_b: &mut TestAppContext, diff --git a/crates/livekit_api/Cargo.toml b/crates/livekit_api/Cargo.toml index 2b2438c25e..80ba67d994 100644 --- a/crates/livekit_api/Cargo.toml +++ b/crates/livekit_api/Cargo.toml @@ -13,6 +13,9 @@ workspace = true path = "src/livekit_api.rs" doctest = false +[features] +test-support = [] + [dependencies] anyhow.workspace = true async-trait.workspace = true diff --git a/crates/livekit_api/src/livekit_api.rs b/crates/livekit_api/src/livekit_api.rs index 745f511b12..250f56b3fd 100644 --- a/crates/livekit_api/src/livekit_api.rs +++ b/crates/livekit_api/src/livekit_api.rs @@ -31,10 +31,25 @@ pub struct LiveKitClient { url: Arc, key: Arc, secret: Arc, + timestamp_source: Arc, } impl LiveKitClient { - pub fn new(mut url: String, key: String, secret: String) -> Self { + pub fn new(url: String, key: String, secret: String) -> Self { + Self::new_with_timestamp_source( + url, + key, + secret, + Arc::new(token::SystemUnixTimestampSource), + ) + } + + pub(crate) fn new_with_timestamp_source( + mut url: String, + key: String, + secret: String, + timestamp_source: Arc, + ) -> Self { if url.ends_with('/') { url.pop(); } @@ -47,6 +62,7 @@ impl LiveKitClient { url: url.into(), key: key.into(), secret: secret.into(), + timestamp_source, } } @@ -61,7 +77,13 @@ impl LiveKitClient { Res: Default + Message, { let client = self.http.clone(); - let token = token::create(&self.key, &self.secret, None, grant); + let token = token::create_with_timestamp_source( + &self.key, + &self.secret, + None, + grant, + self.timestamp_source.as_ref(), + ); let url = format!("{}/{}", self.url, path); log::info!("Request {}: {:?}", url, body); async move { @@ -163,20 +185,22 @@ impl Client for LiveKitClient { } fn room_token(&self, room: &str, identity: &str) -> Result { - token::create( + token::create_with_timestamp_source( &self.key, &self.secret, Some(identity), token::VideoGrant::to_join(room), + self.timestamp_source.as_ref(), ) } fn guest_token(&self, room: &str, identity: &str) -> Result { - token::create( + token::create_with_timestamp_source( &self.key, &self.secret, Some(identity), token::VideoGrant::for_guest(room), + self.timestamp_source.as_ref(), ) } } diff --git a/crates/livekit_api/src/token.rs b/crates/livekit_api/src/token.rs index 6f12d78855..5229a83bb6 100644 --- a/crates/livekit_api/src/token.rs +++ b/crates/livekit_api/src/token.rs @@ -1,14 +1,25 @@ -use anyhow::Result; +use anyhow::{Context as _, Result}; use jsonwebtoken::{DecodingKey, EncodingKey, Header, Validation}; use serde::{Deserialize, Serialize}; use std::{ borrow::Cow, - ops::Add, time::{Duration, SystemTime, UNIX_EPOCH}, }; const DEFAULT_TTL: Duration = Duration::from_secs(6 * 60 * 60); // 6 hours +pub trait UnixTimestampSource: Send + Sync { + fn unix_timestamp(&self) -> Result; +} + +pub struct SystemUnixTimestampSource; + +impl UnixTimestampSource for SystemUnixTimestampSource { + fn unix_timestamp(&self) -> Result { + Ok(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs()) + } +} + #[derive(Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ClaimGrants<'a> { @@ -73,22 +84,55 @@ pub fn create( identity: Option<&str>, video_grant: VideoGrant, ) -> Result { - if video_grant.room_join.is_some() && identity.is_none() { + create_with_timestamp_source( + api_key, + secret_key, + identity, + video_grant, + &SystemUnixTimestampSource, + ) +} + +pub fn create_with_timestamp_source( + api_key: &str, + secret_key: &str, + identity: Option<&str>, + video_grant: VideoGrant, + timestamp_source: &dyn UnixTimestampSource, +) -> Result { + let issued_at = timestamp_source.unix_timestamp()?; + create_with_issued_at(api_key, secret_key, identity, video_grant, issued_at) +} + +fn create_with_issued_at( + api_key: &str, + secret_key: &str, + identity: Option<&str>, + video_grant: VideoGrant, + issued_at: u64, +) -> Result { + let room_join = video_grant.room_join.unwrap_or(false); + if room_join && identity.is_none() { anyhow::bail!("identity is required for room_join grant, but it is none"); } - let now = SystemTime::now(); + let expires_at = issued_at + .checked_add(DEFAULT_TTL.as_secs()) + .context("token expiration overflow")?; + let not_before = if room_join { + // LiveKit Cloud applies participant revocations by comparing the + // revocation timestamp to room-join token `nbf`. + issued_at + } else { + 0 + }; let claims = ClaimGrants { iss: Cow::Borrowed(api_key), sub: identity.map(Cow::Borrowed), - iat: now.duration_since(UNIX_EPOCH).unwrap().as_secs(), - exp: now - .add(DEFAULT_TTL) - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs(), - nbf: 0, + iat: issued_at, + exp: expires_at, + nbf: not_before, jwtid: identity.map(Cow::Borrowed), video: video_grant, }; @@ -108,3 +152,127 @@ pub fn validate<'a>(token: &'a str, secret_key: &str) -> Result> Ok(token.claims) } + +#[cfg(any(test, feature = "test-support"))] +pub fn validate_with_timestamp_source<'a>( + token: &'a str, + secret_key: &str, + timestamp_source: &dyn UnixTimestampSource, +) -> Result> { + let mut validation = Validation::default(); + validation.validate_exp = false; + validation.validate_nbf = false; + let token: jsonwebtoken::TokenData> = jsonwebtoken::decode( + token, + &DecodingKey::from_secret(secret_key.as_ref()), + &validation, + )?; + let claims = token.claims; + let timestamp = timestamp_source.unix_timestamp()?; + + anyhow::ensure!(claims.nbf <= timestamp, "token is not yet valid"); + anyhow::ensure!(claims.exp > timestamp, "token has expired"); + + Ok(claims) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Client as _, LiveKitClient}; + use std::sync::Arc; + + const ISSUED_AT: u64 = 1_234_567; + + struct FixedUnixTimestampSource(u64); + + impl UnixTimestampSource for FixedUnixTimestampSource { + fn unix_timestamp(&self) -> Result { + Ok(self.0) + } + } + + #[test] + fn token_not_before_matches_issue_time() -> Result<()> { + let token = create_with_timestamp_source( + "api-key", + "secret-key", + Some("participant"), + VideoGrant::to_join("room"), + &FixedUnixTimestampSource(ISSUED_AT), + )?; + + assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?; + Ok(()) + } + + #[test] + fn room_token_not_before_matches_issue_time() -> Result<()> { + let client = LiveKitClient::new_with_timestamp_source( + "http://livekit.test".into(), + "api-key".into(), + "secret-key".into(), + Arc::new(FixedUnixTimestampSource(ISSUED_AT)), + ); + let token = client.room_token("room", "participant")?; + + let claims = assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?; + assert_eq!(claims.video.room_join, Some(true)); + assert_eq!(claims.video.can_publish, Some(true)); + assert_eq!(claims.video.can_subscribe, Some(true)); + + Ok(()) + } + + #[test] + fn guest_token_not_before_matches_issue_time() -> Result<()> { + let client = LiveKitClient::new_with_timestamp_source( + "http://livekit.test".into(), + "api-key".into(), + "secret-key".into(), + Arc::new(FixedUnixTimestampSource(ISSUED_AT)), + ); + let token = client.guest_token("room", "participant")?; + + let claims = assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?; + assert_eq!(claims.video.room_join, Some(true)); + assert_eq!(claims.video.can_publish, Some(false)); + assert_eq!(claims.video.can_subscribe, Some(true)); + + Ok(()) + } + + #[test] + fn admin_token_not_before_remains_unset() -> Result<()> { + let token = create_with_timestamp_source( + "api-key", + "secret-key", + None, + VideoGrant::to_admin("room"), + &FixedUnixTimestampSource(ISSUED_AT), + )?; + + let claims = assert_claims_timestamp(&token, ISSUED_AT, 0)?; + assert_eq!(claims.video.room_admin, Some(true)); + + Ok(()) + } + + fn assert_claims_timestamp( + token: &str, + issued_at: u64, + expected_not_before: u64, + ) -> Result> { + let claims = validate_with_timestamp_source( + token, + "secret-key", + &FixedUnixTimestampSource(issued_at), + )?; + + assert_eq!(claims.iat, issued_at); + assert_eq!(claims.nbf, expected_not_before); + assert_eq!(claims.exp, issued_at + DEFAULT_TTL.as_secs()); + + Ok(claims) + } +} diff --git a/crates/livekit_client/Cargo.toml b/crates/livekit_client/Cargo.toml index 42c13f094c..a229d39ca8 100644 --- a/crates/livekit_client/Cargo.toml +++ b/crates/livekit_client/Cargo.toml @@ -17,7 +17,7 @@ doctest = false name = "test_app" [features] -test-support = ["collections/test-support", "gpui/test-support"] +test-support = ["collections/test-support", "gpui/test-support", "livekit_api/test-support"] [dependencies] anyhow.workspace = true @@ -65,6 +65,7 @@ objc.workspace = true collections = { workspace = true, features = ["test-support"] } gpui = { workspace = true, features = ["test-support"] } gpui_platform.workspace = true +livekit_api = { workspace = true, features = ["test-support"] } simplelog.workspace = true [build-dependencies] diff --git a/crates/livekit_client/src/test.rs b/crates/livekit_client/src/test.rs index 955f92dc19..d742dd0661 100644 --- a/crates/livekit_client/src/test.rs +++ b/crates/livekit_client/src/test.rs @@ -57,6 +57,25 @@ pub struct TestServer { pub secret_key: String, rooms: Mutex>, executor: BackgroundExecutor, + timestamp_source: Arc, +} + +pub struct ManualUnixTimestampSource(AtomicU64); + +impl ManualUnixTimestampSource { + pub fn new(timestamp: u64) -> Self { + Self(AtomicU64::new(timestamp)) + } + + pub fn advance(&self) { + self.0.fetch_add(1, SeqCst); + } +} + +impl token::UnixTimestampSource for ManualUnixTimestampSource { + fn unix_timestamp(&self) -> Result { + Ok(self.0.load(SeqCst)) + } } impl TestServer { @@ -65,6 +84,22 @@ impl TestServer { api_key: String, secret_key: String, executor: BackgroundExecutor, + ) -> Result> { + Self::create_with_timestamp_source( + url, + api_key, + secret_key, + executor, + Arc::new(token::SystemUnixTimestampSource), + ) + } + + pub fn create_with_timestamp_source( + url: String, + api_key: String, + secret_key: String, + executor: BackgroundExecutor, + timestamp_source: Arc, ) -> Result> { let mut servers = SERVERS.lock(); if let BTreeEntry::Vacant(e) = servers.entry(url.clone()) { @@ -74,6 +109,7 @@ impl TestServer { secret_key, rooms: Default::default(), executor, + timestamp_source, }); e.insert(server.clone()); Ok(server) @@ -104,6 +140,20 @@ impl TestServer { } } + #[cfg(any(test, feature = "test-support"))] + fn validate_token<'a>(&self, token: &'a str) -> Result> { + token::validate_with_timestamp_source( + token, + &self.secret_key, + self.timestamp_source.as_ref(), + ) + } + + #[cfg(not(any(test, feature = "test-support")))] + fn validate_token<'a>(&self, token: &'a str) -> Result> { + token::validate(token, &self.secret_key) + } + pub async fn create_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; @@ -129,11 +179,19 @@ impl TestServer { async fn join_room(&self, token: String, client_room: Room) -> Result { self.simulate_random_delay().await; - let claims = livekit_api::token::validate(&token, &self.secret_key)?; - let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); - let room_name = claims.video.room.unwrap(); + let claims = self.validate_token(&token)?; + let identity = ParticipantIdentity( + claims + .sub + .context("missing participant identity")? + .to_string(), + ); + let room_name = claims.video.room.context("missing room name")?.to_string(); let mut server_rooms = self.rooms.lock(); - let room = (*server_rooms).entry(room_name.to_string()).or_default(); + let room = (*server_rooms).entry(room_name.clone()).or_default(); + if let Some(revoked_before) = room.token_revocations.get(&identity) { + anyhow::ensure!(claims.nbf >= *revoked_before, "invalid token: revoked"); + } if let Entry::Vacant(e) = room.client_rooms.entry(identity.clone()) { for server_track in &room.video_tracks { @@ -192,7 +250,7 @@ impl TestServer { async fn leave_room(&self, token: String) -> Result<()> { self.simulate_random_delay().await; - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); @@ -209,7 +267,7 @@ impl TestServer { &self, token: String, ) -> Result> { - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let local_identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap().to_string(); @@ -244,14 +302,25 @@ impl TestServer { identity: ParticipantIdentity, ) -> Result<()> { self.simulate_random_delay().await; + let revoked_before = self.timestamp_source.unix_timestamp()?; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .with_context(|| format!("room {room_name} does not exist"))?; - room.client_rooms + let removed_room = room + .client_rooms .remove(&identity) .with_context(|| format!("participant {identity:?} did not join room {room_name:?}"))?; + room.token_revocations.insert(identity, revoked_before); + let mut removed_room = removed_room.0.lock(); + removed_room.connection_state = ConnectionState::Disconnected; + removed_room + .updates_tx + .blocking_send(RoomEvent::Disconnected { + reason: "PARTICIPANT_REMOVED", + }) + .ok(); Ok(()) } @@ -262,13 +331,18 @@ impl TestServer { permission: proto::ParticipantPermission, ) -> Result<()> { self.simulate_random_delay().await; + let revoked_before = self.timestamp_source.unix_timestamp()?; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .with_context(|| format!("room {room_name} does not exist"))?; + let identity = ParticipantIdentity(identity); room.participant_permissions - .insert(ParticipantIdentity(identity), permission); + .insert(identity.clone(), permission); + // Permission changes in LiveKit Cloud invalidate existing participant + // tokens, so the mock needs to reject tokens minted before the update. + room.token_revocations.insert(identity, revoked_before); Ok(()) } @@ -298,7 +372,7 @@ impl TestServer { ) -> Result { self.simulate_random_delay().await; - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); @@ -362,7 +436,7 @@ impl TestServer { ) -> Result { self.simulate_random_delay().await; - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); @@ -421,7 +495,7 @@ impl TestServer { } pub(crate) async fn unpublish_track(&self, token: String, track_sid: &TrackSid) -> Result<()> { - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); @@ -503,7 +577,7 @@ impl TestServer { track_sid: &TrackSid, muted: bool, ) -> Result<()> { - let claims = livekit_api::token::validate(token, &self.secret_key)?; + let claims = self.validate_token(token)?; let room_name = claims.video.room.unwrap(); let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let mut server_rooms = self.rooms.lock(); @@ -557,7 +631,7 @@ impl TestServer { } pub(crate) fn is_track_muted(&self, token: &str, track_sid: &TrackSid) -> Option { - let claims = livekit_api::token::validate(token, &self.secret_key).ok()?; + let claims = self.validate_token(token).ok()?; let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); @@ -572,7 +646,7 @@ impl TestServer { } pub(crate) fn video_tracks(&self, token: String) -> Result> { - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let room_name = claims.video.room.unwrap(); let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); @@ -595,7 +669,7 @@ impl TestServer { } pub(crate) fn audio_tracks(&self, token: String) -> Result> { - let claims = livekit_api::token::validate(&token, &self.secret_key)?; + let claims = self.validate_token(&token)?; let room_name = claims.video.room.unwrap(); let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); @@ -629,6 +703,7 @@ struct TestServerRoom { video_tracks: Vec>, audio_tracks: Vec>, participant_permissions: HashMap, + token_revocations: HashMap, } #[derive(Debug)] @@ -690,21 +765,23 @@ impl livekit_api::Client for TestApiClient { fn room_token(&self, room: &str, identity: &str) -> Result { let server = TestServer::get(&self.url)?; - token::create( + token::create_with_timestamp_source( &server.api_key, &server.secret_key, Some(identity), token::VideoGrant::to_join(room), + server.timestamp_source.as_ref(), ) } fn guest_token(&self, room: &str, identity: &str) -> Result { let server = TestServer::get(&self.url)?; - token::create( + token::create_with_timestamp_source( &server.api_key, &server.secret_key, Some(identity), token::VideoGrant::for_guest(room), + server.timestamp_source.as_ref(), ) } } @@ -853,3 +930,168 @@ impl WeakRoom { self.0.upgrade().map(Room) } } + +#[cfg(test)] +mod tests { + use super::*; + use gpui::TestAppContext; + use livekit_api::Client as _; + use std::{ops::Deref, sync::atomic::AtomicUsize}; + + struct TestServerGuard { + server: Arc, + timestamp_source: Arc, + } + + impl TestServerGuard { + fn advance_timestamp(&self) { + self.timestamp_source.advance(); + } + } + + impl Deref for TestServerGuard { + type Target = TestServer; + + fn deref(&self) -> &Self::Target { + self.server.as_ref() + } + } + + impl Drop for TestServerGuard { + fn drop(&mut self) { + self.server.teardown().ok(); + } + } + + fn create_test_server(name: &str, executor: BackgroundExecutor) -> TestServerGuard { + static NEXT_SERVER_ID: AtomicUsize = AtomicUsize::new(0); + let server_id = NEXT_SERVER_ID.fetch_add(1, SeqCst); + let timestamp_source = Arc::new(ManualUnixTimestampSource::new(1_234_567)); + let server = TestServer::create_with_timestamp_source( + format!("http://livekit-{name}-{server_id}.test"), + format!("api-key-{server_id}"), + format!("secret-key-{server_id}"), + executor, + timestamp_source.clone(), + ) + .expect("create LiveKit test server"); + TestServerGuard { + server, + timestamp_source, + } + } + + async fn assert_token_was_revoked(server: &TestServer, token: String, cx: &mut TestAppContext) { + match Room::connect(server.url.clone(), token, &mut cx.to_async()).await { + Ok(_) => panic!("revoked token unexpectedly connected"), + Err(error) => { + let error = format!("{error:#}"); + assert!( + error.contains("invalid token: revoked"), + "expected revoked token error, got {error}" + ); + } + } + } + + #[gpui::test] + async fn token_created_after_participant_removal_can_join( + executor: BackgroundExecutor, + cx: &mut TestAppContext, + ) { + let server = create_test_server("room-token", executor); + server + .create_room("room".into()) + .await + .expect("create LiveKit test room"); + let api_client = server.create_api_client(); + + let initial_token = api_client + .room_token("room", "participant") + .expect("create initial room token"); + let (initial_room, _) = Room::connect( + server.url.clone(), + initial_token.clone(), + &mut cx.to_async(), + ) + .await + .expect("connect with initial room token"); + + server.advance_timestamp(); + api_client + .remove_participant("room".into(), "participant".into()) + .await + .expect("remove participant"); + + assert_eq!( + initial_room.connection_state(), + ConnectionState::Disconnected + ); + assert_token_was_revoked(&server, initial_token, cx).await; + + let fresh_token = api_client + .room_token("room", "participant") + .expect("create fresh room token"); + let (fresh_room, _) = Room::connect(server.url.clone(), fresh_token, &mut cx.to_async()) + .await + .expect("connect with fresh room token"); + + assert_eq!(fresh_room.connection_state(), ConnectionState::Connected); + } + + #[gpui::test] + async fn guest_token_created_after_permission_update_can_join( + executor: BackgroundExecutor, + cx: &mut TestAppContext, + ) { + let server = create_test_server("guest-token", executor); + server + .create_room("room".into()) + .await + .expect("create LiveKit test room"); + let api_client = server.create_api_client(); + + let initial_token = api_client + .guest_token("room", "participant") + .expect("create initial guest token"); + let (initial_room, _) = Room::connect( + server.url.clone(), + initial_token.clone(), + &mut cx.to_async(), + ) + .await + .expect("connect with initial guest token"); + + server.advance_timestamp(); + api_client + .update_participant( + "room".into(), + "participant".into(), + proto::ParticipantPermission { + can_subscribe: true, + can_publish: true, + can_publish_data: true, + hidden: false, + recorder: false, + }, + ) + .await + .expect("update participant permissions"); + assert_token_was_revoked(&server, initial_token, cx).await; + + server.disconnect_client("participant".into()).await; + assert_eq!( + initial_room.connection_state(), + ConnectionState::Disconnected + ); + + let fresh_token = api_client + .guest_token("room", "participant") + .expect("create fresh guest token"); + let (fresh_room, _) = Room::connect(server.url.clone(), fresh_token, &mut cx.to_async()) + .await + .expect("connect with fresh guest token"); + + assert_eq!(fresh_room.connection_state(), ConnectionState::Connected); + } +}