feat(confidential): refuse a stream handle in a confidential body

crates/tinybus/src/broker.rs

@@ -1290,6 +1290,54 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "modules")]
+    #[tokio::test]
+    async fn a_confidential_call_carrying_a_stream_handle_is_refused_before_it_is_sent() {
+        // The footgun this closes: a stream's bytes travel as their own
+        // unflagged `Write` calls, so a handle inside a confidential body would
+        // attest the recipient of the *handle* while the payload it stands for
+        // went out unattested — and the caller would have every reason to
+        // believe otherwise. Refused in the sender's own process, because the
+        // broker would have to read a confidential body to see it.
+        let (_bus, _broker, _service, client) = attested_bus().await;
+        let voice = client.proxy(VOICE_NAME, VOICE_PATH, VOICE_NAME).unwrap();
+
+        // The recipient really is attested, so the refusal below is about the
+        // stream handle and nothing else.
+        assert!(voice.attestation().await.unwrap().is_some());
+
+        let handle = crate::stream::StreamRef {
+            id: "s1".to_string(),
+            content_type: None,
+            len: Some(4096),
+        };
+        let error = voice
+            .call_confidential::<Value>("Transcribe", (handle,))
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("stream handle"), "{error}");
+
+        // The same handle in a *non*-confidential call is not intercepted: this
+        // guards a confidentiality claim, it does not ban streams. The call
+        // still fails, because this fixture's `Transcribe` takes a string — but
+        // it fails at the service, having been sent, rather than being refused
+        // here. Asserting on which error distinguishes the two.
+        let sent = voice
+            .call::<Value>(
+                "Transcribe",
+                (crate::stream::StreamRef {
+                    id: "s1".to_string(),
+                    content_type: None,
+                    len: Some(4096),
+                },),
+            )
+            .await
+            .unwrap_err();
+        // Only that the guard did not fire — which error the fixture's own
+        // signature mismatch produces downstream is not this test's business.
+        assert!(!sent.to_string().contains("stream handle"), "{sent}");
+    }
+
     #[tokio::test]
     async fn get_attestation_answers_for_a_name_nobody_owns() {
         let (_bus, _service, client) = bus().await;

crates/tinybus/src/connection.rs

@@ -569,6 +569,19 @@ impl Connection {
         let serial = self.inner.serial.fetch_add(1, Ordering::Relaxed);
         message.header.serial = serial;
         message.validate()?;
+        // Refused here, in the sender's own process, rather than at the broker.
+        // A stream's bytes travel as their own unflagged `Write` calls, so a
+        // handle in a confidential body protects the handle and not the payload
+        // — and the broker cannot catch that for us, because seeing the handle
+        // would mean reading a confidential body. Checked only when the flag is
+        // set, so ordinary traffic pays nothing.
+        if message.header.confidential && crate::stream::body_contains_stream_ref(&message.body) {
+            return Err(Error::protocol(
+                "a confidential call cannot carry a stream handle: the stream's bytes \
+                 travel as separate unattested writes, so the payload would not be \
+                 confidential even though the handle was",
+            ));
+        }
         let member = message.member_or_unknown();
 
         let (tx, rx) = oneshot::channel();

crates/tinybus/src/stream/mod.rs

@@ -789,5 +789,63 @@ impl StreamReader {
     }
 }
 
+/// Whether `body` carries a [`StreamRef`] anywhere inside it.
+///
+/// # Why this exists, and why it is not a broker check
+///
+/// A stream's bytes do not travel in the call that carries its handle. They
+/// travel as their own `Write` calls, which are ordinary method calls with no
+/// `confidential` flag on them — so a handle placed in a confidential body
+/// attests the recipient of the *handle*, while the payload it stands for goes
+/// out unattested. A caller who wrote `call_confidential(…, stream_ref)` would
+/// reasonably believe the payload was covered. It is not.
+///
+/// The refusal therefore has to happen in the **sending** peer's own process,
+/// which already owns the body it just built. It deliberately cannot be a
+/// broker rule: spotting a handle means reading the body, and a broker that
+/// read a confidential body would be the very thing confidentiality exists to
+/// prevent. See `docs/modules/attest/README.md`.
+///
+/// # Precision
+///
+/// Matching is structural — an object whose keys are exactly a [`StreamRef`]'s,
+/// with `id` present — and never looks *inside* `id`, which is documented as
+/// opaque and may be minted in any form by any implementation. The cost is that
+/// a bare `{"id": "…"}` in a confidential body is refused even when it was
+/// never a stream handle. That is the direction to be wrong in: the failure is
+/// loud, local, and recoverable by restructuring the call, whereas the
+/// alternative failure is a secret leaving unattested and nobody finding out.
+pub(crate) fn body_contains_stream_ref(body: &Value) -> bool {
+    /// Detection-only mirror of [`StreamRef`]. `deny_unknown_fields` is the
+    /// whole point: it stops every JSON object that merely happens to carry an
+    /// `id` alongside other fields from matching.
+    #[derive(Deserialize)]
+    #[serde(deny_unknown_fields)]
+    struct StreamRefShape {
+        #[allow(dead_code)]
+        id: String,
+        #[serde(default)]
+        #[allow(dead_code)]
+        content_type: Option<String>,
+        #[serde(default)]
+        #[allow(dead_code)]
+        len: Option<u64>,
+    }
+
+    match body {
+        Value::Object(_) => {
+            if serde_json::from_value::<StreamRefShape>(body.clone()).is_ok() {
+                return true;
+            }
+            body.as_object()
+                .expect("matched Value::Object")
+                .values()
+                .any(body_contains_stream_ref)
+        }
+        Value::Array(values) => values.iter().any(body_contains_stream_ref),
+        _ => false,
+    }
+}
+
 #[cfg(test)]
 mod stream_test;

crates/tinybus/src/stream/stream_test.rs

@@ -981,3 +981,44 @@ async fn a_receiver_does_not_reserve_memory_for_a_length_the_sender_merely_claim
         bytes.len()
     );
 }
