diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f9762f6..d3a2a5a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,7 @@ run xdg-desktop-portal while we work on upstreaming the changes. - daemon: Add support for CTAP2 hybrid over BLE behind a feature flag. - daemon: Deduplicate USB state events emitted over D-Bus. - daemon: Don't use hybrid when not available +- daemon: Cancel other transports, if one succeeded/failed - ui: Add Georgian translations. (Thank you, @EkaterinePopova!) - ui: Add a portal backend API to credentialsd-ui. - ui: Allow setting client PIN during the flow when required. diff --git a/Cargo.lock b/Cargo.lock index aa6e0644..9293ea4f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -821,6 +821,7 @@ dependencies = [ "serde_json", "tokio", "tokio-stream", + "tokio-util", "tracing", "tracing-subscriber", "zbus", @@ -4076,14 +4077,13 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.19" +version = "0.7.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +checksum = "d7fcaa8d55a2bdd6b83ace262b016eca0d79ee02818c5c1bcdf0305114081078" dependencies = [ "bytes", "futures-core", "futures-sink", - "libc", "pin-project-lite", "tokio", ] diff --git a/credentialsd/Cargo.toml b/credentialsd/Cargo.toml index 04f54f3a..347e7ee0 100644 --- a/credentialsd/Cargo.toml +++ b/credentialsd/Cargo.toml @@ -25,6 +25,7 @@ ring = "=0.17.14" serde_json = "=1.0.151" tokio = { version = "=1.53.1", features = ["rt-multi-thread"] } tokio-stream = "=0.1.19" +tokio-util = "=0.7.13" [dev-dependencies] gio = "=0.22.8" diff --git a/credentialsd/src/credential_service/hybrid.rs b/credentialsd/src/credential_service/hybrid.rs index 5f136346..be6bb644 100644 --- a/credentialsd/src/credential_service/hybrid.rs +++ b/credentialsd/src/credential_service/hybrid.rs @@ -7,14 +7,18 @@ use tokio::sync::{ broadcast, mpsc::{self, Sender}, }; +use tokio_util::sync::CancellationToken; use tracing::{debug, error}; -use libwebauthn::transport::cable::channel::{CableUpdate, CableUxUpdate}; use libwebauthn::transport::cable::qr_code_device::{ CableQrCodeDevice, CableTransports, QrCodeOperationHint, }; use libwebauthn::transport::{Channel, ChannelSettings, Device}; use libwebauthn::webauthn::{WebAuthn, error::WebAuthnError}; +use libwebauthn::{ + proto::CtapError, + transport::cable::channel::{CableUpdate, CableUxUpdate}, +}; use credentialsd_common::{ memfd::write_secret, @@ -29,6 +33,7 @@ pub(crate) trait HybridHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Unpin + Send + Sized + 'static; } @@ -44,6 +49,7 @@ impl HybridHandler for InternalHybridHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Unpin + Send + Sized + 'static { tracing::debug!("Starting hybrid operation"); let request = request.clone(); @@ -93,65 +99,65 @@ impl HybridHandler for InternalHybridHandler { debug!("Reached end of Hybrid updates stream."); }); - tracing::debug!("Polling hybrid channel for updates."); - let response: Result = loop { - match &request { - CredentialRequest::CreatePublicKeyCredentialRequest(make_request) => { - match channel.webauthn_make_credential(make_request).await { - Ok(response) => break Ok(response.into()), - Err(WebAuthnError::Ctap(ctap_error)) => { - if ctap_error.is_retryable_user_error() { - tracing::debug!( - "Retrying credential creation operation because of CTAP error: {:?}", - ctap_error - ); - continue; - } else { - tracing::error!( - "Received CTAP unrecoverable CTAP error: {:?}", - ctap_error - ); - break Err(Error::AuthenticatorError); - } - } - Err(err) => { - tracing::error!( - "Received unrecoverable error from authenticator: {:?}", - err - ); - break Err(Error::AuthenticatorError); - } - }; + let wait_for_response_fut = async { + loop { + let response: Result = match &request { + CredentialRequest::CreatePublicKeyCredentialRequest(make_request) => { + channel + .webauthn_make_credential(make_request) + .await + .map(|response| response.into()) + } + CredentialRequest::GetPublicKeyCredentialRequest(get_request) => { + channel + .webauthn_get_assertion(get_request) + .await + .map(|response| response.into()) + } + }; + match response { + Ok(response) => { + tracing::debug!("Received credential from hybrid authenticator"); + break Ok(response); + } + Err(WebAuthnError::Ctap(ctap_error)) + if ctap_error.is_retryable_user_error() => + { + tracing::debug!(%ctap_error, "Retrying WebAuthn operation"); + continue; + } + Err(err) => { + tracing::error!(%err, + "Failed to make/get credential with hybrid authenticator" + ); + break Err(err); + } + } + } + .map_err(|err| match err { + WebAuthnError::Ctap(CtapError::PINAuthBlocked) => { + Error::PinAttemptsExhausted } - CredentialRequest::GetPublicKeyCredentialRequest(get_request) => { - match channel.webauthn_get_assertion(get_request).await { - Ok(response) => break Ok(response.into()), - Err(WebAuthnError::Ctap(ctap_error)) => { - if ctap_error.is_retryable_user_error() { - tracing::debug!( - "Retrying assertion operation because of CTAP error: {:?}", - ctap_error - ); - continue; - } else { - tracing::error!( - "Received CTAP unrecoverable CTAP error: {:?}", - ctap_error - ); - break Err(Error::AuthenticatorError); - } - } - Err(err) => { - tracing::error!( - "Received unrecoverable error from authenticator: {:?}", - err - ); - break Err(Error::AuthenticatorError); - } - }; + WebAuthnError::Ctap(CtapError::NoCredentials) => Error::NoCredentials, + WebAuthnError::Ctap(CtapError::CredentialExcluded) => { + Error::CredentialExcluded } + _ => Error::AuthenticatorError, + }) + }; + + tracing::debug!("Polling hybrid channel for updates."); + let response = match cancellation + .run_until_cancelled(wait_for_response_fut) + .await + { + Some(resp) => resp, + None => { + tracing::debug!("Hybrid handler cancelled, stopping processing"); + Err(Error::Internal("Request cancelled".to_string())) } }; + let terminal_state = match response { Ok(auth_response) => HybridStateInternal::Completed(Box::new(auth_response)), Err(_) => HybridStateInternal::Failed, diff --git a/credentialsd/src/credential_service/mod.rs b/credentialsd/src/credential_service/mod.rs index 39435bf9..24cc9523 100644 --- a/credentialsd/src/credential_service/mod.rs +++ b/credentialsd/src/credential_service/mod.rs @@ -21,6 +21,7 @@ use libwebauthn::{ }; use nfc::{NfcEvent, NfcHandler, NfcState, NfcStateInternal}; use tokio::sync::oneshot; +use tokio_util::sync::CancellationToken; use credentialsd_common::model::{ BackgroundEvent, Device, Error as CredentialServiceError, Transport, @@ -41,6 +42,19 @@ pub use usb::UsbState; /// Identifier for a request to be used for cancellation. pub type RequestId = u32; +/// Helper function to sleep with cancellation support. +async fn cancellable_sleep( + duration: std::time::Duration, + cancellation: &CancellationToken, +) -> Result<(), CredentialServiceError> { + tokio::select! { + _ = tokio::time::sleep(duration) => Ok(()), + _ = cancellation.cancelled() => { + Err(CredentialServiceError::Internal("Request cancelled".to_string())) + } + } +} + /// Process-wide in-memory store so a security key's pinUvAuthToken is reused across ceremonies. fn persistent_token_store() -> Arc { static STORE: OnceLock> = OnceLock::new(); @@ -54,6 +68,7 @@ struct RequestContext { request: CredentialRequest, response_channel: oneshot::Sender>, request_id: RequestId, + cancellation: CancellationToken, } impl RequestContext { @@ -73,7 +88,7 @@ pub trait ManageDevice { &self, request: &CredentialRequest, tx: oneshot::Sender>, - ) -> Result; + ) -> Result<(RequestId, CancellationToken), CredentialServiceError>; async fn cancel_request(&self, request_id: RequestId); async fn get_available_public_key_devices(&self) -> Result, ()>; async fn start_discovery( @@ -112,8 +127,17 @@ impl &self, ) -> Pin + Send + 'static>> { let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { ref request, .. }) = *guard { - let stream = self.hybrid_handler.lock().unwrap().start(request); + if let Some(RequestContext { + ref request, + ref cancellation, + .. + }) = *guard + { + let stream = self + .hybrid_handler + .lock() + .unwrap() + .start(request, cancellation.clone()); let ctx = self.ctx.clone(); Box::pin(HybridStateStream { inner: stream, ctx }) } else { @@ -126,8 +150,17 @@ impl async fn get_usb_credential(&self) -> Pin + Send + 'static>> { let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { ref request, .. }) = *guard { - let stream = self.usb_handler.lock().unwrap().start(request); + if let Some(RequestContext { + ref request, + ref cancellation, + .. + }) = *guard + { + let stream = self + .usb_handler + .lock() + .unwrap() + .start(request, cancellation.clone()); let ctx = self.ctx.clone(); Box::pin(UsbStateStream { inner: stream, ctx }) } else { @@ -140,8 +173,17 @@ impl async fn _get_nfc_credential(&self) -> Pin + Send + 'static>> { let guard = self.ctx.lock().unwrap(); - if let Some(RequestContext { ref request, .. }) = *guard { - let stream = self._nfc_handler.lock().unwrap().start(request); + if let Some(RequestContext { + ref request, + ref cancellation, + .. + }) = *guard + { + let stream = self + ._nfc_handler + .lock() + .unwrap() + .start(request, cancellation.clone()); let ctx = self.ctx.clone(); Box::pin(NfcStateStream { inner: stream, ctx }) } else { @@ -161,23 +203,26 @@ impl Manage &self, request: &CredentialRequest, tx: oneshot::Sender>, - ) -> Result { + ) -> Result<(RequestId, CancellationToken), CredentialServiceError> { let mut cred_request = self.ctx.lock().unwrap(); if cred_request.is_some() { Err(CredentialServiceError::Internal( "Already a request in progress.".to_string(), )) } else { - let request_id: RequestId = rand::random(); + // Generate non-zero request ID + let request_id: RequestId = rand::random_range(1..=u32::MAX); + let cancellation = CancellationToken::new(); // TODO: Spawn a task here that will listen to the signals from ui_control_client. // Move the get_*_credential(), etc. from gateway to here. let ctx = RequestContext { request: request.clone(), response_channel: tx, request_id, + cancellation: cancellation.clone(), }; _ = cred_request.insert(ctx); - Ok(request_id) + Ok((request_id, cancellation)) } } @@ -187,7 +232,7 @@ impl Manage && request_id == ctx.request_id { tracing::debug!("Cancelling request {request_id}"); - // TODO: cancel sub-tasks: hybrid and USB streams. + ctx.cancellation.cancel(); // It's fine if the requestor is no longer listening for the response. // TODO: create Cancelled variant @@ -282,27 +327,33 @@ where match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { Poll::Pending => Poll::Pending, Poll::Ready(Some(HybridEvent { state })) => { - if let HybridStateInternal::Completed(hybrid_response) = &state { - let response = match &**hybrid_response { - AuthenticatorResponse::CredentialCreated(make_credential_response) => { - CredentialResponse::from_make_credential( - make_credential_response, - &["hybrid"], - "cross-platform", - ) - } - AuthenticatorResponse::CredentialsAsserted(get_assertion_response) => { - CredentialResponse::from_get_assertion( - // When doing hybrid, the authenticator is capable of displaying it's own UI. - // So we assume here, it only ever returns one assertion. - // In case this doesn't hold true, we have to implement credential selection here, - // as is done for USB. - &get_assertion_response.assertions[0], - "cross-platform", - ) - } - }; - complete_request(ctx, response.clone()); + match &state { + HybridStateInternal::Completed(hybrid_response) => { + let response = match &**hybrid_response { + AuthenticatorResponse::CredentialCreated(make_credential_response) => { + CredentialResponse::from_make_credential( + make_credential_response, + &["hybrid"], + "cross-platform", + ) + } + AuthenticatorResponse::CredentialsAsserted(get_assertion_response) => { + CredentialResponse::from_get_assertion( + // When doing hybrid, the authenticator is capable of displaying it's own UI. + // So we assume here, it only ever returns one assertion. + // In case this doesn't hold true, we have to implement credential selection here, + // as is done for USB. + &get_assertion_response.assertions[0], + "cross-platform", + ) + } + }; + complete_request(ctx, Ok(response.clone())); + } + HybridStateInternal::Failed => { + complete_request(ctx, Err(CredentialServiceError::AuthenticatorError)); + } + _ => {} } Poll::Ready(Some(state.into())) } @@ -330,8 +381,14 @@ where match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { Poll::Pending => Poll::Pending, Poll::Ready(Some(UsbEvent { state })) => { - if let UsbStateInternal::Completed(response) = &state { - complete_request(ctx, response.clone()); + 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())) } @@ -360,8 +417,14 @@ where match Box::pin(Box::pin(self).as_mut().inner.next()).poll(cx) { Poll::Pending => Poll::Pending, Poll::Ready(Some(NfcEvent { state })) => { - if let NfcStateInternal::Completed(response) = &state { - complete_request(ctx, response.clone()); + 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())) } @@ -404,10 +467,14 @@ impl From for DeviceStateUpdate { } } -fn complete_request(ctx: &Mutex>, response: CredentialResponse) { +fn complete_request( + ctx: &Mutex>, + response: Result, +) { match ctx.lock().unwrap().take() { Some(ctx) => { - ctx.send_response(Ok(response)); + ctx.cancellation.cancel(); + ctx.send_response(response); } _ => { tracing::error!("Tried to consume context to respond to caller, but none was found.") @@ -432,3 +499,712 @@ impl From for AuthenticatorResponse { Self::CredentialsAsserted(value) } } + +#[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() + } + } + + fn create_test_credential_response() -> CredentialResponse { + use libwebauthn::ops::webauthn::GetAssertionResponse; + + // Create a minimal GetAssertion response for testing + let get_assertion_response = GetAssertionResponse { + assertions: vec![libwebauthn::ops::webauthn::Assertion { + credential_id: None, + authenticator_data: libwebauthn::fido::AuthenticatorData { + rp_id_hash: [0u8; 32], + flags: libwebauthn::fido::AuthenticatorDataFlags::empty(), + signature_count: 0, + attested_credential: None, + extensions: None, + raw: None, + }, + signature: vec![], + user: None, + credentials_count: None, + user_selected: None, + unsigned_extensions_output: None, + transport: None, + }], + }; + + CredentialResponse::from_get_assertion( + &get_assertion_response.assertions[0], + "cross-platform", + ) + } + + async fn create_test_request() -> CredentialRequest { + use libwebauthn::ops::webauthn::{ + MakeCredentialRequest, OriginValidation, RequestSettings, idl::origin::RequestOrigin, + }; + + let request_json = r#" + { + "rp": { + "id": "example.com", + "name": "Example Relying Party" + }, + "user": { + "id": "MTIzNDU2NzgxMjM0NTY3ODEyMzQ1Njc4MTIzNDU2Nzg", + "name": "test@example.com", + "displayName": "Test User" + }, + "challenge": "MTIzNDU2NzgxMjM0NTY3ODEyMzQ1Njc4MTIzNDU2Nzg", + "pubKeyCredParams": [ + {"type": "public-key", "alg": -7} + ], + "timeout": 60000, + "excludeCredentials": [], + "authenticatorSelection": { + "residentKey": "discouraged", + "userVerification": "preferred" + }, + "attestation": "none" + } + "#; + + let request_origin: RequestOrigin = + "https://example.com".try_into().expect("Invalid origin"); + + let settings = RequestSettings { + origin: OriginValidation::Trust, + }; + + let make_credentials_request = + MakeCredentialRequest::prepare(&request_origin, request_json, &settings) + .await + .expect("Failed to parse request JSON"); + + CredentialRequest::CreatePublicKeyCredentialRequest(make_credentials_request) + } + + #[tokio::test] + async fn test_init_request_returns_token_and_id() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let (tx, _rx) = oneshot::channel(); + let request = create_test_request().await; + + let result = service.init_request(&request, tx).await; + + assert!(result.is_ok()); + let (request_id, cancellation_token) = result.unwrap(); + assert!(request_id > 0); + assert!(!cancellation_token.is_cancelled()); + } + + #[tokio::test] + async fn test_cancel_request_triggers_cancellation() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let (tx, _rx) = oneshot::channel(); + let request = create_test_request().await; + + let (request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + assert!(!cancellation_token.is_cancelled()); + + service.cancel_request(request_id).await; + assert!(cancellation_token.is_cancelled()); + } + + #[tokio::test] + async fn test_cancellable_sleep_completes_normally() { + let token = CancellationToken::new(); + let start = tokio::time::Instant::now(); + + let result = cancellable_sleep(Duration::from_millis(50), &token).await; + + assert!(result.is_ok()); + assert!(start.elapsed() >= Duration::from_millis(50)); + } + + #[tokio::test] + async fn test_cancellable_sleep_respects_cancellation() { + let token = CancellationToken::new(); + token.cancel(); // Pre-cancel the token + + let start = tokio::time::Instant::now(); + let result = cancellable_sleep(Duration::from_secs(5), &token).await; + + assert!(result.is_err()); + // Should return immediately, not after 5 seconds + assert!(start.elapsed() < Duration::from_millis(100)); + } + + #[tokio::test] + async fn test_init_request_rejects_concurrent() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let (tx1, _rx1) = oneshot::channel(); + let (tx2, _rx2) = oneshot::channel(); + let request = create_test_request().await; + + // First request should succeed + let result1 = service.init_request(&request, tx1).await; + assert!(result1.is_ok()); + + // Second concurrent request should fail + let result2 = service.init_request(&request, tx2).await; + assert!(result2.is_err()); + assert!( + result2 + .unwrap_err() + .to_string() + .contains("Already a request in progress") + ); + } + + // 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; }; + loop { + tokio::select! { + biased; + _ = cancellation.cancelled() => { + cancelled.store(true, Ordering::SeqCst); + break; + } + maybe = rx.recv() => match maybe { + Some(state) => yield wrap(state), + 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 request = create_test_request().await; + let (tx, _rx) = oneshot::channel(); + + let (request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + + // Token should not be cancelled initially + assert!(!cancellation_token.is_cancelled()); + + // Cancel by ID + service.cancel_request(request_id).await; + + // Token should now be cancelled + assert!(cancellation_token.is_cancelled()); + } + + #[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 service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + 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 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.shift_state(UsbStateInternal::Waiting); + hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Init(_)) + )); + + // 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); + + // 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; + + assert!( + usb_remaining.is_empty(), + "USB should not emit any more states after cancellation" + ); + assert!( + hybrid_remaining.is_empty(), + "Hybrid should not emit any more states after cancellation" + ); + } + + #[tokio::test] + async fn test_cancellation_cleans_up_request_context() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let request = create_test_request().await; + let (tx, _rx) = oneshot::channel(); + + let (request_id, _token) = service.init_request(&request, tx).await.unwrap(); + + // Cancel the request + service.cancel_request(request_id).await; + + // Should be able to start a new request now (context cleaned up) + let (tx2, _rx2) = oneshot::channel(); + let result = service.init_request(&request, tx2).await; + assert!( + result.is_ok(), + "Should be able to init new request after cancel" + ); + } + + #[tokio::test] + async fn test_cancel_with_unknown_id_is_noop() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let request = create_test_request().await; + let (tx, _rx) = oneshot::channel(); + + let (request_id, cancellation_token) = service.init_request(&request, tx).await.unwrap(); + + // Cancel with a different ID (should be a no-op) + let wrong_id = request_id.wrapping_add(1); + service.cancel_request(wrong_id).await; + + // Original request should still be active + assert!( + !cancellation_token.is_cancelled(), + "Token should not be cancelled with wrong ID" + ); + + // Now cancel with correct ID + service.cancel_request(request_id).await; + assert!( + cancellation_token.is_cancelled(), + "Token should be cancelled with correct ID" + ); + } + + #[tokio::test] + async fn test_cancel_with_no_active_request_is_noop() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + + // Cancel when no request is active (should not crash or panic) + service.cancel_request(12345).await; + + // Should still be able to start a new request + let request = create_test_request().await; + let (tx, _rx) = oneshot::channel(); + let result = service.init_request(&request, tx).await; + assert!( + result.is_ok(), + "Should be able to init request after no-op cancel" + ); + } + + #[tokio::test] + async fn test_request_id_matches_on_init() { + let service = CredentialService::new(MockHybridHandler, MockNfcHandler, MockUsbHandler); + let request = create_test_request().await; + let (tx, _rx) = oneshot::channel(); + + let (request_id_1, _token_1) = service.init_request(&request, tx).await.unwrap(); + + // Cancel to free up the service + service.cancel_request(request_id_1).await; + + // Start a new request + let (tx2, _rx2) = oneshot::channel(); + let (request_id_2, _token_2) = service.init_request(&request, tx2).await.unwrap(); + + // IDs should be different (random) + assert_ne!( + request_id_1, request_id_2, + "Sequential requests should (almost certainly) have different IDs" + ); + } + + #[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(); + + assert!(!usb_ref.was_cancelled()); + assert!(!hybrid_ref.was_cancelled()); + + let service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + 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 usb_stream = service.get_usb_credential().await; + 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())); + assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Init(_)) + )); + + // 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); + + // Explicitly cancel — token becomes cancelled synchronously + service.cancel_request(request_id).await; + assert!( + cancellation_token.is_cancelled(), + "Cancellation token should be triggered after cancel_request" + ); + + // biased select! polls cancellation first, discarding the queued states + let usb_remaining: Vec<_> = usb_stream.collect().await; + let hybrid_remaining: Vec<_> = hybrid_stream.collect().await; + + assert!( + usb_remaining.is_empty(), + "USB should not emit any more states after cancellation" + ); + assert!( + hybrid_remaining.is_empty(), + "Hybrid should not emit any more states after cancellation" + ); + + // Flags are set by run_tracking_stream when it observes the cancelled token + assert!( + usb_ref.was_cancelled(), + "USB handler should have detected cancellation" + ); + assert!( + hybrid_ref.was_cancelled(), + "Hybrid handler should have detected cancellation" + ); + } + + #[tokio::test] + 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 service = CredentialService::new(MockHybridHandler, MockNfcHandler, usb_handler); + 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 usb_stream = service.get_usb_credential().await; + assert!(!cancellation_token.is_cancelled()); + + usb_ref.shift_state(UsbStateInternal::Waiting); + assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); + + usb_ref.shift_state(UsbStateInternal::Failed(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 + assert!( + cancellation_token.is_cancelled(), + "Cancellation token should be triggered when request fails" + ); + } + + #[tokio::test] + 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 service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + 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 usb_stream = service.get_usb_credential().await; + let mut hybrid_stream = service.get_hybrid_credential().await; + + // Confirm hybrid stream is live + hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Init(_)) + )); + + // 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); + + // 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(), + ))); + assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); + assert!(matches!(usb_stream.next().await, Some(UsbState::Failed(_)))); + + assert!( + cancellation_token.is_cancelled(), + "Cancellation token should be triggered when USB fails" + ); + + // biased select! polls cancellation first, discarding the queued Connecting state + let hybrid_remaining: Vec<_> = hybrid_stream.collect().await; + assert!( + hybrid_remaining.is_empty(), + "Hybrid should not emit any more states after USB fails" + ); + assert!( + hybrid_ref.was_cancelled(), + "Hybrid handler should have detected cancellation when USB failed" + ); + } + + #[tokio::test] + 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 service = CredentialService::new(MockHybridHandler, MockNfcHandler, usb_handler); + 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 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)); + 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!( + cancellation_token.is_cancelled(), + "Cancellation token should be triggered when request completes successfully" + ); + } + + #[tokio::test] + 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 service = CredentialService::new(hybrid_handler, MockNfcHandler, usb_handler); + 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 usb_stream = service.get_usb_credential().await; + let mut hybrid_stream = service.get_hybrid_credential().await; + + // Confirm hybrid stream is live + hybrid_ref.shift_state(HybridStateInternal::Init("qr".to_string())); + assert!(matches!( + hybrid_stream.next().await, + Some(HybridState::Init(_)) + )); + + // 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); + + // USB completes — UsbStateStream calls complete_request → token cancelled + usb_ref.shift_state(UsbStateInternal::Waiting); + usb_ref.shift_state(UsbStateInternal::Completed(credential_response)); + assert!(matches!(usb_stream.next().await, Some(UsbState::Waiting))); + assert!(matches!(usb_stream.next().await, Some(UsbState::Completed))); + + assert!( + cancellation_token.is_cancelled(), + "Cancellation token should be triggered when USB completes" + ); + + // biased select! polls cancellation first, discarding the queued Connecting state + let hybrid_remaining: Vec<_> = hybrid_stream.collect().await; + assert!( + hybrid_remaining.is_empty(), + "Hybrid should not emit any more states after USB completes" + ); + assert!( + hybrid_ref.was_cancelled(), + "Hybrid handler should have detected cancellation when USB completed" + ); + } +} diff --git a/credentialsd/src/credential_service/nfc.rs b/credentialsd/src/credential_service/nfc.rs index 96b71021..44ab0a74 100644 --- a/credentialsd/src/credential_service/nfc.rs +++ b/credentialsd/src/credential_service/nfc.rs @@ -13,6 +13,7 @@ use libwebauthn::{ }; use tokio::sync::broadcast; use tokio::sync::mpsc::{self, Receiver, Sender, WeakSender}; +use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use credentialsd_common::model::{BackgroundEvent, Credential, Error, PinNotSetError}; @@ -26,6 +27,7 @@ pub(crate) trait NfcHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Send + Sized + Unpin + 'static; } @@ -36,8 +38,17 @@ impl InProcessNfcHandler { async fn process_idle_waiting( failures: &mut usize, prev_nfc_state: &NfcStateInternal, + cancellation: &CancellationToken, ) -> Result { - match libwebauthn::transport::nfc::get_nfc_device().await { + let list_device_fut = libwebauthn::transport::nfc::get_nfc_device(); + let Some(result) = cancellation.run_until_cancelled(list_device_fut).await else { + // TODO: We should introduce a cancelled-error variant and return this here, + // so we can differentiate between internal errors, cancellation by user + // and cancellation because other transfers finished + tracing::debug!("NFC idle polling cancelled"); + return Err(Error::Internal("Request cancelled".to_string())); + }; + match result { Ok(Some(nfc_device)) => Ok(NfcStateInternal::Connected(nfc_device)), Ok(None) => { let state = NfcStateInternal::Waiting; @@ -55,7 +66,7 @@ impl InProcessNfcHandler { "Failed to list NFC authenticators: {:?}. Throttling NFC state updates", err ); - tokio::time::sleep(Duration::from_secs(1)).await; + super::cancellable_sleep(Duration::from_secs(1), cancellation).await?; Ok(prev_nfc_state.clone()) } } @@ -158,6 +169,7 @@ impl InProcessNfcHandler { async fn process( tx: Sender, cred_request: CredentialRequest, + cancellation: CancellationToken, ) -> Result<(), Error> { let mut state = NfcStateInternal::Idle; let (signal_tx, mut signal_rx) = mpsc::channel(256); @@ -169,34 +181,58 @@ impl InProcessNfcHandler { loop { tracing::debug!("current nfc state: {:?}", state); let prev_nfc_state = state; - let next_nfc_state = match prev_nfc_state { - NfcStateInternal::Idle | NfcStateInternal::Waiting => { - Self::process_idle_waiting(&mut failures, &prev_nfc_state).await - } - NfcStateInternal::Connected(device) => { - let signal_tx2 = signal_tx.clone(); - let cred_request = cred_request.clone(); - tokio::spawn(async move { - handle_events(&cred_request, device, &signal_tx2).await; - }); - Self::process_user_interaction(&mut signal_rx, &cred_tx).await - } - NfcStateInternal::NeedsPin { .. } - | NfcStateInternal::PinNotSet { .. } - | NfcStateInternal::NeedsUserVerification { .. } => { - Self::process_user_interaction(&mut signal_rx, &cred_tx).await + + let select_next_nfc_state_fut = async { + match prev_nfc_state { + NfcStateInternal::Idle | NfcStateInternal::Waiting => { + Self::process_idle_waiting(&mut failures, &prev_nfc_state, &cancellation) + .await + } + NfcStateInternal::Connected(device) => { + let signal_tx2 = signal_tx.clone(); + let cred_request = cred_request.clone(); + let cancellation = cancellation.clone(); + tokio::spawn(async move { + handle_events(&cred_request, device, &signal_tx2, cancellation).await; + }); + Self::process_user_interaction(&mut signal_rx, &cred_tx).await + } + NfcStateInternal::NeedsPin { .. } + | NfcStateInternal::PinNotSet { .. } + | NfcStateInternal::NeedsUserVerification { .. } => { + Self::process_user_interaction(&mut signal_rx, &cred_tx).await + } + NfcStateInternal::SelectCredential { + response, + cred_tx: _, + } => Self::process_select_credential(response, &mut cred_rx).await, + // Terminal states - preserve state unchanged, will break loop after sending + NfcStateInternal::Completed(_) | NfcStateInternal::Failed(_) => { + Ok(prev_nfc_state.clone()) + } } - NfcStateInternal::SelectCredential { - response, - cred_tx: _, - } => Self::process_select_credential(response, &mut cred_rx).await, - NfcStateInternal::Completed(_) => break Ok(()), - NfcStateInternal::Failed(err) => break Err(err), }; + + let Some(next_nfc_state) = cancellation + .run_until_cancelled(select_next_nfc_state_fut) + .await + else { + tracing::debug!("NFC handler cancelled, stopping processing"); + break Err(Error::Internal("Request cancelled".to_string())); + }; + state = next_nfc_state.unwrap_or_else(NfcStateInternal::Failed); + tx.send(state.clone()).await.map_err(|_| { Error::Internal("NFC state channel receiver closed prematurely".to_string()) })?; + + // Check for terminal states AFTER sending + match state { + NfcStateInternal::Completed(_) => break Ok(()), + NfcStateInternal::Failed(err) => break Err(err), + _ => {} + } } } } @@ -205,6 +241,7 @@ async fn handle_events( cred_request: &CredentialRequest, mut device: NfcDevice, signal_tx: &Sender>, + cancellation: CancellationToken, ) { let device_debug = device.to_string(); match device @@ -230,49 +267,71 @@ async fn handle_events( "Polling for credential from NFC authenticator {}", &device_debug ); - let response: Result = loop { - let response = match cred_request { - CredentialRequest::CreatePublicKeyCredentialRequest(make_cred_request) => { - channel - .webauthn_make_credential(make_cred_request) - .await - .map(|response| { - NfcUvMessage::ReceivedCredentials(Box::new(response.into())) - }) - } - CredentialRequest::GetPublicKeyCredentialRequest(get_cred_request) => channel - .webauthn_get_assertion(get_cred_request) - .await - .map(|response| { - NfcUvMessage::ReceivedCredentials(Box::new(response.into())) - }), - }; - match response { - Ok(response) => { - tracing::debug!("Received credential from NFC authenticator"); - break Ok(response); - } - Err(WebAuthnError::Ctap(ctap_error)) - if ctap_error.is_retryable_user_error() => - { - warn!("Retrying WebAuthn credential operation"); - continue; - } - Err(err) => { - tracing::warn!( - "Failed to make/get credential with NFC authenticator: {:?}", - err - ); - break Err(err); + + // Un-awaited async block + let wait_for_cred_response_fut = async { + loop { + let response = match cred_request { + CredentialRequest::CreatePublicKeyCredentialRequest(make_cred_request) => { + channel + .webauthn_make_credential(make_cred_request) + .await + .map(|response| { + NfcUvMessage::ReceivedCredentials(Box::new(response.into())) + }) + } + CredentialRequest::GetPublicKeyCredentialRequest(get_cred_request) => { + channel + .webauthn_get_assertion(get_cred_request) + .await + .map(|response| { + NfcUvMessage::ReceivedCredentials(Box::new(response.into())) + }) + } + }; + match response { + Ok(response) => { + tracing::debug!("Received credential from NFC authenticator"); + break Ok(response); + } + Err(WebAuthnError::Ctap(ctap_error)) + if ctap_error.is_retryable_user_error() => + { + warn!("Retrying WebAuthn credential operation"); + continue; + } + Err(err) => { + tracing::warn!( + "Failed to make/get credential with NFC authenticator: {:?}", + err + ); + break Err(err); + } } } - } - .map_err(|err| match err { - WebAuthnError::Ctap(CtapError::PINAuthBlocked) => Error::PinAttemptsExhausted, - WebAuthnError::Ctap(CtapError::NoCredentials) => Error::NoCredentials, - WebAuthnError::Ctap(CtapError::CredentialExcluded) => Error::CredentialExcluded, - _ => Error::AuthenticatorError, - }); + .map_err(|err| match err { + WebAuthnError::Ctap(CtapError::PINAuthBlocked) => Error::PinAttemptsExhausted, + WebAuthnError::Ctap(CtapError::NoCredentials) => Error::NoCredentials, + WebAuthnError::Ctap(CtapError::CredentialExcluded) => Error::CredentialExcluded, + _ => Error::AuthenticatorError, + }) + }; + + let response = match cancellation + .run_until_cancelled(wait_for_cred_response_fut) + .await + { + Some(resp) => resp, + None => { + tracing::debug!("NFC ceremony cancelled, stopping authenticator operation"); + // Unlike USB, NfcChannelHandle::cancel_ongoing_operation() is a no-op + // because libwebauthn drops _handle_rx in NfcChannel::new(). Cancellation + // takes effect at the next inter-APDU .await point when the future is + // dropped. + Err(Error::Internal("Request cancelled".to_string())) + } + }; + if let Err(err) = signal_tx.send(response).await { tracing::error!("Failed to notify that ceremony completed: {:?}", err); } @@ -284,13 +343,14 @@ impl NfcHandler for InProcessNfcHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Send + Sized + Unpin + 'static { let request = request.clone(); let (tx, mut rx) = mpsc::channel(32); tokio::spawn(async move { // TODO: instead of logging error here, push the errors into the // stream so credential service can handle/forward them to the UI - if let Err(err) = InProcessNfcHandler::process(tx, request).await { + if let Err(err) = InProcessNfcHandler::process(tx, request, cancellation).await { tracing::error!("Error getting credential from NFC: {:?}", err); } }); diff --git a/credentialsd/src/credential_service/usb.rs b/credentialsd/src/credential_service/usb.rs index 3ea66f07..5406b247 100644 --- a/credentialsd/src/credential_service/usb.rs +++ b/credentialsd/src/credential_service/usb.rs @@ -18,6 +18,7 @@ use tokio::sync::{ Mutex as AsyncMutex, broadcast, mpsc::{self, Receiver, Sender, WeakSender}, }; +use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use credentialsd_common::model::{BackgroundEvent, Credential, Error, PinNotSetError}; @@ -30,6 +31,7 @@ pub(crate) trait UsbHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Send + Sized + Unpin + 'static; } @@ -40,13 +42,22 @@ impl InProcessUsbHandler { async fn process_idle_waiting( failures: &mut usize, prev_usb_state: &UsbStateInternal, + cancellation: &CancellationToken, ) -> Result { - match libwebauthn::transport::hid::list_devices().await { + let list_device_fut = libwebauthn::transport::hid::list_devices(); + let Some(result) = cancellation.run_until_cancelled(list_device_fut).await else { + tracing::debug!("USB idle polling cancelled"); + // TODO: We should introduce a cancelled-error variant and return this here, + // so we can differentiate between internal errors, cancellation by user + // and cancellation because other transfers finished + return Err(Error::Internal("Request cancelled".to_string())); + }; + + match result { Ok(hid_devices) => { if hid_devices.is_empty() { - tokio::time::sleep(Duration::from_millis(50)).await; - let state = UsbStateInternal::Waiting; - Ok(state) + super::cancellable_sleep(Duration::from_millis(50), cancellation).await?; + Ok(UsbStateInternal::Waiting) } else { Ok(UsbStateInternal::SelectingDevice(hid_devices)) } @@ -63,7 +74,7 @@ impl InProcessUsbHandler { "Failed to list USB authenticators: {:?}. Throttling USB state updates", err ); - tokio::time::sleep(Duration::from_secs(1)).await; + super::cancellable_sleep(Duration::from_secs(1), cancellation).await?; Ok(prev_usb_state.clone()) } } @@ -72,6 +83,7 @@ impl InProcessUsbHandler { async fn process_selecting_device( hid_devices: &[HidDevice], + cancellation: &CancellationToken, ) -> Result { let expected_answers = hid_devices.len(); let (blinking_tx, mut blinking_rx) = @@ -125,7 +137,26 @@ impl InProcessUsbHandler { tracing::info!("Waiting for user interaction"); drop(blinking_tx); let mut state = UsbStateInternal::Idle; - while let Some(msg) = blinking_rx.recv().await { + + loop { + let maybe_msg_fut = blinking_rx.recv(); + let Some(maybe_msg) = cancellation.run_until_cancelled(maybe_msg_fut).await else { + // The request was cancelled (e.g. another transport completed, or + // the user cancelled). Stop all blinking devices. This interrupts + // the blocking HID read within the transport (≤100ms) and sends a + // CTAP CANCEL frame to each device. + tracing::debug!("USB device selection cancelled"); + for (_key, (device, handle)) in channel_map.into_iter() { + tracing::info!("Cancelling blinking device {device:?}."); + handle.cancel_ongoing_operation().await; + } + return Err(Error::Internal("Request cancelled".to_string())); + }; + + let Some(msg) = maybe_msg else { + // All blink tasks finished without a selection. + break; + }; match msg { Some(idx) => { let (device, _handle) = channel_map.remove(&idx).unwrap(); @@ -242,6 +273,7 @@ impl InProcessUsbHandler { async fn process( tx: Sender, cred_request: CredentialRequest, + cancellation: CancellationToken, ) -> Result<(), Error> { let mut state = UsbStateInternal::Idle; let (signal_tx, mut signal_rx) = mpsc::channel(256); @@ -253,41 +285,58 @@ impl InProcessUsbHandler { loop { tracing::trace!("current usb state: {:?}", state); let prev_usb_state = state; - let next_usb_state = match prev_usb_state { - UsbStateInternal::Idle | UsbStateInternal::Waiting => { - Self::process_idle_waiting(&mut failures, &prev_usb_state).await - } - UsbStateInternal::SelectingDevice(ref hid_devices) => { - Self::process_selecting_device(hid_devices.as_slice()).await - } - UsbStateInternal::Connected(ref device) => { - let device = std::sync::Arc::clone(device); - let signal_tx2 = signal_tx.clone(); - let cred_request = cred_request.clone(); - tokio::spawn(async move { - handle_events(&cred_request, device.clone(), &signal_tx2).await; - }); - Self::process_user_interaction(&mut signal_rx, &cred_tx).await - } - UsbStateInternal::NeedsPin { .. } - | UsbStateInternal::PinNotSet { .. } - | UsbStateInternal::NeedsUserVerification { .. } - | UsbStateInternal::NeedsUserPresence => { - Self::process_user_interaction(&mut signal_rx, &cred_tx).await + let select_next_usb_state_fut = async { + match prev_usb_state { + UsbStateInternal::Idle | UsbStateInternal::Waiting => { + Self::process_idle_waiting(&mut failures, &prev_usb_state, &cancellation) + .await + } + UsbStateInternal::SelectingDevice(ref hid_devices) => { + Self::process_selecting_device(hid_devices.as_slice(), &cancellation).await + } + UsbStateInternal::Connected(ref device) => { + let device = std::sync::Arc::clone(device); + let signal_tx2 = signal_tx.clone(); + let cred_request = cred_request.clone(); + let cancellation = cancellation.clone(); + tokio::spawn(async move { + handle_events(&cred_request, device.clone(), &signal_tx2, cancellation) + .await; + }); + Self::process_user_interaction(&mut signal_rx, &cred_tx).await + } + UsbStateInternal::NeedsPin { .. } + | UsbStateInternal::PinNotSet { .. } + | UsbStateInternal::NeedsUserVerification { .. } + | UsbStateInternal::NeedsUserPresence => { + Self::process_user_interaction(&mut signal_rx, &cred_tx).await + } + UsbStateInternal::SelectCredential { + ref response, + cred_tx: _, + } => Self::process_select_credential(response, &mut cred_rx).await, + // Terminal states - preserve state unchanged, will break loop after sending + UsbStateInternal::Completed(_) | UsbStateInternal::Failed(_) => { + Ok(prev_usb_state.clone()) + } } - UsbStateInternal::SelectCredential { - ref response, - cred_tx: _, - } => Self::process_select_credential(response, &mut cred_rx).await, - UsbStateInternal::Completed(_) => break Ok(()), - UsbStateInternal::Failed(err) => break Err(err), }; + + let Some(next_usb_state) = cancellation + .run_until_cancelled(select_next_usb_state_fut) + .await + else { + tracing::debug!("USB handler cancelled, stopping processing"); + break Err(Error::Internal("Request cancelled".to_string())); + }; + state = next_usb_state.unwrap_or_else(UsbStateInternal::Failed); - // Usually, comparing the Discrimimant is enough, but PinNotSet can be - // repeated multiple times with different or the same error reasons - // (PIN too short, PIN too long, etc.) + // Usually, comparing the discriminant is enough, but PinNotSet/NeedsPin + // can be repeated multiple times with different or the same error reasons + // (PIN wrong, PIN too short, PIN too long, etc.) let state_changed = match (&state, &prev_usb_state) { (UsbStateInternal::PinNotSet { .. }, UsbStateInternal::PinNotSet { .. }) => true, + (UsbStateInternal::NeedsPin { .. }, UsbStateInternal::NeedsPin { .. }) => true, (new_state, old_state) => { std::mem::discriminant(new_state) != std::mem::discriminant(old_state) } @@ -298,6 +347,13 @@ impl InProcessUsbHandler { Error::Internal("USB state channel receiver closed prematurely".to_string()) })?; } + + // Check for terminal states AFTER sending + match state { + UsbStateInternal::Completed(_) => break Ok(()), + UsbStateInternal::Failed(err) => break Err(err), + _ => {} + } } } } @@ -306,6 +362,7 @@ async fn handle_events( cred_request: &CredentialRequest, device: Arc>, signal_tx: &Sender>, + cancellation: CancellationToken, ) { let mut device = device.lock().await; let device_debug = device.to_string(); @@ -322,6 +379,7 @@ async fn handle_events( ); } Ok(mut channel) => { + let cancel_handle = channel.get_handle(); let signal_tx2 = signal_tx.clone().downgrade(); let ux_updates_rx = channel.get_ux_update_receiver(); tokio::spawn(async move { @@ -332,49 +390,68 @@ async fn handle_events( "Polling for credential from USB authenticator {}", &device_debug ); - let response: Result = loop { - let response = match cred_request { - CredentialRequest::CreatePublicKeyCredentialRequest(make_cred_request) => { - channel - .webauthn_make_credential(make_cred_request) - .await - .map(|response| { - UsbUvMessage::ReceivedCredentials(Box::new(response.into())) - }) - } - CredentialRequest::GetPublicKeyCredentialRequest(get_cred_request) => channel - .webauthn_get_assertion(get_cred_request) - .await - .map(|response| { - UsbUvMessage::ReceivedCredentials(Box::new(response.into())) - }), - }; - match response { - Ok(response) => { - tracing::debug!("Received credential from USB authenticator"); - break Ok(response); - } - Err(WebAuthnError::Ctap(ctap_error)) - if ctap_error.is_retryable_user_error() => - { - warn!("Retrying WebAuthn credential operation"); - continue; - } - Err(err) => { - tracing::warn!( - "Failed to make/get credential with USB authenticator: {:?}", - err - ); - break Err(err); + + // Un-await-ed async-block + let wait_for_cred_response_fut = async { + loop { + let response = match cred_request { + CredentialRequest::CreatePublicKeyCredentialRequest(make_cred_request) => { + channel + .webauthn_make_credential(make_cred_request) + .await + .map(|response| { + UsbUvMessage::ReceivedCredentials(Box::new(response.into())) + }) + } + CredentialRequest::GetPublicKeyCredentialRequest(get_cred_request) => { + channel + .webauthn_get_assertion(get_cred_request) + .await + .map(|response| { + UsbUvMessage::ReceivedCredentials(Box::new(response.into())) + }) + } + }; + match response { + Ok(response) => { + tracing::debug!("Received credential from USB authenticator"); + break Ok(response); + } + Err(WebAuthnError::Ctap(ctap_error)) + if ctap_error.is_retryable_user_error() => + { + warn!("Retrying WebAuthn credential operation"); + continue; + } + Err(err) => { + tracing::warn!( + "Failed to make/get credential with USB authenticator: {:?}", + err + ); + break Err(err); + } } } - } - .map_err(|err| match err { - WebAuthnError::Ctap(CtapError::PINAuthBlocked) => Error::PinAttemptsExhausted, - WebAuthnError::Ctap(CtapError::NoCredentials) => Error::NoCredentials, - WebAuthnError::Ctap(CtapError::CredentialExcluded) => Error::CredentialExcluded, - _ => Error::AuthenticatorError, - }); + .map_err(|err| match err { + WebAuthnError::Ctap(CtapError::PINAuthBlocked) => Error::PinAttemptsExhausted, + WebAuthnError::Ctap(CtapError::NoCredentials) => Error::NoCredentials, + WebAuthnError::Ctap(CtapError::CredentialExcluded) => Error::CredentialExcluded, + _ => Error::AuthenticatorError, + }) + }; + + let response = match cancellation + .run_until_cancelled(wait_for_cred_response_fut) + .await + { + Some(resp) => resp, + None => { + tracing::debug!("USB ceremony cancelled, interrupting authenticator operation"); + cancel_handle.cancel_ongoing_operation().await; + Err(Error::Internal("Request cancelled".to_string())) + } + }; + if let Err(err) = signal_tx.send(response).await { tracing::error!("Failed to notify that ceremony completed: {:?}", err); } @@ -386,13 +463,12 @@ impl UsbHandler for InProcessUsbHandler { fn start( &self, request: &CredentialRequest, + cancellation: CancellationToken, ) -> impl Stream + Send + Sized + Unpin + 'static { let request = request.clone(); let (tx, mut rx) = mpsc::channel(32); tokio::spawn(async move { - // TODO: instead of logging error here, push the errors into the - // stream so credential service can handle/forward them to the UI - if let Err(err) = InProcessUsbHandler::process(tx, request).await { + if let Err(err) = InProcessUsbHandler::process(tx, request, cancellation).await { tracing::error!("Error getting credential from USB: {:?}", err); } }); diff --git a/credentialsd/src/dbus/flow_control.rs b/credentialsd/src/dbus/flow_control.rs index f196a811..9c69b361 100644 --- a/credentialsd/src/dbus/flow_control.rs +++ b/credentialsd/src/dbus/flow_control.rs @@ -95,7 +95,7 @@ async fn handle, ) -> Result { let (request_tx, request_rx) = oneshot::channel(); - let request_id = svc.lock().await.init_request(&msg, request_tx).await?; + let (request_id, cancellation_token) = svc.lock().await.init_request(&msg, request_tx).await?; let operation = msg.operation(); let rp_id = msg.relying_party_id().to_string(); @@ -146,126 +146,147 @@ 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)); - while let Some(ui_request) = flow.receive_ui_event().await { - match ui_request { - UserInteractedEvent::DiscoveryRequested => { - let client_pin_tx = client_pin_tx.clone(); - let set_pin_tx = set_pin_tx.clone(); - let cred_selector_tx = cred_selector_tx.clone(); - let stream = - svc.lock() - .await - .start_discovery() - .await - .map(move |device_update| { - match &device_update { - DeviceStateUpdate::Nfc(NfcState::NeedsPin { - pin_tx, .. - }) => { - *client_pin_tx.lock().unwrap() = Some(pin_tx.clone()); - } - DeviceStateUpdate::Usb(UsbState::NeedsPin { - pin_tx, .. - }) => { - *client_pin_tx.lock().unwrap() = Some(pin_tx.clone()); - } - DeviceStateUpdate::Nfc(NfcState::PinNotSet { - pin_tx, .. - }) => { - *set_pin_tx.lock().unwrap() = Some(pin_tx.clone()); - } + let wait_for_ui_request_fut = async { + loop { + let Some(ui_request) = flow.receive_ui_event().await else { + tracing::debug!("UI event stream closed"); + break; + }; + match ui_request { + UserInteractedEvent::DiscoveryRequested => { + let client_pin_tx = client_pin_tx.clone(); + let set_pin_tx = set_pin_tx.clone(); + let cred_selector_tx = cred_selector_tx.clone(); + let stream = + svc.lock() + .await + .start_discovery() + .await + .map(move |device_update| { + match &device_update { + DeviceStateUpdate::Nfc(NfcState::NeedsPin { + pin_tx, + .. + }) => { + *client_pin_tx.lock().unwrap() = Some(pin_tx.clone()); + } - DeviceStateUpdate::Usb(UsbState::PinNotSet { - pin_tx, .. - }) => { - *set_pin_tx.lock().unwrap() = Some(pin_tx.clone()); - } - DeviceStateUpdate::Usb(UsbState::SelectingCredential { - cred_tx, - .. - }) => { - *cred_selector_tx.lock().unwrap() = Some(cred_tx.clone()); + DeviceStateUpdate::Usb(UsbState::NeedsPin { + pin_tx, + .. + }) => { + *client_pin_tx.lock().unwrap() = Some(pin_tx.clone()); + } + DeviceStateUpdate::Nfc(NfcState::PinNotSet { + pin_tx, + .. + }) => { + *set_pin_tx.lock().unwrap() = Some(pin_tx.clone()); + } + + DeviceStateUpdate::Usb(UsbState::PinNotSet { + pin_tx, + .. + }) => { + *set_pin_tx.lock().unwrap() = Some(pin_tx.clone()); + } + DeviceStateUpdate::Usb(UsbState::SelectingCredential { + cred_tx, + .. + }) => { + *cred_selector_tx.lock().unwrap() = + Some(cred_tx.clone()); + } + _ => {} } - _ => {} - } - device_update.into() - }); - let flow = flow.clone(); - forward_background_event_stream(flow, stream); - } - UserInteractedEvent::ClientPinEntered(pin_fd) => { - let pin_fd = OwnedFd::from(pin_fd); - let pin = match read_secret(pin_fd) - .map_err(|err| format!("Could not read from file descriptor: {err}")) - .and_then(|bytes| { - String::from_utf8(bytes).map_err(|err| { - format!("Invalid UTF-8 data retrieved from pin: {err}") - }) - }) { - Ok(pin) => pin, - // TODO: need to send an error to the UI, cancel the request and terminate the loop. - Err(err) => { - tracing::error!(%err, "Failed to read client PIN. Stopping event loop. TODO: cancel the request"); - break; - } - }; - let tx = { client_pin_tx.lock().unwrap().take() }; - if let Some(tx) = tx { - if tx.send(pin).await.is_err() { - tracing::error!("Failed to send client PIN to device"); - } - } else { - tracing::error!( - "Invalid state: received a client PIN with no pending request." - ); + device_update.into() + }); + let flow = flow.clone(); + forward_background_event_stream(flow, stream); } - } - UserInteractedEvent::SetDevicePin(pin_fd) => { - let pin_fd = OwnedFd::from(pin_fd); - let pin = match read_secret(pin_fd) - .map_err(|err| format!("Could not read from file descriptor: {err}")) - .and_then(|bytes| { - String::from_utf8(bytes).map_err(|err| { - format!("Invalid UTF-8 data retrieved from pin: {err}") - }) - }) { - Ok(pin) => pin, - // TODO: need to send an error to the UI, cancel the request and terminate the loop. - Err(err) => { - tracing::error!(%err, "Failed to read new device PIN. Stopping event loop. TODO: cancel the request"); - break; + UserInteractedEvent::ClientPinEntered(pin_fd) => { + let pin_fd = OwnedFd::from(pin_fd); + let pin = match read_secret(pin_fd) + .map_err(|err| format!("Could not read from file descriptor: {err}")) + .and_then(|bytes| { + String::from_utf8(bytes).map_err(|err| { + format!("Invalid UTF-8 data retrieved from pin: {err}") + }) + }) { + Ok(pin) => pin, + // TODO: need to send an error to the UI, cancel the request and terminate the loop. + Err(err) => { + tracing::error!(%err, "Failed to read client PIN. Stopping event loop. TODO: cancel the request"); + break; + } + }; + let tx = { client_pin_tx.lock().unwrap().take() }; + if let Some(tx) = tx { + if tx.send(pin).await.is_err() { + tracing::error!("Failed to send client PIN to device"); + } + } else { + tracing::error!( + "Invalid state: received a client PIN with no pending request." + ); } - }; - let tx = { set_pin_tx.lock().unwrap().take() }; - if let Some(tx) = tx { - if tx.send(pin).await.is_err() { - tracing::error!("Failed to send client PIN to device"); + } + UserInteractedEvent::SetDevicePin(pin_fd) => { + let pin_fd = OwnedFd::from(pin_fd); + let pin = match read_secret(pin_fd) + .map_err(|err| format!("Could not read from file descriptor: {err}")) + .and_then(|bytes| { + String::from_utf8(bytes).map_err(|err| { + format!("Invalid UTF-8 data retrieved from pin: {err}") + }) + }) { + Ok(pin) => pin, + // TODO: need to send an error to the UI, cancel the request and terminate the loop. + Err(err) => { + tracing::error!(%err, "Failed to read new device PIN. Stopping event loop. TODO: cancel the request"); + break; + } + }; + let tx = { set_pin_tx.lock().unwrap().take() }; + if let Some(tx) = tx { + if tx.send(pin).await.is_err() { + tracing::error!("Failed to send client PIN to device"); + } + } else { + tracing::error!( + "Invalid state: received a client PIN with no pending request." + ); } - } else { - tracing::error!( - "Invalid state: received a client PIN with no pending request." - ); } - } - UserInteractedEvent::CredentialSelected(id) => { - let tx = { cred_selector_tx.lock().unwrap().take() }; - if let Some(tx) = tx { - if tx.send(id).await.is_err() { - tracing::error!("Failed to send credential selection to device"); + UserInteractedEvent::CredentialSelected(id) => { + let tx = { cred_selector_tx.lock().unwrap().take() }; + if let Some(tx) = tx { + if tx.send(id).await.is_err() { + tracing::error!("Failed to send credential selection to device"); + } + } else { + tracing::error!( + "Invalid state: received a credential selection ID with no pending request." + ); } - } else { - tracing::error!( - "Invalid state: received a credential selection ID with no pending request." - ); } - } - UserInteractedEvent::RequestCancelled => { - tracing::debug!(%request_id, "Cancelling request"); - svc.lock().await.cancel_request(request_id).await; + UserInteractedEvent::RequestCancelled => { + tracing::debug!(%request_id, "Cancelling request"); + svc.lock().await.cancel_request(request_id).await; + break; + } } } - } + }; + + let Some(_) = cancellation_token + .run_until_cancelled(wait_for_ui_request_fut) + .await + else { + tracing::debug!("Request cancelled, stopping UI event handler"); + return; + }; }); tracing::debug!("Finished setting up request {request_id}"); @@ -286,6 +307,7 @@ fn forward_background_event_stream( break; } } + tracing::debug!("Background event stream ended"); }); }