feat(network): inspect HTTP responses before return

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

@@ -16,6 +16,7 @@ pub const MAX_HEADER_MUTATION_BYTES: usize = 32 * 1024;
 pub enum HeaderAuthority {
     Request,
     Response,
+    ResponseTrailers,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -131,7 +132,7 @@ pub fn apply(
     for mutation in mutations {
         match mutation.operation.as_ref() {
             Some(header_mutation::Operation::Write(write)) => {
-                let name = validate_name(&write.name)?;
+                let name = normalize_name(&write.name)?;
                 validate_authority(authority, MutationKind::Write, &write.name, &name)?;
                 if is_connection_nominated(connection_nominated_headers, &name) {
                     return Err(HeaderMutationError::HopByHop {
@@ -177,7 +178,7 @@ pub fn apply(
                 }
             }
             Some(header_mutation::Operation::Remove(remove)) => {
-                let name = validate_name(&remove.name)?;
+                let name = normalize_name(&remove.name)?;
                 validate_authority(authority, MutationKind::Remove, &remove.name, &name)?;
                 if is_connection_nominated(connection_nominated_headers, &name) {
                     return Err(HeaderMutationError::HopByHop {
@@ -201,7 +202,7 @@ fn enforce_size_limit(mutation_bytes: usize) -> Result<(), HeaderMutationError>
     Ok(())
 }
 
-fn validate_name(name: &str) -> Result<String, HeaderMutationError> {
+pub fn normalize_name(name: &str) -> Result<String, HeaderMutationError> {
     let lower = name.to_ascii_lowercase();
     if lower.is_empty() || !lower.bytes().all(is_name_token_byte) {
         return Err(HeaderMutationError::InvalidName {
@@ -229,6 +230,7 @@ fn validate_authority(
             is_response_protected(normalized_name)
                 || (kind == MutationKind::Write && is_response_remove_only(normalized_name))
         }
+        HeaderAuthority::ResponseTrailers => is_response_protected(normalized_name),
     };
     if protected {
         return Err(HeaderMutationError::Protected {

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

@@ -5,8 +5,15 @@
 
 pub mod headers;
 mod remote;
+mod response;
 mod websocket;
 
+pub use response::{
+    HttpResponseFinish, HttpResponseInvocation, HttpResponseInvocationOutcome,
+    HttpResponseMiddlewareFailure, HttpResponsePreflightInput, HttpResponsePreflightOutcome,
+    HttpResponseSession, MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES,
+};
+
 pub use websocket::{
     WebSocketCoverage, WebSocketCoverageState, WebSocketInvocation, WebSocketInvocationOutcome,
     WebSocketMessageAdmission, WebSocketMessageOutcome, WebSocketMessageType,
@@ -626,6 +633,16 @@ impl MiddlewareDispatch {
             Self::Grpc(service) => service.open_websocket_session(receiver).await,
         }
     }
+
+    async fn open_http_response_pre_return(
+        &self,
+        receiver: tokio::sync::mpsc::Receiver<openshell_core::proto::HttpResponseEvent>,
+    ) -> std::result::Result<HttpResponseResultStream, tonic::Status> {
+        match self {
+            Self::InProcess(service) => service.open_http_response_pre_return(receiver).await,
+            Self::Grpc(service) => service.open_http_response_pre_return(receiver).await,
+        }
+    }
 }
 
 struct MiddlewareServiceState {
@@ -831,6 +848,7 @@ fn validate_payload_limit(source: &str, binding: &MiddlewareBinding) -> Result<u
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum SupportedBinding {
     HttpPreCredentials,
+    HttpResponsePreReturn,
     WebSocketPreCredentials,
 }
 
@@ -846,9 +864,7 @@ fn supported_binding(source: &str, binding: &MiddlewareBinding) -> Result<Suppor
         (
             Some(SupervisorMiddlewareOperation::HttpResponse),
             Some(SupervisorMiddlewarePhase::PreReturn),
-        ) => Err(miette!(
-            "{source} advertises HTTP_RESPONSE/PRE_RETURN, which is not yet supported"
-        )),
+        ) => Ok(SupportedBinding::HttpResponsePreReturn),
         (
             Some(SupervisorMiddlewareOperation::WebsocketMessage),
             Some(SupervisorMiddlewarePhase::PreCredentials),
@@ -3686,7 +3702,7 @@ mod tests {
     }
 
     #[test]
-    fn manifest_rejects_http_response_pre_return_binding_until_dispatch_is_available() {
+    fn manifest_accepts_http_response_pre_return_binding_when_dispatch_is_available() {
         let registration = external_registration(4096);
         let manifest = MiddlewareManifest {
             name: "example/response".into(),
@@ -3700,13 +3716,8 @@ mod tests {
             expected_audience: String::new(),
         };
 
-        let error = validate_external_manifest(&registration, &manifest, 4096, false)
-            .expect_err("HTTP response pre-return binding must remain unavailable");
-        assert!(
-            error
-                .to_string()
-                .contains("HTTP_RESPONSE/PRE_RETURN, which is not yet supported")
-        );
+        validate_external_manifest(&registration, &manifest, 4096, false)
+            .expect("HTTP response pre-return binding is supported");
     }
 
     #[test]

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

@@ -103,6 +103,14 @@ impl GrpcMiddlewareService {
     ) -> std::result::Result<WebSocketResponseStream, Status> {
         self.service.open_websocket_session(receiver).await
     }
+
+    /// Open a remote HTTP response pre-return stream through the gRPC adapter.
+    pub async fn open_http_response_pre_return(
+        &self,
+        receiver: tokio::sync::mpsc::Receiver<HttpResponseEvent>,
+    ) -> std::result::Result<HttpResponseResultStream, Status> {
+        self.service.open_http_response_pre_return(receiver).await
+    }
 }
 
 #[derive(Clone)]