diff --git a/Cargo.toml b/Cargo.toml index 1d10463..5a69270 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -59,7 +59,7 @@ indexeddb = ["dep:idb", "serde"] redb = ["dep:redb", "dep:cbor4ii", "serde"] sqlite = ["dep:sqlx", "dep:cbor4ii", "serde"] -webrtc =["dep:libp2p-webrtc", "dep:libp2p-webrtc-websys"] +webrtc = ["dep:libp2p-webrtc", "dep:libp2p-webrtc-websys"] websocket = ["libp2p/websocket", "rcgen", "dep:pem", "libp2p/websocket-websys"] webtransport = ["libp2p/webtransport-websys"] @@ -80,15 +80,15 @@ futures-timer = "3.0.0" futures-timeout = "0.1.3" getrandom = { version = "0.2.15" } getrandom_03 = { version = "0.3.3", package = "getrandom" } -hickory-resolver = "0.25.2" +hickory-resolver = "0.26.1" idb = "0.6.5" indexmap = "2.10.0" -libp2p = { version = "0.56.0" } -libp2p-allow-block-list = "0.6.0" -libp2p-connection-limits = "0.6.0" -libp2p-stream = { version = "=0.4.0-alpha" } -libp2p-webrtc = { version = "=0.9.0-alpha.1", features = ["pem"] } -libp2p-webrtc-websys = "0.4.0" +libp2p = { version = "0.57.0" } +libp2p-allow-block-list = "0.7.0" +libp2p-connection-limits = "0.7.0" +libp2p-stream = { version = "0.5.0-alpha" } +libp2p-webrtc = { version = "=0.10.0-alpha", features = ["pem"] } +libp2p-webrtc-websys = "0.5.0" other-error = "0.1.1" pollable-map = "0.1.7" pem = { version = "3.0.5" } @@ -156,10 +156,18 @@ futures-timer = { workspace = true, features = ["wasm-bindgen"] } getrandom = { workspace = true, features = ["js"] } getrandom_03 = { workspace = true, features = ["wasm_js"] } idb = { workspace = true, optional = true } -libp2p = { features = ["macros", "serde", "wasm-bindgen"], workspace = true } +libp2p = { features = ["macros", "serde"], workspace = true } libp2p-webrtc-websys = { workspace = true, optional = true } send_wrapper = { workspace = true, features = ["futures"] } serde-wasm-bindgen.workspace = true tokio = { default-features = false, features = ["sync", "macros"], workspace = true } wasm-bindgen.workspace = true wasm-bindgen-futures.workspace = true + +[patch.crates-io] +libp2p = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } +libp2p-allow-block-list = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } +libp2p-connection-limits = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } +libp2p-stream = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } +libp2p-webrtc = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } +libp2p-webrtc-websys = { git = "https://github.com/libp2p/rust-libp2p.git", branch = "master" } \ No newline at end of file diff --git a/examples/upnp/src/main.rs b/examples/upnp/src/main.rs index d1f16ed..af34719 100644 --- a/examples/upnp/src/main.rs +++ b/examples/upnp/src/main.rs @@ -13,9 +13,11 @@ async fn main() -> std::io::Result<()> { println!("New listen address: {addr}") } SwarmEvent::Behaviour(BehaviourEvent::Upnp(event)) => match event { - UpnpEvent::NewExternalAddr(addr) => println!("New external address: {addr}"), - UpnpEvent::ExpiredExternalAddr(addr) => { - println!("Expired external address: {addr}") + UpnpEvent::NewExternalAddr { external_addr, .. } => { + println!("New external address: {external_addr}") + } + UpnpEvent::ExpiredExternalAddr { external_addr, .. } => { + println!("Expired external address: {external_addr}") } UpnpEvent::GatewayNotFound => println!("Gateway not found"), UpnpEvent::NonRoutableGateway => println!("Gateway is not routable"), diff --git a/src/behaviour/request_response/codec.rs b/src/behaviour/request_response/codec.rs index fc775de..ec04907 100644 --- a/src/behaviour/request_response/codec.rs +++ b/src/behaviour/request_response/codec.rs @@ -1,4 +1,3 @@ -use async_trait::async_trait; use bytes::Bytes; use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use libp2p::StreamProtocol; @@ -18,7 +17,6 @@ impl Codec { } } -#[async_trait] impl libp2p::request_response::Codec for Codec { type Protocol = StreamProtocol; type Request = Bytes; diff --git a/src/builder/transport.rs b/src/builder/transport.rs index 8564281..e5a6a74 100644 --- a/src/builder/transport.rs +++ b/src/builder/transport.rs @@ -156,6 +156,8 @@ pub enum DnsResolver { /// Cloudflare DNS Resolver #[default] Cloudflare, + /// Quad9 DNS Resolver + Quad9, /// Local DNS Resolver Local, /// No DNS Resolver @@ -167,12 +169,22 @@ pub enum DnsResolver { impl From for (ResolverConfig, ResolverOpts) { fn from(value: DnsResolver) -> Self { match value { - DnsResolver::Google => (ResolverConfig::google(), Default::default()), - DnsResolver::Cloudflare => (ResolverConfig::cloudflare(), Default::default()), + DnsResolver::Google => ( + ResolverConfig::udp_and_tcp(&hickory_resolver::config::GOOGLE), + Default::default(), + ), + DnsResolver::Cloudflare => ( + ResolverConfig::udp_and_tcp(&hickory_resolver::config::CLOUDFLARE), + Default::default(), + ), + DnsResolver::Quad9 => ( + ResolverConfig::udp_and_tcp(&hickory_resolver::config::QUAD9), + Default::default(), + ), DnsResolver::Local => { hickory_resolver::system_conf::read_system_conf().unwrap_or_default() } - DnsResolver::None => (ResolverConfig::new(), Default::default()), + DnsResolver::None => (ResolverConfig::default(), Default::default()), } } } diff --git a/src/handle.rs b/src/handle.rs index 359fed0..0d7d7be 100644 --- a/src/handle.rs +++ b/src/handle.rs @@ -8,6 +8,10 @@ pub(crate) mod floodsub; #[cfg(feature = "gossipsub")] pub(crate) mod gossipsub; mod peer_store; + +#[cfg(not(target_arch = "wasm32"))] +#[cfg(feature = "relay")] +mod relay_server; #[cfg(feature = "rendezvous")] pub(crate) mod rendezvous; #[cfg(feature = "request-response")] @@ -28,6 +32,8 @@ use crate::handle::floodsub::ConnexaFloodsub; #[cfg(feature = "gossipsub")] use crate::handle::gossipsub::ConnexaGossipsub; use crate::handle::peer_store::ConnexaPeerstore; +#[cfg(not(target_arch = "wasm32"))] +use crate::handle::relay_server::ConnexaRelayServer; #[cfg(feature = "rendezvous")] use crate::handle::rendezvous::ConnexaRendezvous; #[cfg(feature = "request-response")] @@ -137,6 +143,13 @@ where ConnexaRendezvous::new(self) } + /// Returns a handle for relay server functions + #[cfg(not(target_arch = "wasm32"))] + #[cfg(feature = "relay")] + pub fn relay_server(&self) -> ConnexaRelayServer<'_, T, K> { + ConnexaRelayServer::new(self) + } + /// Returns a handle to manage peer whitelist functionality pub fn whitelist(&self) -> ConnexaWhitelist<'_, T, K> { ConnexaWhitelist::new(self) diff --git a/src/handle/relay_server.rs b/src/handle/relay_server.rs new file mode 100644 index 0000000..78d1fcb --- /dev/null +++ b/src/handle/relay_server.rs @@ -0,0 +1,31 @@ +use crate::handle::Connexa; +use crate::types::RelayServerCommand; +use libp2p::relay::Status as RelayServerStatus; + +#[derive(Copy, Clone)] +pub struct ConnexaRelayServer<'a, T = (), K = crate::keystore::store::memory::MemoryKeystore> { + connexa: &'a Connexa, +} + +impl<'a, T, K> ConnexaRelayServer<'a, T, K> +where + T: Send + Sync + 'static, +{ + pub(crate) fn new(connexa: &'a Connexa) -> Self { + Self { connexa } + } + + pub async fn change_status( + &self, + status: impl Into>, + ) -> std::io::Result<()> { + let (tx, rx) = futures::channel::oneshot::channel(); + let status = status.into(); + self.connexa + .to_task + .clone() + .send(RelayServerCommand::StatusChanged { status, resp: tx }.into()) + .await?; + rx.await.map_err(std::io::Error::other)? + } +} diff --git a/src/task.rs b/src/task.rs index b285702..7f3a811 100644 --- a/src/task.rs +++ b/src/task.rs @@ -52,7 +52,7 @@ use futures::channel::{mpsc, oneshot}; use futures::future::BoxFuture; use futures::{FutureExt, StreamExt}; use futures_timer::Delay; -use indexmap::IndexMap; +use indexmap::{IndexMap, IndexSet}; #[cfg(feature = "gossipsub")] use libp2p::gossipsub::{MessageAcceptance, MessageId}; #[cfg(feature = "kad")] @@ -583,6 +583,11 @@ where Command::Rendezvous(rendezvous_command) => { self.process_rendezvous_command(rendezvous_command) } + #[cfg(not(target_arch = "wasm32"))] + #[cfg(feature = "relay")] + Command::RelayServer(relay_server_command) => { + self.process_relay_server_command(relay_server_command) + } Command::Custom(custom_command) => { (self.custom_task_callback)( swarm, diff --git a/src/task/relay.rs b/src/task/relay.rs index 219b797..793c9f6 100644 --- a/src/task/relay.rs +++ b/src/task/relay.rs @@ -1,5 +1,7 @@ use crate::behaviour::peer_store::store::Store; use crate::task::ConnexaTask; +#[cfg(not(target_arch = "wasm32"))] +use crate::types::RelayServerCommand; use libp2p::relay::{Event as RelayServerEvent, client::Event as RelayClientEvent}; use libp2p::swarm::NetworkBehaviour; use std::fmt::Debug; @@ -32,6 +34,7 @@ where } } + #[cfg(not(target_arch = "wasm32"))] pub fn process_relay_server_event(&mut self, event: RelayServerEvent) { match event { RelayServerEvent::ReservationReqAccepted { @@ -69,7 +72,37 @@ where } => { tracing::warn!(%src_peer_id, %dst_peer_id, ?error, "relay server circuit closed"); } + RelayServerEvent::StatusChanged { status } => { + tracing::info!(?status, "relay server status changed"); + } _ => {} } } } + +#[cfg(not(target_arch = "wasm32"))] +impl ConnexaTask +where + X: Default + Send + 'static, + C: Send, + C::ToSwarm: Debug, + S: Store, +{ + pub fn process_relay_server_command(&mut self, command: RelayServerCommand) { + let swarm = self.swarm.as_mut().expect("swarm is active"); + match command { + RelayServerCommand::StatusChanged { status, resp } => { + let Some(relay) = swarm.behaviour_mut().relay.as_mut() else { + let _ = resp.send(Err(std::io::Error::other( + "relay server protocol is not enabled", + ))); + return; + }; + + relay.set_status(status); + + let _ = resp.send(Ok(())); + } + } + } +} diff --git a/src/task/swarm.rs b/src/task/swarm.rs index f596b77..f75309b 100644 --- a/src/task/swarm.rs +++ b/src/task/swarm.rs @@ -291,6 +291,7 @@ where }; match event { + #[cfg(not(target_arch = "wasm32"))] #[cfg(feature = "relay")] BehaviourEvent::Relay(event) => self.process_relay_server_event(event), #[cfg(feature = "relay")] diff --git a/src/task/upnp.rs b/src/task/upnp.rs index 838c38f..5ed3835 100644 --- a/src/task/upnp.rs +++ b/src/task/upnp.rs @@ -13,11 +13,21 @@ where { pub fn process_upnp_event(&mut self, event: UpnpEvent) { match event { - UpnpEvent::NewExternalAddr(addr) => { - tracing::info!(?addr, "upnp external address discovered"); + UpnpEvent::NewExternalAddr { + external_addr, + local_addr, + } => { + tracing::info!( + ?external_addr, + ?local_addr, + "upnp external address discovered" + ); } - UpnpEvent::ExpiredExternalAddr(addr) => { - tracing::info!(?addr, "upnp external address expired"); + UpnpEvent::ExpiredExternalAddr { + external_addr, + local_addr, + } => { + tracing::info!(?external_addr, ?local_addr, "upnp external address expired"); } UpnpEvent::GatewayNotFound => { tracing::warn!("upnp gateway not found"); diff --git a/src/types.rs b/src/types.rs index d9d868f..a0d1be6 100644 --- a/src/types.rs +++ b/src/types.rs @@ -46,6 +46,9 @@ pub enum Command { Rendezvous(RendezvousCommand), #[cfg(feature = "autonat")] Autonat(AutonatCommand), + #[cfg(not(target_arch = "wasm32"))] + #[cfg(feature = "relay")] + RelayServer(RelayServerCommand), Whitelist(WhitelistCommand), Blacklist(BlacklistCommand), ConnectionLimits(ConnectionLimitsCommand), @@ -108,6 +111,14 @@ impl From for Command { } } +#[cfg(not(target_arch = "wasm32"))] +#[cfg(feature = "relay")] +impl From for Command { + fn from(cmd: RelayServerCommand) -> Self { + Command::RelayServer(cmd) + } +} + impl From for Command { fn from(cmd: WhitelistCommand) -> Self { Command::Whitelist(cmd) @@ -505,6 +516,16 @@ pub enum RendezvousCommand { }, } +#[cfg(not(target_arch = "wasm32"))] +#[cfg(feature = "relay")] +#[derive(Debug)] +pub enum RelayServerCommand { + StatusChanged { + status: Option, + resp: oneshot::Sender>, + }, +} + #[cfg(feature = "kad")] #[derive(Clone, Debug)] pub enum DHTEvent {