diff --git a/librice/src/stream.rs b/librice/src/stream.rs index 5cdbc87..3d40572 100644 --- a/librice/src/stream.rs +++ b/librice/src/stream.rs @@ -368,6 +368,22 @@ impl Stream { self.state.proto_stream.remote_credentials() } + /// Restart this stream with fresh random local ICE credentials. + pub fn restart(&self) -> Result { + self.state + .proto_stream + .restart(Instant::from_std(self.state.base_instant))?; + + if let Some(agent) = self.state.weak_agent_inner.upgrade() { + let mut agent = agent.lock().unwrap(); + if let Some(waker) = agent.waker.take() { + waker.wake(); + } + } + + Ok(self.state.proto_stream.local_credentials().unwrap()) + } + /// Add a remote candidate for connection checks for use with this stream /// /// # Examples @@ -1085,4 +1101,23 @@ mod tests { assert_eq!(remote_cands.len(), 1); assert_eq!(remote_cands[0], candidate); } + + #[test] + fn restart_resets_remote_credentials() { + #[cfg(feature = "runtime-tokio")] + let _runtime = crate::tests::tokio_runtime().enter(); + init(); + let agent = Agent::default(); + let stream = agent.add_stream(); + let local = Credentials::new("luser", "lpass"); + let remote = Credentials::new("ruser", "rpass"); + + stream.set_local_credentials(&local); + stream.set_remote_credentials(&remote); + let restarted = stream.restart().unwrap(); + + assert_ne!(restarted, local); + assert_eq!(stream.local_credentials(), Some(restarted)); + assert!(stream.remote_credentials().is_none()); + } } diff --git a/rice-c/src/agent.rs b/rice-c/src/agent.rs index 2308e6e..935365f 100644 --- a/rice-c/src/agent.rs +++ b/rice-c/src/agent.rs @@ -162,6 +162,11 @@ impl Agent { unsafe { crate::ffi::rice_agent_close(self.ffi, now.as_nanos()) } } + /// Restart every stream in this agent with fresh per-stream local ICE credentials. + pub fn restart(&self, now: Instant) { + unsafe { crate::ffi::rice_agent_restart(self.ffi, now.as_nanos()) } + } + /// The controlling state of this ICE agent. This value may change throughout the ICE /// negotiation process. pub fn controlling(&self) -> bool { @@ -779,4 +784,27 @@ mod tests { .build(); let _ = agent; } + + #[test] + fn restart_all_streams() { + let _log = crate::tests::test_init_log(); + let agent = Agent::builder().controlling(true).build(); + let stream1 = agent.add_stream(); + let stream2 = agent.add_stream(); + let local1 = crate::stream::Credentials::new("luser1", "lpass1"); + let local2 = crate::stream::Credentials::new("luser2", "lpass2"); + let remote = crate::stream::Credentials::new("ruser", "rpass"); + + stream1.set_local_credentials(&local1); + stream1.set_remote_credentials(&remote); + stream2.set_local_credentials(&local2); + stream2.set_remote_credentials(&remote); + + agent.restart(Instant::ZERO); + + assert_ne!(stream1.local_credentials().unwrap(), local1); + assert!(stream1.remote_credentials().is_none()); + assert_ne!(stream2.local_credentials().unwrap(), local2); + assert!(stream2.remote_credentials().is_none()); + } } diff --git a/rice-c/src/bindings.rs b/rice-c/src/bindings.rs index 5c00ea3..7f9b90b 100644 --- a/rice-c/src/bindings.rs +++ b/rice-c/src/bindings.rs @@ -3,6 +3,29 @@ pub const RICE_PROTO_MAJOR: u32 = 0; pub const RICE_PROTO_MINOR: u32 = 4; pub const RICE_PROTO_PATCH: u32 = 3; +#[doc = " The UDP transport"] +pub const RICE_TRANSPORT_TYPE_UDP: RiceTransportType = 0; +#[doc = " The TCP transport"] +pub const RICE_TRANSPORT_TYPE_TCP: RiceTransportType = 1; +pub type RiceTransportType = u32; +#[doc = " The candidate is a local network interface"] +pub const RICE_CANDIDATE_TYPE_HOST: RiceCandidateType = 0; +#[doc = " The candidate was discovered from incoming data"] +pub const RICE_CANDIDATE_TYPE_PEER_REFLEXIVE: RiceCandidateType = 1; +#[doc = " The candidate was discovered by asking an external server (STUN/TURN)"] +pub const RICE_CANDIDATE_TYPE_SERVER_REFLEXIVE: RiceCandidateType = 2; +#[doc = " The candidate will relay all data through an external server (TURN)."] +pub const RICE_CANDIDATE_TYPE_RELAYED: RiceCandidateType = 3; +pub type RiceCandidateType = u32; +#[doc = " Not a TCP candidate."] +pub const RICE_TCP_TYPE_NONE: RiceTcpType = 0; +#[doc = " The candidate address will connect to a remote address."] +pub const RICE_TCP_TYPE_ACTIVE: RiceTcpType = 1; +#[doc = " The candidate will listen for incominng TCP connections."] +pub const RICE_TCP_TYPE_PASSIVE: RiceTcpType = 2; +#[doc = " Simultaneous open. The candidate will both listen for incoming connections, and connect to\n remote addresses."] +pub const RICE_TCP_TYPE_SO: RiceTcpType = 3; +pub type RiceTcpType = u32; #[doc = " Component is in initial state and no connectivity checks are in progress."] pub const RICE_COMPONENT_CONNECTION_STATE_NEW: RiceComponentConnectionState = 0; #[doc = " Connectivity checks are in progress for this candidate"] @@ -17,15 +40,25 @@ pub const RICE_ADDRESS_FAMILY_IPV4: RiceAddressFamily = 1; #[doc = " IP version 6."] pub const RICE_ADDRESS_FAMILY_IPV6: RiceAddressFamily = 2; pub type RiceAddressFamily = u32; -#[doc = " The candidate is a local network interface"] -pub const RICE_CANDIDATE_TYPE_HOST: RiceCandidateType = 0; -#[doc = " The candidate was discovered from incoming data"] -pub const RICE_CANDIDATE_TYPE_PEER_REFLEXIVE: RiceCandidateType = 1; -#[doc = " The candidate was discovered by asking an external server (STUN/TURN)"] -pub const RICE_CANDIDATE_TYPE_SERVER_REFLEXIVE: RiceCandidateType = 2; -#[doc = " The candidate will relay all data through an external server (TURN)."] -pub const RICE_CANDIDATE_TYPE_RELAYED: RiceCandidateType = 3; -pub type RiceCandidateType = u32; +#[doc = " The SHA-1 HMAC."] +pub const RICE_INTEGRITY_ALGORITHM_SHA1: RiceIntegrityAlgorithm = 0; +#[doc = " The SHA-256 HMAC."] +pub const RICE_INTEGRITY_ALGORITHM_SHA256: RiceIntegrityAlgorithm = 1; +pub type RiceIntegrityAlgorithm = u32; +#[doc = " The configuration will automatically be used when supported."] +pub const RICE_FEATURE_DISABLED: RiceFeature = -1; +#[doc = " The configuration will automatically be used when supported."] +pub const RICE_FEATURE_AUTO: RiceFeature = 0; +#[doc = " The configuration is enabled and required."] +pub const RICE_FEATURE_REQUIRED: RiceFeature = 1; +pub type RiceFeature = i32; +#[doc = " Openssl."] +pub const RICE_TLS_VARIANT_OPENSSL: RiceTlsVariant = 1; +#[doc = " Rustls."] +pub const RICE_TLS_VARIANT_RUSTLS: RiceTlsVariant = 2; +#[doc = " Dimpl."] +pub const RICE_TLS_VARIANT_DIMPL: RiceTlsVariant = 3; +pub type RiceTlsVariant = u32; #[doc = " Not an error. The operation was completed successfully."] pub const RICE_ERROR_SUCCESS: RiceError = 0; #[doc = " The operation failed for an unspecified reason."] @@ -35,18 +68,6 @@ pub const RICE_ERROR_RESOURCE_NOT_FOUND: RiceError = -2; #[doc = " The operation is already in progress."] pub const RICE_ERROR_ALREADY_IN_PROGRESS: RiceError = -3; pub type RiceError = i32; -#[doc = " The configuration will automatically be used when supported."] -pub const RICE_FEATURE_DISABLED: RiceFeature = -1; -#[doc = " The configuration will automatically be used when supported."] -pub const RICE_FEATURE_AUTO: RiceFeature = 0; -#[doc = " The configuration is enabled and required."] -pub const RICE_FEATURE_REQUIRED: RiceFeature = 1; -pub type RiceFeature = i32; -#[doc = " The SHA-1 HMAC."] -pub const RICE_INTEGRITY_ALGORITHM_SHA1: RiceIntegrityAlgorithm = 0; -#[doc = " The SHA-256 HMAC."] -pub const RICE_INTEGRITY_ALGORITHM_SHA256: RiceIntegrityAlgorithm = 1; -pub type RiceIntegrityAlgorithm = u32; #[doc = " No error."] pub const RICE_PARSE_CANDIDATE_ERROR_SUCCESS: RiceParseCandidateError = 0; #[doc = " Not a candidate message."] @@ -68,27 +89,6 @@ pub const RICE_PARSE_CANDIDATE_ERROR_BAD_EXTENSION: RiceParseCandidateError = -8 #[doc = " Data is not well formed."] pub const RICE_PARSE_CANDIDATE_ERROR_MALFORMED: RiceParseCandidateError = -9; pub type RiceParseCandidateError = i32; -#[doc = " Not a TCP candidate."] -pub const RICE_TCP_TYPE_NONE: RiceTcpType = 0; -#[doc = " The candidate address will connect to a remote address."] -pub const RICE_TCP_TYPE_ACTIVE: RiceTcpType = 1; -#[doc = " The candidate will listen for incominng TCP connections."] -pub const RICE_TCP_TYPE_PASSIVE: RiceTcpType = 2; -#[doc = " Simultaneous open. The candidate will both listen for incoming connections, and connect to\n remote addresses."] -pub const RICE_TCP_TYPE_SO: RiceTcpType = 3; -pub type RiceTcpType = u32; -#[doc = " Openssl."] -pub const RICE_TLS_VARIANT_OPENSSL: RiceTlsVariant = 1; -#[doc = " Rustls."] -pub const RICE_TLS_VARIANT_RUSTLS: RiceTlsVariant = 2; -#[doc = " Dimpl."] -pub const RICE_TLS_VARIANT_DIMPL: RiceTlsVariant = 3; -pub type RiceTlsVariant = u32; -#[doc = " The UDP transport"] -pub const RICE_TRANSPORT_TYPE_UDP: RiceTransportType = 0; -#[doc = " The TCP transport"] -pub const RICE_TRANSPORT_TYPE_TCP: RiceTransportType = 1; -pub type RiceTransportType = u32; #[repr(C)] #[derive(Debug, Copy, Clone)] pub struct Credentials { @@ -620,6 +620,10 @@ unsafe extern "C" { #[doc = " Close the `RiceAgent`.\n\n Closure does involve closing network resources (signalled through calls to\n `rice_agent_poll()`) and will only succesfully complete once `rice_agent_poll`() returns\n `Closed`."] pub fn rice_agent_close(agent: *const RiceAgent, now_nanos: i64); } +unsafe extern "C" { + #[doc = " Restart every stream in the `RiceAgent` with fresh local ICE credentials."] + pub fn rice_agent_restart(agent: *const RiceAgent, now_nanos: i64); +} unsafe extern "C" { #[doc = " Return the process-local unique id for this agent."] pub fn rice_agent_id(agent: *const RiceAgent) -> u64; @@ -961,6 +965,18 @@ unsafe extern "C" { credentials: *const RiceCredentials, ); } +unsafe extern "C" { + #[doc = " Restart this `RiceStream` with explicit local ICE credentials."] + pub fn rice_stream_restart_with_credentials( + stream: *mut RiceStream, + credentials: *const RiceCredentials, + now_nanos: i64, + ) -> RiceError; +} +unsafe extern "C" { + #[doc = " Restart this `RiceStream` with fresh random local ICE credentials."] + pub fn rice_stream_restart(stream: *mut RiceStream, now_nanos: i64) -> RiceError; +} unsafe extern "C" { #[doc = " Construct a `RiceCandidate` from a string as formatted in an SDP and specified in RFC5245\n Section 15.1.\n\n Takes the form 'a=candidate:foundation 1 UDP 12345 127.0.0.1 23456 typ host'."] pub fn rice_candidate_new_from_sdp_string( @@ -1156,6 +1172,18 @@ unsafe extern "C" { component_id: usize, ) -> *mut RiceComponent; } +unsafe extern "C" { + #[doc = " Start gathering candidates for every component in a stream with the provided local socket\n addresses.\n\n Components are started in stream order. If a later component fails after an earlier component\n has already begun gathering, the earlier component continues gathering."] + pub fn rice_stream_gather_candidates( + stream: *mut RiceStream, + sockets_len: usize, + sockets_addr: *const *const RiceAddress, + sockets_transports: *const RiceTransportType, + turn_len: usize, + turn_sockets: *const *const RiceAddress, + turn_config: *const *mut RiceTurnConfig, + ) -> RiceError; +} unsafe extern "C" { #[doc = " Start gathering candidates for a component with the provided local socket addresses.\n\n - `component`: The component to start gathering.\n - `sockets_len`: The number of entries in both `sockets_addr` and `sockets_transports`.\n - `sockets_addr`: An array of addresses for producing host and STUN server-reflexive\n candidates.\n - `sockets_transports`: An array of transport types for producing host and STUN\n server-reflexive candidates.\n - `turn_len`: the number of entries in both `turn_sockets` and `turn_config`.\n - `turn_sockets`: An array of local addresses for producing TURN candidates.\n - `turn_config`: An array of TURN server configurations.\n\n Candidates will be generated as follows (if they succeed):\n\n 1. A host candidate for each `(sockets_transports[i], socket_addr[i])`. If TCP, then both an\n active and passive host candidate will be generated.\n 2. For each configured STUN server, a reflexive candidate for each\n `(sockets_transports[i], socket_addr[i])` if different from any other candidate\n produced. The local address for each STUN server connection will be one of the entries\n provided in `sockets_addr`.\n 3. For each `(turn_sockets[i], turn_config[i])` a TURN allocation will be attempted and a\n relayed candidate produced on success. If you would like multiple options for relayed\n candidates, e.g. UDP, TCP, TCP/TLS, then provide each options as different entries in the\n provided array. The `turn_sockets[i]` value is the local address to communicate with the\n TURN server in `turn_config[i]` and should be different than any value provided through\n `sockets_addr`."] pub fn rice_component_gather_candidates( diff --git a/rice-c/src/stream.rs b/rice-c/src/stream.rs index 134665b..8ec2319 100644 --- a/rice-c/src/stream.rs +++ b/rice-c/src/stream.rs @@ -134,6 +134,31 @@ impl Stream { } } + /// Restart this stream with explicit local ICE credentials. + pub fn restart_with_credentials( + &self, + credentials: &Credentials, + now: Instant, + ) -> Result<(), crate::agent::AgentError> { + unsafe { + crate::agent::AgentError::from_c(crate::ffi::rice_stream_restart_with_credentials( + self.ffi, + credentials.into_c_none(), + now.as_nanos(), + )) + } + } + + /// Restart this stream with fresh random local ICE credentials. + pub fn restart(&self, now: Instant) -> Result<(), crate::agent::AgentError> { + unsafe { + crate::agent::AgentError::from_c(crate::ffi::rice_stream_restart( + self.ffi, + now.as_nanos(), + )) + } + } + /// Signal the end of local candidates. Calling this function may allow ICE processing to /// complete. pub fn end_of_local_candidates(&self) { @@ -228,6 +253,39 @@ impl Stream { } } + /// Start gathering candidates for every component in this stream. + pub fn gather_candidates<'a, 'b>( + &self, + sockets: impl IntoIterator, + turn_servers: impl IntoIterator, + ) -> Result<(), crate::agent::AgentError> { + unsafe { + let mut transports = vec![]; + let mut socket_addr = vec![]; + let mut socket_addresses = vec![]; + for (ttype, addr) in sockets.into_iter() { + transports.push(ttype.into()); + socket_addresses.push(crate::const_override(addr.ffi)); + socket_addr.push(addr); + } + let mut turn_sockets = vec![]; + let mut turn_configs = vec![]; + for (turn_addr, config) in turn_servers.into_iter() { + turn_sockets.push(crate::const_override(turn_addr.ffi)); + turn_configs.push(config.into_c_full()); + } + crate::agent::AgentError::from_c(crate::ffi::rice_stream_gather_candidates( + self.ffi, + transports.len(), + socket_addresses.as_ptr(), + transports.as_ptr(), + turn_sockets.len(), + turn_sockets.as_ptr(), + turn_configs.as_ptr(), + )) + } + } + /// Provide the stream with data that has been received on an external socket. The returned /// value indicates what has been done with the data and any application data that has been /// received. @@ -430,10 +488,13 @@ impl GatheredCandidate { #[cfg(test)] mod tests { + use std::collections::BTreeSet; + use sans_io_time::Instant; use super::*; use crate::agent::{Agent, AgentPoll}; + use crate::candidate::CandidateApi; #[test] fn getters() { @@ -503,4 +564,56 @@ mod tests { let _ = agent.poll(Instant::ZERO); let _ = agent.poll(Instant::ZERO); } + + #[test] + fn restart_clears_remote_credentials() { + let _log = crate::tests::test_init_log(); + let agent = Agent::builder().build(); + let stream = agent.add_stream(); + let local = Credentials::new("luser", "lpass"); + let remote = Credentials::new("ruser", "rpass"); + let restart = Credentials::new("nextuser", "nextpass"); + + stream.set_local_credentials(&local); + stream.set_remote_credentials(&remote); + stream + .restart_with_credentials(&restart, Instant::ZERO) + .unwrap(); + + assert_eq!(stream.local_credentials(), Some(restart)); + assert!(stream.remote_credentials().is_none()); + } + + #[test] + fn stream_gather_candidates_all_components() { + let _log = crate::tests::test_init_log(); + let addr: crate::Address = "192.168.0.1:1000".parse().unwrap(); + let agent = Agent::builder().build(); + let stream = agent.add_stream(); + stream.set_local_credentials(&Credentials::new("luser", "lpass")); + stream.set_remote_credentials(&Credentials::new("ruser", "rpass")); + let _component1 = stream.add_component(); + let _component2 = stream.add_component(); + + stream + .gather_candidates([(TransportType::Udp, &addr)], []) + .unwrap(); + + let mut gathered_components = BTreeSet::new(); + for _ in 0..8 { + let mut poll = agent.poll(Instant::ZERO); + match &mut poll { + AgentPoll::GatheredCandidate(candidate) => { + gathered_components.insert(candidate.gathered.candidate().component_id()); + stream.add_local_gathered_candidate(candidate.gathered.take()); + } + AgentPoll::GatheringComplete(_) | AgentPoll::WaitUntilNanos(_) => (), + other => panic!("unexpected poll result: {other:?}"), + } + if gathered_components.len() == 2 { + break; + } + } + assert_eq!(gathered_components, BTreeSet::from([1, 2])); + } } diff --git a/rice-proto/src/agent.rs b/rice-proto/src/agent.rs index af924e2..5d67859 100644 --- a/rice-proto/src/agent.rs +++ b/rice-proto/src/agent.rs @@ -11,6 +11,7 @@ //! ICE Agent implementation as specified in RFC 8445 use alloc::boxed::Box; +use alloc::string::String; use alloc::vec::Vec; use core::error::Error; @@ -30,7 +31,7 @@ use crate::conncheck::{ use crate::consent::{self, ConsentFreshness, ConsentFreshnessPoll}; use crate::gathering::{GatherPoll, GatheredCandidate}; use crate::rand::rand_u64; -use crate::stream::{Stream, StreamMut, StreamState}; +use crate::stream::{Credentials, Stream, StreamMut, StreamState}; use crate::turn::TurnConfig; use stun_proto::agent::{StunError, Transmit}; use stun_proto::types::message::StunParseError; @@ -106,7 +107,8 @@ pub struct Agent { pub(crate) checklistset: ConnCheckListSet, pub(crate) stun_servers: Vec<(TransportType, SocketAddr)>, pub(crate) turn_servers: Vec, - streams: Vec, + streams: Vec>, + removed_streams: Vec<(usize, usize)>, pub(crate) rto: Option, pub(crate) consent_freshness: Option, pub(crate) consent_freshness_cfg: consent::Config, @@ -251,6 +253,7 @@ impl AgentBuilder { stun_servers: Vec::new(), turn_servers: Vec::new(), streams: Vec::new(), + removed_streams: Vec::new(), rto: self.rto, consent_freshness, consent_freshness_cfg: self.consent_freshness_config, @@ -260,6 +263,28 @@ impl AgentBuilder { static AGENT_COUNT: AtomicU64 = AtomicU64::new(0); +const ICE_RESTART_ALPHABET: &[u8] = + b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +pub(crate) fn generate_restart_credentials() -> Credentials { + fn random_ice_string(len: usize) -> String { + let mut ret = String::with_capacity(len); + while ret.len() < len { + let mut value = rand_u64(); + for _ in 0..11 { + ret.push(ICE_RESTART_ALPHABET[(value % ICE_RESTART_ALPHABET.len() as u64) as usize] as char); + if ret.len() == len { + break; + } + value /= ICE_RESTART_ALPHABET.len() as u64; + } + } + ret + } + + Credentials::new(random_ice_string(4), random_ice_string(22)) +} + impl Default for Agent { fn default() -> Self { Agent::builder().build() @@ -327,7 +352,7 @@ impl Agent { final_retransmit_timeout: Duration, ) { let rto = RequestRto::from_parts(initial, max, retransmits, final_retransmit_timeout); - for stream in self.streams.iter_mut() { + for stream in self.streams.iter_mut().flatten() { stream.set_request_retransmits(rto.clone()); } self.checklistset.set_request_retransmits(rto); @@ -368,12 +393,56 @@ impl Agent { )] pub fn add_stream(&mut self) -> usize { let checklist_id = self.checklistset.new_list(); - let id = self.streams.len(); + let id = self + .streams + .iter() + .enumerate() + .find_map(|(idx, stream)| stream.is_none().then_some(idx)) + .unwrap_or(self.streams.len()); let stream = crate::stream::StreamState::new(id, checklist_id); - self.streams.push(stream); + if let Some(slot) = self.streams.get_mut(id) { + *slot = Some(stream); + } else { + self.streams.push(Some(stream)); + } id } + /// Remove a [`Stream`] from this agent without changing any other stream ids. + pub fn remove_stream(&mut self, id: usize) { + let Some(stream) = self.streams.get_mut(id).and_then(Option::take) else { + return; + }; + let component_ids = stream + .component_ids_iter() + .collect::>(); + for component_id in component_ids { + if let Some(cf) = &mut self.consent_freshness { + cf.stop(id, component_id); + } + } + self.removed_streams.push((stream.checklist_id, id)); + self.checklistset.remove_list(stream.checklist_id); + } + + /// Restart every stream in this agent with fresh per-stream local ICE credentials. + /// + /// Existing local candidates, sockets, and TURN allocations are preserved; callers must + /// exchange the new credentials and remote candidates before connectivity checks can complete + /// again. + pub fn restart(&mut self, now: Instant) { + let stream_ids = self + .streams + .iter() + .enumerate() + .filter_map(|(id, stream)| stream.as_ref().map(|_| id)) + .collect::>(); + for stream_id in stream_ids { + let credentials = generate_restart_credentials(); + let _ = self.restart_stream_with_credentials(stream_id, credentials, now); + } + } + /// Close the agent loop. Applications should wait for [`Agent::poll`] to return /// [`AgentPoll::Closed`] after calling this function. #[tracing::instrument( @@ -386,6 +455,7 @@ impl Agent { pub fn close(&mut self, now: Instant) { info!("closing agent"); self.checklistset.close(now); + self.removed_streams.clear(); if let Some(cf) = &mut self.consent_freshness { cf.close(); @@ -437,7 +507,7 @@ impl Agent { /// /// If the stream does not exist, then `None` will be returned. pub fn stream(&self, id: usize) -> Option> { - if self.streams.get(id).is_some() { + if self.streams.get(id).and_then(Option::as_ref).is_some() { Some(Stream::from_agent(self, id)) } else { None @@ -445,12 +515,12 @@ impl Agent { } pub(crate) fn stream_state(&self, id: usize) -> Option<&crate::stream::StreamState> { - self.streams.get(id) + self.streams.get(id).and_then(Option::as_ref) } /// Get a [`StreamMut`] by id. If the stream does not exist, then `None` will be returned. pub fn mut_stream(&mut self, id: usize) -> Option> { - if self.streams.get_mut(id).is_some() { + if self.streams.get_mut(id).and_then(Option::as_mut).is_some() { Some(StreamMut::from_agent(self, id)) } else { None @@ -461,7 +531,63 @@ impl Agent { &mut self, id: usize, ) -> Option<&mut crate::stream::StreamState> { - self.streams.get_mut(id) + self.streams.get_mut(id).and_then(Option::as_mut) + } + + pub(crate) fn restart_stream_with_credentials( + &mut self, + id: usize, + local_credentials: Credentials, + now: Instant, + ) -> Option<()> { + let (checklist_id, component_ids) = { + let stream = self.mut_stream_state(id)?; + stream.clear_selected_pairs(); + stream.set_local_credentials(local_credentials.clone()); + stream.clear_remote_credentials(); + (stream.checklist_id, stream.component_ids_iter().collect::>()) + }; + let checklist = self.checklistset.mut_list(checklist_id)?; + checklist.set_local_credentials(local_credentials); + checklist.clear_remote_credentials(); + self.checklistset.restart_list(checklist_id, now); + if let Some(cf) = &mut self.consent_freshness { + for component_id in component_ids { + cf.stop(id, component_id); + } + } + Some(()) + } + + pub(crate) fn remove_component(&mut self, stream_id: usize, component_id: usize) { + let checklist_id = { + let Some(stream) = self.mut_stream_state(stream_id) else { + return; + }; + if !stream.remove_component(component_id) { + return; + } + stream.checklist_id + }; + if let Some(cf) = &mut self.consent_freshness { + cf.stop(stream_id, component_id); + } + self.checklistset.remove_component(checklist_id, component_id); + } + + fn stream_id_for_checklist(&self, checklist_id: usize) -> Option { + self.streams + .iter() + .flatten() + .find(|stream| stream.checklist_id == checklist_id) + .map(StreamState::id) + .or_else(|| { + self.removed_streams + .iter() + .find_map(|(retired_checklist_id, stream_id)| { + (*retired_checklist_id == checklist_id).then_some(*stream_id) + }) + }) } /// Poll the [`Agent`] for further progress to be made. The returned value indicates what the @@ -477,7 +603,7 @@ impl Agent { pub fn poll(&mut self, now: Instant) -> AgentPoll { let mut lowest_wait = None; - for stream in self.streams.iter_mut() { + for stream in self.streams.iter_mut().flatten() { let stream_id = stream.id(); match stream.poll_gather(now) { GatherPoll::AllocateSocket { @@ -540,11 +666,9 @@ impl Agent { local_addr: from, remote_addr: to, } => { - if let Some(stream) = - self.streams.iter().find(|s| s.checklist_id == checklist_id) - { + if let Some(stream_id) = self.stream_id_for_checklist(checklist_id) { return AgentPoll::AllocateSocket(AgentSocket { - stream_id: stream.id(), + stream_id, component_id: cid, transport, from, @@ -561,11 +685,9 @@ impl Agent { local_addr: from, remote_addr: to, } => { - if let Some(stream) = - self.streams.iter().find(|s| s.checklist_id == checklist_id) - { + if let Some(stream_id) = self.stream_id_for_checklist(checklist_id) { return AgentPoll::RemoveSocket(AgentSocket { - stream_id: stream.id(), + stream_id, component_id: cid, transport, from, @@ -582,6 +704,7 @@ impl Agent { if let Some(stream) = self .streams .iter_mut() + .flatten() .find(|s| s.checklist_id == checklist_id) { if let Some(component) = stream.mut_component_state(cid) { @@ -601,8 +724,11 @@ impl Agent { checklist_id, event: ConnCheckEvent::SelectedPair(cid, selected), } => { - if let Some(stream) = - self.streams.iter().find(|s| s.checklist_id == checklist_id) + if let Some(stream) = self + .streams + .iter() + .flatten() + .find(|s| s.checklist_id == checklist_id) { if stream.component_state(cid).is_some() { if !self.checklistset.ice_lite() { @@ -645,8 +771,11 @@ impl Agent { checklist_id, event: ConnCheckEvent::ConsentResponseReceived(cid, revoked), } => { - if let Some(stream) = - self.streams.iter().find(|s| s.checklist_id == checklist_id) + if let Some(stream) = self + .streams + .iter() + .flatten() + .find(|s| s.checklist_id == checklist_id) { if let Some(cf) = &mut self.consent_freshness { cf.on_response(cid, now); @@ -675,6 +804,7 @@ impl Agent { let checklist_id = self .streams .iter() + .flatten() .find(|s| s.id() == stream_id) .map(|s| s.checklist_id); @@ -707,6 +837,7 @@ impl Agent { if let Some(component) = self .streams .iter_mut() + .flatten() .find(|s| s.id() == stream_id) .and_then(|stream| stream.mut_component_state(component_id)) { @@ -743,7 +874,7 @@ impl Agent { /// If not-None, then the provided data must be sent to the peer from the provided socket /// address. pub fn poll_transmit(&mut self, now: Instant) -> Option { - for stream in self.streams.iter_mut() { + for stream in self.streams.iter_mut().flatten() { let stream_id = stream.id(); if let Some((_component_id, transmit)) = stream.poll_gather_transmit(now) { return Some(AgentTransmit::from_data(stream_id, transmit)); @@ -753,6 +884,7 @@ impl Agent { if let Some(stream) = self .streams .iter() + .flatten() .find(|s| s.checklist_id == transmit.checklist_id) { Some(AgentTransmit { @@ -951,7 +1083,7 @@ mod tests { .build(); let stream_id = agent.add_stream(); - let _ = agent.streams[stream_id].add_component(); + let _ = agent.streams[stream_id].as_mut().unwrap().add_component(); let component_id = 1; let addr: SocketAddr = "10.0.0.1:1000".parse().unwrap(); @@ -1003,6 +1135,8 @@ mod tests { } let component = agent.streams[stream_id] + .as_ref() + .unwrap() .component_state(component_id) .unwrap(); assert_eq!(component.state(), ComponentConnectionState::Failed); @@ -1026,7 +1160,7 @@ mod tests { .build(); let stream_id = agent.add_stream(); - let _ = agent.streams[stream_id].add_component(); + let _ = agent.streams[stream_id].as_mut().unwrap().add_component(); let component_id = 1; let addr: SocketAddr = "10.0.0.1:1000".parse().unwrap(); @@ -1078,6 +1212,8 @@ mod tests { } let component = agent.streams[stream_id] + .as_ref() + .unwrap() .component_state(component_id) .unwrap(); assert_eq!(component.state(), ComponentConnectionState::Failed); @@ -1100,7 +1236,7 @@ mod tests { .build(); let stream_id = agent.add_stream(); - let _ = agent.streams[stream_id].add_component(); + let _ = agent.streams[stream_id].as_mut().unwrap().add_component(); let component_id = 1; let addr: SocketAddr = "10.0.0.1:1000".parse().unwrap(); @@ -1136,7 +1272,7 @@ mod tests { now, ); - let checklist_id = agent.streams[stream_id].checklist_id; + let checklist_id = agent.streams[stream_id].as_ref().unwrap().checklist_id; agent .mut_stream(stream_id) .unwrap() @@ -1207,6 +1343,61 @@ mod tests { assert_eq!(read_cfg.timeout, Duration::from_secs(60)); } + #[test] + fn remove_stream_keeps_other_ids_stable() { + use crate::candidate::{Candidate, CandidateType, TransportType}; + + let _log = crate::tests::test_init_log(); + let now = Instant::ZERO; + let mut agent = Agent::default(); + let stream1 = agent.add_stream(); + let stream2 = agent.add_stream(); + let component1 = agent.mut_stream(stream1).unwrap().add_component().unwrap(); + let component2 = agent.mut_stream(stream2).unwrap().add_component().unwrap(); + agent + .mut_stream(stream1) + .unwrap() + .add_local_candidate( + Candidate::builder( + component1, + CandidateType::Host, + TransportType::Udp, + "a", + "10.0.0.1:1000".parse().unwrap(), + ) + .priority(100) + .build(), + ); + agent + .mut_stream(stream2) + .unwrap() + .add_local_candidate( + Candidate::builder( + component2, + CandidateType::Host, + TransportType::Udp, + "b", + "10.0.0.1:1002".parse().unwrap(), + ) + .priority(100) + .build(), + ); + + agent.remove_stream(stream1); + assert!(agent.stream(stream1).is_none()); + assert!(agent.stream(stream2).is_some()); + + let AgentPoll::RemoveSocket(removed) = agent.poll(now) else { + panic!("expected RemoveSocket for removed stream"); + }; + assert_eq!(removed.stream_id, stream1); + assert_eq!(removed.component_id, component1); + + let replacement = agent.add_stream(); + assert_eq!(replacement, stream1); + assert!(agent.stream(stream2).is_some()); + } + #[test] fn consent_freshness_close() { let _log = crate::tests::test_init_log(); diff --git a/rice-proto/src/capi.rs b/rice-proto/src/capi.rs index 74156d6..0e10d8c 100644 --- a/rice-proto/src/capi.rs +++ b/rice-proto/src/capi.rs @@ -68,15 +68,15 @@ use crate::turn::OpensslTurnConfig; #[cfg(feature = "rustls")] use crate::turn::RustlsTurnConfig; use crate::turn::{TurnClient, TurnConfig, TurnCredentials, TurnTlsConfig}; -use stun_proto::Instant; use stun_proto::agent::{StunError, Transmit}; use stun_proto::auth::Feature; -use stun_proto::types::AddressFamily; use stun_proto::types::data::{Data, DataOwned, DataSlice}; use stun_proto::types::message::IntegrityAlgorithm; +use stun_proto::types::AddressFamily; +use stun_proto::Instant; -use tracing_subscriber::Layer; use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::Layer; pub use rice_ctypes::{RiceAddress, RiceError, RiceTransportType}; @@ -245,6 +245,19 @@ pub unsafe extern "C" fn rice_agent_close(agent: *const RiceAgent, now_nanos: i6 } } +/// Restart every stream in the `RiceAgent` with fresh local ICE credentials. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn rice_agent_restart(agent: *const RiceAgent, now_nanos: i64) { + unsafe { + let agent = Arc::from_raw(agent); + let mut proto_agent = agent.proto_agent.lock().unwrap(); + proto_agent.restart(Instant::from_nanos(now_nanos)); + + drop(proto_agent); + core::mem::forget(agent); + } +} + /// Return the process-local unique id for this agent. #[unsafe(no_mangle)] pub unsafe extern "C" fn rice_agent_id(agent: *const RiceAgent) -> u64 { @@ -1881,6 +1894,48 @@ pub unsafe extern "C" fn rice_stream_set_remote_credentials( } } +/// Restart this `RiceStream` with explicit local ICE credentials. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn rice_stream_restart_with_credentials( + stream: *mut RiceStream, + credentials: *const RiceCredentials, + now_nanos: i64, +) -> RiceError { + unsafe { + let creds = Box::from_raw(mut_override(credentials)); + let stream = Arc::from_raw(stream); + let mut proto_agent = stream.proto_agent.lock().unwrap(); + let ret = if let Some(mut proto_stream) = proto_agent.mut_stream(stream.stream_id) { + proto_stream.restart(creds.credentials.clone(), Instant::from_nanos(now_nanos)); + RiceError::Success + } else { + RiceError::ResourceNotFound + }; + drop(proto_agent); + core::mem::forget(stream); + core::mem::forget(creds); + ret + } +} + +/// Restart this `RiceStream` with fresh random local ICE credentials. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn rice_stream_restart(stream: *mut RiceStream, now_nanos: i64) -> RiceError { + unsafe { + let stream = Arc::from_raw(stream); + let mut proto_agent = stream.proto_agent.lock().unwrap(); + let ret = if let Some(mut proto_stream) = proto_agent.mut_stream(stream.stream_id) { + let _ = proto_stream.restart_with_random_credentials(Instant::from_nanos(now_nanos)); + RiceError::Success + } else { + RiceError::ResourceNotFound + }; + drop(proto_agent); + core::mem::forget(stream); + ret + } +} + /// The type of the TCP candidate. #[derive(Debug, Copy, Clone, PartialEq, Eq)] #[repr(u32)] @@ -2921,6 +2976,98 @@ pub unsafe extern "C" fn rice_stream_get_component( } } +/// Start gathering candidates for every component in a stream with the provided local socket +/// addresses. +/// +/// Components are started in stream order. If a later component fails after an earlier component +/// has already begun gathering, the earlier component continues gathering. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn rice_stream_gather_candidates( + stream: *mut RiceStream, + sockets_len: usize, + sockets_addr: *const *const RiceAddress, + sockets_transports: *const RiceTransportType, + turn_len: usize, + turn_sockets: *const *const RiceAddress, + turn_config: *const *mut RiceTurnConfig, +) -> RiceError { + unsafe { + let stream = Arc::from_raw(stream); + let stun_servers = { + let Some(agent) = stream.weak_agent.upgrade() else { + core::mem::forget(stream); + return RiceError::ResourceNotFound; + }; + let agent = agent.inner.lock().unwrap(); + agent.stun_servers.clone() + }; + let mut proto_agent = stream.proto_agent.lock().unwrap(); + let Some(mut proto_stream) = proto_agent.mut_stream(stream.stream_id) else { + drop(proto_agent); + core::mem::forget(stream); + return RiceError::ResourceNotFound; + }; + + let sockets_addr = core::slice::from_raw_parts(sockets_addr, sockets_len); + let sockets_transport = core::slice::from_raw_parts(sockets_transports, sockets_len); + + let sockets = sockets_transport + .iter() + .zip(sockets_addr.iter()) + .map(|(&transport, addr)| { + let socket_addr = RiceAddress::into_rice_none(*addr).inner(); + (transport_type_from_c(transport), socket_addr) + }) + .collect::>(); + + let turn_sockets = if turn_len > 0 { + core::slice::from_raw_parts(turn_sockets, turn_len) + } else { + &[] + }; + let turn_configs = if turn_len > 0 { + core::slice::from_raw_parts(turn_config, turn_len) + .iter() + .map(|config| RiceTurnConfig::into_rice_full(*config)) + .collect::>() + } else { + vec![] + }; + let turn_servers = turn_sockets + .iter() + .zip(turn_configs.iter()) + .map(|(socket, config)| { + let turn_addr = RiceAddress::into_rice_none(*socket); + (turn_addr.inner(), &config.0) + }) + .collect::>(); + + let component_ids = proto_stream.component_ids_iter().collect::>(); + for component_id in component_ids { + let Some(mut proto_component) = proto_stream.mut_component(component_id) else { + drop(proto_agent); + core::mem::forget(stream); + return RiceError::ResourceNotFound; + }; + let ret = + proto_component.gather_candidates(&sockets, &stun_servers, turn_servers.as_slice()); + if let Err(err) = ret { + drop(proto_agent); + core::mem::forget(stream); + return match err { + AgentError::AlreadyInProgress => RiceError::AlreadyInProgress, + AgentError::ResourceNotFound => RiceError::ResourceNotFound, + _ => RiceError::Failed, + }; + } + } + + drop(proto_agent); + core::mem::forget(stream); + RiceError::Success + } +} + /// Start gathering candidates for a component with the provided local socket addresses. /// /// - `component`: The component to start gathering. @@ -3268,6 +3415,7 @@ mod tests { use alloc::string::ToString; + use std::collections::BTreeSet; use std::eprintln; #[test] @@ -3440,6 +3588,95 @@ mod tests { } } + #[test] + fn rice_stream_restart_capi() { + unsafe { + let agent = rice_agent_new(true, false); + let stream = rice_agent_add_stream(agent); + let local = + credentials_to_c(Credentials::new("luser".to_string(), "lpass".to_string())); + let remote = + credentials_to_c(Credentials::new("ruser".to_string(), "rpass".to_string())); + + rice_stream_set_local_credentials(stream, local); + rice_credentials_free(local); + rice_stream_set_remote_credentials(stream, remote); + rice_credentials_free(remote); + + let before = rice_stream_get_local_credentials(stream); + assert!(!before.is_null()); + let current_remote = rice_stream_get_remote_credentials(stream); + assert!(!current_remote.is_null()); + rice_credentials_free(current_remote); + + assert_eq!(rice_stream_restart(stream, 0), RiceError::Success); + + let after = rice_stream_get_local_credentials(stream); + assert!(!after.is_null()); + assert!(!rice_credentials_eq(before, after)); + assert!(rice_stream_get_remote_credentials(stream).is_null()); + + rice_credentials_free(before); + rice_credentials_free(after); + rice_stream_unref(stream); + rice_agent_unref(agent); + } + } + + #[test] + fn rice_stream_gather_candidates_capi() { + unsafe { + let addr: SocketAddr = "192.168.0.1:1000".parse().unwrap(); + let addr = RiceAddress::new(addr).into_c_full(); + let agent = rice_agent_new(true, false); + let stream = rice_agent_add_stream(agent); + let component1 = rice_stream_add_component(stream); + let component2 = rice_stream_add_component(stream); + let local = + credentials_to_c(Credentials::new("luser".to_string(), "lpass".to_string())); + let remote = + credentials_to_c(Credentials::new("ruser".to_string(), "rpass".to_string())); + + rice_stream_set_local_credentials(stream, local); + rice_credentials_free(local); + rice_stream_set_remote_credentials(stream, remote); + rice_credentials_free(remote); + + assert_eq!( + rice_stream_gather_candidates( + stream, + 1, + &addr, + &transport_type_to_c(TransportType::Udp), + 0, + core::ptr::null(), + core::ptr::null(), + ), + RiceError::Success + ); + rice_address_free(mut_override(addr)); + + let mut gathered_components = BTreeSet::new(); + for _ in 0..8 { + let mut poll = RiceAgentPoll::Closed; + rice_agent_poll(agent, 0, &mut poll); + if let RiceAgentPoll::GatheredCandidate(ref candidate) = poll { + gathered_components.insert(candidate.gathered.candidate.component_id); + } + rice_agent_poll_clear(&mut poll); + if gathered_components.len() == 2 { + break; + } + } + assert_eq!(gathered_components, BTreeSet::from([1, 2])); + + rice_component_unref(component1); + rice_component_unref(component2); + rice_stream_unref(stream); + rice_agent_unref(agent); + } + } + #[test] fn rice_agent_poll_transmit_null() { unsafe { diff --git a/rice-proto/src/component.rs b/rice-proto/src/component.rs index 5e9e9b3..21c3c94 100644 --- a/rice-proto/src/component.rs +++ b/rice-proto/src/component.rs @@ -380,6 +380,10 @@ impl ComponentState { pub(crate) fn state(&self) -> ComponentConnectionState { self.state } + + pub(crate) fn clear_selected_pair(&mut self) { + self.selected_pair = None; + } } #[cfg(test)] @@ -506,10 +510,11 @@ mod tests { .build(); let pair = CandidatePair::new(local.clone(), remote.clone()); - let mut c = stream.mut_component(component_id).unwrap(); - c.set_selected_pair(pair).unwrap(); - assert!(c.selected_pair().is_some()); - drop(c); + { + let mut c = stream.mut_component(component_id).unwrap(); + c.set_selected_pair(pair).unwrap(); + assert!(c.selected_pair().is_some()); + } let pair2 = CandidatePair::new(local, remote); let mut c = stream.mut_component(component_id).unwrap(); diff --git a/rice-proto/src/conncheck.rs b/rice-proto/src/conncheck.rs index 5079ca0..b2102d6 100644 --- a/rice-proto/src/conncheck.rs +++ b/rice-proto/src/conncheck.rs @@ -43,7 +43,7 @@ use tracing::{debug, error, info, trace, warn}; static STUN_AGENT_COUNT: AtomicUsize = AtomicUsize::new(0); -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] pub(crate) struct StunAgentId(usize); impl core::ops::Deref for StunAgentId { @@ -526,6 +526,7 @@ pub struct ConnCheckList { component_ids: Vec<(usize, ComponentConnectionState)>, local_credentials: Credentials, remote_credentials: Credentials, + remote_credentials_set: bool, local_candidates: Vec, remote_candidates: Vec, // TODO: move to BinaryHeap or similar @@ -675,6 +676,7 @@ impl ConnCheckList { component_ids: Vec::new(), local_credentials, remote_credentials, + remote_credentials_set: true, local_candidates: Vec::new(), remote_candidates: Vec::new(), triggered: VecDeque::new(), @@ -714,6 +716,11 @@ impl ConnCheckList { self.local_credentials = credentials; } + #[cfg(test)] + pub(crate) fn local_credentials(&self) -> &Credentials { + &self.local_credentials + } + /// Set the remote [`Credentials`] for this checklist pub fn set_remote_credentials(&mut self, credentials: Credentials) { trace!( @@ -771,7 +778,26 @@ impl ConnCheckList { request.cancel(); } + let had_remote_credentials = self.remote_credentials_set; self.remote_credentials = credentials; + self.remote_credentials_set = true; + if !had_remote_credentials { + self.generate_checks(); + } + } + + #[cfg(test)] + pub(crate) fn remote_credentials(&self) -> Option<&Credentials> { + self.remote_credentials_set + .then_some(&self.remote_credentials) + } + + pub(crate) fn clear_remote_credentials(&mut self) { + let credentials = generate_random_credentials(); + self.stun_auth_remote + .set_credentials(credentials.clone().into(), IntegrityAlgorithm::Sha1); + self.remote_credentials = credentials; + self.remote_credentials_set = false; } /// Add a component id to this checklist @@ -1209,6 +1235,75 @@ impl ConnCheckList { self.dump_check_state(); } + pub(crate) fn restart(&mut self) { + self.state = CheckListState::Running; + self.remote_candidates.clear(); + self.triggered.clear(); + self.valid.clear(); + self.nominated.clear(); + self.local_end_of_candidates = false; + self.remote_end_of_candidates = false; + self.events.clear(); + self.pending_turn_permissions.clear(); + self.pending_recv.clear(); + self.pending_turn_tcp_connect.clear(); + self.tcp_buffers.clear(); + + let requests = self + .pairs + .iter_mut() + .filter_map(|check| check.agent_id().zip(check.stun_request.take())) + .collect::>(); + for (agent_id, request_id) in requests { + let Some(agent) = self.mut_agent_by_id(agent_id) else { + continue; + }; + let Some(mut request) = agent.mut_request_transaction(request_id) else { + continue; + }; + request.cancel(); + } + self.pairs.clear(); + + for (component_id, state) in self.component_ids.iter_mut() { + *state = ComponentConnectionState::Connecting; + self.events.push_front(ConnCheckEvent::ComponentState( + *component_id, + ComponentConnectionState::Connecting, + )); + } + } + + pub(crate) fn remove_component(&mut self, component_id: usize) { + let removed_check_ids = self + .pairs + .iter() + .filter(|check| check.pair.local.component_id == component_id) + .map(|check| check.conncheck_id) + .collect::>(); + self.component_ids.retain(|(id, _)| *id != component_id); + self.local_candidates + .retain(|candidate| candidate.candidate.component_id != component_id); + self.remote_candidates + .retain(|candidate| candidate.component_id != component_id); + self.triggered + .retain(|check_id| !removed_check_ids.contains(check_id)); + self.valid + .retain(|check_id| !removed_check_ids.contains(check_id)); + self.nominated + .retain(|(_check_id, pair)| pair.local.component_id != component_id); + self.pairs + .retain(|check| check.pair.local.component_id != component_id); + self.pending_recv + .retain(|pending| pending.component_id != component_id); + self.events.retain(|event| match event { + ConnCheckEvent::ComponentState(cid, _) + | ConnCheckEvent::SelectedPair(cid, _) + | ConnCheckEvent::ConsentResponseReceived(cid, _) => *cid != component_id, + }); + self.dump_check_state(); + } + fn next_triggered(&mut self) -> Option<&mut ConnCheck> { // triggered checks referenced by these ids may be removed before the check has a chance to // start. Simply remove them and continue processing. @@ -1430,7 +1525,7 @@ impl ConnCheckList { ) )] fn generate_checks(&mut self) { - if self.ice_lite { + if self.ice_lite || !self.remote_credentials_set { return; }; let mut checks = Vec::new(); @@ -2165,6 +2260,166 @@ impl ConnCheckList { self.pending_turn_permissions.clear(); self.agents.clear(); } + + fn component_for_agent_id(&self, agent_id: StunAgentId) -> Option { + self.pairs + .iter() + .find(|check| matches!(check.variant, ConnCheckVariant::Agent(id) if id == agent_id)) + .map(|check| check.pair.local.component_id) + .or_else(|| { + self.local_candidates.iter().find_map(|candidate| { + matches!(candidate.variant, LocalCandidateVariant::Agent(id) if id == agent_id) + .then_some(candidate.candidate.component_id) + }) + }) + } + + fn component_for_turn_id(&self, turn_id: StunAgentId) -> Option { + self.local_candidates.iter().find_map(|candidate| { + if candidate.candidate.candidate_type != CandidateType::Relayed { + return None; + } + let transport = candidate.candidate.transport_type; + let matches = self + .turn_client_by_allocated_address(transport, candidate.candidate.base_address) + .map(|(id, _)| id == turn_id) + .unwrap_or(false) + || self + .turn_client_by_allocated_address(transport, candidate.candidate.address) + .map(|(id, _)| id == turn_id) + .unwrap_or(false); + matches.then_some(candidate.candidate.component_id) + }) + } + + fn socket_from_agent( + &self, + component_id: usize, + agent: &StunAgent, + checklist_id: usize, + ) -> CheckListSetSocket { + CheckListSetSocket { + checklist_id, + component_id, + transport: agent.transport(), + local_addr: agent.local_addr(), + remote_addr: if agent.transport() == TransportType::Udp { + "0.0.0.0:0".parse().unwrap() + } else { + agent.remote_addr().unwrap() + }, + } + } + + fn socket_from_turn_client( + &self, + component_id: usize, + client: &TurnClient, + checklist_id: usize, + ) -> CheckListSetSocket { + CheckListSetSocket { + checklist_id, + component_id, + transport: client.transport(), + local_addr: client.local_addr(), + remote_addr: client.remote_addr(), + } + } + + fn sockets(&self) -> Vec { + let mut sockets = Vec::new(); + for agent in &self.agents { + let Some(component_id) = self.component_for_agent_id(agent.id) else { + continue; + }; + let socket = self.socket_from_agent(component_id, &agent.agent, self.checklist_id); + if !sockets.contains(&socket) { + sockets.push(socket); + } + } + for client in self + .turn_clients + .iter() + .chain(self.pending_delete_turn_clients.iter()) + { + let component_id = self + .component_for_turn_id(client.id) + .or_else(|| self.component_ids.first().map(|(id, _)| *id)) + .unwrap_or(1); + let socket = self.socket_from_turn_client(component_id, &client.client, self.checklist_id); + if !sockets.contains(&socket) { + sockets.push(socket); + } + } + sockets + } + + fn sockets_for_component(&self, component_id: usize) -> Vec { + self.sockets() + .into_iter() + .filter(|socket| socket.component_id == component_id) + .collect() + } + + fn cleanup_unused_runtime(&mut self) { + let used_agents = self + .pairs + .iter() + .filter_map(|check| match check.variant { + ConnCheckVariant::Agent(agent_id) => Some(agent_id), + _ => None, + }) + .chain(self.local_candidates.iter().filter_map(|candidate| { + if let LocalCandidateVariant::Agent(agent_id) = candidate.variant { + Some(agent_id) + } else { + None + } + })) + .collect::>(); + self.agents.retain(|agent| used_agents.contains(&agent.id)); + + let used_turns = self + .local_candidates + .iter() + .filter(|candidate| candidate.candidate.candidate_type == CandidateType::Relayed) + .filter_map(|candidate| { + self.turn_client_by_allocated_address( + candidate.candidate.transport_type, + candidate.candidate.base_address, + ) + .map(|(id, _)| id) + .or_else(|| { + self.turn_client_by_allocated_address( + candidate.candidate.transport_type, + candidate.candidate.address, + ) + .map(|(id, _)| id) + }) + }) + .collect::>(); + self.turn_clients + .retain(|client| used_turns.contains(&client.id)); + self.pending_delete_turn_clients + .retain(|client| used_turns.contains(&client.id)); + self.pending_turn_permissions + .retain(|(turn_id, _transport, _ip)| used_turns.contains(turn_id)); + self.pending_turn_tcp_connect.retain(|pending| { + used_turns.contains(&pending.turn_id) + && self + .pairs + .iter() + .any(|check| check.conncheck_id == pending.conncheck_id) + }); + let tcp_pairs = self + .agents + .iter() + .filter(|agent| agent.agent.transport() == TransportType::Tcp) + .map(|agent| (agent.agent.local_addr(), agent.agent.remote_addr().unwrap())) + .collect::>(); + self.tcp_buffers + .retain(|key, _| tcp_pairs.contains(key)); + } } /// A builder for a [`ConnCheckListSet`] @@ -2301,6 +2556,139 @@ impl ConnCheckListSet { self.checklists.iter().find(|cl| cl.checklist_id == id) } + pub(crate) fn restart_list(&mut self, checklist_id: usize, _now: Instant) { + let Some(checklist) = self.mut_list(checklist_id) else { + return; + }; + checklist.restart(); + self.pending_messages + .retain(|message| message.checklist_id != checklist_id); + self.pending_transmits + .retain(|transmit| transmit.checklist_id != checklist_id); + self.pending_remove_sockets + .retain(|socket| socket.checklist_id != checklist_id); + self.consent_tids + .retain(|(_tid, cid, _component_id)| *cid != checklist_id); + self.consent_responses + .retain(|(cid, _component_id, _revoked)| *cid != checklist_id); + self.completed = false; + } + + pub(crate) fn remove_component(&mut self, checklist_id: usize, component_id: usize) { + let Some(checklist_i) = self + .checklists + .iter() + .position(|checklist| checklist.checklist_id == checklist_id) + else { + return; + }; + let removed_sockets = self.checklists[checklist_i].sockets_for_component(component_id); + self.checklists[checklist_i].remove_component(component_id); + self.checklists[checklist_i].cleanup_unused_runtime(); + self.pending_remove_sockets.retain(|socket| { + socket.checklist_id != checklist_id || socket.component_id != component_id + }); + self.consent_tids.retain(|(_tid, cid, cid_component)| { + *cid != checklist_id || *cid_component != component_id + }); + self.consent_responses + .retain(|(cid, cid_component, _revoked)| { + *cid != checklist_id || *cid_component != component_id + }); + self.local_consent_revoked + .retain(|(cid, cid_component)| *cid != checklist_id || *cid_component != component_id); + self.completed = false; + + let still_used = self + .checklists + .iter() + .flat_map(|checklist| checklist.sockets()) + .collect::>(); + for socket in removed_sockets { + if !still_used.contains(&socket) && !self.pending_remove_sockets.contains(&socket) { + self.pending_remove_sockets.push_back(socket); + } + } + + let (valid_runtime_ids, valid_sockets) = self + .list(checklist_id) + .map(|checklist| { + let runtime_ids = checklist + .agents + .iter() + .map(|agent| agent.id) + .chain( + checklist + .turn_clients + .iter() + .chain(checklist.pending_delete_turn_clients.iter()) + .map(|client| client.id), + ) + .collect::>(); + (runtime_ids, checklist.sockets()) + }) + .unwrap_or_default(); + + self.pending_messages.retain(|message| { + if message.checklist_id != checklist_id { + return true; + } + valid_runtime_ids.contains(&message.agent_id) + }); + self.pending_transmits.retain(|transmit| { + if transmit.checklist_id != checklist_id { + return true; + } + valid_sockets.iter().any(|socket| { + socket.transport == transmit.transmit.transport + && socket.local_addr == transmit.transmit.from + && (socket.transport == TransportType::Udp + || socket.remote_addr == transmit.transmit.to) + }) + }); + } + + pub(crate) fn remove_list(&mut self, checklist_id: usize) { + let Some(checklist_i) = self + .checklists + .iter() + .position(|checklist| checklist.checklist_id == checklist_id) + else { + return; + }; + let removed_sockets = self.checklists[checklist_i].sockets(); + self.checklists.remove(checklist_i); + if self.checklist_i >= self.checklists.len() && !self.checklists.is_empty() { + self.checklist_i = self.checklists.len() - 1; + } else if self.checklists.is_empty() { + self.checklist_i = 0; + } + self.pending_messages + .retain(|message| message.checklist_id != checklist_id); + self.pending_transmits + .retain(|transmit| transmit.checklist_id != checklist_id); + self.pending_remove_sockets + .retain(|socket| socket.checklist_id != checklist_id); + self.consent_tids + .retain(|(_tid, cid, _component_id)| *cid != checklist_id); + self.consent_responses + .retain(|(cid, _component_id, _revoked)| *cid != checklist_id); + self.local_consent_revoked + .retain(|(cid, _component_id)| *cid != checklist_id); + self.completed = false; + + let still_used = self + .checklists + .iter() + .flat_map(|checklist| checklist.sockets()) + .collect::>(); + for socket in removed_sockets { + if !still_used.contains(&socket) && !self.pending_remove_sockets.contains(&socket) { + self.pending_remove_sockets.push_back(socket); + } + } + } + /// Whether the set is in the controlling mode. This may change during the ICE negotiation /// process. pub fn controlling(&self) -> bool { @@ -4700,7 +5088,7 @@ struct CheckListSetPendingMessage { consent_cid: Option, } -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] struct CheckListSetSocket { checklist_id: usize, component_id: usize, @@ -5267,7 +5655,7 @@ mod tests { .map_or(Err(IgnorableReason::IntegrityFailure), |username| { validate_username(username, local_credentials, remote_credentials) }) - .map_or(false, |_| true); + .is_ok(); let mut response = if ice_controlling.is_err() && ice_controlled.is_err() { warn!("missing ice controlled/controlling attribute"); diff --git a/rice-proto/src/consent.rs b/rice-proto/src/consent.rs index 5c6b1da..9db0a88 100644 --- a/rice-proto/src/consent.rs +++ b/rice-proto/src/consent.rs @@ -364,6 +364,11 @@ impl ConsentFreshness { pub(crate) fn set_config(&mut self, config: Config) { self.config = config; } + + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.entries.len() + } } #[cfg(test)] diff --git a/rice-proto/src/stream.rs b/rice-proto/src/stream.rs index d589454..bd79238 100644 --- a/rice-proto/src/stream.rs +++ b/rice-proto/src/stream.rs @@ -18,7 +18,7 @@ use stun_proto::Instant; use stun_proto::agent::{StunError, Transmit}; use stun_proto::types::data::Data; -use crate::agent::{Agent, AgentError}; +use crate::agent::{generate_restart_credentials, Agent, AgentError}; use crate::component::{Component, ComponentMut, ComponentState, GatherProgress}; use crate::conncheck::{HandleRecvReply, PendingRecv, RecvIgnorable, RequestRto}; @@ -255,6 +255,26 @@ impl<'a> StreamMut<'a> { Ok(component_id) } + /// Remove a `Component` from this stream without changing any other component ids. + /// + /// # Examples + /// + /// ``` + /// # use rice_proto::agent::Agent; + /// let mut agent = Agent::default(); + /// let stream_id = agent.add_stream(); + /// let mut stream = agent.mut_stream(stream_id).unwrap(); + /// let component_id = stream.add_component().unwrap(); + /// stream.remove_component(component_id); + /// assert!(stream.component(component_id).is_none()); + /// ``` + pub fn remove_component(&mut self, index: usize) { + if index < 1 { + return; + } + self.agent.remove_component(self.id, index); + } + /// Retrieve mutable access to a component in this stream. `None` will be returned if the /// component does not exist pub fn mut_component(&mut self, index: usize) -> Option> { @@ -311,6 +331,39 @@ impl<'a> StreamMut<'a> { checklist.set_remote_credentials(credentials); } + /// Restart this stream with explicit local ICE credentials. + /// + /// Existing local candidates, sockets, and TURN allocations are preserved; callers must + /// exchange the new credentials and remote candidates before connectivity checks can complete + /// again. + /// + /// # Examples + /// + /// ``` + /// # use rice_proto::agent::Agent; + /// # use rice_proto::stream::Credentials; + /// # use stun_proto::Instant; + /// let mut agent = Agent::default(); + /// let stream_id = agent.add_stream(); + /// let mut stream = agent.mut_stream(stream_id).unwrap(); + /// let credentials = Credentials::new("ufrag".into(), "a-very-long-password-value".into()); + /// stream.restart(credentials.clone(), Instant::ZERO); + /// assert_eq!(stream.local_credentials(), Some(credentials)); + /// assert!(stream.remote_credentials().is_none()); + /// ``` + pub fn restart(&mut self, local_credentials: Credentials, now: Instant) { + let _ = self + .agent + .restart_stream_with_credentials(self.id, local_credentials, now); + } + + /// Restart this stream with fresh random local ICE credentials. + pub fn restart_with_random_credentials(&mut self, now: Instant) -> Credentials { + let credentials = generate_restart_credentials(); + self.restart(credentials.clone(), now); + credentials + } + /// Add a remote candidate for connection checks for use with this stream /// /// # Examples @@ -424,7 +477,6 @@ impl<'a> StreamMut<'a> { ) )] pub fn end_of_remote_candidates(&mut self) { - // FIXME: how to deal with ice restarts? let stream_state = self.agent.mut_stream_state(self.id).unwrap(); let checklist_id = stream_state.checklist_id; let checklist = self.agent.checklistset.mut_list(checklist_id).unwrap(); @@ -560,7 +612,7 @@ impl StreamState { .unwrap_or((self.components.len(), &None)) .0; info!("adding component {}", index + 1); - if self.components.get(index).is_some() { + if matches!(self.components.get(index), Some(Some(_))) { return Err(AgentError::AlreadyExists); } while self.components.len() <= index { @@ -572,13 +624,21 @@ impl StreamState { Ok(index + 1) } + pub(crate) fn remove_component(&mut self, component_id: usize) -> bool { + if component_id < 1 || component_id > self.components.len() { + return false; + } + self.components[component_id - 1] = None; + true + } + #[tracing::instrument( skip(self), fields( stream.id = self.id ) )] - fn set_local_credentials(&mut self, credentials: Credentials) { + pub(crate) fn set_local_credentials(&mut self, credentials: Credentials) { info!("setting"); self.local_credentials = Some(credentials.clone()); } @@ -593,15 +653,29 @@ impl StreamState { stream.id = self.id() ) )] - fn set_remote_credentials(&mut self, credentials: Credentials) { + pub(crate) fn set_remote_credentials(&mut self, credentials: Credentials) { info!("setting"); self.remote_credentials = Some(credentials.clone()); } + pub(crate) fn clear_remote_credentials(&mut self) { + self.remote_credentials = None; + } + pub(crate) fn remote_credentials(&self) -> Option { self.remote_credentials.clone() } + pub(crate) fn component_ids_iter(&self) -> impl Iterator + '_ { + self.components.iter().flatten().map(|component| component.id) + } + + pub(crate) fn clear_selected_pairs(&mut self) { + for component in self.components.iter_mut().flatten() { + component.clear_selected_pair(); + } + } + pub(crate) fn handle_incoming_data + core::fmt::Debug>( &mut self, component_id: usize, @@ -702,8 +776,102 @@ impl StreamState { #[cfg(test)] mod tests { use super::*; + use alloc::string::String; + use alloc::vec; + use core::time::Duration; + use crate::agent::{Agent, AgentPoll}; use crate::candidate::{Candidate, CandidateType, TcpType, TransportType}; + use crate::component::ComponentConnectionState; + use stun_proto::auth::ShortTermAuth; + use stun_proto::types::attribute::XorMappedAddress; + use stun_proto::types::message::{ + IntegrityAlgorithm, Message, MessageWrite, MessageWriteExt, MessageWriteVec, + }; + + fn host_candidate(component_id: usize, addr: SocketAddr) -> Candidate { + Candidate::builder( + component_id, + CandidateType::Host, + TransportType::Udp, + "foundation", + addr, + ) + .priority(1234) + .build() + } + + fn test_password(fill: char) -> String { + core::iter::repeat_n(fill, 24).collect() + } + + fn reply_to_check( + agent: &mut Agent, + stream_id: usize, + component_id: usize, + transmit: crate::agent::AgentTransmit, + remote: &Credentials, + now: Instant, + ) { + let request = Message::from_bytes(&transmit.transmit.data).unwrap(); + let mut auth = ShortTermAuth::new(); + auth.set_credentials(remote.clone().into(), IntegrityAlgorithm::Sha1); + let mut response = Message::builder_success(&request, MessageWriteVec::new()); + response + .add_attribute(&XorMappedAddress::new(transmit.transmit.from, request.transaction_id())) + .unwrap(); + let mut response = auth.sign_outgoing_message(response).unwrap(); + response.add_fingerprint().unwrap(); + let response = response.finish(); + let reply = Transmit::new( + response.as_slice(), + transmit.transmit.transport, + transmit.transmit.to, + transmit.transmit.from, + ); + let handled = agent + .mut_stream(stream_id) + .unwrap() + .handle_incoming_data(component_id, reply, now); + assert!(handled.handled); + } + + fn complete_check_cycle( + agent: &mut Agent, + stream_id: usize, + component_id: usize, + remote: &Credentials, + mut now: Instant, + ) -> Instant { + for _ in 0..16 { + match agent.poll(now) { + AgentPoll::ComponentStateChange(change) => { + assert_eq!(change.stream_id, stream_id); + assert_eq!(change.component_id, component_id); + assert!( + matches!( + change.state, + ComponentConnectionState::Connecting | ComponentConnectionState::Connected + ), + "unexpected component state {change:?}" + ); + } + AgentPoll::SelectedPair(selected) => { + assert_eq!(selected.stream_id, stream_id); + assert_eq!(selected.component_id, component_id); + return now; + } + AgentPoll::WaitUntil(wait) => { + now = wait; + } + other => panic!("unexpected poll result during conncheck cycle: {other:?}"), + } + if let Some(transmit) = agent.poll_transmit(now) { + reply_to_check(agent, stream_id, component_id, transmit, remote, now); + } + } + panic!("connectivity checks did not complete"); + } #[test] fn getters_setters() { @@ -813,4 +981,169 @@ mod tests { "checklist should emit STUN after allocated_socket" ); } + + #[test] + fn restart_preserves_local_candidates_and_clears_remote_credentials() { + let _log = crate::tests::test_init_log(); + let now = Instant::ZERO; + let mut agent = Agent::default(); + let stream_id = agent.add_stream(); + let component_id = agent.mut_stream(stream_id).unwrap().add_component().unwrap(); + let old_local = Credentials::new("luser".into(), test_password('l')); + let old_remote = Credentials::new("ruser".into(), test_password('r')); + let local_candidate = host_candidate(component_id, "10.0.0.1:1000".parse().unwrap()); + { + let mut stream = agent.mut_stream(stream_id).unwrap(); + stream.set_local_credentials(old_local); + stream.set_remote_credentials(old_remote); + assert!(stream.add_local_candidate(local_candidate.clone())); + stream.add_remote_candidate(host_candidate(component_id, "10.0.0.2:2000".parse().unwrap())); + stream.end_of_local_candidates(); + stream.end_of_remote_candidates(); + } + + let before = agent + .stream(stream_id) + .unwrap() + .local_candidates() + .cloned() + .collect::>(); + let new_local = Credentials::new("next".into(), test_password('n')); + agent.mut_stream(stream_id).unwrap().restart(new_local.clone(), now); + + let stream = agent.stream(stream_id).unwrap(); + let after = stream.local_candidates().cloned().collect::>(); + assert_eq!(after, before); + assert_eq!(stream.local_credentials(), Some(new_local.clone())); + assert!(stream.remote_credentials().is_none()); + + let checklist_id = agent.stream_state(stream_id).unwrap().checklist_id; + let checklist = agent.checklistset.list(checklist_id).unwrap(); + assert_eq!(checklist.local_credentials(), &new_local); + assert!(checklist.remote_credentials().is_none()); + + match agent.poll(now) { + AgentPoll::ComponentStateChange(change) => { + assert_eq!(change.component_id, component_id); + assert_eq!(change.state, ComponentConnectionState::Connecting); + } + other => panic!("expected restart to reset component state, got {other:?}"), + } + assert!( + !matches!(agent.poll(now), AgentPoll::RemoveSocket(_)), + "restart must preserve sockets backing local candidates" + ); + } + + #[test] + fn restart_rearms_end_of_candidates_and_reselects_pair() { + let _log = crate::tests::test_init_log(); + let now = Instant::ZERO; + let config = crate::consent::Config { + interval: Duration::from_secs(5), + timeout: Duration::from_secs(30), + }; + let mut agent = Agent::builder() + .controlling(true) + .consent_freshness_config(config) + .build(); + let stream_id = agent.add_stream(); + let component_id = agent.mut_stream(stream_id).unwrap().add_component().unwrap(); + let local_addr: SocketAddr = "10.0.0.1:1000".parse().unwrap(); + let remote_addr: SocketAddr = "10.0.0.2:2000".parse().unwrap(); + let local_candidate = host_candidate(component_id, local_addr); + let remote_candidate = host_candidate(component_id, remote_addr); + let first_local = Credentials::new("luser".into(), test_password('a')); + let first_remote = Credentials::new("ruser".into(), test_password('b')); + { + let mut stream = agent.mut_stream(stream_id).unwrap(); + stream.set_local_credentials(first_local); + stream.set_remote_credentials(first_remote.clone()); + assert!(stream.add_local_candidate(local_candidate.clone())); + stream.add_remote_candidate(remote_candidate.clone()); + stream.end_of_local_candidates(); + stream.end_of_remote_candidates(); + } + + let mut now = complete_check_cycle(&mut agent, stream_id, component_id, &first_remote, now); + match agent.poll(now) { + AgentPoll::ComponentStateChange(change) => { + assert_eq!(change.component_id, component_id); + assert_eq!(change.state, ComponentConnectionState::Connected); + } + other => panic!("expected Connected after initial SelectedPair, got {other:?}"), + } + assert_eq!(agent.consent_freshness.as_ref().unwrap().len(), 1); + + let second_local = agent + .mut_stream(stream_id) + .unwrap() + .restart_with_random_credentials(now); + assert!(second_local.ufrag.len() >= 4); + assert!(second_local.passwd.len() >= 22); + assert_eq!(agent.consent_freshness.as_ref().unwrap().len(), 0); + + match agent.poll(now) { + AgentPoll::ComponentStateChange(change) => { + assert_eq!(change.component_id, component_id); + assert_eq!(change.state, ComponentConnectionState::Connecting); + } + other => panic!("expected restart to emit Connecting, got {other:?}"), + } + + let second_remote = Credentials::new("rnew".into(), test_password('c')); + { + let mut stream = agent.mut_stream(stream_id).unwrap(); + assert_eq!(stream.local_credentials(), Some(second_local.clone())); + assert!(stream.remote_credentials().is_none()); + stream.set_remote_credentials(second_remote.clone()); + stream.add_remote_candidate(remote_candidate); + stream.end_of_local_candidates(); + stream.end_of_remote_candidates(); + } + + now = complete_check_cycle(&mut agent, stream_id, component_id, &second_remote, now); + assert_eq!(agent.consent_freshness.as_ref().unwrap().len(), 1); + assert!(!matches!(agent.poll(now), AgentPoll::RemoveSocket(_))); + } + + #[test] + fn remove_component_removes_resources_and_reuses_slot() { + let _log = crate::tests::test_init_log(); + let now = Instant::ZERO; + let mut agent = Agent::default(); + let stream_id = agent.add_stream(); + let (component1, component2) = { + let mut stream = agent.mut_stream(stream_id).unwrap(); + let component1 = stream.add_component().unwrap(); + let component2 = stream.add_component().unwrap(); + assert!(stream.add_local_candidate(host_candidate(component1, "10.0.0.1:1000".parse().unwrap()))); + assert!(stream.add_local_candidate(host_candidate(component2, "10.0.0.1:1002".parse().unwrap()))); + (component1, component2) + }; + + agent + .mut_stream(stream_id) + .unwrap() + .remove_component(component1); + let stream = agent.stream(stream_id).unwrap(); + assert!(stream.component(component1).is_none()); + assert!(stream.component(component2).is_some()); + assert_eq!( + stream + .local_candidates() + .map(|candidate| candidate.component_id) + .collect::>(), + vec![component2] + ); + + let AgentPoll::RemoveSocket(removed) = agent.poll(now) else { + panic!("expected removed component socket"); + }; + assert_eq!(removed.stream_id, stream_id); + assert_eq!(removed.component_id, component1); + + let reused = agent.mut_stream(stream_id).unwrap().add_component().unwrap(); + assert_eq!(reused, component1); + } }