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

Generalise MessageReceiver (#3042)

Browse Source 
* Generalise MessageReceiver

* Generics all the way

* Generalise MessageReceiver

* Generics all the way

* Fix Cargo.lock

---------

Co-authored-by: benedettadavico <benedetta.davico@gmail.com>
This commit is contained in:
Drazen Urch
2023-03-21 15:25:51 +01:00
committed by GitHub
parent 41a63a0985
commit 7e109e7f2d
15 changed files with 595 additions and 7687 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+382 -427
View File
File diff suppressed because it is too large Load Diff
clients/client-core/src/client/base_client/mod.rs
+10 -9
View File
@@ -35,7 +35,7 @@ use nym_crypto::asymmetric::{encryption, identity};
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::addressing::nodes::NodeIdentity;
use nym_sphinx::receiver::ReconstructedMessage;
use nym_sphinx::receiver::{ReconstructedMessage, SphinxMessageReceiver};
use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
use nym_task::{TaskClient, TaskManager};
use nym_topology::provider_trait::TopologyProvider;
@@ -294,14 +294,15 @@ where
shutdown: TaskClient,
) {
info!("Starting received messages buffer controller...");
ReceivedMessagesBufferController::new(
local_encryption_keypair,
query_receiver,
mixnet_receiver,
reply_key_storage,
reply_controller_sender,
)
.start_with_shutdown(shutdown)
let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
ReceivedMessagesBufferController::new(
local_encryption_keypair,
query_receiver,
mixnet_receiver,
reply_key_storage,
reply_controller_sender,
);
controller.start_with_shutdown(shutdown)
}
async fn start_gateway_client(
clients/client-core/src/client/received_buffer.rs
+40 -28
View File
@@ -30,13 +30,13 @@ pub type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver<ReceivedBufferM
pub type ReconstructedMessagesSender = mpsc::UnboundedSender<Vec<ReconstructedMessage>>;
pub type ReconstructedMessagesReceiver = mpsc::UnboundedReceiver<Vec<ReconstructedMessage>>;
struct ReceivedMessagesBufferInner {
struct ReceivedMessagesBufferInner<R: MessageReceiver> {
messages: Vec<ReconstructedMessage>,
local_encryption_keypair: Arc<encryption::KeyPair>,
// TODO: looking how it 'looks' here, perhaps `MessageReceiver` should be renamed to something
// else instead.
message_receiver: MessageReceiver,
message_receiver: R,
message_sender: Option<ReconstructedMessagesSender>,
// TODO: this will get cleared upon re-running the client
@@ -45,7 +45,7 @@ struct ReceivedMessagesBufferInner {
recently_reconstructed: HashSet<i32>,
}
impl ReceivedMessagesBufferInner {
impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
fn recover_from_fragment(&mut self, fragment_data: &[u8]) -> Option<NymMessage> {
if nym_sphinx::cover::is_cover(fragment_data) {
trace!("The message was a loop cover message! Skipping it");
@@ -102,13 +102,13 @@ impl ReceivedMessagesBufferInner {
&mut self,
reply_ciphertext: &mut [u8],
reply_key: SurbEncryptionKey,
) -> Option<NymMessage> {
) -> Result<Option<NymMessage>, MessageRecoveryError> {
// note: this performs decryption IN PLACE without extra allocation
self.message_receiver
.recover_plaintext_from_reply(reply_ciphertext, reply_key);
.recover_plaintext_from_reply(reply_ciphertext, reply_key)?;
let fragment_data = reply_ciphertext;
self.recover_from_fragment(fragment_data)
Ok(self.recover_from_fragment(fragment_data))
}
fn process_received_regular_packet(&mut self, mut raw_fragment: Vec<u8>) -> Option<NymMessage> {
@@ -130,13 +130,13 @@ impl ReceivedMessagesBufferInner {
#[derive(Debug, Clone)]
// Note: you should NEVER create more than a single instance of this using 'new()'.
// You should always use .clone() to create additional instances
struct ReceivedMessagesBuffer {
inner: Arc<Mutex<ReceivedMessagesBufferInner>>,
struct ReceivedMessagesBuffer<R: MessageReceiver> {
inner: Arc<Mutex<ReceivedMessagesBufferInner<R>>>,
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
}
impl ReceivedMessagesBuffer {
impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
fn new(
local_encryption_keypair: Arc<encryption::KeyPair>,
reply_key_storage: SentReplyKeys,
@@ -146,7 +146,7 @@ impl ReceivedMessagesBuffer {
inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
messages: Vec::new(),
local_encryption_keypair,
message_receiver: MessageReceiver::new(),
message_receiver: R::new(),
message_sender: None,
recently_reconstructed: HashSet::new(),
})),
@@ -328,7 +328,10 @@ impl ReceivedMessagesBuffer {
})
}
async fn handle_new_received(&mut self, msgs: Vec<Vec<u8>>) {
async fn handle_new_received(
&mut self,
msgs: Vec<Vec<u8>>,
) -> Result<(), MessageRecoveryError> {
trace!(
"Processing {:?} new message that might get added to the buffer!",
msgs.len()
@@ -344,7 +347,7 @@ impl ReceivedMessagesBuffer {
// if yes - this is a reply message
let completed_message =
if let Some((reply_key, reply_message)) = self.get_reply_key(&mut msg) {
inner_guard.process_received_reply(reply_message, reply_key)
inner_guard.process_received_reply(reply_message, reply_key)?
} else {
inner_guard.process_received_regular_packet(msg)
};
@@ -360,6 +363,7 @@ impl ReceivedMessagesBuffer {
if !completed_messages.is_empty() {
self.handle_reconstructed_messages(completed_messages).await
}
Ok(())
}
}
@@ -372,14 +376,14 @@ pub enum ReceivedBufferMessage {
ReceiverDisconnect,
}
struct RequestReceiver {
received_buffer: ReceivedMessagesBuffer,
struct RequestReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
}
impl RequestReceiver {
impl<R: MessageReceiver> RequestReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer,
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
) -> Self {
RequestReceiver {
@@ -422,14 +426,14 @@ impl RequestReceiver {
}
}
struct FragmentedMessageReceiver {
received_buffer: ReceivedMessagesBuffer,
struct FragmentedMessageReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
}
impl FragmentedMessageReceiver {
impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer,
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
) -> Self {
FragmentedMessageReceiver {
@@ -438,13 +442,16 @@ impl FragmentedMessageReceiver {
}
}
async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
async fn run_with_shutdown(
&mut self,
mut shutdown: nym_task::TaskClient,
) -> Result<(), MessageRecoveryError> {
debug!("Started FragmentedMessageReceiver with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
new_messages = self.mixnet_packet_receiver.next() => {
if let Some(new_messages) = new_messages {
self.received_buffer.handle_new_received(new_messages).await;
self.received_buffer.handle_new_received(new_messages).await?;
} else {
log::trace!("FragmentedMessageReceiver: Stopping since channel closed");
break;
@@ -457,15 +464,16 @@ impl FragmentedMessageReceiver {
}
shutdown.recv_timeout().await;
log::debug!("FragmentedMessageReceiver: Exiting");
Ok(())
}
}
pub(crate) struct ReceivedMessagesBufferController {
fragmented_message_receiver: FragmentedMessageReceiver,
request_receiver: RequestReceiver,
pub(crate) struct ReceivedMessagesBufferController<R: MessageReceiver> {
fragmented_message_receiver: FragmentedMessageReceiver<R>,
request_receiver: RequestReceiver<R>,
}
impl ReceivedMessagesBufferController {
impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferController<R> {
pub(crate) fn new(
local_encryption_keypair: Arc<encryption::KeyPair>,
query_receiver: ReceivedBufferRequestReceiver,
@@ -494,9 +502,13 @@ impl ReceivedMessagesBufferController {
let shutdown_handle = shutdown.clone();
spawn_future(async move {
fragmented_message_receiver
match fragmented_message_receiver
.run_with_shutdown(shutdown_handle)
.await;
.await
{
Ok(_) => {}
Err(e) => error!("{e}"),
}
});
spawn_future(async move {
request_receiver.run_with_shutdown(shutdown).await;
common/nymsphinx/Cargo.toml
+3
View File
@@ -26,6 +26,9 @@ nym-sphinx-types = { path = "types" }
nym-crypto = { path = "../crypto", version = "0.2.0" }
nym-topology = { path = "../topology" }
# outfox
nym-outfox = { path = "../../nym-outfox" }
[dev-dependencies]
nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract" }
common/nymsphinx/src/receiver.rs
+104 -38
View File
@@ -7,6 +7,8 @@ use nym_crypto::asymmetric::encryption;
use nym_crypto::shared_key::recompute_shared_key;
use nym_crypto::symmetric::stream_cipher;
use nym_crypto::symmetric::stream_cipher::CipherKey;
use nym_outfox::error::OutfoxError;
use nym_outfox::lion::lion_transform_decrypt;
use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx_anonymous_replies::SurbEncryptionKey;
use nym_sphinx_chunking::fragment::Fragment;
@@ -74,54 +76,76 @@ pub enum MessageRecoveryError {
#[error("Failed to recover message fragment - {0}")]
FragmentRecoveryError(#[from] ChunkingError),
#[error("Outfox: {source}")]
OutfoxRecoveryError {
#[from]
source: OutfoxError,
},
}
pub struct MessageReceiver {
/// High level public structure used to buffer all received data [`Fragment`]s and eventually
/// returning original messages that they encapsulate.
#[derive(Default)]
pub struct OutfoxMessageReceiver {
reconstructor: MessageReconstructor,
/// 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,
}
impl MessageReceiver {
impl OutfoxMessageReceiver {
pub fn new() -> Self {
Default::default()
}
}
/// Allows setting non-default number of expected mix hops in the network.
#[must_use]
pub fn with_mix_hops(mut self, hops: u8) -> Self {
self.num_mix_hops = hops;
self
impl MessageReceiver for OutfoxMessageReceiver {
fn new() -> Self {
Self::default()
}
fn decrypt_raw_message<C>(&self, message: &mut [u8], key: &CipherKey<C>)
fn reconstructor(&self) -> MessageReconstructor {
self.reconstructor.clone()
}
fn num_mix_hops(&self) -> u8 {
DEFAULT_NUM_MIX_HOPS
}
fn decrypt_raw_message<C>(
&self,
message: &mut [u8],
key: &CipherKey<C>,
) -> Result<(), MessageRecoveryError>
where
C: StreamCipher + KeyIvInit,
{
let zero_iv = stream_cipher::zero_iv::<C>();
stream_cipher::decrypt_in_place::<C>(key, &zero_iv, message)
lion_transform_decrypt(message, key)?;
Ok(())
}
}
/// Given raw fragment data, **WITH KEY DIGEST PREFIX ALREADY REMOVED!!**, uses looked up
/// key to decrypt fragment data
pub fn recover_plaintext_from_reply(
pub trait MessageReceiver {
fn new() -> Self;
fn reconstructor(&self) -> MessageReconstructor;
fn num_mix_hops(&self) -> u8;
fn decrypt_raw_message<C>(
&self,
message: &mut [u8],
key: &CipherKey<C>,
) -> Result<(), MessageRecoveryError>
where
C: StreamCipher + KeyIvInit;
fn recover_plaintext_from_reply(
&self,
reply_ciphertext: &mut [u8],
reply_key: SurbEncryptionKey,
) {
) -> Result<(), MessageRecoveryError> {
self.decrypt_raw_message::<ReplySurbEncryptionAlgorithm>(
reply_ciphertext,
reply_key.inner(),
)
}
/// Given raw fragment data, recovers the remote ephemeral key, recomputes shared secret,
/// uses it to decrypt fragment data
pub fn recover_plaintext_from_regular_packet<'a>(
fn recover_plaintext_from_regular_packet<'a>(
&self,
local_key: &encryption::PrivateKey,
raw_enc_frag: &'a mut [u8],
@@ -146,30 +170,25 @@ impl MessageReceiver {
// 3. decrypt fragment data
let fragment_ciphertext = &mut raw_enc_frag[encryption::PUBLIC_KEY_SIZE..];
self.decrypt_raw_message::<PacketEncryptionAlgorithm>(fragment_ciphertext, &encryption_key);
self.decrypt_raw_message::<PacketEncryptionAlgorithm>(
fragment_ciphertext,
&encryption_key,
)?;
let fragment_data = fragment_ciphertext;
Ok(fragment_data)
}
/// Given fragment data recovers [`Fragment`] itself.
pub fn recover_fragment(&self, frag_data: &[u8]) -> Result<Fragment, MessageRecoveryError> {
fn recover_fragment(&self, frag_data: &[u8]) -> Result<Fragment, MessageRecoveryError> {
Ok(Fragment::try_from_bytes(frag_data)?)
}
/// Inserts given [`Fragment`] into the reconstructor.
/// If it was last remaining [`Fragment`] for the original message, the message is reconstructed
/// and returned alongside all (if applicable) set ids used in the message.
///
/// # Returns:
/// - The reconstructed message alongside optional reply SURB,
/// - List of ids of all the [`Set`]s used during reconstruction to detect stale retransmissions.
pub fn insert_new_fragment(
fn insert_new_fragment(
&mut self,
fragment: Fragment,
) -> Result<Option<(NymMessage, Vec<i32>)>, MessageRecoveryError> {
if let Some((message, used_sets)) = self.reconstructor.insert_new_fragment(fragment) {
match PaddedMessage::new_reconstructed(message).remove_padding(self.num_mix_hops) {
if let Some((message, used_sets)) = self.reconstructor().insert_new_fragment(fragment) {
match PaddedMessage::new_reconstructed(message).remove_padding(self.num_mix_hops()) {
Ok(message) => Ok(Some((message, used_sets))),
Err(err) => Err(MessageRecoveryError::MalformedReconstructedMessage {
source: err,
@@ -182,9 +201,56 @@ impl MessageReceiver {
}
}
impl Default for MessageReceiver {
#[derive(Clone)]
pub struct SphinxMessageReceiver {
/// High level public structure used to buffer all received data [`Fragment`]s and eventually
/// returning original messages that they encapsulate.
reconstructor: MessageReconstructor,
/// 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,
}
impl SphinxMessageReceiver {
/// Allows setting non-default number of expected mix hops in the network.
#[must_use]
pub fn with_mix_hops(mut self, hops: u8) -> Self {
self.num_mix_hops = hops;
self
}
}
impl MessageReceiver for SphinxMessageReceiver {
fn new() -> Self {
Default::default()
}
fn decrypt_raw_message<C>(
&self,
message: &mut [u8],
key: &CipherKey<C>,
) -> Result<(), MessageRecoveryError>
where
C: StreamCipher + KeyIvInit,
{
let zero_iv = stream_cipher::zero_iv::<C>();
stream_cipher::decrypt_in_place::<C>(key, &zero_iv, message);
Ok(())
}
fn reconstructor(&self) -> MessageReconstructor {
self.reconstructor.clone()
}
fn num_mix_hops(&self) -> u8 {
self.num_mix_hops
}
}
impl Default for SphinxMessageReceiver {
fn default() -> Self {
MessageReceiver {
SphinxMessageReceiver {
reconstructor: Default::default(),
num_mix_hops: DEFAULT_NUM_MIX_HOPS,
}
nym-api/src/main.rs
+2 -1
View File
@@ -21,6 +21,7 @@ use node_status_api::NodeStatusCache;
use nym_bin_common::logging::setup_logging;
use nym_config::NymConfig;
use nym_contract_cache::cache::NymContractCache;
use nym_sphinx::receiver::SphinxMessageReceiver;
use nym_task::TaskManager;
use rand::rngs::OsRng;
use std::error::Error;
@@ -127,7 +128,7 @@ async fn start_nym_api_tasks(
// if network monitor is enabled, the storage MUST BE available
let storage = maybe_storage.unwrap();
network_monitor::start(
network_monitor::start::<SphinxMessageReceiver>(
&config,
nym_contract_cache_state,
storage,
nym-api/src/network_monitor/mod.rs
+13 -10
View File
@@ -20,6 +20,7 @@ use futures::channel::mpsc;
use gateway_client::bandwidth::BandwidthController;
use nym_credential_storage::PersistentStorage;
use nym_crypto::asymmetric::{encryption, identity};
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskManager;
use std::sync::Arc;
use validator_client::nyxd::SigningNyxdClient;
@@ -32,7 +33,7 @@ pub(crate) mod test_route;
pub(crate) const ROUTE_TESTING_TEST_NONCE: u64 = 0;
pub(crate) fn setup<'a>(
pub(crate) fn setup<'a, R: MessageReceiver>(
config: &'a Config,
nym_contract_cache_state: &NymContractCache,
storage: &NymApiStorage,
@@ -73,7 +74,9 @@ impl<'a> NetworkMonitorBuilder<'a> {
}
}
pub(crate) async fn build(self) -> NetworkMonitorRunnables {
pub(crate) async fn build<R: MessageReceiver + Send + 'static>(
self,
) -> NetworkMonitorRunnables<R> {
// TODO: those keys change constant throughout the whole execution of the monitor.
// and on top of that, they are used with ALL the gateways -> presumably this should change
// in the future
@@ -140,12 +143,12 @@ impl<'a> NetworkMonitorBuilder<'a> {
}
}
pub(crate) struct NetworkMonitorRunnables {
monitor: Monitor,
pub(crate) struct NetworkMonitorRunnables<R: MessageReceiver + Send + 'static> {
monitor: Monitor<R>,
packet_receiver: PacketReceiver,
}
impl NetworkMonitorRunnables {
impl<R: MessageReceiver + Send + 'static> NetworkMonitorRunnables<R> {
// TODO: note, that is not exactly doing what we want, because when
// `ReceivedProcessor` is constructed, it already spawns a future
// this needs to be refactored!
@@ -195,10 +198,10 @@ fn new_packet_sender(
)
}
fn new_received_processor(
fn new_received_processor<R: MessageReceiver + Send + 'static>(
packets_receiver: ReceivedProcessorReceiver,
client_encryption_keypair: Arc<encryption::KeyPair>,
) -> ReceivedProcessor {
) -> ReceivedProcessor<R> {
ReceivedProcessor::new(packets_receiver, client_encryption_keypair)
}
@@ -217,7 +220,7 @@ fn new_packet_receiver(
// TODO: 1) does it still have to have separate builder or could we get rid of it now?
// TODO: 2) how do we make it non-async as other 'start' methods?
pub(crate) async fn start(
pub(crate) async fn start<R: MessageReceiver + Send + 'static>(
config: &Config,
nym_contract_cache_state: &NymContractCache,
storage: &NymApiStorage,
@@ -225,7 +228,7 @@ pub(crate) async fn start(
system_version: &str,
shutdown: &TaskManager,
) {
let monitor_builder = network_monitor::setup(
let monitor_builder = network_monitor::setup::<R>(
config,
nym_contract_cache_state,
storage,
@@ -233,6 +236,6 @@ pub(crate) async fn start(
system_version,
);
info!("Starting network monitor...");
let runnables = monitor_builder.build().await;
let runnables: NetworkMonitorRunnables<R> = monitor_builder.build().await;
runnables.spawn_tasks(shutdown);
}
nym-api/src/network_monitor/monitor/mod.rs
+5 -4
View File
@@ -10,6 +10,7 @@ use crate::network_monitor::test_route::TestRoute;
use crate::storage::NymApiStorage;
use crate::support::config::Config;
use log::{debug, error, info};
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskClient;
use std::collections::{HashMap, HashSet};
use std::process;
@@ -23,11 +24,11 @@ pub(crate) mod receiver;
pub(crate) mod sender;
pub(crate) mod summary_producer;
pub(super) struct Monitor {
pub(super) struct Monitor<R: MessageReceiver + Send + 'static> {
test_nonce: u64,
packet_preparer: PacketPreparer,
packet_sender: PacketSender,
received_processor: ReceivedProcessor,
received_processor: ReceivedProcessor<R>,
summary_producer: SummaryProducer,
node_status_storage: NymApiStorage,
run_interval: Duration,
@@ -45,12 +46,12 @@ pub(super) struct Monitor {
minimum_test_routes: usize,
}
impl Monitor {
impl<R: MessageReceiver + Send> Monitor<R> {
pub(super) fn new(
config: &Config,
packet_preparer: PacketPreparer,
packet_sender: PacketSender,
received_processor: ReceivedProcessor,
received_processor: ReceivedProcessor<R>,
summary_producer: SummaryProducer,
node_status_storage: NymApiStorage,
) -> Self {
nym-api/src/network_monitor/monitor/processor.rs
+13 -13
View File
@@ -43,7 +43,7 @@ enum LockPermit {
Free,
}
struct ReceivedProcessorInner {
struct ReceivedProcessorInner<R: MessageReceiver> {
/// Nonce of the current test run indicating which packets should get rejected.
test_nonce: Option<u64>,
@@ -55,13 +55,13 @@ struct ReceivedProcessorInner {
client_encryption_keypair: Arc<encryption::KeyPair>,
/// Structure responsible for decrypting and recovering plaintext message from received ciphertexts.
message_receiver: MessageReceiver,
message_receiver: R,
/// Vector containing all received (and decrypted) packets in the current test run.
received_packets: Vec<TestPacket>,
}
impl ReceivedProcessorInner {
impl<R: MessageReceiver> ReceivedProcessorInner<R> {
fn on_message(&mut self, mut message: Vec<u8>) -> Result<(), ProcessingError> {
// if the nonce is none it means the packet was received during the 'waiting' for the
// next test run
@@ -101,22 +101,22 @@ impl ReceivedProcessorInner {
}
}
pub(crate) struct ReceivedProcessor {
pub(crate) struct ReceivedProcessor<R: MessageReceiver> {
permit_changer: Option<mpsc::Sender<LockPermit>>,
inner: Arc<Mutex<ReceivedProcessorInner>>,
inner: Arc<Mutex<ReceivedProcessorInner<R>>>,
}
impl ReceivedProcessor {
impl<R: MessageReceiver + Send + 'static> ReceivedProcessor<R> {
pub(crate) fn new(
packets_receiver: ReceivedProcessorReceiver,
client_encryption_keypair: Arc<encryption::KeyPair>,
) -> Self {
let inner: Arc<Mutex<ReceivedProcessorInner>> =
let inner: Arc<Mutex<ReceivedProcessorInner<R>>> =
Arc::new(Mutex::new(ReceivedProcessorInner {
test_nonce: None,
packets_receiver,
client_encryption_keypair,
message_receiver: MessageReceiver::new(),
message_receiver: R::new(),
received_packets: Vec::new(),
}));
@@ -138,9 +138,9 @@ impl ReceivedProcessor {
receive_or_release_permit(&mut permit_receiver, permit).await;
}
async fn receive_or_release_permit(
async fn receive_or_release_permit<Q: MessageReceiver>(
permit_receiver: &mut mpsc::Receiver<LockPermit>,
mut inner: MutexGuard<'_, ReceivedProcessorInner>,
mut inner: MutexGuard<'_, ReceivedProcessorInner<Q>>,
) {
loop {
tokio::select! {
@@ -166,10 +166,10 @@ impl ReceivedProcessor {
// // this lint really looks like a false positive because when lifetimes are elided,
// // the compiler can't figure out appropriate lifetime bounds
// #[allow(clippy::needless_lifetimes)]
async fn wait_for_permit<'a: 'b, 'b>(
async fn wait_for_permit<'a: 'b, 'b, P: MessageReceiver>(
permit_receiver: &'b mut mpsc::Receiver<LockPermit>,
inner: &'a Mutex<ReceivedProcessorInner>,
) -> Option<MutexGuard<'a, ReceivedProcessorInner>> {
inner: &'a Mutex<ReceivedProcessorInner<P>>,
) -> Option<MutexGuard<'a, ReceivedProcessorInner<P>>> {
loop {
match permit_receiver.next().await {
// we should only ever get this on the very first run
nym-connect/mobile/src-tauri/Cargo.lock
Generated
-7142
View File
File diff suppressed because it is too large Load Diff
nym-outfox/src/error.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use chacha20::cipher::InvalidLength;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum OutFoxError {
pub enum OutfoxError {
#[error("Lengths mismatch, expected: {expected}, got: {got}")]
LenMismatch { expected: usize, got: usize },
#[error("{source}")]
nym-outfox/src/format.rs
+8 -8
View File
@@ -70,7 +70,7 @@ const TAGBYTES: usize = 16;
use std::ops::Range;
use crate::error::OutFoxError;
use crate::error::OutfoxError;
use crate::lion::*;
/// A structure that holds mix packet construction parameters. These incluse the length
@@ -184,16 +184,16 @@ impl MixStageParameters {
user_secret_key: &Scalar,
mix_public_key: &MontgomeryPoint,
routing_data: &[u8],
) -> Result<MontgomeryPoint, OutFoxError> {
) -> Result<MontgomeryPoint, OutfoxError> {
if buffer.len() != self.incoming_packet_length() {
return Err(OutFoxError::LenMismatch {
return Err(OutfoxError::LenMismatch {
expected: buffer.len(),
got: self.incoming_packet_length(),
});
}
if routing_data.len() != self.routing_information_length_bytes {
return Err(OutFoxError::LenMismatch {
return Err(OutfoxError::LenMismatch {
expected: routing_data.len(),
got: self.routing_information_length_bytes,
});
@@ -212,7 +212,7 @@ impl MixStageParameters {
let tag = header_aead_key
.encrypt_in_place_detached(&nonce.into(), &[], &mut buffer[self.header_range()])
.map_err(|_| OutFoxError::ChaCha20Poly1305Error)?;
.map_err(|_| OutfoxError::ChaCha20Poly1305Error)?;
// Copy Tag into buffer
buffer[self.tag_range()].copy_from_slice(&tag[..]);
@@ -230,10 +230,10 @@ impl MixStageParameters {
&self,
buffer: &mut [u8],
mix_secret_key: &Scalar,
) -> Result<MontgomeryPoint, OutFoxError> {
) -> Result<MontgomeryPoint, OutfoxError> {
// Check the length of the incoming buffer is correct.
if buffer.len() != self.incoming_packet_length() {
return Err(OutFoxError::LenMismatch {
return Err(OutfoxError::LenMismatch {
expected: buffer.len(),
got: self.incoming_packet_length(),
});
@@ -257,7 +257,7 @@ impl MixStageParameters {
&mut buffer[self.header_range()],
tag.as_slice().try_into().unwrap(),
)
.map_err(|_| OutFoxError::ChaCha20Poly1305Error)?;
.map_err(|_| OutfoxError::ChaCha20Poly1305Error)?;
// Do a round of LION on the payload
lion_transform_decrypt(&mut buffer[self.payload_range()], &shared_key.0)?;
nym-outfox/src/lib.rs
+1
View File
@@ -1,3 +1,4 @@
pub mod error;
pub mod format;
pub mod lion;
pub mod packet;
nym-outfox/src/lion.rs
+6 -6
View File
@@ -36,7 +36,7 @@ use chacha20::XChaCha20;
use chacha20::XNonce;
use zeroize::Zeroize;
use crate::error::OutFoxError;
use crate::error::OutfoxError;
pub const MIN_MESSAGE_LEN: usize = 24 * 2;
const CONTEXT: &str = "LIONKEYS";
@@ -46,7 +46,7 @@ const TAG_LEN: usize = 24;
///
/// The `key` must be 32 bytes, and the `message` >= 48. The message is
/// mutated to the encrypted message.
pub fn lion_transform_encrypt(message: &mut [u8], key: &[u8]) -> Result<(), OutFoxError> {
pub fn lion_transform_encrypt(message: &mut [u8], key: &[u8]) -> Result<(), OutfoxError> {
lion_transform(message, key, [1, 2, 3])
}
@@ -54,7 +54,7 @@ pub fn lion_transform_encrypt(message: &mut [u8], key: &[u8]) -> Result<(), OutF
///
/// The `key` must be 32 bytes, and the `message` >= 48. The message
/// is mutated to the decrypted message.
pub fn lion_transform_decrypt(message: &mut [u8], key: &[u8]) -> Result<(), OutFoxError> {
pub fn lion_transform_decrypt(message: &mut [u8], key: &[u8]) -> Result<(), OutfoxError> {
lion_transform(message, key, [3, 2, 1])
}
@@ -70,13 +70,13 @@ pub fn lion_transform(
message: &mut [u8],
key: &[u8],
key_schedule: [u64; 3],
) -> Result<(), OutFoxError> {
) -> Result<(), OutfoxError> {
if key.len() != 32 {
return Err(OutFoxError::InvalidKeyLength);
return Err(OutfoxError::InvalidKeyLength);
}
if message.len() < MIN_MESSAGE_LEN {
return Err(OutFoxError::InvalidMessageLength);
return Err(OutfoxError::InvalidMessageLength);
}
// Stage 1: Use stream cipher with Nonce from left size, to xor to the right side
nym-outfox/src/packet.rs
+7
View File
@@ -0,0 +1,7 @@
use crate::format::MixCreationParameters;
#[allow(dead_code)]
pub struct OutfoxPacket {
params: MixCreationParameters,
payload: Vec<u8>,
}