praxis_filter/pipeline/
http.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2024 Praxis Contributors
3
4//! HTTP pipeline execution: request, response, and body filter phases.
5
6use bytes::Bytes;
7use tracing::{debug, trace};
8
9use super::{
10    FilterPipeline,
11    branch::BranchOutcome,
12    check_failure_mode,
13    filter::PipelineFilter,
14    http_utils::{
15        BodyFilterOutcome, accumulate_body_bytes, as_request_body_filter, as_response_body_filter,
16        dispatch_body_result, released_or_continue, run_response_filter, skip_by_response_conditions,
17    },
18};
19use crate::{
20    FilterError, actions::FilterAction, any_filter::AnyFilter, condition::should_execute, context::HttpFilterContext,
21};
22
23// -----------------------------------------------------------------------------
24// FilterPipeline HTTP
25// -----------------------------------------------------------------------------
26
27#[expect(
28    clippy::multiple_inherent_impl,
29    reason = "pipeline concerns are split across modules"
30)]
31impl FilterPipeline {
32    /// Run all HTTP request filters in order.
33    ///
34    /// Tracks which filter indices actually executed so the
35    /// response phase can skip filters that were bypassed
36    /// (e.g. by `SkipTo`).
37    ///
38    /// # Errors
39    ///
40    /// Returns [`FilterError`] if any filter fails.
41    #[expect(clippy::indexing_slicing, reason = "while loop bounds idx")]
42    #[expect(clippy::too_many_lines, reason = "filter identity tracking adds lines per branch")]
43    pub async fn execute_http_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
44        ctx.executed_filter_indices.clear();
45        ctx.executed_filter_indices.resize(self.filters.len(), false);
46        ctx.body_done_indices.clear();
47        ctx.body_done_indices.resize(self.filters.len(), false);
48        let mut idx = 0;
49        while idx < self.filters.len() {
50            let pf = &self.filters[idx];
51            ctx.current_filter_id = Some(pf.filter_id);
52            let result = run_request_filter(pf, ctx).await;
53            ctx.current_filter_id = None;
54            match result? {
55                RequestFilterResult::Skip => {
56                    idx += 1;
57                    continue;
58                },
59                RequestFilterResult::Reject(r) => return Ok(FilterAction::Reject(r)),
60                RequestFilterResult::Continue => {},
61            }
62            ctx.executed_filter_indices[idx] = true;
63            match super::evaluate::evaluate_branches(&pf.branches, ctx).await? {
64                BranchOutcome::Continue => idx += 1,
65                BranchOutcome::Terminal => return Ok(FilterAction::Continue),
66                BranchOutcome::SkipTo(t) => idx = t,
67                BranchOutcome::ReEnter(t) => {
68                    ctx.executed_filter_indices[t..=idx].fill(false);
69                    idx = t;
70                },
71                BranchOutcome::Reject(r) => return Ok(FilterAction::Reject(r)),
72            }
73        }
74        Ok(FilterAction::Continue)
75    }
76
77    /// Run all HTTP response filters in reverse order.
78    ///
79    /// Skips filters that did not execute during the request
80    /// phase (tracked by [`executed_filter_indices`]).
81    ///
82    /// # Errors
83    ///
84    /// Returns [`FilterError`] if any filter fails.
85    ///
86    /// [`executed_filter_indices`]: HttpFilterContext::executed_filter_indices
87    pub async fn execute_http_response(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
88        for (idx, pf) in self.filters.iter().enumerate().rev() {
89            if ctx.executed_filter_indices.get(idx) == Some(&false) {
90                trace!(
91                    filter = pf.filter.name(),
92                    "skipped on_response (not executed in request phase)"
93                );
94                continue;
95            }
96            let http_filter = match &pf.filter {
97                AnyFilter::Http(f) => f.as_ref(),
98                AnyFilter::Tcp(_) => continue,
99            };
100            if skip_by_response_conditions(http_filter, &pf.response_conditions, ctx) {
101                continue;
102            }
103            ctx.current_filter_id = Some(pf.filter_id);
104            trace!(filter = http_filter.name(), "on_response");
105            let action = run_response_filter(http_filter, ctx, pf.failure_mode).await;
106            ctx.current_filter_id = None;
107            let action = action?;
108            if let Some(rejection) = action {
109                return Ok(FilterAction::Reject(rejection));
110            }
111        }
112        Ok(FilterAction::Continue)
113    }
114
115    /// Run all HTTP request body filters in order.
116    ///
117    /// Filters that previously returned [`BodyDone`] are skipped.
118    ///
119    /// # Errors
120    ///
121    /// Returns [`FilterError`] if any body filter fails.
122    ///
123    /// [`BodyDone`]: FilterAction::BodyDone
124    #[expect(clippy::indexing_slicing, reason = "idx bounded by filters.len()")]
125    #[expect(clippy::too_many_lines, reason = "filter identity tracking adds lines per branch")]
126    pub async fn execute_http_request_body(
127        &self,
128        ctx: &mut HttpFilterContext<'_>,
129        body: &mut Option<Bytes>,
130        end_of_stream: bool,
131    ) -> Result<FilterAction, FilterError> {
132        ensure_body_done_indices(ctx, self.filters.len());
133        accumulate_body_bytes(&mut ctx.request_body_bytes, body);
134        let mut released = false;
135        for (idx, pf) in self.filters.iter().enumerate() {
136            if ctx.body_done_indices.get(idx) == Some(&true) {
137                trace!(filter = pf.filter.name(), "skipped body (body_done)");
138                continue;
139            }
140            let Some(http_filter) = as_request_body_filter(&pf.filter, &pf.conditions, ctx.request) else {
141                continue;
142            };
143            ctx.current_filter_id = Some(pf.filter_id);
144            trace!(filter = http_filter.name(), "on_request_body");
145            let outcome = dispatch_body_result(
146                http_filter.on_request_body(ctx, body, end_of_stream).await,
147                http_filter.name(),
148                "request body",
149                pf.failure_mode,
150            );
151            ctx.current_filter_id = None;
152            match outcome? {
153                BodyFilterOutcome::Continue => {},
154                BodyFilterOutcome::Released => released = true,
155                BodyFilterOutcome::BodyDone => {
156                    ctx.body_done_indices[idx] = true;
157                },
158                BodyFilterOutcome::Rejected(r) => return Ok(FilterAction::Reject(r)),
159            }
160        }
161        Ok(released_or_continue(released))
162    }
163
164    /// Run all HTTP response body filters in reverse order.
165    ///
166    /// Filters that previously returned [`BodyDone`] are skipped.
167    ///
168    /// # Errors
169    ///
170    /// Returns [`FilterError`] if any body filter fails.
171    ///
172    /// [`BodyDone`]: FilterAction::BodyDone
173    #[expect(clippy::indexing_slicing, reason = "idx bounded by filters.len()")]
174    pub fn execute_http_response_body(
175        &self,
176        ctx: &mut HttpFilterContext<'_>,
177        body: &mut Option<Bytes>,
178        end_of_stream: bool,
179    ) -> Result<FilterAction, FilterError> {
180        ensure_body_done_indices(ctx, self.filters.len());
181        accumulate_body_bytes(&mut ctx.response_body_bytes, body);
182        let mut released = false;
183        for (idx, pf) in self.filters.iter().enumerate().rev() {
184            if ctx.body_done_indices.get(idx) == Some(&true) {
185                trace!(filter = pf.filter.name(), "skipped body (body_done)");
186                continue;
187            }
188            let Some(http_filter) = as_response_body_filter(&pf.filter, &pf.response_conditions, ctx) else {
189                continue;
190            };
191            ctx.current_filter_id = Some(pf.filter_id);
192            trace!(filter = http_filter.name(), "on_response_body");
193            let outcome = dispatch_body_result(
194                http_filter.on_response_body(ctx, body, end_of_stream),
195                http_filter.name(),
196                "response body",
197                pf.failure_mode,
198            );
199            ctx.current_filter_id = None;
200            match outcome? {
201                BodyFilterOutcome::Continue => {},
202                BodyFilterOutcome::Released => released = true,
203                BodyFilterOutcome::BodyDone => {
204                    ctx.body_done_indices[idx] = true;
205                },
206                BodyFilterOutcome::Rejected(r) => return Ok(FilterAction::Reject(r)),
207            }
208        }
209        Ok(released_or_continue(released))
210    }
211}
212
213// -----------------------------------------------------------------------------
214// Body Done Utilities
215// -----------------------------------------------------------------------------
216
217/// Ensure `body_done_indices` is sized to match the filter count.
218fn ensure_body_done_indices(ctx: &mut HttpFilterContext<'_>, filter_count: usize) {
219    if ctx.body_done_indices.len() != filter_count {
220        ctx.body_done_indices.resize(filter_count, false);
221    }
222}
223
224// -----------------------------------------------------------------------------
225// Request Filter Utilities
226// -----------------------------------------------------------------------------
227
228/// Outcome of running a single request filter.
229enum RequestFilterResult {
230    /// Filter executed successfully; continue pipeline.
231    Continue,
232
233    /// Filter rejected the request.
234    Reject(crate::actions::Rejection),
235
236    /// Filter was skipped (TCP or conditions).
237    Skip,
238}
239
240/// Run a single request filter, handling conditions and tracing.
241async fn run_request_filter(
242    pf: &PipelineFilter,
243    ctx: &mut HttpFilterContext<'_>,
244) -> Result<RequestFilterResult, FilterError> {
245    let http_filter = match &pf.filter {
246        AnyFilter::Http(f) => f.as_ref(),
247        AnyFilter::Tcp(_) => return Ok(RequestFilterResult::Skip),
248    };
249    if !should_execute(&pf.conditions, ctx.request) {
250        trace!(filter = http_filter.name(), "skipped by conditions");
251        return Ok(RequestFilterResult::Skip);
252    }
253    trace!(filter = http_filter.name(), "on_request");
254    match http_filter.on_request(ctx).await {
255        Ok(FilterAction::Continue | FilterAction::Release | FilterAction::BodyDone) => {
256            Ok(RequestFilterResult::Continue)
257        },
258        Ok(FilterAction::Reject(rejection)) => {
259            debug!(
260                filter = http_filter.name(),
261                status = rejection.status,
262                "filter rejected request"
263            );
264            Ok(RequestFilterResult::Reject(rejection))
265        },
266        Err(e) => {
267            check_failure_mode(http_filter.name(), e, "request", pf.failure_mode)?;
268            Ok(RequestFilterResult::Continue)
269        },
270    }
271}
272
273// -----------------------------------------------------------------------------
274// Tests
275// -----------------------------------------------------------------------------
276
277#[cfg(test)]
278#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
279#[allow(
280    clippy::unwrap_used,
281    clippy::expect_used,
282    clippy::indexing_slicing,
283    clippy::panic,
284    reason = "tests"
285)]
286mod tests {
287    use bytes::Bytes;
288    use praxis_core::config::{FailureMode, ResponseCondition, ResponseConditionMatch};
289
290    use super::super::http_utils::{
291        accumulate_body_bytes, dispatch_body_result, released_or_continue, skip_by_response_conditions,
292    };
293    use crate::{FilterAction, FilterError, actions::Rejection};
294
295    #[test]
296    fn accumulate_body_bytes_increments_with_some() {
297        let mut counter = 0_u64;
298        let body = Some(Bytes::from_static(b"hello"));
299        accumulate_body_bytes(&mut counter, &body);
300        assert_eq!(counter, 5, "counter should equal body length");
301    }
302
303    #[test]
304    fn accumulate_body_bytes_multiple_chunks() {
305        let mut counter = 0_u64;
306        accumulate_body_bytes(&mut counter, &Some(Bytes::from_static(b"abc")));
307        accumulate_body_bytes(&mut counter, &Some(Bytes::from_static(b"de")));
308        assert_eq!(counter, 5, "counter should accumulate across calls");
309    }
310
311    #[test]
312    fn accumulate_body_bytes_noop_with_none() {
313        let mut counter = 10_u64;
314        accumulate_body_bytes(&mut counter, &None);
315        assert_eq!(counter, 10, "counter should be unchanged when body is None");
316    }
317
318    #[test]
319    fn accumulate_body_bytes_noop_with_empty() {
320        let mut counter = 0_u64;
321        accumulate_body_bytes(&mut counter, &Some(Bytes::new()));
322        assert_eq!(counter, 0, "counter should be unchanged when body is empty");
323    }
324
325    #[test]
326    fn released_or_continue_true_returns_release() {
327        assert!(
328            matches!(released_or_continue(true), FilterAction::Release),
329            "true should yield Release"
330        );
331    }
332
333    #[test]
334    fn released_or_continue_false_returns_continue() {
335        assert!(
336            matches!(released_or_continue(false), FilterAction::Continue),
337            "false should yield Continue"
338        );
339    }
340
341    #[test]
342    fn skip_by_response_conditions_empty_conditions() {
343        let filter = crate::builtins::StaticResponseFilter::from_config(
344            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
345        )
346        .unwrap();
347        let req = crate::test_utils::make_request(http::Method::GET, "/");
348        let mut resp = crate::test_utils::make_response();
349        let mut ctx = crate::test_utils::make_filter_context(&req);
350        ctx.response_header = Some(&mut resp);
351        assert!(
352            !skip_by_response_conditions(filter.as_ref(), &[], &ctx),
353            "empty conditions should not skip"
354        );
355    }
356
357    #[test]
358    fn skip_by_response_conditions_matching_when_does_not_skip() {
359        let filter = crate::builtins::StaticResponseFilter::from_config(
360            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
361        )
362        .unwrap();
363        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
364            status: Some(vec![200]),
365            headers: None,
366        })];
367        let req = crate::test_utils::make_request(http::Method::GET, "/");
368        let mut resp = crate::test_utils::make_response();
369        let mut ctx = crate::test_utils::make_filter_context(&req);
370        ctx.response_header = Some(&mut resp);
371        assert!(
372            !skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
373            "matching 'when' condition should not skip"
374        );
375    }
376
377    #[test]
378    fn skip_by_response_conditions_non_matching_when_skips() {
379        let filter = crate::builtins::StaticResponseFilter::from_config(
380            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
381        )
382        .unwrap();
383        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
384            status: Some(vec![404]),
385            headers: None,
386        })];
387        let req = crate::test_utils::make_request(http::Method::GET, "/");
388        let mut resp = crate::test_utils::make_response();
389        let mut ctx = crate::test_utils::make_filter_context(&req);
390        ctx.response_header = Some(&mut resp);
391        assert!(
392            skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
393            "non-matching 'when' condition should skip"
394        );
395    }
396
397    #[test]
398    fn skip_by_response_conditions_no_response_header_does_not_skip() {
399        let filter = crate::builtins::StaticResponseFilter::from_config(
400            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
401        )
402        .unwrap();
403        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
404            status: Some(vec![200]),
405            headers: None,
406        })];
407        let req = crate::test_utils::make_request(http::Method::GET, "/");
408        let ctx = crate::test_utils::make_filter_context(&req);
409        assert!(
410            !skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
411            "no response header should not skip"
412        );
413    }
414
415    #[test]
416    fn dispatch_body_result_continue() {
417        let outcome =
418            dispatch_body_result(Ok(FilterAction::Continue), "test", "request body", FailureMode::Closed).unwrap();
419        assert!(
420            matches!(outcome, super::super::http_utils::BodyFilterOutcome::Continue),
421            "Continue action should produce Continue outcome"
422        );
423    }
424
425    #[test]
426    fn dispatch_body_result_release() {
427        let outcome =
428            dispatch_body_result(Ok(FilterAction::Release), "test", "request body", FailureMode::Closed).unwrap();
429        assert!(
430            matches!(outcome, super::super::http_utils::BodyFilterOutcome::Released),
431            "Release action should produce Released outcome"
432        );
433    }
434
435    #[test]
436    fn dispatch_body_result_reject() {
437        let outcome = dispatch_body_result(
438            Ok(FilterAction::Reject(Rejection::status(403))),
439            "test",
440            "request body",
441            FailureMode::Closed,
442        )
443        .unwrap();
444        assert!(
445            matches!(outcome, super::super::http_utils::BodyFilterOutcome::Rejected(r) if r.status == 403),
446            "Reject action should produce Rejected outcome with correct status"
447        );
448    }
449
450    #[test]
451    fn dispatch_body_result_body_done() {
452        let outcome =
453            dispatch_body_result(Ok(FilterAction::BodyDone), "test", "request body", FailureMode::Closed).unwrap();
454        assert!(
455            matches!(outcome, super::super::http_utils::BodyFilterOutcome::BodyDone),
456            "BodyDone action should produce BodyDone outcome"
457        );
458    }
459
460    #[test]
461    fn dispatch_body_result_error_closed() {
462        let err: FilterError = "boom".into();
463        let result = dispatch_body_result(Err(err), "test", "request body", FailureMode::Closed);
464        assert!(result.is_err(), "error result should propagate as Err when closed");
465        assert!(
466            result.unwrap_err().to_string().contains("boom"),
467            "error message should be preserved"
468        );
469    }
470
471    #[test]
472    fn dispatch_body_result_error_open() {
473        let err: FilterError = "boom".into();
474        let result = dispatch_body_result(Err(err), "test", "request body", FailureMode::Open);
475        assert!(result.is_ok(), "error result should be Ok when fail-open");
476        assert!(
477            matches!(result.unwrap(), super::super::http_utils::BodyFilterOutcome::Continue),
478            "fail-open error should produce Continue outcome"
479        );
480    }
481}
praxis_filter/pipeline/http.rs

praxis_filter/pipeline/
http.rs
