From 8054b8cc1a4d874bf61288ccb5396cecd1873b27 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 1/2] 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] From 35415ede5f90195ffe121ae8f4408d9cb61f955e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Boug=C3=A9?= Date: Sat, 5 Sep 2026 15:55:00 +0200 Subject: [PATCH 2/2] daemon: Hold discovery permit for the background task lifetime --- CHANGELOG.md | 1 + credentialsd/src/dbus/flow_control.rs | 54 ++++++++++++++++++++++++--- 2 files changed, 49 insertions(+), 6 deletions(-) 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 9af6241..bde1dad 100644 --- a/credentialsd/src/dbus/flow_control.rs +++ b/credentialsd/src/dbus/flow_control.rs @@ -24,6 +24,7 @@ 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, @@ -211,7 +211,14 @@ async fn handle { let pin_fd = OwnedFd::from(pin_fd); @@ -313,14 +320,15 @@ fn claim_discovery(discovery_gate: &Arc) -> Option> + Send>( + mut send_state_update: impl FnMut(BackgroundEvent) -> F + Send + 'static, mut stream: impl Stream + Send + Unpin + 'static, - _discovery_permit: OwnedSemaphorePermit, + 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; @@ -334,6 +342,40 @@ fn forward_background_event_stream( 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));