fix(supervisor-middleware): harden policy enforcement

Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>

Author: pimlock

crates/openshell-sandbox/src/lib.rs

@@ -1494,7 +1494,12 @@ async fn load_policy(
         let opa_engine = Some(Arc::new(engine));
 
         let policy = SandboxPolicy::try_from(proto_policy.clone())?;
-        return Ok((policy, opa_engine, Some(proto_policy), middleware_install_pending));
+        return Ok((
+            policy,
+            opa_engine,
+            Some(proto_policy),
+            middleware_install_pending,
+        ));
     }
 
     // No policy source available

crates/openshell-server/src/grpc/policy.rs

@@ -3135,6 +3135,18 @@ fn deterministic_policy_hash(policy: &ProtoSandboxPolicy) -> String {
         hasher.update(key.as_bytes());
         hasher.update(value.encode_to_vec());
     }
+    if !policy.network_middlewares.is_empty() {
+        hasher.update(b"network_middlewares");
+        for middleware in &policy.network_middlewares {
+            let encoded = middleware.encode_to_vec();
+            hasher.update(
+                u64::try_from(encoded.len())
+                    .expect("protobuf payload length fits in u64")
+                    .to_le_bytes(),
+            );
+            hasher.update(encoded);
+        }
+    }
     hex::encode(hasher.finalize())
 }
 
@@ -9315,6 +9327,26 @@ mod tests {
         assert_ne!(rev_a, rev_b);
     }
 
