Merge branch 'fix/1.6.0-mcp-security-findings' into dev

Author: matthew

crates/busbar/src/ingress/tests/tests.rs

@@ -23,6 +23,7 @@ fn minimal_app() -> Arc<App> {
     Arc::new(App {
         mcp_catalogue: Arc::new(crate::mcp::catalogue::Catalogue::default()),
         mcp_sightings: Default::default(),
+        mcp_spent_approvals: Default::default(),
         mcp_pool: Default::default(),
         mcp_servers: Arc::new(Default::default()),
         // Not an MCP server: the plane is absent and the dispatch table empty, which is what every

crates/busbar/src/main.rs

@@ -1154,19 +1154,18 @@ async fn run() {
     // and the A2A job are: a second job against the same registry would double every fetch and race
     // every ledger stamp. It holds the HANDLE, not the app, so a config apply is picked up on the
     // next tick rather than sweeping a generation the operator has already replaced.
-    if !app_handle.load().mcp_catalogue.is_empty() {
-        tracing::info!(
-            tick_secs = crate::trust::sweep::SWEEP_TICK.as_secs(),
-            "mcp: tool-list refresh job started; registered servers are re-hashed on their own \
-             `refresh_ttl:` and quarantined on drift with no operator present"
-        );
-        // Handle intentionally dropped, exactly as the A2A job's is: it runs for the process
-        // lifetime and exits its own loop on the shutdown broadcast.
-        std::mem::drop(crate::trust::sweep::spawn(
-            crate::mcp::connect::RefreshSweeper(app_handle.clone()),
-            shutdown_tx.subscribe(),
-        ));
-    }
+    //
+    // The decision itself lives in `mcp::spawn_refresh_job` rather than inline here, because
+    // `run()` binds real listeners and joins them and so nothing can test a line of it. While this
+    // was inline, the whole battery in `mcp/tests/timer_dispatch_tests.rs` called `refresh_sweep`
+    // by hand, and deleting this block would have failed exactly nothing.
+    //
+    // Handle intentionally dropped, exactly as the A2A job's is: it runs for the process lifetime
+    // and exits its own loop on the shutdown broadcast.
+    std::mem::drop(crate::mcp::spawn_refresh_job(
+        &app_handle,
+        shutdown_tx.subscribe(),
+    ));
 
     // THE A2A RE-VERIFICATION JOB. An approval is a statement about a document at a moment and
     // nothing keeps it true; the pin catches a change only when somebody looks, and this is what
@@ -3971,6 +3970,13 @@ pub(crate) fn build_app_from_config(
             || Arc::new(crate::mcp::client::catalogue::CatalogueCache::new()),
             |p| p.mcp_sightings.clone(),
         ),
+        // CARRIED ACROSS THE APPLY for the same reason, and it is the same class of mistake: an
+        // approval already spent is evidence, not intent, and a config apply that forgot it would
+        // hand every outstanding confirmation back to whoever still holds it.
+        mcp_spent_approvals: prior.map_or_else(
+            || Arc::new(crate::mcp::askstate::SpentAskStates::new()),
+            |p| p.mcp_spent_approvals.clone(),
+        ),
         credential_cache: prior.map_or_else(
             || Arc::new(auth_cache::CredentialCache::new()),
             |p| p.credential_cache.clone(),

crates/busbar/src/mcp/askstate.rs

@@ -134,6 +134,9 @@ pub(crate) enum Rejected {
     WrongRequest,
     /// Sealed under a catalogue generation that is no longer live — an approval moved underneath it.
     WrongGeneration,
+    /// ALREADY REDEEMED. Perfectly valid, for this caller, for this request, inside its window — and
+    /// already spent on the call it was minted to approve. See [`SpentAskStates`].
+    AlreadySpent,
 }
 
 impl Rejected {
@@ -146,6 +149,7 @@ impl Rejected {
             Rejected::WrongPrincipal => "state_wrong_principal",
             Rejected::WrongRequest => "state_wrong_request",
             Rejected::WrongGeneration => "state_wrong_generation",
+            Rejected::AlreadySpent => "state_already_spent",
         }
     }
 }
@@ -291,6 +295,77 @@ pub(crate) fn digest_arguments(arguments: &serde_json::Value) -> String {
     hex::encode(h.finalize())
 }
 
+/// THE SPENT-APPROVAL LEDGER — what makes an approval SINGLE-USE.
+///
+/// ## What the seal could not do on its own
+///
+/// Everything else about this module is a statement the seal itself can carry: who it was minted
+/// for, what request, which round, until when. Single use is the one property that cannot ride
+/// inside the blob, because a caller presenting the identical blob a second time presents an
+/// identical, perfectly valid blob. The only thing that can tell the second presentation from the
+/// first is a RECORD THAT THE FIRST HAPPENED — and until this existed there was none, so an operator
+/// who gated a money-moving tool behind a confirmation got confirm-once-execute-many.
+///
+/// ## Keyed on the nonce, and only the terminal redemption is recorded
+///
+/// The nonce already exists and is already unique per mint (`mrtr`'s multi-round scenario requires
+/// it), so it is the natural handle and nothing new has to be sealed. What is recorded is the ONE
+/// redemption that dispatches: an intermediate round's state is answered with a fresh ask and a
+/// fresh state, so burning it would refuse the ordinary case of a client retrying a request whose
+/// answer it never saw. The spend therefore happens exactly where the exchange COMPLETES.
+///
+/// ## What a restart does to it, and why that is the right trade
+///
+/// This is PROCESS-LOCAL, and deliberately so rather than for want of a durable store.
+///
+/// - The window a restart reopens is bounded by the state's own life: a state that has lapsed is
+///   already refused by [`Sealer::open`], so the most a restart can restore is the unredeemed
+///   remainder of one [`DEFAULT_TTL_SECS`] window. It is not a standing hole; it closes by itself.
+/// - It is not attacker-triggerable. A caller cannot restart the process, and a caller who could
+///   has a larger primitive than double-spending one confirmation.
+/// - The alternative is a durable spent-nonce table, which means a new `busbar_api::Store` method,
+///   which means the plugin ABI — a substantial change to buy the residual, and one this tree is
+///   the wrong place to spend: it is scheduled for deletion and rebuild on the `rmcp` SDK, and the
+///   rebuilt tree gets to decide where its state lives. The BEHAVIOUR is pinned by test either way,
+///   so the decision can be revisited without the property being lost.
+///
+/// A fleet is the same trade one hop out: two nodes sharing a signing key share the seal but not
+/// this ledger, so a redemption on node A does not stop one on node B. That is a real limit and it
+/// is written down here rather than discovered; closing it needs shared state, which is the same
+/// durable-store decision.
+///
+/// ## The size of it
+///
+/// An entry lives at most as long as the state it records, and every call evicts what has lapsed,
+/// so the table holds at most the approvals minted in one TTL window. Minting one costs the caller
+/// a metered, budget-charged round, so the rate is bounded by governance rather than by this map.
+#[derive(Debug, Default)]
+pub(crate) struct SpentAskStates {
+    /// nonce ⇒ the instant after which the entry is meaningless, because the state it records can
+    /// no longer be opened anyway.
+    seen: std::sync::Mutex<std::collections::HashMap<String, u64>>,
+}
+
+impl SpentAskStates {
+    pub(crate) fn new() -> Self {
+        Self::default()
+    }
+
+    /// SPEND this approval. `true` if it had not been spent before; `false` if it had.
+    ///
+    /// Test-and-set under one lock, and that is not an optimisation: a caller that fires two
+    /// redemptions of one approval concurrently is the obvious way to attack a check that reads and
+    /// then writes, and it is the shape the whole gate exists to refuse.
+    pub(crate) fn spend(&self, nonce: &str, expires_at: u64, now: u64) -> bool {
+        // Poison-recovering, like every other request-path lock in this process: the data behind it
+        // is still valid after a panic, and cascading the poison would turn one stray panic into a
+        // gate that refuses every confirmation for the life of the process.
+        let mut seen = self.seen.lock().unwrap_or_else(|e| e.into_inner());
+        seen.retain(|_, expiry| *expiry >= now);
+        seen.insert(nonce.to_string(), expires_at).is_none()
+    }
+}
+
 /// A fresh nonce. `getrandom` is the same fail-closed entropy source key secrets use; a failure is
 /// not survivable here, because a predictable nonce is a `multi-round` scenario that passes by
 /// accident and a replay window that is wider than it looks.

crates/busbar/src/mcp/callerask.rs

@@ -238,20 +238,41 @@ pub(crate) struct Bind<'a> {
     pub(crate) now: u64,
 }
 
-/// THE DECISION. Pure: config, caller input, and a clock. It is not in scope of any upstream
-/// response and cannot be — see the module header.
+/// THE APPROVAL MACHINERY: what mints continuation state, and what remembers it was spent.
+///
+/// One parameter rather than two because they are one thing viewed from either end — a `Sealer` with
+/// no ledger issues approvals nothing retires, which is precisely the defect
+/// [`askstate::SpentAskStates`] exists to close, and a call site that could pass the first without
+/// the second is a call site that can reintroduce it.
+#[derive(Clone, Copy)]
+pub(crate) struct Approvals<'a> {
+    /// Mints and opens the sealed `requestState`. `None` is a deployment with no signing key, which
+    /// refuses to ask at all rather than issue state it could not verify.
+    pub(crate) sealer: Option<&'a Sealer>,
+    /// The approvals already redeemed. See [`askstate::SpentAskStates`].
+    pub(crate) spent: &'a askstate::SpentAskStates,
+}
+
+/// THE DECISION. Config, caller input, a clock — and the ledger of approvals already spent. It is
+/// not in scope of any upstream response and cannot be — see the module header.
 ///
 /// `rounds` is the operator's ordered list of rounds for THIS capability; an empty list means the
 /// capability declares no ask, which is the default and is today's behaviour.
+///
+/// NOT PURE, and only in one place: the arm that COMPLETES an exchange marks the approval it
+/// consumed as spent. That effect lives here rather than at the call site on purpose — a caller that
+/// had to remember to record the spend is a caller that can forget, and the check and the record
+/// have to be one atomic act or two concurrent redemptions both pass it.
 pub(crate) fn decide(
     rounds: &[AskRoundCfg],
     cap: u32,
     caller_capabilities: &serde_json::Value,
     retry: Retry<'_>,
     bind: Bind<'_>,
     args_digest: &str,
-    sealer: Option<&Sealer>,
+    approvals: Approvals<'_>,
 ) -> AskDecision {
+    let Approvals { sealer, spent } = approvals;
     // (1) A capability that declares NO ask never asks, and never accepts state either. Accepting
     // state here would mean busbar verifying a blob it had no reason to have minted.
     if rounds.is_empty() {
@@ -265,6 +286,10 @@ pub(crate) fn decide(
 
     // (2) WHICH ROUND IS THIS? Read from the sealed state, never from a counter busbar holds between
     // requests and never from anything the caller can write. A caller with no state has not started one.
+    // The approval this request is presenting, if it is presenting one: the nonce that identifies it
+    // and the instant past which it could not be opened anyway. Carried down to the completion arm,
+    // which is the only place it is spent.
+    let mut presented: Option<(String, u64)> = None;
     let next_round = match retry.state {
         None => {
             // `mrtr.mdx` client requirements: a client MUST NOT invent state. Responses without one
@@ -290,6 +315,10 @@ pub(crate) fn decide(
             ) {
                 return AskDecision::Refuse(Refusal::StateRejected(e));
             }
+            presented = Some((
+                opened.nonce.clone(),
+                opened.issued_at.saturating_add(opened.ttl_secs),
+            ));
             opened.round.saturating_add(1)
         }
     };
@@ -307,6 +336,20 @@ pub(crate) fn decide(
     // (4) EVERY ROUND ANSWERED ⇒ the exchange is complete and the call proceeds. This is the ONLY
     // arm that dispatches, and reaching it requires a verified state for the last configured round.
     let Some(this_round) = rounds.get(next_round as usize) else {
+        // AND THE APPROVAL IS NOW SPENT. Everything above proves the state was busbar's, was this
+        // caller's, was for this exact request and has not lapsed — all of which is equally true of
+        // the SECOND presentation of a state already redeemed. Without this, an operator who gated a
+        // tool because it moves money got confirm-once, execute-many.
+        //
+        // `rounds` is non-empty here (the empty case returned at step 1), so an exchange can only
+        // complete by presenting state, and `presented` is therefore always `Some`. The `else` is
+        // the fail-closed reading of a shape that cannot occur rather than a case with a meaning.
+        let Some((nonce, expires_at)) = presented else {
+            return AskDecision::Refuse(Refusal::StateRejected(askstate::Rejected::AlreadySpent));
+        };
+        if !spent.spend(&nonce, expires_at, bind.now) {
+            return AskDecision::Refuse(Refusal::StateRejected(askstate::Rejected::AlreadySpent));
+        }
         return AskDecision::Proceed;
     };
 
@@ -428,3 +471,10 @@ fn declared(capabilities: &serde_json::Value, key: &str) -> bool {
 #[cfg(test)]
 #[path = "tests/callerask_tests.rs"]
 mod callerask_tests;
+
+// THE GATE JUDGED FROM OUTSIDE IT: one approval, presented twice, against a real upstream that
+// records what it was told to do. Separate from the file above because nothing in it may reach into
+// the decision — the claim is about what a caller can make happen, not about which arm answered.
+#[cfg(test)]
+#[path = "tests/confirm_once_tests.rs"]
+mod confirm_once_tests;