Regularly rotate leaf node encryption keys

bindings_ffi/src/mls.rs

@@ -2512,6 +2512,9 @@ mod tests {
         let bo_group = bo.group(alix_group.id()).unwrap();
 
         bo_group.send("bo1".as_bytes().to_vec()).await.unwrap();
+        // Temporary workaround for OpenMLS issue - make sure Alix's epoch is up-to-date
+        // https://github.com/xmtp/libxmtp/issues/1116
+        alix_group.sync().await.unwrap();
         alix_group.send("alix1".as_bytes().to_vec()).await.unwrap();
 
         // Move the group forward by 3 epochs (as Alix's max_past_epochs is
@@ -2715,6 +2718,9 @@ mod tests {
         log::info!("Caro sending fifth message");
         // Caro sends a message in the group
         caro_group.update_installations().await.unwrap();
+        // Temporary workaround for OpenMLS issue - make sure Caro's epoch is up-to-date
+        // https://github.com/xmtp/libxmtp/issues/1116
+        caro_group.sync().await.unwrap();
         caro_group
             .send("Fifth message".as_bytes().to_vec())
             .await

xmtp_mls/migrations/2024-10-03-004750_add_rotated_at_ns/down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE GROUPS
+    DROP COLUMN rotated_at_ns;
+

xmtp_mls/migrations/2024-10-03-004750_add_rotated_at_ns/up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE GROUPS
+    ADD COLUMN rotated_at_ns BIGINT NOT NULL DEFAULT 0;
+

xmtp_mls/migrations/README.md

@@ -22,4 +22,4 @@ Edit the `up.sql` and `down.sql` files created
 cargo run --bin update-schema
 ```
 
-This updates the generated `schema.rs` file. You can now update the models and queries to reference it in `xmtp_mls/src/storage/encrypted_store/`.
+Make sure you run this from `xmtp_mls/`. This updates the generated `schema.rs` file. You can now update the models and queries to reference it in `xmtp_mls/src/storage/encrypted_store/`.

xmtp_mls/src/client.rs

@@ -456,7 +456,7 @@ where
         &self.context
     }
 
-    ///
+    #[allow(clippy::borrowed_box)]
     pub fn smart_contract_signature_verifier(&self) -> &Box<dyn SmartContractSignatureVerifier> {
         &self.scw_verifier
     }

xmtp_mls/src/configuration.rs

@@ -16,6 +16,10 @@ const NS_IN_SEC: i64 = 1_000_000_000;
 
 const NS_IN_HOUR: i64 = NS_IN_SEC * 60 * 60;
 
+const NS_IN_DAY: i64 = NS_IN_HOUR * 24;
+
+pub const GROUP_KEY_ROTATION_INTERVAL_NS: i64 = 30 * NS_IN_DAY;
+
 pub const SYNC_UPDATE_INSTALLATIONS_INTERVAL_NS: i64 = NS_IN_HOUR / 2; // 30 min
 
 pub const SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS: i64 = 5 * NS_IN_SEC;

xmtp_mls/src/groups/intents.rs

@@ -26,6 +26,11 @@ use xmtp_proto::xmtp::mls::database::{
 };
 
 use crate::{
+    configuration::GROUP_KEY_ROTATION_INTERVAL_NS,
+    storage::{
+        db_connection::DbConnection,
+        group_intent::{IntentKind, NewGroupIntent, StoredGroupIntent},
+    },
     types::Address,
     verified_key_package_v2::{KeyPackageVerificationError, VerifiedKeyPackageV2},
 };
@@ -34,6 +39,7 @@ use super::{
     group_membership::GroupMembership,
     group_mutable_metadata::MetadataField,
     group_permissions::{MembershipPolicies, MetadataPolicies, PermissionsPolicies},
+    GroupError, MlsGroup,
 };
 
 #[derive(Debug, Error)]
@@ -48,6 +54,52 @@ pub enum IntentError {
     Generic(String),
 }
 
+impl MlsGroup {
+    pub fn queue_intent(
+        &self,
+        intent_kind: IntentKind,
+        intent_data: Vec<u8>,
+    ) -> Result<StoredGroupIntent, GroupError> {
+        self.context.store.transaction(|provider| {
+            let conn = provider.conn_ref();
+            self.queue_intent_with_conn(conn, intent_kind, intent_data)
+        })
+    }
+
+    pub fn queue_intent_with_conn(
+        &self,
+        conn: &DbConnection,
+        intent_kind: IntentKind,
+        intent_data: Vec<u8>,
+    ) -> Result<StoredGroupIntent, GroupError> {
+        if intent_kind == IntentKind::SendMessage {
+            self.maybe_insert_key_update_intent(conn)?;
+        }
+
+        let intent = conn.insert_group_intent(NewGroupIntent::new(
+            intent_kind,
+            self.group_id.clone(),
+            intent_data,
+        ))?;
+
+        if intent_kind != IntentKind::SendMessage {
+            conn.update_rotated_at_ns(self.group_id.clone())?;
+        }
+
+        Ok(intent)
+    }
+
+    fn maybe_insert_key_update_intent(&self, conn: &DbConnection) -> Result<(), GroupError> {
+        let last_rotated_at_ns = conn.get_rotated_at_ns(self.group_id.clone())?;
+        let now_ns = crate::utils::time::now_ns();
+        let elapsed_ns = now_ns - last_rotated_at_ns;
+        if elapsed_ns > GROUP_KEY_ROTATION_INTERVAL_NS {
+            self.queue_intent_with_conn(conn, IntentKind::KeyUpdate, vec![])?;
+        }
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct SendMessageIntentData {
     pub message: Vec<u8>,
@@ -669,6 +721,15 @@ impl TryFrom<Vec<u8>> for PostCommitAction {
 
 #[cfg(test)]
 mod tests {
+    use openmls::prelude::{MlsMessageBodyIn, MlsMessageIn, ProcessedMessageContent};
+    use tls_codec::Deserialize;
+    use xmtp_cryptography::utils::generate_local_wallet;
+    use xmtp_proto::xmtp::mls::api::v1::{group_message, GroupMessage};
+
+    use crate::{
+        builder::ClientBuilder, groups::GroupMetadataOptions, utils::test::TestClient, Client,
+    };
+
     use super::*;
 
     #[test]
@@ -709,4 +770,107 @@ mod tests {
 
         assert_eq!(intent.field_value, restored_intent.field_value);
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_key_rotation_before_first_message() {
+        let client_a = ClientBuilder::new_test_client(&generate_local_wallet()).await;
+        let client_b = ClientBuilder::new_test_client(&generate_local_wallet()).await;
+
+        // client A makes a group with client B, and then sends a message to client B.
+        let group_a = client_a
+            .create_group(None, GroupMetadataOptions::default())
+            .expect("create group");
+        group_a
+            .add_members_by_inbox_id(&client_a, vec![client_b.inbox_id()])
+            .await
+            .unwrap();
+        group_a
+            .send_message(b"First message from A", &client_a)
+            .await
+            .unwrap();
+
+        // No key rotation needed, because A's commit to add B already performs a rotation.
+        // Group should have a commit to add client B, followed by A's message.
+        verify_num_payloads_in_group(&client_a, &group_a, 2).await;
+
+        // Client B sends a message to Client A
+        let groups_b = client_b.sync_welcomes().await.unwrap();
+        assert_eq!(groups_b.len(), 1);
+        let group_b = groups_b[0].clone();
+        group_b
+            .send_message(b"First message from B", &client_b)
+            .await
+            .expect("send message");
+
+        // B must perform a key rotation before sending their first message.
+        // Group should have a commit to add B, A's message, B's key rotation and then B's message.
+        let payloads_a = verify_num_payloads_in_group(&client_a, &group_a, 4).await;
+        let payloads_b = verify_num_payloads_in_group(&client_b, &group_b, 4).await;
+
+        // Verify key rotation payload
+        for i in 0..payloads_a.len() {
+            assert_eq!(payloads_a[i].encode_to_vec(), payloads_b[i].encode_to_vec());
+        }
+        verify_commit_updates_leaf_node(&client_a, &group_a, &payloads_a[2]);
+
+        // Client B sends another message to Client A, and Client A sends another message to Client B.
+        group_b
+            .send_message(b"Second message from B", &client_b)
+            .await
+            .expect("send message");
+        group_a
+            .send_message(b"Second message from A", &client_a)
+            .await
+            .expect("send message");
+
+        // Group should only have 2 additional messages - no more key rotations needed.
+        verify_num_payloads_in_group(&client_a, &group_a, 6).await;
+        verify_num_payloads_in_group(&client_b, &group_b, 6).await;
+    }
+
+    async fn verify_num_payloads_in_group(
+        client: &Client<TestClient>,
+        group: &MlsGroup,
+        num_messages: usize,
+    ) -> Vec<GroupMessage> {
+        let messages = client
+            .api_client
+            .query_group_messages(group.group_id.clone(), None)
+            .await
+            .unwrap();
+        assert_eq!(messages.len(), num_messages);
+        messages
+    }
+
+    fn verify_commit_updates_leaf_node(
+        client: &Client<TestClient>,
+        group: &MlsGroup,
+        payload: &GroupMessage,
+    ) {
+        let msgv1 = match &payload.version {
+            Some(group_message::Version::V1(value)) => value,
+            _ => panic!("error msgv1"),
+        };
+
+        let mls_message_in = MlsMessageIn::tls_deserialize_exact(&msgv1.data).unwrap();
+        let mls_message = match mls_message_in.extract() {
+            MlsMessageBodyIn::PrivateMessage(mls_message) => mls_message,
+            _ => panic!("error mls_message"),
+        };
+
+        let provider = client.mls_provider().unwrap();
+        let mut openmls_group = group.load_mls_group(&provider).unwrap();
+        let decrypted_message = openmls_group
+            .process_message(&provider, mls_message)
+            .unwrap();
+
+        let staged_commit = match decrypted_message.into_content() {
+            ProcessedMessageContent::StagedCommitMessage(staged_commit) => *staged_commit,
+            _ => panic!("error staged_commit"),
+        };
+
+        // Check there is indeed some updated leaf node, which means the key update works.
+        let path_update_leaf_node = staged_commit.update_path_leaf_node();
+        assert!(path_update_leaf_node.is_some());
+    }
 }

xmtp_mls/src/groups/message_history.rs

@@ -265,7 +265,7 @@ where
             })?;
 
         // publish the intent
-        if let Err(err) = sync_group.publish_intents(&conn.into(), self).await {
+        if let Err(err) = sync_group.publish_messages(self).await {
             tracing::error!("error publishing sync group intents: {:?}", err);
         }
         Ok(())