macltok

public
Star0
Watch0
123 commits 3 branches 0 tags 0 releases3.0 MiB
use std::sync::Arc;

use futures_util::StreamExt;
use matrix_sdk::executor::{JoinHandle, spawn};
use matrix_sdk::{
    Client,
    encryption::{
        VerificationState,
        backups::BackupState,
        recovery::RecoveryState as SdkRecoveryState,
        verification::{
            SasState, SasVerification, Verification, VerificationRequest, VerificationRequestState,
        },
    },
    ruma::events::key::verification::VerificationMethod,
};
use ruma::{OwnedDeviceId, UserId};
use tokio::sync::Mutex;

use crate::{
    EncryptionState, Error, ErrorKind, Event, KeyBackupState, RecoveryState, VerificationAction,
    VerificationState as UiState, VerificationTarget, model::VerificationFailure,
    transport::EventSender,
};

pub(crate) fn encryption_state(
    device_id: OwnedDeviceId,
    recovery: SdkRecoveryState,
    verification: VerificationState,
    backup: BackupState,
) -> EncryptionState {
    EncryptionState {
        device_id,
        device_verified: matches!(verification, VerificationState::Verified),
        recovery: match recovery {
            SdkRecoveryState::Unknown => RecoveryState::Unknown,
            SdkRecoveryState::Disabled => RecoveryState::Disabled,
            SdkRecoveryState::Incomplete => RecoveryState::Incomplete,
            SdkRecoveryState::Enabled => RecoveryState::Enabled,
        },
        key_backup: match backup {
            BackupState::Unknown => KeyBackupState::Unknown,
            BackupState::Creating => KeyBackupState::Creating,
            BackupState::Enabling => KeyBackupState::Enabling,
            BackupState::Resuming => KeyBackupState::Resuming,
            BackupState::Enabled => KeyBackupState::Enabled,
            BackupState::Downloading => KeyBackupState::Downloading,
            BackupState::Disabling => KeyBackupState::Disabling,
        },
    }
}

#[derive(Default)]
pub(crate) struct VerificationController {
    active: Arc<Mutex<VerificationSession>>,
    task: Option<JoinHandle<()>>,
}

#[derive(Default)]
struct VerificationSession {
    request: Option<VerificationRequest>,
    sas: Option<SasVerification>,
}

