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":"additional-ca","cmd":"OPENSHELL_E2E_DOCKER_TEST=additional_ca mise run --no-deps --skip-deps e2e:docker","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":"additional-ca","cmd":"OPENSHELL_E2E_DOCKER_TEST=additional_ca mise run --no-deps --skip-deps e2e:docker","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"middleware-http-hooks","cmd":"mise run --no-deps --skip-deps e2e:middleware-http-hooks","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

@@ -29,6 +29,14 @@ pub const COMPUTE_SUSPEND_RESUME: &str = "openshell.compute.suspend-resume";
 pub const COMPUTE_UPDATE_SANDBOX_CONFIGURATION: &str =
     "openshell.compute.update-sandbox-configuration";
 
+/// Capability for supervisor middleware that implements HTTP hooks.
+///
+/// A service selects HTTP hooks for all of its HTTP bindings by
+/// listing this capability in `required_capabilities`. Peers that predate
+/// HTTP hooks do not support it, so they reject the service at
+/// `Describe` instead of calling the deprecated legacy HTTP hooks.
+pub const SUPERVISOR_MIDDLEWARE_HTTP_HOOKS: &str = "openshell.supervisor-middleware.http-hooks";
+
 const MAX_IMPLEMENTATION_NAME_BYTES: usize = 128;
 const MAX_IMPLEMENTATION_VERSION_BYTES: usize = 128;
 const MAX_CAPABILITY_BYTES: usize = 128;
@@ -129,10 +137,16 @@ 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());
-        supported_capabilities.push(COMPUTE_UPDATE_SANDBOX_CONFIGURATION.to_string());
+    match family {
+        ExtensionFamily::Compute => {
+            supported_capabilities.push(COMPUTE_LAUNCH_AUTHENTICATION.to_string());
+            supported_capabilities.push(COMPUTE_ISOLATION_BACKEND.to_string());
+            supported_capabilities.push(COMPUTE_UPDATE_SANDBOX_CONFIGURATION.to_string());
+        }
+        ExtensionFamily::SupervisorMiddleware => {
+            supported_capabilities.push(SUPERVISOR_MIDDLEWARE_HTTP_HOOKS.to_string());
+        }
+        ExtensionFamily::Credentials | ExtensionFamily::GatewayInterceptor => {}
     }
     PeerMetadata {
         protocol_version: Some(ProtocolVersion {
@@ -176,6 +190,27 @@ fn expected_contract(gateway: &PeerMetadata) -> String {
     )
 }
 
+/// Extension metadata for supervisor middleware that implements HTTP hooks.
+///
+/// It supports and requires [`SUPERVISOR_MIDDLEWARE_HTTP_HOOKS`], so peers
+/// without HTTP hooks reject the service at `Describe`.
+#[must_use]
+pub fn http_hooks_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_HOOKS.to_string()],
+    );
+    metadata
+        .required_capabilities
+        .push(SUPERVISOR_MIDDLEWARE_HTTP_HOOKS.to_string());
+    metadata
+}
+
 pub fn negotiate(
     family: ExtensionFamily,
     configured_name: impl Into<String>,
@@ -512,6 +547,47 @@ mod tests {
         );
     }
 
+    #[test]
+    fn http_hooks_middleware_requires_a_peer_with_http_hooks() {
+        let middleware = http_hooks_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 hooks accepts the service");
+        assert!(
+            negotiated
+                .required_capabilities
+                .iter()
+                .any(|capability| capability == SUPERVISOR_MIDDLEWARE_HTTP_HOOKS)
+        );
+
+        // A peer that predates HTTP hooks advertises only the
+        // contract capability.
+        let mut older = gateway;
+        older
+            .supported_capabilities
+            .retain(|capability| capability != SUPERVISOR_MIDDLEWARE_HTTP_HOOKS);
+        let error = negotiate(
+            ExtensionFamily::SupervisorMiddleware,
+            "example/guard",
+            &older,
+            Some(middleware),
+        )
+        .expect_err("a peer without HTTP hooks rejects the service");
+        assert!(matches!(
+            error,
+            NegotiationError::MissingGatewayCapabilities { .. }
+        ));
+        assert!(
+            error.to_string().contains(SUPERVISOR_MIDDLEWARE_HTTP_HOOKS),
+            "{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,27 @@ use tokio::sync::mpsc;
 use tonic::{Request, Response, Status};
 
 use crate::proto::{
-    HttpHeader, HttpRequestEvaluation, HttpRequestResult, HttpRequestTarget, HttpResponseEvent,
-    HttpResponseEventResult, MiddlewareDescribeRequest, MiddlewareManifest, RequestContext,
-    SupervisorMiddlewarePhase, ValidateConfigRequest, ValidateConfigResponse,
+    HttpHeader, HttpRequestEvaluation, HttpRequestHookEvent, HttpRequestHookResult,
+    HttpRequestResult, HttpRequestTarget, HttpResponseEvent, HttpResponseEventResult,
+    HttpResponseHookEvent, HttpResponseHookResult, MiddlewareDescribeRequest, MiddlewareManifest,
+    RequestContext, SupervisorMiddlewarePhase, ValidateConfigRequest, ValidateConfigResponse,
     WebSocketSessionEvent, WebSocketSessionEventResult,
 };
 
+/// Transport-neutral result stream for one `HttpHooks.EvaluateRequest`
+/// exchange.
+pub type HttpRequestHookResultStream = Pin<
+    Box<dyn tokio_stream::Stream<Item = Result<HttpRequestHookResult, Status>> + Send + 'static>,
+>;
+
+/// Transport-neutral result stream for one `HttpHooks.EvaluateResponse`
+/// exchange.
+pub type HttpResponseHookResultStream = Pin<
+    Box<dyn tokio_stream::Stream<Item = Result<HttpResponseHookResult, Status>> + Send + 'static>,
+>;
+
+/// Legacy 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 +80,26 @@ pub trait SupervisorMiddlewareEndpoint: Send + Sync {
             "middleware does not implement HTTP response pre-return evaluation",
         ))
     }
