daemon: Bound concurrent credential discovery

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]

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<M: ManageDevice + Debug + Send + Sync + 'static, UC: UiControlle
         let client_pin_tx: Arc<Mutex<Option<Sender<String>>>> = Arc::new(Mutex::new(None));
         let set_pin_tx: Arc<Mutex<Option<Sender<String>>>> = 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<M: ManageDevice + Debug + Send + Sync + 'static, UC: UiControlle
                 };
                 match ui_request {
                     UserInteractedEvent::DiscoveryRequested => {
+                        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<M: ManageDevice + Debug + Send + Sync + 'static, UC: UiControlle
                                     device_update.into()
                                 });
                         let flow = flow.clone();
-                        forward_background_event_stream(flow, stream);
+                        forward_background_event_stream(
+                            move |event| {
+                                let flow = flow.clone();
+                                async move { flow.send_state_update(event).await }
+                            },
+                            stream,
+                            discovery_permit,
+                        );
                     }
                     UserInteractedEvent::ClientPinEntered(pin_fd) => {
                         let pin_fd = OwnedFd::from(pin_fd);
@@ -295,13 +310,25 @@ async fn handle<M: ManageDevice + Debug + Send + Sync + 'static, UC: UiControlle
         .expect("Credential service not to drop request channel before responding.")
 }
 
-fn forward_background_event_stream(
-    flow: Ceremony,
+/// Claims the single active discovery slot for a credential request.
+///
+/// The semaphore must be scoped to one request, and the returned permit must be
+/// held until its discovery stream ends. D-Bus sender and session validation is
+/// handled before this point; this additional bound protects the daemon from a
+/// faulty trusted UI without preventing a later discovery attempt.
+fn claim_discovery(discovery_gate: &Arc<Semaphore>) -> Option<OwnedSemaphorePermit> {
+    discovery_gate.clone().try_acquire_owned().ok()
+}
+
+fn forward_background_event_stream<F: Future<Output = Result<(), ()>> + Send>(
+    mut send_state_update: impl FnMut(BackgroundEvent) -> F + Send + 'static,
     mut stream: impl Stream<Item = BackgroundEvent> + 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]