core_crypto/mls/conversation/

pending_conversation.rs

//! When a client joins a group via an external commit, it sometimes receives messages
//! (most of the time renewed external proposals) for the new epoch whereas it does not yet have
//! the confirmation from the DS that the external join commit has been accepted.

use super::Result;
use super::{ConversationWithMls, Error};
use crate::mls::conversation::conversation_guard::decrypt::buffer_messages::MessageRestorePolicy;
use crate::mls::credential::crl::{extract_crl_uris_from_group, get_new_crl_distribution_points};
use crate::mls::credential::ext::CredentialExt as _;
use crate::prelude::{
    ConversationId, MlsBufferedConversationDecryptMessage, MlsCommitBundle, MlsConversation,
    MlsConversationConfiguration, MlsConversationDecryptMessage, MlsCustomConfiguration,
};
use crate::transaction_context::TransactionContext;
use crate::{KeystoreError, LeafError, MlsError, MlsTransportResponse, RecursiveError};
use core_crypto_keystore::CryptoKeystoreMls as _;
use core_crypto_keystore::entities::{MlsPendingMessage, PersistedMlsPendingGroup};
use log::trace;
use mls_crypto_provider::{CryptoKeystore, MlsCryptoProvider};
use openmls::credentials::CredentialWithKey;
use openmls::prelude::{MlsGroup, MlsMessageIn, MlsMessageInBody};
use openmls_traits::OpenMlsCryptoProvider;
use tls_codec::Deserialize as _;

/// A pending conversation is a conversation that has been created via an external join commit
/// locally, while this commit has not yet been approved by the DS.
#[derive(Debug)]
pub struct PendingConversation {
    inner: PersistedMlsPendingGroup,
    context: TransactionContext,
}

impl PendingConversation {
    pub(crate) fn new(inner: PersistedMlsPendingGroup, context: TransactionContext) -> Self {
        Self { inner, context }
    }

    pub(crate) fn from_mls_group(
        group: MlsGroup,
        custom_cfg: MlsCustomConfiguration,
        context: TransactionContext,
    ) -> Result<Self> {
        let serialized_cfg = serde_json::to_vec(&custom_cfg).map_err(MlsError::wrap("serializing custom config"))?;
        let serialized_group =
            core_crypto_keystore::ser(&group).map_err(KeystoreError::wrap("serializing mls group"))?;
        let group_id = group.group_id().to_vec();

        let inner = PersistedMlsPendingGroup {
            id: group_id,
            state: serialized_group,
            custom_configuration: serialized_cfg,
            parent_id: None,
        };
        Ok(Self::new(inner, context))
    }

    async fn mls_provider(&self) -> Result<MlsCryptoProvider> {
        self.context
            .mls_provider()
            .await
            .map_err(RecursiveError::transaction("getting mls provider"))
            .map_err(Into::into)
    }

    async fn keystore(&self) -> Result<CryptoKeystore> {
        let backend = self.mls_provider().await?;
        Ok(backend.keystore())
    }

    fn id(&self) -> &ConversationId {
        &self.inner.id
    }

    pub(crate) async fn save(&self) -> Result<()> {
        let keystore = self.keystore().await?;
        keystore
            .mls_pending_groups_save(self.id(), &self.inner.state, &self.inner.custom_configuration, None)
            .await
            .map_err(KeystoreError::wrap("saving mls pending groups"))
            .map_err(Into::into)
    }

    /// Send the commit via [crate::MlsTransport] and handle the response.
    pub(crate) async fn send_commit(&self, commit: MlsCommitBundle) -> Result<()> {
        let transport = self
            .context
            .mls_transport()
            .await
            .map_err(RecursiveError::transaction("getting mls transport"))?;
        let transport = transport.as_ref().ok_or::<Error>(
            RecursiveError::root("getting mls transport")(crate::Error::MlsTransportNotProvided).into(),
        )?;

        match transport
            .send_commit_bundle(commit.clone())
            .await
            .map_err(RecursiveError::root("sending commit bundle"))?
        {
            MlsTransportResponse::Success => Ok(()),
            MlsTransportResponse::Abort { reason } => Err(Error::MessageRejected { reason }),
            MlsTransportResponse::Retry => Err(Error::CannotRetryWithoutConversation),
        }
    }

