praxis-proxy-filter 0.5.4

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024 Praxis Contributors

//! Pipeline construction and ordering diagnostics.
//!
//! Provides [`FilterPipeline::build`] and [`FilterPipeline::build_with_chains`],
//! which instantiate filters from [`FilterEntry`] slices via the
//! [`FilterRegistry`], compute [`BodyCapabilities`], and run ordering
//! validation checks. Branch-aware builds delegate to
//! [`build_branch::resolve_chain_filters`] for recursive resolution.
//!
//! [`FilterEntry`]: praxis_core::config::FilterEntry
//! [`FilterRegistry`]: crate::registry::FilterRegistry
//! [`BodyCapabilities`]: crate::body::BodyCapabilities
//! [`build_branch::resolve_chain_filters`]: super::build_branch::resolve_chain_filters

use std::{collections::HashMap, mem, sync::Arc};

use praxis_core::{
    config::{FilterEntry, SkipPipelineChecks},
    id::IdGenerator,
    time::SystemTimeSource,
};
use tracing::{debug, warn};

use super::{
    FilterPipeline,
    body::{body_filter_indices, compute_body_capabilities},
    filter::PipelineFilter,
};
use crate::{FilterError, any_filter::AnyFilter, registry::FilterRegistry};

// -----------------------------------------------------------------------------
// FilterPipeline Factory
// -----------------------------------------------------------------------------

impl FilterPipeline {
    /// Build a pipeline by instantiating each filter entry via the registry.
    ///
    /// Conditions are moved out of entries via [`mem::take`] to avoid
    /// cloning. After this call, each entry's condition vecs are empty.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if any filter fails to instantiate.
    pub fn build(entries: &mut [FilterEntry], registry: &FilterRegistry) -> Result<Self, FilterError> {
        let mut filters = Vec::with_capacity(entries.len());
        for (filter_id, entry) in entries.iter_mut().enumerate() {
            let filter = registry.create(&entry.filter_type, &entry.config)?;
            warn_tcp_unsupported_fields(&filter, entry);
            let has_conditions = !entry.conditions.is_empty() || !entry.response_conditions.is_empty();
            debug!(
                filter = filter.name(),
                conditions = has_conditions,
                "filter added to pipeline"
            );
            let mut pf = PipelineFilter::new(
                filter_id,
                filter,
                mem::take(&mut entry.conditions),
                mem::take(&mut entry.response_conditions),
            );
            pf.failure_mode = entry.failure_mode;
            pf.name = entry.name.as_ref().map(|n| Arc::from(n.as_str()));
            filters.push(pf);
        }
        Ok(Self::from_filters(filters))
    }

    /// Build a pipeline with branch chain resolution.
    ///
    /// Like [`build`], but also resolves `branch_chains` on each
    /// filter entry into runtime `ResolvedBranch` types using
    /// the provided chain lookup table.
    ///
    /// The `chains` parameter is the **top-level** chain lookup
    /// table (all `filter_chains` from the config), used to
    /// resolve `ChainRef::Named` entries inside branch
    /// configurations. The actual filters for this pipeline come
    /// from `entries`.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if any filter fails to instantiate
    /// or any branch chain reference is unresolvable.
    ///
    /// [`build`]: FilterPipeline::build
    pub fn build_with_chains(
        entries: &mut [FilterEntry],
        registry: &FilterRegistry,
        chains: &HashMap<&str, &[FilterEntry]>,
    ) -> Result<Self, FilterError> {
        let filters = super::build_branch::resolve_chain_filters(entries, registry, chains, 0)?;
        Ok(Self::from_filters(filters))
    }

