feat: add CPEX policy filter

Cargo.toml

@@ -39,9 +39,10 @@ async-trait = "0.1.89"
 benchmarks = { path = "benchmarks" }
 bytes = "1.12.0"
 chrono = { version = "0.4.45", default-features = false, features = ["clock"] }
-dashmap = "6.2.1"
+cpex = { version = "0.2.0", git = "https://github.com/contextforge-org/cpex.git", rev = "702ab163c64fe762318cb2bc5fb9b088869dd1af", default-features = false, features = ["jwt", "oauth", "pii", "audit", "cedar", "cel", "valkey"] }
 clap = { version = "4.6.1", features = ["derive"] }
 criterion = { version = "0.8.2", features = ["async_tokio"] }
+dashmap = "6.2.1"
 futures = "0.3.32"
 h2 = "0.4.15"
 http = "1.4.2"

deny.toml

@@ -35,4 +35,6 @@ highlight = "all"
 unknown-registry = "deny"
 unknown-git = "deny"
 allow-registry = ["https://github.com/rust-lang/crates.io-index"]
-allow-git = []
+# CPEX policy-runtime crates (feature-gated behind `--features cpex`).
+# Pinned by exact rev in Cargo.toml until the crates are published.
+allow-git = ["https://github.com/contextforge-org/cpex"]

docs/filters/http/security/cpex.md

@@ -0,0 +1,35 @@
+<!-- Generated by: cargo xtask generate-filter-docs -->
+<!-- Do not edit manually -->
+
+# `cpex`
+
+Configuration block for a `cpex` filter slot in a Praxis filter chain.
+
+Requires Cargo feature: `cpex`.
+
+## Configuration Notes
+
+Praxis filter configs are flat: the filter's typed fields sit directly under the `- filter:` entry alongside the structural keys (`name`, `conditions`), not nested under a `config:` wrapper. See `examples/configs/security/cpex.yaml` for a runnable example.
+
+The referenced YAML is the CPEX policy document — plugins, routes, and identity-source declarations. The filter loads it once at construction and rejects misconfigured policy at server startup (fail-fast rather than at first request).
+
+## Configuration
+
+| Field | Type | Required | Description |
+|-------|------|---------|-------------|
+| `config_path` | string | yes | Filesystem path to the CPEX YAML policy document. |
+| `body_access` | `read_only` \| `read_write` | no | Body-access tier. `ReadOnly` (default) lets APL inspect request and response bodies for routing / policy decisions but discards any mutations. `ReadWrite` enables the CMF → JSON-RPC re-serialization round-trip so APL field mutators (e.g. `args.ssn: redact(!perm.view_ssn)`) rewrite the upstream body and response. Pay the round-trip cost only when needed. |
+| `require_mcp_metadata` | bool | no | Fail-closed policy gate for misconfigured chains. When `true` (default), `on_request_body` rejects any request that reaches it without `mcp.method` filter-metadata. The metadata is set by praxis's built-in `mcp` filter, so its absence means either (a) the `mcp` filter is missing from the chain, or (b) it is ordered AFTER `cpex` instead of before. Either is a misconfiguration that would silently bypass CMF/APL policy. Set to `false` only when intentionally fronting non-MCP traffic through `cpex` for identity-only enforcement (legacy behavior). Note: MCP methods that legitimately carry no entity (e.g. `tools/list`, `initialize`, `prompts/list`) still pass — `require_mcp_metadata` only rejects when the metadata is missing entirely. |
+| `init_timeout_secs` | u64 | no | Maximum time, in seconds, to wait for `PluginManager::initialize` at filter construction. Identity plugins fetch JWKS over HTTPS during init; a reachable-but-unresponsive identity provider would otherwise hang startup or hot-reload indefinitely. On expiry, filter construction returns an error and the server fails fast. 30s is generous for legitimate cold-cache JWKS fetches over the public internet, while short enough that misbehavior is noticed during the deploy. |
+| `max_buffer_bytes` | usize | no | Maximum request/response body bytes buffered in `ReadWrite` mode. `ReadWrite` uses `StreamBuffer` to accumulate the whole body before APL field mutators run; without a cap an oversized payload could exhaust memory. Ignored in `ReadOnly` mode, which streams. The pipeline rejects an unbounded buffer at config load, so this always carries a concrete ceiling. |
+
+## Example
+
+```yaml
+filter: cpex
+config_path: /etc/praxis/cpex.yaml
+body_access: read_write       # optional; default read_only
+require_mcp_metadata: true    # optional; default true
+init_timeout_secs: 30         # optional; default 30
+max_buffer_bytes: 10485760    # optional; default 10 MiB (read_write only)
+```

