xmtp_mls/worker/device_sync/
preference_sync.rs

1use super::*;
2pub use crate::core::events::PreferenceUpdate;
3use xmtp_db::consent_record::{
4    ConsentChange, ConsentWrite, Owed, parse_consent_save, write_consent,
5};
6use xmtp_db::group_message::StoredGroupMessage;
7use xmtp_db::user_preferences::{RootKeyWrite, write_root_key};
8use xmtp_proto::xmtp::device_sync::content::{
9    PreferenceUpdate as PreferenceUpdateProto, preference_update::Update as UpdateProto,
10};
11
12/// What one sync message changed.
13#[derive(Default)]
14pub(super) struct Applied {
15    pub consent: Vec<ConsentChange>,
16    /// Set when the root key changed: whether the change was marked owed.
17    pub root_key: Option<bool>,
18}
19
20/// Merge the updates of a sync message, a peer's or this installation's. `updates` is borrowed, so a transaction
21/// retry copies no HMAC root key; the caller wipes the keys. A value from
22/// `selected`, the sync group the publisher selects, keeps its mark; one
23/// from an older group is passed on.
24// implements: CONS-051, CONS-061, SYNC-023, SYNC-027, SYNC-032
25pub(super) fn apply_preference_updates(
26    db: &impl DbQuery,
27    own_inbox: &str,
28    msg: &StoredGroupMessage,
29    updates: &[PreferenceUpdateProto],
30    selected: Option<GroupId>,
31) -> Result<Applied, StorageError> {
32    let owed = if selected == Some(msg.group_id) {
33        Owed::Keep
34    } else {
35        Owed::Mark
36    };
37    let mut applied = Applied::default();
38    for update in updates.iter().filter_map(|update| update.update.as_ref()) {
39        match update {
40            UpdateProto::Consent(save) => {
41                let Some((key, value)) = parse_consent_save(save) else {
42                    tracing::warn!("skipping an unusable synced consent record");
43                    continue;
44                };
45                // A 16-byte DM id is its peer's record.
46                let key = key.pre8_dm(db, own_inbox)?.map_or(key, |(peer, _)| peer);
47                let write = ConsentWrite::from_peer(value, msg.sent_at_ns);
48                applied
49                    .consent
50                    .extend(write_consent(db, &key, write, owed)?);
51            }
52            UpdateProto::Hmac(hmac) => {
53                let write = RootKeyWrite::from_peer(&hmac.key, hmac.cycled_at_ns, msg.sent_at_ns);
54                if write_root_key(db, write, owed)? {
55                    applied.root_key = Some(owed == Owed::Mark);
56                }
57            }
58        }
59    }
60    Ok(applied)
61}
62
63#[cfg(test)]
64mod tests {
65    use super::*;
66    use crate::tester;
67    use xmtp_common::time::now_ns;
68    use xmtp_db::consent_record::{
69        ConsentKey, ConsentState, ConsentType, ConsentValue, MAX_AHEAD_NS,
70    };
71    use xmtp_db::group::{ConversationType, GroupMembershipState, StoredGroup};
72    use xmtp_db::user_preferences::StoredUserPreferences;
73    use xmtp_proto::types::GroupId;
74    use xmtp_proto::xmtp::device_sync::{consent_backup::ConsentSave, content::HmacKeyUpdate};
75
76    /// `Debug` of an HMAC update shows no root key bytes.
77    #[xmtp_common::test]
78    fn hmac_update_debug_omits_the_root_key() {
79        let key: Vec<u8> = (0..42).map(|i| 0x80 | i).collect();
80        let built = PreferenceUpdate::Hmac {
81            key: key.clone().into(),
82            cycled_at_ns: 5,
83        };
84        let forms = [
85            hex::encode(&key),
86            hex::encode_upper(&key),
87            // How `Debug` prints a byte slice, anywhere in the list.
88            format!("{:?}", &key[..4])
89                .trim_matches(['[', ']'])
90                .to_owned(),
91        ];
92        let debug = format!("{:?}", vec![built]);
93        for form in &forms {
94            assert!(
95                !debug.contains(form.as_str()),
96                "PreferenceUpdate Debug shows the key as {form}: {debug}"
97            );
98        }
99    }
100
101    fn consent(entity_type: i32, state: i32, entity: &str, at: i64) -> PreferenceUpdateProto {
102        PreferenceUpdateProto {
103            update: Some(UpdateProto::Consent(ConsentSave {
104                entity_type,
105                state,
106                entity: entity.into(),
107                consented_at_ns: at,
108            })),
109        }
110    }
111
112    fn root_key(key: Vec<u8>, cycled_at_ns: i64) -> PreferenceUpdateProto {
113        PreferenceUpdateProto {
114            update: Some(UpdateProto::Hmac(HmacKeyUpdate { key, cycled_at_ns })),
115        }
116    }
117
118    /// A sync message stored in `group_id` at envelope time `sent_at_ns`.
119    fn message(group_id: GroupId, sent_at_ns: i64) -> StoredGroupMessage {
120        StoredGroupMessage {
121            sent_at_ns,
122            ..crate::test::mock::generate_stored_msg(xmtp_proto::types::Cursor(1), group_id)
123        }
124    }
125
126    fn sync_group(db: &impl DbQuery, id: u8, created_at_ns: i64) -> Result<GroupId, StorageError> {
127        use xmtp_db::Store;
128        let id = GroupId::from([id; 16]);
129        StoredGroup::builder()
130            .id(id)
131            .created_at_ns(created_at_ns)
132            .membership_state(GroupMembershipState::Allowed)
133            .added_by_inbox_id("me")
134            .conversation_type(ConversationType::Sync)
135            .build()?
136            .store(db)?;
137        Ok(id)
138    }
139
140    // verifies: SYNC-023, CONS-051, SYNC-032
141    #[rstest::rstest]
142    #[case::address(3, 2)]
143    #[case::unspecified_state(2, 0)]
144    #[case::unknown_enum(99, 2)]
145    #[case::short_key(2, 2)]
146    #[xmtp_common::test(unwrap_try = true)]
147    async fn unusable_update_is_skipped_and_the_rest_apply(
148        #[case] entity_type: i32,
149        #[case] state: i32,
150    ) {
151        tester!(alix, disable_workers);
152        let db = alix.context.db();
153        let now = now_ns();
154        let bad = if (entity_type, state) == (2, 2) {
155            root_key(vec![1; 41], now + 1)
156        } else {
157            consent(entity_type, state, "bad", now + 1)
158        };
159        let updates = vec![bad, consent(2, 2, "good", now), root_key(vec![2; 42], now)];
160        let group = sync_group(&db, 1, now).unwrap();
161        let applied =
162            apply_preference_updates(&db, "me", &message(group, now), &updates, Some(group))
163                .unwrap();
164        assert_eq!(applied.consent.len(), 1);
165        let stored: Vec<_> = db
166            .consent_records()
167            .unwrap()
168            .iter()
169            .map(|r| (r.entity.clone(), r.state))
170            .collect();
171        assert_eq!(stored, vec![("good".to_string(), ConsentState::Allowed)]);
172        assert_eq!(
173            StoredUserPreferences::load(&db).unwrap().hmac_key,
174            Some(vec![2; 42].into())
175        );
176    }
177
178    /// A received time is capped at a day after its envelope; a 16-byte DM id
179    /// moves to its peer's record; a value from the selected sync group keeps
180    /// its mark and one from an older group is marked to pass it on.
181    // verifies: CONS-054, CONS-061, SYNC-027, SYNC-032
182    #[xmtp_common::test(unwrap_try = true)]
183    async fn received_values_are_capped_keyed_and_marked() {
184        use xmtp_db::Store;
185        tester!(alix, disable_workers);
186        let db = alix.context.db();
187        let own = alix.inbox_id();
188        let envelope = now_ns();
189        let older = sync_group(&db, 1, envelope - 10)?;
190        let selected = sync_group(&db, 2, envelope)?;
191        let peer = "ab".repeat(32);
192        let dm = GroupId::from([7; 16]);
193        StoredGroup::builder()
194            .id(dm)
195            .created_at_ns(1)
196            .membership_state(GroupMembershipState::Restored)
197            .added_by_inbox_id(own)
198            .conversation_type(ConversationType::Dm)
199            .dm_id(Some(format!("dm:{}", {
200                let mut pair = [own.to_string(), peer.clone()];
201                pair.sort();
202                pair.join(":")
203            })))
204            .build()?
205            .store(&db)?;
206        // A 32-byte DM never moves: a former member can build one.
207        let forged = GroupId::from([8; 32]);
208        StoredGroup::builder()
209            .id(forged)
210            .created_at_ns(1)
211            .membership_state(GroupMembershipState::Allowed)
212            .added_by_inbox_id(own)
213            .conversation_type(ConversationType::Dm)
214            .dm_id(Some(format!("dm:{}", {
215                let mut pair = [own.to_string(), "cd".repeat(32)];
216                pair.sort();
217                pair.join(":")
218            })))
219            .build()?
220            .store(&db)?;
221        let far = envelope + 2 * xmtp_common::NS_IN_DAY;
222        let updates = vec![
223            consent(1, 2, &hex::encode(forged), 5),
224            consent(1, 3, &hex::encode(dm).to_uppercase(), 5),
225            consent(2, 2, "far", far),
226            root_key(vec![3; 42], far),
227        ];
228        let applied = apply_preference_updates(
229            &db,
230            own,
231            &message(selected, envelope),
232            &updates,
233            Some(selected),
234        )?;
235        assert_eq!(applied.root_key, Some(false));
236        let cap = envelope + MAX_AHEAD_NS;
237        let row = |key: &ConsentKey| db.consent_row(key).unwrap().unwrap();
238        assert_eq!(
239            row(&ConsentKey::inbox("far")).value(),
240            ConsentValue {
241                state: ConsentState::Allowed,
242                consented_at_ns: cap
243            }
244        );
245        assert_eq!(row(&ConsentKey::inbox(&peer)).state, ConsentState::Denied);
246        assert!(db.consent_row(&ConsentKey::group(dm.as_ref()))?.is_none());
247        assert_eq!(
248            row(&ConsentKey::group(forged.as_ref())).state,
249            ConsentState::Allowed
250        );
251        assert!(
252            db.consent_row(&ConsentKey::inbox(&"cd".repeat(32)))?
253                .is_none()
254        );
255        let prefs = StoredUserPreferences::load(&db)?;
256        assert_eq!(prefs.hmac_key_cycled_at_ns, Some(cap));
257        // From the selected group: nothing is owed.
258        for key in [
259            ConsentKey::inbox("far"),
260            ConsentKey::inbox(&peer),
261            ConsentKey::group(forged.as_ref()),
262        ] {
263            assert_eq!(row(&key).owed_seq, 0);
264        }
265        assert_eq!(prefs.hmac_key_owed_seq, 0);
266        // From an older group: a change is passed on.
267        let later = vec![consent(ConsentType::InboxId as i32, 3, "far", cap + 1)];
268        let applied =
269            apply_preference_updates(&db, own, &message(older, envelope), &later, Some(selected))?;
270        assert!(applied.consent[0].marked);
271        assert!(row(&ConsentKey::inbox("far")).owed_seq > 0);
272    }
273}
xmtp_mls/worker/device_sync/preference_sync.rs

xmtp_mls/worker/device_sync/
preference_sync.rs
