diff --git a/common/cosmwasm-smart-contracts/mixnet-contract/src/key_rotation.rs b/common/cosmwasm-smart-contracts/mixnet-contract/src/key_rotation.rs index 90b6f7ac6b..de3f05d241 100644 --- a/common/cosmwasm-smart-contracts/mixnet-contract/src/key_rotation.rs +++ b/common/cosmwasm-smart-contracts/mixnet-contract/src/key_rotation.rs @@ -25,6 +25,18 @@ impl KeyRotationState { let full_rots = diff / self.validity_epochs; full_rots } + + pub fn next_rotation_starting_epoch_id(&self, current_epoch_id: EpochId) -> EpochId { + let current_rotation_id = self.key_rotation_id(current_epoch_id); + + self.initial_epoch_id + self.validity_epochs * (current_rotation_id + 1) + } + + pub fn current_rotation_starting_epoch_id(&self, current_epoch_id: EpochId) -> EpochId { + let current_rotation_id = self.key_rotation_id(current_epoch_id); + + self.initial_epoch_id + self.validity_epochs * current_rotation_id + } } #[cw_serde] @@ -71,4 +83,76 @@ mod tests { assert_eq!(2, state.key_rotation_id(10048)); assert_eq!(2, state.key_rotation_id(10060)); } + + #[test] + fn next_rotation_starting_epoch_id() { + let state = KeyRotationState { + validity_epochs: 24, + initial_epoch_id: 0, + }; + assert_eq!(24, state.next_rotation_starting_epoch_id(0)); + assert_eq!(24, state.next_rotation_starting_epoch_id(23)); + assert_eq!(48, state.next_rotation_starting_epoch_id(24)); + assert_eq!(48, state.next_rotation_starting_epoch_id(47)); + assert_eq!(72, state.next_rotation_starting_epoch_id(48)); + + let state = KeyRotationState { + validity_epochs: 12, + initial_epoch_id: 0, + }; + assert_eq!(12, state.next_rotation_starting_epoch_id(0)); + assert_eq!(12, state.next_rotation_starting_epoch_id(11)); + assert_eq!(24, state.next_rotation_starting_epoch_id(12)); + assert_eq!(24, state.next_rotation_starting_epoch_id(23)); + assert_eq!(36, state.next_rotation_starting_epoch_id(24)); + + let state = KeyRotationState { + validity_epochs: 24, + initial_epoch_id: 10000, + }; + assert_eq!(10024, state.next_rotation_starting_epoch_id(123)); + assert_eq!(10024, state.next_rotation_starting_epoch_id(10000)); + assert_eq!(10024, state.next_rotation_starting_epoch_id(10001)); + assert_eq!(10024, state.next_rotation_starting_epoch_id(10023)); + assert_eq!(10048, state.next_rotation_starting_epoch_id(10024)); + assert_eq!(10048, state.next_rotation_starting_epoch_id(10047)); + assert_eq!(10072, state.next_rotation_starting_epoch_id(10048)); + assert_eq!(10072, state.next_rotation_starting_epoch_id(10060)); + } + + #[test] + fn current_rotation_starting_epoch_id() { + let state = KeyRotationState { + validity_epochs: 24, + initial_epoch_id: 0, + }; + assert_eq!(0, state.current_rotation_starting_epoch_id(0)); + assert_eq!(0, state.current_rotation_starting_epoch_id(23)); + assert_eq!(24, state.current_rotation_starting_epoch_id(24)); + assert_eq!(24, state.current_rotation_starting_epoch_id(47)); + assert_eq!(48, state.current_rotation_starting_epoch_id(48)); + + let state = KeyRotationState { + validity_epochs: 12, + initial_epoch_id: 0, + }; + assert_eq!(0, state.current_rotation_starting_epoch_id(0)); + assert_eq!(0, state.current_rotation_starting_epoch_id(11)); + assert_eq!(12, state.current_rotation_starting_epoch_id(12)); + assert_eq!(12, state.current_rotation_starting_epoch_id(23)); + assert_eq!(24, state.current_rotation_starting_epoch_id(24)); + + let state = KeyRotationState { + validity_epochs: 24, + initial_epoch_id: 10000, + }; + assert_eq!(10000, state.current_rotation_starting_epoch_id(123)); + assert_eq!(10000, state.current_rotation_starting_epoch_id(10000)); + assert_eq!(10000, state.current_rotation_starting_epoch_id(10001)); + assert_eq!(10000, state.current_rotation_starting_epoch_id(10023)); + assert_eq!(10024, state.current_rotation_starting_epoch_id(10024)); + assert_eq!(10024, state.current_rotation_starting_epoch_id(10047)); + assert_eq!(10048, state.current_rotation_starting_epoch_id(10048)); + assert_eq!(10048, state.current_rotation_starting_epoch_id(10060)); + } } diff --git a/common/nymsphinx/framing/src/processing.rs b/common/nymsphinx/framing/src/processing.rs index 59deb0181e..3703f36d84 100644 --- a/common/nymsphinx/framing/src/processing.rs +++ b/common/nymsphinx/framing/src/processing.rs @@ -110,6 +110,11 @@ pub enum PacketProcessingError { PacketReplay, } +pub struct PartialyUnwrappedPacketWithKeyRotation { + pub packet: PartiallyUnwrappedPacket, + pub used_key_rotation: u32, +} + pub struct PartiallyUnwrappedPacket { received_data: FramedNymPacket, partial_result: PartialMixProcessingResult, @@ -174,6 +179,16 @@ impl PartiallyUnwrappedPacket { pub fn replay_tag(&self) -> Option<&[u8; REPLAY_TAG_SIZE]> { self.partial_result.replay_tag() } + + pub fn with_key_rotation( + self, + used_key_rotation: u32, + ) -> PartialyUnwrappedPacketWithKeyRotation { + PartialyUnwrappedPacketWithKeyRotation { + packet: self, + used_key_rotation, + } + } } impl From<(FramedNymPacket, PartialMixProcessingResult)> for PartiallyUnwrappedPacket { diff --git a/nym-api/nym-api-requests/src/models.rs b/nym-api/nym-api-requests/src/models.rs index 3435717922..60df201a35 100644 --- a/nym-api/nym-api-requests/src/models.rs +++ b/nym-api/nym-api-requests/src/models.rs @@ -1454,6 +1454,23 @@ pub struct KeyRotationInfoResponse { pub epoch_duration: Duration, } +impl KeyRotationInfoResponse { + pub fn current_key_rotation_id(&self) -> u32 { + self.key_rotation_state + .key_rotation_id(self.current_epoch_id) + } + + pub fn next_rotation_starting_epoch_id(&self) -> EpochId { + self.key_rotation_state + .next_rotation_starting_epoch_id(self.current_epoch_id) + } + + pub fn current_rotation_starting_epoch_id(&self) -> EpochId { + self.key_rotation_state + .current_rotation_starting_epoch_id(self.current_epoch_id) + } +} + #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)] pub struct RewardedSetResponse { #[serde(default)] diff --git a/nym-node/nym-node-requests/src/api/mod.rs b/nym-node/nym-node-requests/src/api/mod.rs index 525cba0cf3..990fd95bb6 100644 --- a/nym-node/nym-node-requests/src/api/mod.rs +++ b/nym-node/nym-node-requests/src/api/mod.rs @@ -10,7 +10,6 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; use std::ops::Deref; -use utoipa::ToSchema; #[cfg(feature = "client")] pub mod client; @@ -22,7 +21,7 @@ pub use client::Client; // create the type alias manually if openapi is not enabled pub type SignedHostInformation = SignedData; -#[derive(ToSchema)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] pub struct SignedDataHostInfo { // #[serde(flatten)] pub data: crate::api::v1::node::models::HostInformation, @@ -341,8 +340,7 @@ mod tests { assert!(!current_struct_no_noise.verify(ed22519.public_key())); assert!(current_struct_no_noise.verify_host_information()); - // if noise key is present, the signature is actually valid - assert!(current_struct_noise.verify(ed22519.public_key())); + assert!(!current_struct_noise.verify(ed22519.public_key())); assert!(current_struct_noise.verify_host_information()) } diff --git a/nym-node/src/error.rs b/nym-node/src/error.rs index 40c50e97aa..e779f2659a 100644 --- a/nym-node/src/error.rs +++ b/nym-node/src/error.rs @@ -56,6 +56,15 @@ pub enum KeyIOFailure { err: io::Error, }, + #[error("failed to copy {key} key from '{}' to '{}': {err}", source.display(), destination.display())] + KeyCopyFailure { + key: String, + source: PathBuf, + destination: PathBuf, + #[source] + err: io::Error, + }, + #[error("failed to remove {key} key from '{}': {err}", path.display())] KeyRemovalFailure { key: String, diff --git a/nym-node/src/node/key_rotation/active_keys.rs b/nym-node/src/node/key_rotation/active_keys.rs index cb2dcba51f..94b463e604 100644 --- a/nym-node/src/node/key_rotation/active_keys.rs +++ b/nym-node/src/node/key_rotation/active_keys.rs @@ -16,11 +16,9 @@ struct ActiveSphinxKeysInner { /// Key that's currently used as the default when processing packets with no explicit rotation information primary_key: ArcSwap, - /// Optionally, a key from the previous rotation during the overlap period when the keys are rotated. + /// Optionally, a secondary key associated with this node. depending on the context it could either be + /// the pre-announced key for the following rotation or a key from the previous rotation for the overlap period secondary_key: ArcSwapOption, - - /// Optionally, a key for the upcoming rotation that's being pre-announced to other network entities - pre_announced_key: ArcSwapOption, } impl ActiveSphinxKeys { @@ -37,12 +35,13 @@ impl ActiveSphinxKeys { primary: SphinxPrivateKey, secondary: Option, ) -> Self { - ActiveSphinxKeys { - inner: Arc::new(ActiveSphinxKeysInner { - primary_key: ArcSwap::from_pointee(primary), - secondary_key: ArcSwapOption::from_pointee(secondary), - }), - } + todo!() + // ActiveSphinxKeys { + // inner: Arc::new(ActiveSphinxKeysInner { + // primary_key: ArcSwap::from_pointee(primary), + // secondary_key: ArcSwapOption::from_pointee(secondary), + // }), + // } } pub(crate) fn even(&self) -> Option { @@ -74,17 +73,39 @@ impl ActiveSphinxKeys { Some(SphinxKeyGuard::Secondary(SecondaryKeyGuard { guard })) } - pub(crate) fn rotate(&self, new_primary: SphinxPrivateKey) { - if self.inner.secondary_key.load().is_some() { - // this should NEVER happen, but technically nothing should blow up - error!("somehow our secondary key was still set during the rotation!") - } - - let old_primary = self.inner.primary_key.swap(Arc::new(new_primary)); - self.inner.secondary_key.store(Some(old_primary)); + pub(crate) fn set_secondary(&self, new_key: SphinxPrivateKey) { + self.inner.secondary_key.store(Some(Arc::new(new_key))) } - fn deactivate_secondary(&self) { + pub(crate) fn secondary_key_rotation_id(&self) -> Option { + self.inner + .secondary_key + .load() + .as_ref() + .map(|k| k.rotation_id()) + } + + // set the secondary (pre-announced key) as the primary + // and the current primary as the secondary (for the overlap epoch) + pub(crate) fn rotate(&self) -> bool { + let primary = self.inner.primary_key.load(); + + let Some(pre_announced) = self.inner.secondary_key.load_full() else { + error!("sphinx key inconsistency - attempted to perform key rotation without having pre-announced new key"); + return false; + }; + + if pre_announced.rotation_id() != primary.rotation_id() + 1 { + error!("sphinx key inconsistency - pre-announced key rotation id != primary + 1"); + return false; + } + + let old_primary = self.inner.primary_key.swap(pre_announced); + self.inner.secondary_key.store(Some(old_primary)); + true + } + + pub(crate) fn deactivate_secondary(&self) { self.inner.secondary_key.store(None); } } @@ -106,8 +127,25 @@ impl Deref for SphinxKeyGuard { } } +// enum SecondaryKey { +// PreAnnounced(SphinxPrivateKey), +// PreviousOverlap(SphinxPrivateKey), +// } + +// impl Deref for SecondaryKey { +// type Target = SphinxPrivateKey; +// +// fn deref(&self) -> &Self::Target { +// match self { +// SecondaryKey::PreAnnounced(key) => &key, +// SecondaryKey::PreviousOverlap(key) => &key, +// } +// } +// } + pub(crate) struct SecondaryKeyGuard { guard: Guard>>, + // guard: Guard>>, } impl Deref for SecondaryKeyGuard { diff --git a/nym-node/src/node/key_rotation/controller.rs b/nym-node/src/node/key_rotation/controller.rs index fdd76afece..1a4445deb5 100644 --- a/nym-node/src/node/key_rotation/controller.rs +++ b/nym-node/src/node/key_rotation/controller.rs @@ -3,26 +3,53 @@ use crate::node::key_rotation::manager::SphinxKeyManager; use crate::node::nym_apis_client::NymApisClient; +use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; +use futures::pin_mut; use nym_task::ShutdownToken; +use nym_validator_client::client::NymApiClientExt; +use nym_validator_client::models::KeyRotationInfoResponse; use std::time::Duration; -use tokio::time::interval; -use tracing::{info, trace}; +use time::OffsetDateTime; +use tokio::time::{interval, sleep, Instant}; +use tracing::{error, info, trace, warn}; 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, client: NymApisClient, managed_keys: SphinxKeyManager, shutdown_token: ShutdownToken, } -enum KeyRotationActionState { - // perform key-rotation and pre-announce new key to the nym-api(s) - PreAnnounce, +struct NextAction { + typ: KeyRotationActionState, + deadline: OffsetDateTime, +} - // remove the old key and purge associated data like the replay detection bloomfilter +impl NextAction { + fn until_deadline(&self) -> Duration { + let now = OffsetDateTime::now_utc(); + Duration::try_from(self.deadline - now).unwrap_or_else(|_| { + // deadline is already in the past + Duration::from_nanos(0) + }) + } +} + +#[derive(Clone, Copy)] +enum KeyRotationActionState { + // generate and pre-announce new key to the nym-api(s) + PreAnnounce { rotation_id: u32 }, + + // perform the following exchange + // primary -> secondary + // pre_announced -> primary + SwapDefault, + + // remove the old overlap key and purge associated data like the replay detection bloomfilter PurgeOld, } @@ -31,8 +58,108 @@ impl KeyRotationController { todo!() } - async fn regular_poll(&self) { - todo!() + async fn determine_next_action(&self) -> NextAction { + loop { + if let Some(next) = self.try_determine_next_action().await { + return next; + } + + warn!("failed to determine next key rotation action; will try again in 2min"); + sleep(Duration::from_secs(120)).await; + } + } + + async fn try_determine_next_action(&self) -> Option { + let key_rotation_info = self.try_get_key_rotation_info().await?; + let current_rotation = key_rotation_info.current_key_rotation_id(); + + let current_epoch = key_rotation_info.current_epoch_id; + let next_rotation_epoch = key_rotation_info.next_rotation_starting_epoch_id(); + let current_rotation_epoch = key_rotation_info.current_rotation_starting_epoch_id(); + + let secondary_rotation_id = self.managed_keys.keys.secondary_key_rotation_id(); + let (action, execution_epoch) = match secondary_rotation_id { + None => { + // we don't have any secondary key, meaning the next thing we could possibly do is to pre-announce new key + // an epoch before next rotation + let rotation_id = current_rotation + 1; + + ( + KeyRotationActionState::PreAnnounce { rotation_id }, + next_rotation_epoch - 1, + ) + } + Some(id) if id == current_rotation - 1 => { + // our secondary key is from the previous rotation, meaning the next thing we have to do + // is to remove it (we have clearly already rotated) + (KeyRotationActionState::PurgeOld, current_rotation_epoch + 1) + } + Some(id) if id == current_rotation => { + // our secondary key is from the current epoch, meaning (hopefully) we just have gone into the + // next rotation, and we have to swap it into the primary + (KeyRotationActionState::SwapDefault, current_rotation_epoch) + } + Some(id) if id == current_rotation + 1 => { + // our secondary key is from the upcoming rotation, meaning it's the pre-announced key, meaning + // the next thing we have to do is to swap it into the primary + (KeyRotationActionState::SwapDefault, next_rotation_epoch) + } + Some(other) => { + // this situation should have never occurred, our secondary key is completely unusable, + // so we should just remove it immediately and try again + error!("inconsistent secondary key state. it's marked for rotation {other} while the current value is {current_rotation}"); + (KeyRotationActionState::PurgeOld, current_epoch) + } + }; + + let now = OffsetDateTime::now_utc(); + let since_epoch_start = now - key_rotation_info.current_epoch_start; + let until_execution_epoch = + execution_epoch.saturating_sub(current_epoch) * key_rotation_info.epoch_duration; + + Some(NextAction { + typ: action, + deadline: now - since_epoch_start + until_execution_epoch, + }) + } + + async fn try_get_key_rotation_info(&self) -> Option { + let Ok(rotation_info) = self + .client + .query_exhaustively( + async |c| c.get_key_rotation_info().await, + Duration::from_secs(5), + ) + .await + else { + warn!("failed to retrieve key rotation information from ANY nym-api - we might miss configuration changes"); + return None; + }; + + Some(rotation_info) + } + + async fn execute_next_action(&self, action: KeyRotationActionState) { + match action { + KeyRotationActionState::PreAnnounce { rotation_id } => { + let public_key = match self.managed_keys.generate_key_for_new_rotation(rotation_id) + { + Err(err) => { + error!("failed to generate and store new sphinx key: {err}"); + return; + } + Ok(key) => key, + }; + + 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}") + } + } + KeyRotationActionState::PurgeOld => {} + } } pub(crate) async fn run(&self) { @@ -41,17 +168,27 @@ impl KeyRotationController { let mut polling_interval = interval(self.regular_polling_interval); polling_interval.reset(); + let mut next_action = self.determine_next_action().await; + let mut state_update_future = sleep(next_action.until_deadline()); + pin_mut!(state_update_future); + while !self.shutdown_token.is_cancelled() { tokio::select! { biased; _ = self.shutdown_token.cancelled() => { - trace!("KeyRotationController: Received shutdown"); + trace!("KeyRotationController: Received shutdown"); + break; } - _ = polling_interval.tick() => { - self.regular_poll().await; + _ = polling_interval.tick() => {} + _ = &mut state_update_future => { + self.execute_next_action(next_action.typ).await } - // TODO: } + + next_action = self.determine_next_action().await; + state_update_future + .as_mut() + .reset(Instant::now() + next_action.until_deadline()); } trace!("KeyRotationController: exiting") diff --git a/nym-node/src/node/key_rotation/manager.rs b/nym-node/src/node/key_rotation/manager.rs index 031de7d06f..6bfbed9af6 100644 --- a/nym-node/src/node/key_rotation/manager.rs +++ b/nym-node/src/node/key_rotation/manager.rs @@ -4,7 +4,7 @@ use crate::error::{KeyIOFailure, NymNodeError}; use crate::node::helpers::{load_key, store_key}; use crate::node::key_rotation::active_keys::ActiveSphinxKeys; -use crate::node::key_rotation::key::SphinxPrivateKey; +use crate::node::key_rotation::key::{SphinxPrivateKey, SphinxPublicKey}; use rand::rngs::OsRng; use rand::{CryptoRng, RngCore}; use std::fs; @@ -43,14 +43,17 @@ impl SphinxKeyManager { }) } - fn replace_key_files>( + // moves the primary key to the secondary file + // and vice verse, i.e. secondary to the primary + fn swap_key_files>( primary_path: P, secondary_path: P, ) -> Result<(), NymNodeError> { let tmp_path = primary_path.as_ref().with_extension("tmp"); - fs::rename(primary_path.as_ref(), secondary_path.as_ref()).map_err(|err| { - KeyIOFailure::KeyMoveFailure { + // 1. COPY: primary -> temp + fs::copy(primary_path.as_ref(), secondary_path.as_ref()).map_err(|err| { + KeyIOFailure::KeyCopyFailure { key: "old x25519 sphinx primary".to_string(), source: primary_path.as_ref().to_path_buf(), destination: secondary_path.as_ref().to_path_buf(), @@ -58,34 +61,72 @@ impl SphinxKeyManager { } })?; - fs::rename(&tmp_path, primary_path.as_ref()).map_err(|err| { + // 2. MOVE: secondary -> primary + fs::rename(secondary_path.as_ref(), primary_path.as_ref()).map_err(|err| { KeyIOFailure::KeyMoveFailure { - key: "new x25519 sphinx primary".to_string(), - source: tmp_path, + key: "x25519 sphinx secondary".to_string(), + source: secondary_path.as_ref().to_path_buf(), destination: primary_path.as_ref().to_path_buf(), err, } })?; + + // 3. MOVE temp -> secondary + fs::rename(&tmp_path, secondary_path.as_ref()).map_err(|err| { + KeyIOFailure::KeyMoveFailure { + key: "old x25519 sphinx primary".to_string(), + source: tmp_path.clone(), + destination: primary_path.as_ref().to_path_buf(), + err, + } + })?; + + // 4. REMOVE: temp + fs::remove_file(&tmp_path).map_err(|err| KeyIOFailure::KeyRemovalFailure { + key: "old x25519 sphinx primary (temp location)".to_string(), + path: tmp_path, + err, + })?; Ok(()) } - // 1. generate new key - // 2. save it in a temp file - // 3. move primary key file to the secondary file location (thus losing the secondary) - // 4. move the temp file to the primary file location - // 5. set primary as the secondary - // 6. set new key as the primary - // 7. (outside this method) broadcast update to nym-apis - pub(crate) fn rotate_keys(&mut self, current_rotation_id: u32) -> Result<(), NymNodeError> { + pub(crate) fn generate_key_for_new_rotation( + &self, + expected_rotation: u32, + ) -> Result { let mut rng = OsRng; - let new_primary = SphinxPrivateKey::new(&mut rng, current_rotation_id); + let new = SphinxPrivateKey::new(&mut rng, expected_rotation); + let pub_key = (&new).into(); + store_key( + &new, + &self.secondary_key_path, + "x22519 (pre-announced) sphinx", + )?; - let tmp_path = self.primary_key_path.with_extension("tmp"); - store_key(&new_primary, &tmp_path, "x25519 sphinx")?; + self.keys.set_secondary(new); + Ok(pub_key) + } - Self::replace_key_files(&self.primary_key_path, &self.secondary_key_path)?; + pub(crate) fn rotate_keys(&self) -> Result<(), NymNodeError> { + if !self.keys.rotate() { + // we failed to perform the rotation because the secondary key somehow didn't exist + // we can't do much here, but just generate a brand-new key to rotate into + let primary = self.keys.primary().rotation_id(); + self.generate_key_for_new_rotation(primary + 1)?; + self.keys.rotate(); + } + Self::swap_key_files(&self.primary_key_path, &self.secondary_key_path) + } - self.keys.rotate(new_primary); + pub(crate) fn remove_overlap_key(&self) -> Result<(), NymNodeError> { + self.keys.deactivate_secondary(); + fs::remove_file(&self.secondary_key_path).map_err(|err| { + KeyIOFailure::KeyRemovalFailure { + key: "old x25519 sphinx secondary".to_string(), + path: self.secondary_key_path.clone(), + err, + } + })?; Ok(()) } @@ -94,49 +135,51 @@ impl SphinxKeyManager { primary_key_path: P, secondary_key_path: P, ) -> Result { - // check the temporary location in case we crashed in the middle of rotating the key - let tmp_location = primary_key_path.as_ref().with_extension("tmp"); - if tmp_location.exists() { - warn!("we seem to have crashed in the middle of rotating the sphinx key"); - // if temporary key exists, it means it has never overwritten the primary; - // secondary key might or might have not gotten overwritten, but that doesn't matter, - // we can do it again - Self::replace_key_files(primary_key_path.as_ref(), secondary_key_path.as_ref())?; - } + todo!("check if primary and secondary are correct - we might have crashed during the file swap"); - // primary key should always be present - let primary: SphinxPrivateKey = - load_key(primary_key_path.as_ref(), "x25519 sphinx primary")?; - - // if upon loading it turns out that the node has been inactive for a long time, - // immediately rotate keys (but leave 1h grace period for current primary, i.e. set it as secondary) - if primary.rotation_id() != current_rotation_id { - warn!("this node has been inactive for more than a key rotation duration. the current primary key was generated for rotation {} while the current rotation is {current_rotation_id}. new key will be generated now.", primary.rotation_id()); - let mut this = SphinxKeyManager { - keys: ActiveSphinxKeys::new_loaded(primary, None), - primary_key_path: primary_key_path.as_ref().to_path_buf(), - secondary_key_path: secondary_key_path.as_ref().to_path_buf(), - }; - this.rotate_keys(current_rotation_id)?; - return Ok(this); - } - - // secondary key **might** be present - let secondary_path = secondary_key_path.as_ref(); - - let secondary = if secondary_path.exists() { - Some(load_key::( - secondary_key_path.as_ref(), - "x25519 sphinx secondary", - )?) - } else { - None - }; - - Ok(SphinxKeyManager { - keys: ActiveSphinxKeys::new_loaded(primary, secondary), - primary_key_path: primary_key_path.as_ref().to_path_buf(), - secondary_key_path: secondary_key_path.as_ref().to_path_buf(), - }) + // // check the temporary location in case we crashed in the middle of rotating the key + // let tmp_location = primary_key_path.as_ref().with_extension("tmp"); + // if tmp_location.exists() { + // warn!("we seem to have crashed in the middle of rotating the sphinx key"); + // // if temporary key exists, it means it has never overwritten the primary; + // // secondary key might or might have not gotten overwritten, but that doesn't matter, + // // we can do it again + // Self:swape_key_files(primary_key_path.as_ref(), secondary_key_path.as_ref())?; + // } + // + // // primary key should always be present + // let primary: SphinxPrivateKey = + // load_key(primary_key_path.as_ref(), "x25519 sphinx primary")?; + // + // // if upon loading it turns out that the node has been inactive for a long time, + // // immediately rotate keys (but leave 1h grace period for current primary, i.e. set it as secondary) + // if primary.rotation_id() != current_rotation_id { + // warn!("this node has been inactive for more than a key rotation duration. the current primary key was generated for rotation {} while the current rotation is {current_rotation_id}. new key will be generated now.", primary.rotation_id()); + // let mut this = SphinxKeyManager { + // keys: ActiveSphinxKeys::new_loaded(primary, None), + // primary_key_path: primary_key_path.as_ref().to_path_buf(), + // secondary_key_path: secondary_key_path.as_ref().to_path_buf(), + // }; + // this.rotate_keys(current_rotation_id)?; + // return Ok(this); + // } + // + // // secondary key **might** be present + // let secondary_path = secondary_key_path.as_ref(); + // + // let secondary = if secondary_path.exists() { + // Some(load_key::( + // secondary_key_path.as_ref(), + // "x25519 sphinx secondary", + // )?) + // } else { + // None + // }; + // + // Ok(SphinxKeyManager { + // keys: ActiveSphinxKeys::new_loaded(primary, secondary), + // primary_key_path: primary_key_path.as_ref().to_path_buf(), + // secondary_key_path: secondary_key_path.as_ref().to_path_buf(), + // }) } } diff --git a/nym-node/src/node/mixnet/handler.rs b/nym-node/src/node/mixnet/handler.rs index 15a3f172c4..b355f097c4 100644 --- a/nym-node/src/node/mixnet/handler.rs +++ b/nym-node/src/node/mixnet/handler.rs @@ -8,10 +8,11 @@ use nym_sphinx_framing::codec::NymCodec; use nym_sphinx_framing::packet::FramedNymPacket; use nym_sphinx_framing::processing::{ process_framed_packet, MixProcessingResult, MixProcessingResultData, PacketProcessingError, - PartiallyUnwrappedPacket, ProcessedFinalHop, + PartiallyUnwrappedPacket, PartialyUnwrappedPacketWithKeyRotation, ProcessedFinalHop, }; use nym_sphinx_params::SphinxKeyRotation; use nym_sphinx_types::{Delay, REPLAY_TAG_SIZE}; +use std::collections::HashMap; use std::mem; use std::net::SocketAddr; use tokio::net::TcpStream; @@ -20,41 +21,50 @@ use tokio_util::codec::Framed; use tracing::{debug, error, instrument, trace, warn}; struct PendingReplayCheckPackets { - packets: Vec, + // map of rotation id used for packet creation to the packets + packets: HashMap>, last_acquired_mutex: Instant, } impl PendingReplayCheckPackets { fn new() -> PendingReplayCheckPackets { PendingReplayCheckPackets { - packets: vec![], + packets: Default::default(), last_acquired_mutex: Instant::now(), } } - fn reset(&mut self, now: Instant) -> Vec { + fn reset(&mut self, now: Instant) -> HashMap> { self.last_acquired_mutex = now; mem::take(&mut self.packets) } - fn push(&mut self, now: Instant, packet: PartiallyUnwrappedPacket) { + fn push(&mut self, now: Instant, packet: PartialyUnwrappedPacketWithKeyRotation) { if self.packets.is_empty() { self.last_acquired_mutex = now; } - self.packets.push(packet); + self.packets + .entry(packet.used_key_rotation) + .or_default() + .push(packet.packet) } - fn replay_tags(&self) -> Vec<&[u8; REPLAY_TAG_SIZE]> { - let mut replay_tags = Vec::with_capacity(self.packets.len()); - for packet in &self.packets { - let Some(replay_tag) = packet.replay_tag() else { - error!( - "corrupted batch of {} packets - replay tag was missing", - self.packets.len() - ); - return Vec::new(); - }; - replay_tags.push(replay_tag); + fn replay_tags(&self) -> HashMap> { + let mut replay_tags = HashMap::with_capacity(self.packets.len()); + 'outer: for (rotation_id, packets) in &self.packets { + let mut rotation_replay_tags = Vec::with_capacity(packets.len()); + for packet in packets { + let Some(replay_tag) = packet.replay_tag() else { + error!( + "corrupted batch of {} packets - replay tag was missing", + self.packets.len() + ); + replay_tags.insert(*rotation_id, Vec::new()); + continue 'outer; + }; + rotation_replay_tags.push(replay_tag); + } + replay_tags.insert(*rotation_id, rotation_replay_tags); } replay_tags } @@ -216,22 +226,26 @@ impl ConnectionHandler { fn try_partially_unwrap_packet( &self, packet: FramedNymPacket, - ) -> Result { + ) -> Result { // based on the received sphinx key rotation information, // attempt to choose appropriate key for processing the packet match packet.header().key_rotation { SphinxKeyRotation::Unknown => { + let primary = self.shared.sphinx_keys.primary(); + let primary_rotation = primary.rotation_id(); + // we have to try both keys, start with the primary as it has higher likelihood of being correct // if let Ok(partially_unwrapped) = PartiallyUnwrappedPacket::new() - match PartiallyUnwrappedPacket::new( - packet, - self.shared.sphinx_keys.primary().inner().as_ref(), - ) { - Ok(unwrapped_packet) => Ok(unwrapped_packet), + match PartiallyUnwrappedPacket::new(packet, primary.inner().as_ref()) { + Ok(unwrapped_packet) => { + Ok(unwrapped_packet.with_key_rotation(primary_rotation)) + } Err((packet, err)) => { if let Some(secondary) = self.shared.sphinx_keys.secondary() { + let secondary_rotation = secondary.rotation_id(); PartiallyUnwrappedPacket::new(packet, secondary.inner().as_ref()) .map_err(|(_, err)| err) + .map(|p| p.with_key_rotation(secondary_rotation)) } else { Err(err) } @@ -242,15 +256,19 @@ impl ConnectionHandler { let Some(odd_key) = self.shared.sphinx_keys.odd() else { return Err(PacketProcessingError::ExpiredKey); }; + let odd_rotation = odd_key.rotation_id(); PartiallyUnwrappedPacket::new(packet, odd_key.inner().as_ref()) .map_err(|(_, err)| err) + .map(|p| p.with_key_rotation(odd_rotation)) } SphinxKeyRotation::EvenRotation => { let Some(even_key) = self.shared.sphinx_keys.even() else { return Err(PacketProcessingError::ExpiredKey); }; + let even_rotation = even_key.rotation_id(); PartiallyUnwrappedPacket::new(packet, even_key.inner().as_ref()) .map_err(|(_, err)| err) + .map(|p| p.with_key_rotation(even_rotation)) } } } @@ -317,17 +335,24 @@ impl ConnectionHandler { async fn handle_post_replay_detection_packets( &self, now: Instant, - packets: Vec, - replay_check_results: Vec, + packets: HashMap>, + replay_check_results: HashMap>, ) { - for (packet, replayed) in packets.into_iter().zip(replay_check_results) { - let unwrapped_packet = if replayed { - Err(PacketProcessingError::PacketReplay) - } else { - packet.finalise_unwrapping() + for (rotation_id, packets) in packets { + let Some(replay_checks) = replay_check_results.get(&rotation_id) else { + // this should never happen, but if we messed up, and it does, don't panic, just drop the packets + error!("inconsistent replay check result - no values for rotation {rotation_id}"); + continue; }; + for (packet, &replayed) in packets.into_iter().zip(replay_checks) { + let unwrapped_packet = if replayed { + Err(PacketProcessingError::PacketReplay) + } else { + packet.finalise_unwrapping() + }; - self.handle_unwrapped_packet(now, unwrapped_packet).await; + self.handle_unwrapped_packet(now, unwrapped_packet).await; + } } } diff --git a/nym-node/src/node/mixnet/shared/mod.rs b/nym-node/src/node/mixnet/shared/mod.rs index bd6e815f44..37318505a7 100644 --- a/nym-node/src/node/mixnet/shared/mod.rs +++ b/nym-node/src/node/mixnet/shared/mod.rs @@ -5,7 +5,7 @@ use crate::config::Config; use crate::node::key_rotation::active_keys::ActiveSphinxKeys; use crate::node::mixnet::handler::ConnectionHandler; use crate::node::mixnet::SharedFinalHopData; -use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter; +use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; use nym_gateway::node::GatewayStorageError; use nym_mixnet_client::forwarder::{MixForwardingSender, PacketToForward}; use nym_node_metrics::mixnet::PacketKind; @@ -66,7 +66,7 @@ impl ProcessingConfig { pub(crate) struct SharedData { pub(super) processing_config: ProcessingConfig, pub(super) sphinx_keys: ActiveSphinxKeys, - pub(super) replay_protection_filter: ReplayProtectionBloomfilter, + pub(super) replay_protection_filter: ReplayProtectionBloomfilters, // used for FORWARD mix packets and FINAL ack packets pub(super) mixnet_forwarder: MixForwardingSender, @@ -89,7 +89,7 @@ impl SharedData { pub(crate) fn new( processing_config: ProcessingConfig, sphinx_keys: ActiveSphinxKeys, - replay_protection_filter: ReplayProtectionBloomfilter, + replay_protection_filter: ReplayProtectionBloomfilters, mixnet_forwarder: MixForwardingSender, final_hop: SharedFinalHopData, metrics: NymNodeMetrics, diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index e438728736..96e60c7702 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -31,7 +31,7 @@ use crate::node::mixnet::shared::ProcessingConfig; use crate::node::mixnet::SharedFinalHopData; use crate::node::nym_apis_client::NymApisClient; use crate::node::replay_protection::background_task::ReplayProtectionBackgroundTask; -use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter; +use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; use crate::node::routing_filter::{OpenFilter, RoutingFilter}; use crate::node::shared_network::{ CachedNetwork, CachedTopologyProvider, LocalGatewayNode, NetworkRefresher, @@ -956,9 +956,9 @@ impl NymNode { pub(crate) async fn setup_replay_detection( &self, - ) -> Result { + ) -> Result { if self.config.mixnet.replay_protection.debug.unsafe_disabled { - return Ok(ReplayProtectionBloomfilter::new_disabled()); + return Ok(ReplayProtectionBloomfilters::new_disabled()); } // create the background task for the bloomfilter diff --git a/nym-node/src/node/nym_apis_client.rs b/nym-node/src/node/nym_apis_client.rs index 61794033de..7acbc64441 100644 --- a/nym-node/src/node/nym_apis_client.rs +++ b/nym-node/src/node/nym_apis_client.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::NymNodeError; +use crate::node::key_rotation::key::SphinxPublicKey; use crate::node::NymNode; use futures::{stream, StreamExt}; use nym_crypto::asymmetric::ed25519; @@ -55,17 +56,17 @@ impl NymApisClient { }) } - async fn use_next_endpoint(&mut self) { - let mut guard = self.inner.write().await; - if guard.available_urls.len() == 1 { - return; - } - - let next_index = (guard.currently_used_api + 1) % guard.available_urls.len(); - let next = guard.available_urls[next_index].clone(); - guard.currently_used_api = next_index; - guard.active_client.change_nym_api(next) - } + // async fn use_next_endpoint(&self) { + // let mut guard = self.inner.write().await; + // if guard.available_urls.len() == 1 { + // return; + // } + // + // let next_index = (guard.currently_used_api + 1) % guard.available_urls.len(); + // let next = guard.available_urls[next_index].clone(); + // guard.currently_used_api = next_index; + // guard.active_client.change_nym_api(next) + // } pub(crate) async fn query_exhaustively( &self, @@ -75,11 +76,19 @@ impl NymApisClient { where R: AsyncFn(Client) -> Result, { - self.inner - .read() - .await - .query_exhaustively(req, timeout_duration) - .await + let guard = self.inner.read().await; + let (res, last_working_endpoint) = guard.query_exhaustively(req, timeout_duration).await?; + + // if we had to use a different api, update our starting point for the future calls + if guard.currently_used_api != last_working_endpoint { + drop(guard); + let mut guard = self.inner.write().await; + let next_url = guard.available_urls[last_working_endpoint].clone(); + guard.currently_used_api = last_working_endpoint; + guard.active_client.change_nym_api(next_url); + } + + Ok(res) } pub(crate) async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) { @@ -90,8 +99,12 @@ impl NymApisClient { .await; } - pub(crate) async fn broadcast_key_rotation(&self) { - self.inner.read().await.broadcast_key_rotation().await; + pub(crate) async fn broadcast_pre_announced_key(&self, public_key: SphinxPublicKey) { + self.inner + .read() + .await + .broadcast_pre_announced_key(public_key) + .await; } } @@ -117,20 +130,30 @@ impl InnerClient { } } - pub(crate) async fn query_exhaustively( + async fn query_exhaustively( &self, req: R, timeout_duration: Duration, - ) -> Result + ) -> Result<(T, usize), NymNodeError> where R: AsyncFn(Client) -> Result, { - // this is DESIGNED to query sequentially (but exhaustively) and not to try to send queries to ALL apis at once + let last_working = self.currently_used_api; + + // start from the last working api and progress from there + // also, note this is DESIGNED to query sequentially (but exhaustively) + // and not to try to send queries to ALL apis at once // and check which resolves first - for url in &self.available_urls { + for (idx, url) in self + .available_urls + .iter() + .enumerate() + .skip(last_working) + .chain(self.available_urls.iter().enumerate().take(last_working)) + { let nym_api = self.active_client.nym_api.clone_with_new_url(url.clone()); match timeout(timeout_duration, req(nym_api)).await { - Ok(Ok(res)) => return Ok(res), + Ok(Ok(res)) => return Ok((res, idx)), Ok(Err(err)) => { warn!("failed to resolve query for {url}: {err}") } @@ -143,7 +166,7 @@ impl InnerClient { Err(NymNodeError::NymApisExhausted) } - pub(crate) async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) { + async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) { let request = NodeRefreshBody::new(private_key); self.broadcast( @@ -154,7 +177,7 @@ impl InnerClient { .await; } - pub(crate) async fn broadcast_key_rotation(&self) { + async fn broadcast_pre_announced_key(&self, public_key: SphinxPublicKey) { todo!() } } diff --git a/nym-node/src/node/replay_protection/background_task.rs b/nym-node/src/node/replay_protection/background_task.rs index 37b5463a5f..7ff939f0cc 100644 --- a/nym-node/src/node/replay_protection/background_task.rs +++ b/nym-node/src/node/replay_protection/background_task.rs @@ -3,7 +3,7 @@ use crate::config::Config; use crate::error::NymNodeError; -use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter; +use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; use crate::node::replay_protection::items_in_bloomfilter; use human_repr::HumanCount; use nym_node_metrics::NymNodeMetrics; @@ -72,7 +72,7 @@ pub struct ReplayProtectionBackgroundTask { config: ReplayProtectionBackgroundTaskConfig, last_reset: LastResetData, - filter: ReplayProtectionBloomfilter, + filter: ReplayProtectionBloomfilters, metrics: NymNodeMetrics, shutdown_token: ShutdownToken, } @@ -99,7 +99,7 @@ impl ReplayProtectionBackgroundTask { // if there's nothing on disk, we must create a new filter let bloomfilter = if task_config.current_bloomfilter_path.exists() { - ReplayProtectionBloomfilter::load(&task_config.current_bloomfilter_path).await? + ReplayProtectionBloomfilters::load(&task_config.current_bloomfilter_path).await? } else { let bf_items = items_in_bloomfilter( task_config.filter_reset_rate, @@ -110,7 +110,7 @@ impl ReplayProtectionBackgroundTask { .initial_expected_packets_per_second, ); - ReplayProtectionBloomfilter::new_empty(bf_items, task_config.false_positive_rate)? + ReplayProtectionBloomfilters::new_empty(bf_items, task_config.false_positive_rate)? }; Ok(ReplayProtectionBackgroundTask { @@ -125,7 +125,7 @@ impl ReplayProtectionBackgroundTask { }) } - pub(crate) fn global_bloomfilter(&self) -> ReplayProtectionBloomfilter { + pub(crate) fn global_bloomfilter(&self) -> ReplayProtectionBloomfilters { self.filter.clone() } diff --git a/nym-node/src/node/replay_protection/bloomfilter.rs b/nym-node/src/node/replay_protection/bloomfilter.rs index 087b7b5444..0391ebbdbc 100644 --- a/nym-node/src/node/replay_protection/bloomfilter.rs +++ b/nym-node/src/node/replay_protection/bloomfilter.rs @@ -5,6 +5,7 @@ use crate::error::NymNodeError; use bloomfilter::Bloom; use human_repr::HumanDuration; use nym_sphinx_types::REPLAY_TAG_SIZE; +use std::collections::HashMap; use std::path::Path; use std::sync::{Arc, PoisonError, TryLockError}; use tokio::fs::File; @@ -12,34 +13,58 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::time::Instant; use tracing::{debug, 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 { + // 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, + + /// 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, + + rotation_id: u32, +} + // it appears that now std Mutex is faster (or comparable) to parking_lot // in high contention situations: https://github.com/rust-lang/rust/pull/95035#issuecomment-1073966631 // (tokio's async Mutex has too much overhead due to the number of access required) #[derive(Clone)] -pub(crate) struct ReplayProtectionBloomfilter { +pub(crate) struct ReplayProtectionBloomfilters { disabled: bool, - inner: Arc>, + inner: Arc>, } -impl ReplayProtectionBloomfilter { +impl ReplayProtectionBloomfilters { pub(crate) fn new_empty(items_count: usize, fp_p: f64) -> Result { - Ok(ReplayProtectionBloomfilter { - disabled: false, - inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { - current_filter: Bloom::new_for_fp_rate(items_count, fp_p) - .map_err(NymNodeError::bloomfilter_failure)?, - })), - }) + todo!() + // Ok(ReplayProtectionBloomfilter { + // disabled: false, + // inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { + // current_filter: Bloom::new_for_fp_rate(items_count, fp_p) + // .map_err(NymNodeError::bloomfilter_failure)?, + // })), + // }) } // SAFETY: the hardcoded values of 1,1 are valid #[allow(clippy::unwrap_used)] pub(crate) fn new_disabled() -> Self { // well, technically it's not fully empty, but the memory footprint is negligible - ReplayProtectionBloomfilter { + ReplayProtectionBloomfilters { disabled: true, - inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { - current_filter: Bloom::new(1, 1).unwrap(), + inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfiltersInner { + primary: RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata { + creation_time: Instant::now(), + packets_received_at_creation: 0, + rotation_id: u32::MAX, + }, + data: Bloom::new(1, 1).unwrap(), + }, + secondary: None, + pre_announced: None, })), } } @@ -50,174 +75,188 @@ impl ReplayProtectionBloomfilter { pub(crate) fn reset(&self, items_count: usize, fp_p: f64) -> Result<(), NymNodeError> { // 1. build the new filter - let new_inner = ReplayProtectionBloomfilterInner { - current_filter: Bloom::new_for_fp_rate(items_count, fp_p) - .map_err(NymNodeError::bloomfilter_failure)?, - }; - - // 2. swap it - let mut guard = self - .inner - .lock() - .map_err(|_| NymNodeError::BloomfilterFailure { - message: "mutex got poisoned", - })?; - - *guard = new_inner; - Ok(()) + todo!() + // let new_inner = ReplayProtectionBloomfilterInner { + // current_filter: Bloom::new_for_fp_rate(items_count, fp_p) + // .map_err(NymNodeError::bloomfilter_failure)?, + // }; + // + // // 2. swap it + // let mut guard = self + // .inner + // .lock() + // .map_err(|_| NymNodeError::BloomfilterFailure { + // message: "mutex got poisoned", + // })?; + // + // *guard = new_inner; + // Ok(()) } // NOTE: with key rotations we'll have to check whether the file is still valid and which // key it corresponds to, but that's a future problem pub(crate) async fn load>(path: P) -> Result { - info!("attempting to load prior replay detection bloomfilter..."); - let path = path.as_ref(); - let mut file = - File::open(path) - .await - .map_err(|source| NymNodeError::BloomfilterIoFailure { - source, - path: path.to_path_buf(), - })?; - - let mut buf = Vec::new(); - file.read_to_end(&mut buf) - .await - .map_err(|source| NymNodeError::BloomfilterIoFailure { - source, - path: path.to_path_buf(), - })?; - - Ok(ReplayProtectionBloomfilter { - disabled: false, - inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { - current_filter: Bloom::from_bytes(buf) - .map_err(NymNodeError::bloomfilter_failure)?, - })), - }) + todo!() + // info!("attempting to load prior replay detection bloomfilter..."); + // let path = path.as_ref(); + // let mut file = + // File::open(path) + // .await + // .map_err(|source| NymNodeError::BloomfilterIoFailure { + // source, + // path: path.to_path_buf(), + // })?; + // + // let mut buf = Vec::new(); + // file.read_to_end(&mut buf) + // .await + // .map_err(|source| NymNodeError::BloomfilterIoFailure { + // source, + // path: path.to_path_buf(), + // })?; + // + // Ok(ReplayProtectionBloomfilter { + // disabled: false, + // inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { + // current_filter: Bloom::from_bytes(buf) + // .map_err(NymNodeError::bloomfilter_failure)?, + // })), + // }) } // average HDD has the write speed of ~80MB/s so a 2GB bloomfilter would take almost 30s to write... // and this function is explicitly async and using tokio's async operations, because otherwise // we'd have to go through the whole hassle of using spawn_blocking and awaiting that one instead pub(crate) async fn flush_to_disk>(&self, path: P) -> Result<(), NymNodeError> { - debug!("flushing replay protection bloomfilter to disk..."); - let start = Instant::now(); - let path = path.as_ref(); - - let mut file = - File::create(path) - .await - .map_err(|source| NymNodeError::BloomfilterIoFailure { - source, - path: path.to_path_buf(), - })?; - let data = self.bytes().map_err(|_| NymNodeError::BloomfilterFailure { - message: "mutex got poisoned", - })?; - file.write_all(&data) - .await - .map_err(|source| NymNodeError::BloomfilterIoFailure { - source, - path: path.to_path_buf(), - })?; - - let elapsed = start.elapsed(); - - info!( - "flushed replay protection bloomfilter to disk. it took: {}", - elapsed.human_duration() - ); - - Ok(()) + todo!() + // debug!("flushing replay protection bloomfilter to disk..."); + // let start = Instant::now(); + // let path = path.as_ref(); + // + // let mut file = + // File::create(path) + // .await + // .map_err(|source| NymNodeError::BloomfilterIoFailure { + // source, + // path: path.to_path_buf(), + // })?; + // let data = self.bytes().map_err(|_| NymNodeError::BloomfilterFailure { + // message: "mutex got poisoned", + // })?; + // file.write_all(&data) + // .await + // .map_err(|source| NymNodeError::BloomfilterIoFailure { + // source, + // path: path.to_path_buf(), + // })?; + // + // let elapsed = start.elapsed(); + // + // info!( + // "flushed replay protection bloomfilter to disk. it took: {}", + // elapsed.human_duration() + // ); + // + // Ok(()) } } -struct ReplayProtectionBloomfilterInner { - // metadata to do with epochs, etc. - current_filter: Bloom<[u8; REPLAY_TAG_SIZE]>, - // overlap_filter: bloomfilter::Bloom<[u8; REPLAY_TAG_SIZE]>, +struct RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata, + data: Bloom<[u8; REPLAY_TAG_SIZE]>, } -impl ReplayProtectionBloomfilter { - #[allow(dead_code)] - pub(crate) fn check_and_set( - &self, - replay_tag: &[u8; REPLAY_TAG_SIZE], - ) -> Result> { - let Ok(mut guard) = self.inner.lock() else { - return Err(PoisonError::new(())); - }; - - Ok(guard.current_filter.check_and_set(replay_tag)) - } - - #[allow(dead_code)] - pub(crate) fn try_check_and_set( - &self, - replay_tag: &[u8; REPLAY_TAG_SIZE], - ) -> Option>> { - let mut guard = match self.inner.try_lock() { - Ok(guard) => guard, - Err(TryLockError::Poisoned(_)) => return Some(Err(PoisonError::new(()))), - Err(TryLockError::WouldBlock) => return None, - }; - - Some(Ok(guard.current_filter.check_and_set(replay_tag))) - } - +impl ReplayProtectionBloomfilters { pub(crate) fn batch_try_check_and_set( &self, - reply_tags: &[&[u8; REPLAY_TAG_SIZE]], - ) -> Option, PoisonError<()>>> { + reply_tags: &HashMap>, + ) -> Option>, PoisonError<()>>> { let mut guard = match self.inner.try_lock() { Ok(guard) => guard, Err(TryLockError::Poisoned(_)) => return Some(Err(PoisonError::new(()))), Err(TryLockError::WouldBlock) => return None, }; - let mut result = Vec::with_capacity(reply_tags.len()); - for tag in reply_tags { - result.push(guard.current_filter.check_and_set(tag)); - } - - // for testing throughput without disabling checks: - // return Some(Ok(vec![false; reply_tags.len()])); - - Some(Ok(result)) + Some(Ok(guard.batch_check_and_set(&reply_tags))) } pub(crate) fn batch_check_and_set( &self, - reply_tags: &[&[u8; REPLAY_TAG_SIZE]], - ) -> Result, PoisonError<()>> { + reply_tags: &HashMap>, + ) -> Result>, PoisonError<()>> { let Ok(mut guard) = self.inner.lock() else { return Err(PoisonError::new(())); }; - let mut result = Vec::with_capacity(reply_tags.len()); - for tag in reply_tags { - result.push(guard.current_filter.check_and_set(tag)); - } - - // for testing throughput without disabling checks: - // return Ok(vec![false; reply_tags.len()]); - - Ok(result) - } - - #[allow(dead_code)] - pub(crate) fn clear(&self) -> Result<(), PoisonError<()>> { - let mut guard = self.inner.lock().map_err(|_| PoisonError::new(()))?; - guard.current_filter.clear(); - Ok(()) + Ok(guard.batch_check_and_set(&reply_tags)) } // due to the size of the bloomfilter, extra caution has to be applied when using this method // note: we're not getting reference to bytes as this method is used when flushing data to the disk // (which takes ~30s) and we can't block the mutex for that long. fn bytes(&self) -> Result, PoisonError<()>> { - let guard = self.inner.lock().map_err(|_| PoisonError::new(()))?; - Ok(guard.current_filter.to_bytes()) + todo!() + // let guard = self.inner.lock().map_err(|_| PoisonError::new(()))?; + // Ok(guard.current_filter.to_bytes()) + } +} + +struct ReplayProtectionBloomfiltersInner { + primary: RotationFilter, + + // don't worry, we'll never have 3 active filters at once, + // we will either have a secondary (during the first epoch of a new rotation) + // or a pre_announced (during the last epoch of the current rotation) + // during epoch transition, the following change will happen: + // primary -> secondary + // pre_announced -> primary + // I'm not using an enum because it's easier to reason about those as separate fields + secondary: Option, + pre_announced: Option, +} + +impl ReplayProtectionBloomfiltersInner { + fn batch_check_and_set( + &mut self, + reply_tags: &HashMap>, + ) -> HashMap> { + let mut result = HashMap::with_capacity(reply_tags.len()); + for (&rotation_id, reply_tags) in reply_tags { + // try to 'find' the relevant filter. we might be doing 3 reads here, but realistically it's + // going to be 'primary' most of the time and even if not, it's just few ns of overhead... + let mut filter = if self.primary.metadata.rotation_id == rotation_id { + Some(&mut self.primary.data) + } else if let Some(secondary) = &mut self.secondary { + // if let chaining won't be stable until 1.88 so we have to do the Option workaround + if secondary.metadata.rotation_id == rotation_id { + Some(&mut secondary.data) + } else { + None + } + } else if let Some(pre_announced) = &mut self.pre_announced { + if pre_announced.metadata.rotation_id == rotation_id { + Some(&mut pre_announced.data) + } else { + None + } + } else { + None + }; + + let Some(mut filter) = filter else { + // if we've received a packet from an unknown rotation, it most likely means it has been replayed + // from an older rotation, so mark it as such + result.insert(rotation_id, vec![false; reply_tags.len()]); + continue; + }; + + let mut rotation_results = Vec::with_capacity(reply_tags.len()); + for tag in reply_tags { + rotation_results.push(filter.check_and_set(tag)) + } + result.insert(rotation_id, rotation_results); + } + + result } }