feat(middleware): introduce improved HTTP request/response hooks for the middleware

.github/workflows/e2e-docker-test.yml

@@ -29,7 +29,7 @@ on:
         required: false
         type: string
         default: >-
-          [{"suite":"python","cmd":"mise run --no-deps --skip-deps e2e:python","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-python","cmd":"mise run --no-deps --skip-deps e2e:oidc-python:docker","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-pkce-docker","cmd":"mise run --no-deps --skip-deps e2e:oidc-pkce:docker","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker","cmd":"mise run --no-deps --skip-deps e2e:rust","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker-push","cmd":"mise run --no-deps --skip-deps e2e:rust:push","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"oci-image","cmd":"mise run --no-deps --skip-deps e2e:oci-image","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"mcp","cmd":"mise run --no-deps --skip-deps e2e:mcp","apt_packages":"","python_proto":false,"mcp":true}]
+          [{"suite":"python","cmd":"mise run --no-deps --skip-deps e2e:python","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-python","cmd":"mise run --no-deps --skip-deps e2e:oidc-python:docker","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-pkce-docker","cmd":"mise run --no-deps --skip-deps e2e:oidc-pkce:docker","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker","cmd":"mise run --no-deps --skip-deps e2e:rust","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker-push","cmd":"mise run --no-deps --skip-deps e2e:rust:push","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"oci-image","cmd":"mise run --no-deps --skip-deps e2e:oci-image","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"middleware-http-session","cmd":"mise run --no-deps --skip-deps e2e:middleware-http-session","apt_packages":"","python_proto":false,"mcp":false},{"suite":"mcp","cmd":"mise run --no-deps --skip-deps e2e:mcp","apt_packages":"","python_proto":false,"mcp":true}]
 
 permissions:
   actions: read

crates/openshell-core/src/extension_protocol.rs

@@ -22,6 +22,14 @@ pub const COMPUTE_LAUNCH_AUTHENTICATION: &str = "openshell.compute.launch-authen
 /// Driver stages operator-selected external isolation routes for the supervisor.
 pub const COMPUTE_ISOLATION_BACKEND: &str = "openshell.compute.isolation-backend";
 
+/// Capability for supervisor middleware that implements HTTP session hooks.
+///
+/// A service selects HTTP session hooks for all of its HTTP bindings by
+/// listing this capability in `required_capabilities`. Peers that predate
+/// HTTP session hooks do not support it, so they reject the service at
+/// `Describe` instead of calling the deprecated v1 HTTP hooks.
+pub const SUPERVISOR_MIDDLEWARE_HTTP_SESSION: &str = "openshell.supervisor-middleware.http-session";
+
 const MAX_IMPLEMENTATION_NAME_BYTES: usize = 128;
 const MAX_IMPLEMENTATION_VERSION_BYTES: usize = 128;
 const MAX_CAPABILITY_BYTES: usize = 128;
