starknet_transaction_prover,tower_ohttp: OHTTP-unlinkable request-id for decapsulated content

Tags downstream content logs with a request-id via a new `RequestSpanLayer`
placed below the OHTTP layer. For plaintext it reuses the envelope id from
`RequestLogLayer`; for an OHTTP-decapsulated request (marked with a new
`tower_ohttp::Decapsulated` extension) it mints a FRESH UUID and discards any
client-supplied inner id.

The fresh inner id is never echoed back, so the relay-visible envelope id and
the gateway's content-log id cannot be joined — preserving OHTTP unlinkability
while still giving every request's downstream logs a correlatable id.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: avi-starkware

crates/starknet_transaction_prover/src/server.rs

@@ -34,12 +34,14 @@ pub mod log_redact;
 #[cfg(test)]
 pub mod mock_rpc;
 pub mod request_log;
+pub mod request_span;
 pub mod rpc_api;
 pub mod rpc_impl;
 pub mod tls;
 
 pub use health::{HealthLayer, HEALTH_PATH};
 pub use request_log::{RequestLogLayer, REQUEST_ID_HEADER};
+pub use request_span::RequestSpanLayer;
 
 #[cfg(test)]
 mod rpc_spec_test;
@@ -83,13 +85,17 @@ pub async fn start_server(
                 .set_http_middleware(
                     // `RequestLogLayer` is outermost so the latency it measures
                     // covers every other layer. `HealthLayer` sits inside it so
-                    // probes complete before CORS/OHTTP.
+                    // probes complete before CORS/OHTTP. `RequestSpanLayer` sits
+                    // BELOW `OhttpLayer` so it spans the decapsulated inner
+                    // request with a fresh, envelope-unlinkable id (see
+                    // `request_span`).
                     ServiceBuilder::new()
                         .layer(RequestLogLayer)
                         .layer(HealthLayer)
                         .option_layer(cors_layer)
                         .layer(MapRequestBodyLayer::new(HttpBody::new))
                         .option_layer(ohttp_layer)
+                        .layer(RequestSpanLayer)
                         .layer(MapResponseBodyLayer::new(HttpBody::new))
                         .layer(CompressionLayer::new()),
                 )

crates/starknet_transaction_prover/src/server/ohttp_integration_test.rs

@@ -30,11 +30,12 @@ use tower_ohttp::test_utils::{
 };
 use tower_ohttp::OhttpLayer;
 
+use crate::server::request_log::{RequestLogLayer, REQUEST_ID_HEADER};
+use crate::server::request_span::RequestSpanLayer;
+
 const DEFAULT_BODY_LIMIT: usize = 102_400;
 const KEY_CACHE_SECS: u64 = 3600;
 
-/// Body builder for jsonrpsee's `HttpBody`. Returned as a `fn` pointer to
-/// give `OhttpLayer` a sized, `Copy` closure type without an `as` cast.
 fn body_builder() -> fn(Full<bytes::Bytes>) -> HttpBody {
     HttpBody::new
 }
@@ -95,12 +96,15 @@ async fn production_chain_compresses_inner_not_outer() {
     let ohttp_layer =
         OhttpLayer::new(gateway.clone(), DEFAULT_BODY_LIMIT, KEY_CACHE_SECS, body_builder());
 
-    // Replicates the production ServiceBuilder chain from `server.rs`/`tls.rs`.
-    // Must be kept in sync with those files.
+    // Replicates the OHTTP body-handling portion of the production chain in
+    // `server.rs`/`tls.rs` (the outermost observability layers — request log,
+    // health, metrics — don't affect body/compression handling and are
+    // omitted). `RequestSpanLayer` is included since it sits inside OHTTP.
     let mut svc = tower::ServiceBuilder::new()
         .option_layer(None::<CorsLayer>)
         .layer(MapRequestBodyLayer::new(HttpBody::new))
         .option_layer(Some(ohttp_layer))
+        .layer(RequestSpanLayer)
         .layer(MapResponseBodyLayer::new(HttpBody::new))
         .layer(CompressionLayer::new())
         .service(tower::service_fn(jsonrpsee_echo_service));
@@ -176,3 +180,75 @@ async fn non_ohttp_request_passes_through_jsonrpsee() {
     let body = response.into_body().collect().await.unwrap().to_bytes();
     assert_eq!(body.as_ref(), json_body);
 }
+
+/// End-to-end OHTTP unlinkability: the request-id echoed on the OUTER
+/// (relay-visible) response must differ from the fresh id bound to the
+/// decapsulated inner dispatch, and the client-supplied inner id must be
+/// discarded — so no shared key links the relay's view to the gateway's.
+/// Exercises the real decapsulation path through
+/// `RequestLogLayer → OhttpLayer → RequestSpanLayer`.
+#[tokio::test]
+async fn ohttp_inner_request_id_unlinkable_from_envelope() {
+    let gateway = test_gateway();
+    let ohttp_layer =
+        OhttpLayer::new(gateway.clone(), DEFAULT_BODY_LIMIT, KEY_CACHE_SECS, body_builder());
+
+    // Inner service echoes the request-id it observes into the response body.
+    let echo_id = tower::service_fn(|req: http::Request<HttpBody>| async move {
+        let id = req.headers().get(REQUEST_ID_HEADER).map(|v| v.to_str().unwrap()).unwrap_or("");
+        Ok::<_, BoxError>(
+            http::Response::builder()
+                .status(http::StatusCode::OK)
+                .body(HttpBody::from(id.as_bytes().to_vec()))
+                .unwrap(),
+        )
+    });
+
+    let mut svc = tower::ServiceBuilder::new()
+        .layer(RequestLogLayer)
+        .layer(MapRequestBodyLayer::new(HttpBody::new))
+        .option_layer(Some(ohttp_layer))
+        .layer(RequestSpanLayer)
+        .layer(MapResponseBodyLayer::new(HttpBody::new))
+        .service(echo_id);
+
+    // The envelope carries a client-chosen inner id that must be discarded.
+    let (encapsulated, client_response) = encapsulate_bhttp_request(
+        &gateway,
+        "POST",
+        "/",
+        b"",
+        &[("x-request-id", b"inner-client-id")],
+    );
+
+    // The outer envelope request carries the relay-visible id.
+    let mut outer = ohttp_http_request(encapsulated);
+    outer
+        .headers_mut()
+        .insert(REQUEST_ID_HEADER, http::HeaderValue::from_static("envelope-relay-id"));
+
+    let response = svc.call(outer).await.unwrap();
+
+    // The outer (relay-visible) response echoes the envelope id.
+    let envelope_id =
+        response.headers().get(REQUEST_ID_HEADER).unwrap().to_str().unwrap().to_owned();
+    assert_eq!(envelope_id, "envelope-relay-id");
+
+    let encrypted_body = response.into_body().collect().await.unwrap().to_bytes();
+    let decapsulated = decapsulate_bhttp_response(client_response, &encrypted_body);
+    assert_eq!(decapsulated.status, 200);
+    let inner_id = String::from_utf8(decapsulated.body).expect("utf8 inner id");
+
+    assert_ne!(inner_id, envelope_id, "inner id must not equal the relay-visible envelope id");
+    assert_ne!(inner_id, "inner-client-id", "client-supplied inner id must be discarded");
+    assert!(
+        uuid::Uuid::parse_str(&inner_id).is_ok(),
+        "inner id must be a fresh UUID, got {inner_id:?}"
+    );
+    // No id is set on the inner *response*, so nothing — neither the envelope
+    // id nor the fresh content id — leaks into the encrypted reply's headers.
+    assert!(
+        decapsulated.bhttp_message.header().get(b"x-request-id").is_none(),
+        "inner OHTTP response must not carry an x-request-id header"
+    );
+}

crates/starknet_transaction_prover/src/server/request_span.rs

@@ -0,0 +1,75 @@
+//! tower middleware that binds an `http_request` tracing span over the
+//! downstream dispatch. It sits BELOW the OHTTP layer, so it sees the
+//! decapsulated inner request (or a plaintext pass-through), and picks the id:
+//!
+//! - **plaintext** — reuse the `x-request-id` the outer layer already assigned;
+//! - **OHTTP-decapsulated** ([`tower_ohttp::Decapsulated`]) — mint a fresh id, overwriting any
+//!   client-supplied inner id. The relay never observes it (it lives in content logs and, at most,
+//!   the encrypted inner response), so the relay-visible envelope id and this content-log id can't
+//!   be joined.
+//!
+//! See [`super::request_log`] for why the envelope and content ids are kept
+//! separate (OHTTP unlinkability).
+
+use std::task::{Context, Poll};
+
+use http::{HeaderValue, Request, Response};
+use tower::{Layer, Service};
+use tower_ohttp::Decapsulated;
+use tracing::instrument::Instrumented;
+use tracing::{info_span, Instrument};
+
+use crate::server::request_log::{
+    extract_or_generate_request_id,
+    new_request_id,
+    REQUEST_ID_HEADER,
+};
+
+#[cfg(test)]
+#[path = "request_span_test.rs"]
+mod request_span_test;
+
+/// tower [`Layer`] producing [`RequestSpanService`].
+#[derive(Clone, Copy, Default)]
+pub struct RequestSpanLayer;
+
+impl<S> Layer<S> for RequestSpanLayer {
+    type Service = RequestSpanService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        RequestSpanService { inner }
+    }
+}
+
+#[derive(Clone)]
+pub struct RequestSpanService<S> {
+    inner: S,
+}
+
+impl<S, ReqB, RespB> Service<Request<ReqB>> for RequestSpanService<S>
+where
+    S: Service<Request<ReqB>, Response = Response<RespB>>,
+{
+    type Response = Response<RespB>;
+    type Error = S::Error;
+    type Future = Instrumented<S::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut request: Request<ReqB>) -> Self::Future {
+        let request_id = if request.extensions().get::<Decapsulated>().is_some() {
+            // Fresh id, distinct from the relay-visible envelope id (OHTTP unlinkability).
+            new_request_id()
+        } else {
+            // Reuse the id the outer layer assigned; re-validate here so the
+            // layer is self-contained even without `RequestLogLayer` upstream.
+            extract_or_generate_request_id(&request)
+        };
+        if let Ok(header_value) = HeaderValue::from_str(&request_id) {
+            request.headers_mut().insert(REQUEST_ID_HEADER, header_value);
+        }
+        self.inner.call(request).instrument(info_span!("http_request", request_id = %request_id))
+    }
+}