impl VerificationController {
    pub(crate) async fn apply(
        &mut self,
        sender: EventSender,
        action: VerificationAction,
    ) -> Result<(), Error> {
        match action {
            VerificationAction::Accept => self.accept().await,
            VerificationAction::Confirm => {
                // Retain a handle across the operation: the watcher clears
                // `active` when it publishes a terminal event, but this clone
                // still observes the shared SDK state afterwards.
                let sas = self.active.lock().await.sas.clone();
                match self.confirm().await {
                    Ok(()) => Ok(()),
                    Err(error) if error.kind == ErrorKind::VerificationInactive => Err(error),
                    // The confirm advanced local SDK state before its send failed,
                    // so the retained session could report success on retry without
                    // network traffic. Best-effort cancel (no lock held across the
                    // network), clear for restart, then report the terminal state.
                    Err(error) => {
                        // Stop the watcher first so it cannot emit a stale
                        // Confirmed from the local transition we are unwinding.
                        self.stop_task().await;
                        // The peer can complete or cancel the verification
                        // while our own Done send fails. The stopped watcher may
                        // already have published that terminal outcome (losing the
                        // claim below); otherwise this path owns it and must emit
                        // it instead of regressing to Cancelled.
                        if sas.as_ref().is_some_and(|sas| sas.is_done()) {
                            if take_verification(&self.active).await {
                                let _ =
                                    sender.send(Event::VerificationChanged(UiState::Done)).await;
                            }
                            return Err(error);
                        }
                        if let Some(reason) = sas
                            .as_ref()
                            .and_then(|sas| sas.cancel_info())
                            .map(|info| info.reason().to_owned())
                        {
                            if take_verification(&self.active).await {
                                let _ = sender
                                    .send(Event::VerificationChanged(UiState::Cancelled(reason)))
                                    .await;
                            }
                            return Err(error);
                        }
                        if let Some(sas) = sas {
                            if sas.cancel().await.is_err() {
                                // The best-effort unwind already advanced
                                // local state; clearing here keeps the failed
                                // cancellation from looking retryable. The
                                // send is owned only if the watcher has not
                                // already claimed the session for its own
                                // terminal outcome.
                                if take_verification(&self.active).await {
                                    let _ = sender
                                        .send(Event::VerificationChanged(UiState::Cancelled(
                                            error.to_string(),
                                        )))
                                        .await;
                                }
                                return Err(error);
                            }
                            // The peer can complete while the unwind cancel is in
                            // flight and the SDK call no-ops; the Done then owns the
                            // terminal outcome, so report it instead of attributing
                            // it to the send failure.
                            if sas.is_done() {
                                if take_verification(&self.active).await {
                                    let _ = sender
                                        .send(Event::VerificationChanged(UiState::Done))
                                        .await;
                                }
                                return Err(error);
                            }
                            if let Some(info) = sas.cancel_info() {
                                // Our own unwind cancel just marked the flow; it does
                                // not own the outcome yet, so fall through and report
                                // the send failure as Cancelled below. Only a peer
                                // cancellation racing the unwind owns it instead.
                                if !info.cancelled_by_us() {
                                    if take_verification(&self.active).await {
                                        let _ = sender
                                            .send(Event::VerificationChanged(UiState::Cancelled(
                                                info.reason().to_owned(),
                                            )))
                                            .await;
                                    }
                                    return Err(Error::new(
                                        ErrorKind::VerificationInactive,
                                        info.reason().to_owned(),
                                    ));
                                }
                            }
                        }
                        if take_verification(&self.active).await {
                            let _ = sender
                                .send(Event::VerificationChanged(UiState::Cancelled(
                                    error.to_string(),
                                )))
                                .await;
                        }
                        Err(error)
                    }
                }
            }
            VerificationAction::Reject => self.reject().await,
            VerificationAction::Cancel => {
                // The SDK no-ops a cancel of an already-settled flow, so a
                // failed claim inside `cancel` already reported the true
                // outcome as an error and owns no event here. Only the claim
                // winner emits the local Cancelled.
                let claimed = self.cancel_claimed(Some(&sender)).await?;
                if claimed {
                    sender
                        .send(Event::VerificationChanged(UiState::Cancelled(
                            "cancelled by this client".to_owned(),
                        )))
                        .await
                        .map_err(|source| {
                            Error::with_source(
                                ErrorKind::Interrupted,
                                "event receiver has closed",
                                source,
                            )
                        })?;
                }
                Ok(())
            }
        }
    }

    pub(crate) async fn start(
        &mut self,
        client: &Client,
        sender: EventSender,
        target: VerificationTarget,
    ) -> Result<(), Error> {
        client
            .encryption()
            .wait_for_e2ee_initialization_tasks()
            .await;
        let own_user_id = client.user_id().ok_or_else(|| {
            Error::new(
                ErrorKind::Unauthenticated,
                "the Matrix client is not authenticated",
            )
        })?;
        if self.has_active_verification().await {
            self.cancel().await?;
        } else {
            self.stop_task().await;
        }
        let request = match &target {
            VerificationTarget::OwnIdentity => {
                request_identity_verification(client, own_user_id).await?
            }
            VerificationTarget::User(user_id) => {
                request_identity_verification(client, user_id).await?
            }
            VerificationTarget::Device { user_id, device_id } => {
                let device = client
                    .encryption()
                    .get_device(user_id, device_id)
                    .await
                    .map_err(|source| {
                        Error::with_source(
                            ErrorKind::Encryption,
                            "reading the verification device failed",
                            source,
                        )
                    })?
                    .ok_or_else(|| {
                        Error::new(
                            ErrorKind::VerificationUnsupported,
                            "the verification device is unavailable",
                        )
                    })?;
                device
                    .request_verification_with_methods(vec![VerificationMethod::SasV1])
                    .await
                    .map_err(|source| {
                        Error::from_sdk("starting device verification failed", source)
                    })?
            }
        };
        // subscribed before anything can wait on the event channel, or a quick peer's
        // answer lands while we're blocked and nobody ever hears it
        let changes = request.changes();
        self.active.lock().await.request = Some(request.clone());
        sender
            .send(Event::VerificationTargetChanged(target))
            .await
            .map_err(|source| {
                Error::with_source(ErrorKind::Interrupted, "event receiver has closed", source)
            })?;
        sender
            .send(Event::VerificationChanged(UiState::Waiting))
            .await
            .map_err(|source| {
                Error::with_source(ErrorKind::Interrupted, "event receiver has closed", source)
            })?;
        let active = self.active.clone();
        self.task = Some(spawn(async move {
            follow_request(request, changes, active, sender).await;
        }));
        Ok(())
    }

