feat(filter): outbound chain binding with authority-bound deferred credentials

[leseb]

What does this PR do?

Implements issue #1074: lets a filter bind a nested "outbound" filter chain (named or inline) that runs against the resolved upstream, exposing a small public FilteredSubrequestExecutor::run() API that returns either a buffered or a streaming response. Bound chains are held to the same validation top-level chains face — filter-count cap, empty-predicate rejection, inline-cluster SSRF/insecure-TLS gating, branch-chain constraints (max-iterations ceiling, nesting depth, per-filter and configuration-wide total-branch caps), and rejection of terminal/TCP filters — all of which they would otherwise bypass by never appearing in Config::filter_chains, with every scan traversing branch sub-chains. Credentials are staged during request processing and materialized only into a request bound for the canonical logical authority they were issued for, after header sanitization, and only when actually injected, so a secret cannot reach a shared-vhost endpoint under a different Host. Bound outbound chains participate fully in the config-reload contract, being rebuilt and re-injected with runtime resources (KV stores, session stores, subrequest client) including into nested and branch-embedded pipelines. No IRR or AI-client migration is included.

Which issue(s) does this relate to?

Fixes #1074

Checklist

• Signed off all commits (git commit -s)
• Tests added or updated
• Documentation updated (if applicable)
• make lint && make test && make test-integration passes locally

Does this introduce a breaking change?

No.

[praxis-bot-app]

PR too large: 5334 lines added (limit: 500, excludes Cargo files, tests, docs, examples, and benchmarks). Please split into smaller PRs.

[praxis-bot]

PR Review: Outbound chain binding with authority-bound deferred credentials

Adds a chain-binding mechanism that lets application filters (e.g. AI callouts) own a prebuilt outbound FilterPipeline resolved at construction time, plus an authority-bound deferred credential system that materializes secrets only after destination resolution. Bound chains face the same validation gates as top-level chains (cardinality, SSRF, branch limits, terminal/TCP filter rejection, ordering). Budgets are shared across binding boundaries so the ceiling cannot be evaded.

Assessment

This is a well-structured, security-conscious PR with thorough build-time validation and extensive test coverage across all modules. The credential binding model is sound -- secrets are sealed until the resolved authority matches, and injection happens after sanitization with Host pinning to prevent vhost-targeted exfiltration. The build-time budget sharing (materialization instances and branch definitions) correctly prevents fan-out evasion across binding boundaries. Two medium-severity test coverage gaps are noted below.

Findings

Severity	File	Finding
Medium	credentials.rs	IPv6 authority parsing paths lack test coverage
Medium	credentials.rs	inject_canonical silently swallows invariant violation

[twghu]

CalloutStreamingBody: a response-ceiling breach leaves the stream runnable

The defect

CalloutStreamingBody::checked (crates/filter/src/filtered_subrequest/streaming.rs) returns Err when a chunk would push the stream past max_response_bytes, but leaves the body in a runnable state — finished stays false and inner stays Some. Tracing next_chunk: a caller that polls again finds pending empty, deferred_error None, finished false and inner live, so it pulls the next upstream chunk and returns it if it fits.

Two things compound:

• The rejected chunk is never emitted, so emitted_bytes never advances — the return Err at the ceiling check precedes self.emitted_bytes = total. Later chunks are therefore measured against a pre-breach count.
• The breach is recorded nowhere, so nothing after the error knows the body is no longer contiguous.

Net effect: the oversized chunk is dropped out of the middle of the body while smaller subsequent chunks keep flowing. The consumer receives a body with a hole in it rather than an error-terminated stream — for a callout response (an SSE stream, a JSON document) a worse outcome than overshooting the ceiling would have been.

The deferred_error arm twelve lines below already does the right thing (finished = true; inner = None). That inconsistency is what suggests the omission at the ceiling check is an oversight rather than a deliberate "callers must stop on Err" assumption.

Evidence

Added run_streaming_ceiling_breach_ends_the_stream: 20-byte upstream body, 15-byte ceiling, test_terminal_event in the chain so a 14-byte data: [DONE] completion event follows EOF. Because the rejected 20-byte chunk never advances the counter, that 14-byte event still fits under 15. Against the unpatched code:

