diff --git a/nym-api/nym-api-requests/src/models.rs b/nym-api/nym-api-requests/src/models.rs index 60df201a35..8f7d4572b5 100644 --- a/nym-api/nym-api-requests/src/models.rs +++ b/nym-api/nym-api-requests/src/models.rs @@ -21,7 +21,7 @@ use nym_mixnet_contract_common::nym_node::Role; use nym_mixnet_contract_common::reward_params::{Performance, RewardingParams}; use nym_mixnet_contract_common::rewarding::RewardEstimate; use nym_mixnet_contract_common::{ - EpochId, GatewayBond, IdentityKey, Interval, KeyRotationState, MixNode, NodeId, Percent, + EpochId, GatewayBond, IdentityKey, Interval, MixNode, NodeId, Percent, }; use nym_network_defaults::{DEFAULT_MIX_LISTENING_PORT, DEFAULT_VERLOC_LISTENING_PORT}; use nym_node_requests::api::v1::authenticator::models::Authenticator; @@ -41,6 +41,7 @@ use thiserror::Error; use time::{Date, OffsetDateTime}; use utoipa::{IntoParams, ToResponse, ToSchema}; +pub use nym_mixnet_contract_common::KeyRotationState; pub use nym_node_requests::api::v1::node::models::BinaryBuildInformationOwned; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] diff --git a/nym-node/src/node/key_rotation/controller.rs b/nym-node/src/node/key_rotation/controller.rs index 1a4445deb5..82ebc9bb82 100644 --- a/nym-node/src/node/key_rotation/controller.rs +++ b/nym-node/src/node/key_rotation/controller.rs @@ -1,24 +1,46 @@ // Copyright 2025 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 +// SPDX-License-Identifier: GPL-3.0-only use crate::node::key_rotation::manager::SphinxKeyManager; use crate::node::nym_apis_client::NymApisClient; use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; +use crate::node::replay_protection::manager::ReplayProtectionBloomfiltersManager; use futures::pin_mut; use nym_task::ShutdownToken; use nym_validator_client::client::NymApiClientExt; -use nym_validator_client::models::KeyRotationInfoResponse; +use nym_validator_client::models::{KeyRotationInfoResponse, KeyRotationState}; use std::time::Duration; use time::OffsetDateTime; use tokio::time::{interval, sleep, Instant}; use tracing::{error, info, trace, warn}; +struct RotationConfig { + epoch_duration: Duration, + rotation_state: KeyRotationState, +} + +impl RotationConfig { + fn rotation_lifetime(&self) -> Duration { + (self.rotation_state.validity_epochs + 1) * self.epoch_duration + } +} + +impl From for RotationConfig { + fn from(value: KeyRotationInfoResponse) -> Self { + RotationConfig { + epoch_duration: value.epoch_duration, + rotation_state: value.key_rotation_state, + } + } +} + pub(crate) struct KeyRotationController { // regular polling rate to catch any changes in the system config. they shouldn't happen too often // so the requests can be sent quite infrequently regular_polling_interval: Duration, - replay_protection_bloomfilters: ReplayProtectionBloomfilters, + rotation_config: RotationConfig, + replay_protection_manager: ReplayProtectionBloomfiltersManager, client: NymApisClient, managed_keys: SphinxKeyManager, shutdown_token: ShutdownToken, @@ -54,8 +76,21 @@ enum KeyRotationActionState { } impl KeyRotationController { - pub(crate) fn new(client: NymApisClient, shutdown_token: ShutdownToken) -> Self { + pub(crate) async fn new( + regular_polling_interval: Duration, + client: NymApisClient, + replay_protection_manager: ReplayProtectionBloomfiltersManager, + managed_keys: SphinxKeyManager, + shutdown_token: ShutdownToken, + ) -> Self { todo!() + // KeyRotationController { + // regular_polling_interval, + // replay_protection_manager, + // client, + // managed_keys, + // shutdown_token, + // } } async fn determine_next_action(&self) -> NextAction { @@ -151,14 +186,41 @@ impl KeyRotationController { Ok(key) => key, }; + if self + .replay_protection_manager + .allocate_pre_announced(rotation_id, self.rotation_config.rotation_lifetime()) + .is_err() + { + // mutex poisoning - we have to exit + self.shutdown_token.cancel(); + return; + } self.client.broadcast_pre_announced_key(public_key).await; } KeyRotationActionState::SwapDefault => { if let Err(err) = self.managed_keys.rotate_keys() { error!("failed to perform sphinx key swap: {err}") + }; + if self + .replay_protection_manager + .promote_pre_announced() + .is_err() + { + // mutex poisoning - we have to exit + self.shutdown_token.cancel(); + return; + } + } + KeyRotationActionState::PurgeOld => { + if let Err(err) = self.managed_keys.remove_overlap_key() { + error!("failed to remove old sphinx key: {err}"); + }; + if self.replay_protection_manager.purge_secondary().is_err() { + // mutex poisoning - we have to exit + self.shutdown_token.cancel(); + return; } } - KeyRotationActionState::PurgeOld => {} } } diff --git a/nym-node/src/node/nym_apis_client.rs b/nym-node/src/node/nym_apis_client.rs index 7acbc64441..202aaa3b37 100644 --- a/nym-node/src/node/nym_apis_client.rs +++ b/nym-node/src/node/nym_apis_client.rs @@ -1,5 +1,5 @@ // Copyright 2025 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 +// SPDX-License-Identifier: GPL-3.0-only use crate::error::NymNodeError; use crate::node::key_rotation::key::SphinxPublicKey; diff --git a/nym-node/src/node/replay_protection/bloomfilter.rs b/nym-node/src/node/replay_protection/bloomfilter.rs index 0391ebbdbc..906a3dde8f 100644 --- a/nym-node/src/node/replay_protection/bloomfilter.rs +++ b/nym-node/src/node/replay_protection/bloomfilter.rs @@ -6,25 +6,27 @@ use bloomfilter::Bloom; use human_repr::HumanDuration; use nym_sphinx_types::REPLAY_TAG_SIZE; use std::collections::HashMap; +use std::mem; use std::path::Path; use std::sync::{Arc, PoisonError, TryLockError}; use tokio::fs::File; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::time::Instant; -use tracing::{debug, info}; +use tracing::{debug, error, info}; // auxiliary data associated with the bloomfilter to get some statistics from the time of its creation // this is needed in order to more accurately resize it upon reset -struct ReplayProtectionBloomfilterMetadata { +#[derive(Copy, Clone)] +pub(crate) struct ReplayProtectionBloomfilterMetadata { // used in the unlikely case of epoch durations being changed. it doesn't really cost us anything // to include it, so might as well - creation_time: Instant, + pub(crate) creation_time: Instant, /// Number of packets that this node has received since startup, as recorded when this bloomfilter was created. /// Used for determining the approximate packet rate and thus number of entries in the bloomfilter - packets_received_at_creation: usize, + pub(crate) packets_received_at_creation: usize, - rotation_id: u32, + pub(crate) rotation_id: u32, } // it appears that now std Mutex is faster (or comparable) to parking_lot @@ -73,6 +75,83 @@ impl ReplayProtectionBloomfilters { self.disabled } + pub(crate) fn allocate_pre_announced( + &self, + items_count: usize, + fp_p: f64, + packets_received_at_creation: usize, + rotation_id: u32, + ) -> Result<(), NymNodeError> { + // build the new filter + let filter = + Bloom::new_for_fp_rate(items_count, fp_p).map_err(NymNodeError::bloomfilter_failure)?; + + let mut guard = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })?; + + guard.pre_announced = Some(RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata { + creation_time: Instant::now(), + packets_received_at_creation, + rotation_id, + }, + data: filter, + }); + Ok(()) + } + + pub(crate) fn promote_pre_announced(&self) -> Result<(), NymNodeError> { + let mut guard = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })?; + + let Some(mut pre_announced) = guard.pre_announced.take() else { + error!("there was no pre-announced bloomfilter to promote"); + return Ok(()); + }; + + // pre_announced -> primary + // primary -> temp (pre_announced) + mem::swap(&mut guard.primary, &mut pre_announced); + + // temp (pre_announced) -> secondary + guard.secondary = Some(pre_announced); + Ok(()) + } + + pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> { + let mut guard = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })?; + guard.secondary = None; + Ok(()) + } + + pub(crate) fn primary_metadata( + &self, + ) -> Result { + let metadata = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })? + .primary + .metadata; + + Ok(metadata) + } + pub(crate) fn reset(&self, items_count: usize, fp_p: f64) -> Result<(), NymNodeError> { // 1. build the new filter todo!() diff --git a/nym-node/src/node/replay_protection/manager.rs b/nym-node/src/node/replay_protection/manager.rs new file mode 100644 index 0000000000..e067d15e56 --- /dev/null +++ b/nym-node/src/node/replay_protection/manager.rs @@ -0,0 +1,65 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::error::NymNodeError; +use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; +use crate::node::replay_protection::items_in_bloomfilter; +use human_repr::HumanCount; +use nym_node_metrics::NymNodeMetrics; +use std::cmp::max; +use std::time::Duration; +use tracing::info; + +pub(crate) struct ReplayProtectionBloomfiltersManager { + target_fp_p: f64, + minimum_bloomfilter_packets_per_second: usize, + bloomfilter_size_multiplier: f64, + + metrics: NymNodeMetrics, + filters: ReplayProtectionBloomfilters, +} + +impl ReplayProtectionBloomfiltersManager { + pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> { + self.filters.purge_secondary() + } + + pub(crate) fn promote_pre_announced(&self) -> Result<(), NymNodeError> { + self.filters.promote_pre_announced() + } + + // TODO: actually do add some metrics + pub(crate) fn allocate_pre_announced( + &self, + rotation_id: u32, + rotation_lifetime: Duration, + ) -> Result<(), NymNodeError> { + // 1. estimated the number of items in the filter based on the extrapolated items received + // by the primary filter + let received = self.metrics.mixnet.ingress.forward_hop_packets_received() + + self.metrics.mixnet.ingress.final_hop_packets_received(); + + let primary = self.filters.primary_metadata()?; + let time_delta = primary.creation_time.elapsed(); + let received_since_creation = received - primary.packets_received_at_creation; + let received_per_second = + (received_since_creation as f64 / time_delta.as_secs_f64()).round() as usize; + + let bf_received = max( + received_per_second, + self.minimum_bloomfilter_packets_per_second, + ); + let items_in_new_filter = items_in_bloomfilter(rotation_lifetime, bf_received); + let adjusted = + (items_in_new_filter as f64 * self.bloomfilter_size_multiplier).round() as usize; + + info!( + "allocating new bloom filter. new expected number of packets: {} that preserve fp rate of {}", + adjusted.human_count_bare(), + self.target_fp_p + ); + + self.filters + .allocate_pre_announced(adjusted, self.target_fp_p, received, rotation_id) + } +} diff --git a/nym-node/src/node/replay_protection/mod.rs b/nym-node/src/node/replay_protection/mod.rs index 5d5c9d57e2..00b2548f1a 100644 --- a/nym-node/src/node/replay_protection/mod.rs +++ b/nym-node/src/node/replay_protection/mod.rs @@ -6,6 +6,7 @@ use std::time::Duration; pub(crate) mod background_task; pub(crate) mod bloomfilter; +pub(crate) mod manager; pub fn bitmap_size(false_positive_rate: f64, items_in_filter: usize) -> usize { /// Equivalent to ln(1 / 2^ln(2)) = −ln^2(2)