praxis-proxy-filter 0.5.6

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

//! The [`HttpFilter`] trait definition.
//!
//! Every HTTP filter implements this trait.

use async_trait::async_trait;
use bytes::Bytes;
use praxis_core::config::InsecureOptions;

pub(crate) use crate::context::HttpFilterContext;
use crate::{
    actions::{FilterAction, SelectedUpstreamBodyOutcome},
    body::{BodyAccess, BodyMode},
    builtins::http::payload_processing::compression_config::CompressionConfig,
    pipeline::FilterPipeline,
};

// -----------------------------------------------------------------------------
// Backward-compatible Aliases
// -----------------------------------------------------------------------------

/// Backward-compatible alias for [`HttpFilter`].
pub type Filter = dyn HttpFilter;

/// Backward-compatible alias for [`HttpFilterContext`].
///
/// [`HttpFilterContext`]: crate::context::HttpFilterContext
pub type FilterContext<'a> = HttpFilterContext<'a>;

// -----------------------------------------------------------------------------
// HttpFilter Trait
// -----------------------------------------------------------------------------

/// A filter that participates in HTTP request/response processing.
///
/// # Example
///
/// ```
/// use async_trait::async_trait;
/// use praxis_filter::{FilterAction, FilterError, HttpFilter, HttpFilterContext};
///
/// struct NoopFilter;
///
/// #[async_trait]
/// impl HttpFilter for NoopFilter {
///     fn name(&self) -> &'static str {
///         "noop"
///     }
///
///     async fn on_request(
///         &self,
///         _ctx: &mut HttpFilterContext<'_>,
///     ) -> Result<FilterAction, FilterError> {
///         Ok(FilterAction::Continue)
///     }
/// }
///
/// let filter = NoopFilter;
/// assert_eq!(filter.name(), "noop");
/// ```
#[async_trait]
pub trait HttpFilter: Send + Sync {
    /// Unique name identifying this filter type (e.g. `"router"`, `"rate_limit"`).
    fn name(&self) -> &'static str;

    /// Called for each incoming request, in pipeline order.
    async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError>;

    // -------------------------------------------------------------------------
    // Pipeline Capability Declarations
    // -------------------------------------------------------------------------

    /// Whether this filter can assign [`HttpFilterContext::cluster`].
    ///
    /// Pipeline validation uses this to detect `load_balancer`
    /// configurations that cannot receive a cluster selection and to
    /// reject multiple competing cluster selectors before the same
    /// load balancer. Filters that route to named clusters should
    /// override this.
    ///
    /// [`HttpFilterContext::cluster`]: crate::HttpFilterContext::cluster
    fn selects_cluster(&self) -> bool {
        false
    }

    /// The cluster names this filter may select at runtime.
    ///
    /// Return every configured cluster target so pipeline validation
    /// can verify that downstream load balancers define matching
    /// clusters. Filters without static cluster targets can leave the
    /// default empty list.
    fn selected_clusters(&self) -> Vec<String> {
        Vec::new()
    }

    /// The cluster names this filter can load balance.
    ///
    /// Load-balancing filters override this so validation can compare
    /// selector outputs against the configured upstream cluster set.
    fn load_balancer_clusters(&self) -> Vec<String> {
        Vec::new()
    }

    /// Whether this filter may select a streaming sub-request response.
    ///
    /// Pipeline validation uses this declaration to reject response
    /// pipelines that require [`BodyMode::StreamBuffer`], which is
    /// incompatible with incremental terminal delivery. Filters that can
    /// call [`HttpFilterContext::set_subrequest_response_mode`] with
    /// `Streaming` must override this method.
    ///
    /// A runtime guard still validates an actual streaming terminal action,
    /// protecting dynamically registered or incorrectly declared filters.
    ///
    /// [`HttpFilterContext::set_subrequest_response_mode`]: crate::HttpFilterContext::set_subrequest_response_mode
    fn may_select_streaming_subrequest_response(&self) -> bool {
        false
    }

