feat(l7): support JSON-RPC batch calls

Parse JSON-RPC batch arrays into per-call metadata and evaluate each batch item
with the existing method and params policy rules.

Deny the whole batch when any call is denied.

Signed-off-by: Kris Hicks <khicks@nvidia.com>

Author: krishicks

crates/openshell-sandbox/src/l7/jsonrpc.rs

@@ -31,19 +31,15 @@ pub(crate) async fn parse_jsonrpc_http_request<C: AsyncRead + AsyncWrite + Unpin
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct JsonRpcRequestInfo {
-    pub method: Option<String>,
-    pub params: HashMap<String, String>,
+    pub calls: Vec<JsonRpcCallInfo>,
+    pub is_batch: bool,
     pub error: Option<String>,
 }
 
-/// Returns true if the parsed request's method matches the given `rpc_method` rule pattern.
-///
-/// An empty `rpc_method` pattern matches any method.
-pub fn rpc_method_rule_matches(info: &JsonRpcRequestInfo, rpc_method: &str) -> bool {
-    if rpc_method.is_empty() {
-        return true;
-    }
-    info.method.as_deref() == Some(rpc_method)
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct JsonRpcCallInfo {
+    pub method: String,
+    pub params: HashMap<String, String>,
 }
 
 /// Parse a JSON-RPC 2.0 request body and extract the `method` field.
@@ -53,25 +49,60 @@ pub fn rpc_method_rule_matches(info: &JsonRpcRequestInfo, rpc_method: &str) -> b
 pub fn parse_jsonrpc_body(body: &[u8]) -> JsonRpcRequestInfo {
     let Ok(value) = serde_json::from_slice::<serde_json::Value>(body) else {
         return JsonRpcRequestInfo {
-            method: None,
-            params: HashMap::new(),
+            calls: Vec::new(),
+            is_batch: false,
             error: Some("invalid JSON".to_string()),
         };
     };
-    let Some(method) = value.get("method").and_then(|m| m.as_str()) else {
+
+    if let serde_json::Value::Array(items) = value {
+        if items.is_empty() {
+            return JsonRpcRequestInfo {
+                calls: Vec::new(),
+                is_batch: true,
+                error: Some("empty batch".to_string()),
+            };
+        }
+        let mut calls = Vec::new();
+        for item in &items {
+            let Some(call) = parse_jsonrpc_call(item) else {
+                return JsonRpcRequestInfo {
+                    calls: Vec::new(),
+                    is_batch: true,
+                    error: Some("batch item missing or non-string 'method' field".to_string()),
+                };
+            };
+            calls.push(call);
+        }
+        return JsonRpcRequestInfo {
+            calls,
+            is_batch: true,
+            error: None,
+        };
+    }
+
+    let Some(call) = parse_jsonrpc_call(&value) else {
         return JsonRpcRequestInfo {
-            method: None,
-            params: HashMap::new(),
+            calls: Vec::new(),
+            is_batch: false,
             error: Some("missing or non-string 'method' field".to_string()),
         };
     };
     JsonRpcRequestInfo {
-        method: Some(method.to_string()),
+        calls: vec![call],
+        is_batch: false,
+        error: None,
+    }
+}
+
+fn parse_jsonrpc_call(value: &serde_json::Value) -> Option<JsonRpcCallInfo> {
+    let method = value.get("method").and_then(|m| m.as_str())?;
+    Some(JsonRpcCallInfo {
+        method: method.to_string(),
         params: value
             .get("params")
             .map_or_else(HashMap::new, flatten_jsonrpc_params),
-        error: None,
-    }
+    })
 }
 
 fn flatten_jsonrpc_params(value: &serde_json::Value) -> HashMap<String, String> {
@@ -113,39 +144,45 @@ mod tests {
     fn parses_method_from_request_body() {
         let body = br#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#;
         let info = parse_jsonrpc_body(body);
-        assert_eq!(info.method.as_deref(), Some("initialize"));
+        assert_eq!(
+            info.calls.first().map(|call| call.method.as_str()),
+            Some("initialize")
+        );
+        assert_eq!(info.calls.len(), 1);
+        assert!(!info.is_batch);
         assert!(info.error.is_none());
     }
 
     #[test]
     fn flattens_object_params_for_policy_matching() {
         let body = br#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"submit_report","arguments":{"scope":"workspace/main"}}}"#;
         let info = parse_jsonrpc_body(body);
+        let params = &info.calls.first().expect("single request call").params;
         assert_eq!(
-            info.params.get("name").map(String::as_str),
+            params.get("name").map(String::as_str),
             Some("submit_report")
         );
         assert_eq!(
-            info.params.get("arguments.scope").map(String::as_str),
+            params.get("arguments.scope").map(String::as_str),
             Some("workspace/main")
         );
     }
 
     #[test]
-    fn rpc_method_rule_empty_matches_any() {
-        let info = parse_jsonrpc_body(br#"{"jsonrpc":"2.0","id":1,"method":"tools/call"}"#);
-        assert!(rpc_method_rule_matches(&info, ""));
-    }
-
-    #[test]
-    fn rpc_method_rule_matches_exact_method() {
-        let info = parse_jsonrpc_body(br#"{"jsonrpc":"2.0","id":1,"method":"initialize"}"#);
-        assert!(rpc_method_rule_matches(&info, "initialize"));
-    }
-
-    #[test]
-    fn rpc_method_rule_does_not_match_different_method() {
-        let info = parse_jsonrpc_body(br#"{"jsonrpc":"2.0","id":1,"method":"tools/call"}"#);
-        assert!(!rpc_method_rule_matches(&info, "initialize"));
+    fn parses_valid_batch_without_error() {
+        let body = br#"[
+            {"jsonrpc":"2.0","id":1,"method":"tools/list"},
+            {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"read_status"}}
+        ]"#;
+        let info = parse_jsonrpc_body(body);
+        assert!(info.error.is_none());
+        assert!(info.is_batch);
+        assert_eq!(info.calls.len(), 2);
+        assert_eq!(info.calls[0].method, "tools/list");
+        assert_eq!(info.calls[1].method, "tools/call");
+        assert_eq!(
+            info.calls[1].params.get("name").map(String::as_str),
+            Some("read_status")
+        );
     }
 }

crates/openshell-sandbox/src/l7/relay.rs

@@ -981,7 +981,11 @@ where
                     SeverityId::Informational,
                 ),
             };
-            let rpc_method = jsonrpc_info.method.as_deref().unwrap_or("-");
+            let rpc_method = jsonrpc_info
+                .calls
+                .first()
+                .map(|call| call.method.as_str())
+                .unwrap_or("-");
             let event = HttpActivityBuilder::new(crate::ocsf_ctx())
                 .activity(ActivityId::Other)
                 .action(action_id)
@@ -1295,6 +1299,33 @@ pub fn evaluate_l7_request(
     engine: &TunnelPolicyEngine,
     ctx: &L7EvalContext,
     request: &L7RequestInfo,
+) -> Result<(bool, String)> {
+    if let Some(jsonrpc) = &request.jsonrpc
+        && jsonrpc.is_batch
+        && !jsonrpc.calls.is_empty()
+    {
+        for call in &jsonrpc.calls {
+            let mut item_request = request.clone();
+            item_request.jsonrpc = Some(crate::l7::jsonrpc::JsonRpcRequestInfo {
+                calls: vec![call.clone()],
+                is_batch: false,
+                error: None,
+            });
+            let (allowed, reason) = evaluate_l7_request_once(engine, ctx, &item_request)?;
+            if !allowed {
+                return Ok((false, reason));
+            }
+        }
+        return Ok((true, String::new()));
+    }
+
+    evaluate_l7_request_once(engine, ctx, request)
+}
+
+fn evaluate_l7_request_once(
+    engine: &TunnelPolicyEngine,
+    ctx: &L7EvalContext,
+    request: &L7RequestInfo,
 ) -> Result<(bool, String)> {
     if engine.is_stale() {
         return Err(miette!(
@@ -1319,11 +1350,14 @@ pub fn evaluate_l7_request(
             "path": request.target,
             "query_params": request.query_params.clone(),
             "graphql": request.graphql.clone(),
-            "jsonrpc": request.jsonrpc.as_ref().map(|j| serde_json::json!({
-                "method": j.method,
-                "params": j.params,
-                "error": j.error,
-            })),
+            "jsonrpc": request.jsonrpc.as_ref().map(|j| {
+                let call = if j.is_batch { None } else { j.calls.first() };
+                serde_json::json!({
+                    "method": call.map(|call| call.method.as_str()),
+                    "params": call.map(|call| call.params.clone()).unwrap_or_default(),
+                    "error": j.error,
+                })
+            }),
         }
     });
 
@@ -1966,6 +2000,77 @@ network_policies:
         assert!(reason.contains("WEBSOCKET_TEXT /ws not permitted"));
     }
 
+    #[test]
+    fn jsonrpc_batch_evaluates_each_call() {
+        let data = r#"
+network_policies:
+  jsonrpc_api:
+    name: jsonrpc_api
+    endpoints:
+      - host: api.example.test
+        port: 443
+        protocol: json-rpc
+        enforcement: enforce
+        rules:
+          - allow:
+              method: POST
+              path: "/mcp"
+              rpc_method: "tools/list"
+          - allow:
+              method: POST
+              path: "/mcp"
+              rpc_method: "tools/call"
+              params:
+                name: read_status
+        deny_rules:
+          - rpc_method: "tools/call"
+            params:
+              name: blocked_action
+    binaries:
+      - { path: /usr/bin/node }
+"#;
+        let engine = OpaEngine::from_strings(TEST_POLICY, data).unwrap();
+        let tunnel_engine = engine
+            .clone_engine_for_tunnel(engine.current_generation())
+            .unwrap();
+        let ctx = L7EvalContext {
+            host: "api.example.test".into(),
+            port: 443,
+            policy_name: "jsonrpc_api".into(),
+            binary_path: "/usr/bin/node".into(),
+            ancestors: vec![],
+            cmdline_paths: vec![],
+            secret_resolver: None,
+            activity_tx: None,
+            dynamic_credentials: None,
+            token_grant_resolver: None,
+        };
+        let mut request = L7RequestInfo {
+            action: "POST".into(),
+            target: "/mcp".into(),
+            query_params: std::collections::HashMap::new(),
+            graphql: None,
+            jsonrpc: Some(crate::l7::jsonrpc::parse_jsonrpc_body(
+                br#"[
+                    {"jsonrpc":"2.0","id":1,"method":"tools/list"},
+                    {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"read_status"}}
+                ]"#,
+            )),
+        };
+
+        let (allowed, reason) = evaluate_l7_request(&tunnel_engine, &ctx, &request).unwrap();
+        assert!(allowed, "{reason}");
+
+        request.jsonrpc = Some(crate::l7::jsonrpc::parse_jsonrpc_body(
+            br#"[
+                {"jsonrpc":"2.0","id":1,"method":"tools/list"},
+                {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"blocked_action"}}
+            ]"#,
+        ));
+        let (allowed, _) = evaluate_l7_request(&tunnel_engine, &ctx, &request).unwrap();
+        assert!(!allowed);
+    }
+
     #[tokio::test]
     async fn route_selected_websocket_upgrade_rejects_invalid_accept_without_forwarding_101() {
         let data = r#"