fix(network): preserve pipelined requests after chunked inspection (#3861)

* fix(network): preserve pipelined requests after chunked inspection

Stop chunked MCP and JSON-RPC body reads at each framing boundary so the
connection reader retains the next request for independent inspection.
Keep payload reads bounded by the remaining chunk length and scan framing
lines incrementally.

Cover buffered prefixes, fragmented framing, trailers, and malformed input.
Verify allowed and denied pipelined requests through both relay entry paths.

Signed-off-by: Shiju <shiju@nvidia.com>

* fix(network): share bounded HTTP body inspection with GraphQL

Remove GraphQL's duplicate chunk decoder so all buffered HTTP inspectors
preserve the next request on a persistent connection. Keep GraphQL's
header checks, configured body limit and query classification.

Cover GraphQL trailers and fragmented framing, and exercise subsequent
request authorization for REST, GraphQL, MCP and JSON-RPC through both
persistent relay entry paths.

Signed-off-by: Shiju <shiju@nvidia.com>

---------

Signed-off-by: Shiju <shiju@nvidia.com>

Author: shiju-nv

architecture/sandbox.md

@@ -397,6 +397,8 @@ support `params` matchers; generic JSON-RPC rules match only the method.
 JSON-RPC responses and server-to-client MCP messages on response or SSE streams
 are relayed but are not currently parsed for policy enforcement.
 
+Buffered GraphQL, MCP, and JSON-RPC request inspection shares an HTTP body reader that consumes exactly one chunked message before normalizing its body to `Content-Length`. Chunk extensions and trailers do not become part of the decoded body. REST streams chunked framing with the same message boundary. On persistent inspected connections, bytes for a subsequent pipelined request stay in the connection reader and receive a separate policy decision.
+
 Every `protocol: mcp` endpoint carries a canonical, nonempty `mcp.versions` allowlist drawn from OpenShell's exact revision registry: `2025-03-26`, `2025-06-18`, `2025-11-25`, and `2026-07-28`. Omitting the `mcp` object or its `versions` key resolves immediately to `["2025-11-25"]`. Normalization stores and serializes the materialized allowlist in semantic order, so adding a supported revision never widens a normalized policy. Explicit `mcp: null`, `versions: null`, and `versions: []` are invalid; at protobuf ingress, an empty repeated field means omission because repeated fields do not preserve presence. The registry is a closed set, so duplicate or padded values, unknown dates, and moving aliases such as `draft` or `latest` are rejected. A version names a core protocol revision; separately named extensions have no policy version selector.
 
 For the 2025 revisions, a valid standalone `initialize` proposes a version in its body. Every subsequent HTTP request selects its inspection revision through one `MCP-Protocol-Version` header. An absent header selects the transport's `2025-03-26` compatibility fallback, which is permitted only when the endpoint allowlist contains it. The supervisor does not inspect initialization responses, bind `MCP-Session-Id`, or infer a version from an earlier request. Duplicate, empty, or unsupported version headers receive `400 Bad Request`; a supported revision outside the allowlist receives `403 Forbidden`. The `2025-03-26` profile permits nonempty same-side top-level JSON-RPC batches capped at 64 members. Later profiles prohibit top-level arrays.

crates/openshell-supervisor-network/src/l7/graphql.rs

@@ -3,14 +3,14 @@
 
 //! GraphQL-over-HTTP L7 inspection.
 
-use crate::l7::provider::{BodyLength, L7Provider, L7Request};
+use crate::l7::provider::{L7Provider, L7Request};
 use apollo_parser::Parser;
 use apollo_parser::cst;
-use miette::{IntoDiagnostic, Result, miette};
+use miette::{Result, miette};
 use serde::Serialize;
 use serde_json::Value;
 use std::collections::{HashMap, HashSet};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite};
 
 pub const DEFAULT_MAX_BODY_BYTES: usize = 64 * 1024;
 
@@ -85,7 +85,7 @@ pub(crate) async fn inspect_graphql_request<C: AsyncRead + Unpin>(
 ) -> Result<GraphqlRequestInfo> {
     let header_str = header_str(request)?;
     reject_unsupported_headers(header_str)?;
-    let body = read_body_for_inspection(client, request, max_body_bytes).await?;
+    let body = crate::l7::http::read_body_for_inspection(client, request, max_body_bytes).await?;
     Ok(classify_request(request, &body))
 }
 
@@ -379,195 +379,6 @@ fn unique_persisted_query_id(
     Ok(selected.map(|(_, value)| value))
 }
 