    /// If the given message is the commit generated by [TransactionContext::join_by_external_commit],
    /// merge the pending group and restore any buffered messages.
    ///
    /// Otherwise, the given message will be buffered.
    pub async fn try_process_own_join_commit(
        &mut self,
        message: impl AsRef<[u8]>,
    ) -> Result<MlsConversationDecryptMessage> {
        // If the confirmation tag of the pending group and this incoming message are identical, we can merge the pending group.
        if self.incoming_message_is_own_join_commit(message.as_ref()).await? {
            return self.merge_and_restore_messages().await;
        }

        let keystore = self.keystore().await?;

        let pending_msg = MlsPendingMessage {
            foreign_id: self.id().clone(),
            message: message.as_ref().to_vec(),
        };
        keystore
            .save::<MlsPendingMessage>(pending_msg)
            .await
            .map_err(KeystoreError::wrap("saving mls pending message"))?;
        Err(Error::BufferedForPendingConversation)
    }

    /// If the message confirmation tag and the group confirmation tag are the same, it means that
    /// the external join commit has been accepted by the DS and the pending group can be merged.
    async fn incoming_message_is_own_join_commit(&self, message: impl AsRef<[u8]>) -> Result<bool> {
        let backend = self.mls_provider().await?;
        let keystore = backend.keystore();
        // Instantiate the pending group
        let (group, _cfg) = keystore
            .mls_pending_groups_load(self.id())
            .await
            .map_err(KeystoreError::wrap("loading mls pending groups"))?;
        let mut mls_group = core_crypto_keystore::deser::<MlsGroup>(&group)
            .map_err(KeystoreError::wrap("deserializing mls pending groups"))?;

        // The commit is only merged on this temporary instance of the pending group, to enable
        // calculation of the confirmation tag.
        mls_group
            .merge_pending_commit(&backend)
            .await
            .map_err(MlsError::wrap("merging pending commit"))?;
        let message_in = MlsMessageIn::tls_deserialize(&mut message.as_ref())
            .map_err(MlsError::wrap("deserializing mls message"))?;
        let MlsMessageInBody::PublicMessage(public_message) = message_in.extract() else {
            return Ok(false);
        };
        let Some(msg_ct) = public_message.confirmation_tag() else {
            return Ok(false);
        };
        let group_ct = mls_group
            .compute_confirmation_tag(&backend)
            .map_err(MlsError::wrap("computing confirmation tag"))?;
        Ok(*msg_ct == group_ct)
    }

    /// Merges the [Self] instance and restores any buffered messages.
    async fn merge_and_restore_messages(&mut self) -> Result<MlsConversationDecryptMessage> {
        let buffered_messages = self.merge().await?;
        let context = &self.context;
        let backend = self.mls_provider().await?;
        let id = self.id();

        // This is the now merged conversation
        let conversation = context
            .conversation(id)
            .await
            .map_err(RecursiveError::transaction("getting conversation by id"))?;
        let conversation = conversation.conversation().await;
        let own_leaf = conversation.group.own_leaf().ok_or(LeafError::InternalMlsError)?;

        // We return self identity here, probably not necessary to check revocation
        let own_leaf_credential_with_key = CredentialWithKey {
            credential: own_leaf.credential().clone(),
            signature_key: own_leaf.signature_key().clone(),
        };
        let identity = own_leaf_credential_with_key
            .extract_identity(conversation.ciphersuite(), None)
            .map_err(RecursiveError::mls_credential("extracting identity"))?;

        let crl_new_distribution_points = get_new_crl_distribution_points(
            &backend,
            extract_crl_uris_from_group(&conversation.group)
                .map_err(RecursiveError::mls_credential("extracting crl uris from group"))?,
        )
        .await
        .map_err(RecursiveError::mls_credential("getting new crl distribution points"))?;

        // Note that though we return `has_epoch_changed: true` here, we don't notify the observer.
        // This function can only be reached via a code path going through `MlsConversation::decrypt_message`
        // which already notifies the observer; it would be redundant to notify here also.

        // we still support the `has_epoch_changed` field, though we'll remove it later
        #[expect(deprecated)]
        Ok(MlsConversationDecryptMessage {
            app_msg: None,
            proposals: vec![],
            is_active: conversation.group.is_active(),
            delay: conversation.compute_next_commit_delay(),
            sender_client_id: None,
            has_epoch_changed: true,
            identity,
            buffered_messages,
            crl_new_distribution_points,
        })
    }

