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

combining storage traits

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-05-04 12:17:55 +01:00
parent 88f57e6d87
commit 81c3ad74ca
16 changed files with 492 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/native/src/client/mod.rs
+4 -10
View File
@@ -8,6 +8,7 @@ use futures::channel::mpsc;
use log::*;
use nym_bandwidth_controller::BandwidthController;
use nym_client_core::client::base_client::non_wasm_helpers::create_bandwidth_controller;
use nym_client_core::client::base_client::storage::OnDiskPersistent;
use nym_client_core::client::base_client::{
non_wasm_helpers, BaseClientBuilder, ClientInput, ClientOutput, ClientState,
};
@@ -16,7 +17,6 @@ use nym_client_core::client::key_manager::persistence::OnDiskKeys;
use nym_client_core::client::received_buffer::{
ReceivedBufferMessage, ReceivedBufferRequestSender, ReconstructedMessagesReceiver,
};
use nym_client_core::client::replies::reply_storage::fs_backend;
use nym_client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
use nym_credential_storage::persistent_storage::PersistentStorage;
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
@@ -32,13 +32,7 @@ pub use nym_sphinx::receiver::ReconstructedMessage;
pub mod config;
type NativeClientBuilder<'a> = BaseClientBuilder<
'a,
fs_backend::Backend,
Client<QueryNyxdClient>,
OnDiskKeys,
PersistentStorage,
>;
type NativeClientBuilder<'a> = BaseClientBuilder<'a, Client<QueryNyxdClient>, OnDiskPersistent>;
pub struct SocketClient {
/// Client configuration options, including, among other things, packet sending rates,
@@ -128,8 +122,8 @@ impl SocketClient {
self.key_store(),
bandwidth_controller,
non_wasm_helpers::setup_fs_reply_surb_backend(
Some(self.config.get_base().get_reply_surb_database_path()),
self.config.get_debug_settings(),
self.config.get_base().get_reply_surb_database_path(),
&self.config.get_debug_settings().reply_surbs,
)
.await?,
);
clients/webassembly/src/client/mod.rs
+2 -1
View File
@@ -7,6 +7,7 @@ use crate::client::response_pusher::ResponsePusher;
use crate::constants::NODE_TESTER_CLIENT_ID;
use crate::error::WasmClientError;
use crate::helpers::{parse_recipient, parse_sender_tag, setup_reply_surb_storage_backend};
use crate::storage::traits::FullWasmClientStorage;
use crate::storage::ClientStorage;
use crate::topology::WasmNymTopology;
use js_sys::Promise;
@@ -166,7 +167,7 @@ impl NymClientBuilder {
let key_store =
ClientStorage::new_async(&self.config.id, self.storage_passphrase.take()).await?;
let mut base_builder = BaseClientBuilder::new(
let mut base_builder: BaseClientBuilder<_, FullWasmClientStorage> = BaseClientBuilder::new(
&self.config.gateway_endpoint,
&self.config.debug,
key_store,
clients/webassembly/src/error.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::storage::ClientStorageError;
use crate::storage::errors::ClientStorageError;
use crate::topology::WasmTopologyError;
use js_sys::Promise;
use nym_client_core::error::ClientCoreError;
clients/webassembly/src/storage/errors.rs
+25
View File
@@ -0,0 +1,25 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use thiserror::Error;
use wasm_bindgen::JsValue;
use wasm_utils::simple_js_error;
use wasm_utils::storage::error::StorageError;
#[derive(Debug, Error)]
pub enum ClientStorageError {
#[error("failed to use the storage: {source}")]
StorageError {
#[from]
source: StorageError,
},
#[error("{typ} cryptographic key is not available in storage")]
CryptoKeyNotInStorage { typ: String },
}
impl From<ClientStorageError> for JsValue {
fn from(value: ClientStorageError) -> Self {
simple_js_error(value.to_string())
}
}
clients/webassembly/src/storage/mod.rs
+5 -57
View File
@@ -1,21 +1,21 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use async_trait::async_trait;
use crate::storage::errors::ClientStorageError;
use js_sys::Promise;
use nym_client_core::client::key_manager::{persistence::KeyStore, KeyManager};
use nym_crypto::asymmetric::{encryption, identity};
use nym_gateway_client::SharedKeys;
use nym_sphinx::acknowledgements::AckKey;
use std::sync::Arc;
use thiserror::Error;
use wasm_bindgen::prelude::*;
use wasm_bindgen_futures::future_to_promise;
use wasm_utils::storage::error::StorageError;
use wasm_utils::storage::{IdbVersionChangeEvent, WasmStorage};
use wasm_utils::{console_log, simple_js_error, PromisableResult};
use wasm_utils::PromisableResult;
use zeroize::Zeroizing;
pub(crate) mod errors;
pub(crate) mod traits;
const STORAGE_NAME_PREFIX: &str = "wasm-client-storage";
const STORAGE_VERSION: u32 = 1;
@@ -33,24 +33,6 @@ mod v1 {
pub const AES128CTR_BLAKE3_HMAC_GATEWAY_KEYS: &str = "aes128ctr_blake3_hmac_gateway_keys";
}
#[derive(Debug, Error)]
pub enum ClientStorageError {
#[error("failed to use the storage: {source}")]
StorageError {
#[from]
source: StorageError,
},
#[error("{typ} cryptographic key is not available in storage")]
CryptoKeyNotInStorage { typ: String },
}
impl From<ClientStorageError> for JsValue {
fn from(value: ClientStorageError) -> Self {
simple_js_error(value.to_string())
}
}
#[wasm_bindgen]
pub struct ClientStorage {
#[allow(dead_code)]
@@ -247,37 +229,3 @@ impl ClientStorage {
.map_err(Into::into)
}
}
#[async_trait(?Send)]
impl KeyStore for ClientStorage {
type StorageError = ClientStorageError;
async fn load_keys(&self) -> Result<KeyManager, Self::StorageError> {
console_log!("attempting to load cryptographic keys...");
// all keys implement `ZeroizeOnDrop`, so if we return an Error, whatever was already loaded will be cleared
let identity_keypair = self.must_read_identity_keypair().await?;
let encryption_keypair = self.must_read_encryption_keypair().await?;
let ack_keypair = self.must_read_ack_key().await?;
let gateway_shared_key = self.must_read_gateway_shared_key().await?;
Ok(KeyManager::from_keys(
identity_keypair,
encryption_keypair,
gateway_shared_key,
ack_keypair,
))
}
async fn store_keys(&self, keys: &KeyManager) -> Result<(), Self::StorageError> {
console_log!("attempting to store cryptographic keys...");
self.store_identity_keypair(&keys.identity_keypair())
.await?;
self.store_encryption_keypair(&keys.encryption_keypair())
.await?;
self.store_ack_key(&keys.ack_key()).await?;
self.store_gateway_shared_key(&keys.gateway_shared_key())
.await
}
}
clients/webassembly/src/storage/traits.rs
+76
View File
@@ -0,0 +1,76 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::storage::errors::ClientStorageError;
use crate::storage::ClientStorage;
use async_trait::async_trait;
use nym_client_core::client::base_client::storage::MixnetClientStorage;
use nym_client_core::client::key_manager::persistence::KeyStore;
use nym_client_core::client::key_manager::KeyManager;
use nym_client_core::client::replies::reply_storage::browser_backend;
use nym_credential_storage::ephemeral_storage::EphemeralStorage as EphemeralCredentialStorage;
use wasm_utils::console_log;
// temporary until other variants are properly implemented (probably it should get changed into `ClientStorage`
// implementing all traits and everything getting combined
pub struct FullWasmClientStorage {
key_store: ClientStorage,
reply_storage: browser_backend::Backend,
credential_storage: EphemeralCredentialStorage,
}
impl MixnetClientStorage for FullWasmClientStorage {
type KeyStore = ClientStorage;
type ReplyStore = browser_backend::Backend;
type CredentialStore = EphemeralCredentialStorage;
fn into_split(self) -> (Self::KeyStore, Self::ReplyStore, Self::CredentialStore) {
(self.key_store, self.reply_storage, self.credential_storage)
}
fn key_store(&self) -> &Self::KeyStore {
&self.key_store
}
fn reply_store(&self) -> &Self::ReplyStore {
&self.reply_storage
}
fn credential_store(&self) -> &Self::CredentialStore {
&self.credential_storage
}
}
#[async_trait(?Send)]
impl KeyStore for ClientStorage {
type StorageError = ClientStorageError;
async fn load_keys(&self) -> Result<KeyManager, Self::StorageError> {
console_log!("attempting to load cryptographic keys...");
// all keys implement `ZeroizeOnDrop`, so if we return an Error, whatever was already loaded will be cleared
let identity_keypair = self.must_read_identity_keypair().await?;
let encryption_keypair = self.must_read_encryption_keypair().await?;
let ack_keypair = self.must_read_ack_key().await?;
let gateway_shared_key = self.must_read_gateway_shared_key().await?;
Ok(KeyManager::from_keys(
identity_keypair,
encryption_keypair,
gateway_shared_key,
ack_keypair,
))
}
async fn store_keys(&self, keys: &KeyManager) -> Result<(), Self::StorageError> {
console_log!("attempting to store cryptographic keys...");
self.store_identity_keypair(&keys.identity_keypair())
.await?;
self.store_encryption_keypair(&keys.encryption_keypair())
.await?;
self.store_ack_key(&keys.ack_key()).await?;
self.store_gateway_shared_key(&keys.gateway_shared_key())
.await
}
}
common/client-core/src/client/base_client/mod.rs
+45 -50
View File
@@ -2,8 +2,10 @@
// SPDX-License-Identifier: Apache-2.0
use super::received_buffer::ReceivedBufferMessage;
use crate::client::base_client::storage::MixnetClientStorage;
use crate::client::cover_traffic_stream::LoopCoverTrafficStream;
use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender};
use crate::client::key_manager::persistence::KeyStore;
use crate::client::key_manager::ManagedKeys;
use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
use crate::client::real_messages_control;
@@ -43,11 +45,9 @@ use std::sync::Arc;
use tap::TapFallible;
use url::Url;
use nym_credential_storage::storage::Storage;
#[cfg(not(target_arch = "wasm32"))]
use nym_validator_client::nyxd::traits::DkgQueryClient;
use crate::client::key_manager::persistence::KeyStore;
#[cfg(target_arch = "wasm32")]
use nym_bandwidth_controller::wasm_mockups::DkgQueryClient;
@@ -55,6 +55,7 @@ use nym_bandwidth_controller::wasm_mockups::DkgQueryClient;
pub mod non_wasm_helpers;
pub mod helpers;
pub mod storage;
#[derive(Clone)]
pub struct ClientInput {
@@ -153,33 +154,32 @@ impl From<bool> for CredentialsToggle {
}
}
pub struct BaseClientBuilder<'a, B, C, Kst, St: Storage> {
pub struct BaseClientBuilder<'a, C, S: MixnetClientStorage> {
// due to wasm limitations I had to split it like this : (
gateway_config: &'a GatewayEndpointConfig,
debug_config: &'a DebugConfig,
disabled_credentials: bool,
nym_api_endpoints: Vec<Url>,
reply_storage_backend: B,
key_store: Kst,
reply_storage_backend: S::ReplyStore,
key_store: S::KeyStore,
custom_topology_provider: Option<Box<dyn TopologyProvider>>,
bandwidth_controller: Option<BandwidthController<C, St>>,
bandwidth_controller: Option<BandwidthController<C, S::CredentialStore>>,
managed_keys: ManagedKeys,
}
impl<'a, B, C, Kst, St> BaseClientBuilder<'a, B, C, Kst, St>
impl<'a, C, S> BaseClientBuilder<'a, C, S>
where
B: ReplyStorageBackend + Send + Sync + 'static,
C: DkgQueryClient + Sync + Send + 'static,
Kst: KeyStore,
St: Storage + 'static,
S: MixnetClientStorage + 'static,
C: DkgQueryClient + Send + Sync + 'static,
{
// TODO: combine all storages
pub fn new_from_base_config<T>(
base_config: &'a Config<T>,
key_store: Kst,
bandwidth_controller: Option<BandwidthController<C, St>>,
reply_storage_backend: B,
) -> BaseClientBuilder<'a, B, C, Kst, St> {
key_store: S::KeyStore,
bandwidth_controller: Option<BandwidthController<C, S::CredentialStore>>,
reply_storage_backend: S::ReplyStore,
) -> BaseClientBuilder<'a, C, S> {
BaseClientBuilder {
gateway_config: base_config.get_gateway_endpoint_config(),
debug_config: base_config.get_debug_config(),
@@ -193,15 +193,16 @@ where
}
}
// TODO: combine all storages
pub fn new(
gateway_config: &'a GatewayEndpointConfig,
debug_config: &'a DebugConfig,
key_store: Kst,
bandwidth_controller: Option<BandwidthController<C, St>>,
reply_storage_backend: B,
key_store: S::KeyStore,
bandwidth_controller: Option<BandwidthController<C, S::CredentialStore>>,
reply_storage_backend: S::ReplyStore,
credentials_toggle: CredentialsToggle,
nym_api_endpoints: Vec<Url>,
) -> BaseClientBuilder<'a, B, C, Kst, St> {
) -> BaseClientBuilder<'a, C, S> {
BaseClientBuilder {
gateway_config,
debug_config,
@@ -315,9 +316,9 @@ where
mixnet_message_sender: MixnetMessageSender,
ack_sender: AcknowledgementSender,
shutdown: TaskClient,
) -> Result<GatewayClient<C, St>, ClientCoreError>
) -> Result<GatewayClient<C, S::CredentialStore>, ClientCoreError>
where
Kst::StorageError: Send + Sync + 'static,
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
{
let gateway_id = self.gateway_config.gateway_id.clone();
if gateway_id.is_empty() {
@@ -425,7 +426,7 @@ where
// over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
// requests?
fn start_mix_traffic_controller(
gateway_client: GatewayClient<C, St>,
gateway_client: GatewayClient<C, S::CredentialStore>,
shutdown: TaskClient,
) -> BatchMixMessageSender {
info!("Starting mix traffic controller...");
@@ -434,39 +435,32 @@ where
mix_tx
}
// TODO: rename it as it implies the data is persistent whilst one can use InMemBackend
async fn setup_persistent_reply_storage(
backend: B,
backend: S::ReplyStore,
shutdown: TaskClient,
) -> Result<CombinedReplyStorage, ClientCoreError>
where
<B as ReplyStorageBackend>::StorageError: Sync + Send,
<S::ReplyStore as ReplyStorageBackend>::StorageError: Sync + Send,
S::ReplyStore: Send + Sync,
{
if backend.is_active() {
log::trace!("Setup persistent reply storage");
let persistent_storage = PersistentReplyStorage::new(backend);
let mem_store = persistent_storage
.load_state_from_backend()
log::trace!("Setup persistent reply storage");
let persistent_storage = PersistentReplyStorage::new(backend);
let mem_store = persistent_storage
.load_state_from_backend()
.await
.map_err(|err| ClientCoreError::SurbStorageError {
source: Box::new(err),
})?;
let store_clone = mem_store.clone();
spawn_future(async move {
persistent_storage
.flush_on_shutdown(store_clone, shutdown)
.await
.map_err(|err| ClientCoreError::SurbStorageError {
source: Box::new(err),
})?;
});
let store_clone = mem_store.clone();
spawn_future(async move {
persistent_storage
.flush_on_shutdown(store_clone, shutdown)
.await
});
Ok(mem_store)
} else {
log::trace!("Setup inactive reply storage");
Ok(backend
.get_inactive_storage()
.map_err(|err| ClientCoreError::SurbStorageError {
source: Box::new(err),
})?)
}
Ok(mem_store)
}
async fn initial_key_setup(&mut self) {
@@ -477,8 +471,9 @@ where
pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError>
where
<B as ReplyStorageBackend>::StorageError: Sync + Send,
Kst::StorageError: Send + Sync + 'static,
<S::ReplyStore as ReplyStorageBackend>::StorageError: Sync + Send,
S::ReplyStore: Send + Sync,
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
{
info!("Starting nym client");
self.initial_key_setup().await;
common/client-core/src/client/base_client/non_wasm_helpers.rs
+25 -37
View File
@@ -4,8 +4,8 @@
use crate::client::replies::reply_storage::{
fs_backend, CombinedReplyStorage, ReplyStorageBackend,
};
use crate::config;
use crate::config::Config;
use crate::config::DebugConfig;
use crate::error::ClientCoreError;
use log::{error, info};
use nym_bandwidth_controller::BandwidthController;
@@ -18,7 +18,7 @@ use time::OffsetDateTime;
async fn setup_fresh_backend<P: AsRef<Path>>(
db_path: P,
debug_config: &DebugConfig,
surb_config: &config::ReplySurbs,
) -> Result<fs_backend::Backend, ClientCoreError> {
info!("creating fresh surb database");
let mut storage_backend = match fs_backend::Backend::init(db_path).await {
@@ -35,12 +35,8 @@ async fn setup_fresh_backend<P: AsRef<Path>>(
// it will only be happening on the very first run and in practice won't incur huge
// costs since the storage is going to be empty
let mem_store = CombinedReplyStorage::new(
debug_config
.reply_surbs
.minimum_reply_surb_storage_threshold,
debug_config
.reply_surbs
.maximum_reply_surb_storage_threshold,
surb_config.minimum_reply_surb_storage_threshold,
surb_config.maximum_reply_surb_storage_threshold,
);
storage_backend
.init_fresh(&mem_store)
@@ -52,17 +48,13 @@ async fn setup_fresh_backend<P: AsRef<Path>>(
Ok(storage_backend)
}
fn setup_inactive_backend(debug_config: &DebugConfig) -> fs_backend::Backend {
info!("creating inactive surb database");
fs_backend::Backend::new_inactive(
debug_config
.reply_surbs
.minimum_reply_surb_storage_threshold,
debug_config
.reply_surbs
.maximum_reply_surb_storage_threshold,
)
}
// fn setup_inactive_backend(surb_config: &config::ReplySurbs) -> fs_backend::Backend {
// info!("creating inactive surb database");
// fs_backend::Backend::new_inactive(
// surb_config.minimum_reply_surb_storage_threshold,
// surb_config.maximum_reply_surb_storage_threshold,
// )
// }
fn archive_corrupted_database<P: AsRef<Path>>(db_path: P) -> io::Result<()> {
let db_path = db_path.as_ref();
@@ -86,29 +78,25 @@ fn archive_corrupted_database<P: AsRef<Path>>(db_path: P) -> io::Result<()> {
}
pub async fn setup_fs_reply_surb_backend<P: AsRef<Path>>(
db_path: Option<P>,
debug_config: &DebugConfig,
db_path: P,
surb_config: &config::ReplySurbs,
) -> Result<fs_backend::Backend, ClientCoreError> {
if let Some(db_path) = db_path {
// if the database file doesnt exist, initialise fresh storage, otherwise attempt to load
// the existing one
let db_path = db_path.as_ref();
if db_path.exists() {
info!("loading existing surb database");
match fs_backend::Backend::try_load(db_path).await {
Ok(backend) => Ok(backend),
Err(err) => {
error!("failed to setup persistent storage backend for our reply needs: {err}. We're going to create a fresh database instead. This behaviour might change in the future");
// if the database file doesnt exist, initialise fresh storage, otherwise attempt to load
// the existing one
let db_path = db_path.as_ref();
if db_path.exists() {
info!("loading existing surb database");
match fs_backend::Backend::try_load(db_path).await {
Ok(backend) => Ok(backend),
Err(err) => {
error!("failed to setup persistent storage backend for our reply needs: {err}. We're going to create a fresh database instead. This behaviour might change in the future");
archive_corrupted_database(db_path)?;
setup_fresh_backend(db_path, debug_config).await
}
archive_corrupted_database(db_path)?;
setup_fresh_backend(db_path, surb_config).await
}
} else {
setup_fresh_backend(db_path, debug_config).await
}
} else {
Ok(setup_inactive_backend(debug_config))
setup_fresh_backend(db_path, surb_config).await
}
}
common/client-core/src/client/base_client/storage/mod.rs
+109
View File
@@ -0,0 +1,109 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// TODO: combine those more closely. Perhaps into a single underlying store.
// Like for persistent, on-disk, storage, what's the point of having 3 different databases?
use crate::client::key_manager::persistence::{InMemEphemeralKeys, KeyStore};
use crate::client::replies::reply_storage;
use crate::client::replies::reply_storage::ReplyStorageBackend;
use nym_credential_storage::ephemeral_storage::{
EphemeralStorage as EphemeralCredentialStorage, EphemeralStorage,
};
use nym_credential_storage::storage::Storage as CredentialStorage;
#[cfg(not(target_arch = "wasm32"))]
use crate::client::key_manager::persistence::OnDiskKeys;
#[cfg(not(target_arch = "wasm32"))]
use nym_credential_storage::persistent_storage::PersistentStorage as PersistentCredentialStorage;
#[cfg(not(target_arch = "wasm32"))]
use crate::client::replies::reply_storage::fs_backend;
pub trait MixnetClientStorage {
type KeyStore: KeyStore;
type ReplyStore: ReplyStorageBackend;
type CredentialStore: CredentialStorage;
// this is a TERRIBLE name...
fn into_split(self) -> (Self::KeyStore, Self::ReplyStore, Self::CredentialStore);
fn key_store(&self) -> &Self::KeyStore;
fn reply_store(&self) -> &Self::ReplyStore;
fn credential_store(&self) -> &Self::CredentialStore;
}
pub struct Ephemeral {
key_store: InMemEphemeralKeys,
reply_store: reply_storage::Empty,
credential_store: EphemeralStorage,
}
impl Ephemeral {
pub fn new() -> Self {
Default::default()
}
}
impl Default for Ephemeral {
fn default() -> Self {
Ephemeral {
key_store: Default::default(),
reply_store: Default::default(),
credential_store: Default::default(),
}
}
}
impl MixnetClientStorage for Ephemeral {
type KeyStore = InMemEphemeralKeys;
type ReplyStore = reply_storage::Empty;
type CredentialStore = EphemeralCredentialStorage;
fn into_split(self) -> (Self::KeyStore, Self::ReplyStore, Self::CredentialStore) {
(self.key_store, self.reply_store, self.credential_store)
}
fn key_store(&self) -> &Self::KeyStore {
&self.key_store
}
fn reply_store(&self) -> &Self::ReplyStore {
&self.reply_store
}
fn credential_store(&self) -> &Self::CredentialStore {
&self.credential_store
}
}
#[cfg(not(target_arch = "wasm32"))]
pub struct OnDiskPersistent {
pub(crate) key_store: OnDiskKeys,
pub(crate) reply_store: fs_backend::Backend,
pub(crate) credential_store: PersistentCredentialStorage,
}
#[cfg(not(target_arch = "wasm32"))]
impl MixnetClientStorage for OnDiskPersistent {
type KeyStore = OnDiskKeys;
type ReplyStore = fs_backend::Backend;
type CredentialStore = PersistentCredentialStorage;
fn into_split(self) -> (Self::KeyStore, Self::ReplyStore, Self::CredentialStore) {
(self.key_store, self.reply_store, self.credential_store)
}
fn key_store(&self) -> &Self::KeyStore {
&self.key_store
}
fn reply_store(&self) -> &Self::ReplyStore {
&self.reply_store
}
fn credential_store(&self) -> &Self::CredentialStore {
&self.credential_store
}
}
common/client-core/src/client/key_manager/persistence.rs
+1
View File
@@ -94,6 +94,7 @@ impl KeyStore for OnDiskKeys {
}
}
#[derive(Default)]
pub struct InMemEphemeralKeys;
#[derive(Debug, thiserror::Error)]
common/client-core/src/client/replies/reply_storage/backend/browser_backend.rs
+18 -20
View File
@@ -5,8 +5,6 @@ use crate::client::replies::reply_storage::backend::Empty;
use crate::client::replies::reply_storage::{CombinedReplyStorage, ReplyStorageBackend};
use async_trait::async_trait;
use std::path::PathBuf;
// well, right now we don't have the browser storage : (
// so we keep everything in memory
#[derive(Debug)]
@@ -29,21 +27,21 @@ impl Backend {
impl ReplyStorageBackend for Backend {
type StorageError = <Empty as ReplyStorageBackend>::StorageError;
async fn new(
debug_config: &crate::config::DebugConfig,
_db_path: Option<PathBuf>,
) -> Result<Self, Self::StorageError> {
Ok(Backend {
empty: Empty {
min_surb_threshold: debug_config
.reply_surbs
.minimum_reply_surb_storage_threshold,
max_surb_threshold: debug_config
.reply_surbs
.maximum_reply_surb_storage_threshold,
},
})
}
// async fn new(
// debug_config: &crate::config::DebugConfig,
// _db_path: Option<PathBuf>,
// ) -> Result<Self, Self::StorageError> {
// Ok(Backend {
// empty: Empty {
// min_surb_threshold: debug_config
// .reply_surbs
// .minimum_reply_surb_storage_threshold,
// max_surb_threshold: debug_config
// .reply_surbs
// .maximum_reply_surb_storage_threshold,
// },
// })
// }
async fn flush_surb_storage(
&mut self,
@@ -60,7 +58,7 @@ impl ReplyStorageBackend for Backend {
self.empty.load_surb_storage().await
}
fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
self.empty.get_inactive_storage()
}
// fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
// self.empty.get_inactive_storage()
// }
}
common/client-core/src/client/replies/reply_storage/backend/fs_backend/mod.rs
+94 -100
View File
@@ -1,7 +1,6 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::client::base_client::non_wasm_helpers;
use crate::client::replies::reply_storage::backend::fs_backend::manager::StorageManager;
use crate::client::replies::reply_storage::backend::fs_backend::models::{
ReplySurbStorageMetadata, StoredReplyKey, StoredReplySurb, StoredSenderTag, StoredSurbSender,
@@ -23,49 +22,50 @@ mod error;
mod manager;
mod models;
#[derive(Debug)]
enum StorageManagerState {
Storage(StorageManager),
Inactive(InactiveMetadata),
}
// When the storage backaed is initialized as inactive, it will still contain metadata parameters
// that will be needed when the in-mem storage is fetched for use.
#[derive(Debug)]
struct InactiveMetadata {
pub minimum_reply_surb_storage_threshold: usize,
pub maximum_reply_surb_storage_threshold: usize,
}
impl StorageManagerState {
fn get(&self) -> &StorageManager {
match self {
StorageManagerState::Storage(manager) => manager,
StorageManagerState::Inactive(_) => {
panic!("tried to get storage of an inactive backend")
}
}
}
fn get_mut(&mut self) -> &mut StorageManager {
match self {
StorageManagerState::Storage(manager) => manager,
StorageManagerState::Inactive(_) => {
panic!("tried to get storage of an inactive backend")
}
}
}
fn is_active(&self) -> bool {
matches!(self, StorageManagerState::Storage(_))
}
}
// #[derive(Debug)]
// enum StorageManagerState {
// Storage(StorageManager),
// Inactive(InactiveMetadata),
// }
//
// // When the storage backaed is initialized as inactive, it will still contain metadata parameters
// // that will be needed when the in-mem storage is fetched for use.
// #[derive(Debug)]
// struct InactiveMetadata {
// pub minimum_reply_surb_storage_threshold: usize,
// pub maximum_reply_surb_storage_threshold: usize,
// }
//
// impl StorageManagerState {
// fn get(&self) -> &StorageManager {
// match self {
// StorageManagerState::Storage(manager) => manager,
// StorageManagerState::Inactive(_) => {
// panic!("tried to get storage of an inactive backend")
// }
// }
// }
//
// fn get_mut(&mut self) -> &mut StorageManager {
// match self {
// StorageManagerState::Storage(manager) => manager,
// StorageManagerState::Inactive(_) => {
// panic!("tried to get storage of an inactive backend")
// }
// }
// }
//
// fn is_active(&self) -> bool {
// matches!(self, StorageManagerState::Storage(_))
// }
// }
#[derive(Debug)]
pub struct Backend {
temporary_old_path: Option<PathBuf>,
database_path: PathBuf,
manager: StorageManagerState,
manager: StorageManager,
// manager: StorageManagerState,
}
impl Backend {
@@ -85,25 +85,26 @@ impl Backend {
let backend = Backend {
temporary_old_path: None,
database_path: owned_path,
manager: StorageManagerState::Storage(manager),
// manager: StorageManagerState::Storage(manager),
manager,
};
Ok(backend)
}
pub fn new_inactive(
minimum_reply_surb_storage_threshold: usize,
maximum_reply_surb_storage_threshold: usize,
) -> Self {
Backend {
temporary_old_path: None,
database_path: PathBuf::new(),
manager: StorageManagerState::Inactive(InactiveMetadata {
minimum_reply_surb_storage_threshold,
maximum_reply_surb_storage_threshold,
}),
}
}
// pub fn new_inactive(
// minimum_reply_surb_storage_threshold: usize,
// maximum_reply_surb_storage_threshold: usize,
// ) -> Self {
// Backend {
// temporary_old_path: None,
// database_path: PathBuf::new(),
// manager: StorageManagerState::Inactive(InactiveMetadata {
// minimum_reply_surb_storage_threshold,
// maximum_reply_surb_storage_threshold,
// }),
// }
// }
pub async fn try_load<P: AsRef<Path>>(database_path: P) -> Result<Self, StorageError> {
let owned_path: PathBuf = database_path.as_ref().into();
@@ -176,12 +177,14 @@ impl Backend {
Ok(Backend {
temporary_old_path: None,
database_path: owned_path,
manager: StorageManagerState::Storage(manager),
// manager: StorageManagerState::Storage(manager),
manager,
})
}
async fn close_pool(&mut self) {
self.manager.get_mut().connection_pool.close().await;
self.manager.connection_pool.close().await;
// self.manager.get_mut().connection_pool.close().await;
}
async fn rotate(&mut self) -> Result<(), StorageError> {
@@ -200,9 +203,8 @@ impl Backend {
fs::rename(&self.database_path, &temp_old)
.map_err(|err| StorageError::DatabaseRenameError { source: err })?;
self.manager =
StorageManagerState::Storage(StorageManager::init(&self.database_path, true).await?);
self.manager.get_mut().create_status_table().await?;
self.manager = StorageManager::init(&self.database_path, true).await?;
self.manager.create_status_table().await?;
self.temporary_old_path = Some(temp_old);
Ok(())
@@ -219,27 +221,26 @@ impl Backend {
}
async fn start_storage_flush(&self) -> Result<(), StorageError> {
Ok(self.manager.get().set_flush_status(true).await?)
Ok(self.manager.set_flush_status(true).await?)
}
async fn end_storage_flush(&self) -> Result<(), StorageError> {
self.manager
.get()
.set_previous_flush_timestamp(OffsetDateTime::now_utc().unix_timestamp())
.await?;
Ok(self.manager.get().set_flush_status(false).await?)
Ok(self.manager.set_flush_status(false).await?)
}
async fn start_client_use(&self) -> Result<(), StorageError> {
Ok(self.manager.get().set_client_in_use_status(true).await?)
Ok(self.manager.set_client_in_use_status(true).await?)
}
async fn stop_client_use(&self) -> Result<(), StorageError> {
Ok(self.manager.get().set_client_in_use_status(false).await?)
Ok(self.manager.set_client_in_use_status(false).await?)
}
async fn get_stored_tags(&self) -> Result<UsedSenderTags, StorageError> {
let stored = self.manager.get().get_tags().await?;
let stored = self.manager.get_tags().await?;
// stop at the first instance of corruption. if even a single entry is malformed,
// something weird has happened and we can't trust the rest of the data
@@ -255,7 +256,6 @@ impl Backend {
for map_ref in tags.as_raw_iter() {
let (recipient, tag) = map_ref.pair();
self.manager
.get()
.insert_tag(StoredSenderTag::new(*recipient, *tag))
.await?;
}
@@ -263,7 +263,7 @@ impl Backend {
}
async fn get_stored_reply_keys(&self) -> Result<SentReplyKeys, StorageError> {
let stored = self.manager.get().get_reply_keys().await?;
let stored = self.manager.get_reply_keys().await?;
// stop at the first instance of corruption. if even a single entry is malformed,
// something weird has happened and we can't trust the rest of the data
@@ -279,7 +279,6 @@ impl Backend {
for map_ref in reply_keys.as_raw_iter() {
let (digest, key) = map_ref.pair();
self.manager
.get()
.insert_reply_key(StoredReplyKey::new(*digest, *key))
.await?;
}
@@ -287,7 +286,7 @@ impl Backend {
}
async fn get_stored_reply_surbs(&self) -> Result<ReceivedReplySurbsMap, StorageError> {
let surb_senders = self.manager.get().get_surb_senders().await?;
let surb_senders = self.manager.get_surb_senders().await?;
let metadata = self.get_reply_surb_storage_metadata().await?;
let mut received_surbs = Vec::with_capacity(surb_senders.len());
@@ -297,7 +296,6 @@ impl Backend {
sender.try_into()?;
let stored_surbs = self
.manager
.get()
.get_reply_surbs(sender_id)
.await?
.into_iter()
@@ -325,7 +323,6 @@ impl Backend {
let (tag, received_surbs) = map_ref.pair();
let sender_id = self
.manager
.get()
.insert_surb_sender(StoredSurbSender::new(
*tag,
received_surbs.surbs_last_received_at(),
@@ -334,7 +331,6 @@ impl Backend {
for reply_surb in received_surbs.surbs_ref() {
self.manager
.get()
.insert_reply_surb(StoredReplySurb::new(sender_id, reply_surb))
.await?
}
@@ -346,7 +342,6 @@ impl Backend {
&self,
) -> Result<ReplySurbStorageMetadata, StorageError> {
self.manager
.get()
.get_reply_surb_storage_metadata()
.await
.map_err(Into::into)
@@ -357,7 +352,6 @@ impl Backend {
reply_surbs: &ReceivedReplySurbsMap,
) -> Result<(), StorageError> {
self.manager
.get()
.insert_reply_surb_storage_metadata(ReplySurbStorageMetadata::new(
reply_surbs.min_surb_threshold(),
reply_surbs.max_surb_threshold(),
@@ -371,23 +365,23 @@ impl Backend {
impl ReplyStorageBackend for Backend {
type StorageError = error::StorageError;
async fn new(
debug_config: &crate::config::DebugConfig,
db_path: Option<PathBuf>,
) -> Result<Self, Self::StorageError> {
non_wasm_helpers::setup_fs_reply_surb_backend(db_path, debug_config)
.await
.map_err(|err| {
log::error!("Failed to create storage: {err}");
Self::StorageError::FailedToCreateStorage {
source: Box::new(err),
}
})
}
// async fn new(
// surb_config: &crate::config::ReplySurbs,
// db_path: Option<PathBuf>,
// ) -> Result<Self, Self::StorageError> {
// non_wasm_helpers::setup_fs_reply_surb_backend(db_path, surb_config)
// .await
// .map_err(|err| {
// log::error!("Failed to create storage: {err}");
// Self::StorageError::FailedToCreateStorage {
// source: Box::new(err),
// }
// })
// }
fn is_active(&self) -> bool {
self.manager.is_active()
}
// fn is_active(&self) -> bool {
// self.manager.is_active()
// }
async fn start_storage_session(&self) -> Result<(), Self::StorageError> {
self.start_client_use().await
@@ -426,17 +420,17 @@ impl ReplyStorageBackend for Backend {
Ok(CombinedReplyStorage::load(reply_keys, reply_surbs, tags))
}
fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
match self.manager {
StorageManagerState::Storage(_) => {
panic!("tried to get inactive storage from an active storage backend")
}
StorageManagerState::Inactive(ref state) => Ok(CombinedReplyStorage::new(
state.minimum_reply_surb_storage_threshold,
state.maximum_reply_surb_storage_threshold,
)),
}
}
// fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
// match self.manager {
// StorageManagerState::Storage(_) => {
// panic!("tried to get inactive storage from an active storage backend")
// }
// StorageManagerState::Inactive(ref state) => Ok(CombinedReplyStorage::new(
// state.minimum_reply_surb_storage_threshold,
// state.maximum_reply_surb_storage_threshold,
// )),
// }
// }
async fn stop_storage_session(self) -> Result<(), Self::StorageError> {
self.stop_client_use().await
common/client-core/src/client/replies/reply_storage/backend/mod.rs
+41 -32
View File
@@ -3,7 +3,7 @@
use crate::client::replies::reply_storage::CombinedReplyStorage;
use async_trait::async_trait;
use std::{error::Error, path::PathBuf};
use std::error::Error;
use thiserror::Error;
#[cfg(target_arch = "wasm32")]
@@ -26,23 +26,32 @@ pub struct Empty {
pub max_surb_threshold: usize,
}
impl Default for Empty {
fn default() -> Self {
Empty {
min_surb_threshold: 20,
max_surb_threshold: 200,
}
}
}
#[async_trait]
impl ReplyStorageBackend for Empty {
type StorageError = UndefinedError;
async fn new(
debug_config: &crate::config::DebugConfig,
_db_path: Option<PathBuf>,
) -> Result<Self, Self::StorageError> {
Ok(Self {
min_surb_threshold: debug_config
.reply_surbs
.minimum_reply_surb_storage_threshold,
max_surb_threshold: debug_config
.reply_surbs
.maximum_reply_surb_storage_threshold,
})
}
// async fn new(
// debug_config: &crate::config::DebugConfig,
// _db_path: Option<PathBuf>,
// ) -> Result<Self, Self::StorageError> {
// Ok(Self {
// min_surb_threshold: debug_config
// .reply_surbs
// .minimum_reply_surb_storage_threshold,
// max_surb_threshold: debug_config
// .reply_surbs
// .maximum_reply_surb_storage_threshold,
// })
// }
async fn flush_surb_storage(
&mut self,
@@ -65,26 +74,26 @@ impl ReplyStorageBackend for Empty {
))
}
fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
Ok(CombinedReplyStorage::new(
self.min_surb_threshold,
self.max_surb_threshold,
))
}
// fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError> {
// Ok(CombinedReplyStorage::new(
// self.min_surb_threshold,
// self.max_surb_threshold,
// ))
// }
}
#[async_trait]
pub trait ReplyStorageBackend: Sized {
type StorageError: Error + 'static;
async fn new(
debug_config: &crate::config::DebugConfig,
db_path: Option<PathBuf>,
) -> Result<Self, Self::StorageError>;
fn is_active(&self) -> bool {
true
}
// async fn new(
// surb_config: &crate::config::ReplySurbs,
// db_path: Option<PathBuf>,
// ) -> Result<Self, Self::StorageError>;
//
// fn is_active(&self) -> bool {
// true
// }
async fn start_storage_session(&self) -> Result<(), Self::StorageError> {
Ok(())
@@ -103,10 +112,10 @@ pub trait ReplyStorageBackend: Sized {
async fn load_surb_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError>;
/// In the case the storage backend is initialized in an inactive state (persisting data is
/// disabled), we might still need to fetch the (in-mem) storage and the parameters it was
/// created with.
fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError>;
// /// In the case the storage backend is initialized in an inactive state (persisting data is
// /// disabled), we might still need to fetch the (in-mem) storage and the parameters it was
// /// created with.
// fn get_inactive_storage(&self) -> Result<CombinedReplyStorage, Self::StorageError>;
async fn stop_storage_session(self) -> Result<(), Self::StorageError> {
Ok(())
common/client-core/src/init/helpers.rs
+19 -12
View File
@@ -30,10 +30,7 @@ type WsConn = WebSocketStream<MaybeTlsStream<TcpStream>>;
#[cfg(not(target_arch = "wasm32"))]
use crate::client::key_manager::persistence::OnDiskKeys;
#[cfg(not(target_arch = "wasm32"))]
use crate::{
client::key_manager::KeyManager,
config::{persistence::key_pathfinder::ClientKeyPathfinder, Config},
};
use crate::config::{persistence::key_pathfinder::ClientKeyPathfinder, Config};
#[cfg(not(target_arch = "wasm32"))]
use nym_config::NymConfig;
@@ -249,17 +246,27 @@ pub(super) async fn register_with_gateway<St: Storage>(
Ok(shared_keys)
}
// #[cfg(not(target_arch = "wasm32"))]
// pub(super) async fn store_keys_on_disk<T>(
// key_manager: &KeyManager,
// config: &Config<T>,
// ) -> Result<(), ClientCoreError>
// where
// T: NymConfig,
// {
// let pathfinder = ClientKeyPathfinder::new_from_config(config);
// Ok(key_manager
// .persist_keys(&OnDiskKeys::new(pathfinder))
// .await
// .tap_err(|err| log::error!("Failed to generate keys: {err}"))?)
// }
// TODO: make it generic
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn store_keys_on_disk<T>(
key_manager: &KeyManager,
config: &Config<T>,
) -> Result<(), ClientCoreError>
pub(super) fn on_disk_key_store<T>(config: &Config<T>) -> OnDiskKeys
where
T: NymConfig,
{
let pathfinder = ClientKeyPathfinder::new_from_config(config);
Ok(key_manager
.persist_keys(&OnDiskKeys::new(pathfinder))
.await
.tap_err(|err| log::error!("Failed to generate keys: {err}"))?)
OnDiskKeys::new(pathfinder)
}
common/client-core/src/init/mod.rs
+19 -26
View File
@@ -3,20 +3,6 @@
//! Collection of initialization steps used by client implementations
use std::fmt::Display;
use std::sync::Arc;
use nym_sphinx::addressing::{clients::Recipient, nodes::NodeIdentity};
use rand::rngs::OsRng;
use serde::Serialize;
use tap::TapFallible;
use nym_config::NymConfig;
use nym_credential_storage::storage::Storage;
use nym_crypto::asymmetric::{encryption, identity};
use nym_gateway_requests::registration::handshake::SharedKeys;
use url::Url;
use crate::client::key_manager::{KeyManager, KeyManagerBuilder};
use crate::{
config::{
@@ -25,6 +11,16 @@ use crate::{
},
error::ClientCoreError,
};
use nym_config::NymConfig;
use nym_credential_storage::storage::Storage;
use nym_crypto::asymmetric::{encryption, identity};
use nym_gateway_requests::registration::handshake::SharedKeys;
use nym_sphinx::addressing::{clients::Recipient, nodes::NodeIdentity};
use serde::Serialize;
use std::fmt::Display;
use std::sync::Arc;
use tap::TapFallible;
use url::Url;
mod helpers;
@@ -66,12 +62,6 @@ impl Display for InitResults {
}
}
/// Create a new set of client keys.
pub fn new_client_keys() -> KeyManagerBuilder {
let mut rng = OsRng;
KeyManagerBuilder::new(&mut rng)
}
/// Authenticate and register with a gateway.
/// Either pick one at random by querying the available gateways from the nym-api, or use the
/// chosen one if it's among the available ones.
@@ -139,18 +129,21 @@ where
return Ok(gateway.into());
}
let key_store = helpers::on_disk_key_store(config);
let mut rng = rand::thread_rng();
let mut managed_keys =
crate::client::key_manager::ManagedKeys::load_or_generate(&mut rng, &key_store).await;
// Create new keys and derive our identity
let key_manager_builder = new_client_keys();
let our_identity = key_manager_builder.identity_keypair();
let our_identity = managed_keys.identity_keypair();
// Establish connection, authenticate and generate keys for talking with the gateway
eprintln!("Registering with new gateway");
let shared_keys = helpers::register_with_gateway::<St>(&gateway, our_identity).await?;
let key_manager = key_manager_builder.insert_gateway_shared_key(shared_keys);
managed_keys
.deal_with_gateway_key(shared_keys, &key_store)
.await?;
// Write all keys to storage and just return the gateway endpoint config. It is assumed that we
// will load keys from storage when actually connecting.
helpers::store_keys_on_disk(&key_manager, config).await?;
Ok(gateway.into())
}
common/socks5-client-core/src/lib.rs
+8 -19
View File
@@ -24,7 +24,7 @@ use std::error::Error;
#[cfg(target_os = "android")]
use nym_client_core::client::{
base_client::helpers::setup_empty_reply_surb_backend,
base_client::helpers::setup_empty_reply_surb_backend, base_client::storage,
key_manager::persistence::InMemEphemeralKeys, replies::reply_storage,
};
#[cfg(target_os = "android")]
@@ -32,8 +32,8 @@ use nym_credential_storage::ephemeral_storage::EphemeralStorage;
#[cfg(not(target_os = "android"))]
use nym_client_core::client::{
base_client::non_wasm_helpers, key_manager::persistence::OnDiskKeys,
replies::reply_storage::fs_backend,
base_client::non_wasm_helpers, base_client::storage::OnDiskPersistent,
key_manager::persistence::OnDiskKeys, replies::reply_storage::fs_backend,
};
#[cfg(not(target_os = "android"))]
use nym_credential_storage::persistent_storage::PersistentStorage;
@@ -47,22 +47,11 @@ pub type Socks5ControlMessageSender = mpsc::UnboundedSender<Socks5ControlMessage
pub type Socks5ControlMessageReceiver = mpsc::UnboundedReceiver<Socks5ControlMessage>;
#[cfg(target_os = "android")]
type AndroidSocks5ClientBuilder<'a> = BaseClientBuilder<
'a,
reply_storage::Empty,
Client<QueryNyxdClient>,
InMemEphemeralKeys,
EphemeralStorage,
>;
type AndroidSocks5ClientBuilder<'a> =
BaseClientBuilder<'a, Client<QueryNyxdClient>, storage::Ephemeral>;
#[cfg(not(target_os = "android"))]
type Socks5ClientBuilder<'a> = BaseClientBuilder<
'a,
fs_backend::Backend,
Client<QueryNyxdClient>,
OnDiskKeys,
PersistentStorage,
>;
type Socks5ClientBuilder<'a> = BaseClientBuilder<'a, Client<QueryNyxdClient>, OnDiskPersistent>;
#[derive(Debug)]
pub enum Socks5ControlMessage {
@@ -251,8 +240,8 @@ impl NymClient {
&self,
) -> Result<fs_backend::Backend, Socks5ClientCoreError> {
non_wasm_helpers::setup_fs_reply_surb_backend(
Some(self.config.get_base().get_reply_surb_database_path()),
self.config.get_debug_settings(),
self.config.get_base().get_reply_surb_database_path(),
&self.config.get_debug_settings().reply_surbs,
)
.await
.map_err(Into::into)