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>

Author: shiju-nv

architecture/sandbox.md

@@ -397,7 +397,7 @@ 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.
 
-Chunked MCP and JSON-RPC request inspection consumes exactly one HTTP message before normalizing its body to `Content-Length`. Bytes for a subsequent pipelined request stay in the connection reader and receive a separate policy decision. Chunk extensions and trailers do not become part of the decoded body.
+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.
 

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 } }"}"#;