GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'release/2024.10-caramello' into feature/2024.10-caramello-merge

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2024-09-11 12:41:15 +01:00
parent f655fe81d2 5f7499604d
commit e8e86c273e
101 changed files with 1677 additions and 654 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/src/node/client_handling/websocket/connection_handler/fresh.rs
+119 -86
View File
@@ -20,6 +20,7 @@ use nym_gateway_requests::{
};
use nym_mixnet_client::forwarder::MixForwardingSender;
use nym_sphinx::DestinationAddressBytes;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::net::SocketAddr;
use std::time::Duration;
@@ -111,6 +112,9 @@ pub(crate) enum InitialAuthenticationError {
#[error("failed to upgrade the client handler: {source}")]
HandlerUpgradeFailure { source: RequestHandlingError },
#[error("received shutdown")]
ReceivedShutdown,
}
impl InitialAuthenticationError {
@@ -127,6 +131,7 @@ pub(crate) struct FreshHandler<R, S, St> {
pub(crate) outbound_mix_sender: MixForwardingSender,
pub(crate) socket_connection: SocketStream<S>,
pub(crate) peer_address: SocketAddr,
pub(crate) shutdown: TaskClient,
// currently unused (but populated)
pub(crate) negotiated_protocol: Option<u8>,
@@ -149,6 +154,7 @@ where
active_clients_store: ActiveClientsStore,
shared_state: CommonHandlerState<St>,
peer_address: SocketAddr,
shutdown: TaskClient,
) -> Self {
FreshHandler {
rng,
@@ -158,6 +164,7 @@ where
peer_address,
negotiated_protocol: None,
shared_state,
shutdown,
}
}
@@ -201,6 +208,7 @@ where
ws_stream,
self.shared_state.local_identity.as_ref(),
init_msg,
self.shutdown.clone(),
)
.await
}
@@ -521,6 +529,11 @@ where
/// * `client_address`: address of the client wishing to authenticate.
/// * `encrypted_address`: ciphertext of the address of the client wishing to authenticate.
/// * `iv`: fresh IV received with the request.
#[instrument(skip_all
fields(
address = %address,
)
)]
async fn handle_authenticate(
&mut self,
client_protocol_version: Option<u8>,
@@ -531,6 +544,8 @@ where
where
S: AsyncRead + AsyncWrite + Unpin,
{
debug!("handling client registration");
let negotiated_protocol = self.negotiate_client_protocol(client_protocol_version)?;
// populate the negotiated protocol for future uses
self.negotiated_protocol = Some(negotiated_protocol);
@@ -650,6 +665,8 @@ where
let remote_identity = Self::extract_remote_identity_from_register_init(&init_data)?;
let remote_address = remote_identity.derive_destination_address();
debug!(remote_client = %remote_identity);
if self.active_clients_store.is_active(remote_address) {
return Err(InitialAuthenticationError::DuplicateConnection);
}
@@ -657,6 +674,8 @@ where
let shared_keys = self.perform_registration_handshake(init_data).await?;
let client_id = self.register_client(remote_address, &shared_keys).await?;
debug!(client_id = %client_id, "managed to finalize client registration");
let client_details = ClientDetails::new(client_id, remote_address, shared_keys);
Ok(InitialAuthResult::new(
@@ -716,111 +735,125 @@ where
{
trace!("Started waiting for authenticate/register request...");
while let Some(msg) = self.read_websocket_message().await {
let msg = match msg {
Ok(msg) => msg,
Err(source) => {
debug!("failed to obtain message from websocket stream! stopping connection handler: {source}");
return Err(InitialAuthenticationError::FailedToReadMessage { source });
}
};
let mut shutdown = self.shutdown.clone();
loop {
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("received shutdown signal while performing initial authentication");
return Err(InitialAuthenticationError::ReceivedShutdown);
},
msg = self.read_websocket_message() => {
let Some(msg) = msg else {
break;
};
if msg.is_close() {
return Err(InitialAuthenticationError::CloseMessage);
}
let msg = match msg {
Ok(msg) => msg,
Err(source) => {
debug!("failed to obtain message from websocket stream! stopping connection handler: {source}");
return Err(InitialAuthenticationError::FailedToReadMessage { source });
}
};
// ONLY handle 'Authenticate' or 'Register' requests, ignore everything else
match msg {
// we have explicitly checked for close message
Message::Close(_) => unreachable!(),
Message::Text(text_msg) => {
let (mix_sender, mix_receiver) = mpsc::unbounded();
return match self.handle_initial_authentication_request(text_msg).await {
Err(err) => {
debug!("authentication failure: {err}");
if msg.is_close() {
return Err(InitialAuthenticationError::CloseMessage);
}
// try to send error to the client
if let Err(source) =
self.send_websocket_message(err.to_error_message()).await
// ONLY handle 'Authenticate' or 'Register' requests, ignore everything else
match msg {
// we have explicitly checked for close message
Message::Close(_) => unreachable!(),
Message::Text(text_msg) => {
let (mix_sender, mix_receiver) = mpsc::unbounded();
return match self.handle_initial_authentication_request(text_msg).await {
Err(err) => {
debug!("authentication failure: {err}");
// try to send error to the client
if let Err(source) =
self.send_websocket_message(err.to_error_message()).await
{
debug!("Failed to send authentication error response: {source}");
return Err(InitialAuthenticationError::ErrorResponseSendFailure {
source,
});
}
// return the underlying error
Err(err)
}
Ok(auth_result) => {
// try to send auth response back to the client
if let Err(source) = self
.send_websocket_message(auth_result.server_response.into())
.await
{
debug!("Failed to send authentication response: {source}");
return Err(InitialAuthenticationError::ResponseSendFailure {
source,
});
}
if let Some(client_details) = auth_result.client_details {
// Channel for handlers to ask other handlers if they are still active.
let (is_active_request_sender, is_active_request_receiver) =
mpsc::unbounded();
self.active_clients_store.insert_remote(
client_details.address,
mix_sender,
is_active_request_sender,
);
AuthenticatedHandler::upgrade(
self,
client_details,
mix_receiver,
is_active_request_receiver,
)
.await
.map_err(|source| {
InitialAuthenticationError::HandlerUpgradeFailure { source }
})
} else {
// honestly, it's been so long I don't remember under what conditions its possible (if at all)
// to have empty client details
Err(InitialAuthenticationError::EmptyClientDetails)
}
}
};
}
Message::Binary(_) => {
// perhaps logging level should be reduced here, let's leave it for now and see what happens
// if client is working correctly, this should have never happened
debug!("possibly received a sphinx packet without prior authentication. Request is going to be ignored");
if let Err(source) = self
.send_websocket_message(
ServerResponse::new_error(
"binary request without prior authentication",
)
.into(),
)
.await
{
debug!("Failed to send authentication error response: {source}");
return Err(InitialAuthenticationError::ErrorResponseSendFailure {
source,
});
}
// return the underlying error
Err(err)
return Err(InitialAuthenticationError::BinaryRequestWithoutAuthentication);
}
Ok(auth_result) => {
// try to send auth response back to the client
if let Err(source) = self
.send_websocket_message(auth_result.server_response.into())
.await
{
debug!("Failed to send authentication response: {source}");
return Err(InitialAuthenticationError::ResponseSendFailure {
source,
});
}
if let Some(client_details) = auth_result.client_details {
// Channel for handlers to ask other handlers if they are still active.
let (is_active_request_sender, is_active_request_receiver) =
mpsc::unbounded();
self.active_clients_store.insert_remote(
client_details.address,
mix_sender,
is_active_request_sender,
);
AuthenticatedHandler::upgrade(
self,
client_details,
mix_receiver,
is_active_request_receiver,
)
.await
.map_err(|source| {
InitialAuthenticationError::HandlerUpgradeFailure { source }
})
} else {
// honestly, it's been so long I don't remember under what conditions its possible (if at all)
// to have empty client details
Err(InitialAuthenticationError::EmptyClientDetails)
}
}
_ => continue,
};
}
Message::Binary(_) => {
// perhaps logging level should be reduced here, let's leave it for now and see what happens
// if client is working correctly, this should have never happened
debug!("possibly received a sphinx packet without prior authentication. Request is going to be ignored");
if let Err(source) = self
.send_websocket_message(
ServerResponse::new_error(
"binary request without prior authentication",
)
.into(),
)
.await
{
return Err(InitialAuthenticationError::ErrorResponseSendFailure {
source,
});
}
return Err(InitialAuthenticationError::BinaryRequestWithoutAuthentication);
}
_ => continue,
};
}
}
Err(InitialAuthenticationError::ClosedConnection)
}
pub(crate) async fn start_handling(self, shutdown: nym_task::TaskClient)
pub(crate) async fn start_handling(self)
where
S: AsyncRead + AsyncWrite + Unpin + Send,
{
super::handle_connection(self, shutdown).await
super::handle_connection(self).await
}
}
gateway/src/node/client_handling/websocket/connection_handler/mod.rs
+26 -23
View File
@@ -2,13 +2,14 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::Config;
use fresh::InitialAuthenticationError;
use nym_credential_verification::BandwidthFlushingBehaviourConfig;
use nym_gateway_requests::registration::handshake::SharedKeys;
use nym_gateway_requests::ServerResponse;
use nym_gateway_storage::Storage;
use nym_sphinx::DestinationAddressBytes;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_tungstenite::WebSocketStream;
use tracing::{instrument, trace, warn};
@@ -20,6 +21,8 @@ pub(crate) use self::fresh::FreshHandler;
pub(crate) mod authenticated;
mod fresh;
const WEBSOCKET_HANDSHAKE_TIMEOUT: Duration = Duration::from_millis(1_500);
// TODO: note for my future self to consider the following idea:
// split the socket connection into sink and stream
// stream will be for reading explicit requests
@@ -83,47 +86,47 @@ impl InitialAuthResult {
// imo there's no point in including the peer address in anything higher than debug
#[instrument(level = "debug", skip_all, fields(peer = %handle.peer_address))]
pub(crate) async fn handle_connection<R, S, St>(
mut handle: FreshHandler<R, S, St>,
mut shutdown: TaskClient,
) where
pub(crate) async fn handle_connection<R, S, St>(mut handle: FreshHandler<R, S, St>)
where
R: Rng + CryptoRng,
S: AsyncRead + AsyncWrite + Unpin + Send,
St: Storage + Clone + 'static,
{
// If the connection handler abruptly stops, we shouldn't signal global shutdown
shutdown.disarm();
handle.shutdown.disarm();
match shutdown
.run_future(handle.perform_websocket_handshake())
.await
match tokio::time::timeout(
WEBSOCKET_HANDSHAKE_TIMEOUT,
handle.perform_websocket_handshake(),
)
.await
{
None => {
trace!("received shutdown signal while performing websocket handshake");
Err(timeout_err) => {
warn!("websocket handshake timedout: {timeout_err}");
return;
}
Some(Err(err)) => {
Ok(Err(err)) => {
warn!("Failed to complete WebSocket handshake: {err}. Stopping the handler");
return;
}
_ => (),
_ => {}
}
trace!("Managed to perform websocket handshake!");
match shutdown
.run_future(handle.perform_initial_authentication())
.await
{
None => {
trace!("received shutdown signal while performing initial authentication");
return;
}
Some(Err(err)) => {
let shutdown = handle.shutdown.clone();
match handle.perform_initial_authentication().await {
// For storage error, we want to print the extended storage error, but without
// including it in the error that's returned to the clients
Err(InitialAuthenticationError::StorageError(err)) => {
warn!("authentication has failed: {err}");
return;
}
Some(Ok(auth_handle)) => auth_handle.listen_for_requests(shutdown).await,
Err(err) => {
warn!("authentication has failed: {err}");
return;
}
Ok(auth_handle) => auth_handle.listen_for_requests(shutdown).await,
}
trace!("The handler is done!");
gateway/src/node/client_handling/websocket/listener.rs
+3 -2
View File
@@ -54,6 +54,7 @@ where
connection = tcp_listener.accept() => {
match connection {
Ok((socket, remote_addr)) => {
let shutdown = shutdown.clone().named(format!("ClientConnectionHandler_{remote_addr}"));
trace!("received a socket connection from {remote_addr}");
// TODO: I think we *REALLY* need a mechanism for having a maximum number of connected
// clients or spawned tokio tasks -> perhaps a worker system?
@@ -64,9 +65,9 @@ where
active_clients_store.clone(),
self.shared_state.clone(),
remote_addr,
shutdown,
);
let shutdown = shutdown.clone().named(format!("ClientConnectionHandler_{remote_addr}"));
tokio::spawn(async move { handle.start_handling(shutdown).await });
tokio::spawn(handle.start_handling());
}
Err(err) => warn!("failed to get client: {err}"),
}
gateway/src/node/mod.rs
+5 -9
View File
@@ -51,7 +51,6 @@ struct StartedNetworkRequester {
// TODO: should this struct live here?
#[allow(unused)]
struct StartedAuthenticator {
#[cfg(feature = "wireguard")]
wg_api: Arc<nym_wireguard::WgApiWrapper>,
/// Handle to interact with the local authenticator
@@ -146,7 +145,6 @@ pub struct Gateway<St = PersistentStorage> {
storage: St,
#[cfg(all(feature = "wireguard", target_os = "linux"))]
wireguard_data: Option<nym_wireguard::WireguardData>,
run_http_server: bool,
@@ -169,7 +167,6 @@ impl<St> Gateway<St> {
network_requester_opts,
ip_packet_router_opts,
authenticator_opts: None,
#[cfg(all(feature = "wireguard", target_os = "linux"))]
wireguard_data: None,
run_http_server: true,
task_client: None,
@@ -193,7 +190,6 @@ impl<St> Gateway<St> {
identity_keypair,
sphinx_keypair,
storage,
#[cfg(all(feature = "wireguard", target_os = "linux"))]
wireguard_data: None,
run_http_server: true,
task_client: None,
@@ -208,7 +204,6 @@ impl<St> Gateway<St> {
self.task_client = Some(task_client)
}
#[cfg(all(feature = "wireguard", target_os = "linux"))]
pub fn set_wireguard_data(&mut self, wireguard_data: nym_wireguard::WireguardData) {
self.wireguard_data = Some(wireguard_data)
}
@@ -246,7 +241,7 @@ impl<St> Gateway<St> {
mixnet_handling::Listener::new(listening_address, shutdown).start(connection_handler);
}
#[cfg(all(feature = "wireguard", target_os = "linux"))]
#[cfg(target_os = "linux")]
async fn start_authenticator(
&mut self,
forwarding_channel: MixForwardingSender,
@@ -342,7 +337,7 @@ impl<St> Gateway<St> {
}
}
#[cfg(all(feature = "wireguard", not(target_os = "linux")))]
#[cfg(not(target_os = "linux"))]
async fn start_authenticator(
&self,
_forwarding_channel: MixForwardingSender,
@@ -673,14 +668,15 @@ impl<St> Gateway<St> {
info!("embedded ip packet router is disabled");
};
#[cfg(feature = "wireguard")]
let _wg_api = {
let _wg_api = if self.wireguard_data.is_some() {
let embedded_auth = self
.start_authenticator(mix_forwarding_channel, shutdown.fork("authenticator"))
.await
.map_err(|source| GatewayError::AuthenticatorStartError { source })?;
active_clients_store.insert_embedded(embedded_auth.handle);
Some(embedded_auth.wg_api)
} else {
None
};
if self.run_http_server {