Redis client

rust/xaynet-core/src/crypto/encrypt.rs

@@ -13,7 +13,7 @@ use super::ByteObject;
 /// Number of additional bytes in a ciphertext compared to the corresponding plaintext.
 pub const SEALBYTES: usize = sealedbox::SEALBYTES;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 /// A `C25519` key pair for asymmetric authenticated encryption.
 pub struct EncryptKeyPair {
     /// The `C25519` public key.

rust/xaynet-core/src/mask/object/mod.rs

@@ -18,7 +18,7 @@ use crate::mask::config::MaskConfig;
 /// Errors related to invalid mask objects.
 pub struct InvalidMaskObjectError;
 
-#[derive(Debug, Hash, PartialEq, Eq, Clone)]
+#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
 /// A mask object which represents either a mask or a masked model.
 pub struct MaskObject {
     pub data: Vec<BigUint>,

rust/xaynet-core/src/mask/seed.rs

@@ -69,7 +69,7 @@ impl MaskSeed {
     }
 }
 
-#[derive(AsRef, AsMut, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(AsRef, AsMut, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 /// An encrypted mask seed.
 pub struct EncryptedMaskSeed(Vec<u8>);
 

rust/xaynet-server/Cargo.toml

@@ -45,7 +45,7 @@ config = "0.10.1"
 validator = "0.10"
 validator_derive = "0.10"
 structopt = "0.3"
-paste = "0.1.15"
+paste = "1.0.0"
 tower = "0.3.1"
 tracing = "0.1.16"
 tracing-futures = "0.2.4"
@@ -55,7 +55,7 @@ rayon = "1.3.0"
 async-trait = "0.1.35"
 xaynet-macros = { path = "../xaynet-macros" }
 xaynet-core = { path = "../xaynet-core" }
-redis = { version = "0.17.0", default-features = false, features = ["tokio-rt-core"] }
+redis = { version = "0.17.0", default-features = false, features = ["connection-manager", "aio", "tokio-rt-core"] }
 
 # optional dependencies
 influxdb = { version = "0.1.0", features = ["derive"], optional = true }

rust/xaynet-server/src/lib.rs

@@ -93,6 +93,7 @@ pub mod rest;
 pub mod services;
 pub mod settings;
 pub mod state_machine;
+pub mod storage;
 pub mod utils;
 pub(crate) mod vendor;
 

rust/xaynet-server/src/state_machine/coordinator.rs

@@ -10,7 +10,7 @@ use xaynet_core::{
 use crate::settings::{MaskSettings, ModelSettings, PetSettings};
 
 /// The coordinator state.
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Serialize, Deserialize)]
 pub struct CoordinatorState {
     /// The credentials of the coordinator.
     pub keys: EncryptKeyPair,

rust/xaynet-server/src/state_machine/mod.rs

@@ -259,4 +259,4 @@ where
 }
 
 #[cfg(test)]
-mod tests;
+pub(crate) mod tests;

rust/xaynet-server/src/storage/impls.rs