+
+    /// Open one HTTP hook request exchange (`HttpHooks.EvaluateRequest`).
+    async fn open_http_request_hook(
+        &self,
+        _requests: mpsc::Receiver<HttpRequestHookEvent>,
+    ) -> Result<HttpRequestHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP hook requests",
+        ))
+    }
+
+    /// Open one HTTP hook response exchange (`HttpHooks.EvaluateResponse`).
+    async fn open_http_response_hook(
+        &self,
+        _requests: mpsc::Receiver<HttpResponseHookEvent>,
+    ) -> Result<HttpResponseHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP hook responses",
+        ))
+    }
 }
 
 /// Borrowed request state exposed to one in-process middleware invocation.
@@ -193,7 +228,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 +313,33 @@ pub trait InProcessMiddleware: Send + Sync {
             "middleware does not implement HTTP response pre-return evaluation",
         ))
     }
+
+    /// Open one HTTP hook request exchange (`HttpHooks.EvaluateRequest`),
+    /// for one HTTP request.
+    ///
+    /// Implementations that do not require the HTTP hooks capability
+    /// may keep the default unsupported response.
+    async fn open_http_request_hook(
+        &self,
+        _requests: mpsc::Receiver<HttpRequestHookEvent>,
+    ) -> std::result::Result<HttpRequestHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP hook requests",
+        ))
+    }
+
+    /// Open one HTTP hook response exchange (`HttpHooks.EvaluateResponse`).
+    ///
+    /// Implementations that do not require the HTTP hooks capability
+    /// may keep the default unsupported response.
+    async fn open_http_response_hook(
+        &self,
+        _requests: mpsc::Receiver<HttpResponseHookEvent>,
+    ) -> std::result::Result<HttpResponseHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement HTTP hook responses",
+        ))
+    }
 }
 
 /// Default timeout for one supervisor middleware RPC.
@@ -292,12 +354,32 @@ 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 legacy 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 hooks use
+/// [`MAX_HTTP_HOOK_BUFFERING_SLOTS`] and
+/// [`MAX_HTTP_HOOK_STREAMING_SLOTS`] instead.
 pub const MAX_CONCURRENT_MIDDLEWARE_SESSIONS: usize = 32;
+/// Process-wide bound on HTTP requests and responses in HTTP hook preflight
+/// or in a `WHOLE_BODY` stage.
+///
+/// A request and its response each hold one slot, and only while a stage
+/// inspects them: one whose stages all continue at preflight releases its
+/// slot then. A request or response waits for a slot within its chain
+/// deadline.
+pub const MAX_HTTP_HOOK_BUFFERING_SLOTS: usize = 64;
+/// Process-wide bound on HTTP requests and responses whose HTTP hook chain
+/// has a DUPLEX stage or deferred stages.
+///
+/// Such a 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 requests and responses. 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_HTTP_HOOK_STREAMING_SLOTS: 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

@@ -3292,6 +3292,83 @@ network_policies:
         assert!(errors.contains("invalid host pattern"), "{errors}");
     }
 