docs/filters/reference.md

@@ -44,6 +44,7 @@ Built-in filters organized by protocol and category.
 | Filter | Feature | Description |
 |--------|---------|-------------|
 | [`cors`](http/security/cors.md) | - | Spec-compliant CORS filter implementing origin validation, preflight handling, and response header injection. |
+| [`cpex`](http/security/cpex.md) | `cpex` | Configuration block for a `cpex` filter slot in a Praxis filter chain. |
 | [`credential_injection`](http/security/credential_injection.md) | - | Injects per-cluster API credentials into upstream requests. |
 | [`csrf`](http/security/csrf.md) | - | CSRF protection filter that validates request origins against a trusted allowlist. |
 | [`forwarded_headers`](http/security/forwarded_headers.md) | - | Injects `X-Forwarded-For`, `X-Forwarded-Proto`, and `X-Forwarded-Host` headers into upstream requests. |

examples/README.md

@@ -129,6 +129,7 @@ page.
 | File | Description |
 | ------ | ------------- |
 | [cors.yaml](configs/security/cors.yaml) | Spec-compliant CORS filter with preflight handling, origin validation, and credential support |
+| [cpex.yaml](configs/security/cpex.yaml) | Embeds the CPEX policy runtime in-process to enforce multi-source JWT identity, APL route policy, RFC 8693 OAuth 2.0 token exchange, PII scanning, audit emission, and (under `body_access: read_write`) request / response body rewriting |
 | [csrf.yaml](configs/security/csrf.yaml) | Cross-site request forgery protection via origin validation |
 | [downstream-read-timeout.yaml](configs/security/downstream-read-timeout.yaml) | Protects against slow client attacks by limiting how long the proxy waits for data from downstream clients |
 | [forwarded-headers.yaml](configs/security/forwarded-headers.yaml) | Injects X-Forwarded-For, X-Forwarded-Proto, and X-Forwarded-Host into upstream requests |

examples/configs/security/cpex.yaml