@@ -0,0 +1,226 @@
+use crate::state_machine::coordinator::CoordinatorState;
+use derive_more::{From, Into};
+use paste::paste;
+use redis::{ErrorKind, FromRedisValue, RedisError, RedisResult, RedisWrite, ToRedisArgs, Value};
+use xaynet_core::{
+    crypto::{ByteObject, PublicEncryptKey, PublicSigningKey},
+    mask::{EncryptedMaskSeed, MaskObject},
+};
+
+fn redis_type_error(desc: &'static str, details: Option<String>) -> RedisError {
+    if let Some(details) = details {
+        RedisError::from((ErrorKind::TypeError, desc, details))
+    } else {
+        RedisError::from((ErrorKind::TypeError, desc))
+    }
+}
+
+/// Implements ['FromRedisValue'] and ['ToRedisArgs'] for types that implement ['ByteObject'].
+/// The Redis traits as well as the crypto types are both defined in foreign crates.
+/// To bypass the restrictions of orphan rule, we use `Newtypes` for the crypto types.
+///
+/// Each crypto type has two `Newtypes`, one for reading and one for writing.
+/// The difference between `Read` and `Write` is that the write `Newtype` does not take the
+/// ownership of the value but only a reference. This allows us to use references in the
+/// [`Client`] methods. The `Read` Newtype also implements [`ToRedisArgs`] to reduce the
+/// conversion overhead that you would get if you wanted to reuse a `Read` value for another
+/// Redis query.
+///
+/// Example:
+///
+/// ```ignore
+/// let sum_pks: Vec<PublicSigningKeyRead> = self.connection.hkeys("sum_dict").await?;
+/// for sum_pk in sum_pks {
+///    let sum_pk_seed_dict: HashMap<PublicSigningKeyRead, EncryptedMaskSeedRead>
+///       = self.connection.hgetall(&sum_pk).await?; // no need to convert sum_pk from PublicSigningKeyRead to PublicSigningKeyWrite
+/// }
+/// ```
+///
+/// [`Client`]: crate::storage::redis::Client
+macro_rules! impl_byte_object_redis_traits {
+    ($ty: ty) => {
+        paste! {
+            #[derive(Into, Hash, Eq, PartialEq)]
+            pub(crate) struct [<$ty Read>]($ty);
+
+            impl FromRedisValue for [<$ty Read>] {
+                fn from_redis_value(v: &Value) -> RedisResult<[<$ty Read>]> {
+                    match *v {
+                        Value::Data(ref bytes) => {
+                            let inner = <$ty>::from_slice(bytes).ok_or_else(|| {
+                                redis_type_error(concat!("Invalid ", stringify!($ty)), None)
+                            })?;
+                            Ok([<$ty Read>](inner))
+                        }
+                        _ => Err(redis_type_error(
+                            concat!("Response not ", stringify!($ty), " compatible"),
+                            None,
+                        )),
+                    }
+                }
+            }
+
+            impl ToRedisArgs for [<$ty Read>] {
+                fn write_redis_args<W>(&self, out: &mut W)
+                where
+                    W: ?Sized + RedisWrite,
+                {
+                    self.0.as_slice().write_redis_args(out)
+                }
+            }
+
+            impl<'a> ToRedisArgs for &'a [<$ty Read>] {
+                fn write_redis_args<W>(&self, out: &mut W)
+                where
+                    W: ?Sized + RedisWrite,
+                {
+                    self.0.as_slice().write_redis_args(out)
+                }
+            }
+
+            #[derive(From)]
+            pub(crate) struct [<$ty Write>]<'a>(&'a $ty);
+
+            impl ToRedisArgs for [<$ty Write>]<'_> {
+                fn write_redis_args<W>(&self, out: &mut W)
+                where
+                    W: ?Sized + RedisWrite,
+                {
+                    self.0.as_slice().write_redis_args(out)
+                }
+            }
+
+            impl<'a> ToRedisArgs for &'a [<$ty Write>]<'a> {
+                fn write_redis_args<W>(&self, out: &mut W)
+                where
+                    W: ?Sized + RedisWrite,
+                {
+                    self.0.as_slice().write_redis_args(out)
+                }
+            }
+        }
+    };
+}
+
+impl_byte_object_redis_traits!(PublicEncryptKey);
+impl_byte_object_redis_traits!(PublicSigningKey);
+impl_byte_object_redis_traits!(EncryptedMaskSeed);
+
+/// Implements ['FromRedisValue'] and ['ToRedisArgs'] for types that implement
+/// ['Serialize`] and [`Deserialize']. The data is de/serialized via bincode.
+///
+/// # Panics
+///
+/// `write_redis_args` will panic if the data cannot be serialized with `bincode`
+///
+/// More information about what can cause a panic in bincode:
+/// - https://github.com/servo/bincode/issues/293
+/// - https://github.com/servo/bincode/issues/255
+/// - https://github.com/servo/bincode/issues/130#issuecomment-284641263
+macro_rules! impl_bincode_redis_traits {
+    ($ty: ty) => {
+        impl FromRedisValue for $ty {
+            fn from_redis_value(v: &Value) -> RedisResult<$ty> {
+                match *v {
+                    Value::Data(ref bytes) => bincode::deserialize(bytes).map_err(|e| {
+                        redis_type_error("Invalid CoordinatorState", Some(e.to_string()))
+                    }),
+                    _ => Err(redis_type_error(
+                        "Response not CoordinatorState compatible",
+                        None,
+                    )),
+                }
+            }
+        }
+
+        impl ToRedisArgs for $ty {
+            fn write_redis_args<W>(&self, out: &mut W)
+            where
+                W: ?Sized + RedisWrite,
+            {
+                let data = bincode::serialize(self).unwrap();
+                data.write_redis_args(out)
+            }
+        }
+
+        impl<'a> ToRedisArgs for &'a $ty {
+            fn write_redis_args<W>(&self, out: &mut W)
+            where
+                W: ?Sized + RedisWrite,
+            {
+                (*self).write_redis_args(out)
+            }
+        }
+    };
+}
+
+// CoordinatorState is pretty straightforward:
+// - all the sequences have known length (
+// - no untagged enum
+// so bincode will not panic.
+impl_bincode_redis_traits!(CoordinatorState);
+
+#[derive(From, Into, Serialize, Deserialize)]
+pub(crate) struct MaskObjectRead(MaskObject);
+
+impl_bincode_redis_traits!(MaskObjectRead);
+
+#[derive(From, Serialize)]
+pub(crate) struct MaskObjectWrite<'a>(&'a MaskObject);
+
+impl ToRedisArgs for MaskObjectWrite<'_> {
+    fn write_redis_args<W>(&self, out: &mut W)
+    where
+        W: ?Sized + RedisWrite,
+    {
+        let data = bincode::serialize(self).unwrap();
+        data.write_redis_args(out)
+    }
+}
+
+impl<'a> ToRedisArgs for &'a MaskObjectWrite<'a> {
+    fn write_redis_args<W>(&self, out: &mut W)
+    where
+        W: ?Sized + RedisWrite,
+    {
+        (*self).write_redis_args(out)
+    }
+}
+
+#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]
+pub enum AddSumParticipant {
+    Ok,
+    AlreadyExists,
+}
+
+impl FromRedisValue for AddSumParticipant {
+    fn from_redis_value(v: &Value) -> RedisResult<AddSumParticipant> {
+        match *v {
+            Value::Int(0) => Ok(AddSumParticipant::AlreadyExists),
+            Value::Int(1) => Ok(AddSumParticipant::Ok),
+            _ => Err(redis_type_error(
+                "Response status not valid integer",
+                Some(format!("Response was {:?}", v)),
+            )),
+        }
+    }
+}
+
+#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]
+pub enum DeleteSumParticipant {
+    Ok,
+    DoesNotExist,
+}
+
+impl FromRedisValue for DeleteSumParticipant {
+    fn from_redis_value(v: &Value) -> RedisResult<DeleteSumParticipant> {
+        match *v {
+            Value::Int(0) => Ok(DeleteSumParticipant::DoesNotExist),
+            Value::Int(1) => Ok(DeleteSumParticipant::Ok),
+            _ => Err(redis_type_error(
+                "Response status not valid integer",
+                Some(format!("Response was {:?}", v)),
+            )),
+        }
+    }
+}

rust/xaynet-server/src/storage/mod.rs

@@ -0,0 +1,4 @@
+pub(crate) mod impls;
+pub mod redis;
+
+pub use self::impls::AddSumParticipant;