    /// another device asked us. only taken when nothing else is in flight, and not accepted
    /// until the user says so with `VerificationAction::Accept`
    pub(crate) async fn adopt_incoming(
        &mut self,
        sender: EventSender,
        request: VerificationRequest,
    ) {
        if request.we_started() || request.is_done() || request.cancel_info().is_some() {
            return;
        }
        if self.has_active_verification().await {
            tracing::debug!("ignoring a verification request while another is in flight");
            return;
        }
        self.stop_task().await;
        let changes = request.changes();
        self.active.lock().await.request = Some(request.clone());
        let incoming = UiState::Incoming {
            user_id: request.other_user_id().to_owned(),
            device_id: match request.state() {
                VerificationRequestState::Requested {
                    other_device_data, ..
                }
                | VerificationRequestState::Ready {
                    other_device_data, ..
                } => Some(other_device_data.device_id().to_owned()),
                _ => None,
            },
        };
        let target = VerificationTarget::User(request.other_user_id().to_owned());
        for event in [
            Event::VerificationTargetChanged(target),
            Event::VerificationChanged(incoming),
        ] {
            if sender.send(event).await.is_err() {
                return;
            }
        }
        let active = self.active.clone();
        self.task = Some(spawn(async move {
            follow_request(request, changes, active, sender).await;
        }));
    }

    async fn accept(&self) -> Result<(), Error> {
        let inactive = |message: &str| Error::new(ErrorKind::VerificationInactive, message);
        let (request, sas) = {
            let active = self.active.lock().await;
            (active.request.clone(), active.sas.is_some())
        };
        let request = request.ok_or_else(|| inactive("no verification request is pending"))?;
        if request.we_started() {
            return Err(inactive("nobody asked us, we asked them"));
        }
        if request.is_done() {
            return Err(inactive("verification is already complete"));
        }
        if let Some(info) = request.cancel_info() {
            return Err(inactive(info.reason()));
        }
        if sas {
            return Err(inactive("verification has already started"));
        }
        request
            .accept_with_methods(vec![VerificationMethod::SasV1])
            .await
            .map_err(|source| Error::from_sdk("accepting verification request failed", source))
    }

