diff --git a/protocols/relay/CHANGELOG.md b/protocols/relay/CHANGELOG.md index fbee49cd3be..1d9b4a854ad 100644 --- a/protocols/relay/CHANGELOG.md +++ b/protocols/relay/CHANGELOG.md @@ -1,5 +1,8 @@ ## 0.22.0 +- Avoid panic when dst has connections without reservations. + See [PR XXXX](https://github.com/libp2p/rust-libp2p/pull/XXXX). + - Use `futures-timer` instead of tokio's timer for stream timeouts so the bounded `Delay` works on `wasm32`; tokio's timer has no driver in the browser and panics at runtime. See [PR 6488](https://github.com/libp2p/rust-libp2p/pull/6488). diff --git a/protocols/relay/src/behaviour.rs b/protocols/relay/src/behaviour.rs index dc129dcaa1a..145c53c640f 100644 --- a/protocols/relay/src/behaviour.rs +++ b/protocols/relay/src/behaviour.rs @@ -711,13 +711,11 @@ impl NetworkBehaviour for Behaviour { status: proto::Status::ResourceLimitExceeded, }), } - } else if let Some((dst_conn, status)) = self + } else if let Some((dst_conn, _)) = self .connections .get(&inbound_circuit_req.dst()) - .and_then(|cs| cs.iter().next()) + .and_then(|cs| cs.iter().find(|(_, status)| status.is_active())) { - assert_eq!(*status, Reservation::Active); - // Accept circuit request if reservation present. let circuit_id = self.circuits.insert(Circuit { status: CircuitStatus::Accepting, diff --git a/protocols/relay/tests/lib.rs b/protocols/relay/tests/lib.rs index 35e276727e1..2120e0b7ebc 100644 --- a/protocols/relay/tests/lib.rs +++ b/protocols/relay/tests/lib.rs @@ -475,6 +475,87 @@ async fn propagate_connect_error_to_unknown_peer_to_dialer() { )); } +#[tokio::test] +async fn deny_circuit_to_connected_peer_without_reservation() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let relay_addr = Multiaddr::empty().with(Protocol::Memory(rand::random::())); + let mut relay = build_relay(); + let relay_peer_id = *relay.local_peer_id(); + + relay.listen_on(relay_addr.clone()).unwrap(); + relay.add_external_address(relay_addr.clone()); + + let mut dst = build_client(); + let dst_peer_id = *dst.local_peer_id(); + dst.dial(relay_addr.clone()).unwrap(); + + let mut relay_saw_dst = false; + let mut dst_connected = false; + while !(relay_saw_dst && dst_connected) { + tokio::select! { + event = relay.select_next_some() => { + if let SwarmEvent::ConnectionEstablished { peer_id, .. } = event + && peer_id == dst_peer_id { + relay_saw_dst = true; + } + } + event = dst.select_next_some() => { + if let SwarmEvent::ConnectionEstablished { peer_id, .. } = event + && peer_id == relay_peer_id { + dst_connected = true; + } + } + } + } + + let mut src = build_client(); + let dst_addr = relay_addr + .with(Protocol::P2p(relay_peer_id)) + .with(Protocol::P2pCircuit) + .with(Protocol::P2p(dst_peer_id)); + + let opts = DialOpts::from(dst_addr.clone()); + let circuit_connection_id = opts.connection_id(); + + src.dial(opts).unwrap(); + + let (failed_address, error) = loop { + tokio::select! { + _ = relay.select_next_some() => {} + _ = dst.select_next_some() => {} + event = src.select_next_some() => { + if let SwarmEvent::OutgoingConnectionError { + connection_id, + error: DialError::Transport(mut errors), + .. + } = event + && connection_id == circuit_connection_id { + assert_eq!(errors.len(), 1); + break errors.remove(0); + } + } + } + }; + + // This is a bit wonky but we need to get the source error + let error = error + .source() + .unwrap() + .source() + .unwrap() + .downcast_ref::() + .unwrap(); + + assert_eq!(failed_address, dst_addr); + assert!(matches!( + error, + relay::outbound::hop::ConnectError::NoReservation + )); +} + #[tokio::test] async fn reuse_connection() { let _ = tracing_subscriber::fmt()