From 35ea6fd1799383dbbd4e5d6b3995c232bc464dd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Tue, 24 Sep 2024 05:01:48 +0200 Subject: [PATCH] Send unrequested disconnect messages --- .../ip-packet-requests/src/v6/conversion.rs | 16 ++++ .../ip-packet-router/src/mixnet_listener.rs | 77 +++++++++++++++---- 2 files changed, 77 insertions(+), 16 deletions(-) diff --git a/common/ip-packet-requests/src/v6/conversion.rs b/common/ip-packet-requests/src/v6/conversion.rs index be2327b2e6..90d035b446 100644 --- a/common/ip-packet-requests/src/v6/conversion.rs +++ b/common/ip-packet-requests/src/v6/conversion.rs @@ -67,3 +67,19 @@ impl From for v6::response::InfoLevel { } } } + +impl From for v6::response::UnrequestedDisconnectReason { + fn from(reason: v7::response::UnrequestedDisconnectReason) -> Self { + match reason { + v7::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout => { + v6::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout + } + v7::response::UnrequestedDisconnectReason::ClientTunTrafficTimeout => { + v6::response::UnrequestedDisconnectReason::ClientTunTrafficTimeout + } + v7::response::UnrequestedDisconnectReason::Other(reason) => { + v6::response::UnrequestedDisconnectReason::Other(reason) + } + } + } +} diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index 0949c59e49..7713cc0541 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -113,6 +113,7 @@ impl ConnectedClients { ips: IpPair, nym_address: Recipient, mix_hops: Option, + client_version: SupportedClientVersion, forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender>, close_tx: tokio::sync::oneshot::Sender<()>, handle: tokio::task::JoinHandle<()>, @@ -124,6 +125,7 @@ impl ConnectedClients { ipv6: ips.ipv6, mix_hops, last_activity: Arc::new(RwLock::new(std::time::Instant::now())), + client_version, _close_tx: Arc::new(CloseTx { nym_address, inner: Some(close_tx), @@ -156,7 +158,7 @@ impl ConnectedClients { } // Identify connected client handlers that have stopped without being told to stop - fn get_finished_client_handlers(&mut self) -> Vec<(IpPair, Recipient)> { + fn get_finished_client_handlers(&mut self) -> Vec<(IpPair, Recipient, SupportedClientVersion)> { self.clients_ipv4_mapping .iter_mut() .filter_map(|(ip, connected_client)| { @@ -164,6 +166,7 @@ impl ConnectedClients { Some(( IpPair::new(*ip, connected_client.ipv6), connected_client.nym_address, + connected_client.client_version, )) } else { None @@ -172,7 +175,7 @@ impl ConnectedClients { .collect() } - async fn get_inactive_clients(&mut self) -> Vec<(IpPair, Recipient)> { + async fn get_inactive_clients(&mut self) -> Vec<(IpPair, Recipient, SupportedClientVersion)> { let now = std::time::Instant::now(); let mut ret = vec![]; for (ip, connected_client) in self.clients_ipv4_mapping.iter() { @@ -182,14 +185,18 @@ impl ConnectedClients { ret.push(( IpPair::new(*ip, connected_client.ipv6), connected_client.nym_address, + connected_client.client_version, )) } } ret } - fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IpPair, Recipient)>) { - for (ips, _) in &stopped_clients { + fn disconnect_stopped_client_handlers( + &mut self, + stopped_clients: Vec<(IpPair, Recipient, SupportedClientVersion)>, + ) { + for (ips, _, _) in &stopped_clients { log::info!("Disconnect stopped client: {ips}"); self.clients_ipv4_mapping.remove(&ips.ipv4); self.clients_ipv6_mapping.remove(&ips.ipv6); @@ -202,8 +209,11 @@ impl ConnectedClients { } } - fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IpPair, Recipient)>) { - for (ips, _) in &inactive_clients { + fn disconnect_inactive_clients( + &mut self, + inactive_clients: Vec<(IpPair, Recipient, SupportedClientVersion)>, + ) { + for (ips, _, _) in &inactive_clients { log::info!("Disconnect inactive client: {ips}"); self.clients_ipv4_mapping.remove(&ips.ipv4); self.clients_ipv6_mapping.remove(&ips.ipv6); @@ -238,11 +248,15 @@ pub(crate) struct ConnectedClient { pub(crate) ipv6: Ipv6Addr, // Number of mix node hops that the client has requested to use + // WIP(JON): remove me pub(crate) mix_hops: Option, // Keep track of last activity so we can disconnect inactive clients pub(crate) last_activity: Arc>, + // The version of the client, since we need to know this to send the correct response + pub(crate) client_version: SupportedClientVersion, + pub(crate) _close_tx: Arc, // Handle for the connected client handler @@ -414,6 +428,21 @@ impl Response { } } + fn new_unrequested_disconnect( + reply_to: Recipient, + reason: v7::response::UnrequestedDisconnectReason, + client_version: SupportedClientVersion, + ) -> Self { + match client_version { + SupportedClientVersion::V6 => Response::V6( + v6::response::IpPacketResponse::new_unrequested_disconnect(reply_to, reason.into()), + ), + SupportedClientVersion::V7 => Response::V7( + v7::response::IpPacketResponse::new_unrequested_disconnect(reply_to, reason), + ), + } + } + fn to_bytes(&self) -> Result> { match self { Response::V6(response) => response.to_bytes(), @@ -511,6 +540,7 @@ impl MixnetListener { requested_ips, reply_to, reply_to_hops, + client_version, forward_from_tun_tx, close_tx, handle, @@ -606,6 +636,7 @@ impl MixnetListener { new_ips, reply_to, reply_to_hops, + client_version, forward_from_tun_tx, close_tx, handle, @@ -799,9 +830,9 @@ impl MixnetListener { IpPacketRequestData::Ping(ping_request) => { Ok(vec![self.on_ping_request(ping_request, client_version)]) } - IpPacketRequestData::Health(health_request) => Ok(vec![ - self.on_health_request(health_request, client_version).await, - ]), + IpPacketRequestData::Health(health_request) => { + Ok(vec![self.on_health_request(health_request, client_version)]) + } } } @@ -809,13 +840,27 @@ impl MixnetListener { let stopped_clients = self.connected_clients.get_finished_client_handlers(); let inactive_clients = self.connected_clients.get_inactive_clients().await; - // TODO: Send disconnect responses to all disconnected clients - //for (ip, nym_address) in stopped_clients.iter().chain(disconnected_clients.iter()) { - // let response = IpPacketResponse::new_unrequested_disconnect(...) - // if let Err(err) = self.handle_response(response).await { - // log::error!("Failed to send disconnect response: {err}"); - // } - //} + // WIP(JON): confirm we should send disconnect on handle stopped + for (_ip, nym_address, client_version) in &stopped_clients { + let response = Response::new_unrequested_disconnect( + *nym_address, + v7::response::UnrequestedDisconnectReason::Other("handler stopped".to_string()), + *client_version, + ); + if let Err(err) = self.handle_response(response).await { + log::error!("Failed to send disconnect response: {err}"); + } + } + for (_ip, nym_address, client_version) in &inactive_clients { + let response = Response::new_unrequested_disconnect( + *nym_address, + v7::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout, + *client_version, + ); + if let Err(err) = self.handle_response(response).await { + log::error!("Failed to send disconnect response: {err}"); + } + } self.connected_clients .disconnect_stopped_client_handlers(stopped_clients);