| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| use std::{ |
| collections::HashSet, |
| path::PathBuf, |
| sync::{Arc, Mutex, RwLock}, |
| time::Duration, |
| }; |
| |
| use libp2p::{Multiaddr, PeerId}; |
| use napi::bindgen_prelude::{Buffer, Error, Result, Status}; |
| use napi_derive::napi; |
| use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot, watch}; |
| |
| use crate::engine::{ |
| self, DirectTransport, EngineCommand, PeerConnectionPath, PeerError, StreamCommand, |
| }; |
| |
| type IncomingStreamReceiver = mpsc::Receiver<std::result::Result<Vec<u8>, PeerError>>; |
| const IDENTITY_PAYLOAD_MAX_BYTES: usize = 8 * 1024; |
| const MAX_CONNECT_ROUTES_PER_CLASS: usize = 32; |
| const MAX_TRANSIT_PEERS: usize = 64; |
| const MAX_TRANSIT_RELAY_ADDRESSES: usize = 256; |
| const MAX_WEBRTC_STUN_URLS: usize = 8; |
| const MAX_WEBRTC_STUN_URL_BYTES: usize = 512; |
| |
| #[napi(object)] |
| pub struct StartPeerEndpointOptions { |
| pub key_path: String, |
| pub relay_anchor_path: Option<String>, |
| pub expected_peer_id: Option<String>, |
| pub listen_addresses: Option<Vec<String>>, |
| pub coordination_relays: Option<Vec<String>>, |
| pub automatic_relay_discovery: Option<bool>, |
| pub web_rtc_stun_urls: Option<Vec<String>>, |
| } |
| |
| #[napi(object)] |
| pub struct ConnectPeerOptions { |
| pub request_id: u32, |
| pub peer_id: String, |
| pub route_hints: Vec<String>, |
| pub coordination_relays: Option<Vec<String>>, |
| pub transit_relay_peer_ids: Option<Vec<String>>, |
| pub direct_deadline_ms: u32, |
| } |
| |
| #[napi(object)] |
| pub struct UpdatePeerConnectOptions { |
| pub request_id: u32, |
| pub route_hints: Vec<String>, |
| pub coordination_relays: Option<Vec<String>>, |
| pub transit_relay_peer_ids: Option<Vec<String>>, |
| } |
| |
| #[napi(object)] |
| pub struct ConfigurePeerTransitOptions { |
| pub allowed_peer_ids: Vec<String>, |
| pub approved_relay_peer_ids: Vec<String>, |
| pub relay_candidates: Vec<PeerTransitRelayCandidate>, |
| } |
| |
| #[napi(object)] |
| pub struct PeerTransitRelayCandidate { |
| pub peer_id: String, |
| pub addresses: Vec<String>, |
| pub coordination_relays: Vec<String>, |
| } |
| |
| #[napi(object)] |
| pub struct PeerTransitSnapshot { |
| pub allowed_peer_count: u32, |
| pub active_reservation_count: u32, |
| pub active_circuit_count: u32, |
| pub max_reservation_count: u32, |
| pub max_circuit_count: u32, |
| pub max_circuits_per_peer: u32, |
| pub max_circuit_duration_seconds: u32, |
| pub max_circuit_bytes: u32, |
| } |
| |
| #[napi(object)] |
| pub struct PeerIdentitySignature { |
| pub public_key: Buffer, |
| pub signature: Buffer, |
| } |
| |
| #[napi(object)] |
| #[derive(Clone)] |
| pub struct PeerReachabilitySnapshot { |
| pub generation: u32, |
| pub listen_addresses: Vec<String>, |
| pub active_coordination_relays: Vec<String>, |
| } |
| |
| #[napi(object)] |
| pub struct PeerConnectivitySnapshot { |
| pub generation: u32, |
| pub connected_peer_ids: Vec<String>, |
| } |
| |
| #[napi] |
| pub struct PeerEndpoint { |
| peer_id: String, |
| reachability: watch::Receiver<engine::ReachabilitySnapshot>, |
| connectivity: watch::Receiver<engine::ConnectivitySnapshot>, |
| transit_snapshot: Arc<RwLock<engine::TransitSnapshot>>, |
| commands: mpsc::Sender<EngineCommand>, |
| incoming: Arc<AsyncMutex<mpsc::Receiver<engine::PeerStream>>>, |
| mesh_incoming: Arc<AsyncMutex<mpsc::Receiver<engine::PeerStream>>>, |
| terminal: Arc<AsyncMutex<mpsc::Receiver<PeerError>>>, |
| thread: Arc<Mutex<Option<std::thread::JoinHandle<()>>>>, |
| } |
| |
| #[napi] |
| impl PeerEndpoint { |
| #[napi(getter)] |
| pub fn peer_id(&self) -> String { |
| self.peer_id.clone() |
| } |
| |
| #[napi(getter)] |
| pub fn reachability_snapshot(&self) -> PeerReachabilitySnapshot { |
| reachability_snapshot(&self.reachability.borrow()) |
| } |
| |
| #[napi] |
| pub async fn watch_reachability(&self, after_generation: u32, timeout_ms: u32) -> Result<u32> { |
| if !(1..=300_000).contains(&timeout_ms) { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "reachability watch timeout must be between 1 and 300000 milliseconds", |
| )); |
| } |
| let mut receiver = self.reachability.clone(); |
| match tokio::time::timeout( |
| Duration::from_millis(u64::from(timeout_ms)), |
| receiver.wait_for(|snapshot| snapshot.generation != after_generation), |
| ) |
| .await |
| { |
| Ok(Ok(snapshot)) => return Ok(snapshot.generation), |
| Ok(Err(_)) => return Err(native_closed_error()), |
| Err(_) => {} |
| } |
| Ok(receiver.borrow().generation) |
| } |
| |
| #[napi(getter)] |
| pub fn connectivity_snapshot(&self) -> PeerConnectivitySnapshot { |
| connectivity_snapshot(&self.connectivity.borrow()) |
| } |
| |
| #[napi] |
| pub async fn watch_connectivity( |
| &self, |
| after_generation: u32, |
| timeout_ms: u32, |
| ) -> Result<PeerConnectivitySnapshot> { |
| if !(1..=300_000).contains(&timeout_ms) { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "connectivity watch timeout must be between 1 and 300000 milliseconds", |
| )); |
| } |
| let mut receiver = self.connectivity.clone(); |
| match tokio::time::timeout( |
| Duration::from_millis(u64::from(timeout_ms)), |
| receiver.wait_for(|snapshot| snapshot.generation != after_generation), |
| ) |
| .await |
| { |
| Ok(Ok(snapshot)) => return Ok(connectivity_snapshot(&snapshot)), |
| Ok(Err(_)) => return Err(native_closed_error()), |
| Err(_) => {} |
| } |
| Ok(connectivity_snapshot(&receiver.borrow())) |
| } |
| |
| #[napi(getter)] |
| pub fn transit_snapshot(&self) -> PeerTransitSnapshot { |
| let snapshot = self |
| .transit_snapshot |
| .read() |
| .map(|snapshot| snapshot.clone()) |
| .unwrap_or_default(); |
| PeerTransitSnapshot { |
| allowed_peer_count: snapshot.allowed_peer_count as u32, |
| active_reservation_count: snapshot.active_reservation_count as u32, |
| active_circuit_count: snapshot.active_circuit_count as u32, |
| max_reservation_count: snapshot.max_reservation_count as u32, |
| max_circuit_count: snapshot.max_circuit_count as u32, |
| max_circuits_per_peer: snapshot.max_circuits_per_peer as u32, |
| max_circuit_duration_seconds: snapshot.max_circuit_duration_seconds as u32, |
| max_circuit_bytes: snapshot.max_circuit_bytes as u32, |
| } |
| } |
| |
| #[napi] |
| pub async fn configure_transit(&self, options: ConfigurePeerTransitOptions) -> Result<()> { |
| let allowed_peers = parse_peer_ids(options.allowed_peer_ids)?; |
| let approved_relays = parse_peer_ids(options.approved_relay_peer_ids)?; |
| let relays = parse_transit_relay_candidates(options.relay_candidates)?; |
| let trusted_relays = relays |
| .iter() |
| .map(|candidate| candidate.peer_id) |
| .collect::<HashSet<_>>(); |
| let local_peer_id = parse_peer_id(&self.peer_id)?; |
| if allowed_peers.contains(&local_peer_id) |
| || approved_relays.contains(&local_peer_id) |
| || trusted_relays.contains(&local_peer_id) |
| { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "peer endpoint cannot configure itself as a transit peer", |
| )); |
| } |
| let (result_tx, result_rx) = oneshot::channel(); |
| self.commands |
| .send(EngineCommand::ConfigureTransit { |
| policy: engine::TransitPolicy { |
| allowed_peers, |
| approved_relays, |
| relays, |
| }, |
| result: result_tx, |
| }) |
| .await |
| .map_err(|_| native_closed_error())?; |
| result_rx.await.map_err(|_| native_closed_error()) |
| } |
| |
| #[napi] |
| pub async fn connect(&self, options: ConnectPeerOptions) -> Result<PeerStream> { |
| connect_peer(self, options, engine::StreamKind::Application).await |
| } |
| |
| #[napi] |
| pub async fn connect_mesh_control(&self, options: ConnectPeerOptions) -> Result<PeerStream> { |
| connect_peer(self, options, engine::StreamKind::MeshControl).await |
| } |
| |
| #[napi] |
| pub async fn accept_mesh_control(&self) -> Result<Option<PeerStream>> { |
| self.mesh_incoming |
| .lock() |
| .await |
| .recv() |
| .await |
| .map(wrap_stream) |
| .transpose() |
| } |
| |
| #[napi] |
| pub async fn cancel_connect(&self, request_id: u32) -> Result<bool> { |
| let (result_tx, result_rx) = oneshot::channel(); |
| self.commands |
| .send(EngineCommand::CancelConnect { |
| request_id, |
| result: result_tx, |
| }) |
| .await |
| .map_err(|_| native_closed_error())?; |
| result_rx.await.map_err(|_| native_closed_error()) |
| } |
| |
| #[napi] |
| pub async fn update_connect(&self, options: UpdatePeerConnectOptions) -> Result<bool> { |
| let route_hints = parse_connect_addresses(options.route_hints, "route hint")?; |
| let coordination_relays = parse_connect_addresses( |
| options.coordination_relays.unwrap_or_default(), |
| "coordination relay", |
| )?; |
| let transit_relay_peers = |
| parse_peer_id_list(options.transit_relay_peer_ids.unwrap_or_default())?; |
| let (result_tx, result_rx) = oneshot::channel(); |
| self.commands |
| .send(EngineCommand::UpdateConnect { |
| request_id: options.request_id, |
| candidates: engine::ConnectCandidates { |
| route_hints, |
| coordination_relays, |
| transit_relay_peers, |
| }, |
| result: result_tx, |
| }) |
| .await |
| .map_err(|_| native_closed_error())?; |
| result_rx |
| .await |
| .map_err(|_| native_closed_error())? |
| .map_err(peer_error) |
| } |
| |
| #[napi] |
| pub async fn accept(&self) -> Result<Option<PeerStream>> { |
| let mut incoming = self.incoming.lock().await; |
| let mut terminal = self.terminal.lock().await; |
| tokio::select! { |
| error = terminal.recv() => match error { |
| Some(error) => Err(peer_error(error)), |
| None => Ok(None), |
| }, |
| stream = incoming.recv() => stream.map(wrap_stream).transpose(), |
| } |
| } |
| |
| #[napi] |
| pub async fn close(&self) -> Result<()> { |
| let (result_tx, result_rx) = oneshot::channel(); |
| if self |
| .commands |
| .send(EngineCommand::Stop { result: result_tx }) |
| .await |
| .is_ok() |
| { |
| let _ = result_rx.await; |
| } |
| let thread = self |
| .thread |
| .lock() |
| .map_err(|_| Error::new(Status::GenericFailure, "peer endpoint lock poisoned"))? |
| .take(); |
| if let Some(thread) = thread { |
| tokio::task::spawn_blocking(move || thread.join()) |
| .await |
| .map_err(|error| Error::new(Status::GenericFailure, error.to_string()))? |
| .map_err(|_| Error::new(Status::GenericFailure, "peer endpoint thread panicked"))?; |
| } |
| Ok(()) |
| } |
| } |
| |
| async fn connect_peer( |
| endpoint: &PeerEndpoint, |
| options: ConnectPeerOptions, |
| stream_kind: engine::StreamKind, |
| ) -> Result<PeerStream> { |
| let peer_id = parse_peer_id(&options.peer_id)?; |
| let route_hints = parse_connect_addresses(options.route_hints, "route hint")?; |
| let coordination_relays = parse_connect_addresses( |
| options.coordination_relays.unwrap_or_default(), |
| "coordination relay", |
| )?; |
| let transit_relay_peers = |
| parse_peer_id_list(options.transit_relay_peer_ids.unwrap_or_default())?; |
| if !(1..=120_000).contains(&options.direct_deadline_ms) { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "direct deadline must be between 1 and 120000 milliseconds", |
| )); |
| } |
| let (result_tx, result_rx) = oneshot::channel(); |
| endpoint |
| .commands |
| .send(EngineCommand::Connect { |
| options: engine::ConnectOptions { |
| request_id: options.request_id, |
| peer_id, |
| route_hints, |
| coordination_relays, |
| transit_relay_peers, |
| deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)), |
| }, |
| stream_kind, |
| result: result_tx, |
| }) |
| .await |
| .map_err(|_| { |
| peer_error(PeerError { |
| code: "peer_native_failed", |
| message: "peer endpoint is closed".to_owned(), |
| }) |
| })?; |
| wrap_stream( |
| result_rx |
| .await |
| .map_err(|_| native_closed_error())? |
| .map_err(peer_error)?, |
| ) |
| } |
| |
| impl Drop for PeerEndpoint { |
| fn drop(&mut self) { |
| if Arc::strong_count(&self.thread) == 1 { |
| let (result, _) = oneshot::channel(); |
| let _ = self.commands.try_send(EngineCommand::Stop { result }); |
| } |
| } |
| } |
| |
| #[napi] |
| pub struct PeerStream { |
| peer_id: String, |
| path: PeerStreamPath, |
| incoming: Arc<AsyncMutex<IncomingStreamReceiver>>, |
| commands: mpsc::Sender<StreamCommand>, |
| abort: watch::Sender<bool>, |
| } |
| |
| #[napi(object)] |
| #[derive(Clone)] |
| pub struct PeerStreamPath { |
| pub kind: String, |
| pub transport: Option<String>, |
| pub relay_peer_id: Option<String>, |
| } |
| |
| #[napi] |
| impl PeerStream { |
| #[napi(getter)] |
| pub fn peer_id(&self) -> String { |
| self.peer_id.clone() |
| } |
| |
| #[napi(getter)] |
| pub fn path(&self) -> PeerStreamPath { |
| self.path.clone() |
| } |
| |
| #[napi] |
| pub async fn read(&self) -> Result<Option<Buffer>> { |
| match self.incoming.lock().await.recv().await { |
| Some(Ok(bytes)) => Ok(Some(bytes.into())), |
| Some(Err(error)) => Err(peer_error(error)), |
| None => Ok(None), |
| } |
| } |
| |
| #[napi] |
| pub async fn write(&self, bytes: Buffer) -> Result<()> { |
| let (result_tx, result_rx) = oneshot::channel(); |
| self.commands |
| .send(StreamCommand::Write { |
| bytes: bytes.to_vec(), |
| result: result_tx, |
| }) |
| .await |
| .map_err(|_| native_closed_error())?; |
| result_rx |
| .await |
| .map_err(|_| native_closed_error())? |
| .map_err(peer_error) |
| } |
| |
| #[napi] |
| pub async fn close(&self) -> Result<()> { |
| let (result_tx, result_rx) = oneshot::channel(); |
| if self |
| .commands |
| .send(StreamCommand::Close { result: result_tx }) |
| .await |
| .is_err() |
| { |
| return Ok(()); |
| } |
| match result_rx.await { |
| Ok(result) => result.map_err(peer_error), |
| Err(_) => Ok(()), |
| } |
| } |
| |
| #[napi] |
| pub fn abort(&self) { |
| self.abort.send_replace(true); |
| } |
| } |
| |
| #[napi] |
| pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result<PeerEndpoint> { |
| let started = engine::start(engine::StartOptions { |
| key_path: PathBuf::from(options.key_path), |
| relay_anchor_path: options.relay_anchor_path.map(PathBuf::from), |
| expected_peer_id: options |
| .expected_peer_id |
| .map(|value| parse_peer_id(&value)) |
| .transpose()?, |
| listen_addresses: parse_addresses(options.listen_addresses.unwrap_or_default(), "listen")?, |
| coordination_relays: parse_addresses( |
| options.coordination_relays.unwrap_or_default(), |
| "coordination relay", |
| )?, |
| automatic_relay_discovery: options.automatic_relay_discovery.unwrap_or(false), |
| web_rtc_stun_urls: options |
| .web_rtc_stun_urls |
| .map(parse_webrtc_stun_urls) |
| .transpose()?, |
| }) |
| .map_err(peer_error)?; |
| Ok(PeerEndpoint { |
| peer_id: started.peer_id.to_string(), |
| reachability: started.reachability, |
| connectivity: started.connectivity, |
| transit_snapshot: started.transit_snapshot, |
| commands: started.commands, |
| incoming: Arc::new(AsyncMutex::new(started.incoming)), |
| mesh_incoming: Arc::new(AsyncMutex::new(started.mesh_incoming)), |
| terminal: Arc::new(AsyncMutex::new(started.terminal)), |
| thread: Arc::new(Mutex::new(Some(started.thread))), |
| }) |
| } |
| |
| fn reachability_snapshot(snapshot: &engine::ReachabilitySnapshot) -> PeerReachabilitySnapshot { |
| PeerReachabilitySnapshot { |
| generation: snapshot.generation, |
| listen_addresses: snapshot |
| .listen_addresses |
| .iter() |
| .map(ToString::to_string) |
| .collect(), |
| active_coordination_relays: snapshot |
| .active_coordination_relays |
| .iter() |
| .map(ToString::to_string) |
| .collect(), |
| } |
| } |
| |
| fn connectivity_snapshot(snapshot: &engine::ConnectivitySnapshot) -> PeerConnectivitySnapshot { |
| PeerConnectivitySnapshot { |
| generation: snapshot.generation, |
| connected_peer_ids: snapshot |
| .connected_peers |
| .iter() |
| .map(ToString::to_string) |
| .collect(), |
| } |
| } |
| |
| #[napi] |
| pub async fn ensure_peer_identity(key_path: String) -> Result<String> { |
| engine::ensure_identity(PathBuf::from(key_path)) |
| .await |
| .map(|peer_id| peer_id.to_string()) |
| .map_err(peer_error) |
| } |
| |
| #[napi] |
| pub async fn sign_peer_identity( |
| key_path: String, |
| expected_peer_id: String, |
| payload: Buffer, |
| ) -> Result<PeerIdentitySignature> { |
| validate_identity_payload(&payload)?; |
| let signed = engine::sign_identity( |
| PathBuf::from(key_path), |
| parse_peer_id(&expected_peer_id)?, |
| &payload, |
| ) |
| .await |
| .map_err(peer_error)?; |
| Ok(PeerIdentitySignature { |
| public_key: signed.public_key.into(), |
| signature: signed.signature.into(), |
| }) |
| } |
| |
| #[napi] |
| pub fn verify_peer_identity( |
| peer_id: String, |
| public_key: Buffer, |
| payload: Buffer, |
| signature: Buffer, |
| ) -> Result<bool> { |
| validate_identity_payload(&payload)?; |
| engine::verify_identity(parse_peer_id(&peer_id)?, &public_key, &payload, &signature) |
| .map_err(peer_error) |
| } |
| |
| fn wrap_stream(stream: engine::PeerStream) -> Result<PeerStream> { |
| Ok(PeerStream { |
| peer_id: stream.peer_id.to_string(), |
| path: peer_stream_path(stream.path), |
| incoming: Arc::new(AsyncMutex::new(stream.incoming)), |
| commands: stream.commands, |
| abort: stream.abort, |
| }) |
| } |
| |
| fn peer_stream_path(path: PeerConnectionPath) -> PeerStreamPath { |
| match path { |
| PeerConnectionPath::Direct(transport) => PeerStreamPath { |
| kind: "direct".to_owned(), |
| transport: Some( |
| match transport { |
| DirectTransport::Quic => "quic", |
| DirectTransport::Tcp => "tcp", |
| DirectTransport::WebRtc => "webrtc", |
| DirectTransport::Other => "other", |
| } |
| .to_owned(), |
| ), |
| relay_peer_id: None, |
| }, |
| PeerConnectionPath::Transit { relay_peer_id } => PeerStreamPath { |
| kind: "transit".to_owned(), |
| transport: None, |
| relay_peer_id: Some(relay_peer_id.to_string()), |
| }, |
| } |
| } |
| |
| fn parse_peer_id(value: &str) -> Result<PeerId> { |
| value |
| .parse() |
| .map_err(|_| Error::new(Status::InvalidArg, "peer id is invalid")) |
| } |
| |
| fn parse_addresses(values: Vec<String>, label: &str) -> Result<Vec<Multiaddr>> { |
| values |
| .into_iter() |
| .map(|value| { |
| value.parse().map_err(|_| { |
| Error::new(Status::InvalidArg, format!("{label} multiaddr is invalid")) |
| }) |
| }) |
| .collect() |
| } |
| |
| fn parse_connect_addresses(values: Vec<String>, label: &str) -> Result<Vec<Multiaddr>> { |
| if values.len() > MAX_CONNECT_ROUTES_PER_CLASS { |
| return Err(Error::new( |
| Status::InvalidArg, |
| format!("peer connection cannot contain more than 32 {label}s"), |
| )); |
| } |
| parse_addresses(values, label) |
| } |
| |
| fn parse_webrtc_stun_urls(values: Vec<String>) -> Result<Vec<String>> { |
| if values.len() > MAX_WEBRTC_STUN_URLS { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "WebRTC cannot use more than 8 STUN URLs", |
| )); |
| } |
| for value in &values { |
| if value.len() > MAX_WEBRTC_STUN_URL_BYTES |
| || !value.starts_with("stun:") |
| || value.chars().any(char::is_whitespace) |
| { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "WebRTC STUN URL must use the stun: scheme and contain no whitespace", |
| )); |
| } |
| } |
| Ok(values) |
| } |
| |
| fn parse_transit_relay_candidates( |
| candidates: Vec<PeerTransitRelayCandidate>, |
| ) -> Result<Vec<engine::TransitRelayCandidate>> { |
| let address_count = candidates.iter().try_fold(0usize, |count, candidate| { |
| count |
| .checked_add(candidate.addresses.len())? |
| .checked_add(candidate.coordination_relays.len()) |
| }); |
| if address_count.is_none_or(|count| count > MAX_TRANSIT_RELAY_ADDRESSES) { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "transit policy cannot contain more than 256 relay addresses", |
| )); |
| } |
| let mut relays = Vec::new(); |
| for candidate in candidates { |
| let Ok(expected_peer) = candidate.peer_id.parse::<PeerId>() else { |
| continue; |
| }; |
| let mut addresses = Vec::new(); |
| for value in candidate.addresses { |
| let Ok(address) = value.parse::<Multiaddr>() else { |
| continue; |
| }; |
| if engine::transit_relay_peer_id(&address).ok() == Some(expected_peer) { |
| addresses.push(address); |
| } |
| } |
| let mut coordination_relays = candidate |
| .coordination_relays |
| .into_iter() |
| .filter_map(|value| value.parse::<Multiaddr>().ok()) |
| .filter(|address| { |
| engine::coordination_relay_peer_id(address) |
| .is_ok_and(|peer_id| peer_id != expected_peer) |
| }) |
| .collect::<Vec<_>>(); |
| addresses.sort_unstable_by_key(ToString::to_string); |
| addresses.dedup(); |
| coordination_relays.sort_unstable_by_key(ToString::to_string); |
| coordination_relays.dedup(); |
| if !addresses.is_empty() || !coordination_relays.is_empty() { |
| relays.push(engine::TransitRelayCandidate { |
| peer_id: expected_peer, |
| addresses, |
| coordination_relays, |
| }); |
| } |
| } |
| relays.sort_unstable_by_key(|candidate| candidate.peer_id.to_string()); |
| relays.dedup_by_key(|candidate| candidate.peer_id); |
| Ok(relays) |
| } |
| |
| fn parse_peer_ids(values: Vec<String>) -> Result<HashSet<PeerId>> { |
| Ok(parse_peer_id_list(values)?.into_iter().collect()) |
| } |
| |
| fn parse_peer_id_list(values: Vec<String>) -> Result<Vec<PeerId>> { |
| if values.len() > MAX_TRANSIT_PEERS { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "transit policy cannot contain more than 64 peers", |
| )); |
| } |
| let mut peers = Vec::new(); |
| for value in values { |
| let peer = parse_peer_id(&value)?; |
| if !peers.contains(&peer) { |
| peers.push(peer); |
| } |
| } |
| Ok(peers) |
| } |
| |
| fn validate_identity_payload(payload: &[u8]) -> Result<()> { |
| if payload.is_empty() || payload.len() > IDENTITY_PAYLOAD_MAX_BYTES { |
| return Err(Error::new( |
| Status::InvalidArg, |
| "identity payload must be between 1 and 8192 bytes", |
| )); |
| } |
| Ok(()) |
| } |
| |
| fn peer_error(error: PeerError) -> Error { |
| Error::new( |
| Status::GenericFailure, |
| format!("{}: {}", error.code, error.message), |
| ) |
| } |
| |
| fn native_closed_error() -> Error { |
| peer_error(PeerError { |
| code: "peer_native_failed", |
| message: "peer stream is closed".to_owned(), |
| }) |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| |
| fn endpoint_for_watch_tests() -> ( |
| PeerEndpoint, |
| watch::Sender<engine::ReachabilitySnapshot>, |
| watch::Sender<engine::ConnectivitySnapshot>, |
| ) { |
| let (reachability_tx, reachability) = watch::channel(Default::default()); |
| let (connectivity_tx, connectivity) = watch::channel(Default::default()); |
| let (commands, _command_rx) = mpsc::channel(1); |
| let (_incoming_tx, incoming) = mpsc::channel(1); |
| let (_mesh_incoming_tx, mesh_incoming) = mpsc::channel(1); |
| let (_terminal_tx, terminal) = mpsc::channel(1); |
| ( |
| PeerEndpoint { |
| peer_id: PeerId::random().to_string(), |
| reachability, |
| connectivity, |
| transit_snapshot: Arc::new(RwLock::new(Default::default())), |
| commands, |
| incoming: Arc::new(AsyncMutex::new(incoming)), |
| mesh_incoming: Arc::new(AsyncMutex::new(mesh_incoming)), |
| terminal: Arc::new(AsyncMutex::new(terminal)), |
| thread: Arc::new(Mutex::new(None)), |
| }, |
| reachability_tx, |
| connectivity_tx, |
| ) |
| } |
| |
| #[tokio::test] |
| async fn reachability_watch_waits_for_a_newer_generation() { |
| let (endpoint, reachability, _) = endpoint_for_watch_tests(); |
| reachability.send_replace(engine::ReachabilitySnapshot { |
| generation: 1, |
| ..Default::default() |
| }); |
| let watched = endpoint.watch_reachability(1, 1_000); |
| tokio::pin!(watched); |
| |
| assert!( |
| tokio::time::timeout(Duration::from_millis(10), &mut watched) |
| .await |
| .is_err() |
| ); |
| reachability.send_replace(engine::ReachabilitySnapshot { |
| generation: 2, |
| ..Default::default() |
| }); |
| assert_eq!(watched.await.expect("reachability watch"), 2); |
| } |
| |
| #[tokio::test] |
| async fn connectivity_watch_waits_for_a_newer_generation() { |
| let (endpoint, _, connectivity) = endpoint_for_watch_tests(); |
| connectivity.send_replace(engine::ConnectivitySnapshot { |
| generation: 1, |
| ..Default::default() |
| }); |
| let watched = endpoint.watch_connectivity(1, 1_000); |
| tokio::pin!(watched); |
| |
| assert!( |
| tokio::time::timeout(Duration::from_millis(10), &mut watched) |
| .await |
| .is_err() |
| ); |
| connectivity.send_replace(engine::ConnectivitySnapshot { |
| generation: 2, |
| ..Default::default() |
| }); |
| assert_eq!(watched.await.expect("connectivity watch").generation, 2); |
| } |
| |
| #[test] |
| fn transit_relay_addresses_are_bound_to_the_declared_peer() { |
| let expected = PeerId::random(); |
| let other = PeerId::random(); |
| let accepted = format!("/ip4/192.0.2.1/tcp/4001/p2p/{expected}"); |
| let coordination = format!("/ip4/198.51.100.1/tcp/4001/p2p/{other}"); |
| let relays = parse_transit_relay_candidates(vec![PeerTransitRelayCandidate { |
| peer_id: expected.to_string(), |
| addresses: vec![ |
| accepted.clone(), |
| format!("/ip4/192.0.2.2/tcp/4001/p2p/{other}"), |
| "not-a-multiaddr".to_owned(), |
| ], |
| coordination_relays: vec![ |
| coordination.clone(), |
| format!("/ip4/198.51.100.2/tcp/4001/p2p/{expected}"), |
| ], |
| }]) |
| .expect("candidate policy"); |
| |
| assert_eq!(relays.len(), 1); |
| assert_eq!( |
| relays[0].addresses, |
| vec![accepted.parse().expect("accepted multiaddr")], |
| ); |
| assert_eq!( |
| relays[0].coordination_relays, |
| vec![coordination.parse().expect("coordination multiaddr")], |
| ); |
| } |
| |
| #[test] |
| fn webrtc_configuration_accepts_explicit_host_only_ice_and_rejects_turn() { |
| assert_eq!( |
| parse_webrtc_stun_urls(Vec::new()).expect("host-only ICE"), |
| Vec::<String>::new() |
| ); |
| assert!(parse_webrtc_stun_urls(vec!["turn:relay.example:3478".to_owned()]).is_err()); |
| } |
| } |