    /// Create a pipeline from an already-resolved filter list.
    fn from_filters(filters: Vec<PipelineFilter>) -> Self {
        let body_capabilities = compute_body_capabilities(&filters);
        let compression = extract_compression_config(&filters);
        let may_select_streaming_subrequest_response = filters_may_select_streaming_subrequest_response(&filters);
        let (request_body_filter_indices, response_body_filter_indices) = body_filter_indices(&filters);
        let id_generator = Arc::new(IdGenerator::new());
        let time_source: Arc<dyn praxis_core::time::TimeSource> = Arc::new(SystemTimeSource);
        let mut pipeline = Self {
            body_capabilities,
            compression,
            filters,
            request_body_filter_indices,
            response_body_filter_indices,
            health_registry: None,
            id_generator: Arc::clone(&id_generator),
            kv_stores: None,
            session_stores: None,
            pipeline_extensions: Vec::new(),
            record_filter_duration_metrics: false,
            subrequest_client: None,
            may_select_streaming_subrequest_response,
            time_source: Arc::clone(&time_source),
            request_body_ceiling: None,
            response_body_ceiling: None,
        };
        pipeline.set_id_generator(id_generator);
        pipeline.set_time_source(time_source);
        pipeline
    }

    /// Validate the pipeline for structural misconfigurations that
    /// would cause runtime failures (502s, unreachable filters,
    /// cluster mismatches).
    ///
    /// Individual checks can be skipped via [`SkipPipelineChecks`]
    /// flags. Use [`SkipPipelineChecks::default()`] to run all checks.
    ///
    /// ```
    /// use praxis_core::config::SkipPipelineChecks;
    /// use praxis_filter::{FailureMode, FilterEntry, FilterPipeline, FilterRegistry};
    ///
    /// let registry = FilterRegistry::with_builtins();
    /// let mut entries = vec![FilterEntry {
    ///     branch_chains: None,
    ///     filter_type: "load_balancer".into(),
    ///     config: serde_yaml::from_str("clusters:\n  - name: web\n    endpoints: [\"10.0.0.1:80\"]")
    ///         .unwrap(),
    ///     conditions: vec![],
    ///     name: None,
    ///     response_conditions: vec![],
    ///     failure_mode: FailureMode::default(),
    /// }];
    /// let pipeline = FilterPipeline::build(&mut entries, &registry).unwrap();
    /// let no_skip = SkipPipelineChecks::default();
    /// let errors = pipeline.ordering_errors(&entries, false, &no_skip);
    /// assert!(
    ///     errors
    ///         .iter()
    ///         .any(|e| e.contains("without a preceding router"))
    /// );
    /// ```
    ///
    /// [`build`]: FilterPipeline::build
    /// [`SkipPipelineChecks`]: praxis_core::config::SkipPipelineChecks
    #[expect(
        clippy::too_many_lines,
        reason = "streaming capability check adds one validation pass"
    )]
    pub fn ordering_errors(
        &self,
        entries: &[FilterEntry],
        allow_open_security: bool,
        skip: &SkipPipelineChecks,
    ) -> Vec<String> {
        let names: Vec<&str> = self.filters.iter().map(|pf| pf.filter.name()).collect();

        let mut errors = Vec::new();

        if !skip.lb_without_router {
            super::checks::check_lb_without_cluster_selector(&self.filters, &mut errors);
        }
        if !skip.unreachable_filters {
            super::checks::check_unconditional_static_response(&names, &self.filters, &mut errors);
        }
        if !skip.conditional_security {
            super::checks::check_conditional_security(&names, &self.filters, &mut errors);
        }
        super::checks::check_open_security_filters(&names, &self.filters, allow_open_security, &mut errors);
        if !skip.duplicate_routers {
            super::checks::check_duplicate_routers(&names, &mut errors);
        }
        if !skip.duplicate_load_balancers {
            super::checks::check_duplicate_load_balancers(&names, &mut errors);
        }
        if !skip.conflicting_cluster_selectors {
            super::checks::check_conflicting_cluster_selectors(&self.filters, &mut errors);
        }
        if !skip.misaligned_clusters {
            super::checks::check_misaligned_clusters(&self.filters, &mut errors);
        }
        if !skip.duplicate_rewrite_filters {
            super::checks::check_duplicate_rewrite_filters(&names, entries, &mut errors);
        }
        super::checks::check_skip_to_bypasses_security(&self.filters, &mut errors);
        super::checks::check_terminal_rejoin_bypasses_security(&self.filters, &mut errors);
        super::checks::check_branch_body_filters(&self.filters, &mut errors);
        super::checks::check_irr_with_router_or_lb(&names, &mut errors);
        if self.may_select_streaming_subrequest_response
            && matches!(
                self.body_capabilities.response_body_mode,
                crate::BodyMode::StreamBuffer { .. }
            )
        {
            errors.push(
                "pipeline contains a filter that may select a streaming sub-request response, \
                 but its response body mode is StreamBuffer"
                    .to_owned(),
            );
        }

        errors
    }

    /// Check for non-fatal ordering advisories.
    ///
    /// Currently detects: all routers conditional with no fallback.
    ///
    /// ```
    /// use praxis_filter::{FailureMode, FilterEntry, FilterPipeline, FilterRegistry};
    ///
    /// let registry = FilterRegistry::with_builtins();
    /// let mut entries = vec![FilterEntry {
    ///     branch_chains: None,
    ///     filter_type: "router".into(),
    ///     config: serde_yaml::from_str("routes:\n  - path_prefix: \"/\"\n    cluster: web").unwrap(),
    ///     conditions: vec![praxis_core::config::Condition::When(
    ///         praxis_core::config::ConditionMatch {
    ///             path: None,
    ///             path_prefix: Some("/api".to_owned()),
    ///             methods: None,
    ///             headers: None,
    ///         },
    ///     )],
    ///     name: None,
    ///     response_conditions: vec![],
    ///     failure_mode: FailureMode::default(),
    /// }];
    /// let pipeline = FilterPipeline::build(&mut entries, &registry).unwrap();
    /// let warnings = pipeline.ordering_warnings();
    /// assert!(
    ///     warnings
    ///         .iter()
    ///         .any(|w| w.contains("all router filters are conditional"))
    /// );
    /// ```
    pub fn ordering_warnings(&self) -> Vec<String> {
        let names: Vec<&str> = self.filters.iter().map(|pf| pf.filter.name()).collect();

        let mut warnings = Vec::new();

        super::checks::check_router_without_lb(&names, &mut warnings);
        super::checks::check_all_routers_conditional(&names, &self.filters, &mut warnings);

        warnings
    }
}

