matrix_sdk_crypto/verification/
machine.rs

1// Copyright 2020 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{collections::HashMap, sync::Arc};
16
17use matrix_sdk_common::locks::RwLock as StdRwLock;
18use ruma::{
19    events::{
20        key::verification::VerificationMethod, AnyToDeviceEvent, AnyToDeviceEventContent,
21        ToDeviceEvent,
22    },
23    serde::Raw,
24    uint, DeviceId, EventId, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, RoomId,
25    SecondsSinceUnixEpoch, TransactionId, UInt, UserId,
26};
27use tokio::sync::Mutex;
28use tracing::{debug, info, instrument, trace, warn, Span};
29
30use super::{
31    cache::{RequestInfo, VerificationCache},
32    event_enums::{AnyEvent, AnyVerificationContent, OutgoingContent},
33    requests::VerificationRequest,
34    sas::Sas,
35    FlowId, Verification, VerificationResult, VerificationStore,
36};
37use crate::{
38    olm::{PrivateCrossSigningIdentity, StaticAccountData},
39    store::{CryptoStoreError, CryptoStoreWrapper},
40    types::requests::{
41        OutgoingRequest, OutgoingVerificationRequest, RoomMessageRequest, ToDeviceRequest,
42    },
43    DeviceData, OtherUserIdentityData,
44};
45
46#[derive(Clone, Debug)]
47pub struct VerificationMachine {
48    pub(crate) store: VerificationStore,
49    verifications: VerificationCache,
50    requests: Arc<StdRwLock<HashMap<OwnedUserId, HashMap<String, VerificationRequest>>>>,
51}
52
53impl VerificationMachine {
54    pub(crate) fn new(
55        account: StaticAccountData,
56        identity: Arc<Mutex<PrivateCrossSigningIdentity>>,
57        store: Arc<CryptoStoreWrapper>,
58    ) -> Self {
59        Self {
60            store: VerificationStore { account, private_identity: identity, inner: store },
61            verifications: VerificationCache::new(),
62            requests: Default::default(),
63        }
64    }
65
66    pub(crate) fn own_user_id(&self) -> &UserId {
67        &self.store.account.user_id
68    }
69
70    pub(crate) fn own_device_id(&self) -> &DeviceId {
71        &self.store.account.device_id
72    }
73
74    pub(crate) fn request_to_device_verification(
75        &self,
76        user_id: &UserId,
77        recipient_devices: Vec<OwnedDeviceId>,
78        methods: Option<Vec<VerificationMethod>>,
79    ) -> (VerificationRequest, OutgoingVerificationRequest) {
80        let flow_id = FlowId::from(TransactionId::new());
81
82        let verification = VerificationRequest::new(
83            self.verifications.clone(),
84            self.store.clone(),
85            flow_id,
86            user_id,
87            recipient_devices,
88            methods,
89        );
90
91        self.insert_request(verification.clone());
92
93        let request = verification.request_to_device();
94
95        (verification, request.into())
96    }
97
98    pub fn request_verification(
99        &self,
100        identity: &OtherUserIdentityData,
101        room_id: &RoomId,
102        request_event_id: &EventId,
103        methods: Option<Vec<VerificationMethod>>,
104    ) -> VerificationRequest {
105        let flow_id = FlowId::InRoom(room_id.to_owned(), request_event_id.to_owned());
106
107        let request = VerificationRequest::new(
108            self.verifications.clone(),
109            self.store.clone(),
110            flow_id,
111            identity.user_id(),
112            vec![],
113            methods,
114        );
115
116        self.insert_request(request.clone());
117
118        request
119    }
120
121    pub async fn start_sas(
122        &self,
123        device: DeviceData,
124    ) -> Result<(Sas, OutgoingVerificationRequest), CryptoStoreError> {
125        let identities = self.store.get_identities(device.clone()).await?;
126        let (sas, content) = Sas::start(identities, TransactionId::new(), true, None, None);
127
128        let request = match content {
129            OutgoingContent::Room(r, c) => {
130                RoomMessageRequest { room_id: r, txn_id: TransactionId::new(), content: c }.into()
131            }
132            OutgoingContent::ToDevice(c) => {
133                let request = ToDeviceRequest::with_id(
134                    device.user_id(),
135                    device.device_id().to_owned(),
136                    &c,
137                    TransactionId::new(),
138                );
139
140                self.verifications.insert_sas(sas.clone());
141
142                request.into()
143            }
144        };
145
146        Ok((sas, request))
147    }
148
149    pub fn get_request(
150        &self,
151        user_id: &UserId,
152        flow_id: impl AsRef<str>,
153    ) -> Option<VerificationRequest> {
154        self.requests.read().get(user_id)?.get(flow_id.as_ref()).cloned()
155    }
156
157    pub fn get_requests(&self, user_id: &UserId) -> Vec<VerificationRequest> {
158        self.requests.read().get(user_id).map(|v| v.values().cloned().collect()).unwrap_or_default()
159    }
160
161    /// Add a new `VerificationRequest` object to the cache.
162    /// If there are any existing requests with this user (and different
163    /// flow_id), both the existing and new request will be cancelled.
164    fn insert_request(&self, request: VerificationRequest) {
165        if let Some(r) = self.get_request(request.other_user(), request.flow_id().as_str()) {
166            debug!(flow_id = r.flow_id().as_str(), "Ignoring known verification request",);
167            return;
168        }
169
170        let mut requests = self.requests.write();
171        let user_requests = requests.entry(request.other_user().to_owned()).or_default();
172
173        // Cancel all the old verifications requests as well as the new one we
174        // have for this user if someone tries to have two verifications going
175        // on at once.
176        for old_verification in user_requests.values_mut() {
177            if !old_verification.is_cancelled() {
178                warn!(
179                    "Received a new verification request whilst another request \
180                    with the same user is ongoing. Cancelling both requests."
181                );
182
183                if let Some(r) = old_verification.cancel() {
184                    self.verifications.add_request(r.into())
185                }
186
187                if let Some(r) = request.cancel() {
188                    self.verifications.add_request(r.into())
189                }
190            }
191        }
192
193        // We still want to add the new verification request, in case users
194        // want to inspect the verification object a matching
195        // `m.key.verification.request` produced.
196        user_requests.insert(request.flow_id().as_str().to_owned(), request);
197    }
198
199    pub fn get_verification(&self, user_id: &UserId, flow_id: &str) -> Option<Verification> {
200        self.verifications.get(user_id, flow_id)
201    }
202
203    pub fn get_sas(&self, user_id: &UserId, flow_id: &str) -> Option<Box<Sas>> {
204        self.verifications.get_sas(user_id, flow_id)
205    }
206
207    fn is_timestamp_valid(timestamp: MilliSecondsSinceUnixEpoch) -> bool {
208        // The event should be ignored if the event is older than 10 minutes
209        let old_timestamp_threshold: UInt = uint!(600);
210        // The event should be ignored if the event is 5 minutes or more into the
211        // future.
212        let timestamp_threshold: UInt = uint!(300);
213
214        let timestamp = timestamp.as_secs();
215        let now = SecondsSinceUnixEpoch::now().get();
216
217        !(now.saturating_sub(timestamp) > old_timestamp_threshold
218            || timestamp.saturating_sub(now) > timestamp_threshold)
219    }
220
221    fn queue_up_content(
222        &self,
223        recipient: &UserId,
224        recipient_device: &DeviceId,
225        content: OutgoingContent,
226        request_id: Option<RequestInfo>,
227    ) {
228        self.verifications.queue_up_content(recipient, recipient_device, content, request_id)
229    }
230
231    pub fn mark_request_as_sent(&self, request_id: &TransactionId) {
232        self.verifications.mark_request_as_sent(request_id);
233    }
234
235    pub fn outgoing_messages(&self) -> Vec<OutgoingRequest> {
236        self.verifications.outgoing_requests()
237    }
238
239    pub fn garbage_collect(&self) -> Vec<Raw<AnyToDeviceEvent>> {
240        let mut events = vec![];
241
242        let mut requests: Vec<OutgoingVerificationRequest> = {
243            let mut requests = self.requests.write();
244
245            for user_verification in requests.values_mut() {
246                user_verification.retain(|_, v| !(v.is_done() || v.is_cancelled()));
247            }
248            requests.retain(|_, v| !v.is_empty());
249
250            requests.values().flatten().filter_map(|(_, v)| v.cancel_if_timed_out()).collect()
251        };
252
253        requests.extend(self.verifications.garbage_collect());
254
255        for request in requests {
256            if let Ok(OutgoingContent::ToDevice(to_device)) = request.clone().try_into() {
257                if let AnyToDeviceEventContent::KeyVerificationCancel(content) = *to_device {
258                    let event = ToDeviceEvent { content, sender: self.own_user_id().to_owned() };
259
260                    events.push(
261                        Raw::new(&event)
262                            .expect("Failed to serialize m.key_verification.cancel event")
263                            .cast(),
264                    );
265                }
266            }
267
268            self.verifications.add_verification_request(request)
269        }
270
271        events
272    }
273
274    async fn mark_sas_as_done(
275        &self,
276        sas: &Sas,
277        out_content: Option<OutgoingContent>,
278    ) -> Result<(), CryptoStoreError> {
279        match sas.mark_as_done().await? {
280            VerificationResult::Ok => {
281                if let Some(c) = out_content {
282                    self.queue_up_content(sas.other_user_id(), sas.other_device_id(), c, None);
283                }
284            }
285            VerificationResult::Cancel(c) => {
286                if let Some(r) = sas.cancel_with_code(c) {
287                    self.verifications.add_request(r.into());
288                }
289            }
290            VerificationResult::SignatureUpload(r) => {
291                self.verifications.add_request(r.into());
292
293                if let Some(c) = out_content {
294                    self.queue_up_content(sas.other_user_id(), sas.other_device_id(), c, None);
295                }
296            }
297        }
298
299        Ok(())
300    }
301
302    #[instrument(skip_all, fields(flow_id))]
303    pub async fn receive_any_event(
304        &self,
305        event: impl Into<AnyEvent<'_>>,
306    ) -> Result<(), CryptoStoreError> {
307        let event = event.into();
308
309        let Ok(flow_id) = FlowId::try_from(&event) else {
310            // This isn't a verification event, return early.
311            return Ok(());
312        };
313        Span::current().record("flow_id", flow_id.as_str());
314
315        let flow_id_mismatch = || {
316            warn!(
317                flow_id = flow_id.as_str(),
318                "Received a verification event with a mismatched flow id, \
319                 the verification object was created for a in-room \
320                 verification but an event was received over to-device \
321                 messaging or vice versa"
322            );
323        };
324
325        let event_sent_from_us = |event: &AnyEvent<'_>, from_device: &DeviceId| {
326            if event.sender() == self.store.account.user_id {
327                from_device == self.store.account.device_id || event.is_room_event()
328            } else {
329                false
330            }
331        };
332
333        let Some(content) = event.verification_content() else { return Ok(()) };
334        match &content {
335            AnyVerificationContent::Request(r) => {
336                info!(
337                    sender = ?event.sender(),
338                    from_device = r.from_device().as_str(),
339                    "Received a new verification request",
340                );
341
342                let Some(timestamp) = event.timestamp() else {
343                    warn!(
344                        from_device = r.from_device().as_str(),
345                        "The key verification request didn't contain a valid timestamp"
346                    );
347                    return Ok(());
348                };
349
350                if !Self::is_timestamp_valid(timestamp) {
351                    info!(
352                        from_device = r.from_device().as_str(),
353                        ?timestamp,
354                        "The received verification request was too old or too far into the future",
355                    );
356                    return Ok(());
357                }
358
359                if event_sent_from_us(&event, r.from_device()) {
360                    trace!(
361                        from_device = r.from_device().as_str(),
362                        "The received verification request was sent by us, ignoring it",
363                    );
364                    return Ok(());
365                }
366
367                let Some(device_data) =
368                    self.store.get_device(event.sender(), r.from_device()).await?
369                else {
370                    warn!("Could not retrieve the device data for the incoming verification request, ignoring it");
371                    return Ok(());
372                };
373
374                let request = VerificationRequest::from_request(
375                    self.verifications.clone(),
376                    self.store.clone(),
377                    event.sender(),
378                    flow_id,
379                    r,
380                    device_data,
381                );
382
383                self.insert_request(request);
384            }
385            AnyVerificationContent::Cancel(c) => {
386                if let Some(verification) = self.get_request(event.sender(), flow_id.as_str()) {
387                    verification.receive_cancel(event.sender(), c);
388                }
389
390                if let Some(verification) = self.get_verification(event.sender(), flow_id.as_str())
391                {
392                    match verification {
393                        Verification::SasV1(sas) => {
394                            // This won't produce an outgoing content
395                            let _ = sas.receive_any_event(event.sender(), &content);
396                        }
397                        #[cfg(feature = "qrcode")]
398                        Verification::QrV1(qr) => qr.receive_cancel(event.sender(), c),
399                    }
400                }
401            }
402            AnyVerificationContent::Ready(c) => {
403                let Some(request) = self.get_request(event.sender(), flow_id.as_str()) else {
404                    return Ok(());
405                };
406
407                if request.flow_id() == &flow_id {
408                    if let Some(device_data) =
409                        self.store.get_device(event.sender(), c.from_device()).await?
410                    {
411                        request.receive_ready(event.sender(), c, device_data);
412                    } else {
413                        warn!("Could not retrieve the data for the accepting device, ignoring it");
414                    }
415                } else {
416                    flow_id_mismatch();
417                }
418            }
419            AnyVerificationContent::Start(c) => {
420                if let Some(request) = self.get_request(event.sender(), flow_id.as_str()) {
421                    if request.flow_id() == &flow_id {
422                        Box::pin(request.receive_start(event.sender(), c)).await?
423                    } else {
424                        flow_id_mismatch();
425                    }
426                } else if let FlowId::ToDevice(_) = flow_id {
427                    // TODO remove this soon, this has been deprecated by
428                    // MSC3122 https://github.com/matrix-org/matrix-doc/pull/3122
429                    if let Some(device) =
430                        self.store.get_device(event.sender(), c.from_device()).await?
431                    {
432                        let identities = self.store.get_identities(device).await?;
433
434                        match Sas::from_start_event(flow_id, c, identities, None, false) {
435                            Ok(sas) => {
436                                self.verifications.insert_sas(sas);
437                            }
438                            Err(cancellation) => self.queue_up_content(
439                                event.sender(),
440                                c.from_device(),
441                                cancellation,
442                                None,
443                            ),
444                        }
445                    }
446                }
447            }
448            AnyVerificationContent::Accept(_) | AnyVerificationContent::Key(_) => {
449                let Some(sas) = self.get_sas(event.sender(), flow_id.as_str()) else {
450                    return Ok(());
451                };
452
453                if sas.flow_id() != &flow_id {
454                    flow_id_mismatch();
455                    return Ok(());
456                }
457
458                let Some((content, request_info)) = sas.receive_any_event(event.sender(), &content)
459                else {
460                    return Ok(());
461                };
462
463                self.queue_up_content(
464                    sas.other_user_id(),
465                    sas.other_device_id(),
466                    content,
467                    request_info,
468                );
469            }
470            AnyVerificationContent::Mac(_) => {
471                let Some(s) = self.get_sas(event.sender(), flow_id.as_str()) else { return Ok(()) };
472
473                if s.flow_id() != &flow_id {
474                    flow_id_mismatch();
475                    return Ok(());
476                }
477
478                let content = s.receive_any_event(event.sender(), &content);
479
480                if s.is_done() {
481                    Box::pin(self.mark_sas_as_done(&s, content.map(|(c, _)| c))).await?;
482                } else {
483                    // Even if we are not done (yet), there might be content to
484                    // send out, e.g. in the case where we are done with our
485                    // side of the verification process, but the other side has
486                    // not yet sent their "done".
487                    let Some((content, request_id)) = content else { return Ok(()) };
488
489                    self.queue_up_content(
490                        s.other_user_id(),
491                        s.other_device_id(),
492                        content,
493                        request_id,
494                    );
495                }
496            }
497            AnyVerificationContent::Done(c) => {
498                if let Some(verification) = self.get_request(event.sender(), flow_id.as_str()) {
499                    verification.receive_done(event.sender(), c);
500                }
501
502                #[allow(clippy::single_match)]
503                match self.get_verification(event.sender(), flow_id.as_str()) {
504                    Some(Verification::SasV1(sas)) => {
505                        let content = sas.receive_any_event(event.sender(), &content);
506
507                        if sas.is_done() {
508                            Box::pin(self.mark_sas_as_done(&sas, content.map(|(c, _)| c))).await?;
509                        }
510                    }
511                    #[cfg(feature = "qrcode")]
512                    Some(Verification::QrV1(qr)) => {
513                        let (cancellation, request) = Box::pin(qr.receive_done(c)).await?;
514
515                        if let Some(c) = cancellation {
516                            self.verifications.add_request(c.into())
517                        }
518
519                        if let Some(s) = request {
520                            self.verifications.add_request(s.into())
521                        }
522                    }
523                    None => {}
524                }
525            }
526        }
527
528        Ok(())
529    }
530}
531
532#[cfg(test)]
533mod tests {
534    use std::sync::Arc;
535
536    use matrix_sdk_test::async_test;
537    use ruma::TransactionId;
538    use tokio::sync::Mutex;
539
540    use super::{Sas, VerificationMachine};
541    use crate::{
542        olm::PrivateCrossSigningIdentity,
543        store::{CryptoStoreWrapper, MemoryStore},
544        verification::{
545            cache::VerificationCache,
546            event_enums::{AcceptContent, KeyContent, MacContent, OutgoingContent},
547            tests::{alice_device_id, alice_id, setup_stores, wrap_any_to_device_content},
548            FlowId, VerificationStore,
549        },
550        Account, VerificationRequest,
551    };
552
553    async fn verification_machine() -> (VerificationMachine, VerificationStore) {
554        let (_account, store, _bob, bob_store) = setup_stores().await;
555
556        let machine = VerificationMachine {
557            store,
558            verifications: VerificationCache::new(),
559            requests: Default::default(),
560        };
561
562        (machine, bob_store)
563    }
564
565    async fn setup_verification_machine() -> (VerificationMachine, Sas) {
566        let (machine, bob_store) = verification_machine().await;
567
568        let alice_device =
569            bob_store.get_device(alice_id(), alice_device_id()).await.unwrap().unwrap();
570
571        let identities = bob_store.get_identities(alice_device).await.unwrap();
572        let (bob_sas, start_content) =
573            Sas::start(identities, TransactionId::new(), true, None, None);
574
575        machine
576            .receive_any_event(&wrap_any_to_device_content(bob_sas.user_id(), start_content))
577            .await
578            .unwrap();
579
580        (machine, bob_sas)
581    }
582
583    #[async_test]
584    async fn test_create() {
585        let alice = Account::with_device_id(alice_id(), alice_device_id());
586        let identity = Arc::new(Mutex::new(PrivateCrossSigningIdentity::empty(alice_id())));
587        let _ = VerificationMachine::new(
588            alice.static_data,
589            identity,
590            Arc::new(CryptoStoreWrapper::new(alice_id(), alice_device_id(), MemoryStore::new())),
591        );
592    }
593
594    #[async_test]
595    async fn test_full_flow() {
596        let (alice_machine, bob) = setup_verification_machine().await;
597
598        let alice = alice_machine.get_sas(bob.user_id(), bob.flow_id().as_str()).unwrap();
599
600        let request = alice.accept().unwrap();
601
602        let content = OutgoingContent::try_from(request).unwrap();
603        let content = AcceptContent::try_from(&content).unwrap().into();
604
605        let (content, request_info) = bob.receive_any_event(alice.user_id(), &content).unwrap();
606
607        let event = wrap_any_to_device_content(bob.user_id(), content);
608
609        assert!(alice_machine.verifications.outgoing_requests().is_empty());
610        alice_machine.receive_any_event(&event).await.unwrap();
611        assert!(!alice_machine.verifications.outgoing_requests().is_empty());
612
613        let request = alice_machine.verifications.outgoing_requests().first().cloned().unwrap();
614        let txn_id = request.request_id().to_owned();
615        let content = OutgoingContent::try_from(request).unwrap();
616        let content = KeyContent::try_from(&content).unwrap().into();
617
618        alice_machine.mark_request_as_sent(&txn_id);
619
620        assert!(bob.receive_any_event(alice.user_id(), &content).is_none());
621
622        assert!(alice.emoji().is_some());
623        // Bob can only show the emoji if it marks the request carrying the
624        // m.key.verification.key event as sent.
625        assert!(bob.emoji().is_none());
626        bob.mark_request_as_sent(&request_info.unwrap().request_id);
627        assert!(bob.emoji().is_some());
628        assert_eq!(alice.emoji(), bob.emoji());
629
630        let mut requests = alice.confirm().await.unwrap().0;
631        assert!(requests.len() == 1);
632        let request = requests.pop().unwrap();
633        let content = OutgoingContent::try_from(request).unwrap();
634        let content = MacContent::try_from(&content).unwrap().into();
635        bob.receive_any_event(alice.user_id(), &content);
636
637        let mut requests = bob.confirm().await.unwrap().0;
638        assert!(requests.len() == 1);
639        let request = requests.pop().unwrap();
640        let content = OutgoingContent::try_from(request).unwrap();
641        let content = MacContent::try_from(&content).unwrap().into();
642        alice.receive_any_event(bob.user_id(), &content);
643
644        assert!(alice.is_done());
645        assert!(bob.is_done());
646    }
647
648    #[cfg(not(target_os = "macos"))]
649    #[allow(unknown_lints, clippy::unchecked_duration_subtraction)]
650    #[async_test]
651    async fn test_timing_out() {
652        use std::time::Duration;
653
654        use ruma::time::Instant;
655
656        let (alice_machine, bob) = setup_verification_machine().await;
657        let alice = alice_machine.get_sas(bob.user_id(), bob.flow_id().as_str()).unwrap();
658
659        assert!(!alice.timed_out());
660        assert!(alice_machine.verifications.outgoing_requests().is_empty());
661
662        // This line panics on macOS, so we're disabled for now.
663        alice.set_creation_time(Instant::now() - Duration::from_secs(60 * 15));
664        assert!(alice.timed_out());
665        assert!(alice_machine.verifications.outgoing_requests().is_empty());
666        alice_machine.garbage_collect();
667        assert!(!alice_machine.verifications.outgoing_requests().is_empty());
668        alice_machine.garbage_collect();
669        assert!(alice_machine.verifications.is_empty());
670    }
671
672    /// Test to ensure that we cancel both verifications if a second one gets
673    /// started while another one is going on.
674    #[async_test]
675    async fn test_double_verification_cancellation() {
676        let (machine, bob_store) = verification_machine().await;
677
678        let alice_device =
679            bob_store.get_device(alice_id(), alice_device_id()).await.unwrap().unwrap();
680        let identities = bob_store.get_identities(alice_device).await.unwrap();
681
682        // Start the first sas verification.
683        let (bob_sas, start_content) =
684            Sas::start(identities.clone(), TransactionId::new(), true, None, None);
685
686        machine
687            .receive_any_event(&wrap_any_to_device_content(bob_sas.user_id(), start_content))
688            .await
689            .unwrap();
690
691        let alice_sas = machine.get_sas(bob_sas.user_id(), bob_sas.flow_id().as_str()).unwrap();
692
693        // We're not yet cancelled.
694        assert!(!alice_sas.is_cancelled());
695
696        let second_transaction_id = TransactionId::new();
697        let (bob_sas, start_content) =
698            Sas::start(identities, second_transaction_id.clone(), true, None, None);
699        machine
700            .receive_any_event(&wrap_any_to_device_content(bob_sas.user_id(), start_content))
701            .await
702            .unwrap();
703
704        let second_sas = machine.get_sas(bob_sas.user_id(), bob_sas.flow_id().as_str()).unwrap();
705
706        // Make sure we fetched the new one.
707        assert_eq!(second_sas.flow_id().as_str(), second_transaction_id);
708
709        // Make sure both of them are cancelled.
710        assert!(alice_sas.is_cancelled());
711        assert!(second_sas.is_cancelled());
712    }
713
714    /// Test to ensure that we cancel both verification requests if a second one
715    /// gets started while another one is going on.
716    #[async_test]
717    async fn test_double_verification_request_cancellation() {
718        let (machine, bob_store) = verification_machine().await;
719
720        // Start the first verification request.
721        let flow_id = FlowId::ToDevice("TEST_FLOW_ID".into());
722
723        let bob_request = VerificationRequest::new(
724            VerificationCache::new(),
725            bob_store.clone(),
726            flow_id.clone(),
727            alice_id(),
728            vec![],
729            None,
730        );
731
732        let request = bob_request.request_to_device();
733        let content: OutgoingContent = request.try_into().unwrap();
734
735        machine
736            .receive_any_event(&wrap_any_to_device_content(bob_request.own_user_id(), content))
737            .await
738            .unwrap();
739
740        let alice_request =
741            machine.get_request(bob_request.own_user_id(), bob_request.flow_id().as_str()).unwrap();
742
743        // We're not yet cancelled.
744        assert!(!alice_request.is_cancelled());
745
746        let second_transaction_id = TransactionId::new();
747        let bob_request = VerificationRequest::new(
748            VerificationCache::new(),
749            bob_store,
750            second_transaction_id.clone().into(),
751            alice_id(),
752            vec![],
753            None,
754        );
755
756        let request = bob_request.request_to_device();
757        let content: OutgoingContent = request.try_into().unwrap();
758
759        machine
760            .receive_any_event(&wrap_any_to_device_content(bob_request.own_user_id(), content))
761            .await
762            .unwrap();
763
764        let second_request =
765            machine.get_request(bob_request.own_user_id(), bob_request.flow_id().as_str()).unwrap();
766
767        // Make sure we fetched the new one.
768        assert_eq!(second_request.flow_id().as_str(), second_transaction_id);
769
770        // Make sure both of them are cancelled.
771        assert!(alice_request.is_cancelled());
772        assert!(second_request.is_cancelled());
773    }
774
775    /// Ensure that if a duplicate request is added (i.e. matching user and
776    /// flow_id) the existing request is not cancelled and the new one is
777    /// ignored
778    #[async_test]
779    async fn test_ignore_identical_verification_request() {
780        let (machine, bob_store) = verification_machine().await;
781
782        // Start the first verification request.
783        let flow_id = FlowId::ToDevice("TEST_FLOW_ID".into());
784
785        let bob_request = VerificationRequest::new(
786            VerificationCache::new(),
787            bob_store.clone(),
788            flow_id.clone(),
789            alice_id(),
790            vec![],
791            None,
792        );
793
794        let request = bob_request.request_to_device();
795        let content: OutgoingContent = request.try_into().unwrap();
796
797        machine
798            .receive_any_event(&wrap_any_to_device_content(bob_request.own_user_id(), content))
799            .await
800            .unwrap();
801
802        let first_request =
803            machine.get_request(bob_request.own_user_id(), bob_request.flow_id().as_str()).unwrap();
804
805        // We're not yet cancelled.
806        assert!(!first_request.is_cancelled());
807
808        // Bob is adding a second request with the same flow_id as before
809        let bob_request = VerificationRequest::new(
810            VerificationCache::new(),
811            bob_store,
812            flow_id.clone(),
813            alice_id(),
814            vec![],
815            None,
816        );
817
818        let request = bob_request.request_to_device();
819        let content: OutgoingContent = request.try_into().unwrap();
820
821        machine
822            .receive_any_event(&wrap_any_to_device_content(bob_request.own_user_id(), content))
823            .await
824            .unwrap();
825
826        let second_request =
827            machine.get_request(bob_request.own_user_id(), bob_request.flow_id().as_str()).unwrap();
828
829        // None of the requests are cancelled
830        assert!(!first_request.is_cancelled());
831        assert!(!second_request.is_cancelled());
832    }
833}