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

Feature/multi surbs invalidation (#1858)

Browse Source 
* Cleaned up RealMessagesController constructor

* introduced config field for maximum_reply_surb_age

* Handling edge-case reply-surb failures

* invalidating old reply surbs

* Removing old reply keys from cache

* Invalidating old reply keys

* missing config changes
This commit is contained in:
Jędrzej Stuczyński
2022-12-07 14:19:41 +00:00
committed by GitHub
parent 11fe5cfbf8
commit 2c4611044c
13 changed files with 515 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/fs_surbs_migrations/20221130120000_create_initial_tables.sql
+3 -2
View File
@@ -19,8 +19,9 @@ CREATE TABLE sender_tag
CREATE TABLE reply_key
(
key_digest BLOB NOT NULL UNIQUE,
reply_key BLOB NOT NULL UNIQUE
key_digest BLOB NOT NULL UNIQUE,
reply_key BLOB NOT NULL UNIQUE,
sent_at_timestamp INTEGER NOT NULL
);
CREATE TABLE reply_surb_sender
clients/client-core/src/client/real_messages_control/message_handler.rs
+86 -17
View File
@@ -16,9 +16,11 @@ use nymsphinx::anonymous_replies::requests::{AnonymousSenderTag, RepliableMessag
use nymsphinx::anonymous_replies::{ReplySurb, SurbEncryptionKey};
use nymsphinx::chunking::fragment::Fragment;
use nymsphinx::message::NymMessage;
use nymsphinx::params::{PacketSize, DEFAULT_NUM_MIX_HOPS};
use nymsphinx::preparer::{MessagePreparer, PreparedFragment};
use rand::{CryptoRng, Rng};
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
use topology::{NymTopology, NymTopologyError};
@@ -88,10 +90,63 @@ impl SurbWrappedPreparationError {
}
#[derive(Clone)]
pub(crate) struct MessageHandler<R> {
rng: R,
pub(crate) struct Config {
/// Key used to decrypt contents of received SURBAcks
ack_key: Arc<AckKey>,
self_address: Recipient,
/// Address of this client which also represent an address to which all acknowledgements
/// and surb-based are going to be sent.
sender_address: Recipient,
/// Average delay a data packet is going to get delay at a single mixnode.
average_packet_delay: Duration,
/// Average delay an acknowledgement packet is going to get delay at a single mixnode.
average_ack_delay: Duration,
/// Number of mix hops each packet ('real' message, ack, reply) is expected to take.
/// Note that it does not include gateway hops.
num_mix_hops: u8,
/// Predefined packet size used for the encapsulated messages.
packet_size: PacketSize,
}
impl Config {
pub fn new(
ack_key: Arc<AckKey>,
sender_address: Recipient,
average_packet_delay: Duration,
average_ack_delay: Duration,
) -> Self {
Config {
ack_key,
sender_address,
average_packet_delay,
average_ack_delay,
num_mix_hops: DEFAULT_NUM_MIX_HOPS,
packet_size: PacketSize::default(),
}
}
/// Allows setting non-default number of expected mix hops in the network.
#[allow(dead_code)]
pub fn with_mix_hops(mut self, hops: u8) -> Self {
self.num_mix_hops = hops;
self
}
/// Allows setting non-default size of the sphinx packets sent out.
pub fn with_custom_packet_size(mut self, packet_size: PacketSize) -> Self {
self.packet_size = packet_size;
self
}
}
#[derive(Clone)]
pub(crate) struct MessageHandler<R> {
config: Config,
rng: R,
message_preparer: MessagePreparer<R>,
action_sender: AckActionSender,
real_message_sender: BatchRealMessageSender,
@@ -104,22 +159,30 @@ impl<R> MessageHandler<R>
where
R: CryptoRng + Rng,
{
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
config: Config,
rng: R,
ack_key: Arc<AckKey>,
self_address: Recipient,
message_preparer: MessagePreparer<R>,
action_sender: AckActionSender,
real_message_sender: BatchRealMessageSender,
topology_access: TopologyAccessor,
reply_key_storage: SentReplyKeys,
tag_storage: UsedSenderTags,
) -> Self {
MessageHandler {
) -> Self
where
R: Copy,
{
let message_preparer = MessagePreparer::new(
rng,
config.sender_address,
config.average_packet_delay,
config.average_ack_delay,
)
.with_custom_real_message_packet_size(config.packet_size)
.with_mix_hops(config.num_mix_hops);
MessageHandler {
config,
rng,
ack_key,
self_address,
message_preparer,
action_sender,
real_message_sender,
@@ -145,7 +208,7 @@ where
&self,
permit: &'a TopologyReadPermit<'a>,
) -> Result<&'a NymTopology, PreparationError> {
match permit.try_get_valid_topology_ref(&self.self_address, None) {
match permit.try_get_valid_topology_ref(&self.config.sender_address, None) {
Ok(topology_ref) => Ok(topology_ref),
Err(err) => {
warn!("Could not process the packet - the network topology is invalid - {err}");
@@ -224,7 +287,8 @@ where
) -> Result<(), SurbWrappedPreparationError> {
debug!("requesting {amount} reply SURBs from {from:?}");
let surbs_request = ReplyMessage::new_surb_request_message(self.self_address, amount);
let surbs_request =
ReplyMessage::new_surb_request_message(self.config.sender_address, amount);
self.try_send_single_surb_message(from, surbs_request, reply_surb, true)
.await
}
@@ -267,7 +331,12 @@ where
let chunk_clone = fragment.clone();
let prepared_fragment = self
.message_preparer
.prepare_reply_chunk_for_sending(chunk_clone, topology, &self.ack_key, reply_surb)
.prepare_reply_chunk_for_sending(
chunk_clone,
topology,
&self.config.ack_key,
reply_surb,
)
.unwrap();
let real_message =
@@ -319,7 +388,7 @@ where
let prepared_fragment = self.message_preparer.prepare_chunk_for_sending(
chunk_clone,
topology,
&self.ack_key,
&self.config.ack_key,
&recipient,
)?;
@@ -399,7 +468,7 @@ where
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(chunk, topology, &self.ack_key, &recipient)
.prepare_chunk_for_sending(chunk, topology, &self.config.ack_key, &recipient)
.unwrap();
Ok(prepared_fragment)
@@ -418,7 +487,7 @@ where
let prepared_fragment = self
.message_preparer
.prepare_reply_chunk_for_sending(chunk, topology, &self.ack_key, reply_surb)
.prepare_reply_chunk_for_sending(chunk, topology, &self.config.ack_key, reply_surb)
.unwrap();
Ok(prepared_fragment)
clients/client-core/src/client/real_messages_control/mod.rs
+77 -41
View File
@@ -28,11 +28,11 @@ use log::*;
use nymsphinx::acknowledgements::AckKey;
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::params::PacketSize;
use nymsphinx::preparer::MessagePreparer;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::sync::Arc;
use std::time::Duration;
use crate::client::replies::reply_controller;
use crate::config;
pub(crate) use acknowledgement_control::{AckActionSender, Action};
@@ -85,6 +85,64 @@ pub struct Config {
/// Defines maximum amount of time the client is going to wait for reply surbs before explicitly asking
/// for more even though in theory they wouldn't need to.
maximum_reply_surb_waiting_period: Duration,
/// Defines maximum amount of time given reply surb is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
maximum_reply_surb_age: Duration,
/// Defines maximum amount of time given reply key is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
maximum_reply_key_age: Duration,
}
impl<'a> From<&'a Config> for acknowledgement_control::Config {
fn from(cfg: &'a Config) -> Self {
acknowledgement_control::Config::new(
cfg.ack_wait_addition,
cfg.ack_wait_multiplier,
cfg.retransmission_reply_surb_request_size,
)
.with_custom_packet_size(cfg.packet_size)
}
}
impl<'a> From<&'a Config> for real_traffic_stream::Config {
fn from(cfg: &'a Config) -> Self {
real_traffic_stream::Config::new(
Arc::clone(&cfg.ack_key),
cfg.self_recipient,
cfg.average_ack_delay_duration,
cfg.average_packet_delay_duration,
cfg.average_message_sending_delay,
cfg.disable_main_poisson_packet_distribution,
)
.with_custom_cover_packet_size(cfg.packet_size)
}
}
impl<'a> From<&'a Config> for reply_controller::Config {
fn from(cfg: &'a Config) -> Self {
reply_controller::Config::new(
cfg.minimum_reply_surb_request_size,
cfg.maximum_reply_surb_request_size,
cfg.maximum_allowed_reply_surb_request_size,
cfg.maximum_reply_surb_waiting_period,
cfg.maximum_reply_surb_age,
cfg.maximum_reply_key_age,
)
}
}
impl<'a> From<&'a Config> for message_handler::Config {
fn from(cfg: &'a Config) -> Self {
message_handler::Config::new(
Arc::clone(&cfg.ack_key),
cfg.self_recipient,
cfg.average_packet_delay_duration,
cfg.average_ack_delay_duration,
)
.with_custom_packet_size(cfg.packet_size)
}
}
impl Config {
@@ -114,6 +172,8 @@ impl Config {
.retransmission_reply_surb_request_size,
maximum_reply_surb_waiting_period: base_client_debug_config
.maximum_reply_surb_waiting_period,
maximum_reply_surb_age: base_client_debug_config.maximum_reply_surb_age,
maximum_reply_key_age: base_client_debug_config.maximum_reply_key_age,
}
}
@@ -150,10 +210,10 @@ impl RealMessagesController<OsRng> {
) -> Self {
let rng = OsRng;
// create channels for inter-task communication
let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(1);
let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded();
let (ack_action_tx, ack_action_rx) = mpsc::unbounded();
let ack_controller_connectors = AcknowledgementControllerConnectors::new(
input_receiver,
sent_notifier_rx,
@@ -162,26 +222,16 @@ impl RealMessagesController<OsRng> {
ack_action_rx,
);
let ack_control_config = acknowledgement_control::Config::new(
config.ack_wait_addition,
config.ack_wait_multiplier,
config.retransmission_reply_surb_request_size,
)
.with_custom_packet_size(config.packet_size);
// create all configs for the components
let ack_control_config = (&config).into();
let out_queue_config = (&config).into();
let reply_controller_config = (&config).into();
let message_handler_config = (&config).into();
// TODO: construct MessagePreparer itself inside the MessageHandler
let message_preparer = MessagePreparer::new(
rng,
config.self_recipient,
config.average_packet_delay_duration,
config.average_ack_delay_duration,
)
.with_custom_real_message_packet_size(config.packet_size);
// create the actual components
let message_handler = MessageHandler::new(
message_handler_config,
rng,
Arc::clone(&config.ack_key),
config.self_recipient,
message_preparer,
ack_action_tx,
real_message_sender,
topology_access.clone(),
@@ -189,42 +239,28 @@ impl RealMessagesController<OsRng> {
reply_storage.tags_storage(),
);
let reply_control = ReplyController::new(
message_handler.clone(),
reply_storage.surbs_storage(),
reply_storage.tags_storage(),
reply_controller_receiver,
config.minimum_reply_surb_request_size,
config.maximum_reply_surb_request_size,
config.maximum_allowed_reply_surb_request_size,
config.maximum_reply_surb_waiting_period,
);
let ack_control = AcknowledgementController::new(
ack_control_config,
Arc::clone(&config.ack_key),
ack_controller_connectors,
message_handler,
message_handler.clone(),
reply_controller_sender,
reply_storage.surbs_storage(),
);
let out_queue_config = real_traffic_stream::Config::new(
config.average_ack_delay_duration,
config.average_packet_delay_duration,
config.average_message_sending_delay,
config.disable_main_poisson_packet_distribution,
)
.with_custom_cover_packet_size(config.packet_size);
let reply_control = ReplyController::new(
reply_controller_config,
message_handler,
reply_storage,
reply_controller_receiver,
);
let out_queue_control = OutQueueControl::new(
out_queue_config,
config.ack_key,
rng,
sent_notifier_tx,
mix_sender,
real_message_receiver,
rng,
config.self_recipient,
topology_access,
lane_queue_lengths,
client_connection_rx,
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+15 -15
View File
@@ -48,6 +48,12 @@ fn get_time_now() -> wasm_timer::Instant {
/// Configurable parameters of the `OutQueueControl`
pub(crate) struct Config {
/// Key used to encrypt and decrypt content of an ACK packet.
ack_key: Arc<AckKey>,
/// Represents full address of this client.
our_full_destination: Recipient,
/// Average delay an acknowledgement packet is going to get delay at a single mixnode.
average_ack_delay: Duration,
@@ -67,12 +73,16 @@ pub(crate) struct Config {
impl Config {
pub(crate) fn new(
ack_key: Arc<AckKey>,
our_full_destination: Recipient,
average_ack_delay: Duration,
average_packet_delay: Duration,
average_message_sending_delay: Duration,
disable_poisson_packet_distribution: bool,
) -> Self {
Config {
ack_key,
our_full_destination,
average_ack_delay,
average_packet_delay,
average_message_sending_delay,
@@ -94,9 +104,6 @@ where
/// Configurable parameters of the `ActionController`
config: Config,
/// Key used to encrypt and decrypt content of an ACK packet.
ack_key: Arc<AckKey>,
/// Channel used for notifying of a real packet being sent out. Used to start up retransmission timer.
sent_notifier: SentPacketNotificationSender,
@@ -120,9 +127,6 @@ where
/// before being sent out into the network.
real_receiver: BatchRealMessageReceiver,
/// Represents full address of this client.
our_full_destination: Recipient,
/// Instance of a cryptographically secure random number generator.
rng: R,
@@ -186,25 +190,21 @@ where
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
config: Config,
ack_key: Arc<AckKey>,
rng: R,
sent_notifier: SentPacketNotificationSender,
mix_tx: BatchMixMessageSender,
real_receiver: BatchRealMessageReceiver,
rng: R,
our_full_destination: Recipient,
topology_access: TopologyAccessor,
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
) -> Self {
OutQueueControl {
config,
ack_key,
sent_notifier,
next_delay: None,
sending_delay_controller: Default::default(),
mix_tx,
real_receiver,
our_full_destination,
rng,
topology_access,
transmission_buffer: Default::default(),
@@ -232,8 +232,8 @@ where
let topology_permit = self.topology_access.get_read_permit().await;
// the ack is sent back to ourselves (and then ignored)
let topology_ref = match topology_permit.try_get_valid_topology_ref(
&self.our_full_destination,
Some(&self.our_full_destination),
&self.config.our_full_destination,
Some(&self.config.our_full_destination),
) {
Ok(topology) => topology,
Err(err) => {
@@ -246,8 +246,8 @@ where
generate_loop_cover_packet(
&mut self.rng,
topology_ref,
&self.ack_key,
&self.our_full_destination,
&self.config.ack_key,
&self.config.our_full_destination,
self.config.average_ack_delay,
self.config.average_packet_delay,
self.config.cover_packet_size,
clients/client-core/src/client/received_buffer.rs
+1 -1
View File
@@ -328,7 +328,7 @@ impl ReceivedMessagesBuffer {
.try_pop(possible_key_digest)
.map(|reply_encryption_key| {
(
reply_encryption_key,
*reply_encryption_key,
&mut raw_message[reply_surb_digest_size..],
)
})
clients/client-core/src/client/replies/reply_controller.rs
+223 -76
View File
@@ -2,11 +2,11 @@
// SPDX-License-Identifier: Apache-2.0
use crate::client::real_messages_control::message_handler::{MessageHandler, PreparationError};
use crate::client::replies::reply_storage::{ReceivedReplySurbsMap, UsedSenderTags};
use crate::client::replies::reply_storage::CombinedReplyStorage;
use client_connections::TransmissionLane;
use futures::channel::mpsc;
use futures::StreamExt;
use log::{debug, trace, warn};
use log::{debug, error, info, trace, warn};
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::anonymous_replies::requests::AnonymousSenderTag;
use nymsphinx::anonymous_replies::ReplySurb;
@@ -15,12 +15,13 @@ use rand::{CryptoRng, Rng};
use std::cmp::{max, min};
use std::collections::{HashMap, VecDeque};
use std::time::Duration;
use time::OffsetDateTime;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::Instant;
type IntervalStream = tokio_stream::wrappers::IntervalStream;
#[cfg(target_arch = "wasm32")]
use wasm_timer::Instant;
type IntervalStream = gloo_timers::future::IntervalStream;
pub fn new_control_channels() -> (ReplyControllerSender, ReplyControllerReceiver) {
let (tx, rx) = mpsc::unbounded();
@@ -101,6 +102,35 @@ pub enum ReplyControllerMessage {
},
}
pub struct Config {
min_surb_request_size: u32,
max_surb_request_size: u32,
maximum_allowed_reply_surb_request_size: u32,
max_surb_waiting_period: Duration,
max_reply_surb_age: Duration,
max_reply_key_age: Duration,
}
impl Config {
pub(crate) fn new(
min_surb_request_size: u32,
max_surb_request_size: u32,
maximum_allowed_reply_surb_request_size: u32,
max_surb_waiting_period: Duration,
max_reply_surb_age: Duration,
max_reply_key_age: Duration,
) -> Self {
Self {
min_surb_request_size,
max_surb_request_size,
maximum_allowed_reply_surb_request_size,
max_surb_waiting_period,
max_reply_surb_age,
max_reply_key_age,
}
}
}
// the purpose of this task:
// - buffers split messages from input message listener if there were insufficient surbs to send them
// - upon getting extra surbs, resends them
@@ -111,6 +141,8 @@ pub enum ReplyControllerMessage {
// TODO: this should be split into ingress and egress controllers
// because currently its trying to perform two distinct jobs
pub struct ReplyController<R> {
config: Config,
// TODO: incorporate that field at some point
// and use binomial distribution to determine the expected required number
// of surbs required to send the message through
@@ -118,44 +150,33 @@ pub struct ReplyController<R> {
request_receiver: ReplyControllerReceiver,
pending_replies: HashMap<AnonymousSenderTag, VecDeque<Fragment>>,
message_handler: MessageHandler<R>,
received_reply_surbs: ReceivedReplySurbsMap,
tag_storage: UsedSenderTags,
min_surb_request_size: u32,
max_surb_request_size: u32,
maximum_allowed_reply_surb_request_size: u32,
max_surb_waiting_period: Duration,
full_reply_storage: CombinedReplyStorage,
}
impl<R> ReplyController<R>
where
R: CryptoRng + Rng,
{
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
config: Config,
message_handler: MessageHandler<R>,
received_reply_surbs: ReceivedReplySurbsMap,
tag_storage: UsedSenderTags,
full_reply_storage: CombinedReplyStorage,
request_receiver: ReplyControllerReceiver,
min_surb_request_size: u32,
max_surb_request_size: u32,
maximum_allowed_reply_surb_request_size: u32,
max_surb_waiting_period: Duration,
) -> Self {
ReplyController {
config,
request_receiver,
pending_replies: Default::default(),
pending_replies: HashMap::new(),
message_handler,
received_reply_surbs,
tag_storage,
min_surb_request_size,
max_surb_request_size,
maximum_allowed_reply_surb_request_size,
max_surb_waiting_period,
full_reply_storage,
}
}
fn insert_pending_replies(&mut self, recipient: &AnonymousSenderTag, fragments: Vec<Fragment>) {
fn insert_pending_replies<V: Into<VecDeque<Fragment>>>(
&mut self,
recipient: &AnonymousSenderTag,
fragments: V,
) {
if let Some(existing) = self.pending_replies.get_mut(recipient) {
existing.append(&mut fragments.into())
} else {
@@ -173,10 +194,22 @@ where
None => return false,
};
let available_surbs = self.received_reply_surbs.available_surbs(target);
let pending_surbs = self.received_reply_surbs.pending_reception(target) as usize;
let min_surbs_threshold = self.received_reply_surbs.min_surb_threshold();
let max_surbs_threshold = self.received_reply_surbs.max_surb_threshold();
let available_surbs = self
.full_reply_storage
.surbs_storage_ref()
.available_surbs(target);
let pending_surbs = self
.full_reply_storage
.surbs_storage_ref()
.pending_reception(target) as usize;
let min_surbs_threshold = self
.full_reply_storage
.surbs_storage_ref()
.min_surb_threshold();
let max_surbs_threshold = self
.full_reply_storage
.surbs_storage_ref()
.max_surb_threshold();
debug!("queue size: {queue_size}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..{max_surbs_threshold}");
@@ -190,7 +223,11 @@ where
data: Vec<u8>,
lane: TransmissionLane,
) {
if !self.received_reply_surbs.contains_surbs_for(&recipient_tag) {
if !self
.full_reply_storage
.surbs_storage_ref()
.contains_surbs_for(&recipient_tag)
{
warn!("received reply request for {:?} but we don't have any surbs stored for that recipient!", recipient_tag);
return;
}
@@ -206,7 +243,8 @@ where
// (but at some point we run out of surbs for surb requests)
let (surbs, _surbs_left) = self
.received_reply_surbs
.full_reply_storage
.surbs_storage_ref()
.get_reply_surbs(&recipient_tag, required_surbs);
if let Some(reply_surbs) = surbs {
@@ -215,8 +253,10 @@ where
.try_send_reply_chunks(recipient_tag, fragments, reply_surbs, lane)
.await
{
// TODO: perhaps there should be some timer here to repeat the request once topology recovers
let err = err.return_unused_surbs(&self.received_reply_surbs, &recipient_tag);
let err = err.return_unused_surbs(
self.full_reply_storage.surbs_storage_ref(),
&recipient_tag,
);
warn!("failed to send reply to {:?} - {err}", recipient_tag);
}
} else {
@@ -236,7 +276,8 @@ where
amount: u32,
) -> Result<(), PreparationError> {
let reply_surb = self
.received_reply_surbs
.full_reply_storage
.surbs_storage_ref()
.get_reply_surb_ignoring_threshold(&target)
.and_then(|(reply_surb, _)| reply_surb)
.ok_or(PreparationError::NotEnoughSurbs {
@@ -249,15 +290,15 @@ where
.try_request_additional_reply_surbs(target, reply_surb, amount)
.await
{
let err = err.return_unused_surbs(&self.received_reply_surbs, &target);
let err = err.return_unused_surbs(self.full_reply_storage.surbs_storage_ref(), &target);
warn!(
"failed to request additional surbs from {:?} - {err}",
target
);
// TODO: perhaps there should be some timer here to repeat the request once topology recovers
return Err(err);
} else {
self.received_reply_surbs
self.full_reply_storage
.surbs_storage_ref()
.increment_pending_reception(&target, amount);
}
@@ -289,8 +330,14 @@ where
async fn try_clear_pending_queue(&mut self, target: AnonymousSenderTag) {
trace!("trying to clear pending queue");
let available_surbs = self.received_reply_surbs.available_surbs(&target);
let min_surbs_threshold = self.received_reply_surbs.min_surb_threshold();
let available_surbs = self
.full_reply_storage
.surbs_storage_ref()
.available_surbs(&target);
let min_surbs_threshold = self
.full_reply_storage
.surbs_storage_ref()
.min_surb_threshold();
let max_to_clear = if available_surbs > min_surbs_threshold {
available_surbs - min_surbs_threshold
@@ -302,8 +349,7 @@ where
// we're guaranteed to not get more entries than we have reply surbs for
if let Some(to_send) = self.pop_at_most_pending_replies(&target, max_to_clear) {
// TODO: optimise: we're cloning the fragments every time to re-insert them into the buffer in case of failure
let to_send_vec = to_send.into_iter().collect::<Vec<_>>();
let to_send_vec = to_send.iter().cloned().collect::<Vec<_>>();
if to_send_vec.is_empty() {
panic!(
@@ -312,10 +358,16 @@ where
}
let (surbs_for_reply, _) = self
.received_reply_surbs
.full_reply_storage
.surbs_storage_ref()
.get_reply_surbs(&target, to_send_vec.len());
let surbs_for_reply =
surbs_for_reply.expect("is is possible for this to ever show up?");
let Some(surbs_for_reply) = surbs_for_reply else {
// probably retransmission
debug!("somehow different task has stolen our reply surbs!");
self.insert_pending_replies(&target, to_send);
return
};
if let Err(err) = self
.message_handler
@@ -327,9 +379,10 @@ where
)
.await
{
let err = err.return_unused_surbs(&self.received_reply_surbs, &target);
let err =
err.return_unused_surbs(self.full_reply_storage.surbs_storage_ref(), &target);
self.insert_pending_replies(&target, to_send);
warn!("failed to clear pending queue for {:?} - {err}", target);
// TODO: perhaps there should be some timer here to repeat the request once topology recovers
}
} else {
trace!("the pending queue is empty");
@@ -345,15 +398,19 @@ where
trace!("handling received surbs");
// clear the requesting flag since we should have been asking for surbs
self.received_reply_surbs
self.full_reply_storage
.surbs_storage_ref()
.reset_surbs_last_received_at(&from);
if from_surb_request {
self.received_reply_surbs
self.full_reply_storage
.surbs_storage_ref()
.decrement_pending_reception(&from, reply_surbs.len() as u32);
}
// store received surbs
self.received_reply_surbs.insert_surbs(&from, reply_surbs);
self.full_reply_storage
.surbs_storage_ref()
.insert_surbs(&from, reply_surbs);
// use as many as we can for clearing pending queue
self.try_clear_pending_queue(from).await;
@@ -367,15 +424,19 @@ where
async fn handle_surb_request(&mut self, recipient: Recipient, mut amount: u32) {
// 1. check whether we sent any surbs in the past to this recipient, otherwise
// they have no business in asking for more
if !self.tag_storage.exists(&recipient) {
if !self
.full_reply_storage
.tags_storage_ref()
.exists(&recipient)
{
warn!("{recipient} asked us for reply SURBs even though we never sent them any anonymous messages before!");
return;
}
// 2. check whether the requested amount is within sane range
if amount > self.maximum_allowed_reply_surb_request_size {
warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.maximum_allowed_reply_surb_request_size);
amount = self.maximum_allowed_reply_surb_request_size;
if amount > self.config.maximum_allowed_reply_surb_request_size {
warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.config.maximum_allowed_reply_surb_request_size);
amount = self.config.maximum_allowed_reply_surb_request_size;
}
// 3. construct and send the surbs away
@@ -435,8 +496,8 @@ where
}
let request_size = min(
self.max_surb_request_size,
max(queue_size, self.min_surb_request_size),
self.config.max_surb_request_size,
max(queue_size, self.config.min_surb_request_size),
);
if let Err(err) = self
@@ -449,23 +510,33 @@ where
async fn inspect_stale_entries(&mut self) {
let mut to_request = Vec::new();
let mut to_remove = Vec::new();
let now = Instant::now();
let now = OffsetDateTime::now_utc();
for (pending_reply_target, vals) in &self.pending_replies {
if vals.is_empty() {
// TODO: remove it from the map before getting here
continue;
}
let last_received = self
.received_reply_surbs
.surbs_last_received_at(pending_reply_target)
.expect("I think this shouldnt fail? to be verified.");
let Some(last_received) = self
.full_reply_storage.surbs_storage_ref()
.surbs_last_received_at(pending_reply_target) else {
error!("we have {} pending replies for {pending_reply_target}, but we somehow never received any reply surbs from them!", vals.len());
to_remove.push(*pending_reply_target);
continue
};
let diff = now - last_received;
// this should never ever happen (famous last words, eh?), but in case it DOES happen eventually
// purge that malformed data
let Ok(last_received_time) = OffsetDateTime::from_unix_timestamp(last_received) else {
error!("somehow our stored timestamp ({last_received}) for surbs from {pending_reply_target} is corrupted!. Going to remove all the associated entries");
to_remove.push(*pending_reply_target);
continue
};
let diff = now - last_received_time;
if diff > self.max_surb_waiting_period {
warn!("We haven't received any surbs in {:?} from {:?}. Going to explicitly ask for more", diff, pending_reply_target);
if diff > self.config.max_surb_waiting_period {
warn!("We haven't received any surbs in {:?} from {pending_reply_target}. Going to explicitly ask for more", diff);
to_request.push(*pending_reply_target);
}
}
@@ -473,23 +544,96 @@ where
for pending_reply_target in to_request {
self.request_reply_surbs_for_queue_clearing(pending_reply_target)
.await;
self.received_reply_surbs
self.full_reply_storage
.surbs_storage_ref()
.reset_pending_reception(&pending_reply_target)
}
for to_remove in to_remove {
self.pending_replies.remove(&to_remove);
}
}
async fn invalidate_old_data(&self) {
let now = OffsetDateTime::now_utc();
let mut to_remove_surbs = Vec::new();
let mut to_remove_keys = Vec::new();
for map_ref in self.full_reply_storage.surbs_storage_ref().as_raw_iter() {
let (sender, received) = map_ref.pair();
// TODO: handle the following edge case:
// there's a malicious client sending us exactly one reply surb just before we should have invalidated
// the data thus making us keep everything in memory
// possible solution: keep timestamp PER reply surb (but that seems like an overkill)
// but I doubt this is ever going to be a problem...
// ...
// However, if you're reading this message, it probably became a legit problem,
// so I guess add timestamp per surb then? chop-chop.
let last_received = received.surbs_last_received_at();
// this should never ever happen (famous last words, eh?), but in case it DOES happen eventually
// purge that malformed data
let Ok(last_received_time) = OffsetDateTime::from_unix_timestamp(last_received) else {
error!("somehow our stored timestamp ({last_received}) for surbs from {sender} is corrupted!. Going to remove all the associated entries");
to_remove_surbs.push(*sender);
continue
};
let diff = now - last_received_time;
if diff > self.config.max_reply_surb_age {
info!("it's been {diff:?} since we last received any reply surb from {sender}. Going to remove all stored entries...");
to_remove_surbs.push(*sender);
}
}
for map_ref in self.full_reply_storage.key_storage_ref().as_raw_iter() {
let (digest, reply_key) = map_ref.pair();
// this should never ever happen (famous last words, eh?), but in case it DOES happen eventually
// purge that malformed data
let Ok(sent_at) = OffsetDateTime::from_unix_timestamp(reply_key.sent_at_timestamp) else {
error!("somehow our stored timestamp ({}) for one of our reply key is corrupted!. Going to remove all the entry", reply_key.sent_at_timestamp);
to_remove_keys.push(*digest);
continue
};
let diff = now - sent_at;
if diff > self.config.max_reply_key_age {
debug!("it's been {diff:?} since we created this reply key. it's probably never going to get used, so we're going to purge it...");
to_remove_keys.push(*digest);
}
}
for to_remove in to_remove_surbs {
self.full_reply_storage
.surbs_storage_ref()
.remove(&to_remove);
}
for to_remove in to_remove_keys {
self.full_reply_storage.key_storage().remove(to_remove)
}
}
fn create_interval_stream(polling_rate: Duration) -> IntervalStream {
#[cfg(not(target_arch = "wasm32"))]
return tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(polling_rate));
#[cfg(target_arch = "wasm32")]
return gloo_timers::future::IntervalStream::new(polling_rate.as_millis() as u32);
}
pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started ReplyController with graceful shutdown support");
let polling_rate = Duration::from_secs(5);
let mut stale_inspection = Self::create_interval_stream(polling_rate);
#[cfg(not(target_arch = "wasm32"))]
let mut interval =
tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(polling_rate));
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(polling_rate.as_millis() as u32);
// this is in the order of hours/days so we don't have to poll it that often
let polling_rate = Duration::from_secs(self.config.max_reply_surb_age.as_secs() / 10);
let mut invalidation_inspection = Self::create_interval_stream(polling_rate);
while !shutdown.is_shutdown() {
tokio::select! {
@@ -504,9 +648,12 @@ where
break;
}
},
_ = interval.next() => {
_ = stale_inspection.next() => {
self.inspect_stale_entries().await
},
_ = invalidation_inspection.next() => {
self.invalidate_old_data().await
}
}
}
assert!(shutdown.is_shutdown_poll());
@@ -518,12 +665,12 @@ where
debug!("Started ReplyController without graceful shutdown support");
let polling_rate = Duration::from_secs(5);
let mut interval_timer = tokio::time::interval(polling_rate);
let mut interval = Self::create_interval_stream(polling_rate);
loop {
tokio::select! {
req = self.request_receiver.next() => self.handle_request(req.unwrap()).await,
_ = interval_timer.tick() => self.inspect_stale_entries().await
_ = interval.next() => self.inspect_stale_entries().await
}
}
}
clients/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs
+2 -3
View File
@@ -16,7 +16,6 @@ use nymsphinx::anonymous_replies::requests::AnonymousSenderTag;
use std::fs;
use std::path::{Path, PathBuf};
use time::OffsetDateTime;
use tokio::time::Instant;
mod error;
mod manager;
@@ -221,7 +220,7 @@ impl Backend {
let mut received_surbs = Vec::with_capacity(surb_senders.len());
for sender in surb_senders {
let sender_id = sender.id;
let (sender_tag, surbs_last_received_at): (AnonymousSenderTag, Instant) =
let (sender_tag, surbs_last_received_at_timestamp): (AnonymousSenderTag, i64) =
sender.try_into()?;
let stored_surbs = self
.manager
@@ -233,7 +232,7 @@ impl Backend {
received_surbs.push((
sender_tag,
ReceivedReplySurbs::new_retrieved(stored_surbs, surbs_last_received_at),
ReceivedReplySurbs::new_retrieved(stored_surbs, surbs_last_received_at_timestamp),
))
}
clients/client-core/src/client/replies/reply_storage/backend/fs_backend/models.rs
+17 -40
View File
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::client::replies::reply_storage::backend::fs_backend::error::StorageError;
use crate::client::replies::reply_storage::key_storage::UsedReplyKey;
use crypto::generic_array::typenum::Unsigned;
use crypto::Digest;
use nymsphinx::addressing::clients::{Recipient, RecipientBytes};
@@ -9,9 +10,6 @@ use nymsphinx::anonymous_replies::encryption_key::EncryptionKeyDigest;
use nymsphinx::anonymous_replies::requests::{AnonymousSenderTag, SENDER_TAG_SIZE};
use nymsphinx::anonymous_replies::{ReplySurb, SurbEncryptionKey, SurbEncryptionKeySize};
use nymsphinx::params::ReplySurbKeyDigestAlgorithm;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::time::Instant;
#[derive(Debug, Clone)]
pub(crate) struct StoredSenderTag {
@@ -63,21 +61,20 @@ impl TryFrom<StoredSenderTag> for (RecipientBytes, AnonymousSenderTag) {
pub(crate) struct StoredReplyKey {
pub(crate) key_digest: Vec<u8>,
pub(crate) reply_key: Vec<u8>,
pub(crate) sent_at_timestamp: i64,
}
impl StoredReplyKey {
pub(crate) fn new(
key_digest: EncryptionKeyDigest,
reply_key: SurbEncryptionKey,
) -> StoredReplyKey {
pub(crate) fn new(key_digest: EncryptionKeyDigest, reply_key: UsedReplyKey) -> StoredReplyKey {
StoredReplyKey {
key_digest: key_digest.to_vec(),
reply_key: reply_key.to_bytes(),
reply_key: (*reply_key).to_bytes(),
sent_at_timestamp: reply_key.sent_at_timestamp,
}
}
}
impl TryFrom<StoredReplyKey> for (EncryptionKeyDigest, SurbEncryptionKey) {
impl TryFrom<StoredReplyKey> for (EncryptionKeyDigest, UsedReplyKey) {
type Error = StorageError;
fn try_from(value: StoredReplyKey) -> Result<Self, Self::Error> {
@@ -102,7 +99,10 @@ impl TryFrom<StoredReplyKey> for (EncryptionKeyDigest, SurbEncryptionKey) {
});
};
Ok((digest, reply_key))
Ok((
digest,
UsedReplyKey::new(reply_key, value.sent_at_timestamp),
))
}
}
@@ -113,25 +113,18 @@ pub(crate) struct StoredSurbSender {
}
impl StoredSurbSender {
pub(crate) fn new(tag: AnonymousSenderTag, last_sent: Instant) -> Self {
// this doesn't have to be sub-second accurate
// as a matter of fact even if it's off by few minutes or even hours,
// it would still be good enough
let elapsed = last_sent.elapsed();
let now = OffsetDateTime::now_utc();
let last_sent = now - elapsed;
pub(crate) fn new(tag: AnonymousSenderTag, last_sent_timestamp: i64) -> Self {
StoredSurbSender {
// for the purposes of STORING data,
// we ignore that field anyway
id: 0,
tag: tag.to_bytes().to_vec(),
last_sent_timestamp: last_sent.unix_timestamp(),
last_sent_timestamp,
}
}
}
impl TryFrom<StoredSurbSender> for (AnonymousSenderTag, Instant) {
impl TryFrom<StoredSurbSender> for (AnonymousSenderTag, i64) {
type Error = StorageError;
fn try_from(value: StoredSurbSender) -> Result<Self, Self::Error> {
@@ -145,26 +138,10 @@ impl TryFrom<StoredSurbSender> for (AnonymousSenderTag, Instant) {
});
};
let datetime =
OffsetDateTime::from_unix_timestamp(value.last_sent_timestamp).map_err(|err| {
StorageError::CorruptedData {
details: format!("failed to parse stored timestamp - {err}"),
}
})?;
let duration_since: Duration =
(OffsetDateTime::now_utc() - datetime)
.try_into()
.map_err(|err| StorageError::CorruptedData {
details: format!(
"failed to extract valid duration from the stored timestamp - {err}"
),
})?;
let now = Instant::now();
let instant = now.checked_sub(duration_since).unwrap_or(now);
Ok((AnonymousSenderTag::from_bytes(sender_tag_bytes), instant))
Ok((
AnonymousSenderTag::from_bytes(sender_tag_bytes),
value.last_sent_timestamp,
))
}
}
clients/client-core/src/client/replies/reply_storage/combined.rs
-3
View File
@@ -46,17 +46,14 @@ impl CombinedReplyStorage {
self.used_tags.clone()
}
#[cfg(not(target_arch = "wasm32"))]
pub fn key_storage_ref(&self) -> &SentReplyKeys {
&self.sent_reply_keys
}
#[cfg(not(target_arch = "wasm32"))]
pub fn surbs_storage_ref(&self) -> &ReceivedReplySurbsMap {
&self.received_reply_surbs
}
#[cfg(not(target_arch = "wasm32"))]
pub fn tags_storage_ref(&self) -> &UsedSenderTags {
&self.used_tags
}
clients/client-core/src/client/replies/reply_storage/key_storage.rs
+38 -12
View File
@@ -1,24 +1,22 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use dashmap::iter::Iter;
use dashmap::DashMap;
use nymsphinx::anonymous_replies::encryption_key::EncryptionKeyDigest;
use nymsphinx::anonymous_replies::SurbEncryptionKey;
use std::ops::Deref;
use std::sync::Arc;
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
use dashmap::iter::Iter;
use time::OffsetDateTime;
#[derive(Debug, Clone)]
// TODO: we might have to also put the tag here
// TODO2: some timestamp to indicate when entries should get purged if we expect to never get the reply back
pub struct SentReplyKeys {
inner: Arc<SentReplyKeysInner>,
}
#[derive(Debug)]
struct SentReplyKeysInner {
data: DashMap<EncryptionKeyDigest, SurbEncryptionKey>,
data: DashMap<EncryptionKeyDigest, UsedReplyKey>,
}
impl SentReplyKeys {
@@ -31,7 +29,7 @@ impl SentReplyKeys {
}
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
pub(crate) fn from_raw(raw: Vec<(EncryptionKeyDigest, SurbEncryptionKey)>) -> SentReplyKeys {
pub(crate) fn from_raw(raw: Vec<(EncryptionKeyDigest, UsedReplyKey)>) -> SentReplyKeys {
SentReplyKeys {
inner: Arc::new(SentReplyKeysInner {
data: raw.into_iter().collect(),
@@ -39,22 +37,50 @@ impl SentReplyKeys {
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
pub(crate) fn as_raw_iter(&self) -> Iter<'_, EncryptionKeyDigest, SurbEncryptionKey> {
pub(crate) fn as_raw_iter(&self) -> Iter<'_, EncryptionKeyDigest, UsedReplyKey> {
self.inner.data.iter()
}
pub(crate) fn insert_multiple(&self, keys: Vec<SurbEncryptionKey>) {
let now = OffsetDateTime::now_utc().unix_timestamp();
for key in keys {
self.insert(key)
self.insert(UsedReplyKey::new(key, now))
}
}
pub(crate) fn insert(&self, key: SurbEncryptionKey) {
pub(crate) fn insert(&self, key: UsedReplyKey) {
self.inner.data.insert(key.compute_digest(), key);
}
pub(crate) fn try_pop(&self, digest: EncryptionKeyDigest) -> Option<SurbEncryptionKey> {
pub(crate) fn try_pop(&self, digest: EncryptionKeyDigest) -> Option<UsedReplyKey> {
self.inner.data.remove(&digest).map(|(_k, v)| v)
}
pub(crate) fn remove(&self, digest: EncryptionKeyDigest) {
self.inner.data.remove(&digest);
}
}
#[derive(Debug, Copy, Clone)]
pub(crate) struct UsedReplyKey {
key: SurbEncryptionKey,
// the purpose of this field is to perform invalidation at relatively very long intervals
pub(crate) sent_at_timestamp: i64,
}
impl UsedReplyKey {
pub(crate) fn new(key: SurbEncryptionKey, sent_at_timestamp: i64) -> Self {
UsedReplyKey {
key,
sent_at_timestamp,
}
}
}
impl Deref for UsedReplyKey {
type Target = SurbEncryptionKey;
fn deref(&self) -> &Self::Target {
&self.key
}
}
clients/client-core/src/client/replies/reply_storage/surb_storage.rs
+15 -19
View File
@@ -1,6 +1,7 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use dashmap::iter::Iter;
use dashmap::DashMap;
use log::trace;
use nymsphinx::anonymous_replies::requests::AnonymousSenderTag;
@@ -8,15 +9,7 @@ use nymsphinx::anonymous_replies::ReplySurb;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::Instant;
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
use dashmap::iter::Iter;
#[cfg(target_arch = "wasm32")]
use wasm_timer::Instant;
use time::OffsetDateTime;
#[derive(Debug, Clone)]
pub struct ReceivedReplySurbsMap {
@@ -63,18 +56,21 @@ impl ReceivedReplySurbsMap {
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
pub(crate) fn as_raw_iter(&self) -> Iter<'_, AnonymousSenderTag, ReceivedReplySurbs> {
self.inner.data.iter()
}
pub(crate) fn remove(&self, target: &AnonymousSenderTag) {
self.inner.data.remove(target);
}
pub(crate) fn reset_surbs_last_received_at(&self, target: &AnonymousSenderTag) {
if let Some(mut entry) = self.inner.data.get_mut(target) {
entry.surbs_last_received_at = Instant::now();
entry.surbs_last_received_at_timestamp = OffsetDateTime::now_utc().unix_timestamp();
}
}
pub(crate) fn surbs_last_received_at(&self, target: &AnonymousSenderTag) -> Option<Instant> {
pub(crate) fn surbs_last_received_at(&self, target: &AnonymousSenderTag) -> Option<i64> {
self.inner
.data
.get(target)
@@ -199,7 +195,7 @@ pub(crate) struct ReceivedReplySurbs {
data: VecDeque<ReplySurb>,
pending_reception: u32,
surbs_last_received_at: Instant,
surbs_last_received_at_timestamp: i64,
}
impl ReceivedReplySurbs {
@@ -207,19 +203,19 @@ impl ReceivedReplySurbs {
ReceivedReplySurbs {
data: initial_surbs,
pending_reception: 0,
surbs_last_received_at: Instant::now(),
surbs_last_received_at_timestamp: OffsetDateTime::now_utc().unix_timestamp(),
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "fs-surb-storage"))]
pub(crate) fn new_retrieved(
surbs: Vec<ReplySurb>,
surbs_last_received_at: Instant,
surbs_last_received_at_timestamp: i64,
) -> ReceivedReplySurbs {
ReceivedReplySurbs {
data: surbs.into(),
pending_reception: 0,
surbs_last_received_at,
surbs_last_received_at_timestamp,
}
}
@@ -228,8 +224,8 @@ impl ReceivedReplySurbs {
&self.data
}
pub(crate) fn surbs_last_received_at(&self) -> Instant {
self.surbs_last_received_at
pub(crate) fn surbs_last_received_at(&self) -> i64 {
self.surbs_last_received_at_timestamp
}
pub(crate) fn pending_reception(&self) -> u32 {
@@ -276,7 +272,7 @@ impl ReceivedReplySurbs {
let mut v = surbs.into_iter().collect::<VecDeque<_>>();
trace!("storing {} surbs in the storage", v.len());
self.data.append(&mut v);
self.surbs_last_received_at = Instant::now();
self.surbs_last_received_at_timestamp = OffsetDateTime::now_utc().unix_timestamp();
trace!("we now have {} surbs!", self.data.len());
}
}
clients/client-core/src/config/mod.rs
+26
View File
@@ -47,6 +47,12 @@ const DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE: u32 = 500;
const DEFAULT_RETRANSMISSION_REPLY_SURB_REQUEST_SIZE: u32 = 10;
const DEFAULT_MAXIMUM_REPLY_SURB_WAITING_PERIOD: Duration = Duration::from_secs(10);
// 12 hours
const DEFAULT_MAXIMUM_REPLY_SURB_AGE: Duration = Duration::from_secs(12 * 60 * 60);
// 24 hours
const DEFAULT_MAXIMUM_REPLY_KEY_AGE: Duration = Duration::from_secs(24 * 60 * 60);
pub fn missing_string_value() -> String {
MISSING_VALUE.to_string()
}
@@ -323,6 +329,14 @@ impl<T> Config<T> {
pub fn get_maximum_reply_surb_waiting_period(&self) -> Duration {
self.debug.maximum_reply_surb_waiting_period
}
pub fn get_maximum_reply_surb_age(&self) -> Duration {
self.debug.maximum_reply_surb_age
}
pub fn get_maximum_reply_key_age(&self) -> Duration {
self.debug.maximum_reply_key_age
}
}
impl<T: NymConfig> Default for Config<T> {
@@ -570,6 +584,16 @@ pub struct DebugConfig {
/// for more even though in theory they wouldn't need to.
#[serde(with = "humantime_serde")]
pub maximum_reply_surb_waiting_period: Duration,
/// Defines maximum amount of time given reply surb is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
#[serde(with = "humantime_serde")]
pub maximum_reply_surb_age: Duration,
/// Defines maximum amount of time given reply key is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
#[serde(with = "humantime_serde")]
pub maximum_reply_key_age: Duration,
}
#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -602,6 +626,8 @@ impl Default for DebugConfig {
maximum_allowed_reply_surb_request_size: DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE,
retransmission_reply_surb_request_size: DEFAULT_RETRANSMISSION_REPLY_SURB_REQUEST_SIZE,
maximum_reply_surb_waiting_period: DEFAULT_MAXIMUM_REPLY_SURB_WAITING_PERIOD,
maximum_reply_surb_age: DEFAULT_MAXIMUM_REPLY_SURB_AGE,
maximum_reply_key_age: DEFAULT_MAXIMUM_REPLY_KEY_AGE,
}
}
}
clients/webassembly/src/client/config.rs
+12
View File
@@ -129,6 +129,14 @@ pub struct Debug {
/// Defines maximum amount of time the client is going to wait for reply surbs before explicitly asking
/// for more even though in theory they wouldn't need to.
pub maximum_reply_surb_waiting_period_ms: u64,
/// Defines maximum amount of time given reply surb is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
pub maximum_reply_surb_age_ms: u64,
/// Defines maximum amount of time given reply key is going to be valid for.
/// This is going to be superseded by key rotation once implemented.
pub maximum_reply_key_age_ms: u64,
}
impl From<Debug> for ConfigDebug {
@@ -167,6 +175,8 @@ impl From<Debug> for ConfigDebug {
maximum_reply_surb_waiting_period: Duration::from_millis(
debug.maximum_reply_surb_waiting_period_ms,
),
maximum_reply_surb_age: Duration::from_millis(debug.maximum_reply_surb_age_ms),
maximum_reply_key_age: Duration::from_millis(debug.maximum_reply_key_age_ms),
}
}
}
@@ -198,6 +208,8 @@ impl From<ConfigDebug> for Debug {
maximum_reply_surb_waiting_period_ms: debug
.maximum_reply_surb_waiting_period
.as_millis() as u64,
maximum_reply_surb_age_ms: debug.maximum_reply_surb_age.as_millis() as u64,
maximum_reply_key_age_ms: debug.maximum_reply_key_age.as_millis() as u64,
}
}
}