It works!

libp2p · Sep 12, 2023 · 1ce8d24 · 1ce8d24
1 parent 7daa046
commit 1ce8d24
Show file tree

Hide file tree

Showing 4 changed files with 93 additions and 86 deletions.
diff --git a/.dockerignore b/.dockerignore
@@ -1 +1,2 @@
 target
+**/Dockerfile**
diff --git a/hole-punching-tests/host/Dockerfile b/hole-punching-tests/host/Dockerfile
@@ -25,8 +25,8 @@ RUN --mount=type=cache,target=/var/cache/apt apt-get install -y tcpdump ncat ipu
 
 
 COPY ./hole-punching-tests/host/run.sh /scripts/
-COPY --from=builder /usr/local/bin/hp_client /usr/local/bin/hp_client
-COPY --from=builder /usr/local/bin/relay /usr/local/bin/relay
+COPY --from=builder /usr/local/bin/hp_client /usr/bin/hp_client
+COPY --from=builder /usr/local/bin/relay /usr/bin/relay
 
 RUN chmod +x /scripts/*.sh
 

diff --git a/hole-punching-tests/src/bin/hp_client.rs b/hole-punching-tests/src/bin/hp_client.rs
@@ -18,10 +18,9 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use clap::Parser;
 use futures::{future::Either, stream::StreamExt};
-use libp2p::swarm::dial_opts::DialOpts;
 use libp2p::{
     core::{
         multiaddr::{Multiaddr, Protocol},
@@ -31,12 +30,12 @@ use libp2p::{
     },
     dcutr, identify, identity, noise, ping, quic, relay,
     swarm::{NetworkBehaviour, SwarmBuilder, SwarmEvent},
-    tcp, yamux, PeerId,
+    tcp, yamux, PeerId, Swarm,
 };
 use log::{info, LevelFilter};
 use redis::AsyncCommands;
 use std::collections::HashMap;
-use std::pin::pin;
+use std::net::Ipv4Addr;
 use std::str::FromStr;
 
 #[derive(Debug, Parser)]
@@ -60,76 +59,9 @@ async fn main() -> Result<()> {
 
     let opts = Opts::parse();
 
-    let local_key = identity::Keypair::generate_ed25519();
-    let local_peer_id = PeerId::from(local_key.public());
-    info!("Local peer id: {local_peer_id}");
-
-    let (relay_transport, client) = relay::client::new(local_peer_id);
-
-    let transport = {
-        let relay_tcp_quic_transport = relay_transport
-            .or_transport(tcp::tokio::Transport::new(
-                tcp::Config::default().port_reuse(true),
-            ))
-            .upgrade(upgrade::Version::V1)
-            .authenticate(noise::Config::new(&local_key)?)
-            .multiplex(yamux::Config::default())
-            .or_transport(quic::tokio::Transport::new(quic::Config::new(&local_key)));
-
-        relay_tcp_quic_transport
-            .map(|either_output, _| match either_output {
-                Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
-                Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
-            })
-            .boxed()
-    };
-
-    #[derive(NetworkBehaviour)]
-    struct Behaviour {
-        relay_client: relay::client::Behaviour,
-        ping: ping::Behaviour,
-        identify: identify::Behaviour,
-        dcutr: dcutr::Behaviour,
-    }
-
-    let behaviour = Behaviour {
-        relay_client: client,
-        ping: ping::Behaviour::new(ping::Config::new()),
-        identify: identify::Behaviour::new(identify::Config::new(
-            "/TODO/0.0.1".to_string(),
-            local_key.public(),
-        )),
-        dcutr: dcutr::Behaviour::new(local_peer_id),
-    };
-
     let client = redis::Client::open("redis://redis:6379")?;
     let mut connection = client.get_async_connection().await?;
 
-    let mut swarm = SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build();
-
-    swarm.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse()?)?;
-    swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
-
-    // Wait to listen on all interfaces.
-    let mut delay = pin!(tokio::time::sleep(std::time::Duration::from_secs(1)));
-
-    loop {
-        tokio::select! {
-            Some(event) = swarm.next() => {
-                match event {
-                    SwarmEvent::NewListenAddr { address, .. } => {
-                        info!("Listening on {:?}", address);
-                    }
-                    event => panic!("{event:?}"),
-                }
-            }
-            _ = &mut delay => {
-                // Likely listening on all interfaces now, thus continuing by breaking the loop.
-                break;
-            }
-        }
-    }
-
     let redis_key = match opts.transport {
         TransportProtocol::Tcp => "RELAY_TCP_ADDRESS",
         TransportProtocol::Quic => "RELAY_QUIC_ADDRESS",
@@ -142,20 +74,51 @@ async fn main() -> Result<()> {
         .expect("key that we asked for to be present")
         .parse::<Multiaddr>()?;
 
-    // Connect to the relay server. Not for the reservation or relayed connection, but to (a) learn
-    // our local public address and (b) enable a freshly started relay to learn its public address.
+    let mut swarm = make_swarm()?;
+
+    // Both parties must have a listener for the hole-punch to work.
+    let listen_addr = match opts.transport {
+        TransportProtocol::Tcp => Multiaddr::empty()
+            .with(Protocol::Ip4(Ipv4Addr::UNSPECIFIED))
+            .with(Protocol::Tcp(0)),
+        TransportProtocol::Quic => Multiaddr::empty()
+            .with(Protocol::Ip4(Ipv4Addr::UNSPECIFIED))
+            .with(Protocol::Udp(0))
+            .with(Protocol::QuicV1),
+    };
+    let expected_listener_id = swarm
+        .listen_on(listen_addr)
+        .context("Failed to listen on address")?;
+    let mut listen_addresses = 0;
+
+    // We should have at least two listen addresses, one for localhost and the actual interface.
+    while listen_addresses < 2 {
+        if let SwarmEvent::NewListenAddr {
+            listener_id,
+            address,
+        } = swarm.next().await.unwrap()
+        {
+            if listener_id == expected_listener_id {
+                listen_addresses += 1;
+            }
 
-    let dial_opts = DialOpts::from(relay_address.clone());
-    let relay_connection_id = dial_opts.connection_id();
+            info!("Listening on {address}");
+        }
+    }
 
-    swarm.dial(dial_opts)?;
+    swarm.dial(relay_address.clone())?;
 
+    // Connect to the relay server. Not for the reservation or relayed connection, but to learn our local public address.
+    // FIXME: This should not be necessary. Perhaps dcutr should also already consider external address _candidates_?
     loop {
-        if let SwarmEvent::ConnectionEstablished { connection_id, .. } = swarm.next().await.unwrap()
+        if let SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received {
+            info: identify::Info { observed_addr, .. },
+            ..
+        })) = swarm.next().await.unwrap()
         {
-            if connection_id == relay_connection_id {
-                break;
-            }
+            info!("Relay told us our public address: {:?}", observed_addr);
+            swarm.add_external_address(observed_addr);
+            break;
         }
     }
 
@@ -183,9 +146,6 @@ async fn main() -> Result<()> {
 
     loop {
         match swarm.next().await.unwrap() {
-            SwarmEvent::NewListenAddr { address, .. } => {
-                info!("Listening on {:?}", address);
-            }
             SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                 relay::client::Event::ReservationReqAccepted { .. },
             )) => {
@@ -218,6 +178,44 @@ async fn main() -> Result<()> {
     }
 }
 
+fn make_swarm() -> Result<Swarm<Behaviour>> {
+    let local_key = identity::Keypair::generate_ed25519();
+    let local_peer_id = PeerId::from(local_key.public());
+    info!("Local peer id: {local_peer_id}");
+
+    let (relay_transport, client) = relay::client::new(local_peer_id);
+
+    let transport = {
+        let relay_tcp_quic_transport = relay_transport
+            .or_transport(tcp::tokio::Transport::new(
+                tcp::Config::default().port_reuse(true),
+            ))
+            .upgrade(upgrade::Version::V1)
+            .authenticate(noise::Config::new(&local_key)?)
+            .multiplex(yamux::Config::default())
+            .or_transport(quic::tokio::Transport::new(quic::Config::new(&local_key)));
+
+        relay_tcp_quic_transport
+            .map(|either_output, _| match either_output {
+                Either::Left((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+                Either::Right((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
+            })
+            .boxed()
+    };
+
+    let behaviour = Behaviour {
+        relay_client: client,
+        ping: ping::Behaviour::new(ping::Config::new()),
+        identify: identify::Behaviour::new(identify::Config::new(
+            "/TODO/0.0.1".to_string(),
+            local_key.public(),
+        )),
+        dcutr: dcutr::Behaviour::new(local_peer_id),
+    };
+
+    Ok(SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build())
+}
+
 #[derive(Clone, Debug, PartialEq, Parser)]
 enum Mode {
     Dial,
@@ -251,3 +249,11 @@ impl FromStr for TransportProtocol {
         }
     }
 }
+
+#[derive(NetworkBehaviour)]
+struct Behaviour {
+    relay_client: relay::client::Behaviour,
+    ping: ping::Behaviour,
+    identify: identify::Behaviour,
+    dcutr: dcutr::Behaviour,
+}
diff --git a/hole-punching-tests/src/bin/relay.rs b/hole-punching-tests/src/bin/relay.rs
@@ -102,7 +102,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 address,
                 listener_id,
             } => {
-                // swarm.add_external_address(address); // We know that in our testing network setup, that we are listening on a "publicly-reachable" address.
+                swarm.add_external_address(address.clone()); // We know that in our testing network setup, that we are listening on a "publicly-reachable" address.
 
                 info!("Listening on {address}");