core_crypto/mls/conversation/conversation_guard/
commit.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
//! The methods in this module all produce or handle commits.

use openmls::prelude::KeyPackageIn;

use crate::mls::conversation::{ConversationWithMls as _, Error};
use crate::mls::credential::CredentialBundle;
use crate::prelude::MlsCredentialType;
use crate::{
    LeafError, MlsError, MlsTransportResponse, RecursiveError,
    e2e_identity::init_certificates::NewCrlDistributionPoint,
    mls::{
        conversation::{ConversationGuard, Result, commit::MlsCommitBundle},
        credential::crl::{extract_crl_uris_from_credentials, get_new_crl_distribution_points},
    },
    prelude::ClientId,
};

/// What to do with a commit after it has been sent via [crate::MlsTransport].
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum TransportedCommitPolicy {
    /// Accept and merge the commit.
    Merge,
    /// Do nothing, because intended operation was already done in one in intermediate processing.
    None,
}

impl ConversationGuard {
    async fn send_and_merge_commit(&mut self, commit: MlsCommitBundle) -> Result<()> {
        match self.send_commit(commit).await {
            Ok(TransportedCommitPolicy::None) => Ok(()),
            Ok(TransportedCommitPolicy::Merge) => {
                let backend = self.mls_provider().await?;
                let mut conversation = self.inner.write().await;
                conversation.commit_accepted(&backend).await
            }
            Err(e @ Error::MessageRejected { .. }) => {
                self.clear_pending_commit().await?;
                Err(e)
            }
            Err(e) => Err(e),
        }
    }

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

        let inner = self.conversation().await;
        let epoch_before_sending = inner.group().epoch().as_u64();
        // Drop the lock to allow mutably borrowing self again.
        drop(inner);

