feat(middleware): TCP connection hooks with STATS mode (spike)

crates/openshell-core/src/extension_protocol.rs

@@ -37,6 +37,14 @@ pub const COMPUTE_UPDATE_SANDBOX_CONFIGURATION: &str =
 /// `Describe` instead of calling the deprecated legacy HTTP hooks.
 pub const SUPERVISOR_MIDDLEWARE_HTTP_HOOKS: &str = "openshell.supervisor-middleware.http-hooks";
 
+/// Capability for supervisor middleware that implements TCP connection hooks
+/// (`TcpHooks.EvaluateConnection`).
+///
+/// A service with a `TCP_CONNECTION` binding must list this capability in
+/// `required_capabilities`, so peers without TCP hooks reject it at
+/// `Describe`.
+pub const SUPERVISOR_MIDDLEWARE_TCP_HOOKS: &str = "openshell.supervisor-middleware.tcp-hooks";
+
 const MAX_IMPLEMENTATION_NAME_BYTES: usize = 128;
 const MAX_IMPLEMENTATION_VERSION_BYTES: usize = 128;
 const MAX_CAPABILITY_BYTES: usize = 128;
@@ -145,6 +153,7 @@ pub fn gateway_metadata(family: ExtensionFamily) -> PeerMetadata {
         }
         ExtensionFamily::SupervisorMiddleware => {
             supported_capabilities.push(SUPERVISOR_MIDDLEWARE_HTTP_HOOKS.to_string());
+            supported_capabilities.push(SUPERVISOR_MIDDLEWARE_TCP_HOOKS.to_string());
         }
         ExtensionFamily::Credentials | ExtensionFamily::GatewayInterceptor => {}
     }
@@ -211,6 +220,28 @@ pub fn http_hooks_middleware_metadata(
     metadata
 }
 
+/// Extension metadata for supervisor middleware that implements TCP
+/// connection hooks.
+///
+/// It supports and requires [`SUPERVISOR_MIDDLEWARE_TCP_HOOKS`], so peers
+/// without TCP hooks reject the service at `Describe`.
+#[must_use]
+pub fn tcp_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_TCP_HOOKS.to_string()],
+    );
+    metadata
+        .required_capabilities
+        .push(SUPERVISOR_MIDDLEWARE_TCP_HOOKS.to_string());
+    metadata
+}
+
 pub fn negotiate(
     family: ExtensionFamily,
     configured_name: impl Into<String>,

crates/openshell-core/src/middleware.rs

@@ -14,10 +14,15 @@ use crate::proto::{
     HttpHeader, HttpRequestEvaluation, HttpRequestHookEvent, HttpRequestHookResult,
     HttpRequestResult, HttpRequestTarget, HttpResponseEvent, HttpResponseEventResult,
     HttpResponseHookEvent, HttpResponseHookResult, MiddlewareDescribeRequest, MiddlewareManifest,
-    RequestContext, SupervisorMiddlewarePhase, ValidateConfigRequest, ValidateConfigResponse,
-    WebSocketSessionEvent, WebSocketSessionEventResult,
+    RequestContext, SupervisorMiddlewarePhase, TcpHookEvent, TcpHookResult, ValidateConfigRequest,
+    ValidateConfigResponse, WebSocketSessionEvent, WebSocketSessionEventResult,
 };
 
+/// Transport-neutral result stream for one `TcpHooks.EvaluateConnection`
+/// exchange.
+pub type TcpHookResultStream =
+    Pin<Box<dyn tokio_stream::Stream<Item = Result<TcpHookResult, Status>> + Send + 'static>>;
+
 /// Transport-neutral result stream for one `HttpHooks.EvaluateRequest`
 /// exchange.
 pub type HttpRequestHookResultStream = Pin<
@@ -100,6 +105,16 @@ pub trait SupervisorMiddlewareEndpoint: Send + Sync {
             "middleware does not implement HTTP hook responses",
         ))
     }
+
+    /// Open one TCP connection hook exchange (`TcpHooks.EvaluateConnection`).
+    async fn open_tcp_connection_hook(
+        &self,
+        _requests: mpsc::Receiver<TcpHookEvent>,
+    ) -> Result<TcpHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement TCP connection hooks",
+        ))
+    }
 }
 
 /// Borrowed request state exposed to one in-process middleware invocation.
