From ebf5eb8f39fffc4e86f6a82b02e9f7ed2c4bf45b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 18:46:57 +0200 Subject: [PATCH 1/8] daemon: Extract scripted transport test fixtures The push-based helper was embedded in the credential service tests, and its cancellation-focused names hid that it could script any transport state. Move it to test_support, rename it around that broader role, and share typed completion and failure helpers across USB, hybrid, and NFC tests. --- credentialsd/src/credential_service/mod.rs | 302 ++++-------------- .../src/credential_service/test_support.rs | 212 ++++++++++++ 2 files changed, 267 insertions(+), 247 deletions(-) create mode 100644 credentialsd/src/credential_service/test_support.rs diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 686e80e..f975d16 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -2,6 +2,9 @@ pub mod hybrid; pub mod nfc; pub mod usb; +#[cfg(test)] +mod test_support; + use std::{ fmt::Debug, pin::Pin, @@ -511,45 +514,10 @@ impl From for AuthenticatorResponse { #[cfg(test)] mod tests { - use super::*; use std::time::Duration; - // Mock handlers for testing - #[derive(Debug)] - struct MockUsbHandler; - impl UsbHandler for MockUsbHandler { - fn start( - &self, - _request: &CredentialRequest, - _cancellation: CancellationToken, - ) -> impl Stream + Send + Sized + Unpin + 'static { - futures::stream::empty() - } - } - - #[derive(Debug)] - struct MockHybridHandler; - impl HybridHandler for MockHybridHandler { - fn start( - &self, - _request: &CredentialRequest, - _cancellation: CancellationToken, - ) -> impl Stream + Unpin + Send + Sized + 'static { - futures::stream::empty() - } - } - - #[derive(Debug)] - struct MockNfcHandler; - impl NfcHandler for MockNfcHandler { - fn start( - &self, - _request: &CredentialRequest, - _cancellation: CancellationToken, - ) -> impl Stream + Send + Sized + Unpin + 'static { - futures::stream::empty() - } - } + use super::test_support::{EmptyTransport, ScriptedTransport}; + use super::*; fn create_test_credential_response() -> CredentialResponse { use libwebauthn::ops::webauthn::GetAssertionResponse; @@ -628,7 +596,7 @@ mod tests { #[tokio::test] async fn test_init_request_returns_token_and_id() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let (tx, _rx) = oneshot::channel(); let request = create_test_request().await; @@ -642,7 +610,7 @@ mod tests { #[tokio::test] async fn test_cancel_request_triggers_cancellation() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let (tx, _rx) = oneshot::channel(); let request = create_test_request().await; @@ -679,7 +647,7 @@ mod tests { #[tokio::test] async fn test_init_request_rejects_concurrent() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let (tx1, _rx1) = oneshot::channel(); let (tx2, _rx2) = oneshot::channel(); let request = create_test_request().await; @@ -699,153 +667,9 @@ mod tests { ); } - // Generic push-based handler that tracks cancellation. - // Before moving a handler into the service, call `get_handler_ref()` to obtain - // a `HandlerRef` — a handle that exposes `shift_state()` and `was_cancelled()` - // for use in the test body. - use std::sync::atomic::{AtomicBool, Ordering}; - use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; - - /// Clone-able test handle for a `CancellationTrackingHandler`. - /// Obtained via `handler.get_handler_ref()` before the handler is moved into the - /// service. - #[derive(Clone)] - struct HandlerRef { - tx: UnboundedSender, - cancelled: Arc, - } - - impl HandlerRef { - /// Push the next state to be emitted by the handler's stream. - /// Panics if the stream receiver has been dropped. - fn shift_state(&self, state: T) { - self.tx.send(state).unwrap(); - } - - fn was_cancelled(&self) -> bool { - self.cancelled.load(Ordering::SeqCst) - } - } - - struct CancellationTrackingHandler { - tx: UnboundedSender, - rx: std::sync::Mutex>>, - cancelled: Arc, - } - - impl std::fmt::Debug for CancellationTrackingHandler { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("CancellationTrackingHandler") - .field("cancelled", &self.cancelled.load(Ordering::SeqCst)) - .finish_non_exhaustive() - } - } - - impl CancellationTrackingHandler { - fn new() -> Self { - let (tx, rx) = unbounded_channel(); - Self { - tx, - rx: std::sync::Mutex::new(Some(rx)), - cancelled: Arc::new(AtomicBool::new(false)), - } - } - - /// Return a `HandlerRef` that can be kept by the test after this handler is - /// moved into the service. - fn get_handler_ref(&self) -> HandlerRef { - HandlerRef { - tx: self.tx.clone(), - cancelled: self.cancelled.clone(), - } - } - } - - /// Shared stream body for all three transport trait impls. - /// - /// Uses a `biased` `select!` with the cancellation branch first so that - /// cancellation always wins over a simultaneously-ready channel item. This - /// guarantees that no queued state is emitted once the token is cancelled, - /// making the "no emission after cancel" assertions in tests deterministic. - fn run_tracking_stream( - rx: Option>, - cancellation: CancellationToken, - cancelled: Arc, - wrap: impl Fn(T) -> E + Send + 'static, - ) -> impl Stream + Send + Unpin + 'static - where - T: Send + 'static, - E: Send + 'static, - { - Box::pin(async_stream::stream! { - let Some(mut rx) = rx else { return; }; - // This allows to simulate when the handler detected cancellation, - // but still emit a single event after cancellation to simulate a - // race. - let mut cancel_detected = false; - loop { - tokio::select! { - biased; - _ = cancellation.cancelled(), if !cancel_detected => { - cancel_detected = true; - cancelled.store(true, Ordering::SeqCst); - } - maybe = rx.recv() => match maybe { - Some(state) => { - yield wrap(state) - if cancel_detected { - break; - } - }, - None => break, // all senders dropped - } - } - } - }) - } - - impl UsbHandler for CancellationTrackingHandler { - fn start( - &self, - _request: &CredentialRequest, - cancellation: CancellationToken, - ) -> impl Stream + Send + Sized + Unpin + 'static { - let rx = self.rx.lock().unwrap().take(); - run_tracking_stream(rx, cancellation, self.cancelled.clone(), |state| UsbEvent { - state, - }) - } - } - - impl HybridHandler for CancellationTrackingHandler { - fn start( - &self, - _request: &CredentialRequest, - cancellation: CancellationToken, - ) -> impl Stream + Unpin + Send + Sized + 'static { - let rx = self.rx.lock().unwrap().take(); - run_tracking_stream(rx, cancellation, self.cancelled.clone(), |state| { - HybridEvent { state } - }) - } - } - - impl NfcHandler for CancellationTrackingHandler { - fn start( - &self, - _request: &CredentialRequest, - cancellation: CancellationToken, - ) -> impl Stream + Send + Sized + Unpin + 'static { - let rx = self.rx.lock().unwrap().take(); - run_tracking_stream(rx, cancellation, self.cancelled.clone(), |state| NfcEvent { - state, - }) - } - } - #[tokio::test] async fn test_cancel_request_by_id() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -863,12 +687,10 @@ mod tests { #[tokio::test] async fn test_multiple_handlers_all_cancelled() { - let usb_handler = CancellationTrackingHandler::::new(); - let hybrid_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); - let hybrid_ref = hybrid_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); + let (hybrid_handler, hybrid_ref) = ScriptedTransport::::new(); - let service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + let service = CredentialService::new(hybrid_handler, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -878,8 +700,8 @@ mod tests { let mut hybrid_stream = service.get_hybrid_credential().await; // Push and consume one state from each to confirm streams are live - usb_ref.shift_state(UsbStateInternal::Waiting); - hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + usb_ref.emit(UsbStateInternal::Waiting); + hybrid_ref.emit(HybridStateInternal::Init("qr".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!( hybrid_stream.next().await, @@ -888,9 +710,9 @@ mod tests { // Queue additional states that should never be emitted after cancellation. // These sit in the channel when cancel_request() fires. - usb_ref.shift_state(UsbStateInternal::Waiting); - usb_ref.shift_state(UsbStateInternal::Waiting); - hybrid_ref.shift_state(HybridStateInternal::Connecting); + usb_ref.emit(UsbStateInternal::Waiting); + usb_ref.emit(UsbStateInternal::Waiting); + hybrid_ref.emit(HybridStateInternal::Connecting); // Cancel the request — token is now cancelled synchronously service.cancel_request(request_id).await; @@ -913,7 +735,7 @@ mod tests { #[tokio::test] async fn test_cancellation_cleans_up_request_context() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -933,7 +755,7 @@ mod tests { #[tokio::test] async fn test_cancel_with_unknown_id_is_noop() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -959,7 +781,7 @@ mod tests { #[tokio::test] async fn test_cancel_with_no_active_request_is_noop() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); // Cancel when no request is active (should not crash or panic) service.cancel_request(12345).await; @@ -976,7 +798,7 @@ mod tests { #[tokio::test] async fn test_request_id_matches_on_init() { - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -998,15 +820,13 @@ mod tests { #[tokio::test] async fn test_explicit_cancel_stops_all_transports() { - let usb_handler = CancellationTrackingHandler::::new(); - let hybrid_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); - let hybrid_ref = hybrid_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); + let (hybrid_handler, hybrid_ref) = ScriptedTransport::::new(); assert!(!usb_ref.was_cancelled()); assert!(!hybrid_ref.was_cancelled()); - let service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + let service = CredentialService::new(hybrid_handler, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -1016,8 +836,8 @@ mod tests { let mut hybrid_stream = service.get_hybrid_credential().await; // Push and consume one state each to confirm streams are live - usb_ref.shift_state(UsbStateInternal::Waiting); - hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + usb_ref.emit(UsbStateInternal::Waiting); + hybrid_ref.emit(HybridStateInternal::Init("qr".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!( hybrid_stream.next().await, @@ -1026,9 +846,9 @@ mod tests { // Queue additional states that should never be emitted after cancellation. // These sit in the channel when cancel_request() fires. - usb_ref.shift_state(UsbStateInternal::Waiting); - usb_ref.shift_state(UsbStateInternal::Waiting); - hybrid_ref.shift_state(HybridStateInternal::Connecting); + usb_ref.emit(UsbStateInternal::Waiting); + usb_ref.emit(UsbStateInternal::Waiting); + hybrid_ref.emit(HybridStateInternal::Connecting); // Explicitly cancel — token becomes cancelled synchronously service.cancel_request(request_id).await; @@ -1037,9 +857,7 @@ mod tests { "Cancellation token should be triggered after cancel_request" ); // Add explicit post-cancellation message. - usb_ref.shift_state(UsbStateInternal::Failed(CredentialServiceError::Internal( - "Cancelled".to_string(), - ))); + usb_ref.fail(CredentialServiceError::Internal("Cancelled".to_string())); // biased select! polls cancellation first, discarding the queued states let usb_remaining: Vec<_> = usb_stream.collect().await; @@ -1054,7 +872,7 @@ mod tests { "Hybrid should not emit any more states after cancellation" ); - // Flags are set by run_tracking_stream when it observes the cancelled token + // Flags are set by ScriptedTransport when it observes the cancelled token. assert!( usb_ref.was_cancelled(), "USB handler should have detected cancellation" @@ -1069,10 +887,9 @@ mod tests { async fn test_failed_request_triggers_cancellation() { use credentialsd_common::model::Error; - let usb_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, usb_handler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -1081,12 +898,10 @@ mod tests { let mut usb_stream = service.get_usb_credential().await; assert!(!cancellation_token.is_cancelled()); - usb_ref.shift_state(UsbStateInternal::Waiting); + usb_ref.emit(UsbStateInternal::Waiting); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); - usb_ref.shift_state(UsbStateInternal::Failed(Error::Internal( - "test failure".to_string(), - ))); + usb_ref.fail(Error::Internal("test failure".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Failed(_)))); // UsbStateStream calls complete_request on Failed, which cancels the token @@ -1100,12 +915,10 @@ mod tests { async fn test_failed_request_cancels_other_transports() { use credentialsd_common::model::Error; - let usb_handler = CancellationTrackingHandler::::new(); - let hybrid_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); - let hybrid_ref = hybrid_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); + let (hybrid_handler, hybrid_ref) = ScriptedTransport::::new(); - let service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + let service = CredentialService::new(hybrid_handler, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -1115,7 +928,7 @@ mod tests { let mut hybrid_stream = service.get_hybrid_credential().await; // Confirm hybrid stream is live - hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + hybrid_ref.emit(HybridStateInternal::Init("qr".to_string())); assert!(matches!( hybrid_stream.next().await, Some(HybridState::Init(_)) @@ -1123,13 +936,11 @@ mod tests { // Queue an extra hybrid state that should be discarded once USB fails. // It sits in the channel when complete_request() cancels the token. - hybrid_ref.shift_state(HybridStateInternal::Connecting); + hybrid_ref.emit(HybridStateInternal::Connecting); - // USB fails — UsbStateStream calls complete_request → token cancelled - usb_ref.shift_state(UsbStateInternal::Waiting); - usb_ref.shift_state(UsbStateInternal::Failed(Error::Internal( - "test".to_string(), - ))); + // USB fails — CredentialStateStream calls complete_request → token cancelled + usb_ref.emit(UsbStateInternal::Waiting); + usb_ref.fail(Error::Internal("test".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!(usb_stream.next().await, Some(UsbState::Failed(_)))); @@ -1154,10 +965,9 @@ mod tests { async fn test_completed_request_triggers_cancellation() { let credential_response = create_test_credential_response(); - let usb_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); - let service = CredentialService::new(MockHybridHandler, MockNfcHandler, usb_handler); + let service = CredentialService::new(EmptyTransport, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -1166,8 +976,8 @@ mod tests { let mut usb_stream = service.get_usb_credential().await; assert!(!cancellation_token.is_cancelled()); - usb_ref.shift_state(UsbStateInternal::Waiting); - usb_ref.shift_state(UsbStateInternal::Completed(credential_response)); + usb_ref.emit(UsbStateInternal::Waiting); + usb_ref.complete(credential_response); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!(usb_stream.next().await, Some(UsbState::Completed))); @@ -1182,12 +992,10 @@ mod tests { async fn test_completed_request_cancels_other_transports() { let credential_response = create_test_credential_response(); - let usb_handler = CancellationTrackingHandler::::new(); - let hybrid_handler = CancellationTrackingHandler::::new(); - let usb_ref = usb_handler.get_handler_ref(); - let hybrid_ref = hybrid_handler.get_handler_ref(); + let (usb_handler, usb_ref) = ScriptedTransport::::new(); + let (hybrid_handler, hybrid_ref) = ScriptedTransport::::new(); - let service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + let service = CredentialService::new(hybrid_handler, EmptyTransport, usb_handler); let request = create_test_request().await; let (tx, _rx) = oneshot::channel(); @@ -1197,7 +1005,7 @@ mod tests { let mut hybrid_stream = service.get_hybrid_credential().await; // Confirm hybrid stream is live - hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + hybrid_ref.emit(HybridStateInternal::Init("qr".to_string())); assert!(matches!( hybrid_stream.next().await, Some(HybridState::Init(_)) @@ -1205,11 +1013,11 @@ mod tests { // Queue an extra hybrid state that should be discarded once USB completes. // It sits in the channel when complete_request() cancels the token. - hybrid_ref.shift_state(HybridStateInternal::Connecting); + hybrid_ref.emit(HybridStateInternal::Connecting); - // USB completes — UsbStateStream calls complete_request → token cancelled - usb_ref.shift_state(UsbStateInternal::Waiting); - usb_ref.shift_state(UsbStateInternal::Completed(credential_response)); + // USB completes — CredentialStateStream calls complete_request → token cancelled + usb_ref.emit(UsbStateInternal::Waiting); + usb_ref.complete(credential_response); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!(usb_stream.next().await, Some(UsbState::Completed))); diff --git a/credentialsd/src/credential_service/test_support.rs b/credentialsd/src/credential_service/test_support.rs new file mode 100644 index 0000000..3d6a980 --- /dev/null +++ b/credentialsd/src/credential_service/test_support.rs @@ -0,0 +1,212 @@ +//! Scripted credential transports used by unit tests. + +use std::sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, +}; + +use futures_lite::Stream; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; +use tokio_util::sync::CancellationToken; + +use super::{ + CredentialRequest, CredentialResponse, CredentialServiceError, + hybrid::{HybridEvent, HybridHandler, HybridStateInternal}, + nfc::{NfcEvent, NfcHandler, NfcStateInternal}, + usb::{UsbEvent, UsbHandler, UsbStateInternal}, +}; + +/// A transport that never emits an event. +#[derive(Clone, Copy, Debug, Default)] +pub(super) struct EmptyTransport; + +impl UsbHandler for EmptyTransport { + fn start( + &self, + _request: &CredentialRequest, + _cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + futures::stream::empty() + } +} + +impl HybridHandler for EmptyTransport { + fn start( + &self, + _request: &CredentialRequest, + _cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + futures::stream::empty() + } +} + +impl NfcHandler for EmptyTransport { + fn start( + &self, + _request: &CredentialRequest, + _cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + futures::stream::empty() + } +} + +/// Test-side control handle for a [`ScriptedTransport`]. +#[derive(Clone)] +pub(super) struct ScriptedTransportController { + tx: UnboundedSender, + cancelled: Arc, +} + +impl ScriptedTransportController { + /// Emit the next internal transport state. + pub(super) fn emit(&self, state: T) { + assert!( + self.tx.send(state).is_ok(), + "scripted transport stream has already stopped" + ); + } + + /// Whether the scripted transport observed its cancellation token. + pub(super) fn was_cancelled(&self) -> bool { + self.cancelled.load(Ordering::SeqCst) + } +} + +pub(super) trait ScriptedState { + fn completed(response: CredentialResponse) -> Self; + fn failed(error: CredentialServiceError) -> Self; +} + +impl ScriptedState for UsbStateInternal { + fn completed(response: CredentialResponse) -> Self { + Self::Completed(response) + } + + fn failed(error: CredentialServiceError) -> Self { + Self::Failed(error) + } +} + +impl ScriptedState for HybridStateInternal { + fn completed(response: CredentialResponse) -> Self { + Self::Completed(response) + } + + fn failed(error: CredentialServiceError) -> Self { + Self::Failed(error) + } +} + +impl ScriptedState for NfcStateInternal { + fn completed(response: CredentialResponse) -> Self { + Self::Completed(response) + } + + fn failed(error: CredentialServiceError) -> Self { + Self::Failed(error) + } +} + +impl ScriptedTransportController { + /// Complete the active credential request successfully. + pub(super) fn complete(&self, response: CredentialResponse) { + self.emit(T::completed(response)); + } + + /// Fail the active credential request. + pub(super) fn fail(&self, error: CredentialServiceError) { + self.emit(T::failed(error)); + } +} + +/// A push-based mock transport with deterministic cancellation tracking. +pub(super) struct ScriptedTransport { + rx: Mutex>>, + cancelled: Arc, +} + +impl std::fmt::Debug for ScriptedTransport { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ScriptedTransport") + .field("cancelled", &self.cancelled.load(Ordering::SeqCst)) + .finish_non_exhaustive() + } +} + +impl ScriptedTransport { + pub(super) fn new() -> (Self, ScriptedTransportController) { + let (tx, rx) = unbounded_channel(); + let cancelled = Arc::new(AtomicBool::new(false)); + ( + Self { + rx: Mutex::new(Some(rx)), + cancelled: cancelled.clone(), + }, + ScriptedTransportController { tx, cancelled }, + ) + } + + /// Build a stream that prioritizes cancellation over queued events. + fn start_with( + &self, + cancellation: CancellationToken, + wrap: impl Fn(T) -> E + Send + 'static, + ) -> impl Stream + Send + Unpin + 'static + where + T: Send + 'static, + E: Send + 'static, + { + let mut rx = self + .rx + .lock() + .unwrap() + .take() + .expect("ScriptedTransport can only be started once"); + let cancelled = self.cancelled.clone(); + Box::pin(async_stream::stream! { + loop { + tokio::select! { + biased; + _ = cancellation.cancelled() => { + cancelled.store(true, Ordering::SeqCst); + break; + } + maybe = rx.recv() => match maybe { + Some(state) => yield wrap(state), + None => break, + } + } + } + }) + } +} + +impl UsbHandler for ScriptedTransport { + fn start( + &self, + _request: &CredentialRequest, + cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + self.start_with(cancellation, |state| UsbEvent { state }) + } +} + +impl HybridHandler for ScriptedTransport { + fn start( + &self, + _request: &CredentialRequest, + cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + self.start_with(cancellation, |state| HybridEvent { state }) + } +} + +impl NfcHandler for ScriptedTransport { + fn start( + &self, + _request: &CredentialRequest, + cancellation: CancellationToken, + ) -> impl Stream + Send + Sized + Unpin + 'static { + self.start_with(cancellation, |state| NfcEvent { state }) + } +} From 775f07f9fd1ff6a23b553ab69a97d3043523821a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 18:48:15 +0200 Subject: [PATCH 2/8] daemon: Share credential lifecycle across transports USB, hybrid, and NFC duplicated state conversion and request completion in separate stream wrappers, making their terminal behavior easy to desynchronize. Route typed transport events through one lifecycle stream and cover success and failure for every backend. --- credentialsd/src/credential_service/mod.rs | 322 +++++++++++++-------- credentialsd/src/credential_service/nfc.rs | 6 +- 2 files changed, 200 insertions(+), 128 deletions(-) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index f975d16..592578b 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -9,11 +9,10 @@ use std::{ fmt::Debug, pin::Pin, sync::{Arc, Mutex, OnceLock}, - task::Poll, }; use async_trait::async_trait; -use futures_lite::{FutureExt, Stream, StreamExt}; +use futures_lite::{Stream, StreamExt}; use libwebauthn::{ self, ops::webauthn::{GetAssertionResponse, MakeCredentialResponse}, @@ -142,11 +141,7 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - Box::pin(HybridStateStream { - inner: stream, - ctx, - cancellation_token: cancellation.clone(), - }) + credential_state_stream(stream, ctx, cancellation.clone()) } else { tracing::error!( "Attempted to start hybrid credential flow, but no request context was found." @@ -169,11 +164,7 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - Box::pin(UsbStateStream { - inner: stream, - ctx, - cancellation_token: cancellation.clone(), - }) + credential_state_stream(stream, ctx, cancellation.clone()) } else { tracing::error!( "Attempted to start usb credential flow, but no request context was found." @@ -196,11 +187,7 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - Box::pin(NfcStateStream { - inner: stream, - ctx, - cancellation_token: cancellation.clone(), - }) + credential_state_stream(stream, ctx, cancellation.clone()) } else { tracing::error!( "Attempted to start nfc credential flow, but no request context was found." @@ -323,126 +310,103 @@ impl Manage } } -pub struct HybridStateStream { - inner: H, - ctx: Arc>>, - cancellation_token: CancellationToken, +/// An event emitted by a credential transport. +/// +/// Transport implementations keep their privileged internal states private and +/// use this trait to expose the corresponding public state. Terminal events +/// additionally carry the result used to complete the active request. +trait TransportEvent { + type PublicState; + + fn into_state_and_result( + self, + ) -> ( + Self::PublicState, + Option>, + ); } -impl Stream for HybridStateStream -where - H: Stream + Unpin + Sized, -{ - type Item = HybridState; - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> Poll> { - let ctx = &self.ctx.clone(); - let cancellation_token = self.cancellation_token.clone(); - match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(HybridEvent { state })) => { - if cancellation_token.is_cancelled() { - return Poll::Ready(None); - } - match &state { - HybridStateInternal::Completed(response) => { - complete_request(ctx, Ok(response.clone())); - } - HybridStateInternal::Failed(err) => { - complete_request(ctx, Err(err.clone())); - } - _ => {} - } - Poll::Ready(Some(state.into())) - } - Poll::Ready(None) => Poll::Ready(None), - } +impl TransportEvent for HybridEvent { + type PublicState = HybridState; + + fn into_state_and_result( + self, + ) -> ( + Self::PublicState, + Option>, + ) { + let result = match &self.state { + HybridStateInternal::Completed(response) => Some(Ok(response.clone())), + HybridStateInternal::Failed(error) => Some(Err(error.clone())), + _ => None, + }; + (self.state.into(), result) } } -struct UsbStateStream { - inner: H, - ctx: Arc>>, - cancellation_token: CancellationToken, +impl TransportEvent for UsbEvent { + type PublicState = UsbState; + + fn into_state_and_result( + self, + ) -> ( + Self::PublicState, + Option>, + ) { + let result = match &self.state { + UsbStateInternal::Completed(response) => Some(Ok(response.clone())), + UsbStateInternal::Failed(error) => Some(Err(error.clone())), + _ => None, + }; + (self.state.into(), result) + } } -impl Stream for UsbStateStream -where - H: Stream + Unpin + Sized, -{ - type Item = UsbState; - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> Poll> { - let ctx = &self.ctx.clone(); - let cancellation_token = self.cancellation_token.clone(); - match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(UsbEvent { state })) => { - if cancellation_token.is_cancelled() { - return Poll::Ready(None); - } - match &state { - UsbStateInternal::Completed(response) => { - complete_request(ctx, Ok(response.clone())); - } - UsbStateInternal::Failed(error) => { - complete_request(ctx, Err(error.clone())); - } - _ => {} - } - Poll::Ready(Some(state.into())) - } - Poll::Ready(None) => Poll::Ready(None), - } +impl TransportEvent for NfcEvent { + type PublicState = NfcState; + + fn into_state_and_result( + self, + ) -> ( + Self::PublicState, + Option>, + ) { + let result = match &self.state { + NfcStateInternal::Completed(response) => Some(Ok(response.clone())), + NfcStateInternal::Failed(error) => Some(Err(error.clone())), + _ => None, + }; + (self.state.into(), result) } } -#[expect(unused)] -struct NfcStateStream { - inner: H, +/// Applies request lifecycle handling to a transport's stream of events. +fn credential_state_stream( + mut inner: S, ctx: Arc>>, cancellation_token: CancellationToken, -} - -impl Stream for NfcStateStream +) -> Pin + Send + 'static>> where - H: Stream + Unpin + Sized, + S: Stream + Unpin + Send + 'static, + E: TransportEvent + Send + 'static, + E::PublicState: Send + 'static, { - type Item = NfcState; - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> Poll> { - let ctx = &self.ctx.clone(); - let cancellation_token = self.cancellation_token.clone(); - match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(NfcEvent { state })) => { - if cancellation_token.is_cancelled() { - return Poll::Ready(None); - } - - match &state { - NfcStateInternal::Completed(response) => { - complete_request(ctx, Ok(response.clone())); - } - NfcStateInternal::Failed(error) => { - complete_request(ctx, Err(error.clone())); - } - _ => {} - } - Poll::Ready(Some(state.into())) + Box::pin(async_stream::stream! { + while let Some(event) = inner.next().await { + if cancellation_token.is_cancelled() { + break; + } + + let (state, result) = event.into_state_and_result(); + if let Some(result) = result { + complete_request(&ctx, result); + yield state; + break; } - Poll::Ready(None) => Poll::Ready(None), + + yield state; } - } + }) } pub enum DeviceStateUpdate { @@ -891,7 +855,7 @@ mod tests { let service = CredentialService::new(EmptyTransport, EmptyTransport, usb_handler); let request = create_test_request().await; - let (tx, _rx) = oneshot::channel(); + let (tx, rx) = oneshot::channel(); let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); @@ -904,13 +868,119 @@ mod tests { usb_ref.fail(Error::Internal("test failure".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Failed(_)))); - // UsbStateStream calls complete_request on Failed, which cancels the token + let result = rx.await.expect("request result should be sent"); + assert!(matches!(result, Err(Error::Internal(message)) if message == "test failure")); + + // CredentialStateStream calls complete_request on Failed, which cancels the token assert!( cancellation_token.is_cancelled(), "Cancellation token should be triggered when request fails" ); } + #[tokio::test] + async fn test_scripted_nfc_transport_uses_generic_lifecycle_stream() { + use credentialsd_common::model::Error; + + let (nfc_handler, nfc_controller) = ScriptedTransport::::new(); + let service = CredentialService::new(EmptyTransport, nfc_handler, EmptyTransport); + let request = create_test_request().await; + let (tx, rx) = oneshot::channel(); + + let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + let mut nfc_stream = service._get_nfc_credential().await; + + nfc_controller.emit(NfcStateInternal::Waiting); + assert!(matches!(nfc_stream.next().await, Some(NfcState::Waiting))); + + nfc_controller.fail(Error::Internal("mock NFC failure".to_string())); + assert!(matches!(nfc_stream.next().await, Some(NfcState::Failed(_)))); + assert!(cancellation_token.is_cancelled()); + + let result = rx.await.expect("request result should be sent"); + assert!(matches!(result, Err(Error::Internal(message)) if message == "mock NFC failure")); + } + + #[tokio::test] + async fn test_scripted_nfc_transport_propagates_successful_response() { + let (nfc_handler, nfc_controller) = ScriptedTransport::::new(); + let service = CredentialService::new(EmptyTransport, nfc_handler, EmptyTransport); + let request = create_test_request().await; + let (tx, rx) = oneshot::channel(); + + let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + let mut nfc_stream = service._get_nfc_credential().await; + + nfc_controller.emit(NfcStateInternal::Waiting); + assert!(matches!(nfc_stream.next().await, Some(NfcState::Waiting))); + + nfc_controller.complete(create_test_credential_response()); + assert!(matches!(nfc_stream.next().await, Some(NfcState::Completed))); + assert!(cancellation_token.is_cancelled()); + + let result = rx.await.expect("request result should be sent"); + let Ok(CredentialResponse::GetPublicKeyCredentialResponse(response)) = result else { + panic!("NFC completion should propagate the credential response"); + }; + assert_eq!(response.attachment_modality, "cross-platform"); + } + + #[tokio::test] + async fn test_scripted_hybrid_transport_propagates_failure() { + use credentialsd_common::model::Error; + + let (hybrid_handler, hybrid_controller) = ScriptedTransport::::new(); + let service = CredentialService::new(hybrid_handler, EmptyTransport, EmptyTransport); + let request = create_test_request().await; + let (tx, rx) = oneshot::channel(); + + let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + let mut hybrid_stream = service.get_hybrid_credential().await; + + hybrid_controller.emit(HybridStateInternal::Connecting); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Connecting) + )); + + hybrid_controller.fail(Error::NoCredentials); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Failed(Error::NoCredentials)) + )); + assert!(cancellation_token.is_cancelled()); + + let result = rx.await.expect("request result should be sent"); + assert!(matches!(result, Err(Error::NoCredentials))); + } + + #[tokio::test] + async fn test_lifecycle_stream_discards_event_after_cancellation() { + let request = create_test_request().await; + let (response_channel, _response_rx) = oneshot::channel(); + let cancellation = CancellationToken::new(); + cancellation.cancel(); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id: 1, + cancellation: cancellation.clone(), + }))); + let mut stream = credential_state_stream( + futures::stream::iter([UsbEvent { + state: UsbStateInternal::Waiting, + }]), + ctx.clone(), + cancellation, + ); + + assert!(stream.next().await.is_none()); + assert!( + ctx.lock().unwrap().is_some(), + "discarding a stale event must not complete the active request" + ); + } + #[tokio::test] async fn test_failed_request_cancels_other_transports() { use credentialsd_common::model::Error; @@ -969,7 +1039,7 @@ mod tests { let service = CredentialService::new(EmptyTransport, EmptyTransport, usb_handler); let request = create_test_request().await; - let (tx, _rx) = oneshot::channel(); + let (tx, rx) = oneshot::channel(); let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); @@ -981,7 +1051,9 @@ mod tests { assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); assert!(matches!(usb_stream.next().await, Some(UsbState::Completed))); - // UsbStateStream calls complete_request on Completed, which cancels the token + assert!(rx.await.expect("request result should be sent").is_ok()); + + // CredentialStateStream calls complete_request on Completed, which cancels the token assert!( cancellation_token.is_cancelled(), "Cancellation token should be triggered when request completes successfully" diff --git a/credentialsd/src/credential_service/nfc.rs b/credentialsd/src/credential_service/nfc.rs index bd30d78..a0ddd60 100644 --- a/credentialsd/src/credential_service/nfc.rs +++ b/credentialsd/src/credential_service/nfc.rs @@ -23,7 +23,7 @@ use crate::model::{CredentialRequest, GetAssertionResponseInternal}; use super::{AuthenticatorResponse, CredentialResponse}; pub(crate) trait NfcHandler { - #[expect(unused)] + #[cfg_attr(not(test), expect(unused))] fn start( &self, request: &CredentialRequest, @@ -364,13 +364,13 @@ impl NfcHandler for InProcessNfcHandler { // this exists to prevent making NfcStateInternal type public to the whole crate. /// A message between NFC handler and credential service -#[expect(unused)] +#[cfg_attr(not(test), expect(unused))] pub struct NfcEvent { pub(super) state: NfcStateInternal, } /// Used to share internal state between handler and credential service -#[expect(unused)] +#[cfg_attr(not(test), expect(unused))] #[derive(Clone, Debug, Default)] pub(super) enum NfcStateInternal { /// Not polling for FIDO NFC device. From 0cf344c92031a925dcbbecd9b8632879c68c6729 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sat, 5 Sep 2026 15:52:33 +0200 Subject: [PATCH 3/8] docs: Describe shared transport lifecycle handling --- ARCHITECTURE.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f9551e7..d873c25 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -121,6 +121,9 @@ The `CredentialService` mostly just forwards events over to the UI service, minu any details that are not necessary for the UI to know (like the response channels mentioned above, which cannot be serialized over D-Bus anyway). +USB, hybrid, and NFC handler events pass through shared request lifecycle +handling, while each transport keeps its own public states. + Actual interaction I/O is performed using the [libwebauthn][libwebauthn] library. [libwebauthn]: https://github.com/linux-credentials/libwebauthn From 9655d8da9a2d99f1fd3e2f6fa1f8b34a39a85f24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 17:09:10 +0200 Subject: [PATCH 4/8] daemon: Wake credential streams when requests are cancelled A lifecycle stream only noticed cancellation after another transport event, so idle discovery could remain pending. Await transport polling through CancellationToken::run_until_cancelled so cancellation wakes the stream immediately. --- credentialsd/src/credential_service/mod.rs | 48 +++++++++++++++++++++- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 592578b..d1f498d 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -392,7 +392,15 @@ where E::PublicState: Send + 'static, { Box::pin(async_stream::stream! { - while let Some(event) = inner.next().await { + loop { + let Some(event) = cancellation_token + .run_until_cancelled(inner.next()) + .await + .flatten() + else { + break; + }; + if cancellation_token.is_cancelled() { break; } @@ -478,7 +486,7 @@ impl From for AuthenticatorResponse { #[cfg(test)] mod tests { - use std::time::Duration; + use std::{task::Poll, time::Duration}; use super::test_support::{EmptyTransport, ScriptedTransport}; use super::*; @@ -981,6 +989,42 @@ mod tests { ); } + #[tokio::test] + async fn test_lifecycle_stream_wakes_when_cancelled_while_inner_is_pending() { + let request = create_test_request().await; + let (response_channel, _response_rx) = oneshot::channel(); + let cancellation = CancellationToken::new(); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id: 1, + cancellation: cancellation.clone(), + }))); + let (polled_tx, polled_rx) = oneshot::channel(); + let mut polled_tx = Some(polled_tx); + let pending_inner = futures::stream::poll_fn(move |_cx| { + if let Some(polled_tx) = polled_tx.take() { + _ = polled_tx.send(()); + } + Poll::>::Pending + }); + let mut stream = credential_state_stream(pending_inner, ctx.clone(), cancellation.clone()); + + let waiting_task = tokio::spawn(async move { stream.next().await }); + polled_rx.await.expect("inner stream should be polled"); + cancellation.cancel(); + + let event = tokio::time::timeout(Duration::from_secs(1), waiting_task) + .await + .expect("cancellation should wake the lifecycle stream") + .expect("stream task should not panic"); + assert!(event.is_none()); + assert!( + ctx.lock().unwrap().is_some(), + "cancelling a pending stream must not complete the active request" + ); + } + #[tokio::test] async fn test_failed_request_cancels_other_transports() { use credentialsd_common::model::Error; From 214a719be726248eeb56158a9076dbe5d3617182 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 15:36:37 +0200 Subject: [PATCH 5/8] daemon: Bind terminal events to their originating request A late terminal event could consume whichever request context was active and complete a newer request. Attach the request ID and a unique marker to each lifecycle stream, then ignore terminal results whose ownership no longer matches. --- credentialsd/src/credential_service/mod.rs | 152 +++++++++++++++++++-- 1 file changed, 137 insertions(+), 15 deletions(-) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index d1f498d..407c895 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -65,11 +65,15 @@ fn persistent_token_store() -> Arc { .clone() } +#[derive(Debug)] +struct RequestMarker; + #[derive(Debug)] struct RequestContext { request: CredentialRequest, response_channel: oneshot::Sender>, request_id: RequestId, + request_marker: Arc, cancellation: CancellationToken, } @@ -132,6 +136,8 @@ impl if let Some(RequestContext { ref request, ref cancellation, + ref request_marker, + request_id, .. }) = *guard { @@ -141,7 +147,13 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - credential_state_stream(stream, ctx, cancellation.clone()) + credential_state_stream( + stream, + ctx, + request_id, + request_marker.clone(), + cancellation.clone(), + ) } else { tracing::error!( "Attempted to start hybrid credential flow, but no request context was found." @@ -155,6 +167,8 @@ impl if let Some(RequestContext { ref request, ref cancellation, + ref request_marker, + request_id, .. }) = *guard { @@ -164,7 +178,13 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - credential_state_stream(stream, ctx, cancellation.clone()) + credential_state_stream( + stream, + ctx, + request_id, + request_marker.clone(), + cancellation.clone(), + ) } else { tracing::error!( "Attempted to start usb credential flow, but no request context was found." @@ -178,6 +198,8 @@ impl if let Some(RequestContext { ref request, ref cancellation, + ref request_marker, + request_id, .. }) = *guard { @@ -187,7 +209,13 @@ impl .unwrap() .start(request, cancellation.clone()); let ctx = self.ctx.clone(); - credential_state_stream(stream, ctx, cancellation.clone()) + credential_state_stream( + stream, + ctx, + request_id, + request_marker.clone(), + cancellation.clone(), + ) } else { tracing::error!( "Attempted to start nfc credential flow, but no request context was found." @@ -221,6 +249,7 @@ impl Manage request: request.clone(), response_channel: tx, request_id, + request_marker: Arc::new(RequestMarker), cancellation: cancellation.clone(), }; _ = cred_request.insert(ctx); @@ -306,6 +335,7 @@ impl Manage .boxed(); selected_transports.push(hybrid); } + futures::stream::select_all(selected_transports).boxed() } } @@ -384,6 +414,8 @@ impl TransportEvent for NfcEvent { fn credential_state_stream( mut inner: S, ctx: Arc>>, + request_id: RequestId, + request_marker: Arc, cancellation_token: CancellationToken, ) -> Pin + Send + 'static>> where @@ -407,7 +439,9 @@ where let (state, result) = event.into_state_and_result(); if let Some(result) = result { - complete_request(&ctx, result); + if !complete_request(&ctx, request_id, &request_marker, result) { + break; + } yield state; break; } @@ -453,17 +487,25 @@ impl From for DeviceStateUpdate { fn complete_request( ctx: &Mutex>, + request_id: RequestId, + request_marker: &Arc, response: Result, -) { - match ctx.lock().unwrap().take() { - Some(ctx) => { - ctx.cancellation.cancel(); - ctx.send_response(response); - } - _ => { - tracing::error!("Tried to consume context to respond to caller, but none was found.") - } - } +) -> bool { + let request_ctx = ctx.lock().unwrap().take_if(|request_ctx| { + request_ctx.request_id == request_id + && Arc::ptr_eq(&request_ctx.request_marker, request_marker) + }); + let Some(request_ctx) = request_ctx else { + tracing::debug!( + request_id, + "Ignoring terminal event for a request that is no longer active." + ); + return false; + }; + + request_ctx.cancellation.cancel(); + request_ctx.send_response(response); + true } #[derive(Debug, Clone)] @@ -962,16 +1004,48 @@ mod tests { assert!(matches!(result, Err(Error::NoCredentials))); } + #[tokio::test] + async fn test_scripted_hybrid_transport_propagates_successful_response() { + let (hybrid_handler, hybrid_controller) = ScriptedTransport::::new(); + let service = CredentialService::new(hybrid_handler, EmptyTransport, EmptyTransport); + let request = create_test_request().await; + let (tx, rx) = oneshot::channel(); + + let (_request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + let mut hybrid_stream = service.get_hybrid_credential().await; + + hybrid_controller.emit(HybridStateInternal::Connecting); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Connecting) + )); + + hybrid_controller.complete(create_test_credential_response()); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Completed) + )); + assert!(cancellation_token.is_cancelled()); + + let result = rx.await.expect("request result should be sent"); + let Ok(CredentialResponse::GetPublicKeyCredentialResponse(response)) = result else { + panic!("Hybrid completion should propagate the credential response"); + }; + assert_eq!(response.attachment_modality, "cross-platform"); + } + #[tokio::test] async fn test_lifecycle_stream_discards_event_after_cancellation() { let request = create_test_request().await; let (response_channel, _response_rx) = oneshot::channel(); let cancellation = CancellationToken::new(); + let request_marker = Arc::new(RequestMarker); cancellation.cancel(); let ctx = Arc::new(Mutex::new(Some(RequestContext { request, response_channel, request_id: 1, + request_marker: request_marker.clone(), cancellation: cancellation.clone(), }))); let mut stream = credential_state_stream( @@ -979,6 +1053,8 @@ mod tests { state: UsbStateInternal::Waiting, }]), ctx.clone(), + 1, + request_marker, cancellation, ); @@ -994,10 +1070,12 @@ mod tests { let request = create_test_request().await; let (response_channel, _response_rx) = oneshot::channel(); let cancellation = CancellationToken::new(); + let request_marker = Arc::new(RequestMarker); let ctx = Arc::new(Mutex::new(Some(RequestContext { request, response_channel, request_id: 1, + request_marker: request_marker.clone(), cancellation: cancellation.clone(), }))); let (polled_tx, polled_rx) = oneshot::channel(); @@ -1008,7 +1086,13 @@ mod tests { } Poll::>::Pending }); - let mut stream = credential_state_stream(pending_inner, ctx.clone(), cancellation.clone()); + let mut stream = credential_state_stream( + pending_inner, + ctx.clone(), + 1, + request_marker, + cancellation.clone(), + ); let waiting_task = tokio::spawn(async move { stream.next().await }); polled_rx.await.expect("inner stream should be polled"); @@ -1025,6 +1109,44 @@ mod tests { ); } + #[tokio::test] + async fn test_lifecycle_stream_cannot_complete_a_newer_request() { + let request = create_test_request().await; + let (response_channel, mut response_rx) = oneshot::channel(); + let current_cancellation = CancellationToken::new(); + let current_request_marker = Arc::new(RequestMarker); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id: 1, + request_marker: current_request_marker, + cancellation: current_cancellation.clone(), + }))); + let stale_cancellation = CancellationToken::new(); + let stale_request_marker = Arc::new(RequestMarker); + let mut stale_stream = credential_state_stream( + futures::stream::iter([UsbEvent { + state: UsbStateInternal::Completed(create_test_credential_response()), + }]), + ctx.clone(), + 1, + stale_request_marker, + stale_cancellation.clone(), + ); + + assert!(stale_stream.next().await.is_none()); + assert!(!stale_cancellation.is_cancelled()); + assert!(!current_cancellation.is_cancelled()); + assert_eq!( + ctx.lock().unwrap().as_ref().map(|ctx| ctx.request_id), + Some(1) + ); + assert!(matches!( + response_rx.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + } + #[tokio::test] async fn test_failed_request_cancels_other_transports() { use credentialsd_common::model::Error; From c04af6b820fe70b1d7235f8593e67ab9d76f61bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 18:52:29 +0200 Subject: [PATCH 6/8] daemon: Carry one lifecycle snapshot across transports Discovery let each transport read the active request independently, so a replacement between reads could mix ownership and cancellation data. Snapshot the request ID, marker, and token once and pass that typed lifecycle to every selected transport. --- credentialsd/src/credential_service/mod.rs | 217 +++++++++++---------- 1 file changed, 111 insertions(+), 106 deletions(-) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 407c895..613e34e 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -68,6 +68,13 @@ fn persistent_token_store() -> Arc { #[derive(Debug)] struct RequestMarker; +#[derive(Clone, Debug)] +struct RequestLifecycle { + request_id: RequestId, + request_marker: Arc, + cancellation: CancellationToken, +} + #[derive(Debug)] struct RequestContext { request: CredentialRequest, @@ -78,6 +85,14 @@ struct RequestContext { } impl RequestContext { + fn lifecycle(&self) -> RequestLifecycle { + RequestLifecycle { + request_id: self.request_id, + request_marker: self.request_marker.clone(), + cancellation: self.cancellation.clone(), + } + } + fn send_response(self, response: Result) { if self.response_channel.send(response).is_err() { tracing::error!( @@ -129,31 +144,60 @@ impl impl CredentialService { + fn current_request(&self) -> Option<(CredentialRequest, RequestLifecycle)> { + self.ctx + .lock() + .unwrap() + .as_ref() + .map(|ctx| (ctx.request.clone(), ctx.lifecycle())) + } + + fn hybrid_credential_stream( + &self, + request: &CredentialRequest, + lifecycle: RequestLifecycle, + ) -> Pin + Send + 'static>> { + let stream = self + .hybrid_handler + .lock() + .unwrap() + .start(request, lifecycle.cancellation.clone()); + credential_state_stream(stream, self.ctx.clone(), lifecycle) + } + + fn usb_credential_stream( + &self, + request: &CredentialRequest, + lifecycle: RequestLifecycle, + ) -> Pin + Send + 'static>> { + let stream = self + .usb_handler + .lock() + .unwrap() + .start(request, lifecycle.cancellation.clone()); + credential_state_stream(stream, self.ctx.clone(), lifecycle) + } + + #[cfg_attr(not(test), expect(dead_code))] + fn nfc_credential_stream( + &self, + request: &CredentialRequest, + lifecycle: RequestLifecycle, + ) -> Pin + Send + 'static>> { + let stream = self + ._nfc_handler + .lock() + .unwrap() + .start(request, lifecycle.cancellation.clone()); + credential_state_stream(stream, self.ctx.clone(), lifecycle) + } + + #[cfg(test)] async fn get_hybrid_credential( &self, ) -> Pin + Send + 'static>> { - let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { - ref request, - ref cancellation, - ref request_marker, - request_id, - .. - }) = *guard - { - let stream = self - .hybrid_handler - .lock() - .unwrap() - .start(request, cancellation.clone()); - let ctx = self.ctx.clone(); - credential_state_stream( - stream, - ctx, - request_id, - request_marker.clone(), - cancellation.clone(), - ) + if let Some((request, lifecycle)) = self.current_request() { + self.hybrid_credential_stream(&request, lifecycle) } else { tracing::error!( "Attempted to start hybrid credential flow, but no request context was found." @@ -162,29 +206,10 @@ impl } } + #[cfg(test)] async fn get_usb_credential(&self) -> Pin + Send + 'static>> { - let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { - ref request, - ref cancellation, - ref request_marker, - request_id, - .. - }) = *guard - { - let stream = self - .usb_handler - .lock() - .unwrap() - .start(request, cancellation.clone()); - let ctx = self.ctx.clone(); - credential_state_stream( - stream, - ctx, - request_id, - request_marker.clone(), - cancellation.clone(), - ) + if let Some((request, lifecycle)) = self.current_request() { + self.usb_credential_stream(&request, lifecycle) } else { tracing::error!( "Attempted to start usb credential flow, but no request context was found." @@ -193,29 +218,10 @@ impl } } + #[cfg(test)] async fn _get_nfc_credential(&self) -> Pin + Send + 'static>> { - let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { - ref request, - ref cancellation, - ref request_marker, - request_id, - .. - }) = *guard - { - let stream = self - ._nfc_handler - .lock() - .unwrap() - .start(request, cancellation.clone()); - let ctx = self.ctx.clone(); - credential_state_stream( - stream, - ctx, - request_id, - request_marker.clone(), - cancellation.clone(), - ) + if let Some((request, lifecycle)) = self.current_request() { + self.nfc_credential_stream(&request, lifecycle) } else { tracing::error!( "Attempted to start nfc credential flow, but no request context was found." @@ -223,6 +229,7 @@ impl todo!("Handle error when context is not set up.") } } + } #[async_trait] @@ -301,11 +308,16 @@ impl Manage &self, ) -> Pin + Send + 'static>> { let available_transports = available_transports().await; + let Some((request, lifecycle)) = self.current_request() else { + tracing::error!( + "Attempted to start credential discovery, but no request context was found." + ); + return futures::stream::empty().boxed(); + }; let mut selected_transports = Vec::new(); if available_transports.contains(&libwebauthn::Transport::Usb) { let usb = self - .get_usb_credential() - .await + .usb_credential_stream(&request, lifecycle.clone()) .map(DeviceStateUpdate::from) .boxed(); selected_transports.push(usb); @@ -320,8 +332,7 @@ impl Manage or verification message is sent. if available_transports.contains(&libwebauthn::Transport::Nfc) { let nfc = self - .get_nfc_credential() - .await + .nfc_credential_stream(&request, lifecycle.clone()) .map(DeviceStateUpdate::from) .boxed(); selected_transports.push(nfc); @@ -329,8 +340,7 @@ impl Manage */ if available_transports.contains(&libwebauthn::Transport::Hybrid) { let hybrid = self - .get_hybrid_credential() - .await + .hybrid_credential_stream(&request, lifecycle.clone()) .map(DeviceStateUpdate::from) .boxed(); selected_transports.push(hybrid); @@ -414,9 +424,7 @@ impl TransportEvent for NfcEvent { fn credential_state_stream( mut inner: S, ctx: Arc>>, - request_id: RequestId, - request_marker: Arc, - cancellation_token: CancellationToken, + lifecycle: RequestLifecycle, ) -> Pin + Send + 'static>> where S: Stream + Unpin + Send + 'static, @@ -425,7 +433,7 @@ where { Box::pin(async_stream::stream! { loop { - let Some(event) = cancellation_token + let Some(event) = lifecycle.cancellation .run_until_cancelled(inner.next()) .await .flatten() @@ -433,13 +441,13 @@ where break; }; - if cancellation_token.is_cancelled() { + if lifecycle.cancellation.is_cancelled() { break; } let (state, result) = event.into_state_and_result(); if let Some(result) = result { - if !complete_request(&ctx, request_id, &request_marker, result) { + if !complete_request(&ctx, &lifecycle, result) { break; } yield state; @@ -487,17 +495,16 @@ impl From for DeviceStateUpdate { fn complete_request( ctx: &Mutex>, - request_id: RequestId, - request_marker: &Arc, + lifecycle: &RequestLifecycle, response: Result, ) -> bool { let request_ctx = ctx.lock().unwrap().take_if(|request_ctx| { - request_ctx.request_id == request_id - && Arc::ptr_eq(&request_ctx.request_marker, request_marker) + request_ctx.request_id == lifecycle.request_id + && Arc::ptr_eq(&request_ctx.request_marker, &lifecycle.request_marker) }); let Some(request_ctx) = request_ctx else { tracing::debug!( - request_id, + request_id = lifecycle.request_id, "Ignoring terminal event for a request that is no longer active." ); return false; @@ -713,7 +720,6 @@ mod tests { let mut usb_stream = service.get_usb_credential().await; let mut hybrid_stream = service.get_hybrid_credential().await; - // Push and consume one state from each to confirm streams are live usb_ref.emit(UsbStateInternal::Waiting); hybrid_ref.emit(HybridStateInternal::Init("qr".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); @@ -722,18 +728,13 @@ mod tests { Some(HybridState::Init(_)) )); - // Queue additional states that should never be emitted after cancellation. - // These sit in the channel when cancel_request() fires. usb_ref.emit(UsbStateInternal::Waiting); usb_ref.emit(UsbStateInternal::Waiting); hybrid_ref.emit(HybridStateInternal::Connecting); - // Cancel the request — token is now cancelled synchronously service.cancel_request(request_id).await; assert!(cancellation_token.is_cancelled()); - // biased select! polls cancellation first each iteration, discarding - // the queued states before they can be emitted. let usb_remaining: Vec<_> = usb_stream.collect().await; let hybrid_remaining: Vec<_> = hybrid_stream.collect().await; @@ -921,7 +922,7 @@ mod tests { let result = rx.await.expect("request result should be sent"); assert!(matches!(result, Err(Error::Internal(message)) if message == "test failure")); - // CredentialStateStream calls complete_request on Failed, which cancels the token + // The lifecycle stream completes a failed request and cancels its token. assert!( cancellation_token.is_cancelled(), "Cancellation token should be triggered when request fails" @@ -1048,14 +1049,17 @@ mod tests { request_marker: request_marker.clone(), cancellation: cancellation.clone(), }))); + let lifecycle = RequestLifecycle { + request_id: 1, + request_marker, + cancellation, + }; let mut stream = credential_state_stream( futures::stream::iter([UsbEvent { state: UsbStateInternal::Waiting, }]), ctx.clone(), - 1, - request_marker, - cancellation, + lifecycle, ); assert!(stream.next().await.is_none()); @@ -1078,6 +1082,11 @@ mod tests { request_marker: request_marker.clone(), cancellation: cancellation.clone(), }))); + let lifecycle = RequestLifecycle { + request_id: 1, + request_marker, + cancellation: cancellation.clone(), + }; let (polled_tx, polled_rx) = oneshot::channel(); let mut polled_tx = Some(polled_tx); let pending_inner = futures::stream::poll_fn(move |_cx| { @@ -1086,13 +1095,7 @@ mod tests { } Poll::>::Pending }); - let mut stream = credential_state_stream( - pending_inner, - ctx.clone(), - 1, - request_marker, - cancellation.clone(), - ); + let mut stream = credential_state_stream(pending_inner, ctx.clone(), lifecycle); let waiting_task = tokio::spawn(async move { stream.next().await }); polled_rx.await.expect("inner stream should be polled"); @@ -1129,9 +1132,11 @@ mod tests { state: UsbStateInternal::Completed(create_test_credential_response()), }]), ctx.clone(), - 1, - stale_request_marker, - stale_cancellation.clone(), + RequestLifecycle { + request_id: 1, + request_marker: stale_request_marker, + cancellation: stale_cancellation.clone(), + }, ); assert!(stale_stream.next().await.is_none()); @@ -1174,7 +1179,7 @@ mod tests { // It sits in the channel when complete_request() cancels the token. hybrid_ref.emit(HybridStateInternal::Connecting); - // USB fails — CredentialStateStream calls complete_request → token cancelled + // USB failure completes the request and cancels the token. usb_ref.emit(UsbStateInternal::Waiting); usb_ref.fail(Error::Internal("test".to_string())); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); @@ -1219,7 +1224,7 @@ mod tests { assert!(rx.await.expect("request result should be sent").is_ok()); - // CredentialStateStream calls complete_request on Completed, which cancels the token + // The lifecycle stream completes the request and cancels its token. assert!( cancellation_token.is_cancelled(), "Cancellation token should be triggered when request completes successfully" @@ -1253,7 +1258,7 @@ mod tests { // It sits in the channel when complete_request() cancels the token. hybrid_ref.emit(HybridStateInternal::Connecting); - // USB completes — CredentialStateStream calls complete_request → token cancelled + // USB completion completes the request and cancels the token. usb_ref.emit(UsbStateInternal::Waiting); usb_ref.complete(credential_response); assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); From d785725a3de6f903a829141eb0cbb0fe7c5b9d9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 19:51:36 +0200 Subject: [PATCH 7/8] daemon: Test simultaneous terminal cancellation A terminal event can become ready in the same poll that cancels its request. Without the post-poll cancellation check, that event could still consume the active context. Cancel the token from inside the scripted poll and verify that the terminal result is discarded without sending a response. --- credentialsd/src/credential_service/mod.rs | 39 ++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 613e34e..e0d6977 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -1069,6 +1069,45 @@ mod tests { ); } + #[tokio::test] + async fn test_lifecycle_stream_discards_terminal_event_ready_during_cancellation() { + let request = create_test_request().await; + let (response_channel, mut response_rx) = oneshot::channel(); + let cancellation = CancellationToken::new(); + let cancellation_from_inner = cancellation.clone(); + let request_marker = Arc::new(RequestMarker); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id: 1, + request_marker: request_marker.clone(), + cancellation: cancellation.clone(), + }))); + let lifecycle = RequestLifecycle { + request_id: 1, + request_marker, + cancellation, + }; + let mut event = Some(UsbEvent { + state: UsbStateInternal::Completed(create_test_credential_response()), + }); + let inner = futures::stream::poll_fn(move |_cx| { + cancellation_from_inner.cancel(); + Poll::Ready(event.take()) + }); + let mut stream = credential_state_stream(inner, ctx.clone(), lifecycle); + + assert!(stream.next().await.is_none()); + assert!( + ctx.lock().unwrap().is_some(), + "a terminal event concurrent with cancellation must not complete the request" + ); + assert!(matches!( + response_rx.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + } + #[tokio::test] async fn test_lifecycle_stream_wakes_when_cancelled_while_inner_is_pending() { let request = create_test_request().await; From d3da3878b48152bdc56ae71a03679f35d92465d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sat, 5 Sep 2026 15:53:06 +0200 Subject: [PATCH 8/8] docs: Describe request ownership and prompt cancellation --- ARCHITECTURE.md | 5 ++++- CHANGELOG.md | 2 ++ credentialsd/src/credential_service/mod.rs | 1 - 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d873c25..80ccc70 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -122,7 +122,10 @@ any details that are not necessary for the UI to know (like the response channels mentioned above, which cannot be serialized over D-Bus anyway). USB, hybrid, and NFC handler events pass through shared request lifecycle -handling, while each transport keeps its own public states. +handling, while each transport keeps its own public states. All selected +transports share one snapshot of their originating request. Terminal events +are scoped to that request, so stale streams cannot complete a newer request. +Cancellation wakes pending streams. Actual interaction I/O is performed using the [libwebauthn][libwebauthn] library. diff --git a/CHANGELOG.md b/CHANGELOG.md index f08ad50..aabb02e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Improvements +- daemon: Wake pending credential streams when their request is cancelled. +- daemon: Ignore stale authenticator results from completed requests. - ui: Remove process ID from credential prompt. # 0.3.0 [2026-08-27] diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index e0d6977..b4c251b 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -229,7 +229,6 @@ impl todo!("Handle error when context is not set up.") } } - } #[async_trait]