*: Rework reporting of invalid and wrong PeerIds (#2441)

Previously, the negotiated PeerId was included in the swarm event and
inject_dial_failure’s arguments while the expected one was absent. This
patch adds the negotiated PeerId to the DialError and includes the expected
one in the notifications.

Co-authored-by: Roland Kuhn <rk@rkuhn.info>

core/CHANGELOG.md

@@ -14,14 +14,17 @@
 - Enable overriding _dial concurrency factor_ per dial via
   `DialOpts::override_dial_concurrency_factor`.
 
-    - Introduces `libp2p_core::DialOpts` mirroring `libp2p_swarm::DialOpts`.
+  - Introduces `libp2p_core::DialOpts` mirroring `libp2p_swarm::DialOpts`.
       Passed as an argument to `Network::dial`.
-    - Removes `Peer::dial` in favor of `Network::dial`.
+  - Removes `Peer::dial` in favor of `Network::dial`.
 
   See [PR 2404].
 
 - Implement `Serialize` and `Deserialize` for `PeerId` (see [PR 2408])
 
+- Report negotiated and expected `PeerId` as well as remote address in
+  `DialError::WrongPeerId` (see [PR 2428]).
+
 - Allow overriding role when dialing. This option is needed for NAT and firewall
   hole punching.
 
@@ -39,6 +42,7 @@
 [PR 2392]: https://github.com/libp2p/rust-libp2p/pull/2392
 [PR 2404]: https://github.com/libp2p/rust-libp2p/pull/2404
 [PR 2408]: https://github.com/libp2p/rust-libp2p/pull/2408
+[PR 2428]: https://github.com/libp2p/rust-libp2p/pull/2428
 [PR 2363]: https://github.com/libp2p/rust-libp2p/pull/2363
 
 # 0.30.1 [2021-11-16]
@@ -127,7 +131,7 @@
 
 # 0.28.3 [2021-04-26]
 
-- Fix build with secp256k1 disabled [PR 2057](https://github.com/libp2p/rust-libp2p/pull/2057].
+- Fix build with secp256k1 disabled [PR 2057](https://github.com/libp2p/rust-libp2p/pull/2057).
 
 # 0.28.2 [2021-04-13]
 

core/src/connection/error.rs

@@ -18,9 +18,9 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::connection::ConnectionLimit;
 use crate::transport::TransportError;
 use crate::Multiaddr;
+use crate::{connection::ConnectionLimit, ConnectedPoint, PeerId};
 use std::{fmt, io};
 
 /// Errors that can occur in the context of an established `Connection`.
@@ -84,14 +84,33 @@ pub enum PendingConnectionError<TTransErr> {
     Aborted,
 
     /// The peer identity obtained on the connection did not
-    /// match the one that was expected or is otherwise invalid.
+    /// match the one that was expected or is the local one.
-    InvalidPeerId,
+    WrongPeerId {
+        obtained: PeerId,
+        endpoint: ConnectedPoint,
+    },
 
     /// An I/O error occurred on the connection.
     // TODO: Eventually this should also be a custom error?
     IO(io::Error),
 }
 
+impl<T> PendingConnectionError<T> {
+    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> PendingConnectionError<U> {
+        match self {
+            PendingConnectionError::Transport(t) => PendingConnectionError::Transport(f(t)),
+            PendingConnectionError::ConnectionLimit(l) => {
+                PendingConnectionError::ConnectionLimit(l)
+            }
+            PendingConnectionError::Aborted => PendingConnectionError::Aborted,
+            PendingConnectionError::WrongPeerId { obtained, endpoint } => {
+                PendingConnectionError::WrongPeerId { obtained, endpoint }
+            }
+            PendingConnectionError::IO(e) => PendingConnectionError::IO(e),
+        }
+    }
+}
+
 impl<TTransErr> fmt::Display for PendingConnectionError<TTransErr>
 where
     TTransErr: fmt::Display + fmt::Debug,
@@ -110,8 +129,12 @@ where
             PendingConnectionError::ConnectionLimit(l) => {
                 write!(f, "Connection error: Connection limit: {}.", l)
             }
-            PendingConnectionError::InvalidPeerId => {
+            PendingConnectionError::WrongPeerId { obtained, endpoint } => {
-                write!(f, "Pending connection: Invalid peer ID.")
+                write!(
+                    f,
+                    "Pending connection: Unexpected peer ID {} at {:?}.",
+                    obtained, endpoint
+                )
             }
         }
     }
@@ -125,7 +148,7 @@ where
         match self {
             PendingConnectionError::IO(err) => Some(err),
             PendingConnectionError::Transport(_) => None,
-            PendingConnectionError::InvalidPeerId => None,
+            PendingConnectionError::WrongPeerId { .. } => None,
             PendingConnectionError::Aborted => None,
             PendingConnectionError::ConnectionLimit(..) => None,
         }

core/src/connection/pool.rs

@@ -733,7 +733,7 @@ where
             match event {
                 task::PendingConnectionEvent::ConnectionEstablished {
                     id,
-                    output: (peer_id, muxer),
+                    output: (obtained_peer_id, muxer),
                     outgoing,
                 } => {
                     let PendingConnectionInfo {
@@ -777,49 +777,42 @@ where
                         ),
                     };
 
-                    enum Error {
+                    let error: Result<(), PendingInboundConnectionError<_>> = self
-                        ConnectionLimit(ConnectionLimit),
-                        InvalidPeerId,
-                    }
-
-                    impl<TransportError> From<Error> for PendingConnectionError<TransportError> {
-                        fn from(error: Error) -> Self {
-                            match error {
-                                Error::ConnectionLimit(limit) => {
-                                    PendingConnectionError::ConnectionLimit(limit)
-                                }
-                                Error::InvalidPeerId => PendingConnectionError::InvalidPeerId,
-                            }
-                        }
-                    }
-
-                    let error = self
                         .counters
                         // Check general established connection limit.
                         .check_max_established(&endpoint)
-                        .map_err(Error::ConnectionLimit)
+                        .map_err(PendingConnectionError::ConnectionLimit)
                         // Check per-peer established connection limit.
                         .and_then(|()| {
                             self.counters
                                 .check_max_established_per_peer(num_peer_established(
                                     &self.established,
-                                    peer_id,
+                                    obtained_peer_id,
                                 ))
-                                .map_err(Error::ConnectionLimit)
+                                .map_err(PendingConnectionError::ConnectionLimit)
                         })
                         // Check expected peer id matches.
                         .and_then(|()| {
                             if let Some(peer) = expected_peer_id {
-                                if peer != peer_id {
+                                if peer != obtained_peer_id {
-                                    return Err(Error::InvalidPeerId);
+                                    Err(PendingConnectionError::WrongPeerId {
+                                        obtained: obtained_peer_id,
+                                        endpoint: endpoint.clone(),
+                                    })
+                                } else {
+                                    Ok(())
                                 }
+                            } else {
+                                Ok(())
                             }
-                            Ok(())
                         })
                         // Check peer is not local peer.
                         .and_then(|()| {
-                            if self.local_id == peer_id {
+                            if self.local_id == obtained_peer_id {
-                                Err(Error::InvalidPeerId)
+                                Err(PendingConnectionError::WrongPeerId {
+                                    obtained: obtained_peer_id,
+                                    endpoint: endpoint.clone(),
+                                })
                             } else {
                                 Ok(())
                             }
@@ -832,7 +825,7 @@ where
                                     log::debug!(
                                         "Failed to close connection {:?} to peer {}: {:?}",
                                         id,
-                                        peer_id,
+                                        obtained_peer_id,
                                         e
                                     );
                                 }
@@ -845,9 +838,10 @@ where
                             ConnectedPoint::Dialer { .. } => {
                                 return Poll::Ready(PoolEvent::PendingOutboundConnectionError {
                                     id,
-                                    error: error.into(),
+                                    error: error
+                                        .map(|t| vec![(endpoint.get_remote_address().clone(), t)]),
                                     handler,
-                                    peer: Some(peer_id),
+                                    peer: expected_peer_id.or(Some(obtained_peer_id)),
                                 })
                             }
                             ConnectedPoint::Listener {
@@ -856,7 +850,7 @@ where
                             } => {
                                 return Poll::Ready(PoolEvent::PendingInboundConnectionError {
                                     id,
-                                    error: error.into(),
+                                    error,
                                     handler,
                                     send_back_addr,
                                     local_addr,
@@ -866,7 +860,7 @@ where
                     }
 
                     // Add the connection to the pool.
-                    let conns = self.established.entry(peer_id).or_default();
+                    let conns = self.established.entry(obtained_peer_id).or_default();
                     let other_established_connection_ids = conns.keys().cloned().collect();
                     self.counters.inc_established(&endpoint);
 
@@ -875,20 +869,23 @@ where
                     conns.insert(
                         id,
                         EstablishedConnectionInfo {
-                            peer_id,
+                            peer_id: obtained_peer_id,
                             endpoint: endpoint.clone(),
                             sender: command_sender,
                         },
                     );
 
-                    let connected = Connected { peer_id, endpoint };
+                    let connected = Connected {
+                        peer_id: obtained_peer_id,
+                        endpoint,
+                    };
 
                     let connection =
                         super::Connection::new(muxer, handler.into_handler(&connected));
                     self.spawn(
                         task::new_for_established_connection(
                             id,
-                            peer_id,
+                            obtained_peer_id,
                             connection,
                             command_receiver,
                             self.established_connection_events_tx.clone(),

core/src/network.rs

@@ -37,6 +37,7 @@ use crate::{
     Executor, Multiaddr, PeerId,
 };
 use either::Either;
+use multihash::Multihash;
 use std::{
     convert::TryFrom as _,
     error, fmt,
@@ -241,7 +242,7 @@ where
                     .transpose()
                 {
                     Ok(peer_id) => peer_id,
-                    Err(_) => return Err(DialError::InvalidPeerId { handler }),
+                    Err(multihash) => return Err(DialError::InvalidPeerId { handler, multihash }),
                 };
 
                 (
@@ -576,11 +577,10 @@ pub enum DialError<THandler> {
         handler: THandler,
     },
     /// The dialing attempt is rejected because the peer being dialed is the local peer.
-    LocalPeerId {
+    LocalPeerId { handler: THandler },
-        handler: THandler,
-    },
     InvalidPeerId {
         handler: THandler,
+        multihash: Multihash,
     },
 }
 

core/tests/network_dial_error.rs

@@ -27,7 +27,7 @@ use libp2p_core::{
     connection::PendingConnectionError,
     multiaddr::Protocol,
     network::{NetworkConfig, NetworkEvent},
-    PeerId,
+    ConnectedPoint, Endpoint, PeerId,
 };
 use rand::seq::SliceRandom;
 use std::{io, task::Poll};
@@ -42,7 +42,7 @@ fn deny_incoming_connec() {
 
     swarm1.listen_on("/memory/0".parse().unwrap()).unwrap();
 
-    let address = async_std::task::block_on(future::poll_fn(|cx| match swarm1.poll(cx) {
+    let address = futures::executor::block_on(future::poll_fn(|cx| match swarm1.poll(cx) {
         Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) => {
             Poll::Ready(listen_addr)
         }
@@ -59,7 +59,7 @@ fn deny_incoming_connec() {
         )
         .unwrap();
 
-    async_std::task::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
+    futures::executor::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
         match swarm1.poll(cx) {
             Poll::Ready(NetworkEvent::IncomingConnection { connection, .. }) => drop(connection),
             Poll::Ready(_) => unreachable!(),
@@ -88,6 +88,58 @@ fn deny_incoming_connec() {
     .unwrap();
 }
 
+#[test]
+fn invalid_peer_id() {
+    // Checks whether dialing an address containing the wrong peer id raises an error
+    // for the expected peer id instead of the obtained peer id.
+
+    let mut swarm1 = test_network(NetworkConfig::default());
+    let mut swarm2 = test_network(NetworkConfig::default());
+
+    swarm1.listen_on("/memory/0".parse().unwrap()).unwrap();
+
+    let address = futures::executor::block_on(future::poll_fn(|cx| match swarm1.poll(cx) {
+        Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) => {
+            Poll::Ready(listen_addr)
+        }
+        Poll::Pending => Poll::Pending,
+        _ => panic!("Was expecting the listen address to be reported"),
+    }));
+
+    let other_id = PeerId::random();
+    let other_addr = address.with(Protocol::P2p(other_id.into()));
+
+    swarm2.dial(TestHandler(), other_addr.clone()).unwrap();
+
+    let (peer_id, error) = futures::executor::block_on(future::poll_fn(|cx| {
+        if let Poll::Ready(NetworkEvent::IncomingConnection { connection, .. }) = swarm1.poll(cx) {
+            swarm1.accept(connection, TestHandler()).unwrap();
+        }
+
+        match swarm2.poll(cx) {
+            Poll::Ready(NetworkEvent::DialError { peer_id, error, .. }) => {
+                Poll::Ready((peer_id, error))
+            }
+            Poll::Ready(x) => panic!("unexpected {:?}", x),
+            Poll::Pending => Poll::Pending,
+        }
+    }));
+    assert_eq!(peer_id, other_id);
+    match error {
+        PendingConnectionError::WrongPeerId { obtained, endpoint } => {
+            assert_eq!(obtained, *swarm1.local_peer_id());
+            assert_eq!(
+                endpoint,
+                ConnectedPoint::Dialer {
+                    address: other_addr,
+                    role_override: Endpoint::Dialer,
+                }
+            );
+        }
+        x => panic!("wrong error {:?}", x),
+    }
+}
+
 #[test]
 fn dial_self() {
     // Check whether dialing ourselves correctly fails.
@@ -103,7 +155,7 @@ fn dial_self() {
     let mut swarm = test_network(NetworkConfig::default());
     swarm.listen_on("/memory/0".parse().unwrap()).unwrap();
 
-    let local_address = async_std::task::block_on(future::poll_fn(|cx| match swarm.poll(cx) {
+    let local_address = futures::executor::block_on(future::poll_fn(|cx| match swarm.poll(cx) {
         Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) => {
             Poll::Ready(listen_addr)
         }
@@ -115,12 +167,12 @@ fn dial_self() {
 
     let mut got_dial_err = false;
     let mut got_inc_err = false;
-    async_std::task::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
+    futures::executor::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
         loop {
             match swarm.poll(cx) {
                 Poll::Ready(NetworkEvent::DialError {
                     peer_id,
-                    error: PendingConnectionError::InvalidPeerId { .. },
+                    error: PendingConnectionError::WrongPeerId { .. },
                     ..
                 }) => {
                     assert_eq!(&peer_id, swarm.local_peer_id());
@@ -157,7 +209,7 @@ fn dial_self_by_id() {
     // Trying to dial self by passing the same `PeerId` shouldn't even be possible in the first
     // place.
     let mut swarm = test_network(NetworkConfig::default());
-    let peer_id = swarm.local_peer_id().clone();
+    let peer_id = *swarm.local_peer_id();
     assert!(swarm.peer(peer_id).into_disconnected().is_none());
 }
 
@@ -181,13 +233,13 @@ fn multiple_addresses_err() {
     swarm
         .dial(
             TestHandler(),
-            DialOpts::peer_id(target.clone())
+            DialOpts::peer_id(target)
                 .addresses(addresses.clone())
                 .build(),
         )
         .unwrap();
 
-    async_std::task::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
+    futures::executor::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
         loop {
             match swarm.poll(cx) {
                 Poll::Ready(NetworkEvent::DialError {

misc/metrics/src/swarm.rs

@@ -216,9 +216,12 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
                     libp2p_swarm::DialError::Aborted => {
                         record(OutgoingConnectionErrorError::Aborted)
                     }
-                    libp2p_swarm::DialError::InvalidPeerId => {
+                    libp2p_swarm::DialError::InvalidPeerId { .. } => {
                         record(OutgoingConnectionErrorError::InvalidPeerId)
                     }
+                    libp2p_swarm::DialError::WrongPeerId { .. } => {
+                        record(OutgoingConnectionErrorError::WrongPeerId)
+                    }
                     libp2p_swarm::DialError::ConnectionIo(_) => {
                         record(OutgoingConnectionErrorError::ConnectionIo)
                     }
@@ -292,6 +295,7 @@ enum OutgoingConnectionErrorError {
     DialPeerConditionFalse,
     Aborted,
     InvalidPeerId,
+    WrongPeerId,
     ConnectionIo,
     TransportMultiaddrNotSupported,
     TransportOther,
@@ -304,7 +308,7 @@ struct IncomingConnectionErrorLabels {
 
 #[derive(Encode, Hash, Clone, Eq, PartialEq)]
 enum PendingInboundConnectionError {
-    InvalidPeerId,
+    WrongPeerId,
     TransportErrorMultiaddrNotSupported,
     TransportErrorOther,
     Aborted,
@@ -317,8 +321,8 @@ impl<TTransErr> From<&libp2p_core::connection::PendingInboundConnectionError<TTr
 {
     fn from(error: &libp2p_core::connection::PendingInboundConnectionError<TTransErr>) -> Self {
         match error {
-            libp2p_core::connection::PendingInboundConnectionError::InvalidPeerId => {
+            libp2p_core::connection::PendingInboundConnectionError::WrongPeerId { .. } => {
-                PendingInboundConnectionError::InvalidPeerId
+                PendingInboundConnectionError::WrongPeerId
             }
             libp2p_core::connection::PendingInboundConnectionError::ConnectionLimit(_) => {
                 PendingInboundConnectionError::ConnectionLimit

protocols/kad/src/behaviour.rs

@@ -1919,7 +1919,8 @@ where
             DialError::Banned
             | DialError::ConnectionLimit(_)
             | DialError::LocalPeerId
-            | DialError::InvalidPeerId
+            | DialError::InvalidPeerId { .. }
+            | DialError::WrongPeerId { .. }
             | DialError::Aborted
             | DialError::ConnectionIo(_)
             | DialError::Transport(_)

swarm/CHANGELOG.md

@@ -17,6 +17,9 @@
 - Allow overriding _dial concurrency factor_ per dial via
   `DialOpts::override_dial_concurrency_factor`. See [PR 2404].
 
+- Report negotiated and expected `PeerId` as well as remote address in
+  `DialError::WrongPeerId` (see [PR 2428]).
+
 - Allow overriding role when dialing through `override_role` option on
   `DialOpts`. This option is needed for NAT and firewall hole punching. See [PR
   2363].
@@ -28,6 +31,7 @@
 [PR 2375]: https://github.com/libp2p/rust-libp2p/pull/2375
 [PR 2378]: https://github.com/libp2p/rust-libp2p/pull/2378
 [PR 2404]: https://github.com/libp2p/rust-libp2p/pull/2404
+[PR 2428]: https://github.com/libp2p/rust-libp2p/pull/2428
 [PR 2363]: https://github.com/libp2p/rust-libp2p/pull/2363
 
 # 0.32.0 [2021-11-16]
@@ -41,16 +45,17 @@
 
    Changes required to maintain status quo:
 
-   - Previously `swarm.dial(peer_id)`
+  - Previously `swarm.dial(peer_id)`
      now `swarm.dial(DialOpts::peer_id(peer_id).build())`
      or `swarm.dial(peer_id)` given that `DialOpts` implements `From<PeerId>`.
 
-   - Previously `swarm.dial_addr(addr)`
+  - Previously `swarm.dial_addr(addr)`
      now `swarm.dial(DialOpts::unknown_peer_id().address(addr).build())`
      or `swarm.dial(addr)` given that `DialOpts` implements `From<Multiaddr>`.
 
-   - Previously `NetworkBehaviourAction::DialPeer { peer_id, condition, handler }`
+  - Previously `NetworkBehaviourAction::DialPeer { peer_id, condition, handler }`
      now
+
      ```rust
      NetworkBehaviourAction::Dial {
        opts: DialOpts::peer_id(peer_id)
@@ -60,8 +65,9 @@
      }
      ```
 
-   - Previously `NetworkBehaviourAction::DialAddress { address, handler }`
+  - Previously `NetworkBehaviourAction::DialAddress { address, handler }`
      now
+
      ```rust
      NetworkBehaviourAction::Dial {
        opts: DialOpts::unknown_peer_id()

swarm/src/lib.rs

@@ -81,6 +81,7 @@ use libp2p_core::{
         EstablishedConnection, IntoConnectionHandler, ListenerId, PendingConnectionError,
         PendingInboundConnectionError, PendingOutboundConnectionError, Substream,
     },
+    multihash::Multihash,
     muxing::StreamMuxerBox,
     network::{
         self, peer::ConnectedPeer, ConnectionLimits, Network, NetworkConfig, NetworkEvent,
@@ -1332,10 +1333,13 @@ pub enum DialError {
     DialPeerConditionFalse(dial_opts::PeerCondition),
     /// Pending connection attempt has been aborted.
     Aborted,
-    /// The provided peer identity is invalid or the peer identity obtained on
+    /// The provided peer identity is invalid.
-    /// the connection did not match the one that was expected or is otherwise
+    InvalidPeerId(Multihash),
-    /// invalid.
+    /// The peer identity obtained on the connection did not match the one that was expected.
-    InvalidPeerId,
+    WrongPeerId {
+        obtained: PeerId,
+        endpoint: ConnectedPoint,
+    },
     /// An I/O error occurred on the connection.
     ConnectionIo(io::Error),
     /// An error occurred while negotiating the transport protocol(s) on a connection.
@@ -1349,7 +1353,9 @@ impl DialError {
                 (DialError::ConnectionLimit(limit), handler)
             }
             network::DialError::LocalPeerId { handler } => (DialError::LocalPeerId, handler),
-            network::DialError::InvalidPeerId { handler } => (DialError::InvalidPeerId, handler),
+            network::DialError::InvalidPeerId { handler, multihash } => {
+                (DialError::InvalidPeerId(multihash), handler)
+            }
         }
     }
 }
@@ -1359,7 +1365,9 @@ impl From<PendingOutboundConnectionError<io::Error>> for DialError {
         match error {
             PendingConnectionError::ConnectionLimit(limit) => DialError::ConnectionLimit(limit),
             PendingConnectionError::Aborted => DialError::Aborted,
-            PendingConnectionError::InvalidPeerId => DialError::InvalidPeerId,
+            PendingConnectionError::WrongPeerId { obtained, endpoint } => {
+                DialError::WrongPeerId { obtained, endpoint }
+            }
             PendingConnectionError::IO(e) => DialError::ConnectionIo(e),
             PendingConnectionError::Transport(e) => DialError::Transport(e),
         }
@@ -1384,7 +1392,8 @@ impl fmt::Display for DialError {
                 f,
                 "Dial error: Pending connection attempt has been aborted."
             ),
-            DialError::InvalidPeerId => write!(f, "Dial error: Invalid peer ID."),
+            DialError::InvalidPeerId(multihash) => write!(f, "Dial error: multihash {:?} is not a PeerId", multihash),
+            DialError::WrongPeerId { obtained, endpoint} => write!(f, "Dial error: Unexpected peer ID {} at {:?}.", obtained, endpoint),
             DialError::ConnectionIo(e) => write!(
                 f,
                 "Dial error: An I/O error occurred on the connection: {:?}.", e
@@ -1403,7 +1412,8 @@ impl error::Error for DialError {
             DialError::Banned => None,
             DialError::DialPeerConditionFalse(_) => None,
             DialError::Aborted => None,
-            DialError::InvalidPeerId => None,
+            DialError::InvalidPeerId { .. } => None,
+            DialError::WrongPeerId { .. } => None,
             DialError::ConnectionIo(_) => None,
             DialError::Transport(_) => None,
         }