diff --git a/CHANGELOG.md b/CHANGELOG.md index f08ad50..3857710 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Improvements +- daemon: Prevent concurrent discovery streams for the same credential request. - ui: Remove process ID from credential prompt. # 0.3.0 [2026-08-27] diff --git a/credentialsd/src/dbus/flow_control.rs b/credentialsd/src/dbus/flow_control.rs index 61362a2..bde1dad 100644 --- a/credentialsd/src/dbus/flow_control.rs +++ b/credentialsd/src/dbus/flow_control.rs @@ -18,12 +18,13 @@ 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; use crate::dbus::UiControlServiceClient; +use crate::gateway::WebAuthnError; use crate::{ credential_service::UsbState, dbus::ui_control::UiController, @@ -33,7 +34,6 @@ use crate::{ credential_service::{DeviceStateUpdate, ManageDevice, nfc::NfcState}, model::ClientDetails, }; -use crate::{dbus::ui_control::Ceremony, gateway::WebAuthnError}; pub struct UiRequestContext { request: CredentialRequest, @@ -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,14 @@ async fn handle { let pin_fd = OwnedFd::from(pin_fd); @@ -295,13 +310,25 @@ async fn handle) -> Option { + discovery_gate.clone().try_acquire_owned().ok() +} + +fn forward_background_event_stream> + Send>( + mut send_state_update: impl FnMut(BackgroundEvent) -> F + Send + 'static, mut stream: impl Stream + Send + Unpin + 'static, + discovery_permit: OwnedSemaphorePermit, ) { tokio::spawn(async move { + let _discovery_permit = discovery_permit; while let Some(event) = stream.next().await { - let send_result = flow.send_state_update(event).await; + let send_result = send_state_update(event).await; if send_result.is_err() { tracing::error!("Failed to send state update event to backend. Stopping flow"); break; @@ -311,6 +338,63 @@ fn forward_background_event_stream( }); } +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn discovery_slot_is_held_until_background_stream_ends() { + let discovery_gate = Arc::new(Semaphore::new(1)); + let permit = claim_discovery(&discovery_gate).unwrap(); + let (events_tx, mut events_rx) = tokio::sync::mpsc::unbounded_channel(); + let (polled_tx, polled_rx) = oneshot::channel(); + let mut polled_tx = Some(polled_tx); + let stream = futures::stream::poll_fn(move |cx| { + if let Some(tx) = polled_tx.take() { + let _ = tx.send(()); + } + events_rx.poll_recv(cx) + }); + + forward_background_event_stream(|_| async { Ok(()) }, stream, permit); + assert!(claim_discovery(&discovery_gate).is_none()); + tokio::time::timeout(std::time::Duration::from_secs(1), polled_rx) + .await + .expect("background stream should be polled") + .unwrap(); + assert!(claim_discovery(&discovery_gate).is_none()); + + drop(events_tx); + let next_permit = tokio::time::timeout( + std::time::Duration::from_secs(1), + discovery_gate.clone().acquire_owned(), + ) + .await + .expect("ending the stream should release the discovery slot") + .unwrap(); + drop(next_permit); + assert!(claim_discovery(&discovery_gate).is_some()); + } + + #[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]