From c41ff0d7187bfd6aa75f49afda715cb165bcf42b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 16:13:15 +0200 Subject: [PATCH 01/10] docs: Fix architecture typos Correct grammar and wording in the architecture overview without changing the documented design. --- ARCHITECTURE.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f9551e7..5459234 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -8,7 +8,7 @@ There are a few roles that are used in this architecture: - User: The user who owns the credential and interacts with the client and UI Controller. - Client: The application that is requesting a credential. -- Trusted Caller: A caller trusted by credentialsd to validate the permissions of the client appliation. +- Trusted Caller: A caller trusted by credentialsd to validate the permissions of the client application. - Gateway: Receives requests from the Trusted Caller, verifies request parameters and manages request concurrency. - Flow Controller: Proxies between the UI Controller and Credential Manager - UI Controller: Draws the UI and receives input from the user during the ceremony. @@ -23,7 +23,7 @@ validates the request and caller permissions, then passes the request to the **Flow Controller**, which in turn calls the Credential Manager, and the UI Controller. -The **UI Controller** is launches a UI for the user to respond to authenticator +The **UI Controller** launches a UI for the user to respond to authenticator requests for user interaction, like entering a client PIN or touching an authenticator for user presence. The **Credential Manager** interacts with the OS and hardware, like discovering available transports and authenticators. @@ -34,12 +34,12 @@ through the authentication ceremony, like prompts for a user to enter their PIN or touch the device. The UI Controller and Credential Manager pass user interaction and authenticator -events messages back and forth via the Flow Controller until an Authenticator -releases a credential, an terminal error is returned, or the request is +events back and forth via the Flow Controller until an Authenticator releases a +credential, a terminal error is returned, or the request is cancelled. Then, the Flow Controller sends the response back to the Gateway, which relays the credential to the Client. -Here is a diagram of the interactions between the roles. (User is ommitted for simplicity.) +Here is a diagram of the interactions between the roles. (User is omitted for simplicity.) ```mermaid sequenceDiagram From f13d9bc503e5cd18339fa7d034f3592959d446f8 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 02/10] 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 d2906195915d1ad28141da6870fb528094a9de39 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 03/10] 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 3d118d12d9b37c6f5b162c7b9aea0c0d144473b3 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 04/10] 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 cb9645784e668fd61c12bdeccfeed171c017ff29 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 05/10] 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 76213a0e4ee425dfb501ca691c60045be0eb168e 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 06/10] 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 024b9e4e94d976bcbc9d4a7618440a56cb56f213 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 18:52:57 +0200 Subject: [PATCH 07/10] daemon: Fail discovery when all transports exit silently When the final selected transport ended without a terminal result, discovery returned EOF while the request stayed active and its caller could wait indefinitely. Complete that request with a transport-independent failure while ignoring exhaustion from stale or explicitly cancelled discovery. --- credentialsd/src/credential_service/mod.rs | 313 ++++++++++++++++++++- 1 file changed, 312 insertions(+), 1 deletion(-) diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 613e34e..c859d45 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -230,6 +230,14 @@ impl } } + fn merge_discovery_streams( + &self, + selected_transports: Vec + Send + 'static>>>, + lifecycle: RequestLifecycle, + ) -> Pin + Send + 'static>> { + let selected_transports = futures::stream::select_all(selected_transports); + discovery_state_stream(selected_transports, self.ctx.clone(), lifecycle) + } } #[async_trait] @@ -346,7 +354,7 @@ impl Manage selected_transports.push(hybrid); } - futures::stream::select_all(selected_transports).boxed() + self.merge_discovery_streams(selected_transports, lifecycle) } } @@ -459,7 +467,41 @@ where }) } +/// Completes the request if every selected credential transport stops without +/// producing a terminal event. +fn discovery_state_stream( + mut inner: S, + ctx: Arc>>, + lifecycle: RequestLifecycle, +) -> Pin + Send + 'static>> +where + S: Stream + Unpin + Send + 'static, +{ + Box::pin(async_stream::stream! { + loop { + match lifecycle.cancellation.run_until_cancelled(inner.next()).await { + None => break, + Some(Some(state)) => yield state, + Some(None) => { + if lifecycle.cancellation.is_cancelled() { + break; + } + + let error = CredentialServiceError::Internal( + "All credential transports ended without a terminal event.".to_string(), + ); + if complete_request(&ctx, &lifecycle, Err(error.clone())) { + yield DeviceStateUpdate::Failed(error); + } + break; + } + } + } + }) +} + pub enum DeviceStateUpdate { + Failed(CredentialServiceError), Hybrid(HybridState), Nfc(NfcState), Usb(UsbState), @@ -468,6 +510,15 @@ pub enum DeviceStateUpdate { impl From for BackgroundEvent { fn from(value: DeviceStateUpdate) -> Self { match value { + DeviceStateUpdate::Failed(error) => match error { + CredentialServiceError::AuthenticatorError => BackgroundEvent::ErrorAuthenticator, + CredentialServiceError::NoCredentials => BackgroundEvent::ErrorNoCredentials, + CredentialServiceError::CredentialExcluded => { + BackgroundEvent::ErrorCredentialExcluded + } + CredentialServiceError::PinAttemptsExhausted => BackgroundEvent::ErrorAuthenticator, + CredentialServiceError::Internal(_) => BackgroundEvent::ErrorInternal, + }, DeviceStateUpdate::Hybrid(state) => (&state).into(), DeviceStateUpdate::Nfc(state) => (&state).into(), DeviceStateUpdate::Usb(state) => (&state).into(), @@ -1112,6 +1163,224 @@ mod tests { ); } + #[tokio::test] + async fn test_discovery_continues_when_one_transport_ends_unexpectedly() { + let request = create_test_request().await; + let (response_channel, response_rx) = oneshot::channel(); + let cancellation = CancellationToken::new(); + let request_id = 7; + let request_marker = Arc::new(RequestMarker); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id, + request_marker: request_marker.clone(), + cancellation: cancellation.clone(), + }))); + let lifecycle = RequestLifecycle { + request_id, + request_marker, + cancellation: cancellation.clone(), + }; + let hybrid_stream = credential_state_stream( + futures::stream::empty::(), + ctx.clone(), + lifecycle.clone(), + ) + .map(DeviceStateUpdate::from) + .boxed(); + let usb_stream = credential_state_stream( + futures::stream::iter([ + UsbEvent { + state: UsbStateInternal::Waiting, + }, + UsbEvent { + state: UsbStateInternal::Completed(create_test_credential_response()), + }, + ]), + ctx.clone(), + lifecycle.clone(), + ) + .map(DeviceStateUpdate::from) + .boxed(); + let mut stream = discovery_state_stream( + futures::stream::select_all(vec![hybrid_stream, usb_stream]), + ctx, + lifecycle, + ); + + assert!(matches!( + stream.next().await, + Some(DeviceStateUpdate::Usb(UsbState::Waiting)) + )); + assert!(!cancellation.is_cancelled()); + assert!(matches!( + stream.next().await, + Some(DeviceStateUpdate::Usb(UsbState::Completed)) + )); + + let result = response_rx.await.expect("request result should be sent"); + assert!(result.is_ok()); + assert!(cancellation.is_cancelled()); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn test_discovery_fails_when_all_transports_end_unexpectedly() { + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); + let request = create_test_request().await; + let (response_channel, response_rx) = oneshot::channel(); + let (_request_id, cancellation) = service + .init_request(&request, response_channel) + .await + .unwrap(); + let lifecycle = service.current_request().unwrap().1; + let usb_stream = futures::stream::iter([DeviceStateUpdate::Usb(UsbState::Waiting)]).boxed(); + let hybrid_stream = futures::stream::empty::().boxed(); + let mut stream = + service.merge_discovery_streams(vec![usb_stream, hybrid_stream], lifecycle); + + assert!(matches!( + stream.next().await, + Some(DeviceStateUpdate::Usb(UsbState::Waiting)) + )); + + let state = stream + .next() + .await + .expect("exhausting all transports should emit a failure state"); + assert!(matches!( + &state, + DeviceStateUpdate::Failed(CredentialServiceError::Internal(message)) + if message == "All credential transports ended without a terminal event." + )); + assert_eq!(BackgroundEvent::from(state), BackgroundEvent::ErrorInternal); + + let result = response_rx.await.expect("request result should be sent"); + assert!(matches!( + result, + Err(CredentialServiceError::Internal(message)) + if message == "All credential transports ended without a terminal event." + )); + assert!(cancellation.is_cancelled()); + assert!(service.ctx.lock().unwrap().is_none()); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn test_cancelling_pending_discovery_does_not_report_transport_exhaustion() { + let (usb_handler, usb_controller) = ScriptedTransport::::new(); + let service = CredentialService::new(EmptyTransport, EmptyTransport, usb_handler); + let request = create_test_request().await; + let (response_channel, response_rx) = oneshot::channel(); + let (request_id, cancellation) = service + .init_request(&request, response_channel) + .await + .unwrap(); + let usb_stream = service + .get_usb_credential() + .await + .map(DeviceStateUpdate::from) + .boxed(); + let lifecycle = service.current_request().unwrap().1; + let mut stream = service.merge_discovery_streams(vec![usb_stream], lifecycle); + + usb_controller.emit(UsbStateInternal::Waiting); + assert!(matches!( + stream.next().await, + Some(DeviceStateUpdate::Usb(UsbState::Waiting)) + )); + + let waiting_task = tokio::spawn(async move { stream.next().await }); + tokio::task::yield_now().await; + service.cancel_request(request_id).await; + + let event = tokio::time::timeout(Duration::from_secs(1), waiting_task) + .await + .expect("cancellation should wake the aggregate discovery stream") + .expect("discovery task should not panic"); + assert!(event.is_none()); + assert!(cancellation.is_cancelled()); + assert!(usb_controller.was_cancelled()); + + let result = response_rx.await.expect("request result should be sent"); + assert!(matches!( + result, + Err(CredentialServiceError::Internal(message)) + if message == format!("Cancelled request {request_id}.") + )); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_concurrent_terminal_events_emit_only_one_terminal_state() { + let request = create_test_request().await; + let (response_channel, response_rx) = oneshot::channel(); + let cancellation = CancellationToken::new(); + let request_id = 11; + let request_marker = Arc::new(RequestMarker); + let ctx = Arc::new(Mutex::new(Some(RequestContext { + request, + response_channel, + request_id, + request_marker: request_marker.clone(), + cancellation: cancellation.clone(), + }))); + let lifecycle = RequestLifecycle { + request_id, + request_marker, + cancellation: cancellation.clone(), + }; + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + + let make_discovery = || { + let barrier = barrier.clone(); + let usb_events = futures::stream::once(async move { + barrier.wait().await; + UsbEvent { + state: UsbStateInternal::Completed(create_test_credential_response()), + } + }) + .boxed(); + let usb_stream = credential_state_stream(usb_events, ctx.clone(), lifecycle.clone()) + .map(DeviceStateUpdate::from) + .boxed(); + discovery_state_stream( + futures::stream::select_all(vec![usb_stream]), + ctx.clone(), + lifecycle.clone(), + ) + }; + let mut first_discovery = make_discovery(); + let mut second_discovery = make_discovery(); + let first = tokio::spawn(async move { first_discovery.next().await }); + let second = tokio::spawn(async move { second_discovery.next().await }); + + barrier.wait().await; + let (first, second) = tokio::join!(first, second); + let events = [ + first.expect("first discovery task should not panic"), + second.expect("second discovery task should not panic"), + ]; + assert_eq!( + events + .into_iter() + .filter(|event| { + matches!(event, Some(DeviceStateUpdate::Usb(UsbState::Completed))) + }) + .count(), + 1, + "only the winning terminal event should be emitted" + ); + assert!(cancellation.is_cancelled()); + assert!(ctx.lock().unwrap().is_none()); + assert!( + response_rx + .await + .expect("request result should be sent") + .is_ok() + ); + } + #[tokio::test] async fn test_lifecycle_stream_cannot_complete_a_newer_request() { let request = create_test_request().await; @@ -1152,6 +1421,48 @@ mod tests { )); } + #[tokio::test] + async fn test_stale_discovery_exhaustion_cannot_complete_a_newer_request() { + let service = CredentialService::new(EmptyTransport, EmptyTransport, EmptyTransport); + let request = create_test_request().await; + let (stale_response_tx, stale_response_rx) = oneshot::channel(); + let (stale_request_id, _stale_cancellation) = service + .init_request(&request, stale_response_tx) + .await + .unwrap(); + let stale_lifecycle = service.current_request().unwrap().1; + + service.cancel_request(stale_request_id).await; + assert!(stale_response_rx.await.unwrap().is_err()); + + let (current_response_tx, mut current_response_rx) = oneshot::channel(); + let (current_request_id, current_cancellation) = service + .init_request(&request, current_response_tx) + .await + .unwrap(); + let stale_transport = futures::stream::empty::().boxed(); + let mut stale_discovery = + service.merge_discovery_streams(vec![stale_transport], stale_lifecycle); + + assert!(stale_discovery.next().await.is_none()); + assert!(!current_cancellation.is_cancelled()); + assert_eq!( + service + .ctx + .lock() + .unwrap() + .as_ref() + .map(|ctx| ctx.request_id), + Some(current_request_id) + ); + assert!(matches!( + current_response_rx.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + + service.cancel_request(current_request_id).await; + } + #[tokio::test] async fn test_failed_request_cancels_other_transports() { use credentialsd_common::model::Error; From acfb9c684d17761c334abad152bf30ba7aeb151e 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 08/10] 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 c859d45..0e33d74 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -1120,6 +1120,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 78054b691eae4731e91dc340c506b474c0848898 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 15:50:11 +0200 Subject: [PATCH 09/10] docs: Document credential transport lifecycle Describe request-scoped terminal handling, immediate cancellation wakeups, and transport exhaustion. Keep the changelog focused on the resulting fixes for stalled discovery and stale authenticator results. --- ARCHITECTURE.md | 6 ++++++ CHANGELOG.md | 2 ++ 2 files changed, 8 insertions(+) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5459234..f29aa1d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -121,6 +121,12 @@ 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). +Handler events pass through shared request lifecycle handling. Terminal events +are scoped to the request that started their stream, so stale streams cannot +complete a newer request. Cancellation wakes pending streams, and if every +selected transport ends without a terminal event, the request fails instead of +remaining pending. + Actual interaction I/O is performed using the [libwebauthn][libwebauthn] library. [libwebauthn]: https://github.com/linux-credentials/libwebauthn diff --git a/CHANGELOG.md b/CHANGELOG.md index f08ad50..487f973 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Improvements +- daemon: Prevent credential discovery from hanging after cancellation or transport shutdown. +- daemon: Ignore stale authenticator results from completed requests. - ui: Remove process ID from credential prompt. # 0.3.0 [2026-08-27] From 9e97418013ed6c94fd284a5d03b645d391d1fb4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sun, 30 Aug 2026 20:18:11 +0200 Subject: [PATCH 10/10] daemon: Bound concurrent credential discovery Repeated DiscoveryRequested signals could spawn independent streams for the same request, consuming resources and racing request completion. Hold one request-scoped permit for the lifetime of the forwarded stream. Concurrent signals are ignored, while releasing the permit allows a later discovery attempt. --- credentialsd/src/dbus/flow_control.rs | 46 +++++++++++++++++++++++++-- 1 file changed, 44 insertions(+), 2 deletions(-) diff --git a/credentialsd/src/dbus/flow_control.rs b/credentialsd/src/dbus/flow_control.rs index 61362a2..9af6241 100644 --- a/credentialsd/src/dbus/flow_control.rs +++ b/credentialsd/src/dbus/flow_control.rs @@ -18,7 +18,7 @@ use credentialsd_common::{ use futures_lite::{Stream, StreamExt}; use tokio::sync::mpsc::Receiver; use tokio::sync::oneshot; -use tokio::sync::{Mutex as AsyncMutex, mpsc::Sender}; +use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore, mpsc::Sender}; use tokio::task::AbortHandle; use zbus::connection::Connection; use zbus::zvariant::OwnedObjectPath; @@ -146,6 +146,7 @@ async fn handle>>> = Arc::new(Mutex::new(None)); let set_pin_tx: Arc>>> = Arc::new(Mutex::new(None)); let cred_selector_tx = Arc::new(Mutex::new(None)); + let discovery_gate = Arc::new(Semaphore::new(1)); let wait_for_ui_request_fut = async { loop { @@ -155,6 +156,13 @@ async fn handle { + let Some(discovery_permit) = claim_discovery(&discovery_gate) else { + tracing::debug!( + %request_id, + "Ignoring discovery request while discovery is already active." + ); + continue; + }; let client_pin_tx = client_pin_tx.clone(); let set_pin_tx = set_pin_tx.clone(); let cred_selector_tx = cred_selector_tx.clone(); @@ -203,7 +211,7 @@ async fn handle { let pin_fd = OwnedFd::from(pin_fd); @@ -295,9 +303,20 @@ async fn handle) -> Option { + discovery_gate.clone().try_acquire_owned().ok() +} + fn forward_background_event_stream( flow: Ceremony, mut stream: impl Stream + Send + Unpin + 'static, + _discovery_permit: OwnedSemaphorePermit, ) { tokio::spawn(async move { while let Some(event) = stream.next().await { @@ -311,6 +330,29 @@ fn forward_background_event_stream( }); } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn discovery_can_restart_after_the_active_stream_releases_its_claim() { + let discovery_gate = Arc::new(Semaphore::new(1)); + + let active_discovery = + claim_discovery(&discovery_gate).expect("first discovery should claim the slot"); + assert!( + claim_discovery(&discovery_gate).is_none(), + "a concurrent discovery must not claim the same request" + ); + + drop(active_discovery); + assert!( + claim_discovery(&discovery_gate).is_some(), + "ending the active discovery must allow a later attempt" + ); + } +} + /// Coordinates between user and various devices connected to the machine to /// fulfill credential requests. #[async_trait]