From 39e183755c04bd7a938beaf6379cf3085c262d60 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 14 Aug 2026 09:11:03 +0300 Subject: [PATCH] chore: files changed crates/tinybus/src/broker.rs,crates/tinybus/src/connection.rs,crates/tinybus/sr Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/broker.rs | 48 ++++++++++++++++++++ crates/tinybus/src/connection.rs | 13 ++++++ crates/tinybus/src/stream/mod.rs | 58 ++++++++++++++++++++++++ crates/tinybus/src/stream/stream_test.rs | 41 +++++++++++++++++ docs/modules/attest/README.md | 31 +++++++++++-- docs/protocol.md | 15 ++++-- 6 files changed, 200 insertions(+), 6 deletions(-) diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 124111d..0b3cb09 100644 --- a/crates/tinybus/src/broker.rs +++ b/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::("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::( + "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; diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index fa4f9a4..b3e76a9 100644 --- a/crates/tinybus/src/connection.rs +++ b/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(); diff --git a/crates/tinybus/src/stream/mod.rs b/crates/tinybus/src/stream/mod.rs index 670b212..2d81f41 100644 --- a/crates/tinybus/src/stream/mod.rs +++ b/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, + #[serde(default)] + #[allow(dead_code)] + len: Option, + } + + match body { + Value::Object(_) => { + if serde_json::from_value::(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; diff --git a/crates/tinybus/src/stream/stream_test.rs b/crates/tinybus/src/stream/stream_test.rs index 941ca83..b5fd196 100644 --- a/crates/tinybus/src/stream/stream_test.rs +++ b/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)); +} diff --git a/docs/modules/attest/README.md b/docs/modules/attest/README.md index ef560c1..4ed4333 100644 --- a/docs/modules/attest/README.md +++ b/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 diff --git a/docs/protocol.md b/docs/protocol.md index e8a0e69..6d36544 100644 --- a/docs/protocol.md +++ b/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