plecto_control/
chain.rs

1//! The chain dispatcher (ADR 000007): drive a request through the loaded filters in order,
2//! honouring `decision` and applying `modified` edits; drive the response back in reverse.
3//! Fail-closed — a trapped / timed-out filter (`RunError`) never falls through to upstream.
4
5use std::sync::Arc;
6
7use plecto_host::{
8    Header, HttpRequest, HttpResponse, LoadedFilter, RequestBodyDecision, RequestDecision,
9    RequestEdit, RequestTrace, ResponseDecision, ResponseEdit,
10};
11
12/// The result of driving a request through the chain.
13pub enum ChainOutcome {
14    /// Respond now without reaching upstream: a filter short-circuited, or the chain failed
15    /// closed on a trap / deadline (the synthetic 5xx from `RunError::fail_closed_response`).
16    Respond(HttpResponse),
17    /// The chain passed: forward this (possibly edited) request upstream.
18    Forward(HttpRequest),
19}
20
21/// The result of driving a buffered request body through the chain's `on-request-body` hooks
22/// (ADR 000025).
23pub enum RequestBodyOutcome {
24    /// Respond now without reaching upstream: a body filter short-circuited, or the chain failed
25    /// closed on a trap / deadline.
26    Respond(HttpResponse),
27    /// The chain passed: forward this (possibly transformed) body upstream.
28    Forward(Vec<u8>),
29}
30
31/// The result of driving a response back through the chain (ADR 000073): the typed successor
32/// of the old in-band "non-empty body means synthetic" signal.
33pub enum ResponseOutcome {
34    /// The chain passed: send the (possibly edited) status + headers and stream the upstream
35    /// body through (its `body` is empty — header-only, ADR 000038).
36    Forward(HttpResponse),
37    /// Send this SYNTHESISED response instead of the upstream one (a filter's `replace`, or
38    /// the chain failed closed on a trap / deadline) — the upstream stream is dropped unread.
39    Respond(HttpResponse),
40}
41
42pub(crate) fn dispatch_request(
43    chain: &[Arc<LoadedFilter>],
44    mut request: HttpRequest,
45    trace: &RequestTrace,
46) -> ChainOutcome {
47    for filter in chain {
48        // `trace` parents each filter span (ADR 000009): the host times the call and emits the
49        // span to its sink; we drive only the decision here.
50        match filter.on_request(&request, trace) {
51            Ok((RequestDecision::Continue, _logs)) => {}
52            Ok((RequestDecision::Modified(edit), _logs)) => apply_request_edit(&mut request, edit),
53            Ok((RequestDecision::ShortCircuit(response), _logs)) => {
54                return ChainOutcome::Respond(response);
55            }
56            // fail-closed: a trapped / timed-out filter must not reach upstream.
57            Err(err) => return ChainOutcome::Respond(err.fail_closed_response()),
58        }
59    }
60    ChainOutcome::Forward(request)
61}
62
63/// Drive a buffered request body through the chain's `on-request-body` hooks in order (ADR 000025),
64/// threading the possibly-transformed body from one filter to the next. A short-circuit (or a
65/// fail-closed trap / deadline) stops the chain before upstream. The host already buffered the body;
66/// this only sequences the decisions.
67pub(crate) fn dispatch_request_body(
68    chain: &[Arc<LoadedFilter>],
69    mut body: Vec<u8>,
70    trace: &RequestTrace,
71) -> RequestBodyOutcome {
72    for filter in chain {
73        match filter.on_request_body(&body, trace) {
74            Ok((RequestBodyDecision::Continue(next), _logs)) => body = next,
75            Ok((RequestBodyDecision::ShortCircuit(response), _logs)) => {
76                return RequestBodyOutcome::Respond(response);
77            }
78            // fail-closed: a trapped / timed-out body filter must not reach upstream.
79            Err(err) => return RequestBodyOutcome::Respond(err.fail_closed_response()),
80        }
81    }
82    RequestBodyOutcome::Forward(body)
83}
84
85pub(crate) fn dispatch_response(
86    chain: &[Arc<LoadedFilter>],
87    request: &HttpRequest,
88    mut response: HttpResponse,
89    trace: &RequestTrace,
90) -> ResponseOutcome {
91    // The response side runs the chain in reverse (CONTEXT: request/response are symmetric).
92    // `request` is the as-forwarded snapshot every hook sees (ADR 000073). A `replace` stops
93    // the chain and answers with the synthesised response — same terminal shape as the request
94    // side's short-circuit and the fail-closed arm (and the general proxy-filter form: a local
95    // response skips the remaining filters). The same `trace` as the request side, so request +
96    // response spans share one trace (ADR 000009).
97    for filter in chain.iter().rev() {
98        match filter.on_response(request, &response, trace) {
99            Ok((ResponseDecision::Continue, _logs)) => {}
100            Ok((ResponseDecision::Modified(edit), _logs)) => {
101                apply_response_edit(&mut response, edit)
102            }
103            Ok((ResponseDecision::Replace(replacement), _logs)) => {
104                return ResponseOutcome::Respond(replacement);
105            }
106            Err(err) => return ResponseOutcome::Respond(err.fail_closed_response()),
107        }
108    }
109    ResponseOutcome::Forward(response)
110}
111
112/// Apply a request rewrite: remove the named headers, then set (replace-or-add) the given
113/// ones. Header names match case-insensitively (HTTP semantics).
114fn apply_request_edit(request: &mut HttpRequest, edit: RequestEdit) {
115    remove_headers(&mut request.headers, &edit.remove_headers);
116    set_headers(&mut request.headers, edit.set_headers);
117}
118
119fn apply_response_edit(response: &mut HttpResponse, edit: ResponseEdit) {
120    if let Some(status) = edit.set_status {
121        response.status = status;
122    }
123    remove_headers(&mut response.headers, &edit.remove_headers);
124    set_headers(&mut response.headers, edit.set_headers);
125}
126
127fn remove_headers(headers: &mut Vec<Header>, names: &[String]) {
128    for name in names {
129        headers.retain(|h| !h.name.eq_ignore_ascii_case(name));
130    }
131}
132
133fn set_headers(headers: &mut Vec<Header>, set: Vec<Header>) {
134    for header in set {
135        // Repeatable headers (Set-Cookie) APPEND: each field line is independent, so collapsing
136        // them on `set` would drop all but the last cookie (review f000005 P3#7). Every other
137        // header REPLACES — the semantics a filter relies on to AUTHORITATIVELY overwrite a value
138        // (e.g. a spoofed `x-authenticated-user`), so this carve-out is for repeatables only.
139        if !is_repeatable(&header.name) {
140            headers.retain(|existing| !existing.name.eq_ignore_ascii_case(&header.name));
141        }
142        headers.push(header);
143    }
144}
145
146/// Headers that may legitimately appear multiple times and must not be collapsed by `set`.
147/// `Set-Cookie` is the canonical case (RFC 6265: one cookie per field line).
148fn is_repeatable(name: &str) -> bool {
149    name.eq_ignore_ascii_case("set-cookie")
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    fn h(name: &str, value: &str) -> Header {
157        Header {
158            name: name.to_string(),
159            value: value.as_bytes().to_vec(),
160        }
161    }
162
163    #[test]
164    fn request_edit_sets_replaces_and_removes_case_insensitively() {
165        let mut request = HttpRequest {
166            method: "GET".to_string(),
167            path: "/".to_string(),
168            authority: "a".to_string(),
169            scheme: "https".to_string(),
170            headers: vec![h("X-Keep", "1"), h("X-Drop", "old"), h("X-Replace", "old")],
171        };
172        apply_request_edit(
173            &mut request,
174            RequestEdit {
175                set_headers: vec![h("x-replace", "new"), h("X-Add", "1")],
176                remove_headers: vec!["x-drop".to_string()],
177            },
178        );
179
180        assert!(
181            !request
182                .headers
183                .iter()
184                .any(|x| x.name.eq_ignore_ascii_case("x-drop")),
185            "removed header gone"
186        );
187        let replaced: Vec<_> = request
188            .headers
189            .iter()
190            .filter(|x| x.name.eq_ignore_ascii_case("x-replace"))
191            .collect();
192        assert_eq!(replaced.len(), 1, "set replaces, not duplicates");
193        assert_eq!(replaced[0].value.as_slice(), b"new");
194        assert!(
195            request
196                .headers
197                .iter()
198                .any(|x| x.name.eq_ignore_ascii_case("x-add"))
199        );
200        assert!(request.headers.iter().any(|x| x.name == "X-Keep"));
201    }
202
203    #[test]
204    fn response_edit_sets_status_and_headers() {
205        let mut response = HttpResponse {
206            status: 200,
207            headers: vec![h("X-Old", "1")],
208            body: vec![],
209        };
210        apply_response_edit(
211            &mut response,
212            ResponseEdit {
213                set_status: Some(503),
214                set_headers: vec![h("X-New", "1")],
215                remove_headers: vec!["x-old".to_string()],
216            },
217        );
218
219        assert_eq!(response.status, 503);
220        assert!(response.headers.iter().any(|x| x.name == "X-New"));
221        assert!(
222            !response
223                .headers
224                .iter()
225                .any(|x| x.name.eq_ignore_ascii_case("x-old"))
226        );
227    }
228
229    #[test]
230    fn response_edit_appends_repeatable_set_cookie_but_replaces_others() {
231        // review f000005 P3#7: two Set-Cookie in one edit must BOTH survive (append), not collapse
232        // to the last — otherwise a filter setting a session + a flag cookie silently loses one.
233        // A non-repeatable header still replaces (the `set replaces` contract is unchanged for it).
234        let mut response = HttpResponse {
235            status: 200,
236            headers: vec![h("x-keep", "1")],
237            body: vec![],
238        };
239        apply_response_edit(
240            &mut response,
241            ResponseEdit {
242                set_status: None,
243                set_headers: vec![
244                    h("set-cookie", "session=abc"),
245                    h("set-cookie", "flag=1"),
246                    h("x-keep", "2"),
247                ],
248                remove_headers: vec![],
249            },
250        );
251
252        let cookies: Vec<&str> = response
253            .headers
254            .iter()
255            .filter(|x| x.name.eq_ignore_ascii_case("set-cookie"))
256            .map(|x| std::str::from_utf8(&x.value).expect("utf-8"))
257            .collect();
258        assert_eq!(
259            cookies,
260            vec!["session=abc", "flag=1"],
261            "both Set-Cookie field lines must survive (append, not collapse)"
262        );
263        let keep: Vec<&Header> = response
264            .headers
265            .iter()
266            .filter(|x| x.name.eq_ignore_ascii_case("x-keep"))
267            .collect();
268        assert_eq!(keep.len(), 1, "a non-repeatable header still replaces");
269        assert_eq!(keep[0].value.as_slice(), b"2");
270    }
271}
plecto_control/chain.rs

plecto_control/
chain.rs
