From caea72627493076636ff3b857bf36988c34827fa Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 17:58:22 -0700 Subject: [PATCH 1/6] feat(http-callout): scaffold crate and workspace wiring Create the praxis-proxy-http-callout satellite crate skeleton with an empty export_filters! invocation and wire it into the workspace, server, and integration test infrastructure. - filter/http-callout/Cargo.toml with praxis-filters metadata marker - filter/http-callout/src/lib.rs with empty export_filters! - Workspace Cargo.toml: add member, serde_json_path dep, crate dep - server/Cargo.toml: optional dep + http-callout feature (default-on) - tests/integration/Cargo.toml: dev-dep + http-callout feature Assisted by Opus 4.6 Signed-off-by: usize --- Cargo.lock | 70 ++++++++++++++++++++++++++++++++++ Cargo.toml | 3 ++ filter/http-callout/Cargo.toml | 35 +++++++++++++++++ filter/http-callout/src/lib.rs | 15 ++++++++ server/Cargo.toml | 4 +- tests/integration/Cargo.toml | 4 +- 6 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 filter/http-callout/Cargo.toml create mode 100644 filter/http-callout/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 94f6a502..3aa4b41c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3194,6 +3194,7 @@ dependencies = [ "notify", "praxis-proxy-core", "praxis-proxy-filter", + "praxis-proxy-http-callout", "praxis-proxy-protocol", "serde", "tempfile", @@ -3277,6 +3278,24 @@ dependencies = [ "zeroize", ] +[[package]] +name = "praxis-proxy-http-callout" +version = "0.3.1" +dependencies = [ + "async-trait", + "bytes", + "http", + "praxis-proxy-core", + "praxis-proxy-filter", + "serde", + "serde_json", + "serde_json_path", + "tokio", + "tracing", + "wiremock", + "yaml_serde", +] + [[package]] name = "praxis-proxy-proto" version = "0.3.1" @@ -3388,6 +3407,7 @@ dependencies = [ "jsonwebtoken 9.3.1", "praxis-proxy-core", "praxis-proxy-filter", + "praxis-proxy-http-callout", "praxis-proxy-protocol", "praxis-test-utils", "rustls", @@ -4561,6 +4581,56 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_json_path" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b992cea3194eea663ba99a042d61cea4bd1872da37021af56f6a37e0359b9d33" +dependencies = [ + "inventory", + "nom", + "regex", + "serde", + "serde_json", + "serde_json_path_core", + "serde_json_path_macros", + "thiserror 2.0.18", +] + +[[package]] +name = "serde_json_path_core" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dde67d8dfe7d4967b5a95e247d4148368ddd1e753e500adb34b3ffe40c6bc1bc" +dependencies = [ + "inventory", + "serde", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "serde_json_path_macros" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "517acfa7f77ddaf5c43d5f119c44a683774e130b4247b7d3210f8924506cfac8" +dependencies = [ + "inventory", + "serde_json_path_core", + "serde_json_path_macros_internal", +] + +[[package]] +name = "serde_json_path_macros_internal" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aafbefbe175fa9bf03ca83ef89beecff7d2a95aaacd5732325b90ac8c3bd7b90" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" diff --git a/Cargo.toml b/Cargo.toml index 8cf4f147..a86d4959 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "core", "filter", "filter/ext-proc", + "filter/http-callout", "filter/proto", "protocol", "tls", @@ -56,6 +57,7 @@ pingora-http = { version = "0.8.2", package = "quixotic-plecostomus-http" } pingora-proxy = { version = "0.8.2", package = "quixotic-plecostomus-proxy", features = ["rustls"] } percent-encoding = "2.3.2" praxis-ext-proc = { version = "0.3.1", path = "filter/ext-proc", package = "praxis-proxy-ext-proc" } +praxis-http-callout = { version = "0.3.1", path = "filter/http-callout", package = "praxis-proxy-http-callout" } praxis-proto = { version = "0.3.1", path = "filter/proto", package = "praxis-proxy-proto" } prost = "0.14.4" prost-build = "0.14.4" @@ -74,6 +76,7 @@ praxis-test-utils = { path = "tests/utils" } secrecy = { version = "0.10.3", features = ["serde"] } serde = { version = "1.0.228", features = ["derive", "rc"] } serde_json = "1.0.150" +serde_json_path = "0.7.2" rcgen = "0.14.8" regex = "1.12.4" reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls"] } diff --git a/filter/http-callout/Cargo.toml b/filter/http-callout/Cargo.toml new file mode 100644 index 00000000..81c843c6 --- /dev/null +++ b/filter/http-callout/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "praxis-proxy-http-callout" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +description = "HTTP callout filter for Praxis" +license.workspace = true +repository.workspace = true +readme = "../../README.md" +publish = false + +[lib] +name = "praxis_http_callout" + +[lints] +workspace = true + +[package.metadata.praxis-filters] + +[dependencies] +async-trait = { workspace = true } +bytes = { workspace = true } +http = { workspace = true } +praxis-core = { workspace = true, features = ["callout"] } +praxis-filter = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +serde_json_path = { workspace = true } +serde_yaml = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +wiremock = { workspace = true } diff --git a/filter/http-callout/src/lib.rs b/filter/http-callout/src/lib.rs new file mode 100644 index 00000000..62a9d49a --- /dev/null +++ b/filter/http-callout/src/lib.rs @@ -0,0 +1,15 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +#![deny(unreachable_pub)] + +//! HTTP callout filter for Praxis. +//! +//! Provides an [`HttpFilter`] that makes outbound HTTP requests during +//! request processing, extracts results from the response via `JSONPath`, +//! and feeds them into [`FilterResultSet`] for branch-chain evaluation. +//! +//! [`HttpFilter`]: praxis_filter::HttpFilter +//! [`FilterResultSet`]: praxis_filter::FilterResultSet + +praxis_filter::export_filters! {} diff --git a/server/Cargo.toml b/server/Cargo.toml index 08cc6104..c190777d 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -25,6 +25,7 @@ clap = { workspace = true } notify = { workspace = true } praxis-core = { workspace = true } praxis-filter = { workspace = true } +praxis-http-callout = { workspace = true, optional = true } praxis-protocol = { workspace = true } serde = { workspace = true } serde_yaml = { workspace = true } @@ -43,7 +44,8 @@ cargo_metadata = "0.23.1" tempfile = { workspace = true } [features] -default = ["ai-inference"] +default = ["ai-inference", "http-callout"] ai-inference = ["praxis-filter/ai-inference", "praxis-protocol/response-store"] ext-proc = ["praxis-filter/ext-proc"] cpex-policy-engine = ["praxis-filter/cpex-policy-engine"] +http-callout = ["dep:praxis-http-callout", "praxis-core/callout"] diff --git a/tests/integration/Cargo.toml b/tests/integration/Cargo.toml index d1d02062..eb00a959 100644 --- a/tests/integration/Cargo.toml +++ b/tests/integration/Cargo.toml @@ -8,9 +8,10 @@ readme = "../../README.md" publish = false [features] -default = ["ai-inference"] +default = ["ai-inference", "http-callout"] ai-inference = ["praxis-filter/ai-inference", "praxis-test-utils/ai-inference"] cpex-policy-engine = ["praxis-filter/cpex-policy-engine"] +http-callout = ["praxis-core/callout"] no-mac-cert-rotation-tests = [] # We use this to disable testing cert rotation on macOS [lints] @@ -27,6 +28,7 @@ http = { workspace = true } jsonwebtoken = "9" praxis-core = { workspace = true } praxis-filter = { workspace = true } +praxis-http-callout = { workspace = true } praxis-protocol = { workspace = true } praxis-test-utils = { workspace = true } rustls = { workspace = true } From 58cf883736d5c38f4d280938e28a7c1a832c3484 Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 17:58:46 -0700 Subject: [PATCH 2/6] feat(http-callout): implement HttpCalloutFilter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Core implementation: config types with serde deserialization, SSRF URL validation, env-var ${VAR} expansion in header values, JSONPath extraction with type coercion, and the HttpCalloutFilter struct with full HttpFilter trait implementation. - config.rs: HttpCalloutConfig, TargetConfig, RequestConfig, ResponseConfig, ExtractionConfig, Phase enum, FailureModeConfig, CircuitBreakerConfig, duration parsing, SSRF validation, env-var expansion - extract.rs: CompiledExtraction with JSONPath compilation at config time and runtime evaluation with coercion rules (bool→string, number→string, null→skip, array/object→JSON) - lib.rs: HttpCalloutFilter with from_config factory, callout request building, response extraction, header injection, on-denied headers, and export_filters! registration Assisted by Opus 4.6 Signed-off-by: usize --- filter/http-callout/src/config.rs | 413 +++++++++++++++++++++++++++++ filter/http-callout/src/extract.rs | 194 ++++++++++++++ filter/http-callout/src/lib.rs | 331 ++++++++++++++++++++++- 3 files changed, 937 insertions(+), 1 deletion(-) create mode 100644 filter/http-callout/src/config.rs create mode 100644 filter/http-callout/src/extract.rs diff --git a/filter/http-callout/src/config.rs b/filter/http-callout/src/config.rs new file mode 100644 index 00000000..f10a5652 --- /dev/null +++ b/filter/http-callout/src/config.rs @@ -0,0 +1,413 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Serde configuration types for the HTTP callout filter. + +use std::time::Duration; + +use praxis_filter::FilterError; +use serde::Deserialize; + +// ----------------------------------------------------------------------------- +// Top-Level Config +// ----------------------------------------------------------------------------- + +/// HTTP callout filter configuration. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct HttpCalloutConfig { + /// Callout target configuration. + pub target: TargetConfig, + + /// Request phase and body forwarding options. + #[serde(default)] + pub request: RequestConfig, + + /// Response extraction and header injection. + #[serde(default)] + pub response: ResponseConfig, + + /// Behavior on callout failure. + #[serde(default, alias = "failure_mode")] + pub on_failure: FailureModeConfig, + + /// HTTP status code returned when rejecting on failure. + pub status_on_error: Option, + + /// Circuit breaker configuration. + pub circuit_breaker: Option, + + /// Maximum callout depth for loop prevention. + pub max_depth: Option, +} + +// ----------------------------------------------------------------------------- +// Target +// ----------------------------------------------------------------------------- + +/// Callout target: URL, timeout, and headers. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TargetConfig { + /// Absolute HTTP(S) URL to call. + pub url: String, + + /// Request timeout (e.g. `"2s"`, `"500ms"`). + #[serde(default = "default_timeout", deserialize_with = "deserialize_duration")] + pub timeout: Duration, + + /// Static headers to send with every callout. + #[serde(default)] + pub headers: Vec, + + /// Headers to copy from the downstream request. + #[serde(default)] + pub forward_headers: Vec, +} + +/// A static header entry with optional env-var expansion in the value. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct HeaderEntry { + /// Header name. + pub name: String, + + /// Header value. Supports `${VAR}` env-var expansion. + pub value: String, +} + +// ----------------------------------------------------------------------------- +// Request +// ----------------------------------------------------------------------------- + +/// Controls when the callout fires and how much body to forward. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct RequestConfig { + /// Phase at which the callout executes. + #[serde(default)] + pub phase: Phase, + + /// Maximum request body bytes to buffer and forward. + #[serde(default = "default_max_body_bytes")] + pub max_body_bytes: usize, +} + +impl Default for RequestConfig { + fn default() -> Self { + Self { + phase: Phase::default(), + max_body_bytes: default_max_body_bytes(), + } + } +} + +// ----------------------------------------------------------------------------- +// Response +// ----------------------------------------------------------------------------- + +/// Extraction and header injection from the callout response. +#[derive(Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ResponseConfig { + /// `JSONPath` extractions to write into [`FilterResultSet`]. + /// + /// [`FilterResultSet`]: praxis_filter::FilterResultSet + #[serde(default)] + pub extract: Vec, + + /// Callout response headers to inject into the upstream request + /// on success. + #[serde(default)] + pub inject_headers: Vec, + + /// Callout response headers to include in the rejection + /// response when the callout returns a non-2xx status. + #[serde(default)] + pub on_denied_headers: Vec, +} + +/// A single `JSONPath` extraction rule. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ExtractionConfig { + /// `JSONPath` expression to evaluate against the response body. + pub json_path: String, + + /// Key to write the result under in [`FilterResultSet`]. + /// + /// [`FilterResultSet`]: praxis_filter::FilterResultSet + pub result_key: String, +} + +// ----------------------------------------------------------------------------- +// Phase +// ----------------------------------------------------------------------------- + +/// When the callout fires during request processing. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum Phase { + /// Fire during `on_request` (headers only, no body). + RequestHeaders, + + /// Fire during `on_request_body` (headers + body). + #[default] + RequestBody, +} + +// ----------------------------------------------------------------------------- +// Failure Mode +// ----------------------------------------------------------------------------- + +/// Behavior when a callout fails. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum FailureModeConfig { + /// Reject the original request (fail-closed). + #[default] + Closed, + + /// Allow the original request to proceed (fail-open). + Open, +} + +// ----------------------------------------------------------------------------- +// Circuit Breaker +// ----------------------------------------------------------------------------- + +/// Circuit breaker settings for the callout. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct CircuitBreakerConfig { + /// Consecutive failures to trip the breaker. + pub failure_threshold: u32, + + /// Recovery window (e.g. `"30s"`). + #[serde(deserialize_with = "deserialize_duration")] + pub recovery_timeout: Duration, +} + +// ----------------------------------------------------------------------------- +// Defaults +// ----------------------------------------------------------------------------- + +/// Default timeout: 5 seconds. +fn default_timeout() -> Duration { + Duration::from_secs(5) +} + +/// Default max body bytes: 1 MiB. +fn default_max_body_bytes() -> usize { + 1_048_576 // 1 MiB +} + +// ----------------------------------------------------------------------------- +// Duration Parsing +// ----------------------------------------------------------------------------- + +/// Deserialize a human-readable duration string (`"2s"`, `"500ms"`). +fn deserialize_duration<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + let s = String::deserialize(deserializer)?; + parse_duration(&s).map_err(serde::de::Error::custom) +} + +/// Parse a duration string like `"2s"` or `"500ms"`. +/// +/// # Errors +/// +/// Returns an error if the format is unrecognized. +fn parse_duration(s: &str) -> Result { + if let Some(ms) = s.strip_suffix("ms") { + let n: u64 = ms + .parse() + .map_err(|_err| format!("invalid milliseconds in duration: {s}"))?; + return Ok(Duration::from_millis(n)); + } + if let Some(secs) = s.strip_suffix('s') { + let n: u64 = secs + .parse() + .map_err(|_err| format!("invalid seconds in duration: {s}"))?; + return Ok(Duration::from_secs(n)); + } + Err(format!("unsupported duration format: {s} (use '2s' or '500ms')")) +} + +// ----------------------------------------------------------------------------- +// SSRF Validation +// ----------------------------------------------------------------------------- + +/// Validate that a URL is safe for outbound callouts. +/// +/// # Errors +/// +/// Returns [`FilterError`] if the URL: +/// - Is not absolute +/// - Uses a scheme other than `http` or `https` +/// - Has an empty host +/// - Contains `${...}` template markers +pub(crate) fn validate_callout_url(url: &str) -> Result<(), FilterError> { + if url.contains("${") { + return Err(format!("http_callout: URL must not contain template variables: {url}").into()); + } + + let parsed: http::Uri = url + .parse() + .map_err(|e| -> FilterError { format!("http_callout: invalid URL '{url}': {e}").into() })?; + + match parsed.scheme_str() { + Some("http" | "https") => {}, + Some(scheme) => { + return Err(format!("http_callout: URL scheme must be http or https, got '{scheme}'").into()); + }, + None => { + return Err(format!("http_callout: URL must have an http or https scheme: {url}").into()); + }, + } + + if parsed.host().is_none_or(str::is_empty) { + return Err(format!("http_callout: URL must have a non-empty host: {url}").into()); + } + + Ok(()) +} + +// ----------------------------------------------------------------------------- +// Env Var Expansion +// ----------------------------------------------------------------------------- + +/// Expand `${VAR}` references in a header value using environment +/// variables. +/// +/// # Errors +/// +/// Returns [`FilterError`] if a referenced variable is not set. +pub(crate) fn expand_env_vars(value: &str) -> Result { + let mut result = String::with_capacity(value.len()); + let mut remaining = value; + + while let Some(start) = remaining.find("${") { + // `$` and `{` are single-byte ASCII, so byte indexing is safe. + result.push_str(remaining.get(..start).unwrap_or_default()); + let after_open = remaining.get(start + 2..).unwrap_or_default(); + let end = after_open.find('}').ok_or_else(|| -> FilterError { + format!("http_callout: unclosed '${{' in header value: {value}").into() + })?; + let var_name = after_open.get(..end).unwrap_or_default(); + let var_value = std::env::var(var_name).map_err(|_err| -> FilterError { + format!("http_callout: environment variable '{var_name}' is not set").into() + })?; + result.push_str(&var_value); + remaining = after_open.get(end + 1..).unwrap_or_default(); + } + + result.push_str(remaining); + Ok(result) +} + +// ----------------------------------------------------------------------------- +// 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 super::*; + + #[test] + fn parse_duration_seconds() { + assert_eq!(parse_duration("5s").unwrap(), Duration::from_secs(5)); + } + + #[test] + fn parse_duration_milliseconds() { + assert_eq!(parse_duration("500ms").unwrap(), Duration::from_millis(500)); + } + + #[test] + fn parse_duration_invalid_suffix() { + assert!(parse_duration("5m").is_err(), "unsupported suffix should error"); + } + + #[test] + fn parse_duration_invalid_number() { + assert!(parse_duration("abcs").is_err(), "non-numeric should error"); + } + + #[test] + fn validate_url_accepts_http() { + assert!(validate_callout_url("http://example.com/api").is_ok()); + } + + #[test] + fn validate_url_accepts_https() { + assert!(validate_callout_url("https://api.example.com/v2/guard").is_ok()); + } + + #[test] + fn validate_url_rejects_no_scheme() { + assert!( + validate_callout_url("example.com/api").is_err(), + "URL without scheme should be rejected" + ); + } + + #[test] + fn validate_url_rejects_non_http_scheme() { + let err = validate_callout_url("ftp://example.com/file").unwrap_err(); + assert!( + err.to_string().contains("http or https"), + "should mention allowed schemes: {err}" + ); + } + + #[test] + fn validate_url_rejects_template_in_url() { + let err = validate_callout_url("https://${HOST}/api").unwrap_err(); + assert!( + err.to_string().contains("template"), + "should mention template variables: {err}" + ); + } + + #[test] + fn expand_env_vars_no_vars() { + assert_eq!(expand_env_vars("Bearer token123").unwrap(), "Bearer token123"); + } + + #[test] + fn expand_env_vars_with_var() { + // PATH is reliably set on all platforms. + let result = expand_env_vars("prefix-${PATH}-suffix").unwrap(); + let path_value = std::env::var("PATH").unwrap(); + assert_eq!(result, format!("prefix-{path_value}-suffix")); + } + + #[test] + fn expand_env_vars_unset_var() { + let err = expand_env_vars("${PRAXIS_TEST_NONEXISTENT_VAR_XYZ_12345}").unwrap_err(); + assert!( + err.to_string().contains("not set"), + "should report unset variable: {err}" + ); + } + + #[test] + fn expand_env_vars_unclosed_brace() { + let err = expand_env_vars("${UNCLOSED").unwrap_err(); + assert!( + err.to_string().contains("unclosed"), + "should report unclosed brace: {err}" + ); + } +} diff --git a/filter/http-callout/src/extract.rs b/filter/http-callout/src/extract.rs new file mode 100644 index 00000000..649fad46 --- /dev/null +++ b/filter/http-callout/src/extract.rs @@ -0,0 +1,194 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! `JSONPath` extraction from callout response bodies. + +use praxis_filter::{FilterError, FilterResultSet}; +use serde_json::Value; +use serde_json_path::JsonPath; +use tracing::debug; + +// ----------------------------------------------------------------------------- +// Compiled Extraction +// ----------------------------------------------------------------------------- + +/// A pre-compiled `JSONPath` extraction rule. +#[derive(Debug)] +pub(crate) struct CompiledExtraction { + /// The compiled `JSONPath` expression. + path: JsonPath, + + /// Key to write into [`FilterResultSet`]. + result_key: String, +} + +impl CompiledExtraction { + /// Parse and compile a `JSONPath` expression at config time. + /// + /// # Errors + /// + /// Returns [`FilterError`] if the expression is invalid. + pub(crate) fn compile(json_path: &str, result_key: String) -> Result { + let path = JsonPath::parse(json_path) + .map_err(|e| -> FilterError { format!("http_callout: invalid JSONPath '{json_path}': {e}").into() })?; + Ok(Self { path, result_key }) + } + + /// Evaluate this extraction against a JSON value and write + /// results into the result set. + /// + /// Coercion rules for the first matched node: + /// - `bool` → `"true"` / `"false"` + /// - `number` → decimal string + /// - `string` → as-is + /// - `array` / `object` → compact JSON + /// - `null` or no match → skip (no entry written) + /// + /// # Errors + /// + /// Returns [`FilterError`] if the result set rejects the + /// key or value. + pub(crate) fn evaluate(&self, json: &Value, results: &mut FilterResultSet) -> Result<(), FilterError> { + let node_list = self.path.query(json); + let nodes: Vec<&Value> = node_list.all(); + + let Some(first) = nodes.first() else { + debug!(key = %self.result_key, "JSONPath matched no nodes; skipping"); + return Ok(()); + }; + + let coerced = coerce_value(first); + let Some(value) = coerced else { + debug!(key = %self.result_key, "JSONPath matched null; skipping"); + return Ok(()); + }; + + results.set(self.result_key.clone(), value)?; + Ok(()) + } +} + +// ----------------------------------------------------------------------------- +// Coercion +// ----------------------------------------------------------------------------- + +/// Coerce a JSON value to a string for [`FilterResultSet`]. +/// +/// Returns `None` for null values. +fn coerce_value(value: &Value) -> Option { + match value { + Value::Null => None, + Value::Bool(b) => Some(b.to_string()), + Value::Number(n) => Some(n.to_string()), + Value::String(s) => Some(s.clone()), + Value::Array(_) | Value::Object(_) => Some(value.to_string()), + } +} + +// ----------------------------------------------------------------------------- +// 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 serde_json::json; + + use super::*; + + #[test] + fn compile_valid_expression() { + assert!( + CompiledExtraction::compile("$.flagged", "flagged".into()).is_ok(), + "valid JSONPath should compile" + ); + } + + #[test] + fn compile_invalid_expression() { + let err = CompiledExtraction::compile("$[invalid", "key".into()).unwrap_err(); + assert!( + err.to_string().contains("invalid JSONPath"), + "should report invalid expression: {err}" + ); + } + + #[test] + fn evaluate_bool_true() { + let ext = CompiledExtraction::compile("$.flagged", "flagged".into()).unwrap(); + let json = json!({"flagged": true}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("flagged"), Some("true")); + } + + #[test] + fn evaluate_bool_false() { + let ext = CompiledExtraction::compile("$.flagged", "flagged".into()).unwrap(); + let json = json!({"flagged": false}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("flagged"), Some("false")); + } + + #[test] + fn evaluate_number() { + let ext = CompiledExtraction::compile("$.score", "score".into()).unwrap(); + let json = json!({"score": 0.95}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("score"), Some("0.95")); + } + + #[test] + fn evaluate_string() { + let ext = CompiledExtraction::compile("$.label", "label".into()).unwrap(); + let json = json!({"label": "safe"}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("label"), Some("safe")); + } + + #[test] + fn evaluate_array() { + let ext = CompiledExtraction::compile("$.tags", "tags".into()).unwrap(); + let json = json!({"tags": ["a", "b"]}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("tags"), Some(r#"["a","b"]"#)); + } + + #[test] + fn evaluate_object() { + let ext = CompiledExtraction::compile("$.meta", "meta".into()).unwrap(); + let json = json!({"meta": {"k": "v"}}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert_eq!(rs.get("meta"), Some(r#"{"k":"v"}"#)); + } + + #[test] + fn evaluate_null_skips() { + let ext = CompiledExtraction::compile("$.missing", "missing".into()).unwrap(); + let json = json!({"missing": null}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert!(rs.get("missing").is_none(), "null should be skipped"); + } + + #[test] + fn evaluate_no_match_skips() { + let ext = CompiledExtraction::compile("$.nonexistent", "key".into()).unwrap(); + let json = json!({"other": 1}); + let mut rs = FilterResultSet::new(); + ext.evaluate(&json, &mut rs).unwrap(); + assert!(rs.get("key").is_none(), "no-match should be skipped"); + } +} diff --git a/filter/http-callout/src/lib.rs b/filter/http-callout/src/lib.rs index 62a9d49a..d21ef017 100644 --- a/filter/http-callout/src/lib.rs +++ b/filter/http-callout/src/lib.rs @@ -12,4 +12,333 @@ //! [`HttpFilter`]: praxis_filter::HttpFilter //! [`FilterResultSet`]: praxis_filter::FilterResultSet -praxis_filter::export_filters! {} +mod config; +mod extract; + +use std::borrow::Cow; + +use async_trait::async_trait; +use bytes::Bytes; +use praxis_core::callout::{ + CalloutClient, CalloutConfig, CalloutRequest, CalloutResponse, CalloutResult, + CircuitBreakerConfig as CoreCircuitBreakerConfig, DEPTH_HEADER, FailureMode, +}; +use praxis_filter::{ + BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, Rejection, parse_filter_config, +}; +use tracing::debug; + +use crate::{ + config::{FailureModeConfig, HttpCalloutConfig, Phase, expand_env_vars, validate_callout_url}, + extract::CompiledExtraction, +}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// Filter type name. +const FILTER_NAME: &str = "http_callout"; + +// ----------------------------------------------------------------------------- +// HttpCalloutFilter +// ----------------------------------------------------------------------------- + +/// HTTP callout filter. +/// +/// Makes an outbound HTTP request during request processing, +/// optionally forwarding the request body and downstream headers. +/// Extracts values from the callout response via `JSONPath` and +/// writes them to [`FilterResultSet`] for branch-chain evaluation. +/// +/// [`FilterResultSet`]: praxis_filter::FilterResultSet +struct HttpCalloutFilter { + /// Reusable HTTP callout client. + client: CalloutClient, + + /// Pre-compiled `JSONPath` extraction rules. + extractions: Vec, + + /// Downstream headers to copy into the callout request. + forward_headers: Vec, + + /// Static headers to send with every callout. + headers: Vec<(http::HeaderName, http::HeaderValue)>, + + /// Callout response headers to inject into the upstream + /// request on success. + inject_headers: Vec, + + /// Maximum request body bytes to buffer. + max_body_bytes: usize, + + /// Callout response headers to include in the rejection + /// response when the callout returns non-2xx. + on_denied_headers: Vec, + + /// When the callout fires. + phase: Phase, + + /// Target URL for the callout. + url: String, +} + +impl HttpCalloutFilter { + /// Construct the filter from a YAML config value. + /// + /// # Errors + /// + /// Returns [`FilterError`] if config parsing, SSRF validation, + /// env-var expansion, `JSONPath` compilation, or client + /// construction fails. + fn from_config(config: &serde_yaml::Value) -> Result, FilterError> { + let cfg: HttpCalloutConfig = parse_filter_config(FILTER_NAME, config)?; + + validate_callout_url(&cfg.target.url)?; + + let headers = parse_static_headers(&cfg)?; + let forward_headers = parse_header_names(&cfg.target.forward_headers, "forward_header")?; + let extractions = compile_extractions(&cfg)?; + let inject_headers = parse_header_names(&cfg.response.inject_headers, "inject_header")?; + let on_denied_headers = parse_header_names(&cfg.response.on_denied_headers, "on_denied_header")?; + + let client = build_callout_client(&cfg)?; + + Ok(Box::new(Self { + client, + extractions, + forward_headers, + headers, + inject_headers, + max_body_bytes: cfg.request.max_body_bytes, + on_denied_headers, + phase: cfg.request.phase, + url: cfg.target.url, + })) + } + + /// Build a [`CalloutRequest`] from the current filter context. + fn build_request(&self, ctx: &HttpFilterContext<'_>, body: Option>) -> CalloutRequest { + let depth = ctx + .request + .headers + .get(DEPTH_HEADER) + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + let mut headers = self.headers.clone(); + + for name in &self.forward_headers { + if let Some(value) = ctx.request.headers.get(name) { + headers.push((name.clone(), value.clone())); + } + } + + CalloutRequest { + body, + depth, + headers, + method: http::Method::POST, + url: self.url.clone(), + } + } + + /// Process a successful callout response: extract results and + /// inject headers. + fn handle_success( + &self, + response: &CalloutResponse, + ctx: &mut HttpFilterContext<'_>, + ) -> Result { + if !self.extractions.is_empty() { + if let Ok(json) = serde_json::from_slice::(&response.body) { + let results = ctx.filter_results.entry(self.name()).or_default(); + for extraction in &self.extractions { + extraction.evaluate(&json, results)?; + } + } else { + debug!("callout response body is not valid JSON; skipping extraction"); + } + } + + for name in &self.inject_headers { + if let Some((_, value)) = response.headers.iter().find(|(n, _)| n == name) + && let Ok(value_str) = value.to_str() + { + ctx.extra_request_headers + .push((Cow::Owned(name.to_string()), value_str.to_owned())); + } + } + + Ok(FilterAction::Continue) + } + + /// Build a rejection with on-denied headers from the callout + /// response, if available. + fn build_rejection(&self, response: Option<&CalloutResponse>, status: u16) -> FilterAction { + let mut rejection = Rejection::status(status); + + if let Some(resp) = response { + for name in &self.on_denied_headers { + if let Some((_, value)) = resp.headers.iter().find(|(n, _)| n == name) + && let Ok(value_str) = value.to_str() + { + rejection = rejection.with_header(name.to_string(), value_str.to_owned()); + } + } + } + + FilterAction::Reject(rejection) + } + + /// Execute the callout and process the result. + async fn execute_callout( + &self, + ctx: &mut HttpFilterContext<'_>, + body: Option>, + ) -> Result { + let request = self.build_request(ctx, body); + let result = self.client.execute(request).await; + + match result { + CalloutResult::Success(response) => self.handle_success(&response, ctx), + CalloutResult::Failed => { + debug!("callout failed (open mode); continuing"); + Ok(FilterAction::Continue) + }, + CalloutResult::Rejected(rejection) => Ok(self.build_rejection(None, rejection.status)), + } + } +} + +// ----------------------------------------------------------------------------- +// Config Parsing Helpers +// ----------------------------------------------------------------------------- + +/// Parse static header entries with env-var expansion. +fn parse_static_headers(cfg: &HttpCalloutConfig) -> Result, FilterError> { + cfg.target + .headers + .iter() + .map(|h| { + let expanded = expand_env_vars(&h.value)?; + let name: http::HeaderName = h.name.parse().map_err(|e| -> FilterError { + format!("http_callout: invalid header name '{}': {e}", h.name).into() + })?; + let value: http::HeaderValue = expanded.parse().map_err(|e| -> FilterError { + format!("http_callout: invalid header value for '{}': {e}", h.name).into() + })?; + Ok((name, value)) + }) + .collect() +} + +/// Parse a list of header name strings. +fn parse_header_names(names: &[String], context: &str) -> Result, FilterError> { + names + .iter() + .map(|h| { + h.parse::() + .map_err(|e| -> FilterError { format!("http_callout: invalid {context} '{h}': {e}").into() }) + }) + .collect() +} + +/// Compile `JSONPath` extraction rules from config. +fn compile_extractions(cfg: &HttpCalloutConfig) -> Result, FilterError> { + cfg.response + .extract + .iter() + .map(|e| CompiledExtraction::compile(&e.json_path, e.result_key.clone())) + .collect() +} + +/// Build the [`CalloutClient`] from parsed config. +#[expect( + clippy::cast_possible_truncation, + reason = "durations are bounded by config validation" +)] +fn build_callout_client(cfg: &HttpCalloutConfig) -> Result { + let failure_mode = match cfg.on_failure { + FailureModeConfig::Closed => FailureMode::Closed, + FailureModeConfig::Open => FailureMode::Open, + }; + + let circuit_breaker = cfg.circuit_breaker.as_ref().map(|cb| CoreCircuitBreakerConfig { + consecutive_failures: cb.failure_threshold, + recovery_window_ms: cb.recovery_timeout.as_millis() as u64, + }); + + let callout_config = CalloutConfig { + circuit_breaker, + failure_mode, + max_depth: cfg.max_depth.unwrap_or(1), + status_on_error: cfg.status_on_error.unwrap_or(403), + timeout_ms: cfg.target.timeout.as_millis() as u64, + ..CalloutConfig::default() + }; + + CalloutClient::new(callout_config).map_err(|e| -> FilterError { format!("http_callout: {e}").into() }) +} + +// ----------------------------------------------------------------------------- +// HttpFilter Implementation +// ----------------------------------------------------------------------------- + +#[async_trait] +impl HttpFilter for HttpCalloutFilter { + fn name(&self) -> &'static str { + FILTER_NAME + } + + fn request_body_access(&self) -> BodyAccess { + match self.phase { + Phase::RequestBody => BodyAccess::ReadOnly, + Phase::RequestHeaders => BodyAccess::None, + } + } + + fn request_body_mode(&self) -> BodyMode { + match self.phase { + Phase::RequestBody => BodyMode::StreamBuffer { + max_bytes: Some(self.max_body_bytes), + }, + Phase::RequestHeaders => BodyMode::Stream, + } + } + + fn needs_request_context(&self) -> bool { + true + } + + async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result { + if self.phase != Phase::RequestHeaders { + return Ok(FilterAction::Continue); + } + + self.execute_callout(ctx, None).await + } + + async fn on_request_body( + &self, + ctx: &mut HttpFilterContext<'_>, + body: &mut Option, + end_of_stream: bool, + ) -> Result { + if self.phase != Phase::RequestBody || !end_of_stream { + return Ok(FilterAction::Continue); + } + + let body_bytes = body.as_ref().map(|b| b.to_vec()); + self.execute_callout(ctx, body_bytes).await + } +} + +// ----------------------------------------------------------------------------- +// Filter Registration +// ----------------------------------------------------------------------------- + +praxis_filter::export_filters! { + http "http_callout" => HttpCalloutFilter::from_config, +} From 3ec2df4862f7c6985ccdf8cc402368c74b0c12cf Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 17:59:02 -0700 Subject: [PATCH 3/6] test(http-callout): add unit tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Comprehensive wiremock-based tests covering all filter features: - Config parsing: valid config, missing target, invalid URL (no scheme, template), invalid JSONPath, env-var expansion - SSRF validation: rejects non-http schemes, template URLs - Phase handling: request_headers vs request_body body access - Successful callout: 2xx → extraction → FilterResultSet - JSONPath coercion: bool, number, string, null, no-match - Failure modes: closed (reject) vs open (continue) - Timeout: callout times out → failure mode applied - Circuit breaker: trips after threshold - Depth limiting: depth >= max_depth → failure mode applied - Forward headers: downstream headers copied to callout - Inject headers: callout response headers added to upstream - Request body phase: fires on end_of_stream, skips on_request 46 unit tests total (23 config/extract + 23 filter integration). Assisted by Opus 4.6 Signed-off-by: usize --- filter/http-callout/src/lib.rs | 3 + filter/http-callout/src/tests.rs | 817 +++++++++++++++++++++++++++++++ 2 files changed, 820 insertions(+) create mode 100644 filter/http-callout/src/tests.rs diff --git a/filter/http-callout/src/lib.rs b/filter/http-callout/src/lib.rs index d21ef017..a91fb3b7 100644 --- a/filter/http-callout/src/lib.rs +++ b/filter/http-callout/src/lib.rs @@ -15,6 +15,9 @@ mod config; mod extract; +#[cfg(test)] +mod tests; + use std::borrow::Cow; use async_trait::async_trait; diff --git a/filter/http-callout/src/tests.rs b/filter/http-callout/src/tests.rs new file mode 100644 index 00000000..263c2fe0 --- /dev/null +++ b/filter/http-callout/src/tests.rs @@ -0,0 +1,817 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Unit tests for the HTTP callout filter. +//! +//! Uses [`wiremock`] to simulate the callout backend. + +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::panic, + clippy::needless_pass_by_value, + clippy::too_many_lines, + reason = "tests" +)] +mod filter_tests { + use std::time::Duration; + + use praxis_filter::{BodyAccess, BodyMode, FilterAction}; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + use crate::HttpCalloutFilter; + + // ------------------------------------------------------------------------- + // Config Parsing + // ------------------------------------------------------------------------- + + #[test] + fn config_valid_minimal() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + assert_eq!(filter.name(), "http_callout"); + } + + #[test] + fn config_missing_target() { + let yaml = serde_yaml::from_str::("{}").unwrap(); + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("target"), + "should mention missing target: {err}" + ); + } + + #[test] + fn config_invalid_url_no_scheme() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "example.com/api" + "#, + ) + .unwrap(); + + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("invalid") || err.to_string().contains("http or https"), + "should reject URL without scheme: {err}" + ); + } + + #[test] + fn config_invalid_url_template() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "https://${HOST}/api" + "#, + ) + .unwrap(); + + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("template"), + "should reject template URL: {err}" + ); + } + + #[test] + fn config_invalid_jsonpath() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + response: + extract: + - json_path: "$[invalid" + result_key: "key" + "#, + ) + .unwrap(); + + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("invalid JSONPath"), + "should reject invalid JSONPath: {err}" + ); + } + + #[test] + fn config_env_var_expansion_unset() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + headers: + - name: "Authorization" + value: "Bearer ${PRAXIS_TEST_MISSING_VAR_ABC123}" + "#, + ) + .unwrap(); + + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("not set"), + "should fail on unset env var: {err}" + ); + } + + #[test] + fn config_non_http_scheme_rejected() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "ftp://example.com/file" + "#, + ) + .unwrap(); + + let err = HttpCalloutFilter::from_config(&yaml).err().expect("expected error"); + assert!( + err.to_string().contains("http or https"), + "should reject non-http scheme: {err}" + ); + } + + // ------------------------------------------------------------------------- + // Phase Handling + // ------------------------------------------------------------------------- + + #[test] + fn phase_request_headers_body_access_is_none() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + request: + phase: request_headers + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + assert_eq!(filter.request_body_access(), BodyAccess::None); + assert_eq!(filter.request_body_mode(), BodyMode::Stream); + } + + #[test] + fn phase_request_body_access_is_readonly() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + request: + phase: request_body + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + assert_eq!(filter.request_body_access(), BodyAccess::ReadOnly); + assert!( + matches!( + filter.request_body_mode(), + BodyMode::StreamBuffer { max_bytes: Some(_) } + ), + "request_body phase should use StreamBuffer" + ); + } + + #[test] + fn needs_request_context_is_true() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + assert!(filter.needs_request_context()); + } + + // ------------------------------------------------------------------------- + // Successful Callout + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn successful_callout_extracts_results() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "flagged": true, + "score": 0.95 + }))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + response: + extract: + - json_path: "$.flagged" + result_key: "flagged" + - json_path: "$.score" + result_key: "score" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Continue), + "should continue after success" + ); + + let results = ctx.filter_results.get("http_callout").expect("should have results"); + assert_eq!(results.get("flagged"), Some("true")); + assert_eq!(results.get("score"), Some("0.95")); + } + + // ------------------------------------------------------------------------- + // Failure Modes + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn failure_mode_closed_rejects() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(500)) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + on_failure: closed + status_on_error: 403 + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Reject(r) if r.status == 403), + "fail-closed should reject with 403" + ); + } + + #[tokio::test] + async fn failure_mode_open_continues() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(500)) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + on_failure: open + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!(matches!(action, FilterAction::Continue), "fail-open should continue"); + } + + // ------------------------------------------------------------------------- + // Timeout + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn timeout_triggers_failure_mode() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/slow")) + .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(10))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/slow" + timeout: "100ms" + request: + phase: request_headers + on_failure: closed + status_on_error: 504 + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Reject(r) if r.status == 504), + "timeout should reject with configured status" + ); + } + + // ------------------------------------------------------------------------- + // Depth Limiting + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn depth_limit_rejects_at_max() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) // should not be called + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + on_failure: closed + max_depth: 1 + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let mut headers = http::HeaderMap::new(); + headers.insert("x-praxis-callout-depth", "1".parse().unwrap()); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers, + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Reject(_)), + "depth >= max_depth should reject" + ); + } + + // ------------------------------------------------------------------------- + // Forward Headers + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn forward_headers_copied_to_callout() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .and(wiremock::matchers::header("x-custom", "my-value")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true}))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + forward_headers: + - "x-custom" + request: + phase: request_headers + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let mut headers = http::HeaderMap::new(); + headers.insert("x-custom", "my-value".parse().unwrap()); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers, + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!(matches!(action, FilterAction::Continue), "forward_headers should work"); + } + + // ------------------------------------------------------------------------- + // Inject Headers + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn inject_headers_from_callout_response() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with( + ResponseTemplate::new(200) + .append_header("x-guard-id", "abc-123") + .set_body_json(serde_json::json!({"ok": true})), + ) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + response: + inject_headers: + - "x-guard-id" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!(matches!(action, FilterAction::Continue)); + + let injected = ctx + .extra_request_headers + .iter() + .find(|(name, _)| name.as_ref() == "x-guard-id"); + assert!(injected.is_some(), "x-guard-id should be injected"); + assert_eq!(injected.unwrap().1, "abc-123"); + } + + // ------------------------------------------------------------------------- + // Request Body Phase + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn request_body_phase_skips_on_request() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + request: + phase: request_body + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Continue), + "request_body phase should skip on_request" + ); + } + + #[tokio::test] + async fn request_body_phase_fires_on_end_of_stream() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"flagged": false}))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_body + response: + extract: + - json_path: "$.flagged" + result_key: "flagged" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + let mut body = Some(bytes::Bytes::from(r#"{"prompt":"hello"}"#)); + + let action = filter.on_request_body(&mut ctx, &mut body, true).await.unwrap(); + assert!(matches!(action, FilterAction::Continue)); + + let results = ctx.filter_results.get("http_callout").expect("should have results"); + assert_eq!(results.get("flagged"), Some("false")); + } + + #[tokio::test] + async fn request_body_phase_skips_non_eos() { + let yaml = serde_yaml::from_str::( + r#" + target: + url: "http://example.com/api" + request: + phase: request_body + "#, + ) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + let mut body = Some(bytes::Bytes::from("chunk")); + + let action = filter.on_request_body(&mut ctx, &mut body, false).await.unwrap(); + assert!( + matches!(action, FilterAction::Continue), + "should skip non-end-of-stream chunks" + ); + } + + // ------------------------------------------------------------------------- + // Circuit Breaker + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn circuit_breaker_trips_after_threshold() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(500)) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + on_failure: closed + status_on_error: 503 + circuit_breaker: + failure_threshold: 2 + recovery_timeout: "60s" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + // Fire enough requests to trip the breaker. + for _ in 0..3 { + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + let _action = filter.on_request(&mut ctx).await.unwrap(); + } + + // After the breaker trips, requests should still be rejected + // without hitting the server. + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + let action = filter.on_request(&mut ctx).await.unwrap(); + assert!( + matches!(action, FilterAction::Reject(r) if r.status == 503), + "circuit breaker should reject after threshold" + ); + } + + // ------------------------------------------------------------------------- + // JSONPath Coercion (via filter) + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn extraction_integer_coerced_to_string() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"count": 42}))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + response: + extract: + - json_path: "$.count" + result_key: "count" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let _action = filter.on_request(&mut ctx).await.unwrap(); + let results = ctx.filter_results.get("http_callout").expect("should have results"); + assert_eq!(results.get("count"), Some("42")); + } + + #[tokio::test] + async fn extraction_null_skipped() { + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/guard")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"field": null}))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + request: + phase: request_headers + response: + extract: + - json_path: "$.field" + result_key: "field" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + let _action = filter.on_request(&mut ctx).await.unwrap(); + // No results written when null. + let has_field = ctx + .filter_results + .get("http_callout") + .is_some_and(|rs| rs.get("field").is_some()); + assert!(!has_field, "null field should not be written to results"); + } + + // ------------------------------------------------------------------------- + // Test Utilities + // ------------------------------------------------------------------------- + + /// Build a minimal [`HttpFilterContext`] for unit tests. + fn make_test_context(req: &praxis_filter::Request) -> praxis_filter::HttpFilterContext<'_> { + use std::sync::LazyLock; + + use praxis_core::id::IdGenerator; + + static TEST_ID_GEN: LazyLock = LazyLock::new(|| IdGenerator::with_seed(0)); + + praxis_filter::HttpFilterContext { + body_done_indices: Vec::new(), + branch_iterations: std::collections::HashMap::new(), + client_addr: None, + cluster: None, + current_filter_id: None, + downstream_tls: false, + extensions: praxis_filter::RequestExtensions::default(), + executed_filter_indices: Vec::new(), + extra_request_headers: Vec::new(), + request_headers_to_remove: Vec::new(), + request_headers_to_set: Vec::new(), + filter_metadata: std::collections::HashMap::new(), + filter_results: std::collections::HashMap::new(), + filter_state: std::collections::HashMap::new(), + health_registry: None, + id_generator: &TEST_ID_GEN, + kv_stores: None, + request: req, + request_body_bytes: 0, + request_body_mode: BodyMode::Stream, + request_start: std::time::Instant::now(), + response_body_bytes: 0, + response_body_mode: BodyMode::Stream, + response_header: None, + response_headers_modified: false, + selected_endpoint_index: None, + time_source: &praxis_core::time::SystemTimeSource, + upstream: None, + rewritten_path: None, + } + } +} From 11efccdd7809a24b841af336c4bd57572fc97316 Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 17:59:13 -0700 Subject: [PATCH 4/6] feat(http-callout): add Lakera Guard example and integration test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Working Lakera Guard content safety config with branch-chain rejection of flagged requests, plus end-to-end integration tests. - examples/configs/ai/lakera-guard.yaml: callout to Lakera Guard API with JSONPath extraction of $.flagged, branch chain to static_response 403 on flagged=true, circuit breaker, fail-closed - tests/integration/tests/suite/examples/lakera_guard.rs: two tests using mock Lakera backend (flagged→403, clean→200 with upstream) - Wire lakera_guard module in mod.rs with http-callout feature gate - Regenerated examples/README.md via sync-example-readme Assisted by Opus 4.6 Signed-off-by: usize --- examples/README.md | 1 + examples/configs/ai/lakera-guard.yaml | 66 +++++++++++ .../tests/suite/examples/lakera_guard.rs | 110 ++++++++++++++++++ tests/integration/tests/suite/examples/mod.rs | 2 + 4 files changed, 179 insertions(+) create mode 100644 examples/configs/ai/lakera-guard.yaml create mode 100644 tests/integration/tests/suite/examples/lakera_guard.rs diff --git a/examples/README.md b/examples/README.md index 51e3237f..92874b69 100644 --- a/examples/README.md +++ b/examples/README.md @@ -31,6 +31,7 @@ page. | [unified-gateway.yaml](configs/ai/anthropic/unified-gateway.yaml) | Routes traffic by classifier-promoted headers so a single listener handles Anthropic Messages, OpenAI Chat Completions, and OpenAI Responses requests | | [credential-injection.yaml](configs/ai/credential-injection.yaml) | Injects per-cluster API credentials into upstream requests and strips client-provided credentials to prevent forwarding | | [json-rpc-routing.yaml](configs/ai/json-rpc-routing.yaml) | Routes JSON-RPC 2.0 requests to different backends based on the "method" field in the JSON request body | +| [lakera-guard.yaml](configs/ai/lakera-guard.yaml) | Sends each request body to Lakera Guard for content moderation | | [mcp-classifier-routing.yaml](configs/ai/mcp-classifier-routing.yaml) | Routes MCP requests by body-derived method and tool name | | [model-to-header-routing.yaml](configs/ai/model-to-header-routing.yaml) | Routes LLM API requests to different backends based on the "model" field in the JSON request body | | [format-routing.yaml](configs/ai/openai/responses/format-routing.yaml) | Routes AI API traffic by detected body format | diff --git a/examples/configs/ai/lakera-guard.yaml b/examples/configs/ai/lakera-guard.yaml new file mode 100644 index 00000000..4a56b338 --- /dev/null +++ b/examples/configs/ai/lakera-guard.yaml @@ -0,0 +1,66 @@ +# Lakera Guard Content Safety +# +# Sends each request body to Lakera Guard for content +# moderation. Flagged requests are rejected with 403; +# clean requests continue to the upstream backend. +# +# Requires the LAKERA_API_KEY environment variable. +# +# Usage: +# LAKERA_API_KEY=... cargo run -p praxis-proxy -- -c examples/configs/ai/lakera-guard.yaml + +listeners: + - name: web + address: "0.0.0.0:8080" + filter_chains: [safety-check, routing] + +filter_chains: + - name: safety-check + filters: + - filter: http_callout + target: + url: "https://api.lakera.ai/v2/guard" + timeout: "2s" + headers: + - name: "Authorization" + value: "Bearer ${LAKERA_API_KEY}" + - name: "Content-Type" + value: "application/json" + request: + phase: request_body + max_body_bytes: 1048576 + response: + extract: + - json_path: "$.flagged" + result_key: "flagged" + on_failure: closed + status_on_error: 403 + circuit_breaker: + failure_threshold: 5 + recovery_timeout: "30s" + branch_chains: + - name: block_flagged + on_result: + filter: http_callout + key: flagged + result: "true" + rejoin: terminal + chains: + - name: reject_flagged + filters: + - filter: static_response + status: 403 + body: "request blocked by content safety" + + - name: routing + filters: + - filter: router + routes: + - path_prefix: "/" + cluster: backend + + - filter: load_balancer + clusters: + - name: backend + endpoints: + - "127.0.0.1:3000" diff --git a/tests/integration/tests/suite/examples/lakera_guard.rs b/tests/integration/tests/suite/examples/lakera_guard.rs new file mode 100644 index 00000000..b58aabff --- /dev/null +++ b/tests/integration/tests/suite/examples/lakera_guard.rs @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Functional integration tests for the `lakera-guard` example config. + +use std::collections::HashMap; + +use praxis_filter::FilterRegistry; +use praxis_test_utils::{ + example_config_path, free_port, http_send, parse_status, patch_yaml, start_backend_with_shutdown, + start_proxy_with_registry, +}; + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +#[test] +fn lakera_guard_rejects_flagged() { + let (mock_lakera_port, _lakera_guard) = start_mock_lakera(true); + let backend_guard = start_backend_with_shutdown("upstream ok"); + let proxy_port = free_port(); + + let config = load_lakera_config(proxy_port, mock_lakera_port, backend_guard.port()); + let registry = build_registry(); + let proxy = start_proxy_with_registry(&config, ®istry); + + let payload = r#"{"prompt":"tell me something bad"}"#; + let raw = http_send( + proxy.addr(), + &format!( + "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + payload.len() + ), + ); + + assert_eq!(parse_status(&raw), 403, "flagged request should be rejected with 403"); +} + +#[test] +fn lakera_guard_passes_clean() { + let (mock_lakera_port, _lakera_guard) = start_mock_lakera(false); + let backend_guard = start_backend_with_shutdown("upstream ok"); + let proxy_port = free_port(); + + let config = load_lakera_config(proxy_port, mock_lakera_port, backend_guard.port()); + let registry = build_registry(); + let proxy = start_proxy_with_registry(&config, ®istry); + + let payload = r#"{"prompt":"hello"}"#; + let raw = http_send( + proxy.addr(), + &format!( + "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + payload.len() + ), + ); + + assert_eq!(parse_status(&raw), 200, "clean request should return 200 from upstream"); + assert!( + raw.contains("upstream ok"), + "clean request should receive upstream body" + ); +} + +// ----------------------------------------------------------------------------- +// Helpers +// ----------------------------------------------------------------------------- + +/// Load and patch the Lakera Guard example config. +fn load_lakera_config(proxy_port: u16, lakera_port: u16, backend_port: u16) -> praxis_core::config::Config { + let path = example_config_path("ai/lakera-guard.yaml"); + let yaml = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {path}: {e}")); + + // Replace the real Lakera API URL with our mock and remove + // the Authorization header that references an env var. + let yaml = yaml + .replace( + "https://api.lakera.ai/v2/guard", + &format!("http://127.0.0.1:{lakera_port}/v2/guard"), + ) + .replace( + " - name: \"Authorization\"\n value: \"Bearer ${LAKERA_API_KEY}\"\n", + "", + ); + + let patched = patch_yaml(&yaml, proxy_port, &HashMap::from([("127.0.0.1:3000", backend_port)])); + praxis_core::config::Config::from_yaml(&patched).unwrap_or_else(|e| panic!("parse lakera-guard.yaml: {e}")) +} + +/// Build a filter registry that includes the http_callout filter. +fn build_registry() -> FilterRegistry { + let mut registry = FilterRegistry::with_builtins(); + praxis_http_callout::register_filters(&mut registry); + registry +} + +/// Start a mock Lakera Guard backend that returns a fixed response. +/// +/// Returns `(port, guard)`. The guard keeps the backend alive. +fn start_mock_lakera(flagged: bool) -> (u16, praxis_test_utils::BackendGuard) { + let body = if flagged { + r#"{"flagged":true}"# + } else { + r#"{"flagged":false}"# + }; + let guard = start_backend_with_shutdown(body); + let port = guard.port(); + (port, guard) +} diff --git a/tests/integration/tests/suite/examples/mod.rs b/tests/integration/tests/suite/examples/mod.rs index 8d9f6f20..7017d8be 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -24,6 +24,8 @@ mod default_config; mod full_flow; mod grpc_detection; mod guardrails; +#[cfg(feature = "http-callout")] +mod lakera_guard; mod header_manipulation; mod health_checks; mod hostname_upstream; From 1f084fc1f73609792023bdc654a4c202bbdfa719 Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 18:47:49 -0700 Subject: [PATCH 5/6] feat(http-callout): add JSONPath body shaping for callout requests Add a target.body config option that reshapes the downstream request body before sending it to the callout target. Each key becomes a field in the callout JSON body; each value is a JSONPath expression evaluated against the downstream body. This solves the impedance mismatch between strict callout APIs (e.g. Lakera Guard rejects unknown fields) and upstream APIs (e.g. Ollama requires "model" in the body). The downstream body goes to the upstream untouched while the callout receives only the mapped fields. Also adds structured logging to the callout execution path: debug for callout initiation, info for success/rejection, warn for failures and body shaping errors, and debug for extracted result values. Example config: target: url: "https://api.lakera.ai/v2/guard" body: messages: "$.messages" This sends {"messages":[...]} to Lakera while the full {"model":"...","messages":[...]} body reaches the upstream. Assisted by Opus 4.6 Signed-off-by: usize --- examples/README.md | 2 +- examples/configs/ai/lakera-guard.yaml | 17 +- filter/http-callout/src/config.rs | 13 +- filter/http-callout/src/extract.rs | 173 +++++++++++++++++- filter/http-callout/src/lib.rs | 67 +++++-- filter/http-callout/src/tests.rs | 56 ++++++ .../tests/suite/examples/lakera_guard.rs | 8 +- tests/integration/tests/suite/examples/mod.rs | 4 +- 8 files changed, 316 insertions(+), 24 deletions(-) diff --git a/examples/README.md b/examples/README.md index 92874b69..1f8ce9bd 100644 --- a/examples/README.md +++ b/examples/README.md @@ -31,7 +31,7 @@ page. | [unified-gateway.yaml](configs/ai/anthropic/unified-gateway.yaml) | Routes traffic by classifier-promoted headers so a single listener handles Anthropic Messages, OpenAI Chat Completions, and OpenAI Responses requests | | [credential-injection.yaml](configs/ai/credential-injection.yaml) | Injects per-cluster API credentials into upstream requests and strips client-provided credentials to prevent forwarding | | [json-rpc-routing.yaml](configs/ai/json-rpc-routing.yaml) | Routes JSON-RPC 2.0 requests to different backends based on the "method" field in the JSON request body | -| [lakera-guard.yaml](configs/ai/lakera-guard.yaml) | Sends each request body to Lakera Guard for content moderation | +| [lakera-guard.yaml](configs/ai/lakera-guard.yaml) | Screens every request body through Lakera Guard for content moderation before forwarding to the upstream | | [mcp-classifier-routing.yaml](configs/ai/mcp-classifier-routing.yaml) | Routes MCP requests by body-derived method and tool name | | [model-to-header-routing.yaml](configs/ai/model-to-header-routing.yaml) | Routes LLM API requests to different backends based on the "model" field in the JSON request body | | [format-routing.yaml](configs/ai/openai/responses/format-routing.yaml) | Routes AI API traffic by detected body format | diff --git a/examples/configs/ai/lakera-guard.yaml b/examples/configs/ai/lakera-guard.yaml index 4a56b338..ec78aa00 100644 --- a/examples/configs/ai/lakera-guard.yaml +++ b/examples/configs/ai/lakera-guard.yaml @@ -1,8 +1,14 @@ # Lakera Guard Content Safety # -# Sends each request body to Lakera Guard for content -# moderation. Flagged requests are rejected with 403; -# clean requests continue to the upstream backend. +# Screens every request body through Lakera Guard for +# content moderation before forwarding to the upstream. +# Flagged requests are rejected with 403; clean requests +# continue to the backend. +# +# The target.body option uses JSONPath to send only the +# "messages" array to Lakera, stripping provider-specific +# fields like "model" that Lakera rejects. The full +# downstream body reaches the upstream untouched. # # Requires the LAKERA_API_KEY environment variable. # @@ -26,6 +32,8 @@ filter_chains: value: "Bearer ${LAKERA_API_KEY}" - name: "Content-Type" value: "application/json" + body: + messages: "$.messages" request: phase: request_body max_body_bytes: 1048576 @@ -38,6 +46,9 @@ filter_chains: circuit_breaker: failure_threshold: 5 recovery_timeout: "30s" + conditions: + - when: + methods: [POST] branch_chains: - name: block_flagged on_result: diff --git a/filter/http-callout/src/config.rs b/filter/http-callout/src/config.rs index f10a5652..65c063a1 100644 --- a/filter/http-callout/src/config.rs +++ b/filter/http-callout/src/config.rs @@ -45,7 +45,7 @@ pub(crate) struct HttpCalloutConfig { // Target // ----------------------------------------------------------------------------- -/// Callout target: URL, timeout, and headers. +/// Callout target: URL, timeout, headers, and body shaping. #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct TargetConfig { @@ -63,6 +63,17 @@ pub(crate) struct TargetConfig { /// Headers to copy from the downstream request. #[serde(default)] pub forward_headers: Vec, + + /// Reshape the downstream request body for the callout. + /// + /// Each key becomes a field in the callout JSON body; each + /// value is a `JSONPath` expression evaluated against the + /// downstream body. When set, only the listed fields are + /// sent — the downstream body goes to upstream untouched. + /// + /// When absent, the downstream body is forwarded verbatim. + #[serde(default)] + pub body: std::collections::HashMap, } /// A static header entry with optional env-var expansion in the value. diff --git a/filter/http-callout/src/extract.rs b/filter/http-callout/src/extract.rs index 649fad46..563db3b0 100644 --- a/filter/http-callout/src/extract.rs +++ b/filter/http-callout/src/extract.rs @@ -1,7 +1,9 @@ // SPDX-License-Identifier: MIT // Copyright (c) 2026 Praxis Contributors -//! `JSONPath` extraction from callout response bodies. +//! `JSONPath` extraction and body shaping for the HTTP callout filter. + +use std::collections::HashMap; use praxis_filter::{FilterError, FilterResultSet}; use serde_json::Value; @@ -68,6 +70,66 @@ impl CompiledExtraction { } } +// ----------------------------------------------------------------------------- +// Body Shaping +// ----------------------------------------------------------------------------- + +/// Pre-compiled field→`JSONPath` mappings for reshaping the callout +/// request body. +/// +/// When present, the downstream body is parsed as JSON and a new +/// object is constructed with only the mapped fields. The original +/// downstream body continues to the upstream untouched. +#[derive(Debug)] +pub(crate) struct BodyShaper { + /// Compiled field mappings: `(output_field_name, jsonpath)`. + fields: Vec<(String, JsonPath)>, +} + +impl BodyShaper { + /// Compile a set of field→`JSONPath` mappings at config time. + /// + /// # Errors + /// + /// Returns [`FilterError`] if any `JSONPath` expression is invalid. + pub(crate) fn compile(mappings: &HashMap) -> Result { + let mut fields = Vec::with_capacity(mappings.len()); + for (field, expr) in mappings { + let path = JsonPath::parse(expr).map_err(|e| -> FilterError { + format!("http_callout: invalid body JSONPath for field '{field}': {e}").into() + })?; + fields.push((field.clone(), path)); + } + // Sort for deterministic output. + fields.sort_by(|(a, _), (b, _)| a.cmp(b)); + Ok(Self { fields }) + } + + /// Whether any field mappings are configured. + pub(crate) fn is_empty(&self) -> bool { + self.fields.is_empty() + } + + /// Reshape a raw body using the compiled `JSONPath` mappings. + /// + /// Parses `raw` as JSON, evaluates each mapping, and builds a + /// new JSON object. Returns `None` if `raw` is not valid JSON. + pub(crate) fn shape(&self, raw: &[u8]) -> Option> { + let source: Value = serde_json::from_slice(raw).ok()?; + let mut output = serde_json::Map::with_capacity(self.fields.len()); + + for (field, path) in &self.fields { + let node_list = path.query(&source); + let nodes: Vec<&Value> = node_list.all(); + if let Some(value) = nodes.first() { + output.insert(field.clone(), (*value).clone()); + } + } + + serde_json::to_vec(&Value::Object(output)).ok() + } +} + // ----------------------------------------------------------------------------- // Coercion // ----------------------------------------------------------------------------- @@ -191,4 +253,113 @@ mod tests { ext.evaluate(&json, &mut rs).unwrap(); assert!(rs.get("key").is_none(), "no-match should be skipped"); } + + // ------------------------------------------------------------------------- + // BodyShaper + // ------------------------------------------------------------------------- + + #[test] + fn body_shaper_empty_mappings() { + let shaper = BodyShaper::compile(&HashMap::new()).unwrap(); + assert!(shaper.is_empty(), "empty mappings should be empty"); + } + + #[test] + fn body_shaper_picks_single_field() { + let mut mappings = HashMap::new(); + mappings.insert("messages".into(), "$.messages".into()); + let shaper = BodyShaper::compile(&mappings).unwrap(); + + let input = serde_json::to_vec(&json!({ + "model": "gpt-4", + "messages": [{"role": "user", "content": "hi"}] + })) + .unwrap(); + + let output = shaper.shape(&input).unwrap(); + let parsed: Value = serde_json::from_slice(&output).unwrap(); + + assert!(parsed.get("messages").is_some(), "messages should be present"); + assert!(parsed.get("model").is_none(), "model should be stripped"); + } + + #[test] + fn body_shaper_picks_multiple_fields() { + let mut mappings = HashMap::new(); + mappings.insert("messages".into(), "$.messages".into()); + mappings.insert("stream".into(), "$.stream".into()); + let shaper = BodyShaper::compile(&mappings).unwrap(); + + let input = serde_json::to_vec(&json!({ + "model": "gpt-4", + "messages": [{"role": "user", "content": "hi"}], + "stream": true, + "temperature": 0.7 + })) + .unwrap(); + + let output = shaper.shape(&input).unwrap(); + let parsed: Value = serde_json::from_slice(&output).unwrap(); + + assert!(parsed.get("messages").is_some(), "messages should be present"); + assert!(parsed.get("stream").is_some(), "stream should be present"); + assert!(parsed.get("model").is_none(), "model should be stripped"); + assert!(parsed.get("temperature").is_none(), "temperature should be stripped"); + } + + #[test] + fn body_shaper_missing_field_omitted() { + let mut mappings = HashMap::new(); + mappings.insert("messages".into(), "$.messages".into()); + mappings.insert("absent".into(), "$.nonexistent".into()); + let shaper = BodyShaper::compile(&mappings).unwrap(); + + let input = serde_json::to_vec(&json!({ + "messages": [{"role": "user", "content": "hi"}] + })) + .unwrap(); + + let output = shaper.shape(&input).unwrap(); + let parsed: Value = serde_json::from_slice(&output).unwrap(); + + assert!(parsed.get("messages").is_some(), "messages should be present"); + assert!(parsed.get("absent").is_none(), "missing field should be omitted"); + } + + #[test] + fn body_shaper_invalid_json_returns_none() { + let mut mappings = HashMap::new(); + mappings.insert("x".into(), "$.x".into()); + let shaper = BodyShaper::compile(&mappings).unwrap(); + + assert!(shaper.shape(b"not json").is_none(), "invalid JSON should return None"); + } + + #[test] + fn body_shaper_invalid_jsonpath_rejected() { + let mut mappings = HashMap::new(); + mappings.insert("x".into(), "$[invalid".into()); + let err = BodyShaper::compile(&mappings).expect_err("expected error"); + assert!( + err.to_string().contains("invalid body JSONPath"), + "should report invalid JSONPath: {err}" + ); + } + + #[test] + fn body_shaper_nested_extraction() { + let mut mappings = HashMap::new(); + mappings.insert("content".into(), "$.messages[0].content".into()); + let shaper = BodyShaper::compile(&mappings).unwrap(); + + let input = serde_json::to_vec(&json!({ + "messages": [{"role": "user", "content": "hello world"}] + })) + .unwrap(); + + let output = shaper.shape(&input).unwrap(); + let parsed: Value = serde_json::from_slice(&output).unwrap(); + + assert_eq!(parsed["content"], "hello world", "should extract nested value"); + } } diff --git a/filter/http-callout/src/lib.rs b/filter/http-callout/src/lib.rs index a91fb3b7..5a2f63a0 100644 --- a/filter/http-callout/src/lib.rs +++ b/filter/http-callout/src/lib.rs @@ -29,11 +29,11 @@ use praxis_core::callout::{ use praxis_filter::{ BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, Rejection, parse_filter_config, }; -use tracing::debug; +use tracing::{debug, info, warn}; use crate::{ config::{FailureModeConfig, HttpCalloutConfig, Phase, expand_env_vars, validate_callout_url}, - extract::CompiledExtraction, + extract::{BodyShaper, CompiledExtraction}, }; // ----------------------------------------------------------------------------- @@ -56,6 +56,9 @@ const FILTER_NAME: &str = "http_callout"; /// /// [`FilterResultSet`]: praxis_filter::FilterResultSet struct HttpCalloutFilter { + /// Pre-compiled body shaper for reshaping the callout body. + body_shaper: BodyShaper, + /// Reusable HTTP callout client. client: CalloutClient, @@ -99,6 +102,7 @@ impl HttpCalloutFilter { validate_callout_url(&cfg.target.url)?; + let body_shaper = BodyShaper::compile(&cfg.target.body)?; let headers = parse_static_headers(&cfg)?; let forward_headers = parse_header_names(&cfg.target.forward_headers, "forward_header")?; let extractions = compile_extractions(&cfg)?; @@ -108,6 +112,7 @@ impl HttpCalloutFilter { let client = build_callout_client(&cfg)?; Ok(Box::new(Self { + body_shaper, client, extractions, forward_headers, @@ -160,8 +165,12 @@ impl HttpCalloutFilter { for extraction in &self.extractions { extraction.evaluate(&json, results)?; } + debug!( + results = ?results, + "extracted callout results" + ); } else { - debug!("callout response body is not valid JSON; skipping extraction"); + warn!("callout response body is not valid JSON; skipping extraction"); } } @@ -195,24 +204,58 @@ impl HttpCalloutFilter { FilterAction::Reject(rejection) } - /// Execute the callout and process the result. - async fn execute_callout( + /// Apply body shaping if configured, otherwise pass through. + fn shape_body(&self, body: Option>) -> Option> { + match body { + Some(raw) if !self.body_shaper.is_empty() => { + let result = self.body_shaper.shape(&raw); + if result.is_none() { + warn!(url = %self.url, "body shaping failed (not valid JSON); forwarding raw body"); + } + result.or(Some(raw)) + }, + other => other, + } + } + + /// Map a [`CalloutResult`] to a [`FilterAction`], logging the + /// outcome. + fn handle_result( &self, + result: CalloutResult, ctx: &mut HttpFilterContext<'_>, - body: Option>, ) -> Result { - let request = self.build_request(ctx, body); - let result = self.client.execute(request).await; - match result { - CalloutResult::Success(response) => self.handle_success(&response, ctx), + CalloutResult::Success(response) => { + info!(url = %self.url, status = response.status, "callout succeeded"); + self.handle_success(&response, ctx) + }, CalloutResult::Failed => { - debug!("callout failed (open mode); continuing"); + warn!(url = %self.url, "callout failed; continuing (fail-open)"); Ok(FilterAction::Continue) }, - CalloutResult::Rejected(rejection) => Ok(self.build_rejection(None, rejection.status)), + CalloutResult::Rejected(rejection) => { + info!(url = %self.url, status = rejection.status, "callout rejected request"); + Ok(self.build_rejection(None, rejection.status)) + }, } } + + /// Execute the callout and process the result. + async fn execute_callout( + &self, + ctx: &mut HttpFilterContext<'_>, + body: Option>, + ) -> Result { + let body_len = body.as_ref().map_or(0, Vec::len); + let callout_body = self.shape_body(body); + + debug!(url = %self.url, body_bytes = body_len, "executing callout"); + + let request = self.build_request(ctx, callout_body); + let result = self.client.execute(request).await; + self.handle_result(result, ctx) + } } // ----------------------------------------------------------------------------- diff --git a/filter/http-callout/src/tests.rs b/filter/http-callout/src/tests.rs index 263c2fe0..684bb1a7 100644 --- a/filter/http-callout/src/tests.rs +++ b/filter/http-callout/src/tests.rs @@ -626,6 +626,62 @@ mod filter_tests { ); } + // ------------------------------------------------------------------------- + // Body Shaping + // ------------------------------------------------------------------------- + + #[tokio::test] + async fn body_shaping_strips_extra_fields() { + let mock_server = MockServer::start().await; + + // The mock expects a body with only "messages", no "model". + Mock::given(method("POST")) + .and(path("/guard")) + .and(wiremock::matchers::body_json(serde_json::json!({ + "messages": [{"role": "user", "content": "hi"}] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"flagged": false}))) + .mount(&mock_server) + .await; + + let yaml = serde_yaml::from_str::(&format!( + r#" + target: + url: "{}/guard" + body: + messages: "$.messages" + request: + phase: request_body + response: + extract: + - json_path: "$.flagged" + result_key: "flagged" + "#, + mock_server.uri() + )) + .unwrap(); + + let filter = HttpCalloutFilter::from_config(&yaml).unwrap(); + + let req = praxis_filter::Request { + method: http::Method::POST, + uri: "/test".parse().unwrap(), + headers: http::HeaderMap::new(), + }; + let mut ctx = make_test_context(&req); + + // Downstream body has "model" which Lakera would reject. + let mut body = Some(bytes::Bytes::from( + r#"{"model":"gpt-4","messages":[{"role":"user","content":"hi"}]}"#, + )); + + let action = filter.on_request_body(&mut ctx, &mut body, true).await.unwrap(); + assert!(matches!(action, FilterAction::Continue), "shaped body should succeed"); + + let results = ctx.filter_results.get("http_callout").expect("should have results"); + assert_eq!(results.get("flagged"), Some("false")); + } + // ------------------------------------------------------------------------- // Circuit Breaker // ------------------------------------------------------------------------- diff --git a/tests/integration/tests/suite/examples/lakera_guard.rs b/tests/integration/tests/suite/examples/lakera_guard.rs index b58aabff..e4a32f9f 100644 --- a/tests/integration/tests/suite/examples/lakera_guard.rs +++ b/tests/integration/tests/suite/examples/lakera_guard.rs @@ -25,11 +25,11 @@ fn lakera_guard_rejects_flagged() { let registry = build_registry(); let proxy = start_proxy_with_registry(&config, ®istry); - let payload = r#"{"prompt":"tell me something bad"}"#; + let payload = r#"{"model":"test","messages":[{"role":"user","content":"bad"}]}"#; let raw = http_send( proxy.addr(), &format!( - "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + "POST /v1/chat/completions HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", payload.len() ), ); @@ -47,11 +47,11 @@ fn lakera_guard_passes_clean() { let registry = build_registry(); let proxy = start_proxy_with_registry(&config, ®istry); - let payload = r#"{"prompt":"hello"}"#; + let payload = r#"{"model":"test","messages":[{"role":"user","content":"hello"}]}"#; let raw = http_send( proxy.addr(), &format!( - "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + "POST /v1/chat/completions HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", payload.len() ), ); diff --git a/tests/integration/tests/suite/examples/mod.rs b/tests/integration/tests/suite/examples/mod.rs index 7017d8be..091c265f 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -24,11 +24,11 @@ mod default_config; mod full_flow; mod grpc_detection; mod guardrails; -#[cfg(feature = "http-callout")] -mod lakera_guard; mod header_manipulation; mod health_checks; mod hostname_upstream; +#[cfg(feature = "http-callout")] +mod lakera_guard; mod least_connections; mod logging; mod max_body_guard; From 8c92b41f02c343847dd28746935fd46627436a6a Mon Sep 17 00:00:00 2001 From: usize Date: Fri, 26 Jun 2026 19:41:54 -0700 Subject: [PATCH 6/6] fix(server): suppress cargo-machete false positive for http-callout The praxis-http-callout dependency is consumed via build.rs auto-discovery codegen, not through a direct `use` in source. cargo-machete doesn't see the generated include! and reports it as unused. Assisted by Opus 4.6 Signed-off-by: usize --- server/Cargo.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/server/Cargo.toml b/server/Cargo.toml index c190777d..f64d69e1 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -49,3 +49,6 @@ ai-inference = ["praxis-filter/ai-inference", "praxis-protocol/response-store"] ext-proc = ["praxis-filter/ext-proc"] cpex-policy-engine = ["praxis-filter/cpex-policy-engine"] http-callout = ["dep:praxis-http-callout", "praxis-core/callout"] + +[package.metadata.cargo-machete] +ignored = ["praxis-http-callout"] # consumed via build.rs auto-discovery codegen