    /// This merges the commit generated by [TransactionContext::join_by_external_commit],
    /// persists the group permanently and deletes the temporary one. After merging, the group
    /// is fully functional.
    ///
    /// # Errors
    /// Errors resulting from OpenMls, the KeyStore calls and deserialization
    pub(crate) async fn merge(&mut self) -> Result<Option<Vec<MlsBufferedConversationDecryptMessage>>> {
        let mls_provider = self.mls_provider().await?;
        let id = self.id();
        let group = self.inner.state.clone();
        let cfg = self.inner.custom_configuration.clone();

        let mut mls_group =
            core_crypto_keystore::deser::<MlsGroup>(&group).map_err(KeystoreError::wrap("deserializing mls group"))?;

        // Merge it aka bring the MLS group to life and make it usable
        mls_group
            .merge_pending_commit(&mls_provider)
            .await
            .map_err(MlsError::wrap("merging pending commit"))?;

        // Restore the custom configuration and build a conversation from it
        let custom_cfg =
            serde_json::from_slice(&cfg).map_err(MlsError::wrap("deserializing mls custom configuration"))?;
        let configuration = MlsConversationConfiguration {
            ciphersuite: mls_group.ciphersuite().into(),
            custom: custom_cfg,
            ..Default::default()
        };

        // We have to determine the restore policy before we persist the group, because it depends
        // on whether the group already exists.
        let restore_policy = if mls_provider.key_store().mls_group_exists(id.as_slice()).await {
            // If the group already exists, it means the external commit is about rejoining the group.
            // This is most of the time a last resort measure (for example when a commit is dropped,
            // and you go out of sync), so there's no point in decrypting buffered messages
            trace!("External commit trying to rejoin group");
            MessageRestorePolicy::ClearOnly
        } else {
            MessageRestorePolicy::DecryptAndClear
        };

        // Persist the now usable MLS group in the keystore
        let conversation = MlsConversation::from_mls_group(mls_group, configuration, &mls_provider)
            .await
            .map_err(RecursiveError::mls_conversation(
                "constructing conversation from mls group",
            ))?;

        let context = &self.context;

        context
            .mls_groups()
            .await
            .map_err(RecursiveError::transaction("getting mls groups"))?
            .insert(id.clone(), conversation);

        // This is the now merged conversation
        let mut conversation = context
            .conversation(id)
            .await
            .map_err(RecursiveError::transaction("getting conversation by id"))?;
        let pending_messages = conversation
            .restore_pending_messages(restore_policy)
            .await
            .map_err(RecursiveError::mls_conversation("restoring pending messages"))?;

        if pending_messages.is_some() {
            mls_provider
                .key_store()
                .remove::<MlsPendingMessage, _>(id)
                .await
                .map_err(KeystoreError::wrap("deleting mls pending message by id"))?;
        }

        // cleanup the pending group we no longer need
        self.clear().await?;

        Ok(pending_messages)
    }

