feat: add integration and pipeline tests for set_filter_state

- Add 3 integration tests for xDS config parsing (Istio waypoint)
- Add 5 HTTP pipeline tests for filter application
- Fix skip_if_empty to handle '-' default values
- Add :authority and :method pseudo-headers to RequestContext

Signed-off-by: Eeshu-Yadav <[email protected]>

Author: Eeshu-Yadav

orion-lib/src/listeners/http_connection_manager.rs

@@ -274,6 +274,13 @@ fn apply_set_filter_state_filters<B>(
     use orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SharedWithUpstream as ConfigSharedWithUpstream;
     use crate::SharedWithUpstream as RuntimeSharedWithUpstream;
 
+    tracing::trace!(
+        "Applying set_filter_state filters: method={}, uri={}, filters={}",
+        request.method(),
+        request.uri(),
+        http_filters.len()
+    );
+
     // Build request context from HTTP request
     let mut ctx = RequestContext::new();
 
@@ -284,9 +291,19 @@ fn apply_set_filter_state_filters<B>(
         }
     }
 
+    if ctx.get_header(":authority").is_none() {
+        if let Some(host) = ctx.get_header("host") {
+            let host = host.clone();
+            ctx = ctx.with_header(":authority", host);
+        }
+    }
+    
     // Extract method
     ctx = ctx.with_method(request.method().as_str());
 
+    // Also add :method pseudo-header for compatibility with %REQ(:method)%
+    ctx = ctx.with_header(":method", request.method().as_str());
+    
     // Extract path
     ctx = ctx.with_path(request.uri().path());
 