// -----------------------------------------------------------------------------
// Utility Functions
// -----------------------------------------------------------------------------

/// Warn when a TCP filter has conditions or branch chains configured.
///
/// TCP filters do not support conditions or branching; these fields
/// are silently ignored at runtime. Logging at build time helps
/// operators catch misconfigurations.
fn warn_tcp_unsupported_fields(filter: &AnyFilter, entry: &FilterEntry) {
    if !matches!(filter, AnyFilter::Tcp(_)) {
        return;
    }
    if !entry.conditions.is_empty() || !entry.response_conditions.is_empty() {
        warn!(
            filter = filter.name(),
            "TCP filter has conditions that will be ignored; \
             conditions are only evaluated for HTTP filters"
        );
    }
    if entry.branch_chains.is_some() {
        warn!(
            filter = filter.name(),
            "TCP filter has branch_chains that will be ignored; \
             branching is only supported for HTTP filters"
        );
    }
}

/// Scan the filter list for a compression filter and extract its config.
fn extract_compression_config(
    filters: &[PipelineFilter],
) -> Option<crate::builtins::http::payload_processing::compression_config::CompressionConfig> {
    filters.iter().find_map(|pf| match &pf.filter {
        AnyFilter::Http(f) => f.compression_config().cloned(),
        AnyFilter::Tcp(_) => None,
    })
}

/// Recursively detect filters that may select streaming sub-request delivery.
fn filters_may_select_streaming_subrequest_response(filters: &[PipelineFilter]) -> bool {
    filters.iter().any(|pf| {
        let selects_streaming = match &pf.filter {
            AnyFilter::Http(filter) => filter.may_select_streaming_subrequest_response(),
            AnyFilter::Tcp(_) => false,
        };
        selects_streaming
            || pf
                .branches
                .iter()
                .any(|branch| filters_may_select_streaming_subrequest_response(&branch.filters))
    })
}