    async fn confirm(&self) -> Result<(), Error> {
        let sas = self.active.lock().await.sas.clone().ok_or_else(|| {
            Error::new(
                ErrorKind::VerificationInactive,
                "verification has not reached comparison",
            )
        })?;
        // Report a settled flow truthfully instead of as "not reached
        // comparison": the SDK would no-op the confirm with success.
        if sas.is_done() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "verification is already complete",
            ));
        }
        if let Some(info) = sas.cancel_info() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                info.reason().to_owned(),
            ));
        }
        // Only the short-auth-string stage is confirmable. The SDK reports the
        // remote-MAC-arrived state through the same presentable flag, while an
        // early confirm would locally no-op and falsely report success.
        if !sas.can_be_presented() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "verification has not reached comparison",
            ));
        }
        // Our own confirm can move the flow to Done, so only Done that predates
        // the send means the SDK no-op'd on a stale session; a cancellation
        // racing the send must not report success.
        let was_done = sas.is_done();
        sas.confirm()
            .await
            .map_err(|source| Error::from_sdk("confirming verification failed", source))?;
        if sas.is_done() {
            if was_done {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    "verification is already complete",
                ));
            }
        } else if sas.is_cancelled() {
            let reason = sas
                .cancel_info()
                .map(|info| info.reason().to_owned())
                .unwrap_or_else(|| "verification was cancelled".to_owned());
            return Err(Error::new(ErrorKind::VerificationInactive, reason));
        }
        Ok(())
    }

    async fn reject(&self) -> Result<(), Error> {
        let sas = self.active.lock().await.sas.clone().ok_or_else(|| {
            Error::new(
                ErrorKind::VerificationInactive,
                "verification has not reached comparison",
            )
        })?;
        // Report a settled flow truthfully instead of as "not reached
        // comparison": the SDK would no-op the mismatch with success.
        if sas.is_done() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "verification is already complete",
            ));
        }
        if let Some(info) = sas.cancel_info() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                info.reason().to_owned(),
            ));
        }
        // Rejecting before the comparison stage would cancel a flow the user
        // never compared, so it requires the same presentable state.
        if !sas.can_be_presented() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "verification has not reached comparison",
            ));
        }
        sas.mismatch()
            .await
            .map_err(|source| Error::from_sdk("rejecting verification failed", source))?;
        // A peer completing the flow concurrently must not be reported as
        // rejected. A remote cancellation racing the send turns the SDK call
        // into a no-op success, so report it instead of a false success.
        if sas.is_done() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "verification is already complete",
            ));
        }
        if let Some(info) = sas.cancel_info() {
            // Our own mismatch just marked the flow cancelled-by-us; that is the
            // expected reject path and stays Ok. Only a peer cancel reports an
            // error.
            if !info.cancelled_by_us() {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    info.reason().to_owned(),
                ));
            }
        }
        Ok(())
    }
    pub(crate) async fn cancel(&mut self) -> Result<(), Error> {
        self.cancel_claimed(None).await.map(|_| ())
    }

    /// Return whether this action won the claim for a local cancellation.
    /// The optional sender reports a peer terminal outcome when it races after
    /// the claim; starting a replacement verification does not publish it.
    async fn cancel_claimed(&mut self, sender: Option<&EventSender>) -> Result<bool, Error> {
        let (sas, request) = {
            let active = self.active.lock().await;
            (active.sas.clone(), active.request.clone())
        };
        if sas.is_none() && request.is_none() {
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "no verification is active",
            ));
        }
        // The watcher owns terminal emission and claims the session when it
        // publishes. If the SDK is already terminal, the watcher either already
        // reported it or is about to: leave session and watcher alone so the
        // outcome is emitted exactly once, and report it instead of letting the
        // SDK no-op success regress Done to "cancelled by this client".
        if let Some(sas) = sas.as_ref() {
            if sas.is_done() {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    "verification is already complete",
                ));
            }
            if let Some(info) = sas.cancel_info() {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    info.reason().to_owned(),
                ));
            }
        } else if let Some(request) = request.as_ref() {
            if request.is_done() {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    "verification is already complete",
                ));
            }
            if let Some(info) = request.cancel_info() {
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    info.reason().to_owned(),
                ));
            }
        }
        // Own the terminal outcome from here: claim the session before
        // stopping the watcher so a watcher that already claimed (and will
        // emit its own terminal event) turns this into a truthful error
        // instead of a duplicate local cancel; no lock is held across sends.
        if !take_verification(&self.active).await {
            // Re-check the SDK: the claim loss may be a stale session after a
            // terminal outcome, or a concurrent start of a new flow. Only the
            // terminal case reports here; a live session belongs to the new
            // flow and must not be disturbed.
            if let Some(sas) = sas.as_ref() {
                if sas.is_done() {
                    return Err(Error::new(
                        ErrorKind::VerificationInactive,
                        "verification is already complete",
                    ));
                }
                if let Some(info) = sas.cancel_info() {
                    return Err(Error::new(
                        ErrorKind::VerificationInactive,
                        info.reason().to_owned(),
                    ));
                }
            } else if let Some(request) = request.as_ref() {
                if request.is_done() {
                    return Err(Error::new(
                        ErrorKind::VerificationInactive,
                        "verification is already complete",
                    ));
                }
                if let Some(info) = request.cancel_info() {
                    return Err(Error::new(
                        ErrorKind::VerificationInactive,
                        info.reason().to_owned(),
                    ));
                }
            }
            return Err(Error::new(
                ErrorKind::VerificationInactive,
                "no verification is active",
            ));
        }
        self.stop_task().await;
        if let Some(sas) = sas {
            let result = sas.cancel().await;
            if sas.is_done() {
                report_claimed_terminal(sender, UiState::Done).await;
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    "verification is already complete",
                ));
            }
            if let Some(info) = sas.cancel_info()
                && !info.cancelled_by_us()
            {
                let reason = info.reason().to_owned();
                report_claimed_terminal(sender, UiState::Cancelled(reason.clone())).await;
                return Err(Error::new(ErrorKind::VerificationInactive, reason));
            }
            if let Err(source) = result {
                let error = Error::from_sdk("cancelling verification failed", source);
                report_claimed_terminal(sender, UiState::Cancelled(error.to_string())).await;
                return Err(error);
            }
            Ok(true)
        } else if let Some(request) = request {
            let result = request.cancel().await;
            if request.is_done() {
                report_claimed_terminal(sender, UiState::Done).await;
                return Err(Error::new(
                    ErrorKind::VerificationInactive,
                    "verification is already complete",
                ));
            }
            if let Some(info) = request.cancel_info()
                && !info.cancelled_by_us()
            {
                let reason = info.reason().to_owned();
                report_claimed_terminal(sender, UiState::Cancelled(reason.clone())).await;
                return Err(Error::new(ErrorKind::VerificationInactive, reason));
            }
            if let Err(source) = result {
                let error = Error::from_sdk("cancelling verification failed", source);
                report_claimed_terminal(sender, UiState::Cancelled(error.to_string())).await;
                return Err(error);
            }
            Ok(true)
        } else {
            Ok(false)
        }
    }

    async fn has_active_verification(&self) -> bool {
        let active = self.active.lock().await;
        active.request.is_some() || active.sas.is_some()
    }

    async fn stop_task(&mut self) {
        if let Some(task) = self.task.take() {
            task.abort();
            let _ = task.await;
        }
    }
}