thread '...run_streaming_ceiling_breach_ends_the_stream' panicked at
crates/filter/src/filtered_subrequest/tests.rs:1432:5:
a ceiling breach must end the stream; polling again must not resume it,
got resumed with 14 more bytes

That is the completion event being delivered after a fatal error.

The change

Split checked into a teardown wrapper plus an account helper, so every error path out of the accounting — the ceiling breach and the checked_add overflow, which had the same problem — sets finished = true and clears pending before the error propagates. The next poll then hits the existing if self.finished { return Ok(None) } arm.

One deliberate divergence from the deferred_error arm: inner is left in place. That arm can null it safely because drain_completion has already populated held_extensions. At the ceiling check it has not, and checked is synchronous — so taking inner there would drop the upstream body without awaiting SubResponseBody::cancel(), and would strand the parent extensions where swap_extensions cannot reach them (it falls back to held_extensions only when inner is None). Leaving inner alive keeps a later cancel()/suppress() able to cancel the upstream asynchronously and recover the extensions. finished = true is what stops the stream; dropping inner was never the part doing the work.

The test's pull loop tolerates the upstream splitting those 20 bytes across several chunks, since spawn_raw_backend issues a single write_all and the test shouldn't rest on non-coalescing.

Verification

• cargo test -p praxis-proxy-filter --lib filtered_subrequest::tests::run_streaming — 7/7 pass
• cargo clippy -p praxis-proxy-filter --all-targets --all-features -- -D warnings — clean
• RUSTDOCFLAGS=-Dwarnings cargo doc -p praxis-proxy-filter --no-deps --document-private-items — clean
• The same test fails as shown above with the streaming.rs hunk reverted and the test retained

Not run: cargo +nightly fmt --all -- --check (no nightly toolchain locally; rustfmt.toml relies on nightly-only options, so a stable run would be a false pass).

Patches:

git --no-pager diff -- crates/filter/src/filtered_subrequest/streaming.rs
diff --git a/crates/filter/src/filtered_subrequest/streaming.rs b/crates/filter/src/filtered_subrequest/streaming.rs
index 7bb94737..d7b42f52 100644
--- a/crates/filter/src/filtered_subrequest/streaming.rs
+++ b/crates/filter/src/filtered_subrequest/streaming.rs
@@ -349,8 +349,34 @@ impl CalloutStreamingBody {
         }
     }
 
-    /// Account one outgoing chunk against the response byte ceiling.
+    /// Account one outgoing chunk against the response byte ceiling, ending the
+    /// stream if it does not fit.
+    ///
+    /// A breach is *terminal*. The rejected chunk is never emitted, so
+    /// `emitted_bytes` does not advance; without marking the stream finished a
+    /// caller that polled again would resume around the hole and keep delivering
+    /// later chunks against the stale count, handing the consumer a body with a
+    /// gap in it instead of an error-terminated stream. Marking it finished and
+    /// dropping queued chunks makes the next poll report end-of-stream, matching
+    /// the `deferred_error` arm of [`next_chunk`](StreamingResponseBody::next_chunk).
+    ///
+    /// The inner body is deliberately left in place: tearing it down here would
+    /// have to drop the upstream synchronously, whereas a subsequent
+    /// [`cancel`](StreamingResponseBody::cancel) or
+    /// [`suppress`](StreamingResponseBody::suppress) can still cancel it
+    /// asynchronously and recover the parent extensions.
     fn checked(&mut self, chunk: Bytes) -> Result<Option<Bytes>, FilterError> {
+        let outcome = self.account(chunk);
+        if outcome.is_err() {
+            self.finished = true;
+            self.pending.clear();
+        }
+        outcome
+    }
+
+    /// Add `chunk` to the emitted-byte total, rejecting a counter overflow or a
+    /// breach of the response ceiling.
+    fn account(&mut self, chunk: Bytes) -> Result<Option<Bytes>, FilterError> {
         let total = self
             .emitted_bytes
             .checked_add(chunk.len())

git --no-pager diff -- crates/filter/src/filtered_subrequest/tests.rs
diff --git a/crates/filter/src/filtered_subrequest/tests.rs b/crates/filter/src/filtered_subrequest/tests.rs
index 9d99ac70..d5c023d1 100644
--- a/crates/filter/src/filtered_subrequest/tests.rs
+++ b/crates/filter/src/filtered_subrequest/tests.rs
@@ -1360,6 +1360,81 @@ async fn run_streaming_enforces_response_byte_ceiling() {
     );
 }
 