-async fn read_body_for_inspection<C: AsyncRead + Unpin>(
-    client: &mut C,
-    request: &mut L7Request,
-    max_body_bytes: usize,
-) -> Result<Vec<u8>> {
-    let header_end = request
-        .raw_header
-        .windows(4)
-        .position(|w| w == b"\r\n\r\n")
-        .map_or(request.raw_header.len(), |p| p + 4);
-    let overflow = request.raw_header[header_end..].to_vec();
-
-    match request.body_length {
-        BodyLength::None => Ok(Vec::new()),
-        BodyLength::ContentLength(len) => {
-            let len = usize::try_from(len)
-                .map_err(|_| miette!("GraphQL request body length exceeds platform limit"))?;
-            if len > max_body_bytes {
-                return Err(miette!(
-                    "GraphQL request body exceeds {max_body_bytes} byte inspection limit"
-                ));
-            }
-            if overflow.len() > len {
-                return Err(miette!(
-                    "GraphQL request contains more body bytes than Content-Length"
-                ));
-            }
-            let remaining = len - overflow.len();
-            let mut body = overflow;
-            if remaining > 0 {
-                let start = body.len();
-                body.resize(len, 0);
-                client
-                    .read_exact(&mut body[start..])
-                    .await
-                    .into_diagnostic()?;
-            }
-            request.raw_header.truncate(header_end);
-            request.raw_header.extend_from_slice(&body);
-            Ok(body)
-        }
-        BodyLength::Chunked => {
-            let body = read_chunked_body_for_inspection(
-                client,
-                request,
-                header_end,
-                overflow,
-                max_body_bytes,
-            )
-            .await?;
-            normalize_chunked_request_to_content_length(request, header_end, &body)?;
-            Ok(body)
-        }
-    }
-}
-
-fn normalize_chunked_request_to_content_length(
-    request: &mut L7Request,
-    header_end: usize,
-    body: &[u8],
-) -> Result<()> {
-    let header_str = std::str::from_utf8(&request.raw_header[..header_end])
-        .map_err(|_| miette!("GraphQL HTTP headers contain invalid UTF-8"))?;
-    let header_str = header_str
-        .strip_suffix("\r\n\r\n")
-        .ok_or_else(|| miette!("GraphQL HTTP headers missing terminator"))?;
-
-    let mut normalized = Vec::with_capacity(header_str.len() + body.len() + 32);
-    for (idx, line) in header_str.split("\r\n").enumerate() {
-        if idx > 0 {
-            let name = line
-                .split_once(':')
-                .map(|(name, _)| name.trim().to_ascii_lowercase());
-            if matches!(
-                name.as_deref(),
-                Some("transfer-encoding" | "content-length" | "trailer")
-            ) {
-                continue;
-            }
-        }
-        normalized.extend_from_slice(line.as_bytes());
-        normalized.extend_from_slice(b"\r\n");
-    }
-    normalized.extend_from_slice(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes());
-    normalized.extend_from_slice(body);
-
-    request.raw_header = normalized;
-    request.body_length = BodyLength::ContentLength(body.len() as u64);
-    Ok(())
-}
-
-async fn read_chunked_body_for_inspection<C: AsyncRead + Unpin>(
-    client: &mut C,
-    request: &mut L7Request,
-    header_end: usize,
-    overflow: Vec<u8>,
-    max_body_bytes: usize,
-) -> Result<Vec<u8>> {
-    let mut raw = overflow;
-    let mut decoded = Vec::new();
-    let mut pos = 0usize;
-
-    loop {
-        let size_line_end = loop {
-            if let Some(end) = find_crlf(&raw, pos) {
-                break end;
-            }
-            read_more(client, &mut raw, max_body_bytes).await?;
-        };
-        let size_line = std::str::from_utf8(&raw[pos..size_line_end])
-            .into_diagnostic()
-            .map_err(|_| miette!("Invalid UTF-8 in GraphQL chunk-size line"))?;
-        let size_token = size_line
-            .split(';')
-            .next()
-            .map(str::trim)
-            .unwrap_or_default();
-        let chunk_size = usize::from_str_radix(size_token, 16)
-            .into_diagnostic()
-            .map_err(|_| miette!("Invalid GraphQL chunk size token: {size_token:?}"))?;
-        pos = size_line_end + 2;
-
-        if decoded.len().saturating_add(chunk_size) > max_body_bytes {
-            return Err(miette!(
-                "GraphQL request body exceeds {max_body_bytes} byte inspection limit"
-            ));
-        }
-
-        if chunk_size == 0 {
-            loop {
-                let trailer_end = loop {
-                    if let Some(end) = find_crlf(&raw, pos) {
-                        break end;
-                    }
-                    read_more(client, &mut raw, max_body_bytes).await?;
-                };
-                let trailer_line = &raw[pos..trailer_end];
-                pos = trailer_end + 2;
-                if trailer_line.is_empty() {
-                    request.raw_header.truncate(header_end);
-                    request.raw_header.extend_from_slice(&raw[..pos]);
-                    return Ok(decoded);
-                }
-            }
-        }
-
-        let chunk_end = pos
-            .checked_add(chunk_size)
-            .ok_or_else(|| miette!("GraphQL chunk size overflow"))?;
-        let chunk_with_crlf_end = chunk_end
-            .checked_add(2)
-            .ok_or_else(|| miette!("GraphQL chunk size overflow"))?;
-        while raw.len() < chunk_with_crlf_end {
-            read_more(client, &mut raw, max_body_bytes).await?;
-        }
-        decoded.extend_from_slice(&raw[pos..chunk_end]);
-        if raw.get(chunk_end..chunk_with_crlf_end) != Some(&b"\r\n"[..]) {
-            return Err(miette!("GraphQL chunk payload missing terminating CRLF"));
-        }
-        pos = chunk_with_crlf_end;
-    }
-}
-
-async fn read_more<C: AsyncRead + Unpin>(
-    client: &mut C,
-    raw: &mut Vec<u8>,
-    max_body_bytes: usize,
-) -> Result<()> {
-    if raw.len() > max_body_bytes.saturating_mul(2).max(max_body_bytes) {
-        return Err(miette!(
-            "GraphQL chunked request body exceeds inspection framing limit"
-        ));
-    }
-    let mut buf = [0u8; 8192];
-    let n = client.read(&mut buf).await.into_diagnostic()?;
-    if n == 0 {
-        return Err(miette!("GraphQL chunked body ended before terminator"));
-    }
-    raw.extend_from_slice(&buf[..n]);
-    Ok(())
-}
-
-fn find_crlf(buf: &[u8], start: usize) -> Option<usize> {
-    buf.get(start..)?
-        .windows(2)
-        .position(|w| w == b"\r\n")
-        .map(|p| start + p)
-}
-
 fn header_str(request: &L7Request) -> Result<&str> {
     let header_end = request
         .raw_header
@@ -602,6 +413,8 @@ fn reject_unsupported_headers(headers: &str) -> Result<()> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::l7::provider::BodyLength;
+    use tokio::io::AsyncReadExt;
 
     fn request(method: &str, target: &str) -> L7Request {
         L7Request {
@@ -698,6 +511,68 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn chunked_graphql_preserves_pipelined_request() {
+        use tokio::io::{AsyncWriteExt, BufReader};
+
+        let body = r#"{"query":"query Viewer { viewer { login } }"}"#;
+        let next = "GET /next HTTP/1.1\r\nHost: example.com\r\n\r\n";
+        for capacity in [1, 8192] {
+            for trailers in ["", "X-Sig: ignored\r\n"] {
+                let wire = format!(
+                    "POST /graphql HTTP/1.1\r\nHost: example.com\r\nContent-Type: application/json\r\nTransfer-Encoding: chunked\r\n\r\n{:x};ext=yes\r\n{body}\r\n0\r\n{trailers}\r\n{next}",
+                    body.len()
+                );
+                let (mut sender, receiver) = tokio::io::duplex(8192);
+                sender.write_all(wire.as_bytes()).await.unwrap();
+                sender.shutdown().await.unwrap();
+                // A one-byte buffer splits every CRLF; a large buffer makes
+                // the complete next request available during body inspection.
+                let mut reader = BufReader::with_capacity(capacity, receiver);
+                let parsed = parse_graphql_http_request(
+                    &mut reader,
+                    DEFAULT_MAX_BODY_BYTES,
+                    crate::l7::path::CanonicalizeOptions::default(),
+                )
+                .await
+                .unwrap()
+                .unwrap();
+                assert_eq!(parsed.info.error, None);
+                assert_eq!(parsed.info.operations[0].fields, ["viewer"]);
+                let mut remaining = String::new();
+                reader.read_to_string(&mut remaining).await.unwrap();
+                assert_eq!(
+                    remaining, next,
+                    "GraphQL inspection consumed the next request"
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn graphql_inspection_preserves_configured_body_limit() {
+        let body = br#"{"query":"query { viewer }"}"#;
+        for chunked in [false, true] {
+            let mut req = request("POST", "/graphql");
+            let wire = if chunked {
+                req.body_length = BodyLength::Chunked;
+                format!(
+                    "{:x}\r\n{}\r\n0\r\n\r\n",
+                    body.len(),
+                    std::str::from_utf8(body).unwrap()
+                )
+                .into_bytes()
+            } else {
+                req.body_length = BodyLength::ContentLength(body.len() as u64);
+                body.to_vec()
+            };
+            let error = inspect_graphql_request(&mut wire.as_slice(), &mut req, body.len() - 1)
+                .await
+                .expect_err("the configured body limit must apply to either HTTP framing");
+            assert!(error.to_string().contains("inspection limit"));
+        }
+    }
+
     #[tokio::test]
     async fn chunked_graphql_post_is_normalized_after_inspection() {
         let body = br#"{"query":"query Viewer { viewer { login } }"}"#;