@@ -340,6 +355,16 @@ pub trait InProcessMiddleware: Send + Sync {
             "middleware does not implement HTTP hook responses",
         ))
     }
+
+    /// Open one TCP connection hook exchange (`TcpHooks.EvaluateConnection`).
+    async fn open_tcp_connection_hook(
+        &self,
+        _requests: mpsc::Receiver<TcpHookEvent>,
+    ) -> std::result::Result<TcpHookResultStream, Status> {
+        Err(Status::unimplemented(
+            "middleware does not implement TCP connection hooks",
+        ))
+    }
 }
 
 /// Default timeout for one supervisor middleware RPC.

crates/openshell-policy/src/lib.rs

@@ -3379,6 +3379,7 @@ network_policies:
     }
 
     #[test]
+    #[ignore = "spike 02b: tls: skip conflict moved to gateway validation"]
     fn validate_rejects_middleware_selector_matching_tls_skip_endpoint() {
         let mut policy = restrictive_default_policy();
         add_middleware(
@@ -3436,6 +3437,7 @@ network_policies:
     }
 
     #[test]
+    #[ignore = "spike 02b: tls: skip conflict moved to gateway validation"]
     fn validate_rejects_explicit_fail_closed_middleware_on_tls_skip_endpoint() {
         let mut policy = restrictive_default_policy();
         let mut middleware = middleware_config("openshell/regex");
@@ -3464,6 +3466,7 @@ network_policies:
     }
 
     #[test]
+    #[ignore = "spike 02b: tls: skip conflict moved to gateway validation"]
     fn validate_rejects_concrete_selector_overlapping_tls_skip_wildcard() {
         let mut policy = restrictive_default_policy();
         let mut middleware = middleware_config("openshell/regex");

crates/openshell-policy/src/middleware.rs

@@ -283,7 +283,11 @@ pub fn validate(policy: &SandboxPolicy) -> Vec<PolicyViolation> {
             None
         };
 
-        let requires_inspection = matches!(middleware.on_error.as_str(), "" | "fail_closed");
+        // Spike 02b: TCP-only middleware may select tls: skip endpoints under
+        // fail_closed, which this static check cannot tell apart. The
+        // gateway's registry-aware validation (validate_tcp_hook_rules)
+        // enforces the conflict for services that inspect content.
+        let requires_inspection = false;
         for (key, rule) in &policy.network_policies {
             let policy_name = if rule.name.is_empty() {
                 key

crates/openshell-server/src/middleware.rs

@@ -13,7 +13,8 @@ pub async fn validate_policy(
 ) -> Result<(), Status> {
     async {
         registry.validate_policy_configs(policy).await?;
-        registry.validate_http_hook_rules(policy).await
+        registry.validate_http_hook_rules(policy).await?;
+        registry.validate_tcp_hook_rules(policy).await
     }
     .await
     .map_err(|error| {

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

@@ -9,6 +9,7 @@ pub mod headers;
 mod http_hooks;
 mod remote;
 mod response;
+mod tcp_hooks;
 mod websocket;
 
 pub use http_hooks::{
@@ -44,8 +45,8 @@ use miette::{Result, miette};
 use prost::Message;
 
 use openshell_core::extension_protocol::{
-    ExtensionFamily, NegotiatedExtension, SUPERVISOR_MIDDLEWARE_HTTP_HOOKS, gateway_metadata,
-    negotiate,
+    ExtensionFamily, NegotiatedExtension, SUPERVISOR_MIDDLEWARE_HTTP_HOOKS,
+    SUPERVISOR_MIDDLEWARE_TCP_HOOKS, gateway_metadata, negotiate,
 };
 use openshell_core::proto::middleware::v1::supervisor_middleware_server::SupervisorMiddleware;
 use openshell_core::proto::{
@@ -62,7 +63,11 @@ use binding_policy::BindingPolicy;
 pub use openshell_core::middleware::{
     HttpRequestHookResultStream, HttpRequestView, HttpResponseHookResultStream,
     HttpResponseResultStream, InProcessMiddleware, SupervisorMiddlewareEndpoint,
-    WebSocketResponseStream,
+    TcpHookResultStream, WebSocketResponseStream,
+};
+pub use tcp_hooks::{
+    MAX_TCP_STATS_INTERVAL, MIN_TCP_STATS_INTERVAL, MeteredStream, TcpCloseObserver,
+    TcpConnectionMeter, TcpHookAdmission, TcpHookDenial, TcpHookRequest, TcpHookSession,
 };
 
 pub type MiddlewareService =
@@ -235,6 +240,13 @@ impl InProcessMiddleware for EndpointInProcessAdapter {
     ) -> std::result::Result<HttpResponseHookResultStream, tonic::Status> {
         self.endpoint.open_http_response_hook(requests).await
     }
+
+    async fn open_tcp_connection_hook(
+        &self,
+        requests: tokio::sync::mpsc::Receiver<openshell_core::proto::TcpHookEvent>,
+    ) -> std::result::Result<TcpHookResultStream, tonic::Status> {
+        self.endpoint.open_tcp_connection_hook(requests).await
+    }
 }
 
 /// Adapt a transport-neutral endpoint to the in-process registry contract.
@@ -490,13 +502,51 @@ fn validate_manifest_http_hooks_capability(
     Ok(())
 }
 
+/// True when the service requires the TCP hooks capability.
+fn requires_tcp_hooks(manifest: &MiddlewareManifest) -> bool {
+    manifest.extension.as_ref().is_some_and(|extension| {
+        extension
+            .required_capabilities
+            .iter()
+            .any(|capability| capability == SUPERVISOR_MIDDLEWARE_TCP_HOOKS)
+    })
+}
+
+fn is_tcp_binding(binding: &MiddlewareBinding) -> bool {
+    binding.operation == SupervisorMiddlewareOperation::TcpConnection as i32
+}
+
+/// A service with a TCP connection binding must require the TCP hooks
+/// capability, so peers without TCP hooks reject it at Describe.
+fn validate_manifest_tcp_hooks_capability(
+    source: &str,
+    manifest: &MiddlewareManifest,
+) -> Result<()> {
+    if manifest.bindings.iter().any(is_tcp_binding) && !requires_tcp_hooks(manifest) {
+        return Err(miette!(
+            "{source} declares a TCP_CONNECTION binding without requiring {SUPERVISOR_MIDDLEWARE_TCP_HOOKS}"
+        ));
+    }
+    Ok(())
+}
+
+/// True when every binding of the service is a TCP connection binding. Such a
+/// service observes connections without inspecting content, so it never
+/// requires traffic to be inspectable.
+fn manifest_is_tcp_only(manifest: &MiddlewareManifest) -> bool {
+    !manifest.bindings.is_empty() && manifest.bindings.iter().all(is_tcp_binding)
+}
+
 /// Legacy HTTP hooks and WebSocket bindings carry payloads. An HTTP hook
 /// binding that advertises no payload limit is offered no body modes, so it
 /// can only continue or deny at preflight.
 fn binding_requires_payload_limit(
     manifest: &MiddlewareManifest,
     binding: &MiddlewareBinding,
 ) -> bool {
+    if is_tcp_binding(binding) {
+        return false;
+    }
     binding_http_hook_version(manifest, binding) != Some(HttpHookVersion::Current)
         || binding.max_payload_bytes > 0
 }
@@ -869,6 +919,16 @@ impl MiddlewareDispatch {
             Self::Grpc(service) => service.open_http_response_hook(receiver).await,
         }
     }
+
+    async fn open_tcp_connection_hook(
+        &self,
+        receiver: tokio::sync::mpsc::Receiver<openshell_core::proto::TcpHookEvent>,
+    ) -> std::result::Result<TcpHookResultStream, tonic::Status> {
+        match self {
+            Self::InProcess(service) => service.open_tcp_connection_hook(receiver).await,
+            Self::Grpc(service) => service.open_tcp_connection_hook(receiver).await,
+        }
+    }
 }
 
 struct MiddlewareServiceState {
@@ -1119,6 +1179,10 @@ fn supported_binding(
             Some(SupervisorMiddlewareOperation::WebsocketMessage),
             Some(SupervisorMiddlewarePhase::PreCredentials),
         ) => Ok(SupervisorMiddlewareBinding::WebsocketMessagePreCredentials),
+        (
+            Some(SupervisorMiddlewareOperation::TcpConnection),
+            Some(SupervisorMiddlewarePhase::PreConnect),
+        ) => Ok(SupervisorMiddlewareBinding::TcpConnectionPreConnect),
         (
             Some(SupervisorMiddlewareOperation::WebsocketMessage),
             Some(SupervisorMiddlewarePhase::PreReturn),
@@ -1140,6 +1204,7 @@ fn validate_manifest_bindings(
     }
 
     validate_manifest_http_hooks_capability(source, manifest)?;
+    validate_manifest_tcp_hooks_capability(source, manifest)?;
     let mut bindings = BTreeMap::new();
     for binding in &manifest.bindings {
         let selector = supported_binding(source, binding)?;
@@ -1541,6 +1606,78 @@ impl MiddlewareRegistry {
         })
     }
 
+    /// True unless this registry described the middleware and every one of its
+    /// bindings is a TCP connection binding. Only such a service may select
+    /// traffic `OpenShell` cannot inspect (`tls: skip`, raw TCP) under
+    /// `fail_closed`: it observes connections without inspecting content.
+    #[must_use]
+    pub fn requires_inspection(&self, implementation: &str) -> bool {
+        !self.services.iter().any(|state| {
+            state.manifest.get().is_some_and(|manifest| {
+                ChainRunner::attachment_name(state, manifest) == implementation
+                    && manifest_is_tcp_only(manifest)
+            })
+        })
+    }
+
+    /// Gateway rules for policies that use TCP hook middleware.
+    ///
+    /// - `on_error: fail_open` is rejected on an entry whose service has a TCP
+    ///   connection binding: TCP hooks always fail closed.
+    /// - A `fail_closed` entry whose service inspects content (any HTTP or
+    ///   WebSocket binding, or an undescribed service) must not select a
+    ///   `tls: skip` endpoint. TCP-only services may.
+    pub async fn validate_tcp_hook_rules(&self, policy: &SandboxPolicy) -> Result<()> {
+        use openshell_core::host_pattern::{HostPattern, HostSelector};
+        let manifests = ChainRunner::from_registry(self.clone()).manifests().await?;
+        let manifest_for = |implementation: &str| {
+            manifests
+                .iter()
+                .find(|(state, manifest)| {
+                    ChainRunner::attachment_name(state, manifest) == implementation
+                })
+                .map(|(_, manifest)| manifest)
+        };
+        let mut configs: Vec<_> = policy.network_middlewares.iter().collect();
+        configs.sort_unstable_by_key(|(name, _)| name.as_str());
+        for (name, config) in configs {
+            let manifest = manifest_for(&config.middleware);
+            let on_error = OnError::parse(&config.on_error)?;
+            if on_error == OnError::FailOpen
+                && manifest.is_some_and(|manifest| manifest.bindings.iter().any(is_tcp_binding))
+            {
+                return Err(miette!(
+                    "middleware config '{name}' cannot use on_error: fail_open with '{}': TCP hook middleware is always fail-closed; use fail_closed",
+                    config.middleware
+                ));
+            }
+            if on_error == OnError::FailOpen || manifest.is_some_and(manifest_is_tcp_only) {
+                continue;
+            }
+            let Some(selector) = config
+                .endpoints
+                .as_ref()
+                .and_then(|selector| HostSelector::new(&selector.include, &selector.exclude).ok())
+            else {
+                continue;
+            };
+            for (key, rule) in &policy.network_policies {
+                for endpoint in &rule.endpoints {
+                    if endpoint.tls == openshell_core::proto::NetworkTlsMode::Skip as i32
+                        && HostPattern::new(&endpoint.host)
+                            .is_ok_and(|pattern| selector.may_match_pattern(&pattern))
+                    {
+                        return Err(miette!(
+                            "fail-closed middleware config '{name}' selects tls: skip endpoint '{}' in network policy '{key}'; its service inspects content it cannot see there. Exclude the host, or use a TCP-only middleware service",
+                            endpoint.host
+                        ));
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Gateway rules for policies that use HTTP hook middleware.
     ///
     /// - `on_error: fail_open` is rejected on an entry whose service runs HTTP
@@ -1984,6 +2121,12 @@ impl ChainRunner {
         })
     }
 
+    /// See [`MiddlewareRegistry::requires_inspection`].
+    #[must_use]
+    pub fn requires_inspection(&self, implementation: &str) -> bool {
+        self.registry.requires_inspection(implementation)
+    }
+
     /// HTTP hook version of a registered middleware, without a network call. See
     /// [`MiddlewareRegistry::http_hook_version_of`].
     #[must_use]