    /// Whether this filter may return a terminal [`FilterAction`] — one that
    /// short-circuits the request phase with a response and no upstream
    /// (`TerminalResponse` / `StreamingTerminalResponse`).
    ///
    /// Pipeline validation uses this declaration to reject terminal filters from
    /// outbound chains bound into a filtered sub-request: that executor forwards
    /// to a resolved upstream and cannot surface such a response, so it would
    /// drop the action and then error that no upstream resolved. Any filter that
    /// can return a terminal action must override this, so detection is not
    /// limited to the hard-coded builtin terminal names in [`TERMINAL_FILTERS`].
    ///
    /// [`FilterAction`]: crate::FilterAction
    /// [`TERMINAL_FILTERS`]: praxis_core::config::TERMINAL_FILTERS
    fn produces_terminal_response(&self) -> bool {
        false
    }

    /// Visit pipelines owned by this filter.
    ///
    /// Framework filters that embed nested pipelines override this hook so
    /// server-provided runtime resources are propagated into those pipelines.
    /// Ordinary filters should use the default no-op implementation.
    #[doc(hidden)]
    fn visit_nested_pipelines(&mut self, _visitor: &mut dyn FnMut(&mut FilterPipeline)) {}

    /// Called for each response, in reverse pipeline order.
    async fn on_response(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
        let _ = ctx;
        Ok(FilterAction::Continue)
    }

    // -------------------------------------------------------------------------
    // Body Access Declarations
    // -------------------------------------------------------------------------

    /// Declares what access this filter needs to request bodies.
    ///
    /// Return [`BodyAccess::None`] (the default) when the filter
    /// does not inspect or modify request bodies. Return
    /// [`BodyAccess::ReadOnly`] to receive body chunks in
    /// [`on_request_body`] without modification rights, or
    /// [`BodyAccess::ReadWrite`] to mutate body bytes in place.
    ///
    /// [`on_request_body`]: HttpFilter::on_request_body
    fn request_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares what access this filter needs to response bodies.
    ///
    /// Return [`BodyAccess::None`] (the default) when the filter
    /// does not inspect or modify response bodies. Return
    /// [`BodyAccess::ReadOnly`] to receive body chunks in
    /// [`on_response_body`] without modification rights, or
    /// [`BodyAccess::ReadWrite`] to mutate body bytes in place.
    ///
    /// [`on_response_body`]: HttpFilter::on_response_body
    fn response_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares what access this filter needs to the request body during
    /// the selected-upstream phase.
    ///
    /// The selected-upstream request-body phase runs after the request
    /// phase has chosen an upstream cluster, giving a filter a final
    /// opportunity to inspect or rewrite the request body for the
    /// specific upstream it is about to be forwarded to. Return
    /// [`BodyAccess::None`] (the default) to opt out,
    /// [`BodyAccess::ReadOnly`] to observe the body in
    /// [`on_selected_upstream_request_body`], or
    /// [`BodyAccess::ReadWrite`] to mutate it.
    ///
    /// A participating filter must declare a bounded
    /// [`BodyMode::StreamBuffer`] via [`request_body_mode`] so the
    /// complete body is buffered before this phase runs; the phase reuses
    /// the same request-body delivery mode rather than defining its own.
    /// Pipeline validation rejects a participant whose
    /// [`request_body_mode`] is not a bounded `StreamBuffer`.
    ///
    /// [`on_selected_upstream_request_body`]: HttpFilter::on_selected_upstream_request_body
    /// [`request_body_mode`]: HttpFilter::request_body_mode
    /// [`BodyMode::StreamBuffer`]: crate::BodyMode::StreamBuffer
    fn selected_upstream_request_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares the delivery mode for request body chunks.
    ///
    /// [`BodyMode::Stream`] (the default) delivers chunks as they
    /// arrive with minimal latency. [`BodyMode::StreamBuffer`]
    /// accumulates chunks and defers forwarding until the filter
    /// calls [`FilterAction::Release`] or end-of-stream; use this
    /// when the filter needs the full body before making a decision
    /// (e.g. JSON parsing for routing).
    ///
    /// [`FilterAction::Release`]: crate::FilterAction::Release
    fn request_body_mode(&self) -> BodyMode {
        BodyMode::Stream
    }

    /// Declares the delivery mode for response body chunks.
    ///
    /// [`BodyMode::Stream`] (the default) delivers chunks as they
    /// arrive with minimal latency. [`BodyMode::StreamBuffer`]
    /// accumulates chunks and defers forwarding until the filter
    /// calls [`FilterAction::Release`] or end-of-stream; use this
    /// when the filter needs the complete response body (e.g.
    /// response persistence).
    ///
    /// [`FilterAction::Release`]: crate::FilterAction::Release
    fn response_body_mode(&self) -> BodyMode {
        BodyMode::Stream
    }