@@ -122,9 +130,15 @@ pub enum NegotiationError {
 pub fn gateway_metadata(family: ExtensionFamily) -> PeerMetadata {
     let contract = family.contract_capability();
     let mut supported_capabilities = vec![contract.clone()];
-    if family == ExtensionFamily::Compute {
-        supported_capabilities.push(COMPUTE_LAUNCH_AUTHENTICATION.to_string());
-        supported_capabilities.push(COMPUTE_ISOLATION_BACKEND.to_string());
+    match family {
+        ExtensionFamily::Compute => {
+            supported_capabilities.push(COMPUTE_LAUNCH_AUTHENTICATION.to_string());
+            supported_capabilities.push(COMPUTE_ISOLATION_BACKEND.to_string());
+        }
+        ExtensionFamily::SupervisorMiddleware => {
+            supported_capabilities.push(SUPERVISOR_MIDDLEWARE_HTTP_SESSION.to_string());
+        }
+        ExtensionFamily::Credentials | ExtensionFamily::GatewayInterceptor => {}
     }
     PeerMetadata {
         protocol_version: Some(ProtocolVersion {
@@ -168,6 +182,27 @@ fn expected_contract(gateway: &PeerMetadata) -> String {
     )
 }
 
+/// Extension metadata for supervisor middleware that implements HTTP session hooks.
+///
+/// It supports and requires [`SUPERVISOR_MIDDLEWARE_HTTP_SESSION`], so peers
+/// without HTTP session hooks reject the service at `Describe`.
+#[must_use]
+pub fn http_session_middleware_metadata(
+    implementation_name: impl Into<String>,
+    implementation_version: impl Into<String>,
+) -> PeerMetadata {
+    let mut metadata = extension_metadata(
+        ExtensionFamily::SupervisorMiddleware,
+        implementation_name,
+        implementation_version,
+        [SUPERVISOR_MIDDLEWARE_HTTP_SESSION.to_string()],
+    );
+    metadata
+        .required_capabilities
+        .push(SUPERVISOR_MIDDLEWARE_HTTP_SESSION.to_string());
+    metadata
+}
+
 pub fn negotiate(
     family: ExtensionFamily,
     configured_name: impl Into<String>,
@@ -487,6 +522,49 @@ mod tests {
         );
     }
 
+    #[test]
+    fn http_session_middleware_requires_a_peer_with_http_session_hooks() {
+        let middleware = http_session_middleware_metadata("example/guard", "test");
+        let gateway = gateway_metadata(ExtensionFamily::SupervisorMiddleware);
+        let negotiated = negotiate(
+            ExtensionFamily::SupervisorMiddleware,
+            "example/guard",
+            &gateway,
+            Some(middleware.clone()),
+        )
+        .expect("a peer with HTTP session hooks accepts the service");
+        assert!(
+            negotiated
+                .required_capabilities
+                .iter()
+                .any(|capability| capability == SUPERVISOR_MIDDLEWARE_HTTP_SESSION)
+        );
+
+        // A peer that predates HTTP session hooks advertises only the
+        // contract capability.
+        let mut older = gateway;
+        older
+            .supported_capabilities
+            .retain(|capability| capability != SUPERVISOR_MIDDLEWARE_HTTP_SESSION);
+        let error = negotiate(
+            ExtensionFamily::SupervisorMiddleware,
+            "example/guard",
+            &older,
+            Some(middleware),
+        )
+        .expect_err("a peer without HTTP session hooks rejects the service");
+        assert!(matches!(
+            error,
+            NegotiationError::MissingGatewayCapabilities { .. }
+        ));
+        assert!(
+            error
+                .to_string()
+                .contains(SUPERVISOR_MIDDLEWARE_HTTP_SESSION),
+            "{error}"
+        );
+    }
+
     #[test]
     fn both_peers_require_capabilities_from_the_other() {
         let (mut gateway, mut extension) = compatible();

crates/openshell-core/src/middleware.rs

@@ -11,12 +11,29 @@ use tokio::sync::mpsc;
 use tonic::{Request, Response, Status};
 
 use crate::proto::{
-    HttpHeader, HttpRequestEvaluation, HttpRequestResult, HttpRequestTarget, HttpResponseEvent,
-    HttpResponseEventResult, MiddlewareDescribeRequest, MiddlewareManifest, RequestContext,
-    SupervisorMiddlewarePhase, ValidateConfigRequest, ValidateConfigResponse,
-    WebSocketSessionEvent, WebSocketSessionEventResult,
+    HttpHeader, HttpRequestEvaluation, HttpRequestResult, HttpRequestSessionEvent,
+    HttpRequestSessionResult, HttpRequestTarget, HttpResponseEvent, HttpResponseEventResult,
+    HttpResponseSessionEvent, HttpResponseSessionResult, MiddlewareDescribeRequest,
+    MiddlewareManifest, RequestContext, SupervisorMiddlewarePhase, ValidateConfigRequest,
+    ValidateConfigResponse, WebSocketSessionEvent, WebSocketSessionEventResult,
 };
 
+/// Transport-neutral result stream for one `EvaluateHttpRequestSession`
+/// exchange.
+pub type HttpRequestSessionResultStream = Pin<
+    Box<dyn tokio_stream::Stream<Item = Result<HttpRequestSessionResult, Status>> + Send + 'static>,
+>;
+
+/// Transport-neutral result stream for one `EvaluateHttpResponseSession`
+/// exchange.
+pub type HttpResponseSessionResultStream = Pin<
+    Box<
+        dyn tokio_stream::Stream<Item = Result<HttpResponseSessionResult, Status>> + Send + 'static,
+    >,
+>;
+
+/// v1 HTTP hooks. Removed in 0.2.0.
+///
 /// Transport-neutral result stream for one HTTP response middleware stage.
 pub type HttpResponseResultStream = Pin<
     Box<dyn tokio_stream::Stream<Item = Result<HttpResponseEventResult, Status>> + Send + 'static>,
@@ -65,6 +82,26 @@ pub trait SupervisorMiddlewareEndpoint: Send + Sync {
             "middleware does not implement HTTP response pre-return evaluation",
         ))
     }
+
+    /// Open one HTTP session hook request exchange (`EvaluateHttpRequestSession`).
+    async fn open_http_request_session(
+        &self,
+        _requests: mpsc::Receiver<HttpRequestSessionEvent>,
+    ) -> Result<HttpRequestSessionResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP session hook requests",
+        ))
+    }
+
+    /// Open one HTTP session hook response exchange (`EvaluateHttpResponseSession`).
+    async fn open_http_response_session(
+        &self,
+        _requests: mpsc::Receiver<HttpResponseSessionEvent>,
+    ) -> Result<HttpResponseSessionResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP session hook responses",
+        ))
+    }
 }
 
 /// Borrowed request state exposed to one in-process middleware invocation.
