lumina_node/p2p/
swarm.rs

1use libp2p::identity::Keypair;
2use libp2p::swarm::{NetworkBehaviour, Swarm};
3use std::time::Duration;
4
5use crate::p2p::{P2pError, Result};
6
7pub(crate) use self::imp::new_swarm;
8
9#[cfg(not(target_arch = "wasm32"))]
10mod imp {
11    use std::env;
12    use std::io::Cursor;
13    use std::path::Path;
14
15    use futures::future::Either;
16    use libp2p::core::muxing::StreamMuxerBox;
17    use libp2p::core::upgrade::Version;
18    use libp2p::{PeerId, Transport, dns, noise, quic, swarm, tcp, websocket, yamux};
19    use rustls_pki_types::{CertificateDer, PrivateKeyDer};
20    use tokio::fs;
21
22    use super::*;
23
24    pub(crate) async fn new_swarm<B>(keypair: Keypair, behaviour: B) -> Result<Swarm<B>>
25    where
26        B: NetworkBehaviour,
27    {
28        let tls_key = match env::var("LUMINA_TLS_KEY_FILE") {
29            Ok(path) => Some(read_tls_key(path).await?),
30            Err(_) => None,
31        };
32
33        let tls_certs = match env::var("LUMINA_TLS_CERT_FILE") {
34            Ok(path) => Some(read_tls_certs(path).await?),
35            Err(_) => None,
36        };
37
38        // We do not use system's DNS because libp2p caches system DNS
39        // servers when `Swarm` get constructed, and doesn't update them
40        // later. This can be a problem, if device roams between networks
41        // (and old DNS addresses may not be reachable from the new network).
42        //
43        // Similarly, if node is started when there's no Internet connection,
44        // it won't use the DNS servers offered when Internet connectivity
45        // is restored. Instead we per-define globally-accessible public DNS servers.
46        let dns_config = dns::ResolverConfig::cloudflare();
47
48        let noise_config =
49            noise::Config::new(&keypair).map_err(|e| P2pError::NoiseInit(e.to_string()))?;
50
51        let wss_transport = {
52            let config = if let (Some(key), Some(certs)) = (tls_key, tls_certs) {
53                let key = websocket::tls::PrivateKey::new(key.secret_der().to_vec());
54                let certs = certs
55                    .iter()
56                    .map(|cert| websocket::tls::Certificate::new(cert.to_vec()));
57
58                websocket::tls::Config::new(key, certs)
59                    .map_err(|e| P2pError::TlsInit(format!("server config: {e}")))?
60            } else {
61                websocket::tls::Config::client()
62            };
63
64            let mut wss_transport = websocket::Config::new(dns::tokio::Transport::custom(
65                tcp::tokio::Transport::new(tcp::Config::default()),
66                dns_config.clone(),
67                dns::ResolverOpts::default(),
68            ));
69
70            wss_transport.set_tls_config(config);
71
72            wss_transport
73                .upgrade(Version::V1Lazy)
74                .authenticate(noise_config.clone())
75                .multiplex(yamux::Config::default())
76        };
77
78        let tcp_transport = tcp::tokio::Transport::new(tcp::Config::default())
79            .upgrade(Version::V1Lazy)
80            .authenticate(noise_config)
81            .multiplex(yamux::Config::default());
82
83        let quic_transport = quic::tokio::Transport::new(quic::Config::new(&keypair));
84
85        // WSS must be before TCP transport and must not be wrapped in DNS transport.
86        let transport = wss_transport
87            .or_transport(dns::tokio::Transport::custom(
88                tcp_transport
89                    .or_transport(quic_transport)
90                    .map(|either, _| match either {
91                        Either::Left((peer_id, conn)) => (peer_id, StreamMuxerBox::new(conn)),
92                        Either::Right((peer_id, conn)) => (peer_id, StreamMuxerBox::new(conn)),
93                    }),
94                dns_config,
95                dns::ResolverOpts::default(),
96            ))
97            .map(|either, _| match either {
98                Either::Left((peer_id, conn)) => (peer_id, StreamMuxerBox::new(conn)),
99                Either::Right((peer_id, conn)) => (peer_id, StreamMuxerBox::new(conn)),
100            })
101            .boxed();
102
103        let local_peer_id = PeerId::from_public_key(&keypair.public());
104
105        Ok(Swarm::new(
106            transport,
107            behaviour,
108            local_peer_id,
109            swarm::Config::with_tokio_executor()
110                // This closes any connection that is idle for 15 seconds and not protected
111                // by `p2p::Worker::protect` (check p2p.rs for more info).
112                .with_idle_connection_timeout(Duration::from_secs(15)),
113        ))
114    }
115
116    impl From<noise::Error> for P2pError {
117        fn from(e: noise::Error) -> Self {
118            P2pError::NoiseInit(e.to_string())
119        }
120    }
121
122    async fn read_tls_key(path: impl AsRef<Path>) -> Result<PrivateKeyDer<'static>, P2pError> {
123        let path = path.as_ref();
124
125        // TODO: read key in a preallocated memory and zero it after use
126        let data = fs::read(&path)
127            .await
128            .map_err(|e| P2pError::TlsInit(format!("{}: {e}", path.display())))?;
129
130        let mut data = Cursor::new(data);
131
132        rustls_pemfile::private_key(&mut data)
133            .map_err(|e| P2pError::TlsInit(format!("{}: {e}", path.display())))?
134            .ok_or_else(|| P2pError::TlsInit(format!("{}: Key not found in file", path.display())))
135    }
136
137    async fn read_tls_certs(
138        path: impl AsRef<Path>,
139    ) -> Result<Vec<CertificateDer<'static>>, P2pError> {
140        let path = path.as_ref();
141
142        let data = fs::read(path)
143            .await
144            .map_err(|e| P2pError::TlsInit(format!("{}: {e}", path.display())))?;
145
146        let mut data = Cursor::new(data);
147        let certs = rustls_pemfile::certs(&mut data)
148            .collect::<Result<Vec<_>, std::io::Error>>()
149            .map_err(|e| P2pError::TlsInit(format!("{}: {e}", path.display())))?;
150
151        if certs.is_empty() {
152            let e = format!("{}: Certificate not found in file", path.display());
153            Err(P2pError::TlsInit(e))
154        } else {
155            Ok(certs)
156        }
157    }
158}
159
160#[cfg(target_arch = "wasm32")]
161mod imp {
162    use super::*;
163    use libp2p::core::upgrade::Version;
164    use libp2p::{SwarmBuilder, Transport, noise, websocket_websys, webtransport_websys, yamux};
165
166    pub(crate) async fn new_swarm<B>(keypair: Keypair, behaviour: B) -> Result<Swarm<B>>
167    where
168        B: NetworkBehaviour,
169    {
170        let noise_config =
171            noise::Config::new(&keypair).map_err(|e| P2pError::NoiseInit(e.to_string()))?;
172
173        Ok(SwarmBuilder::with_existing_identity(keypair)
174            .with_wasm_bindgen()
175            .with_other_transport(move |_| {
176                Ok(websocket_websys::Transport::default()
177                    .upgrade(Version::V1Lazy)
178                    .authenticate(noise_config)
179                    .multiplex(yamux::Config::default()))
180            })
181            .expect("websocket_websys::Transport is infallible")
182            .with_other_transport(|local_keypair| {
183                let config = webtransport_websys::Config::new(local_keypair);
184                webtransport_websys::Transport::new(config)
185            })
186            .expect("webtransport_websys::Transport is infallible")
187            .with_behaviour(|_| behaviour)
188            .expect("Moving behaviour doesn't fail")
189            .with_swarm_config(|config| {
190                // This closes any connection that is idle for 15 seconds and not protected
191                // by `p2p::Worker::protect` (check p2p.rs for more info).
192                config.with_idle_connection_timeout(Duration::from_secs(15))
193            })
194            .build())
195    }
196}
lumina_node/p2p/swarm.rs

lumina_node/p2p/
swarm.rs