@@ -0,0 +1,78 @@
+# CPEX Security Filter
+#
+# Embeds the CPEX policy runtime in-process to enforce multi-source
+# JWT identity, APL route policy, RFC 8693 OAuth 2.0 token exchange,
+# PII scanning, audit emission, and (under `body_access: read_write`)
+# request / response body rewriting.
+#
+# The cpex filter consumes `mcp.method` / `mcp.name` metadata stashed
+# by praxis's built-in `mcp` filter, which MUST be ordered before it
+# in the chain.
+#
+# `config_path` points at the CPEX policy YAML (plugins + routes) that
+# the filter loads once at construction. It is operator-supplied, not
+# shipped here: a real policy carries your own issuers, secrets, routes,
+# PDPs, and delegators. See the praxis-demos repo
+# (`demos/cpex/cpex.yaml`) for a fully-featured policy, and
+# `tests/integration/fixtures/cpex-policy.yaml` for the minimal
+# single-HS256-identity-plugin policy the integration test loads.
+#
+# `require_mcp_metadata: true` (the default) fail-closes when the
+# `mcp` filter is missing or ordered after `cpex` — this guards
+# against a misconfigured chain silently bypassing policy.
+#
+# This example builds only when the `cpex` cargo feature is enabled:
+#
+#   cargo run --features cpex -p praxis -- \
+#     -c examples/configs/security/cpex.yaml
+#
+# Exercise (assumes the backend at :3000 echoes 200):
+#
+#   # Missing Authorization → 401 with WWW-Authenticate and
+#   # X-Cpex-Violation header:
+#   curl -i -X POST http://localhost:8080/mcp \
+#     -H "Content-Type: application/json" \
+#     -d '{"jsonrpc":"2.0","id":1,"method":"tools/call",
+#          "params":{"name":"echo","arguments":{}}}'
+#
+#   # Token rejected by HS256 signature mismatch → 401:
+#   curl -i -X POST http://localhost:8080/mcp \
+#     -H "Authorization: Bearer bogus.token.bytes" \
+#     -H "Content-Type: application/json" \
+#     -d '{"jsonrpc":"2.0","id":1,"method":"tools/call",
+#          "params":{"name":"echo","arguments":{}}}'
+#
+# Supply your own policy document at `config_path` before running this
+# example; the minimal test fixture uses a placeholder shared secret
+# that must not be used beyond local experimentation.
+
+listeners:
+  - name: default
+    address: "127.0.0.1:8080"
+    filter_chains:
+      - main
+
+filter_chains:
+  - name: main
+    filters:
+      - filter: mcp
+        # Default mode parses MCP JSON-RPC bodies and stashes
+        # mcp.method / mcp.name in filter_metadata. Must precede cpex.
+
+      - filter: cpex
+        config_path: /etc/praxis/cpex-policy.yaml
+        # Fail-closed when mcp.method is missing. Set to `false` only
+        # when intentionally fronting non-MCP traffic through cpex for
+        # identity-only enforcement.
+        require_mcp_metadata: true
+
+      - filter: router
+        routes:
+          - path_prefix: "/"
+            cluster: backend
+
+      - filter: load_balancer
+        clusters:
+          - name: backend
+            endpoints:
+              - "127.0.0.1:3000"

filter/Cargo.toml

@@ -17,6 +17,10 @@ name = "praxis_filter"
 default = ["ai-inference"]
 ai-inference = ["dep:secrecy", "dep:sqlx", "dep:tokio"]
 ext-proc = ["dep:praxis-proto", "dep:tonic", "dep:prost-wkt-types"]
+cpex = [
+    "dep:cpex",
+    "dep:tokio",
+]
 
 [package.metadata.cargo-machete]
 ignored = ["praxis-proto", "prost-wkt-types", "tonic"]
@@ -27,6 +31,7 @@ workspace = true
 [dependencies]
 async-trait = { workspace = true }
 bytes = { workspace = true }
+cpex = { workspace = true, optional = true }
 dashmap = { workspace = true }
 http = { workspace = true }
 percent-encoding = { workspace = true }
@@ -41,10 +46,18 @@ serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 sqlx = { workspace = true, optional = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, optional = true }
+# Optional, shared by `ai-inference` and `cpex`. The `cpex` feature
+# drives plugin `initialize()` at filter construction (the praxis
+# filter-factory signature is sync, so we block on a single-thread
+# runtime); the time/net features let JWT identity plugins fetch JWKS
+# at init time.
+tokio = { workspace = true, optional = true, features = ["rt", "time", "net"] }
 tonic = { workspace = true, optional = true }
 tracing = { workspace = true }
 zeroize = { workspace = true }
 
 [dev-dependencies]
+chrono = { workspace = true }
+jsonwebtoken = "9"
+tempfile = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread"] }

filter/src/builtins/http/mod.rs

@@ -40,6 +40,8 @@ pub use ai::token_usage::{TokenUsage, TokenUsageProvider, extract_token_usage};
 pub use ai::{A2aFilter, JsonRpcFilter, McpFilter, TokenUsageHeadersFilter};
 pub use observability::{AccessLogFilter, RequestIdFilter};
 pub use payload_processing::{CompressionFilter, JsonBodyFieldFilter};
+#[cfg(feature = "cpex")]
+pub use security::CpexFilter;
 pub use security::{
     ContainsValue, CorsFilter, CredentialInjectionFilter, CsrfFilter, DisallowedOriginMode, ForwardedHeadersFilter,
     GuardrailsAction, GuardrailsFilter, IpAclFilter, PiiKind, RuleTargetKind,