/// shutdown forgets the flow without telling the sdk, a network send could hang it
impl Drop for VerificationController {
    fn drop(&mut self) {
        if let Some(task) = self.task.take() {
            task.abort();
        }
    }
}

async fn request_identity_verification(
    client: &Client,
    user_id: &UserId,
) -> Result<VerificationRequest, Error> {
    let identity = client
        .encryption()
        .request_user_identity(user_id)
        .await
        .map_err(|source| Error::from_sdk("requesting the user identity failed", source))?
        .ok_or_else(|| {
            Error::new(
                ErrorKind::NoCrossSigningIdentity,
                "the user has no cross-signing identity",
            )
        })?;
    identity
        .request_verification_with_methods(vec![VerificationMethod::SasV1])
        .await
        .map_err(|source| Error::from_verification_request("starting verification failed", source))
}

/// `changes` is subscribed by the caller before the request could move, and the current state
/// is handled first, since the stream only reports what happens after it was taken
async fn follow_request(
    request: VerificationRequest,
    changes: impl futures_util::Stream<Item = VerificationRequestState> + Unpin,
    active: Arc<Mutex<VerificationSession>>,
    sender: EventSender,
) {
    let mut states = futures_util::stream::iter([request.state()]).chain(changes);
    while let Some(state) = states.next().await {
        match state {
            // created is ours waiting on them, requested is theirs waiting on `Accept`
            VerificationRequestState::Created { .. }
            | VerificationRequestState::Requested { .. } => {}
            // both sides starting SAS collide into a cancel, so like Element the side that
            // asked starts and the side that accepted waits for `Transitioned`
            VerificationRequestState::Ready { .. } if !request.we_started() => {}
            VerificationRequestState::Ready { .. } => match request.start_sas().await {
                Ok(Some(sas)) => {
                    follow_sas(sas, active.clone(), sender.clone()).await;
                    return;
                }
                Ok(None) => {}
                Err(error) => {
                    tracing::warn!(error = %error, "starting SAS verification failed");
                    // Claim the session so this terminal outcome is emitted exactly
                    // once; a lost claim means an action path already owns it.
                    report_watcher_terminal(
                        &active,
                        &sender,
                        UiState::Failed(VerificationFailure::SasStartup(error.to_string())),
                        "SAS startup failure",
                    )
                    .await;
                    return;
                }
            },
            VerificationRequestState::Transitioned { verification } => {
                if let Verification::SasV1(sas) = verification {
                    follow_sas(sas, active.clone(), sender.clone()).await;
                } else {
                    report_watcher_terminal(
                        &active,
                        &sender,
                        UiState::Cancelled(
                            "the other device selected an unsupported method".into(),
                        ),
                        "unsupported verification method",
                    )
                    .await;
                }
                return;
            }
            VerificationRequestState::Done => {
                report_watcher_terminal(&active, &sender, UiState::Done, "completion").await;
                return;
            }
            VerificationRequestState::Cancelled(info) => {
                report_watcher_terminal(
                    &active,
                    &sender,
                    UiState::Cancelled(info.reason().to_owned()),
                    "cancellation",
                )
                .await;
                return;
            }
        }
    }
    tracing::debug!("verification request watcher ended unexpectedly");
    // The stream may end after the SDK already settled without delivering the
    // terminal update; report the settled outcome instead of StreamEnded, and
    // stay silent if an action path already claimed the session.
    if request.is_done() {
        report_watcher_terminal(&active, &sender, UiState::Done, "completion").await;
    } else if let Some(info) = request.cancel_info() {
        let reason = info.reason().to_owned();
        report_watcher_terminal(&active, &sender, UiState::Cancelled(reason), "cancellation").await;
    } else {
        report_watcher_terminal(
            &active,
            &sender,
            UiState::Failed(VerificationFailure::RequestStreamEnded),
            "request stream termination",
        )
        .await;
    }
}

