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/3] 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/3] 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/3] 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