@@ -193,7 +230,7 @@ impl<'a> HttpRequestView<'a> {
 ///                 operation: SupervisorMiddlewareOperation::HttpRequest as i32,
 ///                 phase: SupervisorMiddlewarePhase::PreCredentials as i32,
 ///                 max_payload_bytes: 1024,
-///                 request_timeout: None,
+///                 ..Default::default()
 ///             }],
 ///             expected_audience: String::new(),
 ///             extension: Some(openshell_core::extension_protocol::extension_metadata(
@@ -278,6 +315,33 @@ pub trait InProcessMiddleware: Send + Sync {
             "middleware does not implement HTTP response pre-return evaluation",
         ))
     }
+
+    /// Open one HTTP session hook request exchange (`EvaluateHttpRequestSession`),
+    /// for a request or for traffic `OpenShell` cannot inspect.
+    ///
+    /// Implementations that do not require the HTTP session hooks capability
+    /// may keep the default unsupported response.
+    async fn open_http_request_session(
+        &self,
+        _requests: mpsc::Receiver<HttpRequestSessionEvent>,
+    ) -> std::result::Result<HttpRequestSessionResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP session hook requests",
+        ))
+    }
+
+    /// Open one HTTP session hook response exchange (`EvaluateHttpResponseSession`).
+    ///
+    /// Implementations that do not require the HTTP session hooks capability
+    /// may keep the default unsupported response.
+    async fn open_http_response_session(
+        &self,
+        _requests: mpsc::Receiver<HttpResponseSessionEvent>,
+    ) -> std::result::Result<HttpResponseSessionResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP session hook responses",
+        ))
+    }
 }
 
 /// Default timeout for one supervisor middleware RPC.
@@ -292,12 +356,29 @@ pub const MAX_MIDDLEWARE_CHAIN_TIMEOUT: Duration = Duration::from_secs(30);
 pub const MAX_MIDDLEWARE_PREFLIGHT_TIMEOUT: Duration = Duration::from_secs(1);
 /// Process-wide safety valve for concurrently buffered middleware work.
 pub const MAX_CONCURRENT_MIDDLEWARE_WORK: usize = 32;
-/// Process-wide safety valve for retained streaming middleware sessions.
+/// Process-wide safety valve for retained WebSocket and v1 HTTP response
+/// middleware sessions.
 ///
-/// A session consumes one permit regardless of its protocol or stage fan-out.
-/// Persistent middleware protocols must acquire from this shared budget before
-/// opening streams and retain the permit while any stage remains active.
+/// A session consumes one permit regardless of its stage fan-out. These
+/// protocols acquire from this shared budget before opening streams and retain
+/// the permit while any stage remains active. HTTP session hooks use
+/// [`MAX_CONCURRENT_HTTP_MIDDLEWARE_SESSIONS`] and
+/// [`MAX_CONCURRENT_HTTP_MIDDLEWARE_STREAMS`] instead.
 pub const MAX_CONCURRENT_MIDDLEWARE_SESSIONS: usize = 32;