async fn follow_sas(
    sas: SasVerification,
    active: Arc<Mutex<VerificationSession>>,
    sender: EventSender,
) {
    // the sdk may already have moved the flow along before we got here, a peer starting
    // sas is the usual case, so take the stream first and the current state after
    let changes = sas.changes();
    active.lock().await.sas = Some(sas.clone());
    let mut states = futures_util::stream::iter([sas.state()]).chain(changes);
    while let Some(state) = states.next().await {
        let update = match state {
            SasState::Created { .. } | SasState::Accepted { .. } => continue,
            SasState::Started { .. } => {
                if let Err(error) = sas.accept().await {
                    tracing::warn!(error = %error, "accepting SAS verification failed");
                    // Claim the session so this terminal outcome is emitted exactly
                    // once; a lost claim means an action path already owns it.
                    report_watcher_terminal(
                        &active,
                        &sender,
                        UiState::Failed(VerificationFailure::SasAcceptance(error.to_string())),
                        "SAS acceptance failure",
                    )
                    .await;
                    return;
                }
                continue;
            }
            SasState::KeysExchanged { emojis, decimals } => UiState::Comparison {
                emojis: emojis
                    .map(|short_auth| {
                        short_auth
                            .emojis
                            .into_iter()
                            .map(|emoji| (emoji.symbol.to_owned(), emoji.description.to_owned()))
                            .collect()
                    })
                    .unwrap_or_default(),
                decimals,
            },
            SasState::Confirmed => UiState::Confirmed,
            SasState::Done { .. } => UiState::Done,
            SasState::Cancelled(info) => UiState::Cancelled(info.reason().to_owned()),
        };
        let terminal = matches!(update, UiState::Done | UiState::Cancelled(_));
        // Claim the session before the terminal send so a racing Cancel observes
        // no active session and the SDK no-op cannot regress Done back to
        // "cancelled by this client". A lost claim means an action path already
        // owns the terminal outcome, so stay silent instead of duplicating it.
        if terminal && !take_verification(&active).await {
            return;
        }
        if sender
            .send(Event::VerificationChanged(update))
            .await
            .is_err()
        {
            tracing::debug!("SAS watcher stopped because the event receiver closed");
            return;
        }
        if terminal {
            return;
        }
    }
    tracing::debug!("SAS watcher ended unexpectedly");
    // The stream may end after the SDK already settled without delivering the
    // terminal update; report the settled outcome instead of StreamEnded, and
    // stay silent if an action path already claimed the session.
    if sas.is_done() {
        report_watcher_terminal(&active, &sender, UiState::Done, "completion").await;
    } else if let Some(info) = sas.cancel_info() {
        let reason = info.reason().to_owned();
        report_watcher_terminal(&active, &sender, UiState::Cancelled(reason), "cancellation").await;
    } else {
        report_watcher_terminal(
            &active,
            &sender,
            UiState::Failed(VerificationFailure::SasStreamEnded),
            "SAS stream termination",
        )
        .await;
    }
}

