Feature/persistent gateway storage (#784)

* Sqlx struct stub

* Initial schema

* Initial error enum

* Managed for persisted shared keys

* Initial inbox manager

* Comments

* Using new database in clients handler

* Extending gateway storage API

* tokio::main + placeholder values

* Removed old client store

* Simplified logic of async packet processing

* Renamed table + not null restriction

* BandwidthManager

* Removed sled dependency

* Using centralised storage for bandwidth

* Dead code removal

* WIP connection_handler split and simplification

Maybe it doesn't look like it right now, but once completed it will remove bunch of redundant checks for Nones etc

* Further more explicit clients handler split

* Minor cleanup

* Temporary store for active client handles

* Fixed error types

* Error trait on iv and encrypted address

* Authentication and registration moved to the handler

* Removal of clients handler

* Further logic simplification + returned explicit bandwidth values

* Further cleanup and comments

* Updated config with relevant changes

* Basic bandwidth tracking in client

* FreshHandle doc comments + fixed stagger issue

* Removed side-effects from .map

* More doc comments

* Database migration on build

* Increased default claimed bandwidth

* Renaming

* Fixed client determining available bandwidth

* Removed dead sql table that might be used in the future

* Windows workaround

* Comment

* Return error rather than cap credential

Author: jstuczyn

Cargo.lock

@@ -40,7 +40,7 @@ impl MixTrafficController {
     async fn on_messages(&mut self, mut mix_packets: Vec<MixPacket>) {
         debug_assert!(!mix_packets.is_empty());
 
-        let success = if mix_packets.len() == 1 {
+        let result = if mix_packets.len() == 1 {
             let mix_packet = mix_packets.pop().unwrap();
             self.gateway_client.send_mix_packet(mix_packet).await
         } else {
@@ -49,7 +49,7 @@ impl MixTrafficController {
                 .await
         };
 
-        match success {
+        match result {
             Err(e) => {
                 error!("Failed to send sphinx packet(s) to the gateway! - {:?}", e);
                 self.consecutive_gateway_failure_count += 1;

clients/client-core/src/client/mix_traffic.rs

@@ -36,8 +36,7 @@ const DEFAULT_RECONNECTION_BACKOFF: Duration = Duration::from_secs(5);
 
 pub struct GatewayClient {
     authenticated: bool,
-    // TODO: This should be replaced by an actual bandwidth value, with 0 meaning no bandwidth
-    has_bandwidth: bool,
+    bandwidth_remaining: i64,
     gateway_address: String,
     gateway_identity: identity::PublicKey,
     local_identity: Arc<identity::KeyPair>,
@@ -72,7 +71,7 @@ impl GatewayClient {
     ) -> Self {
         GatewayClient {
             authenticated: false,
-            has_bandwidth: false,
+            bandwidth_remaining: 0,
             gateway_address,
             gateway_identity,
             local_identity,
@@ -117,7 +116,7 @@ impl GatewayClient {
 
         GatewayClient {
             authenticated: false,
-            has_bandwidth: false,
+            bandwidth_remaining: 0,
             gateway_address,
             gateway_identity,
             local_identity,
@@ -474,22 +473,29 @@ impl GatewayClient {
         )
         .ok_or(GatewayClientError::SerializeCredential)?
         .into();
-        self.has_bandwidth = match self.send_websocket_message(msg).await? {
-            ServerResponse::Bandwidth { status } => Ok(status),
+        self.bandwidth_remaining = match self.send_websocket_message(msg).await? {
+            ServerResponse::Bandwidth { available_total } => Ok(available_total),
             ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
             _ => Err(GatewayClientError::UnexpectedResponse),
         }?;
         Ok(())
     }
 
+    fn estimate_required_bandwidth(&self, packets: &[MixPacket]) -> i64 {
+        packets
+            .iter()
+            .map(|packet| packet.sphinx_packet().len())
+            .sum::<usize>() as i64
+    }
+
     pub async fn batch_send_mix_packets(
         &mut self,
         packets: Vec<MixPacket>,
     ) -> Result<(), GatewayClientError> {
         if !self.authenticated {
             return Err(GatewayClientError::NotAuthenticated);
         }
-        if !self.has_bandwidth {
+        if self.estimate_required_bandwidth(&packets) < self.bandwidth_remaining {
             return Err(GatewayClientError::NotEnoughBandwidth);
         }
         if !self.connection.is_established() {
@@ -550,7 +556,7 @@ impl GatewayClient {
         if !self.authenticated {
             return Err(GatewayClientError::NotAuthenticated);
         }
-        if !self.has_bandwidth {
+        if (mix_packet.sphinx_packet().len() as i64) > self.bandwidth_remaining {
             return Err(GatewayClientError::NotEnoughBandwidth);
         }
         if !self.connection.is_established() {
@@ -598,7 +604,7 @@ impl GatewayClient {
         if !self.authenticated {
             return Err(GatewayClientError::NotAuthenticated);
         }
-        if !self.has_bandwidth {
+        if self.bandwidth_remaining <= 0 {
             return Err(GatewayClientError::NotEnoughBandwidth);
         }
         if self.connection.is_partially_delegated() {

common/client-libs/gateway-client/src/client.rs

@@ -12,7 +12,7 @@ use crate::error::Error;
 use crate::utils::{obtain_aggregate_signature, prepare_credential_for_spending};
 use coconut_interface::{hash_to_scalar, Credential, Parameters, Signature, VerificationKey};
 
-const BANDWIDTH_VALUE: u64 = 1024 * 1024; // 1 MB
+const BANDWIDTH_VALUE: u64 = 10 * 1024 * 1024 * 1024; // 10 GB
 
 pub const PUBLIC_ATTRIBUTES: u32 = 1;
 pub const PRIVATE_ATTRIBUTES: u32 = 1;

common/credentials/src/bandwidth.rs

@@ -20,12 +20,13 @@ log = "0.4"
 pretty_env_logger = "0.4"
 rand = "0.7"
 serde = { version = "1.0.104", features = ["derive"] }
-sled = "0.34"
+thiserror = "1"
 tokio = { version = "1.4", features = [ "rt-multi-thread", "net", "signal", "fs" ] }
 tokio-util = { version = "0.6", features = [ "codec" ] }
 tokio-stream = { version = "0.1", features = [ "fs" ] }
 tokio-tungstenite = "0.14"
 url = { version = "2.2", features = [ "serde" ] }
+sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
 
 # internal
 coconut-interface = { path = "../common/coconut-interface" }
@@ -39,3 +40,7 @@ nymsphinx = { path = "../common/nymsphinx" }
 pemstore = { path = "../common/pemstore" }
 validator-client = { path = "../common/client-libs/validator-client" }
 version-checker = { path = "../common/version-checker" }
+
+[build-dependencies]
+tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] }
+sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }

gateway/Cargo.toml

@@ -0,0 +1,25 @@
+use sqlx::{Connection, SqliteConnection};
+use std::env;
+
+#[tokio::main]
+async fn main() {
+    let out_dir = env::var("OUT_DIR").unwrap();
+    let database_path = format!("{}/gateway-example.sqlite", out_dir);
+
+    let mut conn = SqliteConnection::connect(&*format!("sqlite://{}?mode=rwc", database_path))
+        .await
+        .expect("Failed to create SQLx database connection");
+
+    sqlx::migrate!("./migrations")
+        .run(&mut conn)
+        .await
+        .expect("Failed to perform SQLx migrations");
+
+    #[cfg(target_family = "unix")]
+    println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
+
+    #[cfg(target_family = "windows")]
+    // for some strange reason we need to add a leading `/` to the windows path even though it's
+    // not a valid windows path... but hey, it works...
+    println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
+}

gateway/build.rs

@@ -6,6 +6,7 @@ use crate::registration::handshake::shared_key::SharedKeys;
 use crypto::symmetric::stream_cipher;
 use nymsphinx::params::GatewayEncryptionAlgorithm;
 use nymsphinx::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH};
+use thiserror::Error;
 
 pub const ENCRYPTED_ADDRESS_SIZE: usize = DESTINATION_ADDRESS_LENGTH;
 
@@ -16,9 +17,11 @@ pub const ENCRYPTED_ADDRESS_SIZE: usize = DESTINATION_ADDRESS_LENGTH;
 #[derive(Debug, PartialEq, Eq, Hash, Clone, Copy)]
 pub struct EncryptedAddressBytes([u8; ENCRYPTED_ADDRESS_SIZE]);
 
-#[derive(Debug)]
+#[derive(Debug, Error)]
 pub enum EncryptedAddressConversionError {
-    DecodeError(bs58::decode::Error),
+    #[error("Failed to decode the encrypted address - {0}")]
+    DecodeError(#[from] bs58::decode::Error),
+    #[error("The decoded address has invalid length")]
     StringOfInvalidLengthError,
 }
 
@@ -54,10 +57,7 @@ impl EncryptedAddressBytes {
     pub fn try_from_base58_string<S: Into<String>>(
         val: S,
     ) -> Result<Self, EncryptedAddressConversionError> {
-        let decoded = match bs58::decode(val.into()).into_vec() {
-            Ok(decoded) => decoded,
-            Err(err) => return Err(EncryptedAddressConversionError::DecodeError(err)),
-        };
+        let decoded = bs58::decode(val.into()).into_vec()?;
 
         if decoded.len() != ENCRYPTED_ADDRESS_SIZE {
             return Err(EncryptedAddressConversionError::StringOfInvalidLengthError);

gateway/gateway-requests/src/authentication/encrypted_address.rs

@@ -5,19 +5,25 @@ use crypto::generic_array::{typenum::Unsigned, GenericArray};
 use crypto::symmetric::stream_cipher::{random_iv, NewCipher, IV as CryptoIV};
 use nymsphinx::params::GatewayEncryptionAlgorithm;
 use rand::{CryptoRng, RngCore};
+use thiserror::Error;
 
 type NonceSize = <GatewayEncryptionAlgorithm as NewCipher>::NonceSize;
 
 // I think 'IV' looks better than 'Iv', feel free to change that.
 #[allow(clippy::upper_case_acronyms)]
 pub struct IV(CryptoIV<GatewayEncryptionAlgorithm>);
 
-#[derive(Debug)]
+#[derive(Error, Debug)]
 // I think 'IV' looks better than 'Iv', feel free to change that.
 #[allow(clippy::upper_case_acronyms)]
 pub enum IVConversionError {
-    DecodeError(bs58::decode::Error),
+    #[error("Failed to decode the iv - {0}")]
+    DecodeError(#[from] bs58::decode::Error),
+
+    #[error("The decoded bytes iv has invalid length")]
     BytesOfInvalidLengthError,
+
+    #[error("The decoded string iv has invalid length")]
     StringOfInvalidLengthError,
 }
 
@@ -47,10 +53,7 @@ impl IV {
     }
 
     pub fn try_from_base58_string<S: Into<String>>(val: S) -> Result<Self, IVConversionError> {
-        let decoded = match bs58::decode(val.into()).into_vec() {
-            Ok(decoded) => decoded,
-            Err(err) => return Err(IVConversionError::DecodeError(err)),
-        };
+        let decoded = bs58::decode(val.into()).into_vec()?;
 
         if decoded.len() != NonceSize::to_usize() {
             return Err(IVConversionError::StringOfInvalidLengthError);

gateway/gateway-requests/src/iv.rs

@@ -22,7 +22,7 @@ type EncryptionKeySize = <GatewayEncryptionAlgorithm as NewCipher>::KeySize;
 /// Shared key used when computing MAC for messages exchanged between client and its gateway.
 pub type MacKey = GenericArray<u8, MacKeySize>;
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Copy, Debug)]
 pub struct SharedKeys {
     encryption_key: CipherKey<GatewayEncryptionAlgorithm>,
     mac_key: MacKey,

gateway/gateway-requests/src/registration/handshake/shared_key.rs

@@ -89,6 +89,8 @@ impl fmt::Display for GatewayRequestsError {
     }
 }
 
+impl std::error::Error for GatewayRequestsError {}
+
 impl From<NymNodeRoutingAddressError> for GatewayRequestsError {
     fn from(_: NymNodeRoutingAddressError) -> Self {
         GatewayRequestsError::IncorrectlyEncodedAddress
@@ -191,9 +193,8 @@ impl TryInto<String> for ClientControlRequest {
 pub enum ServerResponse {
     Authenticate { status: bool },
     Register { status: bool },
-    // Maybe we could return the remaining bandwidth?
-    Bandwidth { status: bool },
-    Send { status: bool },
+    Bandwidth { available_total: i64 },
+    Send { remaining_bandwidth: i64 },
     Error { message: String },
 }
 

gateway/gateway-requests/src/types.rs

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2021 - Nym Technologies SA <[email protected]>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+CREATE TABLE shared_keys
+(
+    client_address_bs58                      TEXT NOT NULL PRIMARY KEY UNIQUE,
+    derived_aes128_ctr_blake3_hmac_keys_bs58 TEXT NOT NULL
+);
+
+CREATE TABLE message_store
+(
+    id                  INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+    client_address_bs58 TEXT    NOT NULL,
+    content             BLOB    NOT NULL
+);
+
+CREATE TABLE available_bandwidth
+(
+    client_address_bs58 TEXT    NOT NULL PRIMARY KEY UNIQUE,
+    available           INTEGER NOT NULL
+);
+
+CREATE INDEX `message_store_index` ON `message_store` (`client_address_bs58`, `content`);

gateway/migrations/20210921120000_create_initial_tables.sql

@@ -44,15 +44,9 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name(INBOXES_ARG_NAME)
-                .long(INBOXES_ARG_NAME)
-                .help("Directory with inboxes where all packets for the clients are stored")
-                .takes_value(true)
-        )
-        .arg(
-            Arg::with_name(CLIENTS_LEDGER_ARG_NAME)
-                .long(CLIENTS_LEDGER_ARG_NAME)
-                .help("Ledger file containing registered clients")
+            Arg::with_name(DATASTORE_PATH)
+                .long(DATASTORE_PATH)
+                .help("Path to sqlite database containing all gateway persistent data")
                 .takes_value(true)
         )
         .arg(
@@ -115,7 +109,7 @@ fn show_bonding_info(config: &Config) {
     );
 }
 
-pub fn execute(matches: &ArgMatches) {
+pub async fn execute(matches: ArgMatches<'static>) {
     let id = matches.value_of(ID_ARG_NAME).unwrap();
     println!("Initialising gateway {}...", id);
 
@@ -128,7 +122,7 @@ pub fn execute(matches: &ArgMatches) {
 
     let mut config = Config::new(id);
 
-    config = override_config(config, matches);
+    config = override_config(config, &matches);
 
     // if gateway was already initialised, don't generate new keys
     if !already_init {

gateway/src/commands/init.rs

@@ -15,8 +15,7 @@ pub(crate) const MIX_PORT_ARG_NAME: &str = "mix-port";
 pub(crate) const CLIENTS_PORT_ARG_NAME: &str = "clients-port";
 pub(crate) const VALIDATORS_ARG_NAME: &str = "validators";
 pub(crate) const ANNOUNCE_HOST_ARG_NAME: &str = "announce-host";
-pub(crate) const INBOXES_ARG_NAME: &str = "inboxes";
-pub(crate) const CLIENTS_LEDGER_ARG_NAME: &str = "clients-ledger";
+pub(crate) const DATASTORE_PATH: &str = "datastore";
 
 fn parse_validators(raw: &str) -> Vec<Url> {
     raw.split(',')
@@ -69,12 +68,8 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
         config = config.with_custom_validator_apis(parse_validators(raw_validators));
     }
 
-    if let Some(inboxes_dir) = matches.value_of(INBOXES_ARG_NAME) {
-        config = config.with_custom_clients_inboxes(inboxes_dir);
-    }
-
-    if let Some(clients_ledger) = matches.value_of(CLIENTS_LEDGER_ARG_NAME) {
-        config = config.with_custom_clients_ledger(clients_ledger);
+    if let Some(datastore_path) = matches.value_of(DATASTORE_PATH) {
+        config = config.with_custom_persistent_store(datastore_path);
     }
 
     config