+    #[test]
+    fn policy_hash_changes_when_network_middlewares_change() {
+        let policy_a = ProtoSandboxPolicy::default();
+        let policy_b = ProtoSandboxPolicy {
+            network_middlewares: vec![openshell_core::proto::NetworkMiddlewareConfig {
+                name: "redact-secrets".into(),
+                middleware: "openshell/secrets".into(),
+                on_error: "fail_closed".into(),
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        assert_ne!(
+            deterministic_policy_hash(&policy_a),
+            deterministic_policy_hash(&policy_b),
+            "middleware-only policy changes must produce a new policy hash"
+        );
+    }
+
     #[test]
     fn config_revision_changes_when_policy_source_changes() {
         let policy = ProtoSandboxPolicy::default();

crates/openshell-supervisor-middleware/src/lib.rs

@@ -712,6 +712,37 @@ impl ChainRunner {
                 }
             };
 
+            if decision == Decision::Deny {
+                for finding in result.findings {
+                    findings.push(NamespacedFinding {
+                        middleware: entry.entry.name.clone(),
+                        finding,
+                    });
+                }
+                if !result.metadata.is_empty() {
+                    metadata.insert(
+                        entry.entry.name.clone(),
+                        result.metadata.into_iter().collect(),
+                    );
+                }
+                applied.push(MiddlewareInvocation {
+                    name: entry.entry.name.clone(),
+                    implementation: entry.entry.implementation.clone(),
+                    decision,
+                    transformed: false,
+                    failed: false,
+                });
+                return Ok(ChainOutcome {
+                    allowed: false,
+                    reason: safe_reason(&result.reason),
+                    body,
+                    added_headers,
+                    findings,
+                    metadata,
+                    applied,
+                });
+            }
+
             if result.has_body && result.body.len() > entry.max_body_bytes {
                 match apply_on_error(entry, "response_body_over_capacity", &mut applied) {
                     OnErrorAction::FailOpen => continue,
@@ -774,17 +805,6 @@ impl ChainRunner {
                 transformed,
                 failed: false,
             });
-            if decision == Decision::Deny {
-                return Ok(ChainOutcome {
-                    allowed: false,
-                    reason: safe_reason(&result.reason),
-                    body,
-                    added_headers,
-                    findings,
-                    metadata,
-                    applied,
-                });
-            }
         }
 
         Ok(ChainOutcome {
@@ -1399,6 +1419,62 @@ mod tests {
         assert!(!outcome.applied[0].failed);
     }
 
+    #[tokio::test]
+    async fn deny_decision_ignores_unsafe_mutations_under_fail_open() {
+        let runner = ChainRunner::new(Arc::new(scripted_service(
+            openshell_core::proto::HttpRequestResult {
+                decision: Decision::Deny as i32,
+                reason: "blocked_by_policy".into(),
+                add_headers: std::iter::once((
+                    "x-openshell-middleware-inject".to_string(),
+                    "ok\r\nHost: evil".to_string(),
+                ))
+                .collect(),
+                ..allow_result()
+            },
+        )));
+
+        let outcome = runner
+            .evaluate(&[entry("guard", OnError::FailOpen)], input("hello"))
+            .await
+            .expect("evaluate");
+
+        assert!(!outcome.allowed);
+        assert_eq!(outcome.reason, "blocked_by_policy");
+        assert!(outcome.added_headers.is_empty());
+        assert_eq!(outcome.applied.len(), 1);
+        assert_eq!(outcome.applied[0].decision, Decision::Deny);
+        assert!(!outcome.applied[0].failed);
+    }
+
+    #[tokio::test]
+    async fn deny_decision_ignores_oversized_replacement_under_fail_open() {
+        let runner = ChainRunner::new(Arc::new(ScriptedService {
+            binding_id: BUILTIN_SECRETS.into(),
+            max_body_bytes: 4,
+            result: openshell_core::proto::HttpRequestResult {
+                decision: Decision::Deny as i32,
+                reason: "blocked_by_policy".into(),
+                body: b"too large".to_vec(),
+                has_body: true,
+                ..allow_result()
+            },
+        }));
+
+        let outcome = runner
+            .evaluate(&[entry("guard", OnError::FailOpen)], input("safe"))
+            .await
+            .expect("evaluate");
+
+        assert!(!outcome.allowed);
+        assert_eq!(outcome.reason, "blocked_by_policy");
+        assert_eq!(outcome.body, b"safe");
+        assert_eq!(outcome.applied.len(), 1);
+        assert_eq!(outcome.applied[0].decision, Decision::Deny);
+        assert!(!outcome.applied[0].transformed);
+        assert!(!outcome.applied[0].failed);
+    }
+
     #[tokio::test]
     async fn metadata_and_findings_are_namespaced_per_config() {
         let runner = ChainRunner::new(Arc::new(scripted_service(

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

@@ -824,7 +824,14 @@ pub(crate) async fn apply_middleware_chain_for_scheme<C: AsyncRead + AsyncWrite
         // body. Apply each entry's `on_error` policy without buffering (an
         // unresolved binding is handled before the body is read) and forward
         // the original request unchanged if the chain allows.
-        let input = middleware_request_input(scheme, &req, ctx, BTreeMap::new(), String::new(), Vec::new());
+        let input = middleware_request_input(
+            scheme,
+            &req,
+            ctx,
+            BTreeMap::new(),
+            String::new(),
+            Vec::new(),
+        );
         let outcome = runner.evaluate_described(&chain, input).await?;
         emit_middleware_events(ctx, &req, &outcome);
         return Ok(if outcome.allowed {
@@ -2823,6 +2830,72 @@ network_policies:
             .unwrap();
     }
 
+    #[tokio::test]
+    async fn l7_rest_middleware_acknowledges_expect_continue_before_reading_body() {
+        let (config, tunnel_engine, ctx) =
+            middleware_relay_context("openshell/secrets", "fail_closed");
+        let (mut app, mut relay_client) = tokio::io::duplex(8192);
+        let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192);
+        let relay = tokio::spawn(async move {
+            relay_with_inspection(
+                &config,
+                tunnel_engine,
+                &mut relay_client,
+                &mut relay_upstream,
+                &ctx,
+            )
+            .await
+        });
+
+        let body = br#"{"api_key":"sk-1234567890abcdef"}"#;
+        let headers = format!(
+            "POST /v1/messages HTTP/1.1\r\nHost: api.example.test\r\nContent-Length: {}\r\nExpect: 100-continue\r\nConnection: close\r\n\r\n",
+            body.len()
+        );
+        app.write_all(headers.as_bytes()).await.unwrap();
+
+        let mut interim = [0u8; 64];
+        let n = tokio::time::timeout(std::time::Duration::from_secs(1), app.read(&mut interim))
+            .await
+            .expect("middleware buffering should acknowledge Expect before reading the body")
+            .unwrap();
+        assert_eq!(&interim[..n], b"HTTP/1.1 100 Continue\r\n\r\n");
+
+        app.write_all(body).await.unwrap();
+
+        let mut upstream_request = [0u8; 1024];
+        let n = tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            upstream.read(&mut upstream_request),
+        )
+        .await
+        .expect("request should reach upstream after the body is released")
+        .unwrap();
+        let upstream_request = String::from_utf8_lossy(&upstream_request[..n]);
+        assert!(upstream_request.contains(r#""api_key":"[REDACTED]""#));
+        assert!(!upstream_request.contains("Expect: 100-continue"));
+
+        upstream
+            .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+            .await
+            .unwrap();
+        let mut client_response = [0u8; 512];
+        let n = tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            app.read(&mut client_response),
+        )
+        .await
+        .expect("response should reach client")
+        .unwrap();
+        assert!(String::from_utf8_lossy(&client_response[..n]).contains("204 No Content"));
+        drop(app);
+        tokio::time::timeout(std::time::Duration::from_secs(1), relay)
+            .await
+            .expect("relay should finish")
+            .unwrap()
+            .unwrap();
+    }
+
     #[tokio::test]
     async fn l7_rest_middleware_fail_closed_does_not_reach_upstream() {
         let (config, tunnel_engine, ctx) =
@@ -3484,7 +3557,8 @@ network_policies:
         let mut response = Vec::new();
         let mut buf = [0u8; 512];
         while !String::from_utf8_lossy(&response).contains("middleware_failed") {
-            match tokio::time::timeout(std::time::Duration::from_secs(1), app.read(&mut buf)).await {
+            match tokio::time::timeout(std::time::Duration::from_secs(1), app.read(&mut buf)).await
+            {
                 Ok(Ok(0)) | Err(_) => break, // clean EOF, or no more data before the deadline
                 Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
                 Ok(Err(e)) => panic!("read from relay failed: {e}"),

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

@@ -828,7 +828,7 @@ pub(crate) enum BufferResult {
     OverCapacity { recoverable: bool },
 }
 
-pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + Unpin>(
+pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + AsyncWrite + Unpin>(
     req: &L7Request,
     client: &mut C,
     generation_guard: Option<&PolicyGenerationGuard>,
@@ -839,7 +839,7 @@ pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + Unpin>(
         .windows(4)
         .position(|w| w == b"\r\n\r\n")
         .map_or(req.raw_header.len(), |p| p + 4);
-    let headers = req.raw_header[..header_end].to_vec();
+    let mut headers = req.raw_header[..header_end].to_vec();
     let already_read = &req.raw_header[header_end..];
     match req.body_length {
         BodyLength::None => Ok(BufferResult::Buffered(BufferedRequestBody {
@@ -860,6 +860,9 @@ pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + Unpin>(
             let mut body = Vec::with_capacity(len);
             body.extend_from_slice(&already_read[..initial_len]);
             let mut remaining = len.saturating_sub(initial_len);
+            if remaining > 0 && already_read.is_empty() {
+                acknowledge_expect_continue(client, &mut headers).await?;
+            }
             let mut buf = [0u8; RELAY_BUF_SIZE];
             while remaining > 0 {
                 let to_read = remaining.min(buf.len());
@@ -887,6 +890,9 @@ pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + Unpin>(
             // we have already consumed wire bytes from the client stream and
             // cannot re-enter the normal raw relay path without a separate
             // splice-through buffer.
+            if already_read.is_empty() {
+                acknowledge_expect_continue(client, &mut headers).await?;
+            }
             Ok(
                 collect_chunked_body(client, already_read, generation_guard, Some(max_body_bytes))
                     .await
@@ -898,6 +904,25 @@ pub(crate) async fn buffer_request_body_for_middleware<C: AsyncRead + Unpin>(
     }
 }
 
+async fn acknowledge_expect_continue<C: AsyncWrite + Unpin>(
+    client: &mut C,
+    headers: &mut Vec<u8>,
+) -> Result<()> {
+    let header_str =
+        std::str::from_utf8(headers).map_err(|_| miette!("HTTP headers contain invalid UTF-8"))?;
+    if !has_expect_continue(header_str) {
+        return Ok(());
+    }
+
+    client
+        .write_all(b"HTTP/1.1 100 Continue\r\n\r\n")
+        .await
+        .into_diagnostic()?;
+    client.flush().await.into_diagnostic()?;
+    *headers = strip_header(headers, "expect")?;
+    Ok(())
+}
+
 pub(crate) fn rebuild_request_with_buffered_body(
     req: &L7Request,
     headers: &[u8],