crates/starknet_transaction_prover/src/server/request_span_test.rs

@@ -0,0 +1,68 @@
+use bytes::Bytes;
+use http::{Method, Request, Response, StatusCode};
+use http_body_util::{BodyExt, Full};
+use jsonrpsee::server::HttpBody;
+use tower::{Layer, ServiceExt};
+use tower_ohttp::Decapsulated;
+
+use crate::server::request_log::REQUEST_ID_HEADER;
+use crate::server::request_span::RequestSpanLayer;
+
+/// Inner service that echoes the request's `x-request-id` into the response
+/// body, so tests can observe which id `RequestSpanLayer` bound.
+fn echo_id_service() -> impl tower::Service<
+    Request<HttpBody>,
+    Response = Response<HttpBody>,
+    Error = std::convert::Infallible,
+    Future = futures::future::Ready<Result<Response<HttpBody>, std::convert::Infallible>>,
+> + Clone {
+    tower::service_fn(|request: Request<HttpBody>| {
+        let id = request
+            .headers()
+            .get(REQUEST_ID_HEADER)
+            .map(|value| value.to_str().unwrap().to_owned())
+            .unwrap_or_default();
+        futures::future::ready(Ok::<_, std::convert::Infallible>(
+            Response::builder()
+                .status(StatusCode::OK)
+                .body(HttpBody::new(Full::new(Bytes::from(id))))
+                .expect("static body is infallible"),
+        ))
+    })
+}
+
+async fn body_string(response: Response<HttpBody>) -> String {
+    let bytes = response.into_body().collect().await.expect("body collect").to_bytes().to_vec();
+    String::from_utf8(bytes).expect("utf8 body")
+}
+
+#[tokio::test]
+async fn plaintext_reuses_inbound_request_id() {
+    let request = Request::builder()
+        .method(Method::POST)
+        .uri("/")
+        .header(REQUEST_ID_HEADER, "reused-xyz")
+        .body(HttpBody::new(Full::new(Bytes::new())))
+        .expect("static body is infallible");
+
+    let response = RequestSpanLayer.layer(echo_id_service()).oneshot(request).await.unwrap();
+
+    assert_eq!(body_string(response).await, "reused-xyz");
+}
+
+#[tokio::test]
+async fn decapsulated_gets_fresh_id_discarding_inbound() {
+    let mut request = Request::builder()
+        .method(Method::POST)
+        .uri("/")
+        .header(REQUEST_ID_HEADER, "envelope-abc")
+        .body(HttpBody::new(Full::new(Bytes::new())))
+        .expect("static body is infallible");
+    request.extensions_mut().insert(Decapsulated);
+
+    let response = RequestSpanLayer.layer(echo_id_service()).oneshot(request).await.unwrap();
+
+    let id = body_string(response).await;
+    assert_ne!(id, "envelope-abc", "must discard the client-supplied inner id");
+    assert!(uuid::Uuid::parse_str(&id).is_ok(), "must mint a fresh UUID, got {id:?}");
+}