async fn report_watcher_terminal(
    active: &Arc<Mutex<VerificationSession>>,
    sender: &EventSender,
    state: UiState,
    context: &'static str,
) {
    if take_verification(active).await
        && sender
            .send(Event::VerificationChanged(state))
            .await
            .is_err()
    {
        tracing::debug!(
            context,
            "verification watcher could not report terminal outcome"
        );
    }
}

async fn report_claimed_terminal(sender: Option<&EventSender>, state: UiState) {
    if let Some(sender) = sender {
        let _ = sender.send(Event::VerificationChanged(state)).await;
    }
}

/// Claim the active session for terminal emission, if one is still present.
///
/// Exactly one party (watcher or action path) wins the claim and may emit the
/// terminal event; the loser stays silent so Done is never overwritten by a
/// later local Cancelled/Failed.
async fn take_verification(active: &Arc<Mutex<VerificationSession>>) -> bool {
    let mut active = active.lock().await;
    let had = active.sas.is_some() || active.request.is_some();
    active.sas.take();
    active.request.take();
    had
}

#[cfg(test)]
mod tests {
    use matrix_sdk::encryption::VerificationState;
    use matrix_sdk::encryption::backups::BackupState;
    use ruma::owned_device_id;

    use super::*;

    #[test]
    fn sdk_security_state_is_normalized() {
        assert_eq!(
            encryption_state(
                owned_device_id!("DEVICE"),
                SdkRecoveryState::Incomplete,
                VerificationState::Unverified,
                BackupState::Unknown,
            ),
            EncryptionState {
                device_id: owned_device_id!("DEVICE"),
                device_verified: false,
                recovery: RecoveryState::Incomplete,
                key_backup: KeyBackupState::Unknown,
            }
        );
        assert_eq!(
            encryption_state(
                owned_device_id!("DEVICE"),
                SdkRecoveryState::Enabled,
                VerificationState::Verified,
                BackupState::Enabled,
            ),
            EncryptionState {
                device_id: owned_device_id!("DEVICE"),
                device_verified: true,
                recovery: RecoveryState::Enabled,
                key_backup: KeyBackupState::Enabled,
            }
        );
    }

    #[tokio::test]
    async fn comparison_actions_require_an_active_sas() {
        let mut controller = VerificationController::default();

        assert_eq!(
            controller.confirm().await.unwrap_err().to_string(),
            "verification has not reached comparison"
        );
        assert_eq!(
            controller.reject().await.unwrap_err().to_string(),
            "verification has not reached comparison"
        );
        assert_eq!(
            controller.cancel().await.unwrap_err().to_string(),
            "no verification is active"
        );
    }
}