        loop {
            match transport
                .send_commit_bundle(commit.clone())
                .await
                .map_err(RecursiveError::root("sending commit bundle"))?
            {
                MlsTransportResponse::Success => {
                    return Ok(TransportedCommitPolicy::Merge);
                }
                MlsTransportResponse::Abort { reason } => {
                    return Err(Error::MessageRejected { reason });
                }
                MlsTransportResponse::Retry => {
                    let mut inner = self.conversation_mut().await;
                    let epoch_after_sending = inner.group().epoch().as_u64();
                    if epoch_before_sending == epoch_after_sending {
                        // No intermediate commits have been processed before returning retry.
                        // This will be the case, e.g., on network failure.
                        // We can just send the exact same commit again.
                        continue;
                    }

                    // The epoch has changed. I.e., a client originally tried sending a commit for an old epoch,
                    // which was rejected by the DS.
                    // Before returning `Retry`, the API consumer has fetched and merged all commits,
                    // so the group state is up-to-date.
                    // The original commit has been `renewed` to a pending proposal, unless the
                    // intended operation was already done in one of the merged commits.
                    let Some(commit_to_retry) = inner.commit_pending_proposals(&client, &backend).await? else {
                        // The intended operation was already done in one of the merged commits.
                        return Ok(TransportedCommitPolicy::None);
                    };
                    commit = commit_to_retry;
                }
            }
        }
    }

    /// Adds new members to the group/conversation
    pub async fn add_members(&mut self, key_packages: Vec<KeyPackageIn>) -> Result<NewCrlDistributionPoint> {
        let backend = self.mls_provider().await?;
        let credential = self.credential_bundle().await?;
        let signer = credential.signature_key();
        let mut conversation = self.conversation_mut().await;

        // No need to also check pending proposals since they should already have been scanned while decrypting the proposal message
        let crl_dps = extract_crl_uris_from_credentials(key_packages.iter().filter_map(|kp| {
            let mls_credential = kp.credential().mls_credential();
            matches!(mls_credential, openmls::prelude::MlsCredentialType::X509(_)).then_some(mls_credential)
        }))
        .map_err(RecursiveError::mls_credential("extracting crl uris from credentials"))?;
        let crl_new_distribution_points = get_new_crl_distribution_points(&backend, crl_dps)
            .await
            .map_err(RecursiveError::mls_credential("getting new crl distribution points"))?;

        let (commit, welcome, group_info) = conversation
            .group
            .add_members(&backend, signer, key_packages)
            .await
            .map_err(MlsError::wrap("group add members"))?;

        // commit requires an optional welcome
        let welcome = Some(welcome);
        let group_info = Self::group_info(group_info)?;

        conversation
            .persist_group_when_changed(&backend.keystore(), false)
            .await?;

        // we don't need the conversation anymore, but we do need to mutably borrow `self` again
        drop(conversation);

        let commit = MlsCommitBundle {
            commit,
            welcome,
            group_info,
        };

        self.send_and_merge_commit(commit).await?;

        Ok(crl_new_distribution_points)
    }

    /// Removes clients from the group/conversation.
    ///
    /// # Arguments
    /// * `id` - group/conversation id
    /// * `clients` - list of client ids to be removed from the group
    pub async fn remove_members(&mut self, clients: &[ClientId]) -> Result<()> {
        let backend = self.mls_provider().await?;
        let credential = self.credential_bundle().await?;
        let signer = credential.signature_key();
        let mut conversation = self.inner.write().await;

        let members = conversation
            .group
            .members()
            .filter_map(|kp| {
                clients
                    .iter()
                    .any(move |client_id| client_id.as_slice() == kp.credential.identity())
                    .then_some(kp.index)
            })
            .collect::<Vec<_>>();

        let (commit, welcome, group_info) = conversation
            .group
            .remove_members(&backend, signer, &members)
            .await
            .map_err(MlsError::wrap("group remove members"))?;

        let group_info = Self::group_info(group_info)?;

        conversation
            .persist_group_when_changed(&backend.keystore(), false)
            .await?;

        // we don't need the conversation anymore, but we do need to mutably borrow `self` again
        drop(conversation);

        self.send_and_merge_commit(MlsCommitBundle {
            commit,
            welcome,
            group_info,
        })
        .await
    }

    /// Self updates the KeyPackage and automatically commits. Pending proposals will be commited.
    ///
    /// # Arguments
    /// * `conversation_id` - the group/conversation id
    ///
    /// see [MlsCentral::update_keying_material]
    pub async fn update_key_material(&mut self) -> Result<()> {
        let client = self.mls_client().await?;
        let backend = self.mls_provider().await?;
        let mut conversation = self.inner.write().await;
        let commit = conversation
            .update_keying_material(&client, &backend, None, None)
            .await?;
        drop(conversation);
        self.send_and_merge_commit(commit).await
    }

    /// Send a commit in a conversation for changing the credential. Requires first
    /// having enrolled a new X509 certificate with either
    /// [crate::context::CentralContext::e2ei_new_activation_enrollment] or
    /// [crate::context::CentralContext::e2ei_new_rotate_enrollment] and having saved it with
    /// [crate::context::CentralContext::save_x509_credential].
    pub async fn e2ei_rotate(&mut self, cb: Option<&CredentialBundle>) -> Result<()> {
        let client = &self.mls_client().await?;
        let backend = &self.mls_provider().await?;
        let mut conversation = self.inner.write().await;

        let cb = match cb {
            Some(cb) => cb,
            None => &client
                .find_most_recent_credential_bundle(
                    conversation.ciphersuite().signature_algorithm(),
                    MlsCredentialType::X509,
                )
                .await
                .map_err(RecursiveError::mls_client("finding most recent x509 credential bundle"))?,
        };

        let mut leaf_node = conversation
            .group
            .own_leaf()
            .ok_or(LeafError::InternalMlsError)?
            .clone();
        leaf_node.set_credential_with_key(cb.to_mls_credential_with_key());

        let commit = conversation
            .update_keying_material(client, backend, Some(cb), Some(leaf_node))
            .await?;
        // we don't need the conversation anymore, but we do need to mutably borrow `self` again
        drop(conversation);

        self.send_and_merge_commit(commit).await
    }

    /// Commits all pending proposals of the group
    pub async fn commit_pending_proposals(&mut self) -> Result<()> {
        let client = self.mls_client().await?;
        let backend = self.mls_provider().await?;
        let mut conversation = self.inner.write().await;
        let commit = conversation.commit_pending_proposals(&client, &backend).await?;
        drop(conversation);
        let Some(commit) = commit else {
            return Ok(());
        };
        self.send_and_merge_commit(commit).await
    }
}