    /// Whether this filter needs the original request context
    /// during body phases.
    ///
    /// When `true`, the pipeline preserves the original request
    /// headers and URI so they are available in [`on_request_body`]
    /// and [`on_response_body`]. Enable this when body-phase
    /// decisions depend on request metadata (method, path, headers).
    ///
    /// [`on_request_body`]: HttpFilter::on_request_body
    /// [`on_response_body`]: HttpFilter::on_response_body
    fn needs_request_context(&self) -> bool {
        false
    }

    /// Filesystem paths this filter reads its configuration from, beyond the
    /// main Praxis config.
    ///
    /// The config watcher uses this to decide which files should trigger a
    /// reload. A filter whose configuration lives entirely inside the Praxis
    /// config has nothing to declare.
    ///
    /// Paths are returned as configured, not canonicalized: the watcher does its
    /// own resolution because it has to handle symlinks and relative paths
    /// consistently with how it already treats the main config.
    fn referenced_files(&self) -> Vec<std::path::PathBuf> {
        Vec::new()
    }

    /// Apply global [`InsecureOptions`] to this filter.
    ///
    /// Filters that support insecure overrides (e.g. CSRF
    /// log-only mode) override this.
    ///
    /// [`InsecureOptions`]: praxis_core::config::InsecureOptions
    fn apply_insecure_options(&self, _options: &InsecureOptions) {}

    /// Returns the compression configuration if this filter enables
    /// response compression.
    ///
    /// Return `Some` to activate response compression with the
    /// given algorithm and level settings. Return `None` (the
    /// default) to leave compression unmodified. Only one filter
    /// in a pipeline should return `Some`.
    fn compression_config(&self) -> Option<&CompressionConfig> {
        None
    }

    // -------------------------------------------------------------------------
    // Body Hooks
    // -------------------------------------------------------------------------

    /// Called for each chunk of request body data, in pipeline order.
    ///
    /// `body` contains the current chunk (`None` when empty).
    /// `end_of_stream` is `true` on the final chunk. Filters may
    /// safely modify `body` before `end_of_stream` when using
    /// [`BodyAccess::ReadWrite`]; buffered modes guarantee all
    /// bytes arrive in a single call with `end_of_stream = true`.
    /// Return [`FilterAction::Reject`] to abort with an error
    /// response.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The
    /// pipeline converts errors into 500 responses.
    ///
    /// [`FilterAction::Reject`]: crate::FilterAction::Reject
    async fn on_request_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
        end_of_stream: bool,
    ) -> Result<FilterAction, FilterError> {
        let _ = (ctx, body, end_of_stream);
        Ok(FilterAction::Continue)
    }

    /// Called for each chunk of response body data, in reverse
    /// pipeline order.
    ///
    /// `body` contains the current chunk (`None` when empty).
    /// `end_of_stream` is `true` on the final chunk. Filters may
    /// safely modify `body` before `end_of_stream` when using
    /// [`BodyAccess::ReadWrite`]; buffered modes guarantee all
    /// bytes arrive in a single call with `end_of_stream = true`.
    /// Return [`FilterAction::Reject`] to abort with an error
    /// response.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The
    /// pipeline converts errors into 502 responses.
    ///
    /// [`FilterAction::Reject`]: crate::FilterAction::Reject
    fn on_response_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
        end_of_stream: bool,
    ) -> Result<FilterAction, FilterError> {
        let _ = (ctx, body, end_of_stream);
        Ok(FilterAction::Continue)
    }

    /// Called once with the fully buffered request body after the request
    /// phase has selected an upstream cluster.
    ///
    /// Runs only for filters that declare
    /// [`selected_upstream_request_body_access`] other than
    /// [`BodyAccess::None`], in pipeline order, and only if the filter
    /// executed during the request phase. `body` holds the complete
    /// request body (`None` when the request had no body). Filters that
    /// declared [`BodyAccess::ReadWrite`] may mutate `body` in place;
    /// this is a working value distinct from any canonical body the
    /// pipeline retains. Return [`SelectedUpstreamBodyOutcome::Reject`] to
    /// abort with an error response; the pipeline stops and does not call
    /// later selected-upstream body filters.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The pipeline
    /// honors the filter's `failure_mode`: a closed filter's error aborts
    /// the request, an open filter's error is logged and treated as
    /// [`SelectedUpstreamBodyOutcome::Continue`].
    ///
    /// [`selected_upstream_request_body_access`]: HttpFilter::selected_upstream_request_body_access
    /// [`SelectedUpstreamBodyOutcome::Reject`]: crate::SelectedUpstreamBodyOutcome::Reject
    /// [`SelectedUpstreamBodyOutcome::Continue`]: crate::SelectedUpstreamBodyOutcome::Continue
    async fn on_selected_upstream_request_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
    ) -> Result<SelectedUpstreamBodyOutcome, FilterError> {
        let _ = (ctx, body);
        Ok(SelectedUpstreamBodyOutcome::Continue)
    }
}