@@ -308,6 +325,12 @@ fn apply_set_filter_state_filters<B>(
 
         // Check if this is a SetFilterState filter
         if let Some(HttpFilterValue::SetFilterState(sfs_config)) = &filter.filter {
+            tracing::debug!(
+                "Processing SetFilterState filter '{}' with {} entries",
+                filter.name,
+                sfs_config.on_request_headers.len()
+            );
+            
             // Process each on_request_headers entry
             for value_config in &sfs_config.on_request_headers {
                 // 1. Evaluate format string
@@ -325,7 +348,7 @@ fn apply_set_filter_state_filters<B>(
                 };
 
                 // 2. Check skip_if_empty
-                if value_config.skip_if_empty && evaluated_value.is_empty() {
+                if value_config.skip_if_empty && (evaluated_value.is_empty() || evaluated_value == "-") {
                     tracing::debug!(
                         "Skipping filter state key '{}' due to empty value",
                         value_config.object_key
@@ -1311,6 +1334,10 @@ fn apply_authorization_rules<B>(rbac: &HttpRbac, req: &Request<B>) -> FilterDeci
 #[cfg(test)]
 mod tests {
     use orion_configuration::config::network_filters::http_connection_manager::MatchHost;
+    use hyper::Method;
+    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+    use crate::filter_state::SharedWithUpstream as RuntimeSharedWithUpstream;
+    use crate::listeners::filter_state::{DownstreamConnectionMetadata, DownstreamMetadata};
 
     use super::*;
 
@@ -1349,4 +1376,328 @@ mod tests {
         let request = Request::builder().header("host", "domain2.com").body(()).unwrap();
         assert_eq!(select_virtual_host(&request, &[vh1.clone(), vh2.clone(), vh3.clone()]), None);
     }
+
+    #[test]
+    fn test_apply_set_filter_state_filters_basic() {
+        use orion_configuration::config::network_filters::http_connection_manager::http_filters::{
+            set_filter_state::{FilterStateValue, FormatString, SharedWithUpstream as ConfigSharedWithUpstream},
+        };
+        use compact_str::CompactString;
+
+        let filter_state = FilterState::new();
+        let format_evaluator = FormatStringEvaluator::new();
+
+        let sfs_config = orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SetFilterState {
+            on_request_headers: vec![
+                FilterStateValue {
+                    object_key: CompactString::from("io.istio.connect_authority"),
+                    format_string: FormatString::Text(CompactString::from("%REQ(:authority)%")),
+                    factory_key: None,
+                    read_only: true,
+                    shared_with_upstream: ConfigSharedWithUpstream::Once,
+                    skip_if_empty: false,
+                },
+            ],
+        };
+
+        let http_filter = Arc::new(HttpFilter {
+            name: CompactString::from("test.set_filter_state"),
+            disabled: false,
+            filter: Some(HttpFilterValue::SetFilterState(sfs_config)),
+        });
+
+        let request = Request::builder()
+            .method(Method::CONNECT)
+            .uri("example.com:443")
+            .header("host", "example.com:443")
+            .body(())
+            .unwrap();
+
+        let downstream_metadata = DownstreamMetadata::new(
+            DownstreamConnectionMetadata::Socket {
+                peer_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 54321),
+                local_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(172, 16, 0, 1)), 15008),
+                original_destination_address: None,
+            },
+            None::<String>,
+        );
+
+        apply_set_filter_state_filters(
+            &request,
+            &downstream_metadata,
+            &[http_filter],
+            &filter_state,
+            &format_evaluator,
+        );
+
+        let (value, read_only, shared_with_upstream) = filter_state.get_with_metadata("io.istio.connect_authority").unwrap();
+        assert_eq!(value.as_str(), "example.com:443");
+        assert!(read_only);
+        assert!(matches!(shared_with_upstream, RuntimeSharedWithUpstream::Once));
+    }
+
+    #[test]
+    fn test_apply_set_filter_state_filters_skip_if_empty() {
+        use orion_configuration::config::network_filters::http_connection_manager::http_filters::{
+            set_filter_state::{FilterStateValue, FormatString, SharedWithUpstream as ConfigSharedWithUpstream},
+        };
+        use compact_str::CompactString;
+
+        let filter_state = FilterState::new();
+        let format_evaluator = FormatStringEvaluator::new();
+
+        // Create config with skip_if_empty=true for a missing header
+        let sfs_config = orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SetFilterState {
+            on_request_headers: vec![
+                FilterStateValue {
+                    object_key: CompactString::from("test.missing_header"),
+                    format_string: FormatString::Text(CompactString::from("%REQ(x-missing-header)%")),
+                    factory_key: None,
+                    read_only: false,
+                    shared_with_upstream: ConfigSharedWithUpstream::None,
+                    skip_if_empty: true,
+                },
+            ],
+        };
+
+        let http_filter = Arc::new(HttpFilter {
+            name: CompactString::from("test.set_filter_state"),
+            disabled: false,
+            filter: Some(HttpFilterValue::SetFilterState(sfs_config)),
+        });
+
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/test")
+            .body(())
+            .unwrap();
+
+        let downstream_metadata = DownstreamMetadata::new(
+            DownstreamConnectionMetadata::Socket {
+                peer_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 12345),
+                local_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
+                original_destination_address: None,
+            },
+            None::<String>,
+        );
+
+        apply_set_filter_state_filters(
+            &request,
+            &downstream_metadata,
+            &[http_filter],
+            &filter_state,
+            &format_evaluator,
+        );
+
+        assert!(filter_state.get("test.missing_header").is_err());
+    }
+
+    #[test]
+    fn test_apply_set_filter_state_filters_client_address() {
+        use orion_configuration::config::network_filters::http_connection_manager::http_filters::{
+            set_filter_state::{FilterStateValue, FormatString, SharedWithUpstream as ConfigSharedWithUpstream},
+        };
+        use compact_str::CompactString;
+
+        let filter_state = FilterState::new();
+        let format_evaluator = FormatStringEvaluator::new();
+
+        // Create config for client address
+        let sfs_config = orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SetFilterState {
+            on_request_headers: vec![
+                FilterStateValue {
+                    object_key: CompactString::from("io.istio.client_address"),
+                    format_string: FormatString::Text(CompactString::from("%DOWNSTREAM_REMOTE_ADDRESS%")),
+                    factory_key: None,
+                    read_only: false,
+                    shared_with_upstream: ConfigSharedWithUpstream::Transitive,
+                    skip_if_empty: false,
+                },
+            ],
+        };
+
+        let http_filter = Arc::new(HttpFilter {
+            name: CompactString::from("test.set_filter_state"),
+            disabled: false,
+            filter: Some(HttpFilterValue::SetFilterState(sfs_config)),
+        });
+
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/")
+            .body(())
+            .unwrap();
+
+        let downstream_metadata = DownstreamMetadata::new(
+            DownstreamConnectionMetadata::Socket {
+                peer_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 45678),
+                local_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 8080),
+                original_destination_address: None,
+            },
+            None::<String>,
+        );
+
+        // Apply filters
+        apply_set_filter_state_filters(
+            &request,
+            &downstream_metadata,
+            &[http_filter],
+            &filter_state,
+            &format_evaluator,
+        );
+
+        // Verify the client address was captured
+        let (value, read_only, shared_with_upstream) = filter_state.get_with_metadata("io.istio.client_address").unwrap();
+        assert_eq!(value.as_str(), "192.168.1.100:45678");
+        assert!(!read_only);
+        assert!(matches!(shared_with_upstream, RuntimeSharedWithUpstream::Transitive));
+    }
+
+    #[test]
+    fn test_apply_set_filter_state_filters_multiple_entries() {
+        use orion_configuration::config::network_filters::http_connection_manager::http_filters::{
+            set_filter_state::{FilterStateValue, FormatString, SharedWithUpstream as ConfigSharedWithUpstream},
+        };
+        use compact_str::CompactString;
+
+        let filter_state = FilterState::new();
+        let format_evaluator = FormatStringEvaluator::new();
+
+        // Create config with multiple filter state entries
+        let sfs_config = orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SetFilterState {
+            on_request_headers: vec![
+                FilterStateValue {
+                    object_key: CompactString::from("io.istio.connect_authority"),
+                    format_string: FormatString::Text(CompactString::from("%REQ(:authority)%")),
+                    factory_key: None,
+                    read_only: true,
+                    shared_with_upstream: ConfigSharedWithUpstream::Once,
+                    skip_if_empty: false,
+                },
+                FilterStateValue {
+                    object_key: CompactString::from("io.istio.client_address"),
+                    format_string: FormatString::Text(CompactString::from("%DOWNSTREAM_REMOTE_ADDRESS%")),
+                    factory_key: None,
+                    read_only: false,
+                    shared_with_upstream: ConfigSharedWithUpstream::Transitive,
+                    skip_if_empty: false,
+                },
+                FilterStateValue {
+                    object_key: CompactString::from("test.method"),
+                    format_string: FormatString::Text(CompactString::from("%REQ(:method)%")),
+                    factory_key: None,
+                    read_only: false,
+                    shared_with_upstream: ConfigSharedWithUpstream::None,
+                    skip_if_empty: false,
+                },
+            ],
+        };
+
+        let http_filter = Arc::new(HttpFilter {
+            name: CompactString::from("test.set_filter_state"),
+            disabled: false,
+            filter: Some(HttpFilterValue::SetFilterState(sfs_config)),
+        });
+
+        let request = Request::builder()
+            .method(Method::CONNECT)
+            .uri("dest.example.com:443")
+            .header("host", "dest.example.com:443")
+            .body(())
+            .unwrap();
+
+        let downstream_metadata = DownstreamMetadata::new(
+            DownstreamConnectionMetadata::Socket {
+                peer_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 1, 2, 3)), 12345),
+                local_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(172, 31, 0, 1)), 15008),
+                original_destination_address: None,
+            },
+            None::<String>,
+        );
+
+        // Apply filters
+        apply_set_filter_state_filters(
+            &request,
+            &downstream_metadata,
+            &[http_filter],
+            &filter_state,
+            &format_evaluator,
+        );
+
+        // Verify all three entries were set
+        assert_eq!(filter_state.get("io.istio.connect_authority").unwrap().as_str(), "dest.example.com:443");
+        assert_eq!(filter_state.get("io.istio.client_address").unwrap().as_str(), "10.1.2.3:12345");
+        assert_eq!(filter_state.get("test.method").unwrap().as_str(), "CONNECT");
+
+        // Verify metadata
+        let (_, read_only1, shared1) = filter_state.get_with_metadata("io.istio.connect_authority").unwrap();
+        assert!(read_only1);
+        assert!(matches!(shared1, RuntimeSharedWithUpstream::Once));
+
+        let (_, read_only2, shared2) = filter_state.get_with_metadata("io.istio.client_address").unwrap();
+        assert!(!read_only2);
+        assert!(matches!(shared2, RuntimeSharedWithUpstream::Transitive));
+
+        let (_, read_only3, shared3) = filter_state.get_with_metadata("test.method").unwrap();
+        assert!(!read_only3);
+        assert!(matches!(shared3, RuntimeSharedWithUpstream::None));
+    }
+
+    #[test]
+    fn test_apply_set_filter_state_filters_disabled_filter() {
+        use orion_configuration::config::network_filters::http_connection_manager::http_filters::{
+            set_filter_state::{FilterStateValue, FormatString, SharedWithUpstream as ConfigSharedWithUpstream},
+        };
+        use compact_str::CompactString;
+
+        let filter_state = FilterState::new();
+        let format_evaluator = FormatStringEvaluator::new();
+
+        let sfs_config = orion_configuration::config::network_filters::http_connection_manager::http_filters::set_filter_state::SetFilterState {
+            on_request_headers: vec![
+                FilterStateValue {
+                    object_key: CompactString::from("test.disabled"),
+                    format_string: FormatString::Text(CompactString::from("value")),
+                    factory_key: None,
+                    read_only: false,
+                    shared_with_upstream: ConfigSharedWithUpstream::None,
+                    skip_if_empty: false,
+                },
+            ],
+        };
+
+        // Create a disabled filter
+        let http_filter = Arc::new(HttpFilter {
+            name: CompactString::from("test.set_filter_state"),
+            disabled: true, // Disabled!
+            filter: Some(HttpFilterValue::SetFilterState(sfs_config)),
+        });
+
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/")
+            .body(())
+            .unwrap();
+
+        let downstream_metadata = DownstreamMetadata::new(
+            DownstreamConnectionMetadata::Socket {
+                peer_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 12345),
+                local_address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
+                original_destination_address: None,
+            },
+            None::<String>,
+        );
+
+        // Apply filters
+        apply_set_filter_state_filters(
+            &request,
+            &downstream_metadata,
+            &[http_filter],
+            &filter_state,
+            &format_evaluator,
+        );
+
+        // Verify nothing was set (filter was disabled)
+        assert!(filter_state.get("test.disabled").is_err());
+    }
 }