+
+#[test]
+fn a_stream_handle_is_detected_wherever_it_sits_in_a_body() {
+    use crate::stream::body_contains_stream_ref;
+
+    let handle = serde_json::to_value(StreamRef {
+        id: "s1".to_string(),
+        content_type: Some("application/pdf".to_string()),
+        len: Some(1024),
+    })
+    .unwrap();
+
+    // Bare, nested in the positional argument array a call actually sends, and
+    // buried inside a struct — a caller can put it anywhere, so all of them
+    // have to be found.
+    assert!(body_contains_stream_ref(&handle));
+    assert!(body_contains_stream_ref(&serde_json::json!([handle])));
+    assert!(body_contains_stream_ref(&serde_json::json!([{
+        "attachment": handle,
+        "subject": "invoice"
+    }])));
+    assert!(body_contains_stream_ref(&serde_json::json!({
+        "id": "s7"
+    })));
+}
+
+#[test]
+fn an_ordinary_body_is_not_mistaken_for_a_stream_handle() {
+    use crate::stream::body_contains_stream_ref;
+
+    // An `id` alongside other fields is an ordinary record, not a handle;
+    // `deny_unknown_fields` is what keeps these out.
+    assert!(!body_contains_stream_ref(&serde_json::json!([{
+        "id": "account-1",
+        "balance": 10
+    }])));
+    assert!(!body_contains_stream_ref(&serde_json::json!(["s1"])));
+    assert!(!body_contains_stream_ref(&serde_json::json!([{ "id": 7 }])));
+    assert!(!body_contains_stream_ref(&serde_json::json!([])));
+    assert!(!body_contains_stream_ref(&serde_json::Value::Null));
+}

docs/modules/attest/README.md

@@ -48,9 +48,34 @@ ineligible to receive them.
 
 A second thing it does not cover: **bulk streams**. A stream's bytes move as
 their own `Stream.Write` calls, which carry no `confidential` flag and so are
-routed without this check. Putting a `StreamRef` in a confidential call attests
-the recipient of the *handle*, not of the payload — so a secret large enough to
-want a stream currently has no attested way to travel. See
+routed without this check. Putting a `StreamRef` in a confidential call would
+attest the recipient of the *handle*, not of the payload.
+
+Rather than leave that as a trap, it is **refused**. A confidential call whose
+body carries a stream handle fails before it is sent:
+
+```text
+a confidential call cannot carry a stream handle: the stream's bytes travel as
+separate unattested writes, so the payload would not be confidential even
+though the handle was
+```
+
+The refusal happens in the **sending peer's own process**, not at the broker,
+and that placement is forced: spotting a handle means reading the body, and a
+broker that read a confidential body would be the very thing confidentiality
+exists to prevent. The sender already owns the body it just built, so it is the
+only party that can look without breaking the rule.
+
+Matching is structural — an object whose keys are exactly a `StreamRef`'s, with
+`id` present — and never looks inside `id`, which is documented as opaque. The
+cost is that a bare `{"id": "…"}` in a confidential body is refused even when it
+was never a handle. That is the direction to be wrong in: the failure is loud,
+local, and fixed by restructuring the call, whereas the alternative failure is a
+secret leaving unattested and nobody finding out.
+
+A secret large enough to want a stream therefore still has no attested way to
+travel. Confidential bulk transfer is its own piece of work; what this closes is
+the silent version of the gap. See
 [the protocol's `confidential` section](../../protocol.md#confidential).
 
 A third case worth naming explicitly: a module loaded from a GitHub release

docs/protocol.md

@@ -94,9 +94,18 @@ old to have the method, both mean the guarantee is unavailable.
 not that body. A stream's bytes travel as separate `Stream.Write` calls (see
 [Bulk streams](#bulk-streams)) which carry no `confidential` flag and are
 therefore routed without an attestation check — putting a `StreamRef` in a
-confidential call protects the handle, not the payload it names. There is
-currently no confidential stream; a secret that must be attested has to fit in
-the body of the call itself.
+confidential call would protect the handle, not the payload it names.
+
+A sender **must** therefore refuse to send a confidential message whose body
+carries a stream handle, and tinybus does: the call fails locally, before the
+message leaves the process. This is a rule for *senders*, not for brokers. A
+broker cannot enforce it, because finding a handle means reading the body, and
+reading a confidential body is precisely what the flag forbids — so a broker
+never attempts it and never relies on peers having got it right.
+
+There is still no confidential stream; a secret that must be attested has to fit
+in the body of the call itself. What the refusal removes is the silent version
+of that gap, where a caller believed otherwise.
 
 ## Names