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

gateway/authenticator: add peer-controller timeouts and lock tweaks to keep registration responsive after suspend

Browse Source
This commit is contained in:
Tommy Verrall
2025-11-21 09:26:41 +01:00
parent 96e3ff2af9
commit 8dbbd2bdf4
3 changed files with 49 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/src/node/internal_service_providers/authenticator/error.rs
+3
View File
@@ -85,6 +85,9 @@ pub enum AuthenticatorError {
#[error("peers can't be interacted with anymore")]
PeerInteractionStopped,
#[error("peers interaction timed out while attempting to {operation}")]
PeerInteractionTimeout { operation: &'static str },
#[error("unknown version number")]
UnknownVersion,
gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs
+23 -7
View File
@@ -185,11 +185,16 @@ impl MixnetListener {
) -> AuthenticatorHandleResult {
let remote_public = init_message.pub_key();
let nonce: u64 = fastrand::u64(..);
let mut registered_and_free = self.registered_and_free.write().await;
if let Some(registration_data) = registered_and_free
.registration_in_progres
.get(&remote_public)
{
let pending_registration = {
let registered_and_free = self.registered_and_free.read().await;
registered_and_free
.registration_in_progres
.get(&remote_public)
.cloned()
};
if let Some(registration_data) = pending_registration {
let gateway_data = registration_data.gateway_data.clone();
let bytes = match AuthenticatorVersion::from(protocol) {
AuthenticatorVersion::V1 => {
@@ -292,7 +297,17 @@ impl MixnetListener {
return Ok((bytes, reply_to));
}
let peer = self.peer_manager.query_peer(remote_public).await?;
let peer = match self.peer_manager.query_peer(remote_public).await {
Ok(peer) => peer,
Err(err) => {
tracing::warn!(
"Failed to query peer {}: {err}. Continuing with fresh registration",
remote_public
);
None
}
};
if let Some(peer) = peer {
let allowed_ipv4 = peer
.allowed_ips
@@ -383,6 +398,7 @@ impl MixnetListener {
return Ok((bytes, reply_to));
}
let mut registered_and_free = self.registered_and_free.write().await;
let private_ip_ref = registered_and_free
.free_private_network_ips
.iter_mut()
@@ -818,7 +834,7 @@ impl MixnetListener {
.to_bytes()
.map_err(AuthenticatorError::response_serialisation)?,
AuthenticatorVersion::V1 | AuthenticatorVersion::V2 | AuthenticatorVersion::UNKNOWN => {
return Err(AuthenticatorError::UnknownVersion)
return Err(AuthenticatorError::UnknownVersion);
}
};
gateway/src/node/internal_service_providers/authenticator/peer_manager.rs
+23 -25
View File
@@ -8,6 +8,9 @@ use nym_credential_verification::{ClientBandwidth, TicketVerifier};
use nym_credentials_interface::CredentialSpendingData;
use nym_wireguard::{peer_controller::PeerControlRequest, WireguardGatewayData};
use nym_wireguard_types::PeerPublicKey;
use tokio::time::{timeout, Duration};
const PEER_MANAGER_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
pub struct PeerManager {
pub(crate) wireguard_gateway_data: WireguardGatewayData,
@@ -28,9 +31,8 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| AuthenticatorError::InternalError("no response for add peer".to_string()))?
recv_with_timeout(response_rx, "add peer")
.await?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"adding peer could not be performed: {err:?}"
@@ -48,11 +50,8 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for remove peer".to_string())
})?
recv_with_timeout(response_rx, "remove peer")
.await?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"removing peer could not be performed: {err:?}"
@@ -73,11 +72,8 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for query peer".to_string())
})?
recv_with_timeout(response_rx, "query peer")
.await?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying peer could not be performed: {err:?}"
@@ -106,13 +102,8 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError(
"no response for query client bandwidth".to_string(),
)
})?
recv_with_timeout(response_rx, "query client bandwidth")
.await?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying client bandwidth could not be performed: {err:?}"
@@ -138,11 +129,8 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for query verifier".to_string())
})?
recv_with_timeout(response_rx, "query verifier")
.await?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying verifier could not be performed: {err:?}"
@@ -151,6 +139,16 @@ impl PeerManager {
}
}
async fn recv_with_timeout<T>(
response_rx: oneshot::Receiver<T>,
operation: &'static str,
) -> Result<T, AuthenticatorError> {
timeout(PEER_MANAGER_RESPONSE_TIMEOUT, response_rx)
.await
.map_err(|_| AuthenticatorError::PeerInteractionTimeout { operation })?
.map_err(|_| AuthenticatorError::PeerInteractionStopped)
}
#[cfg(test)]
mod tests {
use std::{str::FromStr, sync::Arc};