Skip to content

feat(swarm): close connections on multistream-select protocol violations #3482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thomaseizinger wants to merge 1 commit into from
Closed

feat(swarm): close connections on multistream-select protocol violations #3482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 10 additions & 16 deletions protocols/dcutr/src/handler/relayed.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ use futures::future;
use futures::future::{BoxFuture, FutureExt};
use instant::Instant;
use libp2p_core::multiaddr::Multiaddr;
use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
use libp2p_core::upgrade::DeniedUpgrade;
use libp2p_core::ConnectedPoint;
use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
Expand Down Expand Up @@ -229,27 +229,23 @@ impl Handler {
},
));
}
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
ConnectionHandlerUpgrErr::NegotiationFailed => {
// The remote merely doesn't support the DCUtR protocol.
// This is no reason to close the connection, which may
// successfully communicate with other protocols already.
self.keep_alive = KeepAlive::No;
self.queued_events.push_back(ConnectionHandlerEvent::Custom(
Event::InboundNegotiationFailed {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
NegotiationError::Failed,
)),
error: ConnectionHandlerUpgrErr::NegotiationFailed,
},
));
}
_ => {
// Anything else is considered a fatal error or misbehaviour of
// the remote peer and results in closing the connection.
self.pending_error = Some(error.map_upgrade_err(|e| {
e.map_err(|e| match e {
Either::Left(e) => Either::Left(e),
Either::Right(v) => void::unreachable(v),
})
self.pending_error = Some(error.map_upgrade_err(|e| match e {
Either::Left(e) => Either::Left(e),
Either::Right(v) => void::unreachable(v),
}));
}
}
Expand All @@ -272,22 +268,20 @@ impl Handler {
},
));
}
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
ConnectionHandlerUpgrErr::NegotiationFailed => {
// The remote merely doesn't support the DCUtR protocol.
// This is no reason to close the connection, which may
// successfully communicate with other protocols already.
self.queued_events.push_back(ConnectionHandlerEvent::Custom(
Event::OutboundNegotiationFailed {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
NegotiationError::Failed,
)),
error: ConnectionHandlerUpgrErr::NegotiationFailed,
},
));
}
_ => {
// Anything else is considered a fatal error or misbehaviour of
// the remote peer and results in closing the connection.
self.pending_error = Some(error.map_upgrade_err(|e| e.map_err(Either::Right)));
self.pending_error = Some(error.map_upgrade_err(Either::Right));
}
}
}
Expand Down Expand Up @@ -388,7 +382,7 @@ impl ConnectionHandler for Handler {
}
Err(e) => {
return Poll::Ready(ConnectionHandlerEvent::Close(
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(e))),
ConnectionHandlerUpgrErr::Upgrade(Either::Left(e)),
))
}
}
Expand Down
38 changes: 15 additions & 23 deletions protocols/gossipsub/src/handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ use asynchronous_codec::Framed;
use futures::prelude::*;
use futures::StreamExt;
use instant::Instant;
use libp2p_core::upgrade::{NegotiationError, UpgradeError};
use libp2p_swarm::handler::{
ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr,
DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, KeepAlive,
Expand Down Expand Up @@ -299,28 +298,21 @@ impl ConnectionHandler for Handler {
Some(HandlerError::NegotiationTimeout)
}
// There was an error post negotiation, close the connection.
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)) => Some(e),
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(negotiation_error)) => {
match negotiation_error {
NegotiationError::Failed => {
// The protocol is not supported
self.protocol_unsupported = true;
if !self.peer_kind_sent {
self.peer_kind_sent = true;
// clear all substreams so the keep alive returns false
self.inbound_substream = None;
self.outbound_substream = None;
self.keep_alive = KeepAlive::No;
return Poll::Ready(ConnectionHandlerEvent::Custom(
HandlerEvent::PeerKind(PeerKind::NotSupported),
));
} else {
None
}
}
NegotiationError::ProtocolError(e) => {
Some(HandlerError::NegotiationProtocolError(e))
}
ConnectionHandlerUpgrErr::Upgrade(e) => Some(e),
ConnectionHandlerUpgrErr::NegotiationFailed => {
// The protocol is not supported
self.protocol_unsupported = true;
if !self.peer_kind_sent {
self.peer_kind_sent = true;
// clear all substreams so the keep alive returns false
self.inbound_substream = None;
self.outbound_substream = None;
self.keep_alive = KeepAlive::No;
return Poll::Ready(ConnectionHandlerEvent::Custom(
HandlerEvent::PeerKind(PeerKind::NotSupported),
));
} else {
None
}
}
};
Expand Down
11 changes: 3 additions & 8 deletions protocols/identify/src/handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,12 +254,9 @@ impl Handler {
<Self as ConnectionHandler>::OutboundProtocol,
>,
) {
use libp2p_core::upgrade::UpgradeError;

let err = err.map_upgrade_err(|e| match e {
UpgradeError::Select(e) => UpgradeError::Select(e),
UpgradeError::Apply(Either::Left(ioe)) => UpgradeError::Apply(ioe),
UpgradeError::Apply(Either::Right(ioe)) => UpgradeError::Apply(ioe),
Either::Left(ioe) => ioe,
Either::Right(ioe) => ioe,
});
self.events
.push(ConnectionHandlerEvent::Custom(Event::IdentificationError(
Expand Down Expand Up @@ -366,9 +363,7 @@ impl ConnectionHandler for Handler {
Event::Identification(peer_id),
)),
Poll::Ready(Some(Err(err))) => Poll::Ready(ConnectionHandlerEvent::Custom(
Event::IdentificationError(ConnectionHandlerUpgrErr::Upgrade(
libp2p_core::upgrade::UpgradeError::Apply(err),
)),
Event::IdentificationError(ConnectionHandlerUpgrErr::Upgrade(err)),
)),
Poll::Ready(None) | Poll::Pending => Poll::Pending,
}
Expand Down
3 changes: 1 addition & 2 deletions protocols/ping/src/handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ use futures::future::BoxFuture;
use futures::prelude::*;
use futures_timer::Delay;
use libp2p_core::upgrade::ReadyUpgrade;
use libp2p_core::{upgrade::NegotiationError, UpgradeError};
use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
};
Expand Down Expand Up @@ -238,7 +237,7 @@ impl Handler {
self.outbound = None; // Request a new substream on the next `poll`.

let error = match error {
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
ConnectionHandlerUpgrErr::NegotiationFailed => {
debug_assert_eq!(self.state, State::Active);

self.state = State::Inactive { reported: false };
Expand Down
52 changes: 10 additions & 42 deletions protocols/relay/src/behaviour/handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ use futures::io::AsyncWriteExt;
use futures::stream::{FuturesUnordered, StreamExt};
use futures_timer::Delay;
use instant::Instant;
use libp2p_core::{upgrade, ConnectedPoint, Multiaddr, PeerId};
use libp2p_core::{ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
ListenUpgradeError, SendWrapper,
Expand Down Expand Up @@ -506,25 +506,9 @@ impl Handler {
let non_fatal_error = match error {
ConnectionHandlerUpgrErr::Timeout => ConnectionHandlerUpgrErr::Timeout,
ConnectionHandlerUpgrErr::Timer => ConnectionHandlerUpgrErr::Timer,
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Select(
upgrade::NegotiationError::Failed,
)) => ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Select(
upgrade::NegotiationError::Failed,
)),
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Select(
upgrade::NegotiationError::ProtocolError(e),
)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Select(upgrade::NegotiationError::ProtocolError(e)),
));
return;
}
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Apply(
inbound_hop::UpgradeError::Fatal(error),
)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(Either::Left(error)),
));
ConnectionHandlerUpgrErr::NegotiationFailed => ConnectionHandlerUpgrErr::NegotiationFailed,
ConnectionHandlerUpgrErr::Upgrade(inbound_hop::UpgradeError::Fatal(error)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(Either::Left(error)));
return;
}
};
Expand Down Expand Up @@ -553,29 +537,16 @@ impl Handler {
ConnectionHandlerUpgrErr::Timer => {
(ConnectionHandlerUpgrErr::Timer, Status::ConnectionFailed)
}
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Select(
upgrade::NegotiationError::Failed,
)) => {
ConnectionHandlerUpgrErr::NegotiationFailed => {
// The remote has previously done a reservation. Doing a reservation but not
// supporting the stop protocol is pointless, thus disconnecting.
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Select(upgrade::NegotiationError::Failed),
));
self.pending_error = Some(ConnectionHandlerUpgrErr::NegotiationFailed);
return;
}
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Select(
upgrade::NegotiationError::ProtocolError(e),
)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Select(upgrade::NegotiationError::ProtocolError(e)),
));
return;
}
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Apply(error)) => match error {
ConnectionHandlerUpgrErr::Upgrade(error) => match error {
outbound_stop::UpgradeError::Fatal(error) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(Either::Right(error)),
));
self.pending_error =
Some(ConnectionHandlerUpgrErr::Upgrade(Either::Right(error)));
return;
}
outbound_stop::UpgradeError::CircuitFailed(error) => {
Expand All @@ -587,10 +558,7 @@ impl Handler {
Status::PermissionDenied
}
};
(
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Apply(error)),
status,
)
(ConnectionHandlerUpgrErr::Upgrade(error), status)
}
},
};
Expand Down
Loading