    /// In case the external commit generated by [TransactionContext::join_by_external_commit] is
    /// rejected by the Delivery Service, and we want to abort this external commit,
    /// we can wipe out the pending group from the keystore.
    ///
    /// # Errors
    /// Errors resulting from the KeyStore calls
    pub(crate) async fn clear(&mut self) -> Result<()> {
        self.keystore()
            .await?
            .mls_pending_groups_delete(self.id())
            .await
            .map_err(KeystoreError::wrap("deleting pending groups by id"))?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::prelude::MlsConversationDecryptMessage;
    use crate::test_utils::*;
    use wasm_bindgen_test::*;

    wasm_bindgen_test_configure!(run_in_browser);

    #[apply(all_cred_cipher)]
    #[wasm_bindgen_test]
    async fn should_buffer_and_reapply_messages_after_external_commit_merged(case: TestContext) {
        let [alice, bob, charlie, debbie] = case.sessions().await;
        Box::pin(async move {
            let conversation = case.create_conversation([&alice]).await;
            // Bob tries to join Alice's group with an external commit
            let (commit_guard, _pending_conversation) = conversation.external_join_unmerged(&bob).await;
            let external_commit = commit_guard.message();

            // Alice decrypts the external commit...
            let conversation = commit_guard
                .notify_member(&alice)
                .await
                .process_member_changes()
                .await
                .finish();

            // Meanwhile Debbie joins the party by creating an external proposal
            let proposal_guard = conversation.external_join_proposal(&debbie).await;
            let external_proposal = proposal_guard.message();

            // ...then Alice generates new messages for this epoch
            let app_msg = proposal_guard
                .conversation()
                .guard()
                .await
                .encrypt_message(b"Hello Bob !")
                .await
                .unwrap();

            let conversation = proposal_guard.notify_member(&alice).await.finish();
            let proposal_guard = conversation.update_proposal().await;
            let proposal = proposal_guard.message();
            let conversation = proposal_guard.finish();

            let commit_guard = conversation.invite([&charlie]).await;
            let commit = commit_guard.message();
            let conversation = commit_guard.process_member_changes().await.finish();

            // And now Bob will have to decrypt those messages while he hasn't yet merged its external commit
            // To add more fun, he will buffer the messages in exactly the wrong order (to make
            // sure he reapplies them in the right order afterwards)
            let messages = vec![commit, external_proposal, proposal]
                .into_iter()
                .map(|m| m.to_bytes().unwrap());
            let Err(crate::transaction_context::Error::PendingConversation(mut pending_conversation)) =
                bob.transaction.conversation(conversation.id()).await
            else {
                panic!("Bob should not have the conversation yet")
            };
            for m in messages {
                let decrypt = pending_conversation.try_process_own_join_commit(m).await;
                assert!(matches!(decrypt.unwrap_err(), Error::BufferedForPendingConversation));
            }
            let decrypt = pending_conversation.try_process_own_join_commit(app_msg).await;
            assert!(matches!(decrypt.unwrap_err(), Error::BufferedForPendingConversation));

            // Bob should have buffered the messages
            assert_eq!(bob.transaction.count_entities().await.pending_messages, 4);

            let observer = TestEpochObserver::new();
            bob.session()
                .await
                .register_epoch_observer(observer.clone())
                .await
                .unwrap();

            // Finally, Bob receives the green light from the DS and he can merge the external commit
            let MlsConversationDecryptMessage {
                buffered_messages: Some(restored_messages),
                ..
            } = pending_conversation
                .try_process_own_join_commit(external_commit.to_bytes().unwrap())
                .await
                .unwrap()
            else {
                panic!("Alice's messages should have been restored at this point");
            };

            let observed_epochs = observer.observed_epochs().await;
            assert_eq!(
                observed_epochs.len(),
                1,
                "we should see exactly 1 epoch change in these 4 messages"
            );
            assert_eq!(observed_epochs[0].0, *conversation.id(), "conversation id must match");

            for (idx, msg) in restored_messages.iter().enumerate() {
                if idx == 0 {
                    // the only application message
                    assert_eq!(msg.app_msg.as_deref(), Some(b"Hello Bob !" as _));
                } else {
                    assert!(msg.app_msg.is_none());
                }
            }

            assert_eq!(conversation.member_count().await, 4);
            assert!(
                conversation
                    .is_functional_and_contains([&alice, &bob, &charlie, &debbie])
                    .await
            );

            // After merging we should erase all those pending messages
            assert_eq!(bob.transaction.count_entities().await.pending_messages, 0);
        })
        .await
    }

    #[apply(all_cred_cipher)]
    #[wasm_bindgen_test]
    async fn should_not_reapply_buffered_messages_when_rejoining(case: TestContext) {
        use crate::mls;

        let [alice, bob] = case.sessions().await;
        Box::pin(async move {
            let conversation = case.create_conversation([&alice, &bob]).await;

            // Alice will never see this commit
            let conversation = conversation.acting_as(&bob).await;
            let commit_guard = conversation.update().await;
            let conversation = commit_guard.conversation();

            let msg1 = conversation.guard().await.encrypt_message("A").await.unwrap();
            let msg2 = conversation.guard().await.encrypt_message("B").await.unwrap();

            let conversation = commit_guard.finish();
            // Since Alice missed Bob's commit she should buffer this message
            let decrypt = conversation.guard().await.decrypt_message(msg1).await;
            assert!(matches!(
                decrypt.unwrap_err(),
                mls::conversation::Error::BufferedFutureMessage { .. }
            ));
            let decrypt = conversation.guard().await.decrypt_message(msg2).await;
            assert!(matches!(
                decrypt.unwrap_err(),
                mls::conversation::Error::BufferedFutureMessage { .. }
            ));
            assert_eq!(alice.transaction.count_entities().await.pending_messages, 2);

            let conversation = conversation.acting_as(&bob).await.external_join_notify(&alice).await;
            // Alice should have deleted all her buffered messages
            assert_eq!(alice.transaction.count_entities().await.pending_messages, 0);
            assert!(conversation.is_functional_and_contains([&alice, &bob]).await);
        })
        .await
    }
}