/// Boxed error type for filter results.
pub type FilterError = Box<dyn std::error::Error + Send + Sync>;

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    reason = "tests"
)]
mod tests {
    use async_trait::async_trait;

    use super::*;

    #[tokio::test]
    async fn default_on_response_returns_continue() {
        let filter = MinimalFilter;
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        let action = filter.on_response(&mut ctx).await.unwrap();

        assert!(
            matches!(action, FilterAction::Continue),
            "default on_response should return Continue"
        );
    }

    #[test]
    fn default_body_access_is_none() {
        let filter = MinimalFilter;
        assert_eq!(
            filter.request_body_access(),
            BodyAccess::None,
            "default request body access should be None"
        );
        assert_eq!(
            filter.response_body_access(),
            BodyAccess::None,
            "default response body access should be None"
        );
        assert_eq!(
            filter.selected_upstream_request_body_access(),
            BodyAccess::None,
            "default selected-upstream request body access should be None"
        );
        assert_eq!(
            filter.request_body_mode(),
            BodyMode::Stream,
            "default request body mode should be Stream"
        );
        assert_eq!(
            filter.response_body_mode(),
            BodyMode::Stream,
            "default response body mode should be Stream"
        );
        assert!(
            !filter.needs_request_context(),
            "default needs_request_context should be false"
        );
    }

    #[test]
    fn default_referenced_files_is_empty() {
        let filter = MinimalFilter;
        assert!(
            filter.referenced_files().is_empty(),
            "a filter with no external config must declare nothing"
        );
    }

    #[test]
    fn default_cluster_capabilities_are_empty() {
        let filter = MinimalFilter;
        assert!(!filter.selects_cluster(), "default selects_cluster should be false");
        assert!(
            filter.selected_clusters().is_empty(),
            "default selected_clusters should be empty"
        );
        assert!(
            filter.load_balancer_clusters().is_empty(),
            "default load_balancer_clusters should be empty"
        );
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    #[tokio::test]
    async fn default_body_hooks_return_continue() {
        let filter = MinimalFilter;
        let req = crate::test_utils::make_request(http::Method::POST, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut body = Some(Bytes::from_static(b"chunk"));

        let request_action = filter.on_request_body(&mut ctx, &mut body, true).await.unwrap();
        assert!(
            matches!(request_action, FilterAction::Continue),
            "default on_request_body should return Continue"
        );

        let response_action = filter.on_response_body(&mut ctx, &mut body, true).unwrap();
        assert!(
            matches!(response_action, FilterAction::Continue),
            "default on_response_body should return Continue"
        );

        let selected_action = filter
            .on_selected_upstream_request_body(&mut ctx, &mut body)
            .await
            .unwrap();
        assert!(
            matches!(selected_action, SelectedUpstreamBodyOutcome::Continue),
            "default on_selected_upstream_request_body should return Continue"
        );
        assert_eq!(
            body.as_deref(),
            Some(b"chunk".as_slice()),
            "defaults must not touch the body"
        );
    }

    #[test]
    fn default_streaming_selection_is_disabled() {
        let filter = MinimalFilter;
        assert!(
            !filter.may_select_streaming_subrequest_response(),
            "filters must opt in to streaming selection"
        );
    }

    /// Minimal filter for verifying trait defaults.
    struct MinimalFilter;

    #[async_trait]
    impl HttpFilter for MinimalFilter {
        fn name(&self) -> &'static str {
            "minimal"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }
    }
}