xmtp_mls/groups/mls_sync/
post_commit.rs

1//! Post-commit work: installations, welcomes, and HMAC keys.
2
3use super::*;
4
5impl<Context> MlsGroup<Context>
6where
7    Context: XmtpSharedContext,
8{
9    #[tracing::instrument(skip_all)]
10    pub(crate) async fn post_commit(&self) -> Result<(), GroupError> {
11        self.publish_required_welcomes().await
12    }
13
14    pub async fn maybe_update_installations(
15        &self,
16        update_interval_ns: Option<i64>,
17    ) -> Result<(), GroupError> {
18        let db = self.context.db();
19        let Some(stored_group) = db.find_group(&self.group_id)? else {
20            return Err(GroupError::NotFound(NotFound::GroupById(self.group_id)));
21        };
22        if stored_group.conversation_type.is_virtual() {
23            return Ok(());
24        }
25
26        // determine how long of an interval in time to use before updating list
27        let interval_ns = update_interval_ns.unwrap_or(SYNC_UPDATE_INSTALLATIONS_INTERVAL_NS);
28
29        let now_ns = xmtp_common::time::now_ns();
30        let last_ns = db.get_installations_time_checked(&self.group_id)?;
31        let elapsed_ns = now_ns - last_ns;
32        if elapsed_ns > interval_ns && self.is_active()? {
33            self.add_missing_installations().await?;
34            db.update_installations_time_checked(&self.group_id)?;
35        }
36
37        Ok(())
38    }
39
40    /**
41     * Checks each member of the group for `IdentityUpdates` after their current sequence_id. If updates
42     * are found the method will construct an [`UpdateGroupMembershipIntentData`] and create a change
43     * to the [`GroupMembership`] that will add any missing installations.
44     *
45     * This is designed to handle cases where existing members have added a new installation to their inbox or revoked an installation
46     * and the group has not been updated to include it.
47     */
48    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = %self.context.inbox_id()), skip_all))]
49    #[cfg_attr(
50        not(any(test, feature = "test-utils")),
51        tracing::instrument(level = "trace", skip_all)
52    )]
53    pub(crate) async fn add_missing_installations(&self) -> Result<(), GroupError> {
54        let intent_data = self.get_membership_update_intent(&[], &[]).await?;
55
56        // If there is nothing to do, stop here
57        if intent_data.is_empty() {
58            return Ok(());
59        }
60
61        debug!(
62            inbox_id = self.context.inbox_id(),
63            installation_id = %self.context.installation_id(),
64            "Adding missing installations {:?}",
65            intent_data
66        );
67
68        let intent = QueueIntent::update_group_membership()
69            .data(intent_data)
70            .queue(self)?;
71
72        let _ = self.sync_until_intent_resolved(intent.id).await?;
73        Ok(())
74    }
75
76    #[tracing::instrument(level = "trace", skip_all)]
77    /**
78     * get_membership_update_intent will query the network for any new [`IdentityUpdate`]s for any of the existing
79     * group members
80     *
81     * Callers may also include a list of added or removed inboxes
82     */
83    pub(crate) async fn get_membership_update_intent(
84        &self,
85        inbox_ids_to_add: &[InboxIdRef<'_>],
86        inbox_ids_to_remove: &[InboxIdRef<'_>],
87    ) -> Result<UpdateGroupMembershipIntentData, GroupError> {
88        let existing_group_membership = self.with_group_snapshot(|group| {
89            extract_group_membership(group.extensions()).map_err(Into::into)
90        })?;
91        {
92            // TODO:nm prevent querying for updates on members who are being removed
93            let mut inbox_ids = existing_group_membership.inbox_ids();
94            inbox_ids.extend_from_slice(inbox_ids_to_add);
95            let conn = self.context.db();
96            // Load any missing updates from the network
97            load_identity_updates(self.context.api(), &conn, &inbox_ids).await?;
98
99            let latest_sequence_id_map = conn.get_latest_sequence_id(&inbox_ids as &[&str])?;
100
101            // Get a list of all inbox IDs that have increased sequence_id for the group
102            let changed_inbox_ids =
103                inbox_ids
104                    .iter()
105                    .try_fold(HashMap::new(), |mut updates, inbox_id| {
106                        match (
107                            latest_sequence_id_map.get(inbox_id as &str),
108                            existing_group_membership.get(inbox_id),
109                        ) {
110                            // This is an update. We have a new sequence ID and an existing one
111                            (Some(latest_sequence_id), Some(current_sequence_id)) => {
112                                let latest_sequence_id_u64 = *latest_sequence_id as u64;
113                                if latest_sequence_id_u64.gt(current_sequence_id) {
114                                    updates.insert(inbox_id.to_string(), latest_sequence_id_u64);
115                                }
116                            }
117                            // This is for new additions to the group
118                            (Some(latest_sequence_id), None) => {
119                                // This is the case for net new members to the group
120                                updates.insert(inbox_id.to_string(), *latest_sequence_id as u64);
121                            }
122                            (_, _) => {
123                                tracing::warn!(
124                                    "Could not find existing sequence ID for inbox {}",
125                                    inbox_id
126                                );
127                                return Err(GroupError::MissingSequenceId);
128                            }
129                        }
130
131                        Ok(updates)
132                    })?;
133            let old_group_membership = existing_group_membership.clone();
134            let mut new_membership = old_group_membership.clone();
135            for (inbox_id, sequence_id) in changed_inbox_ids.iter() {
136                new_membership.add(inbox_id.clone(), *sequence_id);
137            }
138            for inbox_id in inbox_ids_to_remove {
139                new_membership.remove(inbox_id);
140            }
141
142            let changes_with_kps = calculate_membership_changes_with_keypackages(
143                &self.context,
144                &self.group_id,
145                &new_membership,
146                &old_group_membership,
147            )
148            .await?;
149
150            // If we fail to fetch or verify all the added members' KeyPackage, return an error.
151            // skip if the inbox ids is 0 from the beginning
152            if !inbox_ids_to_add.is_empty()
153                && !changes_with_kps.failed_installations.is_empty()
154                && changes_with_kps.new_installations.is_empty()
155            {
156                return Err(GroupError::FailedToVerifyInstallations(
157                    FailedInstallationIds(changes_with_kps.failed_installations.clone()),
158                ));
159            }
160
161            Ok(UpdateGroupMembershipIntentData::new(
162                changed_inbox_ids,
163                inbox_ids_to_remove
164                    .iter()
165                    .map(|s| s.to_string())
166                    .collect::<Vec<String>>(),
167                changes_with_kps.failed_installations,
168            ))
169        }
170    }
171
172    #[cfg(test)]
173    pub(in crate::groups) async fn send_welcomes(
174        &self,
175        action: SendWelcomesAction,
176        message_cursor: Option<i64>,
177    ) -> Result<(), GroupError> {
178        let message_cursor = u64::try_from(message_cursor.unwrap_or(0))
179            .map_err(|_| xmtp_proto::ConversionError::Unspecified("negative Welcome cursor"))?;
180        let units = crate::state_tx::state_write(self.context.mls_storage(), |_tx| {
181            self.prepare_welcome_envelopes(action, message_cursor)?
182                .into_iter()
183                .map(PublishUnit::single)
184                .collect::<Result<Vec<_>, _>>()
185                .map(Continue)
186                .map_err(GroupError::from)
187        })?
188        .into_continued();
189        self.context.api().publish_units(units).await?;
190        Ok(())
191    }
192
193    /// Provides hmac keys for a range of epochs around current epoch
194    /// `group.hmac_keys(-1..=1)`` will provide 3 keys consisting of last epoch, current epoch, and next epoch
195    /// `group.hmac_keys(0..=0) will provide 1 key, consisting of only the current epoch
196    #[tracing::instrument(level = "trace", skip_all)]
197    pub fn hmac_keys(
198        &self,
199        epoch_delta_range: RangeInclusive<i64>,
200    ) -> Result<Vec<HmacKey>, StorageError> {
201        crate::state_tx::state_write(self.context.mls_storage(), |tx| {
202            self.hmac_keys_in(tx.storage().db(), epoch_delta_range)
203                .map(Continue)
204        })
205        .map(TransactionOutcome::into_continued)
206    }
207
208    pub(crate) fn hmac_keys_in(
209        &self,
210        conn: impl xmtp_db::DbQuery,
211        epoch_delta_range: RangeInclusive<i64>,
212    ) -> Result<Vec<HmacKey>, StorageError> {
213        let preferences = StoredUserPreferences::load(&conn)?;
214        let mut ikm = match preferences.hmac_key {
215            Some(ikm) => ikm,
216            None => {
217                let key = HmacKey::random_key();
218                StoredUserPreferences::store_hmac_key(&conn, &key, None)?;
219                key
220            }
221        };
222        ikm.extend_from_slice(self.group_id.as_ref());
223        let hkdf = Hkdf::<Sha256>::new(Some(HMAC_SALT), &ikm);
224
225        let mut result = vec![];
226        let current_epoch = hmac_epoch();
227        for delta in epoch_delta_range {
228            let epoch = current_epoch + delta;
229
230            let mut info = self.group_id.to_vec();
231            info.extend(&epoch.to_le_bytes());
232
233            let mut key = [0; 42];
234            hkdf.expand(&info, &mut key).expect("Length is correct");
235
236            result.push(HmacKey { key, epoch });
237        }
238
239        Ok(result)
240    }
241
242    #[cfg(test)]
243    #[tracing::instrument(level = "trace", skip_all)]
244    pub(in crate::groups) fn prepare_group_messages(
245        &self,
246        payloads: Vec<(&[u8], bool)>,
247    ) -> Result<Vec<PublishUnit>, GroupError> {
248        crate::state_tx::state_write(self.context.mls_storage(), |tx| {
249            let envelopes = self.prepare_group_envelopes_in(tx.storage().db(), payloads)?;
250            Ok::<_, GroupError>(Continue(vec![PublishUnit::new(envelopes)?]))
251        })
252        .map(TransactionOutcome::into_continued)
253    }
254
255    pub(super) fn prepare_group_envelopes_in(
256        &self,
257        conn: impl xmtp_db::DbQuery,
258        payloads: Vec<(&[u8], bool)>,
259    ) -> Result<Vec<ClientEnvelope>, GroupError> {
260        let hmac_key = self
261            .hmac_keys_in(conn, 0..=0)?
262            .pop()
263            .expect("Range of count 1 was provided.");
264        let sender_hmac =
265            Hmac::<Sha256>::new_from_slice(&hmac_key.key).expect("HMAC can take key of any size");
266
267        let mut result = vec![];
268        for (payload, should_push) in payloads {
269            let mut sender_hmac = sender_hmac.clone();
270            sender_hmac.update(payload);
271            let sender_hmac = sender_hmac.finalize();
272
273            result.push(ClientEnvelope {
274                payload: Some(Payload::GroupMessage(BackendGroupMessage {
275                    data: payload.to_vec(),
276                    sender_hmac: sender_hmac.into_bytes().to_vec(),
277                    should_push,
278                })),
279            });
280        }
281
282        Ok(result)
283    }
284}
xmtp_mls/groups/mls_sync/post_commit.rs

xmtp_mls/groups/mls_sync/
post_commit.rs
