diff --git a/Cargo.lock b/Cargo.lock index 64b63ccc..e7c46cf7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -80,9 +80,9 @@ checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5" [[package]] name = "amplify" -version = "4.0.0" +version = "4.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f26966af46e0d200e8bf2b7f16230997c1c3f2d141bc27ccc091c012ed527b58" +checksum = "8629db306c0bbeb0a402e2918bdcf0026b5ddb24c46460f3bf5410b350d98710" dependencies = [ "amplify_derive", "amplify_num", @@ -92,9 +92,9 @@ dependencies = [ [[package]] name = "amplify_derive" -version = "3.0.1" +version = "4.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c87df0f28e6eb1f2d355f29ba6793fa9ca643967528609608d5cbd70bd68f9d1" +checksum = "759dcbfaf94d838367a86d493ec34ccc8aa6fe365cb7880d6bf89006de24d9c1" dependencies = [ "amplify_syn", "proc-macro2", @@ -1677,9 +1677,9 @@ dependencies = [ [[package]] name = "io-reactor" -version = "0.2.1" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "850495a01d6f0b6d29adf5849a75b5a82449d9181bdd642cd681ce3d6f14b58c" +checksum = "2457e8fb1b1f298809fcd93cd15d485f52ef565f7ad47970583c7a80ae6c7e78" dependencies = [ "amplify", "crossbeam-channel", @@ -1748,9 +1748,9 @@ checksum = "478ee9e62aaeaf5b140bd4138753d1f109765488581444218d3ddda43234f3e8" [[package]] name = "libc" -version = "0.2.150" +version = "0.2.152" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89d92a4743f9a61002fae18374ed11e7973f530cb3a3255fb354818118b2203c" +checksum = "13e3bf6590cbc649f4d1a3eefc9d5d6eb746f5200ffb04e5e142700b8faa56e7" [[package]] name = "libgit2-sys" @@ -1908,9 +1908,9 @@ dependencies = [ [[package]] name = "netservices" -version = "0.4.0" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee13a6ce51c79cf719cea8a3f0d3584098e529c726634bc1197cbbb6967b7de2" +checksum = "e227ff39744a414d32b9b473266c1739a950b89cd03694a58b624f4bde926c10" dependencies = [ "amplify", "cyphernet", diff --git a/radicle-node/Cargo.toml b/radicle-node/Cargo.toml index 055ee1ab..62d1b740 100644 --- a/radicle-node/Cargo.toml +++ b/radicle-node/Cargo.toml @@ -19,12 +19,12 @@ colored = { version = "1.9.0" } crossbeam-channel = { version = "0.5.6" } cyphernet = { version = "0.4.1", features = ["tor", "dns", "ed25519", "p2p-ed25519"] } fastrand = { version = "2.0.0" } -io-reactor = { version = "0.2.1", features = ["popol"] } +io-reactor = { version = "0.3.0", features = ["popol"] } lexopt = { version = "0.2.1" } libc = { version = "0.2.137" } log = { version = "0.4.17", features = ["std"] } localtime = { version = "1.2.0" } -netservices = { version = "0.4.0", features = ["io-reactor", "socket2"] } +netservices = { version = "0.5.0", features = ["io-reactor", "socket2"] } nonempty = { version = "0.8.1", features = ["serialize"] } once_cell = { version = "1.13" } qcheck = { version = "1", default-features = false, optional = true } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 7c15474b..e9e6434f 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -4,8 +4,7 @@ //! The handshake itself is implemented in the external [`cyphernet`] and [`netservices`] crates. use std::collections::hash_map::Entry; use std::collections::VecDeque; -use std::os::unix::io::AsRawFd; -use std::os::unix::prelude::RawFd; +use std::os::unix::io::{AsRawFd, RawFd}; use std::sync::Arc; use std::{io, net, time}; @@ -19,7 +18,7 @@ use localtime::LocalTime; use netservices::resource::{ListenerEvent, NetAccept, NetTransport, SessionEvent}; use netservices::session::{ProtocolArtifact, Socks5Session}; use netservices::{NetConnection, NetProtocol, NetReader, NetWriter}; -use reactor::Timestamp; +use reactor::{ResourceId, Timestamp}; use radicle::collections::RandomMap; use radicle::node::NodeId; @@ -141,15 +140,28 @@ impl Streams { } } +/// The initial state of an outbound peer before handshake is completed. +#[derive(Debug)] +struct Outbound { + /// Resource ID, if registered. + id: Option, + /// Remote address. + addr: NetAddr, + /// Remote Node ID. + nid: NodeId, +} + +/// The initial state of an inbound peer before handshake is completed. +#[derive(Debug)] +struct Inbound { + /// Resource ID, if registered. + id: Option, + /// Remote address. + addr: NetAddr, +} + /// Peer connection state machine. enum Peer { - /// The initial state of an inbound peer before handshake is completed. - Inbound { addr: NetAddr }, - /// The initial state of an outbound peer before handshake is completed. - Outbound { - addr: NetAddr, - nid: NodeId, - }, /// The state after handshake is completed. /// Peers in this state are handled by the underlying service. Connected { @@ -171,8 +183,6 @@ enum Peer { impl std::fmt::Debug for Peer { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Inbound { addr } => write!(f, "Inbound({addr})"), - Self::Outbound { nid, .. } => write!(f, "Outbound({nid})"), Self::Connected { link, nid, .. } => write!(f, "Connected({link:?}, {nid})"), Self::Disconnecting { .. } => write!(f, "Disconnecting"), } @@ -183,58 +193,19 @@ impl Peer { /// Return the peer's id, if any. fn id(&self) -> Option<&NodeId> { match self { - Peer::Outbound { nid, .. } - | Peer::Connected { nid, .. } - | Peer::Disconnecting { nid: Some(nid), .. } => Some(nid), - Peer::Inbound { .. } => None, + Peer::Connected { nid, .. } | Peer::Disconnecting { nid: Some(nid), .. } => Some(nid), Peer::Disconnecting { nid: None, .. } => None, } } - /// Return a new inbound connecting peer. - fn inbound(addr: NetAddr) -> Self { - Self::Inbound { addr } - } - - /// Return a new outbound connecting peer. - fn outbound(addr: NetAddr, nid: NodeId) -> Self { - Self::Outbound { addr, nid } - } - - /// Switch to connected state. - fn connected(&mut self, nid: NodeId) -> (NetAddr, Link) { - if let Self::Inbound { addr } = self { - let link = Link::Inbound; - let addr = addr.clone(); - - *self = Self::Connected { - link, - addr: addr.clone(), - nid, - inbox: Deserializer::default(), - streams: Streams::new(link), - }; - (addr, link) - } else if let Self::Outbound { + /// Connected peer. + fn connected(nid: NodeId, addr: NetAddr, link: Link) -> Self { + Self::Connected { + link, addr, - nid: expected, - } = self - { - assert_eq!(nid, *expected); - - let link = Link::Outbound; - let addr = addr.clone(); - - *self = Self::Connected { - link, - addr: addr.clone(), - nid, - inbox: Deserializer::default(), - streams: Streams::new(link), - }; - (addr, link) - } else { - panic!("Peer::connected: session for {nid} is already established"); + nid, + inbox: Deserializer::default(), + streams: Streams::new(link), } } @@ -243,13 +214,6 @@ impl Peer { if let Self::Connected { nid, streams, .. } = self { streams.shutdown(); - *self = Self::Disconnecting { - nid: Some(*nid), - reason, - }; - } else if let Self::Inbound { .. } = self { - *self = Self::Disconnecting { nid: None, reason }; - } else if let Self::Outbound { nid, .. } = self { *self = Self::Disconnecting { nid: Some(*nid), reason, @@ -260,54 +224,53 @@ impl Peer { } } -struct Peers(RandomMap); +/// Holds connected peers. +struct Peers(RandomMap); impl Peers { - fn get_mut(&mut self, fd: &RawFd) -> Option<&mut Peer> { - self.0.get_mut(fd) + fn get_mut(&mut self, id: &ResourceId) -> Option<&mut Peer> { + self.0.get_mut(id) } - fn entry(&mut self, fd: RawFd) -> Entry { - self.0.entry(fd) + fn entry(&mut self, id: ResourceId) -> Entry { + self.0.entry(id) } - fn insert(&mut self, fd: RawFd, peer: Peer) { - if self.0.insert(fd, peer).is_some() { - log::warn!(target: "wire", "Replacing existing peer fd={fd}"); + fn insert(&mut self, id: ResourceId, peer: Peer) { + if self.0.insert(id, peer).is_some() { + log::warn!(target: "wire", "Replacing existing peer id={id}"); } } - fn remove(&mut self, fd: &RawFd) -> Option { - self.0.remove(fd) + fn remove(&mut self, id: &ResourceId) -> Option { + self.0.remove(id) } - fn lookup(&self, node_id: &NodeId) -> Option<(RawFd, &Peer)> { + fn lookup(&self, node_id: &NodeId) -> Option<(ResourceId, &Peer)> { self.0 .iter() .find(|(_, peer)| peer.id() == Some(node_id)) .map(|(fd, peer)| (*fd, peer)) } - fn lookup_mut(&mut self, node_id: &NodeId) -> Option<(RawFd, &mut Peer)> { + fn lookup_mut(&mut self, node_id: &NodeId) -> Option<(ResourceId, &mut Peer)> { self.0 .iter_mut() .find(|(_, peer)| peer.id() == Some(node_id)) .map(|(fd, peer)| (*fd, peer)) } - fn active(&self) -> impl Iterator { - self.0.iter().filter_map(|(fd, peer)| match peer { - Peer::Inbound { .. } => None, - Peer::Outbound { nid, .. } => Some((*fd, nid)), - Peer::Connected { nid, .. } => Some((*fd, nid)), + fn active(&self) -> impl Iterator { + self.0.iter().filter_map(|(id, peer)| match peer { + Peer::Connected { nid, .. } => Some((*id, nid)), Peer::Disconnecting { .. } => None, }) } - fn connected(&self) -> impl Iterator { - self.0.iter().filter_map(|(fd, peer)| { - if let Peer::Connected { nid: id, .. } = peer { - Some((*fd, id)) + fn connected(&self) -> impl Iterator { + self.0.iter().filter_map(|(id, peer)| { + if let Peer::Connected { nid, .. } = peer { + Some((*id, nid)) } else { None } @@ -325,7 +288,11 @@ pub struct Wire { signer: G, /// Internal queue of actions to send to the reactor. actions: VecDeque>, - /// Peer sessions. + /// Outbound attempted peers without a session. + outbound: RandomMap, + /// Inbound peers without a session. + inbound: RandomMap, + /// Peer (established) sessions. peers: Peers, /// SOCKS5 proxy address. proxy: net::SocketAddr, @@ -351,6 +318,8 @@ where signer, proxy, actions: VecDeque::new(), + inbound: RandomMap::default(), + outbound: RandomMap::default(), peers: Peers(RandomMap::default()), } } @@ -359,19 +328,19 @@ where self.actions.push_back(Action::RegisterListener(socket)); } - fn disconnect(&mut self, fd: RawFd, reason: DisconnectReason) { - match self.peers.get_mut(&fd) { + fn disconnect(&mut self, id: ResourceId, reason: DisconnectReason) { + match self.peers.get_mut(&id) { Some(Peer::Disconnecting { .. }) => { - log::error!(target: "wire", "Peer (fd={fd}) is already disconnecting"); + log::error!(target: "wire", "Peer (id={id}) is already disconnecting"); } Some(peer) => { - log::debug!(target: "wire", "Disconnecting peer (fd={fd}): {reason}"); + log::debug!(target: "wire", "Disconnecting peer (id={id}): {reason}"); peer.disconnecting(reason); - self.actions.push_back(Action::UnregisterTransport(fd)); + self.actions.push_back(Action::UnregisterTransport(id)); } None => { - log::error!(target: "wire", "Unknown peer (fd={fd}) cannot be disconnected"); + log::error!(target: "wire", "Unknown peer (id={id}) cannot be disconnected"); } } } @@ -467,22 +436,20 @@ where fn handle_listener_event( &mut self, - _sock: net::SocketAddr, + _: ResourceId, // Nb. This is the ID of the listener socket. event: ListenerEvent>, _: Timestamp, ) { match event { ListenerEvent::Accepted(connection) => { let addr = connection.remote_addr(); - log::debug!(target: "wire", "Accepting inbound peer connection from {addr}.."); - - self.peers - .insert(connection.as_raw_fd(), Peer::inbound(addr.clone().into())); + let fd = connection.as_raw_fd(); + log::debug!(target: "wire", "Accepting inbound connection from {addr} (fd={fd}).."); // If the service doesn't want to accept this connection, // we drop the connection here, which disconnects the socket. if !self.service.accepted(NetAddr::from(addr.clone()).into()) { - log::debug!(target: "wire", "Dropping inbound connection from {addr}.."); + log::debug!(target: "wire", "Rejecting inbound connection from {addr} (fd={fd}).."); drop(connection); return; @@ -496,6 +463,14 @@ where return; } }; + + self.inbound.insert( + fd, + Inbound { + id: None, + addr: addr.into(), + }, + ); self.actions .push_back(reactor::Action::RegisterTransport(transport)) } @@ -505,45 +480,62 @@ where } } + fn handle_registered(&mut self, fd: RawFd, id: ResourceId) { + if let Some(outbound) = self.outbound.get_mut(&fd) { + log::debug!(target: "wire", "Outbound peer resource registered with id={id} (fd={fd})"); + outbound.id = Some(id); + } else if let Some(inbound) = self.inbound.get_mut(&fd) { + log::debug!(target: "wire", "Inbound peer resource registered with id={id} (fd={fd})"); + inbound.id = Some(id); + } else { + log::error!(target: "wire", "Unknown peer registered with fd={fd} and id={id}"); + } + } + fn handle_transport_event( &mut self, - fd: RawFd, + id: ResourceId, event: SessionEvent>, _: Timestamp, ) { match event { - SessionEvent::Established(ProtocolArtifact { state, .. }) => { + SessionEvent::Established(fd, ProtocolArtifact { state, .. }) => { // SAFETY: With the NoiseXK protocol, there is always a remote static key. - let id: NodeId = state.remote_static_key.unwrap(); + let nid: NodeId = state.remote_static_key.unwrap(); - log::debug!(target: "wire", "Session established with {id} (fd={fd})"); + log::debug!(target: "wire", "Session established with {nid} (id={id}) (fd={fd})"); let conflicting = self .peers .active() - .filter(|(other, d)| **d == id && *other != fd) - .map(|(fd, _)| fd) + .filter(|(other, d)| **d == nid && *other != id) + .map(|(id, _)| id) .collect::>(); - for fd in conflicting { + for id in conflicting { log::warn!( - target: "wire", "Closing conflicting session with {id} (fd={fd})" + target: "wire", "Closing conflicting session with {nid} (id={id})" ); self.disconnect( - fd, + id, DisconnectReason::Dial(Arc::new(io::Error::from( io::ErrorKind::AlreadyExists, ))), ); } - let Some(peer) = self.peers.get_mut(&fd) else { - log::error!(target: "wire", "Session not found for fd {fd}"); + let (addr, link) = if let Some(peer) = self.inbound.remove(&fd) { + (peer.addr, Link::Inbound) + } else if let Some(peer) = self.outbound.remove(&fd) { + assert_eq!(nid, peer.nid); + (peer.addr, Link::Outbound) + } else { + log::error!(target: "wire", "Session for {nid} (id={id}) not found"); return; }; - let (addr, link) = peer.connected(id); - - self.service.connected(id, addr.into(), link); + self.peers + .insert(id, Peer::connected(nid, addr.clone(), link)); + self.service.connected(nid, addr.into(), link); } SessionEvent::Data(data) => { if let Some(Peer::Connected { @@ -551,7 +543,7 @@ where inbox, streams, .. - }) = self.peers.get_mut(&fd) + }) = self.peers.get_mut(&id) { inbox.input(&data); @@ -631,7 +623,7 @@ where log::debug!(target: "wire", "Dropping read buffer for {nid} with {} bytes", inbox.unparsed().count()); } self.disconnect( - fd, + id, DisconnectReason::Session(session::Error::Misbehavior), ); break; @@ -639,11 +631,11 @@ where } } } else { - log::warn!(target: "wire", "Dropping message from unconnected peer (fd={fd})"); + log::warn!(target: "wire", "Dropping message from unconnected peer (id={id})"); } } SessionEvent::Terminated(err) => { - self.disconnect(fd, DisconnectReason::Connection(Arc::new(err))); + self.disconnect(id, DisconnectReason::Connection(Arc::new(err))); } } } @@ -702,36 +694,35 @@ where } } - fn handover_listener(&mut self, _listener: Self::Listener) { + fn handover_listener(&mut self, _id: ResourceId, _listener: Self::Listener) { panic!("Wire::handover_listener: listener handover is not supported"); } - fn handover_transport(&mut self, transport: Self::Transport) { + fn handover_transport(&mut self, id: ResourceId, transport: Self::Transport) { let fd = transport.as_raw_fd(); - log::debug!(target: "wire", "Received transport handover (fd={fd})"); - match self.peers.entry(fd) { + match self.peers.entry(id) { Entry::Occupied(e) => { match e.get() { - Peer::Disconnecting { - nid: id, reason, .. - } => { + Peer::Disconnecting { nid, reason, .. } => { + log::debug!(target: "wire", "Received transport handover for disconnecting peer with id={id} (fd={fd})"); + // Disconnect TCP stream. drop(transport); - // If there is no ID, the service is not aware of the peer. - if let Some(id) = id { - self.service.disconnected(*id, reason); + // If there is no NID, the service is not aware of the peer. + if let Some(nid) = nid { + self.service.disconnected(*nid, reason); } e.remove(); } - _ => { - panic!("Wire::handover_transport: Unexpected peer with fd {fd} handed over from the reactor"); + Peer::Connected { nid, .. } => { + panic!("Wire::handover_transport: Unexpected handover of connected peer {} with id={id} (fd={fd})", nid); } } } Entry::Vacant(_) => { - panic!("Wire::handover_transport: Unknown peer with fd {fd} handed over"); + panic!("Wire::handover_transport: Unknown peer with id={id} (fd={fd}) handed over"); } } } @@ -795,11 +786,13 @@ where }) { Ok(transport) => { self.service.attempted(node_id, addr.clone()); - // TODO: Keep track of peer address for when peer disconnects before - // handshake is complete. - self.peers.insert( + self.outbound.insert( transport.as_raw_fd(), - Peer::outbound(addr.to_inner(), node_id), + Outbound { + id: None, + nid: node_id, + addr: addr.to_inner(), + }, ); self.actions .push_back(reactor::Action::RegisterTransport(transport)); @@ -813,8 +806,8 @@ where } } Io::Disconnect(nid, reason) => { - if let Some((fd, Peer::Connected { .. })) = self.peers.lookup(&nid) { - self.disconnect(fd, reason); + if let Some((id, Peer::Connected { .. })) = self.peers.lookup(&nid) { + self.disconnect(id, reason); } else { log::warn!(target: "wire", "Peer {nid} is not connected: ignoring disconnect"); }