+#[tokio::test]
+#[expect(clippy::large_futures, reason = "drives the full executor future in a test")]
+async fn run_streaming_ceiling_breach_ends_the_stream() {
+    use std::{
+        sync::Arc,
+        time::{Duration, Instant},
+    };
+
+    // A 20-byte upstream body (hex chunk length 14), behind a chain whose
+    // completion hook emits a 14-byte terminal event at EOF.
+    let (addr, backend) = spawn_raw_backend(
+        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n14\r\nAAAAAAAAAAAAAAAAAAAA\r\n0\r\n\r\n",
+    )
+    .await;
+    let registry = callout_registry();
+    let mut entries: Vec<crate::FilterEntry> =
+        serde_yaml::from_str(&routed_chain_yaml(addr, "- filter: test_terminal_event\n")).unwrap();
+    let pipeline = Arc::new(crate::FilterPipeline::build(&mut entries, &registry).unwrap());
+
+    // 20 bytes of body cannot fit under a 15-byte ceiling however the upstream
+    // frames it, so some chunk must be rejected. A rejected chunk is never
+    // emitted and so never advances `emitted_bytes` — leaving room for the
+    // 14-byte completion event to pass the check afterwards. A stream that kept
+    // running would therefore resume around the hole and hand the caller a body
+    // with a gap in it instead of an error-terminated one.
+    let executor = streaming_executor(15);
+    let request = crate::SubRequest {
+        method: http::Method::GET,
+        uri: http::Uri::from_static("/"),
+        headers: HeaderMap::new(),
+        body: bytes::Bytes::new(),
+    };
+    let deadline = Instant::now() + Duration::from_secs(5);
+
+    let mut body = match executor
+        .run(&pipeline, &request, crate::RequestExtensions::default(), deadline)
+        .await
+        .expect("streaming callout should open")
+    {
+        crate::CalloutResponse::Streaming { body, .. } => body,
+        crate::CalloutResponse::Buffered(_) => panic!("the chain selected streaming mode"),
+    };
+
+    // Pull until the ceiling rejects a chunk, whatever framing the upstream used.
+    let mut breach = None;
+    for _ in 0_u8..32 {
+        match body.next_chunk().await {
+            Ok(Some(_)) => {},
+            Ok(None) => break,
+            Err(error) => {
+                breach = Some(error);
+                break;
+            },
+        }
+    }
+    let breach = breach.expect("20 bytes of body must breach a 15-byte ceiling");
+    assert!(
+        breach.to_string().contains("exceeds configured body limit"),
+        "the breach must be the body-limit error: {breach}"
+    );
+
+    let resumed = body.next_chunk().await;
+    backend.abort();
+
+    let observed = match &resumed {
+        Ok(Some(chunk)) => format!("resumed with {} more bytes", chunk.len()),
+        Ok(None) => "end of stream".to_owned(),
+        Err(error) => format!("error: {error}"),
+    };
+    assert!(
+        matches!(resumed, Ok(None)),
+        "a ceiling breach must end the stream; polling again must not resume it, got {observed}"
+    );
+}
+
 #[tokio::test]
 #[expect(clippy::large_futures, reason = "drives the full executor future in a test")]
 async fn run_streaming_surfaces_unhandled_upstream_termination() {

[twghu]

A response-ceiling breach in CalloutStreamingBody::checked doesn't end the stream, so polling again resumes around the rejected chunk and hands the caller a body with a hole in it instead of an error — fix and failing-test evidence in the previous comment.

[twghu]

Once the fix is considered this PR can be approved.

[leseb]

A response-ceiling breach in CalloutStreamingBody::checked doesn't end the stream, so polling again resumes around the rejected chunk and hands the caller a body with a hole in it instead of an error — fix and failing-test evidence in the previous comment.

Addressed in 5013eb9 - PTAL :)