Cargo.toml

@@ -63,6 +63,8 @@ futures = "0.3"
 hex = { version = "0.4", features = ["serde"] }
 jsonrpsee = { version = "0.22", features = ["server", "client", "macros"] }
 openmls = "0.5"
+openmls_rust_crypto = "0.2"
+openmls_traits = "0.2"
 password-hash = "0.5"
 pbkdf2 = { version = "0.12", features = ["simple"] }
 rand = "0.8"

client/Cargo.toml

@@ -11,11 +11,14 @@ aspect-ident = { workspace = true }
 aspect-proto = { workspace = true }
 aspect-util = { workspace = true }
 
+anyhow = { workspace = true }
 async-trait = { workspace = true }
 ecksport-net = { workspace = true }
 ecksport-rpc = { workspace = true }
 futures = { workspace = true }
 openmls = { workspace = true }
+openmls_rust_crypto = { workspace = true }
+openmls_traits = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

client/src/infra/errors.rs

@@ -0,0 +1,20 @@
+use openmls::framing::errors::MlsMessageError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum InfraError {
+    #[error("mls msg: {0}")]
+    MlsMsg(#[from] MlsMessageError),
+
+    #[error("db: {0}")]
+    Db(#[from] aspect_db::errors::Error),
+
+    #[error("llmq: {0}")]
+    LlmqDb(#[from] aspect_db::llmq::errors::DbError),
+
+    #[error("not yet implemented")]
+    Unimplemented,
+
+    #[error("{0}")]
+    Other(String),
+}

client/src/infra/group_io.rs

@@ -0,0 +1,69 @@
+//! Utils for supporting a group emitting application-layer messages.
+
+use openmls::framing::MlsMessageOut;
+use openmls::group::GroupId;
+
+use aspect_db::llmq::query::MqDatastore;
+use aspect_db::{
+    generic_msg_queue::QueueId,
+    group::{
+        traits::GroupOutputDb,
+        types::{AppMsg, MsgEntry},
+    },
+};
+
+use super::errors::InfraError;
+
+/// Trait used to write messages to the group's message log.
+pub trait GroupOutput {
+    /// Writes a standard application message.
+    fn write_msg(&self, msg: AppMsg) -> Result<(), InfraError>;
+}
+
+/// Group output that writes to a database implementing the standard trait.
+pub struct DbGroupOutput<D: GroupOutputDb> {
+    group_id: GroupId,
+    db: D,
+}
+
+impl<D: GroupOutputDb> DbGroupOutput<D> {
+    pub fn new(group_id: GroupId, db: D) -> Self {
+        Self { group_id, db }
+    }
+}
+
+impl<D: GroupOutputDb> GroupOutput for DbGroupOutput<D> {
+    fn write_msg(&self, msg: AppMsg) -> Result<(), InfraError> {
+        self.db
+            .write_next_entry(&self.group_id, MsgEntry::App(msg))?;
+        Ok(())
+    }
+}
+
+/// Broadcast trait used to send messages to the delivery service.
+pub trait GroupBroadcast {
+    /// Broadcasts an MLS message out to the delivery service.
+    fn broadcast_mls_out(&self, out_msg: MlsMessageOut) -> Result<(), InfraError>;
+}
+
+/// Stores the broadcasted messages in an LLMQ database.
+// TODO have some notification system so we can trigger the outbound code
+// to do something?
+pub struct MqBroadcastStore {
+    queue_id: QueueId,
+    mqds: MqDatastore,
+}
+
+impl MqBroadcastStore {
+    pub fn new(queue_id: QueueId, mqds: MqDatastore) -> Self {
+        Self { queue_id, mqds }
+    }
+}
+
+impl GroupBroadcast for MqBroadcastStore {
+    fn broadcast_mls_out(&self, out_msg: MlsMessageOut) -> Result<(), InfraError> {
+        let raw = out_msg.to_bytes()?;
+        self.mqds.submit_generic_msg(&self.queue_id, raw)?;
+        Ok(())
+    }
+}

client/src/infra/message_batch.rs

@@ -0,0 +1,29 @@
+use openmls::framing::ProtocolMessage;
+
+/// Batch of inbound messages that we process all at once to avoid excessive disk IO.
+#[derive(Clone)]
+pub struct MessageBatch {
+    protocol_messages: Vec<ProtocolMessage>,
+}
+
+impl MessageBatch {
+    pub fn new(protocol_messages: Vec<ProtocolMessage>) -> Self {
+        if protocol_messages.is_empty() {
+            panic!("message_batch: must be nonempty list of messages");
+        }
+
+        let pm0 = &protocol_messages[0];
+        if !protocol_messages[1..]
+            .iter()
+            .all(|pm| pm.group_id() == pm0.group_id())
+        {
+            panic!("message_batch: not all messages for same group");
+        }
+
+        Self { protocol_messages }
+    }
+
+    pub fn into_inner(self) -> Vec<ProtocolMessage> {
+        self.protocol_messages
+    }
+}

client/src/infra/mls_provider.rs

@@ -0,0 +1,35 @@
+use openmls_traits::{key_store::OpenMlsKeyStore, OpenMlsCryptoProvider};
+
+/// Shim OpenMLS crypto provider impl around a keystore, which is the only
+/// component that needs external data.
+pub struct SimpleMlsProvder<K> {
+    rust_crypto: openmls_rust_crypto::RustCrypto,
+    keystore: K,
+}
+
+impl<K> SimpleMlsProvder<K> {
+    pub fn new(keystore: K) -> Self {
+        Self {
+            rust_crypto: openmls_rust_crypto::RustCrypto::default(),
+            keystore,
+        }
+    }
+}
+
+impl<K: OpenMlsKeyStore> OpenMlsCryptoProvider for SimpleMlsProvder<K> {
+    type CryptoProvider = openmls_rust_crypto::RustCrypto;
+    type RandProvider = openmls_rust_crypto::RustCrypto;
+    type KeyStoreProvider = K;
+
+    fn crypto(&self) -> &Self::CryptoProvider {
+        &self.rust_crypto
+    }
+
+    fn rand(&self) -> &Self::RandProvider {
+        &self.rust_crypto
+    }
+
+    fn key_store(&self) -> &Self::KeyStoreProvider {
+        &self.keystore
+    }
+}

client/src/infra/mod.rs

@@ -0,0 +1,7 @@
+//! Logic for implementing an active instance of the client (workers, etc.).
+
+pub mod group_io;
+pub mod message_batch;
+pub mod mls_provider;
+
+pub mod errors;