+    #[test]
+    fn validate_rejects_ports_in_middleware_selector_patterns() {
+        for (include, exclude, reason) in [
+            ("api.example.com:443", None, "cannot include one"),
+            ("*.example.com:443", None, "cannot include one"),
+            ("api.example.com:*", None, "cannot include one"),
+            (
+                "api.example.com",
+                Some("api.example.com:22"),
+                "cannot include one",
+            ),
+            ("10.0.0.1:443", None, "cannot include one"),
+            ("10.0.0.1:*", None, "cannot include one"),
+            ("*:443", None, "cannot include one"),
+            ("[2001:db8::1]:443", None, "without brackets"),
+            ("[2001:db8::1]", None, "without brackets"),
+            ("2001:db8::1::2", None, "complete IPv6 address"),
+            ("cafe:*", None, "complete IPv6 address"),
+            ("beef:443", None, "complete IPv6 address"),
+            ("2001:4860:*", None, "complete IPv6 address"),
+            ("fe80::*", None, "complete IPv6 address"),
+        ] {
+            let mut policy = restrictive_default_policy();
+            let mut middleware = middleware_config("openshell/regex");
+            let selector = middleware.endpoints.as_mut().unwrap();
+            selector.include = vec![include.into()];
+            selector.exclude = exclude.into_iter().map(Into::into).collect();
+            add_middleware(&mut policy, "redactor", middleware);
+
+            let errors = validate_sandbox_policy(&policy)
+                .expect_err("a selector pattern with a port")
+                .into_iter()
+                .map(|violation| violation.to_string())
+                .collect::<Vec<_>>()
+                .join("; ");
+            let pattern = exclude.unwrap_or(include);
+            assert!(
+                errors.contains(&format!("pattern '{pattern}' is invalid"))
+                    && errors.contains(reason),
+                "{errors}"
+            );
+        }
+
+        // The supervisor's local-file policy path runs the same check.
+        let data = serde_json::json!({
+            "network_middlewares": {
+                "redactor": {
+                    "middleware": "openshell/regex",
+                    "endpoints": {"include": ["api.example.com:443"]}
+                }
+            }
+        });
+        let violations = validate_network_middleware_json(&data).expect("parse middleware policy");
+        assert!(violations.iter().any(|violation| matches!(
+            violation,
+            PolicyViolation::InvalidMiddlewareConfig { name, reason }
+                if name == "redactor" && reason.contains("cannot include one")
+        )));
+    }
+
+    #[test]
+    fn validate_accepts_ipv6_middleware_selector_addresses() {
+        for pattern in ["2001:db8::1", "2001:DB8::1", "::1", "::ffff:10.0.0.1"] {
+            let mut policy = restrictive_default_policy();
+            let mut middleware = middleware_config("openshell/regex");
+            middleware.endpoints.as_mut().unwrap().include = vec![pattern.into()];
+            add_middleware(&mut policy, "redactor", middleware);
+            if let Err(violations) = validate_sandbox_policy(&policy) {
+                panic!("IPv6 selector '{pattern}' was rejected: {violations:?}");
+            }
+        }
+
+        // The proxy matches IPv6 hosts without brackets.
+        assert!(middleware_host_matches("2001:DB8::1", "2001:db8::1").unwrap());
+        assert!(!middleware_host_matches("2001:db8::1", "2001:db8::2").unwrap());
+    }
+
     #[test]
     fn middleware_host_selector_matching_is_case_insensitive() {
         assert!(middleware_host_matches("*.Example.COM", "API.example.com").unwrap());

crates/openshell-policy/src/middleware.rs

@@ -179,6 +179,26 @@ where
     Ok(violations)
 }
 
+/// Parse one middleware selector pattern. Selectors match hostnames on every
+/// port, so a `host:port` pattern would parse as a hostname that never matches
+/// and silently leave the entry unselected. The only pattern that may contain
+/// `:` is a complete IPv6 address without brackets, such as `2001:db8::1`.
+/// IPv6 globs are rejected: `cafe:*` cannot be told apart from a mistyped
+/// `host:port`.
+fn selector_pattern(pattern: &str) -> Result<HostPattern, String> {
+    if pattern.contains(':') && pattern.parse::<std::net::Ipv6Addr>().is_err() {
+        let reason = if pattern.starts_with('[') {
+            "write IPv6 addresses without brackets"
+        } else {
+            "':' is only allowed in a complete IPv6 address"
+        };
+        return Err(format!(
+            "{reason}; selectors match hosts on every port and cannot include one"
+        ));
+    }
+    HostPattern::new(pattern)
+}
+
 pub fn validate(policy: &SandboxPolicy) -> Vec<PolicyViolation> {
     let mut violations = Vec::new();
     let mut orders = BTreeMap::new();
@@ -249,7 +269,7 @@ pub fn validate(policy: &SandboxPolicy) -> Vec<PolicyViolation> {
         }
         let mut selector_valid = !selector.include.is_empty();
         for pattern in selector.include.iter().chain(&selector.exclude) {
-            if let Err(reason) = HostPattern::new(pattern) {
+            if let Err(reason) = selector_pattern(pattern) {
                 selector_valid = false;
                 violations.push(PolicyViolation::InvalidMiddlewareConfig {
                     name: name.clone(),