+/// Process-wide bound on HTTP session hook messages in preflight or in a
+/// BUFFERED body.
+///
+/// A request and its response each hold one slot, and only while a stage
+/// inspects them: a message whose stages all continue at preflight releases
+/// its slot then. A message waits for a slot within its chain deadline.
+pub const MAX_CONCURRENT_HTTP_MIDDLEWARE_SESSIONS: usize = 64;
+/// Process-wide bound on HTTP session hook messages with a STREAM stage.
+///
+/// A STREAM body has no total deadline, so a server-sent events response can
+/// hold its slot indefinitely. This separate budget keeps such streams from
+/// starving short messages. It allows a request and a response stream for each
+/// of the 96 network streams the `OpenShell` sandbox runtime relays at once.
+pub const MAX_CONCURRENT_HTTP_MIDDLEWARE_STREAMS: usize = 256;
 
 /// Largest number of middleware configurations accepted in one sandbox policy.
 pub const MAX_MIDDLEWARE_CONFIGS: usize = 10;

crates/openshell-policy/src/lib.rs

@@ -51,6 +51,7 @@ pub use merge::{
     policy_covers_rule,
 };
 pub use middleware::middleware_host_matches;
+pub use middleware::tls_skip_conflicts as middleware_tls_skip_conflicts;
 pub use middleware::validate_json as validate_network_middleware_json;
 pub use middleware::validate_json_with_config as validate_network_middleware_json_with_config;
 
@@ -3323,7 +3324,8 @@ network_policies:
             },
         );
 
-        let violations = validate_sandbox_policy(&policy).expect_err("tls skip conflict");
+        validate_sandbox_policy(&policy).expect("the tls: skip rule needs the registry");
+        let violations = middleware_tls_skip_conflicts(&policy, |_| false);
         assert!(violations.iter().any(|violation| matches!(
             violation,
             PolicyViolation::MiddlewareTlsSkipConflict {
@@ -3354,8 +3356,36 @@ network_policies:
             },
         );
 
-        validate_sandbox_policy(&policy)
-            .expect("fail-open middleware may select uninspectable tls: skip traffic");
+        assert!(
+            middleware_tls_skip_conflicts(&policy, |_| false).is_empty(),
+            "fail-open middleware may select uninspectable tls: skip traffic"
+        );
+    }
+
+    #[test]
+    fn tls_skip_rule_exempts_middleware_that_decides_uninspectable_traffic() {
+        let mut policy = restrictive_default_policy();
+        add_middleware(&mut policy, "guard", middleware_config("example/guard"));
+        policy.network_policies.insert(
+            "api".into(),
+            NetworkPolicyRule {
+                name: "api".into(),
+                endpoints: vec![NetworkEndpoint {
+                    host: "api.example.com".into(),
+                    port: 443,
+                    tls: openshell_core::proto::NetworkTlsMode::Skip as i32,
+                    ..Default::default()
+                }],
+                binaries: Vec::new(),
+            },
+        );
+
+        assert_eq!(middleware_tls_skip_conflicts(&policy, |_| false).len(), 1);
+        assert!(
+            middleware_tls_skip_conflicts(&policy, |middleware| middleware == "example/guard")
+                .is_empty(),
+            "HTTP session hook middleware decides about tls: skip tunnels at runtime"
+        );
     }
 
     #[test]
@@ -3378,7 +3408,8 @@ network_policies:
             },
         );
 
-        let violations = validate_sandbox_policy(&policy).expect_err("tls skip conflict");
+        validate_sandbox_policy(&policy).expect("the tls: skip rule needs the registry");
+        let violations = middleware_tls_skip_conflicts(&policy, |_| false);
         assert!(violations.iter().any(|violation| matches!(
             violation,
             PolicyViolation::MiddlewareTlsSkipConflict { middleware_name, .. }
@@ -3406,7 +3437,8 @@ network_policies:
             },
         );
 
-        let violations = validate_sandbox_policy(&policy).expect_err("tls skip conflict");
+        validate_sandbox_policy(&policy).expect("the tls: skip rule needs the registry");
+        let violations = middleware_tls_skip_conflicts(&policy, |_| false);
         assert!(violations.iter().any(|violation| matches!(
             violation,
             PolicyViolation::MiddlewareTlsSkipConflict {