crates/starknet_transaction_prover/src/server/tls.rs

@@ -27,7 +27,7 @@ use tower_http::map_request_body::MapRequestBodyLayer;
 use tower_http::map_response_body::MapResponseBodyLayer;
 use tracing::warn;
 
-use crate::server::{HealthLayer, OhttpJsonrpseeLayer, RequestLogLayer};
+use crate::server::{HealthLayer, OhttpJsonrpseeLayer, RequestLogLayer, RequestSpanLayer};
 
 /// Maximum time allowed for a TLS handshake before the connection is dropped.
 const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
@@ -68,6 +68,7 @@ pub async fn start_tls_server(
                 .option_layer(cors_layer)
                 .layer(MapRequestBodyLayer::new(HttpBody::new))
                 .option_layer(ohttp_layer)
+                .layer(RequestSpanLayer)
                 .layer(MapResponseBodyLayer::new(HttpBody::new))
                 .layer(CompressionLayer::new()),
         )

crates/tower_ohttp/src/layer.rs

@@ -194,7 +194,11 @@ where
                         OhttpError::InvalidFormat("Invalid Binary HTTP message")
                     })?;
 
-                let inner_request = rebuild_request(&bhttp_message)?.map(build_body);
+                let mut inner_request = rebuild_request(&bhttp_message)?.map(build_body);
+                // Mark the request so downstream layers can tell decapsulated
+                // traffic apart from plaintext pass-through (e.g. to assign a
+                // fresh, envelope-unlinkable request id).
+                inner_request.extensions_mut().insert(crate::Decapsulated);
 
                 inner.oneshot(inner_request).await.map_err(|error| {
                     debug!("Inner service error after successful OHTTP decapsulation: {error:?}");

crates/tower_ohttp/src/lib.rs

@@ -34,6 +34,14 @@ pub use errors::OhttpError;
 pub use gateway::OhttpGateway;
 pub use layer::{OhttpLayer, OhttpService};
 
+/// Marker inserted into the extensions of a request rebuilt from a decapsulated
+/// OHTTP envelope, before it is forwarded to the inner service. Downstream
+/// layers can check for it (`request.extensions().get::<Decapsulated>()`) to
+/// distinguish envelope-decapsulated traffic from plaintext pass-through —
+/// the two are otherwise indistinguishable once the inner request is rebuilt.
+#[derive(Clone, Copy, Debug)]
+pub struct Decapsulated;
+
 pub(crate) const OHTTP_REQUEST_CONTENT_TYPE: &str = "message/ohttp-req";
 pub(crate) const OHTTP_RESPONSE_CONTENT_TYPE: &str = "message/ohttp-res";
 pub(crate) const OHTTP_KEYS_PATH: &str = "/ohttp-keys";