Skip to main content

sozu_lib/protocol/kawa_h1/
editor.rs

1//! H1 request/response header editor.
2//!
3//! Captures method/authority/path on parse, rewrites hop-by-hop and
4//! forwarding headers (`X-Forwarded-*`, `Forwarded`, `Connection`,
5//! WebSocket-upgrade signalling, optional `traceparent`), and surfaces the
6//! canonical `LogContext` used by the access-log envelope. Acts as the
7//! Kawa `ParserCallbacks` implementation for the H1 mux path.
8
9use std::{
10    borrow::Cow,
11    io::Write as _,
12    net::{IpAddr, SocketAddr},
13    rc::Rc,
14    str::from_utf8,
15    sync::Arc,
16};
17
18use rusty_ulid::Ulid;
19use sozu_command_lib::{
20    logging::{CachedTags, LogContext},
21    proto::command::ForwardedHeaders,
22};
23
24use crate::metrics::names;
25use crate::{
26    Protocol, RetrieveClusterError,
27    pool::Checkout,
28    protocol::{
29        http::{
30            GenericHttpStream,
31            parser::{Method, compare_no_case},
32        },
33        pipe::WebSocketContext,
34    },
35};
36
37#[cfg(feature = "opentelemetry")]
38fn parse_traceparent(val: &kawa::Store, buf: &[u8]) -> Option<([u8; 32], [u8; 16])> {
39    let val = val.data(buf);
40    let (version, val) = parse_hex::<2>(val)?;
41    if version.as_slice() != b"00" {
42        return None;
43    }
44    let val = skip_separator(val)?;
45    let (trace_id, val) = parse_hex::<32>(val)?;
46    let val = skip_separator(val)?;
47    let (parent_id, val) = parse_hex::<16>(val)?;
48    let val = skip_separator(val)?;
49    let (_, val) = parse_hex::<2>(val)?;
50    val.is_empty().then_some((trace_id, parent_id))
51}
52
53#[cfg(feature = "opentelemetry")]
54fn parse_hex<const N: usize>(buf: &[u8]) -> Option<([u8; N], &[u8])> {
55    let val: [u8; N] = buf.get(..N)?.try_into().unwrap();
56    val.iter()
57        .all(|c| c.is_ascii_hexdigit())
58        .then_some((val, &buf[N..]))
59}
60
61#[cfg(feature = "opentelemetry")]
62fn skip_separator(buf: &[u8]) -> Option<&[u8]> {
63    buf.first().filter(|b| **b == b'-').map(|_| &buf[1..])
64}
65
66#[cfg(feature = "opentelemetry")]
67fn random_id<const N: usize>() -> [u8; N] {
68    use rand::RngExt;
69    const CHARSET: &[u8] = b"0123456789abcdef";
70    let mut rng = rand::rng();
71    let mut buf = [0; N];
72    buf.fill_with(|| {
73        let n = rng.random_range(0..CHARSET.len());
74        CHARSET[n]
75    });
76    buf
77}
78
79#[cfg(feature = "opentelemetry")]
80fn build_traceparent(trace_id: &[u8; 32], parent_id: &[u8; 16]) -> [u8; 55] {
81    // Pre: the ids are hex (they come from a parsed traceparent or our own
82    // hex `random_id`). Hex digits are inherently CR/LF-free, so this also
83    // upholds the anti-injection guarantee for the value we splice into the
84    // `traceparent` header. Inline (not via `is_crlf_free`) so the release
85    // build under `--features opentelemetry` still compiles (HARD RULE 2).
86    debug_assert!(
87        trace_id.iter().all(u8::is_ascii_hexdigit) && parent_id.iter().all(u8::is_ascii_hexdigit),
88        "traceparent ids must be hex (CR/LF-free header value)"
89    );
90    let mut buf = [0; 55];
91    buf[..3].copy_from_slice(b"00-");
92    buf[3..35].copy_from_slice(trace_id);
93    buf[35] = b'-';
94    buf[36..52].copy_from_slice(parent_id);
95    buf[52..55].copy_from_slice(b"-01");
96    // Post: the value is exactly the `00-<trace>-<parent>-01` shape with the
97    // dash separators at the fixed offsets the parser expects.
98    debug_assert!(
99        buf[2] == b'-' && buf[35] == b'-' && buf[52] == b'-',
100        "traceparent must carry dash separators at fixed offsets"
101    );
102    buf
103}
104
105/// Whether a `Connection` field value lists `option` (RFC 9110 §7.6.1): a
106/// comma-separated list of case-insensitive tokens with optional whitespace.
107pub(crate) fn has_connection_option(value: &[u8], option: &[u8]) -> bool {
108    value
109        .split(|byte| *byte == b',')
110        .any(|listed| compare_no_case(listed.trim_ascii(), option))
111}
112
113/// Announce that the connection closes after `message` (RFC 9112 §9.6)
114/// without dropping the options it already lists: removing one would turn
115/// the field it nominates as hop-by-hop into an end-to-end one (RFC 9110
116/// §7.6.1). Every `Connection` line is elided and their options merged into
117/// one pushed line ending in a single `close`; only the `keep-alive` the
118/// close contradicts, and a `close` already listed, are dropped. With
119/// `refuse_upgrade`, the `upgrade` option goes too, with every `Upgrade`
120/// field it nominates, so the message requests no protocol switch.
121fn merge_connection_options_into_close(message: &mut GenericHttpStream, refuse_upgrade: bool) {
122    let buf = message.storage.buffer();
123    let mut options = Vec::new();
124    for block in &mut message.blocks {
125        if let kawa::Block::Header(header) = block
126            && !header.is_elided()
127        {
128            let key = header.key.data(buf);
129            if compare_no_case(key, b"connection") {
130                for option in header.val.data(buf).split(|byte| *byte == b',') {
131                    let option = option.trim_ascii();
132                    let dropped = option.is_empty()
133                        || compare_no_case(option, b"keep-alive")
134                        || compare_no_case(option, b"close")
135                        || (refuse_upgrade && compare_no_case(option, b"upgrade"));
136                    if !dropped {
137                        options.extend_from_slice(option);
138                        options.extend_from_slice(b", ");
139                    }
140                }
141                header.elide();
142            } else if refuse_upgrade && compare_no_case(key, b"upgrade") {
143                header.elide();
144            }
145        }
146    }
147    // A lone `close` needs no allocation: the common shutdown case of a
148    // message that listed no other option.
149    let val = if options.is_empty() {
150        kawa::Store::Static(b"close")
151    } else {
152        options.extend_from_slice(b"close");
153        debug_assert!(
154            is_crlf_free(&options),
155            "the merged Connection options must be CR/LF-free (anti-injection)"
156        );
157        kawa::Store::from_vec(options)
158    };
159    message.push_block(kawa::Block::Header(kawa::Pair {
160        key: kawa::Store::Static(b"Connection"),
161        val,
162    }));
163}
164
165/// `true` when `bytes` contains no CR or LF — the anti-injection
166/// invariant for any header value Sōzu serialises onto the wire. A value
167/// carrying a raw CR/LF could split one header into two (request/response
168/// smuggling, CWE-93/CWE-113).
169///
170/// Compiled in EVERY profile (HARD RULE 2): it is referenced inside plain
171/// `debug_assert!` calls whose arguments still compile with
172/// `debug_assertions` OFF, so a `#[cfg(debug_assertions)]` gate would break
173/// the release build with E0425. In release it is only reached from
174/// optimised-out `debug_assert!` bodies, so the optimiser drops it — hence
175/// `#[allow(dead_code)]` to keep `-D warnings` clean.
176#[allow(dead_code)]
177fn is_crlf_free(bytes: &[u8]) -> bool {
178    !bytes.iter().any(|b| *b == b'\r' || *b == b'\n')
179}
180
181/// Write the ";for=..;by=.." portion of a Forwarded header into `buf`
182/// without heap-allocating a `String`.
183///
184/// RFC 7239 §6 requires IPv6 literals to be enclosed in square brackets
185/// inside a quoted string (the `IP-literal` production is `"[" IPv6address "]"`).
186/// Emitting a bare IPv6 like `for="2001:db8::1:8080"` is ambiguous: the
187/// trailing `:1:8080` could parse as `::1` + port `8080` or as the literal
188/// `:1:8080` with no port. Matches HAProxy's behaviour
189/// (`_7239_print_ip6` in `src/http_ext.c`), which is the de-facto reference
190/// implementation of RFC 7239 in the proxy world. See sozu issue #1254.
191fn write_forwarded_for_by(buf: &mut Vec<u8>, peer_ip: IpAddr, peer_port: u16, public_ip: IpAddr) {
192    let before = buf.len();
193    buf.extend_from_slice(b";for=\"");
194    write_ip_literal(buf, peer_ip);
195    buf.push(b':');
196    let mut port_buf = itoa::Buffer::new();
197    buf.extend_from_slice(port_buf.format(peer_port).as_bytes());
198    buf.extend_from_slice(b"\";by=");
199    match public_ip {
200        IpAddr::V4(_) => {
201            let _ = write!(buf, "{public_ip}");
202        }
203        IpAddr::V6(_) => {
204            buf.push(b'"');
205            write_ip_literal(buf, public_ip);
206            buf.push(b'"');
207        }
208    }
209    // Post: the fragment grew the buffer and the whole appended span is
210    // CR/LF-free — it is spliced verbatim into a `Forwarded` header value,
211    // so a stray CR/LF would let an attacker-influenced peer address split
212    // the header (CWE-93). `peer_ip`/`public_ip` are `IpAddr` and the port
213    // is a `u16`, so no untrusted bytes reach here, but the guard pins the
214    // anti-injection contract against future edits to this fragment.
215    debug_assert!(
216        buf.len() > before,
217        "the Forwarded ;for=;by= fragment must emit bytes"
218    );
219    debug_assert!(
220        is_crlf_free(&buf[before..]),
221        "the Forwarded fragment must be CR/LF-free (anti-injection)"
222    );
223}
224
225/// Write an `IP-literal` per RFC 3986 §3.2.2 / RFC 7239 §6: IPv4 verbatim,
226/// IPv6 wrapped in `[...]`. Caller is responsible for any surrounding quotes
227/// required by the embedding syntax (RFC 7239 mandates them for IPv6).
228fn write_ip_literal(buf: &mut Vec<u8>, ip: IpAddr) {
229    let before = buf.len();
230    match ip {
231        IpAddr::V4(_) => {
232            let _ = write!(buf, "{ip}");
233        }
234        IpAddr::V6(_) => {
235            buf.push(b'[');
236            let _ = write!(buf, "{ip}");
237            buf.push(b']');
238        }
239    }
240    // Post: a non-empty literal was appended (every IpAddr renders to at
241    // least one byte) and the emitted bytes carry no CR/LF — an IpAddr can
242    // only render `[0-9a-fA-F:.]` plus the brackets we add, so this is a
243    // belt-and-suspenders guard on the value we splice into a header.
244    debug_assert!(buf.len() > before, "write_ip_literal must emit bytes");
245    debug_assert!(
246        is_crlf_free(&buf[before..]),
247        "an IP literal spliced into a header must be CR/LF-free"
248    );
249    // IPv6 is bracketed on both ends; IPv4 is bare (RFC 3986 §3.2.2).
250    debug_assert_eq!(
251        matches!(ip, IpAddr::V6(_)),
252        buf[before] == b'[' && buf[buf.len() - 1] == b']',
253        "IPv6 literals are bracketed, IPv4 literals are not"
254    );
255}
256
257/// Write ", proto=<proto>;for=..;by=.." (the suffix appended to an existing Forwarded value).
258fn write_forwarded_suffix(
259    buf: &mut Vec<u8>,
260    proto: &str,
261    peer_ip: IpAddr,
262    peer_port: u16,
263    public_ip: IpAddr,
264) {
265    // Pre: `proto` is the proxy-chosen scheme label ("http"/"https"), never
266    // attacker-controlled — assert it carries no CR/LF before it is spliced
267    // into the `Forwarded` value (anti-injection, CWE-93).
268    debug_assert!(
269        is_crlf_free(proto.as_bytes()),
270        "the proto label spliced into Forwarded must be CR/LF-free"
271    );
272    let before = buf.len();
273    buf.extend_from_slice(b", proto=");
274    buf.extend_from_slice(proto.as_bytes());
275    write_forwarded_for_by(buf, peer_ip, peer_port, public_ip);
276    // Post: the suffix grew the buffer, begins with the `, proto=`
277    // separator (so it appends cleanly to an existing value), and the whole
278    // appended span is CR/LF-free.
279    debug_assert!(
280        buf.len() > before + b", proto=".len(),
281        "the Forwarded suffix must emit the separator plus a value"
282    );
283    debug_assert!(
284        buf[before..].starts_with(b", proto="),
285        "the Forwarded suffix must start with the `, proto=` separator"
286    );
287    debug_assert!(
288        is_crlf_free(&buf[before..]),
289        "the Forwarded suffix must be CR/LF-free (anti-injection)"
290    );
291}
292
293/// The separator a hop is appended to a client-supplied chain with. A
294/// rendered hop starts with it, and a synthesised header shares the same
295/// rendering from just past it (`kawa::Store::Shared`'s start offset).
296const HOP_SEPARATOR: &[u8] = b", ";
297
298/// Whether a listener in `mode` adds the `X-Forwarded-*` family
299/// (`X-Forwarded-For`, `-Proto`, `-Port`, and `-Host` on a host rewrite).
300pub(crate) fn emits_x_forwarded(mode: ForwardedHeaders) -> bool {
301    matches!(mode, ForwardedHeaders::Both | ForwardedHeaders::XForwarded)
302}
303
304/// Whether a listener in `mode` adds an RFC 7239 `Forwarded` element.
305pub(crate) fn emits_rfc7239(mode: ForwardedHeaders) -> bool {
306    matches!(mode, ForwardedHeaders::Both | ForwardedHeaders::Rfc7239)
307}
308
309/// Whether a listener in `mode` removes a client-supplied
310/// `X-Forwarded-For`, `-Proto`, `-Port` or `-Host`: only `rfc7239` does, so
311/// that the backend sees one forwarding family, whose last element Sōzu
312/// wrote (RFC 7239 §8.1: the header fields a client sends cannot be
313/// trusted).
314pub(crate) fn strips_x_forwarded(mode: ForwardedHeaders) -> bool {
315    mode == ForwardedHeaders::Rfc7239
316}
317
318/// Whether `key` names one of the `X-Forwarded-*` headers the editor
319/// manages: the ones `strips_x_forwarded` removes.
320fn is_managed_x_forwarded(key: &[u8]) -> bool {
321    compare_no_case(key, b"X-Forwarded-For")
322        || compare_no_case(key, b"X-Forwarded-Proto")
323        || compare_no_case(key, b"X-Forwarded-Port")
324        || compare_no_case(key, b"X-Forwarded-Host")
325}
326
327/// Whether `byte` is an RFC 9110 §5.6.2 `tchar`.
328fn is_tchar(byte: u8) -> bool {
329    byte.is_ascii_alphanumeric()
330        || matches!(
331            byte,
332            b'!' | b'#'
333                | b'$'
334                | b'%'
335                | b'&'
336                | b'\''
337                | b'*'
338                | b'+'
339                | b'-'
340                | b'.'
341                | b'^'
342                | b'_'
343                | b'`'
344                | b'|'
345                | b'~'
346        )
347}
348
349/// Whether `value` is a well-formed RFC 7239 §4 `Forwarded` field value:
350///
351/// ```text
352/// Forwarded         = 1#forwarded-element
353/// forwarded-element = [ forwarded-pair ] *( ";" [ forwarded-pair ] )
354/// forwarded-pair    = token "=" value
355/// value             = token / quoted-string
356/// ```
357///
358/// with `token` and `quoted-string` from RFC 9110 §5.6.2 / §5.6.4. As RFC
359/// 9110 §5.6.1 asks of a list recipient, empty list elements and optional
360/// whitespace around `,` are accepted; whitespace is also tolerated around
361/// `;`. An empty value is an empty list. What matters is that a parser reads
362/// the value as a sequence of complete elements, so the element Sōzu appends
363/// after a `, ` is read as one of its own: an unclosed quoted-string would
364/// swallow it.
365fn is_valid_forwarded(value: &[u8]) -> bool {
366    let skip_ows = |i: &mut usize| {
367        while *i < value.len() && matches!(value[*i], b' ' | b'\t') {
368            *i += 1;
369        }
370    };
371    let mut i = 0;
372    loop {
373        // One forwarded-element: optional pairs separated by `;`.
374        loop {
375            skip_ows(&mut i);
376            if i < value.len() && is_tchar(value[i]) {
377                // token "="
378                while i < value.len() && is_tchar(value[i]) {
379                    i += 1;
380                }
381                if value.get(i) != Some(&b'=') {
382                    return false;
383                }
384                i += 1;
385                // value = token / quoted-string
386                match value.get(i) {
387                    Some(b'"') => {
388                        i += 1;
389                        loop {
390                            match value.get(i) {
391                                None => return false,
392                                Some(b'"') => {
393                                    i += 1;
394                                    break;
395                                }
396                                Some(b'\\') => match value.get(i + 1) {
397                                    Some(&escaped)
398                                        if escaped == b'\t'
399                                            || escaped == b' '
400                                            || (0x21..=0x7e).contains(&escaped)
401                                            || escaped >= 0x80 =>
402                                    {
403                                        i += 2
404                                    }
405                                    _ => return false,
406                                },
407                                Some(&byte)
408                                    if byte == b'\t'
409                                        || byte == b' '
410                                        || byte == 0x21
411                                        || (0x23..=0x7e).contains(&byte)
412                                        || byte >= 0x80 =>
413                                {
414                                    i += 1
415                                }
416                                Some(_) => return false,
417                            }
418                        }
419                    }
420                    Some(&byte) if is_tchar(byte) => {
421                        while i < value.len() && is_tchar(value[i]) {
422                            i += 1;
423                        }
424                    }
425                    _ => return false,
426                }
427            }
428            skip_ows(&mut i);
429            if value.get(i) == Some(&b';') {
430                i += 1;
431            } else {
432                break;
433            }
434        }
435        match value.get(i) {
436            None => return true,
437            Some(b',') => i += 1,
438            Some(_) => return false,
439        }
440    }
441}
442
443/// The forwarding values of one connection, rendered once from the inputs
444/// they depend on and shared by every request that reuses those inputs.
445///
446/// Neither value depends on the request: the peer and public addresses and
447/// the protocol are connection-scoped, so the second and later requests of
448/// a keep-alive connection forward them without a heap operation
449/// (`kawa::Store::Shared` bumps a reference count). A client-supplied
450/// `X-Forwarded-For` or `Forwarded` chain is request-scoped and is never
451/// kept here: the rendered hop is appended to a copy of it.
452/// `HttpContext::forwarding_hop` renders the values again whenever an input
453/// differs from the ones they were rendered from.
454#[derive(Debug)]
455pub(crate) struct ForwardingHop {
456    protocol: Protocol,
457    public_address: SocketAddr,
458    session_address: Option<SocketAddr>,
459    /// The `X-Forwarded-Port` value: the public port.
460    port: Rc<[u8]>,
461    /// `, <peer ip>`: the hop appended to a client `X-Forwarded-For`; past
462    /// `HOP_SEPARATOR`, the value of a synthesised `X-Forwarded-For` and
463    /// of `X-Real-IP`. `None` without a peer address.
464    x_forwarded_for: Option<Rc<[u8]>>,
465    /// `, proto=<proto>;for="<peer>";by=<public>`: the element appended to a
466    /// client `Forwarded`; past `HOP_SEPARATOR`, the value of a
467    /// synthesised `Forwarded`. `None` without a peer address.
468    forwarded: Option<Rc<[u8]>>,
469}
470
471impl ForwardingHop {
472    /// Render the forwarding values of `protocol`, labelled `proto`, for
473    /// the given public and peer addresses.
474    fn render(
475        protocol: Protocol,
476        proto: &str,
477        public_address: SocketAddr,
478        session_address: Option<SocketAddr>,
479    ) -> Self {
480        let mut port_buf = itoa::Buffer::new();
481        let port = Rc::from(port_buf.format(public_address.port()).as_bytes());
482        let (x_forwarded_for, forwarded) = match session_address {
483            Some(peer) => {
484                // One scratch renders both values, each copied out once at
485                // its exact size.
486                let mut scratch = Vec::with_capacity(128);
487                scratch.extend_from_slice(HOP_SEPARATOR);
488                let _ = write!(scratch, "{}", peer.ip());
489                let x_forwarded_for = Rc::<[u8]>::from(&scratch[..]);
490                scratch.clear();
491                write_forwarded_suffix(
492                    &mut scratch,
493                    proto,
494                    peer.ip(),
495                    peer.port(),
496                    public_address.ip(),
497                );
498                (Some(x_forwarded_for), Some(Rc::<[u8]>::from(&scratch[..])))
499            }
500            None => (None, None),
501        };
502        let hop = Self {
503            protocol,
504            public_address,
505            session_address,
506            port,
507            x_forwarded_for,
508            forwarded,
509        };
510        // Post: both peer values exist exactly when there is a peer, each
511        // starts with the separator and carries a value past it, and every
512        // rendered byte is CR/LF-free — they are spliced verbatim into
513        // header values (anti-injection, CWE-93).
514        debug_assert!(
515            hop.x_forwarded_for.is_some() == session_address.is_some()
516                && hop.forwarded.is_some() == session_address.is_some(),
517            "the peer values are rendered exactly when there is a peer"
518        );
519        debug_assert!(
520            [&hop.x_forwarded_for, &hop.forwarded]
521                .into_iter()
522                .flatten()
523                .all(|value| value.len() > HOP_SEPARATOR.len()
524                    && value.starts_with(HOP_SEPARATOR)
525                    && is_crlf_free(value)),
526            "a rendered hop is the separator followed by a CR/LF-free value"
527        );
528        debug_assert!(
529            !hop.port.is_empty() && hop.port.iter().all(u8::is_ascii_digit),
530            "the X-Forwarded-Port value is the rendered public port"
531        );
532        hop
533    }
534
535    /// Whether these values were rendered from exactly these inputs.
536    fn renders(
537        &self,
538        protocol: Protocol,
539        public_address: SocketAddr,
540        session_address: Option<SocketAddr>,
541    ) -> bool {
542        self.protocol == protocol
543            && self.public_address == public_address
544            && self.session_address == session_address
545    }
546}
547
548/// A synthesised header value: `hop` past its `HOP_SEPARATOR`, shared.
549fn synthesised_from_hop(hop: &Rc<[u8]>) -> kawa::Store {
550    debug_assert!(
551        hop.starts_with(HOP_SEPARATOR),
552        "a rendered hop starts with the separator"
553    );
554    kawa::Store::Shared(hop.clone(), HOP_SEPARATOR.len() as u32)
555}
556
557/// A client-supplied chain `client` extended with `hop`, in one allocation
558/// of the exact size of the result.
559fn extended_chain(client: &[u8], hop: &[u8]) -> kawa::Store {
560    let mut value = Vec::with_capacity(client.len() + hop.len());
561    value.extend_from_slice(client);
562    value.extend_from_slice(hop);
563    // Post: the client chain is kept as a prefix, our CR/LF-free hop is
564    // appended after it (anti-injection), and the buffer is full, so
565    // `kawa::Store::from_vec` keeps it without reallocating.
566    debug_assert!(
567        value.starts_with(client) && value.len() > client.len(),
568        "a client chain is only ever extended"
569    );
570    debug_assert!(
571        is_crlf_free(&value[client.len()..]) && value.len() == value.capacity(),
572        "the appended hop is CR/LF-free and the value exactly sized"
573    );
574    kawa::Store::from_vec(value)
575}
576
577/// This is the container used to store and use information about the session from within a Kawa parser callback
578#[derive(Debug)]
579pub struct HttpContext {
580    // ========== Write only
581    /// set to false if Kawa finds a "Connection" header with a "close" value in the response,
582    /// or if the response is a non-persistent HTTP/1.0 one (RFC 9112 §9.3)
583    pub keep_alive_backend: bool,
584    /// set to false if Kawa finds a "Connection" header with a "close" value in the request
585    pub keep_alive_frontend: bool,
586    /// the value of the sticky session cookie in the request
587    pub sticky_session_found: Option<String>,
588    // ---------- Status Line
589    /// the value of the method in the request line
590    pub method: Option<Method>,
591    /// the value of the authority of the request (in the request line of "Host" header)
592    pub authority: Option<String>,
593    /// the value of the path in the request line
594    pub path: Option<String>,
595    /// the value of the status code in the response line
596    pub status: Option<u16>,
597    /// the value of the reason in the response line: the `'static` phrase
598    /// RFC 9110 §15 registers for the status code when the backend sent
599    /// exactly that phrase, a copy otherwise (`standard_reason`)
600    pub reason: Option<Cow<'static, str>>,
601    // ---------- Additional optional data
602    pub user_agent: Option<String>,
603    /// Value of the `x-request-id` header observed (if propagated from the
604    /// client/upstream LB) or generated (from `self.id`). Universal correlation
605    /// header — populated unconditionally by `on_request_headers` so the access
606    /// log can record the exact value forwarded to the backend. `Rc` because
607    /// a generated value is the one rendering of `self.id` that the forwarded
608    /// `X-Request-Id` and both correlation headers share
609    /// (`kawa::Store::Shared`).
610    pub x_request_id: Option<Rc<str>>,
611    /// Verbatim value of the client-supplied `X-Forwarded-For` header as
612    /// observed before Sōzu appended its own hop. Captured here, not at
613    /// request edit time, so the access log records the upstream-attested
614    /// chain even when Sōzu also appends its own peer to the forwarded
615    /// header. `None` if the request had no `X-Forwarded-For` header.
616    pub xff_chain: Option<String>,
617
618    #[cfg(feature = "opentelemetry")]
619    pub otel: Option<sozu_command::logging::OpenTelemetry>,
620
621    // ========== Read only
622    /// signals wether Kawa should write a "Connection" header with a "close" value (request and response)
623    pub closing: bool,
624    /// Connection/session ULID — stable across all requests multiplexed on this
625    /// TCP or TLS connection. Used as the first slot in the legacy log-context
626    /// bracket `[session req cluster backend]` and emitted into
627    /// `ProtobufAccessLog.session_id`.
628    pub session_id: Ulid,
629    /// Request ULID: the value of the correlation header, named "Sozu-Id" by
630    /// default, that Kawa should write (request and response), of a generated
631    /// `X-Request-Id` and of `%REQUEST_ID`. Request-scoped: `reset` replaces
632    /// it for each keep-alive request, unlike [`Self::session_id`].
633    pub id: Ulid,
634    pub backend_id: Option<Rc<str>>,
635    pub cluster_id: Option<sozu_command_lib::state::ClusterId>,
636    /// Configured cluster lifetime that owns this request's labelled metrics.
637    /// Captured after routing so successive requests on one H1 keep-alive
638    /// connection may belong to successive lifetimes of the same cluster id.
639    pub(crate) cluster_metrics_incarnation: crate::metrics::ClusterMetricsCapture,
640    /// The client affinity key `Router::plan_connect`
641    /// (`lib/src/protocol/mux/router.rs`) derived for this request when its
642    /// cluster selects with `HRW` or `MAGLEV`: the hash of the cluster's
643    /// `affinity_header` / `affinity_cookie` value, else of the client source
644    /// IP. `None` under every other policy. Stored next to `cluster_id` for the
645    /// same reason: a replay finds the request's header blocks drained and
646    /// reuses the key the first attempt derived.
647    pub affinity_key: Option<u64>,
648    /// the value of the protocol Kawa should write in the Forwarded headers of the request
649    pub protocol: Protocol,
650    /// the value of the public address Kawa should write in the Forwarded headers of the request
651    pub public_address: SocketAddr,
652    /// the value of the session address Kawa should write in the Forwarded headers of the request
653    pub session_address: Option<SocketAddr>,
654    /// the name of the cookie Kawa should read from the request to get the sticky session
655    pub sticky_name: String,
656    /// the sticky session that should be used
657    /// used to create a "Set-Cookie" header in the response in case it differs from sticky_session_found
658    pub sticky_session: Option<String>,
659    /// the address of the backend server
660    pub backend_address: Option<SocketAddr>,
661    /// The TLS Server Name Indication (SNI) hostname negotiated at handshake.
662    ///
663    /// Populated for HTTPS listeners when the client sent an SNI extension (see
664    /// `https.rs::upgrade_handshake`). Used by the routing layer to enforce the
665    /// TLS trust boundary against the HTTP `:authority` / `Host` header — without
666    /// this check, an attacker holding a valid certificate for tenant A could
667    /// open TLS with SNI=A then send requests with `:authority=tenantB` and
668    /// reach tenant B's backend (CWE-346 / CWE-444).
669    ///
670    /// `None` when the listener is plaintext HTTP or the client omitted SNI.
671    /// Stored pre-lowercased and without a port for direct exact-match comparison.
672    pub tls_server_name: Option<String>,
673    /// Snapshot of the SAN set of the certificate Sōzu actually served at
674    /// the TLS handshake. Captured once in `https.rs::upgrade_handshake`
675    /// from the resolver and frozen for the connection lifetime so H2
676    /// stream coalescing (RFC 7540 §9.1.1 / RFC 9113 §9.1.1) accepts any
677    /// `:authority` covered by the certificate, with RFC 6125 §6.4.3
678    /// wildcard handling. `None` for plaintext listeners or when SNI was
679    /// absent (router falls back to the legacy exact-match predicate).
680    /// `Some(empty)` when the default cert was served — every
681    /// `:authority` is rejected. `Arc` so the snapshot is shared across
682    /// every per-stream `HttpContext` without re-allocation.
683    pub tls_cert_names: Option<Arc<Vec<String>>>,
684    /// Whether the router must reject this request when `tls_server_name`
685    /// does not exact-match its authority (CWE-346 / CWE-444). Mirrors
686    /// `HttpsListenerConfig::strict_sni_binding`. Set from the mux
687    /// `Context` at stream creation time (see `Context::create_stream`).
688    /// Plaintext listeners still never hit the check because
689    /// `tls_server_name` is `None`.
690    pub strict_sni_binding: bool,
691    /// When `true`, the request-side block walk in `on_request_headers`
692    /// strips any client-supplied `X-Real-IP` header before forwarding
693    /// (anti-spoofing). Mirrors `HttpListenerConfig::elide_x_real_ip` /
694    /// `HttpsListenerConfig::elide_x_real_ip`. Set from the mux `Context`
695    /// at stream creation (see `Context::create_stream`); listener-scoped
696    /// and never reset across keep-alive requests. Independent of
697    /// `send_x_real_ip`. The same flag is plumbed into
698    /// `pkawa::handle_trailer` so trailer HEADERS frames cannot bypass
699    /// the elision.
700    pub elide_x_real_ip: bool,
701    /// When `true`, `on_request_headers` injects a proxy-generated
702    /// `X-Real-IP` header carrying `session_address.ip()` (post-PROXY-v2
703    /// unwrap, i.e. the original client IP). Mirrors
704    /// `HttpListenerConfig::send_x_real_ip` /
705    /// `HttpsListenerConfig::send_x_real_ip`. Set from the mux `Context`
706    /// at stream creation (see `Context::create_stream`); listener-scoped
707    /// and never reset across keep-alive requests. Independent of
708    /// `elide_x_real_ip`. When `session_address` is `None` (raw socket
709    /// without a peer), no header is appended — identical to the
710    /// existing X-Forwarded-For / Forwarded synthesis behaviour.
711    pub send_x_real_ip: bool,
712    /// Which forwarding header family `on_request_headers` adds to the
713    /// request. Mirrors `HttpListenerConfig::forwarded_headers` /
714    /// `HttpsListenerConfig::forwarded_headers`. Set from the mux `Context`
715    /// at stream creation (see `Context::create_stream`); listener-scoped
716    /// and never reset across keep-alive requests. Independent of
717    /// `elide_x_real_ip` and `send_x_real_ip`: `X-Real-IP` belongs to
718    /// neither family. `Both`, the default, is the historical behaviour.
719    pub forwarded_headers: ForwardedHeaders,
720    /// Most trailer fields an H1 chunked request may carry, elided ones
721    /// included; one more makes `HttpContext::filter_request_trailers`
722    /// reject the request. Mirrors the listener's `h2_max_header_fields`,
723    /// the bound `pkawa::handle_trailer` (`lib/src/protocol/mux/pkawa.rs`)
724    /// applies to an H2 trailer block, so both frontends admit the same
725    /// number of trailer fields. Set from the mux `Context` at stream
726    /// creation (see `Context::create_stream`); listener-scoped and never
727    /// reset across keep-alive requests.
728    pub max_trailer_fields: u32,
729    /// Trailer fields of the current H1 request counted so far by
730    /// `HttpContext::filter_request_trailers`. Request-scoped: `reset`
731    /// zeroes it, so each pipelined request has its own budget.
732    pub trailer_fields: u32,
733    /// Negotiated TLS protocol version as a short label (e.g. `"TLSv1.3"`).
734    /// Captured from `rustls_version_label` at handshake completion and
735    /// propagated from the mux `Context`. `None` for plaintext listeners.
736    pub tls_version: Option<&'static str>,
737    /// Negotiated TLS cipher suite as a short label (e.g.
738    /// `"TLS_AES_128_GCM_SHA256"`). Captured from `rustls_ciphersuite_label`
739    /// at handshake completion and propagated from the mux `Context`. `None`
740    /// for plaintext listeners.
741    pub tls_cipher: Option<&'static str>,
742    /// Negotiated ALPN protocol (e.g. `"h2"`, `"http/1.1"`). Captured from
743    /// rustls at handshake completion and propagated from the mux `Context`.
744    /// `None` for plaintext listeners or when no ALPN was negotiated.
745    pub tls_alpn: Option<&'static str>,
746    /// Name of the correlation header Sozu injects into every request and
747    /// response. Defaults to `"Sozu-Id"` via [`crate::L7ListenerHandler::get_sozu_id_header`].
748    /// Populated at stream creation from the listener config's `sozu_id_header`
749    /// knob. Stored as an owned `String` so it survives a listener hot-reload
750    /// that changes the value.
751    pub sozu_id_header: String,
752    /// Resolved `Location` URL stashed by the routing layer when a frontend
753    /// triggers a permanent redirect (`RedirectPolicy::PERMANENT` or the
754    /// legacy `cluster.https_redirect`). Read by the default-answer 301
755    /// path so the response carries the correct target URL — including
756    /// optional `cluster.https_redirect_port` and rewrite-template captures.
757    /// `None` when the request is not redirecting.
758    pub redirect_location: Option<String>,
759    /// `WWW-Authenticate` realm stashed by the routing layer when a
760    /// frontend rejects an unauthenticated request (`required_auth = true`
761    /// without a valid `Authorization: Basic` header, or
762    /// `RedirectPolicy::UNAUTHORIZED`). Read by the default-answer 401
763    /// path so the response carries the cluster's configured
764    /// `www_authenticate` value. `None` falls back to template default
765    /// (header is elided when no realm is configured).
766    pub www_authenticate: Option<String>,
767    /// Original authority captured before a rewrite-host fired; emitted
768    /// back to the backend as `X-Forwarded-Host` so the backend can
769    /// reconstruct the public URL even though `:authority` / `Host` was
770    /// rewritten on the wire. `None` when no host rewrite happened.
771    pub original_authority: Option<String>,
772    /// Per-frontend response-side header edits (set/replace/delete)
773    /// stashed by the routing layer for the emission boundary in
774    /// `mux/h1.rs::writable` and `mux/h2.rs::write_streams` to apply
775    /// before `kawa.prepare(...)`. Empty when the frontend has no
776    /// response-side header policy. An entry with an empty `val`
777    /// deletes the header by name (HAProxy `del-header` parity); a
778    /// non-empty `val` set/replaces.
779    pub headers_response: Vec<HeaderEditSnapshot>,
780    /// Resolved `Retry-After` value (seconds) for an HTTP 429 default
781    /// answer. Computed in `Router::plan_connect` when the per-(cluster,
782    /// source-IP) connection limit is hit, by folding the cluster's
783    /// `retry_after` override over the global default. `None` (or
784    /// `Some(0)`) tells the answer engine to omit the `Retry-After`
785    /// header entirely — `Retry-After: 0` invites an immediate retry
786    /// that defeats the limit. Unused for any other status code.
787    pub retry_after_seconds: Option<u32>,
788    /// Frontend-supplied template body that overrides the listener /
789    /// cluster default `http.301.redirection` for a single
790    /// `RedirectPolicy::PERMANENT` request. Stashed by the routing layer
791    /// from `RouteResult::redirect_template` and consumed by the 301
792    /// branch of `mux::answers::set_default_answer_with_retry_after`,
793    /// which compiles the body via `HttpAnswers::render_inline_301` and
794    /// renders it with the same `(REDIRECT_LOCATION, ROUTE,
795    /// REQUEST_ID)` variable schema as the persistent template chain.
796    /// `None` falls back to the cluster / listener default. Unused for
797    /// any other status code or for the legacy
798    /// `cluster.https_redirect = true` path (which never sets it).
799    pub frontend_redirect_template: Option<String>,
800    /// Resolved redirect status code stashed by the routing layer when
801    /// a frontend's `RedirectPolicy` is one of the redirect variants.
802    /// 301 = `Permanent`, 302 = `Found`, 308 = `PermanentRedirect`.
803    /// `None` falls back to 301 for the legacy
804    /// `cluster.https_redirect = true` path. Closes #1009.
805    pub redirect_status: Option<u16>,
806    /// Access-log tags of the frontend rule that matched this request,
807    /// stashed by the routing layer from `RouteResult::tags` (see
808    /// `mux/router.rs::route_from_request`) and read back by
809    /// `mux/stream.rs::generate_access_log`.
810    ///
811    /// The matched frontend is the only correct owner of these tags: the
812    /// operator configures them per frontend RULE (`*.example.com`,
813    /// `/foo.*/.example.com`, a Pre/Post string), while the request only
814    /// ever carries a concrete authority. Resolving them here — rather
815    /// than by looking the authority up in the listener's
816    /// `BTreeMap<String, CachedTags>` — is what makes a wildcard, regex,
817    /// ported or differently-cased frontend log its tags at all
818    /// (sozu#1379). `Rc` because the same `CachedTags` is shared by every
819    /// stream routed to that frontend, and because it must outlive a
820    /// frontend removed mid-request.
821    ///
822    /// `None` until the request is routed, and for a request that matched
823    /// no frontend at all.
824    pub tags: Option<Rc<CachedTags>>,
825    /// The forwarding values rendered for this connection, reused by every
826    /// request whose inputs match (`HttpContext::forwarding_hop`).
827    /// Connection-scoped: `reset` keeps it. `None` until the first request
828    /// is edited.
829    pub(crate) forwarding_hop: Option<ForwardingHop>,
830    /// Stable, structured discriminator surfaced as the access-log
831    /// `message` field when the session terminates on a timeout. Set by
832    /// the `MuxState::timeout` handler
833    /// **before** the default-answer or `forcefully_terminate_answer`
834    /// path consumes it. The vocabulary is operator-visible API once
835    /// shipped — see the access-log section of `doc/configure.md` for
836    /// the full token list (`client_timeout`,
837    /// `client_timeout_during_response`, `backend_timeout`,
838    /// `backend_response_timeout`). `None` for non-timeout sessions, in
839    /// which case the access log emits `message: None` as before.
840    pub access_log_message: Option<&'static str>,
841    /// The routed cluster has backends, but none could be selected for this
842    /// request — every one failing its health check or backing off after
843    /// connection failures — so Sōzu answered 503 without contacting one.
844    /// That is a backend outage, not a routing miss: the H2 RST caps count
845    /// such a stream as routed to a backend (`routed_to_a_backend`,
846    /// `lib/src/protocol/mux/h2.rs`). Request-scoped: `reset` clears it.
847    pub backends_unavailable: bool,
848}
849
850/// How `apply_response_header_edits` should interpret a per-edit value.
851///
852/// The implicit empty-`val` Append → delete encoding is still supported
853/// (so legacy operator-supplied `[[...frontends.headers]]` entries work
854/// unchanged); the explicit modes give typed policies finer control:
855///
856/// - `Append`: drop the legacy delete shortcut for empty `val`, and
857///   append every other entry before the end-of-headers flag.
858/// - `SetIfAbsent`: skip the insert when `kawa.blocks` already carries
859///   a non-elided header with the same name (case-insensitive). HSTS
860///   uses this by default to preserve a backend-supplied
861///   `Strict-Transport-Security` (RFC 6797 §6.1 single-header
862///   requirement).
863/// - `Set`: delete every existing header with the matching name, then
864///   insert the new entry. Use when the operator wants their typed
865///   policy to override any backend-supplied value (the
866///   `force_replace_backend = true` HSTS shape, for example).
867#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
868pub enum HeaderEditMode {
869    /// Append the header before the end-of-headers flag. Empty `val`
870    /// is interpreted as a delete (legacy behaviour preserved).
871    #[default]
872    Append,
873    /// Skip the insert if `kawa.blocks` already contains a non-elided
874    /// header whose name matches `key` case-insensitively. Otherwise
875    /// behave like `Append`.
876    SetIfAbsent,
877    /// Delete every existing header with the matching name, then
878    /// insert the new value. Equivalent to two operator-defined edits
879    /// (delete + append) but safer to express as one typed entry.
880    Set,
881}
882
883/// Owned snapshot of a per-frontend header edit, captured at routing
884/// time so the emission boundary can apply set/replace/delete without
885/// touching the routing layer's `Rc<[HeaderEdit]>` slices.
886///
887/// `mode` chooses between explicit Append/Delete/SetIfAbsent semantics.
888/// For backwards compatibility `mode = Append` paired with an empty
889/// `val` is still treated as a delete by `apply_response_header_edits`
890/// (HAProxy `del-header` parity), so callers that have not yet migrated
891/// to the explicit `HeaderEditMode::Delete` keep working unchanged.
892#[derive(Debug, Clone)]
893pub struct HeaderEditSnapshot {
894    pub key: Vec<u8>,
895    pub val: Vec<u8>,
896    pub mode: HeaderEditMode,
897}
898
899/// A ULID in its canonical 26-character Crockford base-32 form, rendered on
900/// the stack: `Ulid`'s `Display` allocates a `String` to render it.
901fn render_ulid(id: Ulid) -> [u8; 26] {
902    const CROCKFORD: &[u8; 32] = b"0123456789ABCDEFGHJKMNPQRSTVWXYZ";
903    let value = u128::from(id);
904    let mut rendered = [0u8; 26];
905    // 26 digits of 5 bits carry 130 bits: the first digit holds the top 3
906    // bits of the 128-bit value, every later one the next 5.
907    for (index, digit) in rendered.iter_mut().enumerate() {
908        let shift = 5 * (25 - index);
909        *digit = CROCKFORD[((value >> shift) & 0x1f) as usize];
910    }
911    // Post: every digit is a Crockford base-32 character, so the rendering
912    // is ASCII and CR/LF-free wherever it is spliced into a header.
913    debug_assert!(
914        rendered.iter().all(|digit| CROCKFORD.contains(digit)),
915        "a rendered ULID must only carry Crockford base-32 digits"
916    );
917    rendered
918}
919
920/// The reason phrase RFC 9110 §15 registers for `code`, plus 429 from
921/// RFC 6585 §4, or `None` for a code it registers none for.
922///
923/// `on_response_headers` captures a backend's reason for the access log by
924/// reference to this phrase when the two are byte-identical, and copies it
925/// otherwise: an older name (`Payload Too Large`), a different case or a
926/// custom phrase is recorded verbatim, never normalised.
927fn standard_reason(code: u16) -> Option<&'static str> {
928    let phrase = match code {
929        100 => "Continue",
930        101 => "Switching Protocols",
931        200 => "OK",
932        201 => "Created",
933        202 => "Accepted",
934        203 => "Non-Authoritative Information",
935        204 => "No Content",
936        205 => "Reset Content",
937        206 => "Partial Content",
938        300 => "Multiple Choices",
939        301 => "Moved Permanently",
940        302 => "Found",
941        303 => "See Other",
942        304 => "Not Modified",
943        305 => "Use Proxy",
944        307 => "Temporary Redirect",
945        308 => "Permanent Redirect",
946        400 => "Bad Request",
947        401 => "Unauthorized",
948        402 => "Payment Required",
949        403 => "Forbidden",
950        404 => "Not Found",
951        405 => "Method Not Allowed",
952        406 => "Not Acceptable",
953        407 => "Proxy Authentication Required",
954        408 => "Request Timeout",
955        409 => "Conflict",
956        410 => "Gone",
957        411 => "Length Required",
958        412 => "Precondition Failed",
959        413 => "Content Too Large",
960        414 => "URI Too Long",
961        415 => "Unsupported Media Type",
962        416 => "Range Not Satisfiable",
963        417 => "Expectation Failed",
964        421 => "Misdirected Request",
965        422 => "Unprocessable Content",
966        426 => "Upgrade Required",
967        429 => "Too Many Requests",
968        500 => "Internal Server Error",
969        501 => "Not Implemented",
970        502 => "Bad Gateway",
971        503 => "Service Unavailable",
972        504 => "Gateway Timeout",
973        505 => "HTTP Version Not Supported",
974        _ => return None,
975    };
976    // Post: a registered phrase is non-empty and CR/LF-free; it is only ever
977    // compared with, and logged in place of, identical wire bytes.
978    debug_assert!(
979        !phrase.is_empty() && is_crlf_free(phrase.as_bytes()),
980        "a registered reason phrase is a non-empty single-line token"
981    );
982    Some(phrase)
983}
984
985/// Whether `kawa` carries a non-elided `Content-Length` whose value is not
986/// `1*DIGIT` (RFC 9110 §8.6).
987///
988/// kawa 0.7.1 read the value with `usize::from_str`, which also accepts one
989/// leading `+`, and left the field line in place with its original
990/// spelling (CleverCloud/kawa#25). kawa >= 0.7.2 checks `1*DIGIT` itself
991/// before the header callback (CleverCloud/kawa#26), so under the locked
992/// kawa this helper never finds a non-digit value on a parsed message: it
993/// is defense in depth against a kawa regression. Only non-elided lines are
994/// judged, because they are the ones the H1 serializer forwards: kawa elides
995/// a second line equal to the first, and every Content-Length beside a
996/// Transfer-Encoding.
997fn has_non_digit_content_length(kawa: &GenericHttpStream) -> bool {
998    let buf = kawa.storage.buffer();
999    kawa.blocks.iter().any(|block| match block {
1000        kawa::Block::Header(header)
1001            if !header.is_elided() && compare_no_case(header.key.data(buf), b"content-length") =>
1002        {
1003            let val = header.val.data(buf);
1004            val.is_empty() || !val.iter().all(u8::is_ascii_digit)
1005        }
1006        _ => false,
1007    })
1008}
1009
1010/// The store of a correlation-header name: the default `Sozu-Id` — what
1011/// `L7ListenerHandler::get_sozu_id_header` (`lib/src/lib.rs`) and its
1012/// `HttpListener` and `HttpsListener` implementations answer when the
1013/// listener does not rename it — is a `'static` literal and costs nothing; a
1014/// renamed header is copied.
1015fn correlation_header_name(name: &str) -> kawa::Store {
1016    const DEFAULT: &str = "Sozu-Id";
1017    if name == DEFAULT {
1018        kawa::Store::Static(DEFAULT.as_bytes())
1019    } else {
1020        kawa::Store::from_slice(name.as_bytes())
1021    }
1022}
1023
1024/// Header names a request trailer section may never carry past Sōzu, in
1025/// lower case. This is the single list shared by both frontends:
1026/// `pkawa::handle_trailer` (`lib/src/protocol/mux/pkawa.rs`) drops them from
1027/// an H2 trailer HEADERS frame, and [`HttpContext::filter_request_trailers`]
1028/// from an H1 chunked trailer section. Both also drop
1029/// [`TRAILER_FORBIDDEN_FIELDS`] from a request trailer section.
1030///
1031/// RFC 9110 §6.5.1 forbids trailers from carrying fields that affect message
1032/// routing or request semantics. Each name here is client attribution that
1033/// Sōzu handles on the header block only: `HttpContext::on_request_headers`
1034/// replaces, elides or preserves `X-Real-IP` and `X-Request-Id`, and
1035/// synthesises, extends, removes or passes through `X-Forwarded-For`,
1036/// `Forwarded`, `X-Forwarded-Proto`, `X-Forwarded-Port` and
1037/// `X-Forwarded-Host` according to the listener's `forwarded_headers` mode
1038/// (`apply_request_rewrites_and_headers` in
1039/// `lib/src/protocol/mux/router.rs` also injects `X-Forwarded-Host` on a
1040/// host rewrite). A trailer copy would bypass that handling
1041/// and hand a forged value to any backend that merges trailers into its
1042/// header view (sozu-proxy/sozu#1689), so they are dropped unconditionally.
1043pub const TRAILER_SPOOF_VECTOR_HEADERS: [&[u8]; 7] = [
1044    b"x-real-ip",
1045    b"x-forwarded-for",
1046    b"forwarded",
1047    b"x-request-id",
1048    b"x-forwarded-proto",
1049    b"x-forwarded-port",
1050    b"x-forwarded-host",
1051];
1052
1053/// Returns true if `name`, in any case, is one of
1054/// [`TRAILER_SPOOF_VECTOR_HEADERS`].
1055pub fn is_trailer_spoof_vector(name: &[u8]) -> bool {
1056    TRAILER_SPOOF_VECTOR_HEADERS
1057        .iter()
1058        .any(|spoofed| compare_no_case(name, spoofed))
1059}
1060
1061/// Header names RFC 9110 §6.5.1 keeps out of a trailer section, in lower
1062/// case: fields "whose evaluation is necessary prior to receiving the
1063/// content, such as those that describe message framing, routing,
1064/// authentication, request modifiers, response controls, or content
1065/// format", plus the connection-specific fields of RFC 9110 §7.6.1.
1066///
1067/// RFC 9110 names only the categories. The concrete names are the ones its
1068/// predecessor, RFC 7230 §4.1.2, gave as examples for each category, read
1069/// on the request side:
1070///
1071/// - framing: `Content-Length`, `Transfer-Encoding`;
1072/// - routing: `Host`;
1073/// - request modifiers, "controls and conditionals in Section 5 of
1074///   \[RFC7231\]": `Cache-Control`, `Expect`, `Max-Forwards`, `Pragma`,
1075///   `Range`, `TE` (RFC 7231 §5.1) and `If-Match`, `If-None-Match`,
1076///   `If-Modified-Since`, `If-Unmodified-Since`, `If-Range` (RFC 7231 §5.2);
1077/// - authentication, "see \[RFC7235\] and \[RFC6265\]": `Authorization`,
1078///   `Proxy-Authorization`, `Cookie`;
1079/// - how to process the content: `Content-Encoding`, `Content-Type`,
1080///   `Content-Range`, `Trailer`.
1081///
1082/// Response control data (RFC 7231 §7.1) describes a response, and has no
1083/// request-side meaning to protect. The connection-specific fields
1084/// `Connection`, `Keep-Alive`, `Proxy-Connection` and `Upgrade` are added
1085/// because RFC 9110 §7.6.1 has an intermediary remove them before
1086/// forwarding, trailer or not, and a trailer copy never reaches the
1087/// header-block handling that does so. `TE` and `Transfer-Encoding` are in
1088/// that §7.6.1 list too.
1089///
1090/// RFC 7230 §4.1.2 says why a recipient must not act on them: "A recipient
1091/// MUST ignore (or consider as an error) any fields that are forbidden to be
1092/// sent in a trailer, since processing them as if they were present in the
1093/// header section might bypass external security filters."
1094/// [`HttpContext::filter_request_trailers`] ignores them by eliding them:
1095/// RFC 9112 §7.1.2 lets a recipient "selectively retain or discard the
1096/// received trailer fields". `pkawa::handle_trailer`
1097/// (`lib/src/protocol/mux/pkawa.rs`) elides them from an H2 request trailer
1098/// block the same way (sozu-proxy/sozu#1714), once `classify_invalid_h2_header`
1099/// has refused the connection-specific ones as RFC 9113 §8.2.2 requires.
1100pub const TRAILER_FORBIDDEN_FIELDS: [&[u8]; 25] = [
1101    // framing
1102    b"content-length",
1103    b"transfer-encoding",
1104    // routing
1105    b"host",
1106    // request modifiers: controls
1107    b"cache-control",
1108    b"expect",
1109    b"max-forwards",
1110    b"pragma",
1111    b"range",
1112    b"te",
1113    // request modifiers: conditionals
1114    b"if-match",
1115    b"if-none-match",
1116    b"if-modified-since",
1117    b"if-unmodified-since",
1118    b"if-range",
1119    // authentication
1120    b"authorization",
1121    b"proxy-authorization",
1122    b"cookie",
1123    // how to process the content
1124    b"content-encoding",
1125    b"content-type",
1126    b"content-range",
1127    b"trailer",
1128    // connection-specific
1129    b"connection",
1130    b"keep-alive",
1131    b"proxy-connection",
1132    b"upgrade",
1133];
1134
1135/// Returns true if `name`, in any case, is one of
1136/// [`TRAILER_FORBIDDEN_FIELDS`].
1137pub fn is_trailer_forbidden_field(name: &[u8]) -> bool {
1138    TRAILER_FORBIDDEN_FIELDS
1139        .iter()
1140        .any(|forbidden| compare_no_case(name, forbidden))
1141}
1142
1143/// What [`HttpContext::filter_request_trailers`] did to the trailer fields
1144/// the last parse appended.
1145#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
1146pub struct TrailerFilterOutcome {
1147    /// Fields elided because they are in [`TRAILER_SPOOF_VECTOR_HEADERS`].
1148    pub spoof_vectors: usize,
1149    /// Fields elided because they are in [`TRAILER_FORBIDDEN_FIELDS`].
1150    pub forbidden: usize,
1151    /// The trailer section went over `HttpContext::max_trailer_fields`, and
1152    /// the request was marked in error.
1153    pub over_limit: bool,
1154}
1155
1156impl kawa::h1::ParserCallbacks<Checkout> for HttpContext {
1157    fn on_headers(&mut self, stream: &mut GenericHttpStream) {
1158        match stream.kind {
1159            kawa::Kind::Request => self.on_request_headers(stream),
1160            kawa::Kind::Response => self.on_response_headers(stream),
1161        }
1162    }
1163}
1164
1165impl HttpContext {
1166    /// Creates a new instance
1167    #[allow(clippy::too_many_arguments)]
1168    pub fn new(
1169        session_id: Ulid,
1170        request_id: Ulid,
1171        protocol: Protocol,
1172        public_address: SocketAddr,
1173        session_address: Option<SocketAddr>,
1174        sticky_name: String,
1175        sozu_id_header: String,
1176        elide_x_real_ip: bool,
1177        send_x_real_ip: bool,
1178    ) -> Self {
1179        Self {
1180            session_id,
1181            id: request_id,
1182            backend_id: None,
1183            cluster_id: None,
1184            cluster_metrics_incarnation: Default::default(),
1185            affinity_key: None,
1186
1187            closing: false,
1188            keep_alive_backend: true,
1189            keep_alive_frontend: true,
1190            protocol,
1191            public_address,
1192            session_address,
1193            sticky_name,
1194            sticky_session: None,
1195            sticky_session_found: None,
1196
1197            method: None,
1198            authority: None,
1199            path: None,
1200            status: None,
1201            reason: None,
1202            user_agent: None,
1203            x_request_id: None,
1204            xff_chain: None,
1205
1206            #[cfg(feature = "opentelemetry")]
1207            otel: Default::default(),
1208
1209            backend_address: None,
1210            tls_server_name: None,
1211            tls_cert_names: None,
1212            strict_sni_binding: true,
1213            elide_x_real_ip,
1214            send_x_real_ip,
1215            forwarded_headers: ForwardedHeaders::Both,
1216            max_trailer_fields: crate::protocol::mux::H2FloodConfig::default().max_header_fields(),
1217            trailer_fields: 0,
1218            tls_version: None,
1219            tls_cipher: None,
1220            tls_alpn: None,
1221            sozu_id_header,
1222            redirect_location: None,
1223            www_authenticate: None,
1224            original_authority: None,
1225            headers_response: Vec::new(),
1226            retry_after_seconds: None,
1227            frontend_redirect_template: None,
1228            redirect_status: None,
1229            tags: None,
1230            forwarding_hop: None,
1231            access_log_message: None,
1232            backends_unavailable: false,
1233        }
1234    }
1235
1236    pub(crate) fn set_cluster_metrics_incarnation(
1237        &mut self,
1238        incarnation: crate::metrics::ClusterMetricsCapture,
1239    ) {
1240        self.cluster_metrics_incarnation = incarnation;
1241    }
1242
1243    pub(crate) fn cluster_metrics_incarnation(&self) -> crate::metrics::ClusterMetricsCapture {
1244        self.cluster_metrics_incarnation
1245    }
1246
1247    /// Callback for request:
1248    ///
1249    /// - edit headers (connection, forwarded, sticky cookie, sozu-id,
1250    ///   x-request-id, x-real-ip)
1251    /// - save information:
1252    ///   - method
1253    ///   - authority
1254    ///   - path
1255    ///   - front keep-alive
1256    ///   - sticky cookie
1257    ///   - user-agent
1258    ///   - x-request-id (preserved if present, else derived from `self.id`)
1259    fn on_request_headers(&mut self, request: &mut GenericHttpStream) {
1260        // Editing never drops a block — it only elides in place (key set to
1261        // Empty, length preserved) or pushes new headers. Snapshot the count
1262        // so the postcondition can pin "blocks only grow" for the whole edit.
1263        let blocks_at_entry = request.blocks.len();
1264
1265        // Defense-in-depth against CL.TE request smuggling (CWE-444, RFC 9110 §7.6 /
1266        // RFC 9112 §6.1; reopen of #726). kawa leaves every Transfer-Encoding field line
1267        // in place, so we reason about ALL surviving TE headers rather than its aggregate
1268        // `body_size`. Reject:
1269        //   * more than one non-elided TE header  (RFC 9112 §6.1: chunked must be applied
1270        //     once and be the final coding; several field lines cannot be safely
1271        //     reconciled here, and forwarding them all hands the backend the combining
1272        //     problem we just declined to solve), or
1273        //   * a TE header whose field value does not end in `chunked`, or
1274        //   * a TE header present while kawa did not adopt chunked framing.
1275        //
1276        // WHICH CLAUSE ACTUALLY FIRES, against the kawa this workspace pins (`^0.7.2`,
1277        // locked at 0.7.2; its Transfer-Encoding handling is 0.7.1's). Read from kawa's
1278        // `process_headers` (its `src/protocol/h1/parser/mod.rs`), not measured here: kawa resolves the
1279        // combined Transfer-Encoding BEFORE this callback and, for a REQUEST whose
1280        // combined final coding is not chunked, errors the parse and returns without
1281        // calling `callbacks.on_headers` at all (RFC 9112 §6.3). It judges the LAST TE
1282        // line, per RFC 9110 §5.3 combining, so an earlier `chunked` line can no longer
1283        // latch chunked framing on its own — that latch was a kawa 0.7.0 shape.
1284        //   * the COUNT clause is the one that fires on traffic kawa accepted:
1285        //     `Transfer-Encoding: identity` followed by `Transfer-Encoding: chunked`
1286        //     combines to a chunked-final coding, so kawa parses it clean, and forwarding
1287        //     both lines is what we refuse.
1288        //   * the other two cannot fire on a request kawa accepted: a surviving TE line is
1289        //     chunked-final (or kawa already refused the request), and `body_size` is then
1290        //     Chunked. `chunked, gzip` never reaches this code — kawa refused it. They are
1291        //     defense in depth against a kawa regression. Keep them; do not describe them
1292        //     as closing a live hole.
1293        //
1294        // OWS IS NOT AMBIGUITY, and this guard must not treat it as one. kawa >=0.7.1
1295        // trims OWS from every field value at parse time (`trim_ows` in its
1296        // `parser/primitives.rs`, RFC 9112 §5), so `header.val` here already reads
1297        // `chunked` for `Transfer-Encoding: chunked\t`: the suffix check passes,
1298        // `body_size` is Chunked, the request is ACCEPTED — and it reaches the backend
1299        // spelled `chunked`, with the Content-Length elided. Measured:
1300        // `test_h1_te_ows_forwarded_canonically` (`e2e/src/tests/h1_security_tests.rs`)
1301        // sends exactly that request and pins the forwarded bytes. An obfuscated coding
1302        // over a valid chunked body is legal traffic; refusing it would refuse legal
1303        // traffic.
1304        //
1305        // kawa 0.7.0 is what the suffix check was written against: it framed on the
1306        // trimmed reading but forwarded the field line verbatim, so a backend that did not
1307        // itself trim saw neither a coding it recognized nor a Content-Length, and read
1308        // our chunked body as a pipelined request — a TE.TE desync. 0.7.1 closed that at
1309        // the source. Do not reintroduce a trailing-OWS rejection here to compensate.
1310        //
1311        // The `trailing-tab` / `trailing-space` rows of that module's `TE_SMUGGLING_CASES`
1312        // still answer 400, for their invalid chunked BODY (`Hello` is not a chunk) and
1313        // not through this predicate — as that table's own doc comment says. Their 400 is
1314        // not evidence that this guard rejects an obfuscated coding.
1315        let (te_count, te_all_suffix_chunked) = {
1316            const CHUNKED: &[u8] = b"chunked";
1317            let buf0 = request.storage.buffer();
1318            request
1319                .blocks
1320                .iter()
1321                .filter_map(|block| match block {
1322                    kawa::Block::Header(header)
1323                        if !header.is_elided()
1324                            && compare_no_case(header.key.data(buf0), b"transfer-encoding") =>
1325                    {
1326                        Some(header.val.data(buf0))
1327                    }
1328                    _ => None,
1329                })
1330                .fold((0usize, true), |(count, all_chunked), val| {
1331                    let suffix_chunked = val.len() >= CHUNKED.len()
1332                        && compare_no_case(&val[val.len() - CHUNKED.len()..], CHUNKED);
1333                    (count + 1, all_chunked && suffix_chunked)
1334                })
1335        };
1336        if te_count > 1
1337            || (te_count == 1
1338                && (!te_all_suffix_chunked || request.body_size != kawa::BodySize::Chunked))
1339        {
1340            incr!(names::http::FRONTEND_TE_SMUGGLING);
1341            warn!(
1342                "{} rejecting request: ambiguous Transfer-Encoding framing (possible CL.TE request smuggling)",
1343                self.log_context()
1344            );
1345            request
1346                .parsing_phase
1347                .error("Transfer-Encoding conflicts with message framing".into());
1348            return;
1349        }
1350
1351        // RFC 9110 §8.6: `Content-Length = 1*DIGIT`, and a sender MUST NOT
1352        // forward a message whose Content-Length does not match it. kawa
1353        // 0.7.1 framed `Content-Length: +5` as 5 bytes and left the line to be
1354        // forwarded as sent: a backend that refuses or re-reads `+5` takes the
1355        // body for the start of the next request — never routed, never
1356        // Basic-auth checked (CWE-444, sozu#1652). RFC 9112 §6.3 rule 5 makes
1357        // it an unrecoverable framing error: 400. kawa >= 0.7.2 refuses every
1358        // non-digit value before calling back (CleverCloud/kawa#26), so this
1359        // clause no longer fires on a request kawa accepted: it is defense in
1360        // depth against a kawa regression — keep it, and do not describe it
1361        // as closing a live hole. A list of equal values
1362        // (`5, 5`), which RFC 9110 §8.6 lets a recipient either collapse or
1363        // reject, is rejected — kawa already refuses it. `005` is `1*DIGIT`
1364        // and is forwarded as sent, as the H2 path forwards it. Mirrors
1365        // `RejectReason::DuplicateCl` on the H2 side
1366        // (`pkawa::write_regular_header`), which is why this cannot fire for
1367        // an H2 request.
1368        if has_non_digit_content_length(request) {
1369            incr!(names::http::FRONTEND_CONTENT_LENGTH_INVALID);
1370            warn!(
1371                "{} rejecting request: Content-Length is not 1*DIGIT (possible request smuggling)",
1372                self.log_context()
1373            );
1374            request
1375                .parsing_phase
1376                .error("Content-Length is not 1*DIGIT".into());
1377            return;
1378        }
1379
1380        let buf = request.storage.mut_buffer();
1381
1382        // Captures the request line
1383        if let kawa::StatusLine::Request {
1384            method,
1385            authority,
1386            path,
1387            ..
1388        } = &request.detached.status_line
1389        {
1390            self.method = method.data_opt(buf).map(Method::new);
1391            self.authority = authority
1392                .data_opt(buf)
1393                .and_then(|data| from_utf8(data).ok())
1394                .map(ToOwned::to_owned);
1395            self.path = path
1396                .data_opt(buf)
1397                .and_then(|data| from_utf8(data).ok())
1398                .map(ToOwned::to_owned);
1399        }
1400
1401        // A request with neither Content-Length nor Transfer-Encoding has no
1402        // body, whatever its method or version (RFC 9112 §6.3 rule 7);
1403        // close-delimited framing (rule 8) is for responses only. kawa 0.7.1
1404        // did not make that distinction: `kawa::h1::parse` mapped
1405        // `BodySize::Empty` to `ParsingPhase::Body` for both kinds, and that
1406        // arm takes every byte left in the buffer. A request pipelined behind
1407        // this one would then be forwarded raw as its "body" to this
1408        // request's backend: never routed, never Basic-auth checked, without
1409        // `Sozu-Id` or `X-Forwarded-*` (CWE-444, sozu#1650). End it here
1410        // instead, so the flags block kawa pushes right after this callback
1411        // carries `end_stream`, and the next request stays unparsed for its
1412        // own turn.
1413        //
1414        // kawa >= 0.7.2 applies rule 7 itself (CleverCloud/kawa#27): it maps
1415        // `BodySize::Empty` to `ParsingPhase::Terminated` for a request before
1416        // calling back, so this branch no longer fires on the H1 path. It is
1417        // defense in depth against a kawa regression — keep it.
1418        //
1419        // The `ParsingPhase::Body` conjunct is what confines this to kawa's
1420        // H1 parser, which set that phase before calling back (kawa 0.7.1).
1421        // `pkawa::handle_header` calls this same callback for an H2 request
1422        // while its `body_size` is still `Empty` and its phase still the
1423        // initial one, then frames a request whose DATA follows as chunked
1424        // unless the callback terminated it: `Empty` alone would drop that
1425        // body. Pinned by `an_h2_request_is_left_for_pkawa_to_frame`.
1426        if request.parsing_phase == kawa::ParsingPhase::Body
1427            && request.body_size == kawa::BodySize::Empty
1428        {
1429            request.parsing_phase = kawa::ParsingPhase::Terminated;
1430        }
1431
1432        let public_port = self.public_address.port();
1433        let proto = match self.protocol {
1434            Protocol::HTTP => "http",
1435            Protocol::HTTPS => "https",
1436            _ => unreachable!(),
1437        };
1438
1439        // `proto` is the proxy-resolved scheme label, sourced from the
1440        // listener protocol (never from request bytes). The match above
1441        // already rejects anything but HTTP/HTTPS; pin that it is one of the
1442        // two CR/LF-free literals we splice into forwarding headers.
1443        debug_assert!(
1444            proto == "http" || proto == "https",
1445            "proto must be the http/https scheme label"
1446        );
1447
1448        // Find and remove the sticky_name cookie
1449        // if found its value is stored in sticky_session_found
1450        for cookie in &mut request.detached.jar {
1451            let key = cookie.key.data(buf);
1452            if key == self.sticky_name.as_bytes() {
1453                let val = cookie.val.data(buf);
1454                self.sticky_session_found = from_utf8(val).ok().map(ToOwned::to_owned);
1455                cookie.elide();
1456                // Post: the matched sticky cookie is gone from the forwarded
1457                // jar so the backend never sees Sōzu's own session cookie.
1458                debug_assert!(
1459                    cookie.is_elided(),
1460                    "the matched sticky cookie must be elided after capture"
1461                );
1462            }
1463        }
1464
1465        // If found:
1466        // - set keep_alive_frontend to false if Connection lists "close"
1467        //   (RFC 9110 §7.6.1, RFC 9112 §9.6), as on the response side,
1468        //   shutting down or not
1469        // - update value of X-Forwarded-Proto
1470        // - update value of X-Forwarded-Port
1471        // - store X-Forwarded-For
1472        // - store Forwarded
1473        // - store User-Agent
1474        let forwarded_headers = self.forwarded_headers;
1475        let emits_x_forwarded = emits_x_forwarded(forwarded_headers);
1476        let emits_rfc7239 = emits_rfc7239(forwarded_headers);
1477        let strips_x_forwarded = strips_x_forwarded(forwarded_headers);
1478        // A mode that strips the client X-Forwarded-* family never adds it.
1479        debug_assert!(
1480            !(strips_x_forwarded && emits_x_forwarded),
1481            "a mode that strips X-Forwarded-* must not emit it"
1482        );
1483        let mut x_for = None;
1484        let mut forwarded = None;
1485        let mut has_x_port = false;
1486        let mut has_x_proto = false;
1487        let mut has_x_request_id = false;
1488        #[cfg(feature = "opentelemetry")]
1489        let mut traceparent: Option<&mut kawa::Pair> = None;
1490        #[cfg(feature = "opentelemetry")]
1491        let mut tracestate: Option<&mut kawa::Pair> = None;
1492        for block in &mut request.blocks {
1493            match block {
1494                kawa::Block::Header(header) if !header.is_elided() => {
1495                    let key = header.key.data(buf);
1496                    if compare_no_case(key, b"connection") {
1497                        let val = header.val.data(buf);
1498                        self.keep_alive_frontend &= !has_connection_option(val, b"close");
1499                    } else if strips_x_forwarded && is_managed_x_forwarded(key) {
1500                        // `rfc7239`: a client X-Forwarded-For, -Proto,
1501                        // -Port or -Host is removed, not trusted. The
1502                        // X-Forwarded-For value is still recorded for the
1503                        // access log, which keeps what the client attested.
1504                        if compare_no_case(key, b"X-Forwarded-For") {
1505                            self.xff_chain = header
1506                                .val
1507                                .data_opt(buf)
1508                                .and_then(|data| from_utf8(data).ok())
1509                                .map(ToOwned::to_owned);
1510                        }
1511                        header.elide();
1512                        // Post: the client value never reaches the backend.
1513                        debug_assert!(
1514                            header.is_elided(),
1515                            "a client X-Forwarded-* header must be elided in rfc7239 mode"
1516                        );
1517                    } else if !emits_x_forwarded
1518                        && (compare_no_case(key, b"X-Forwarded-Proto")
1519                            || compare_no_case(key, b"X-Forwarded-Port"))
1520                    {
1521                        // `none`: Sōzu takes no stance on a client
1522                        // X-Forwarded-Proto or -Port — it neither trusts nor
1523                        // replaces them, so they pass through as sent.
1524                        debug_assert!(
1525                            forwarded_headers == ForwardedHeaders::None,
1526                            "only `none` passes a client X-Forwarded-Proto/-Port through"
1527                        );
1528                    } else if compare_no_case(key, b"X-Forwarded-Proto") {
1529                        has_x_proto = true;
1530                        // header.val = kawa::Store::Static(proto.as_bytes());
1531                        incr!(names::http::TRUSTING_X_PROTO);
1532                        let val = header.val.data(buf);
1533                        if !compare_no_case(val, proto.as_bytes()) {
1534                            incr!(names::http::TRUSTING_X_PROTO_DIFF);
1535                            debug!(
1536                                "{} Trusting X-Forwarded-Proto for {:?} even though {:?} != {}",
1537                                self.log_context(),
1538                                self.authority,
1539                                val,
1540                                proto
1541                            );
1542                        }
1543                    } else if compare_no_case(key, b"X-Forwarded-Port") {
1544                        has_x_port = true;
1545                        // header.val = kawa::Store::from_string(public_port.to_string());
1546                        incr!(names::http::TRUSTING_X_PORT);
1547                        let val = header.val.data(buf);
1548                        let mut port_buf = itoa::Buffer::new();
1549                        let expected = port_buf.format(public_port);
1550                        if !compare_no_case(val, expected.as_bytes()) {
1551                            incr!(names::http::TRUSTING_X_PORT_DIFF);
1552                            debug!(
1553                                "{} Trusting X-Forwarded-Port for {:?} even though {:?} != {}",
1554                                self.log_context(),
1555                                self.authority,
1556                                val,
1557                                expected
1558                            );
1559                        }
1560                    } else if compare_no_case(key, b"X-Forwarded-For") {
1561                        // Snapshot the upstream-attested chain before we
1562                        // potentially append our own peer below — the access
1563                        // log records the value the client/upstream LB
1564                        // forwarded, not the rewritten value Sōzu emits.
1565                        self.xff_chain = header
1566                            .val
1567                            .data_opt(buf)
1568                            .and_then(|data| from_utf8(data).ok())
1569                            .map(ToOwned::to_owned);
1570                        x_for = Some(header);
1571                    } else if compare_no_case(key, b"X-Real-IP") && self.elide_x_real_ip {
1572                        // Anti-spoofing: a client cannot supply its own
1573                        // `X-Real-IP` and have it reach the backend. The
1574                        // proxy-injected value (when `send_x_real_ip` is
1575                        // also set) is appended after this loop. Trailer
1576                        // fields bypass this callback; they are covered by
1577                        // `TRAILER_SPOOF_VECTOR_HEADERS`, dropped from H2
1578                        // trailer HEADERS frames by `pkawa::handle_trailer`
1579                        // and from H1 chunked trailers by
1580                        // `HttpContext::filter_request_trailers`.
1581                        debug_assert!(
1582                            self.elide_x_real_ip,
1583                            "X-Real-IP is only elided when anti-spoofing is enabled"
1584                        );
1585                        header.elide();
1586                        // Post: the spoofable client value is stripped before
1587                        // forwarding (anti-spoofing invariant, CWE-348).
1588                        debug_assert!(
1589                            header.is_elided(),
1590                            "client X-Real-IP must be elided when elide_x_real_ip is set"
1591                        );
1592                    } else if compare_no_case(key, b"Forwarded") {
1593                        if emits_rfc7239
1594                            && !is_valid_forwarded(header.val.data_opt(buf).unwrap_or_default())
1595                        {
1596                            // A malformed client chain cannot be extended:
1597                            // an unclosed quoted-string would swallow the
1598                            // element appended after it. RFC 7239 §4 lets a
1599                            // proxy remove the field; the element Sōzu adds
1600                            // then goes to an earlier well-formed line or a
1601                            // synthesised one, so it is always parseable
1602                            // and last. The counter is the operator-visible
1603                            // trace of the removal; the log stays at debug.
1604                            incr!(names::http::FORWARDED_MALFORMED_ELIDED);
1605                            debug!(
1606                                "{} eliding a malformed client Forwarded header",
1607                                self.log_context()
1608                            );
1609                            header.elide();
1610                            debug_assert!(
1611                                header.is_elided(),
1612                                "a malformed client Forwarded must be elided"
1613                            );
1614                        } else {
1615                            forwarded = Some(header);
1616                        }
1617                    } else if compare_no_case(key, b"User-Agent") {
1618                        self.user_agent = header
1619                            .val
1620                            .data_opt(buf)
1621                            .and_then(|data| from_utf8(data).ok())
1622                            .map(ToOwned::to_owned);
1623                    } else if compare_no_case(key, b"X-Request-Id") {
1624                        // RFC: not standardized, but the de-facto correlation
1625                        // header used by Envoy/HAProxy/most LBs. Preserve the
1626                        // client-supplied value verbatim — overwriting it
1627                        // breaks end-to-end request tracing.
1628                        has_x_request_id = true;
1629                        self.x_request_id = header
1630                            .val
1631                            .data_opt(buf)
1632                            .and_then(|data| from_utf8(data).ok())
1633                            .map(Rc::from);
1634                    } else {
1635                        #[cfg(feature = "opentelemetry")]
1636                        if compare_no_case(key, b"traceparent") {
1637                            if let Some(hdr) = traceparent {
1638                                hdr.elide();
1639                            }
1640                            traceparent = Some(header);
1641                        } else if compare_no_case(key, b"tracestate") {
1642                            if let Some(hdr) = tracestate {
1643                                hdr.elide();
1644                            }
1645                            tracestate = Some(header);
1646                        }
1647                    }
1648                }
1649                _ => {}
1650            }
1651        }
1652
1653        #[cfg(feature = "opentelemetry")]
1654        let (otel, has_traceparent) = {
1655            let mut otel = sozu_command_lib::logging::OpenTelemetry::default();
1656            let tp = traceparent
1657                .as_ref()
1658                .and_then(|hdr| parse_traceparent(&hdr.val, buf))
1659                .map(|(trace_id, parent_id)| (trace_id, Some(parent_id)));
1660            // Remove tracestate if no traceparent is present
1661            if let (None, Some(tracestate)) = (tp, tracestate) {
1662                tracestate.elide();
1663            }
1664            let (trace_id, parent_id) = tp.unwrap_or_else(|| (random_id(), None));
1665            otel.trace_id = trace_id;
1666            otel.parent_span_id = parent_id;
1667            otel.span_id = random_id();
1668            // Modify header if present
1669            if let Some(id) = &mut traceparent {
1670                let new_val = build_traceparent(&otel.trace_id, &otel.span_id);
1671                id.val.modify(buf, &new_val);
1672            }
1673            (otel, traceparent.is_some())
1674        };
1675
1676        // The connection's forwarding values, rendered by its first request
1677        // and shared by the next ones: a reference-count bump each, never a
1678        // copy (`HttpContext::forwarding_hop`).
1679        let (port_hop, x_forwarded_for_hop, forwarded_hop) = {
1680            let hop = self.forwarding_hop(proto);
1681            (
1682                hop.port.clone(),
1683                hop.x_forwarded_for.clone(),
1684                hop.forwarded.clone(),
1685            )
1686        };
1687
1688        // If session_address is set:
1689        // - append its ip address to the list of "X-Forwarded-For" if it was found, creates it if not
1690        // - append "proto=[PROTO];for=[PEER];by=[PUBLIC]" to the list of "Forwarded" if it was found, creates it if not
1691        if let (Some(x_forwarded_for_hop), Some(forwarded_hop)) =
1692            (x_forwarded_for_hop, forwarded_hop)
1693        {
1694            let has_x_for = x_for.is_some();
1695            let has_forwarded = forwarded.is_some();
1696
1697            // A client-supplied chain is request-scoped: it is extended into
1698            // one exact-size copy (`extended_chain`), never kept. A mode
1699            // that does not emit a family leaves its client chain as sent
1700            // (`rfc7239` elided the X-Forwarded-For lines in the walk above).
1701            if emits_x_forwarded && let Some(header) = x_for {
1702                header.val = extended_chain(header.val.data(buf), &x_forwarded_for_hop);
1703            }
1704            if emits_rfc7239 && let Some(header) = &mut forwarded {
1705                header.val = extended_chain(header.val.data(buf), &forwarded_hop);
1706            }
1707
1708            if emits_x_forwarded && !has_x_for {
1709                let blocks_before = request.blocks.len();
1710                request.push_block(kawa::Block::Header(kawa::Pair {
1711                    key: kawa::Store::Static(b"X-Forwarded-For"),
1712                    val: synthesised_from_hop(&x_forwarded_for_hop),
1713                }));
1714                debug_assert_eq!(
1715                    request.blocks.len(),
1716                    blocks_before + 1,
1717                    "creating X-Forwarded-For must push exactly one block"
1718                );
1719            }
1720            if emits_rfc7239 && !has_forwarded {
1721                let blocks_before = request.blocks.len();
1722                request.push_block(kawa::Block::Header(kawa::Pair {
1723                    key: kawa::Store::Static(b"Forwarded"),
1724                    val: synthesised_from_hop(&forwarded_hop),
1725                }));
1726                debug_assert_eq!(
1727                    request.blocks.len(),
1728                    blocks_before + 1,
1729                    "creating Forwarded must push exactly one block"
1730                );
1731            }
1732
1733            // Inject a proxy-generated `X-Real-IP` header carrying the
1734            // peer IP (post-PROXY-v2 unwrap, so the original client IP
1735            // even when the upstream presented PROXY-v2). Folded into the
1736            // peer arm so missing peers (raw socket, no PROXY-v2) skip the
1737            // injection silently — identical to the X-Forwarded-For /
1738            // Forwarded synthesis behaviour above. `X-Real-IP` belongs to
1739            // neither forwarding family, so `forwarded_headers` never gates
1740            // it: the hop is rendered in every mode. It shares the
1741            // X-Forwarded-For rendering of the peer. Any client-supplied
1742            // `X-Real-IP` was either elided in the block walk (if
1743            // `elide_x_real_ip` is on) or passes through; this header is
1744            // appended last so order in the resulting block list is
1745            // deterministic for tests.
1746            if self.send_x_real_ip {
1747                let blocks_before = request.blocks.len();
1748                request.push_block(kawa::Block::Header(kawa::Pair {
1749                    key: kawa::Store::Static(b"X-Real-IP"),
1750                    val: synthesised_from_hop(&x_forwarded_for_hop),
1751                }));
1752                debug_assert_eq!(
1753                    request.blocks.len(),
1754                    blocks_before + 1,
1755                    "injecting X-Real-IP must push exactly one block"
1756                );
1757            }
1758        }
1759
1760        #[cfg(feature = "opentelemetry")]
1761        {
1762            if !has_traceparent {
1763                let val = build_traceparent(&otel.trace_id, &otel.span_id);
1764                request.push_block(kawa::Block::Header(kawa::Pair {
1765                    key: kawa::Store::Static(b"traceparent"),
1766                    val: kawa::Store::from_slice(&val),
1767                }));
1768            }
1769            self.otel = Some(otel);
1770        }
1771
1772        if emits_x_forwarded && !has_x_port {
1773            request.push_block(kawa::Block::Header(kawa::Pair {
1774                key: kawa::Store::Static(b"X-Forwarded-Port"),
1775                val: kawa::Store::Shared(port_hop, 0),
1776            }));
1777        }
1778        if emits_x_forwarded && !has_x_proto {
1779            request.push_block(kawa::Block::Header(kawa::Pair {
1780                key: kawa::Store::Static(b"X-Forwarded-Proto"),
1781                val: kawa::Store::Static(proto.as_bytes()),
1782            }));
1783        }
1784        // A shutting-down Sōzu closes the client connection after this
1785        // response (`ConnectionH1::writable`, `lib/src/protocol/mux/h1.rs`),
1786        // so the request announces the close too, keeping the other options
1787        // the client listed. It refuses a protocol upgrade: `writable` tests
1788        // `closing` before it handles a 101, so an upgrade would switch the
1789        // client to a connection that closes at once. Dropping the `upgrade`
1790        // option with its `Upgrade` field lets the backend answer in
1791        // HTTP/1.1, which it always may (RFC 9110 §7.8), and the client
1792        // retry the upgrade on a new connection.
1793        if self.closing {
1794            merge_connection_options_into_close(request, true);
1795        }
1796        // Inject "X-Request-Id" derived from the request ULID when the client
1797        // (or upstream LB) did not already supply one. When already present,
1798        // the header is left untouched in the block list — preserving the
1799        // client-supplied value end-to-end is the whole point of this header.
1800        // Either way, `self.x_request_id` is populated so the access log
1801        // records the exact value forwarded to the backend.
1802        if has_x_request_id {
1803            incr!(names::http::X_REQUEST_ID_PROPAGATED);
1804        } else {
1805            // The one rendering of `self.id` this request allocates: the
1806            // forwarded header, the access log and both correlation headers
1807            // share it (`Self::shared_id`).
1808            let rendered = render_ulid(self.id);
1809            let value: Rc<str> = Rc::from(from_utf8(&rendered).expect("a rendered ULID is ASCII"));
1810            // The generated id is a ULID rendering — Crockford base-32, so
1811            // CR/LF-free by construction.
1812            debug_assert!(
1813                is_crlf_free(value.as_bytes()),
1814                "the generated X-Request-Id must be CR/LF-free"
1815            );
1816            request.push_block(kawa::Block::Header(kawa::Pair {
1817                key: kawa::Store::Static(b"X-Request-Id"),
1818                val: kawa::Store::Shared(Rc::<[u8]>::from(value.clone()), 0),
1819            }));
1820            self.x_request_id = Some(value);
1821            incr!(names::http::X_REQUEST_ID_GENERATED);
1822        }
1823        // Either branch leaves the forwarded value recorded for the access
1824        // log so it matches exactly what the backend receives.
1825        debug_assert!(
1826            self.x_request_id.is_some(),
1827            "on_request_headers must record the forwarded X-Request-Id"
1828        );
1829
1830        // Create a custom correlation header (defaults to "Sozu-Id", can be
1831        // renamed via the `sozu_id_header` listener config knob).
1832        let blocks_before_sozu_id = request.blocks.len();
1833        request.push_block(kawa::Block::Header(kawa::Pair {
1834            key: correlation_header_name(&self.sozu_id_header),
1835            val: kawa::Store::Shared(self.shared_id(), 0),
1836        }));
1837        debug_assert_eq!(
1838            request.blocks.len(),
1839            blocks_before_sozu_id + 1,
1840            "the Sozu-Id correlation header must be pushed exactly once"
1841        );
1842
1843        // Postcondition: the whole edit only ever added or elided headers —
1844        // the block count is monotonically non-decreasing (a removed header
1845        // is elided in place, never popped), so the backend never loses a
1846        // header it should have seen.
1847        debug_assert!(
1848            request.blocks.len() >= blocks_at_entry,
1849            "header editing must never drop a block from the request"
1850        );
1851        // Postcondition: `rfc7239` forwards no X-Forwarded-* header the
1852        // editor manages, client-supplied or synthesised.
1853        debug_assert!(
1854            !strips_x_forwarded
1855                || request.blocks.iter().all(|block| match block {
1856                    kawa::Block::Header(header) if !header.is_elided() =>
1857                        !is_managed_x_forwarded(header.key.data(request.storage.buffer())),
1858                    _ => true,
1859                }),
1860            "rfc7239 mode must forward no X-Forwarded-For/-Proto/-Port/-Host"
1861        );
1862    }
1863
1864    /// Callback for response:
1865    ///
1866    /// - edit headers (connection, set-cookie, sozu-id)
1867    /// - forward an HTTP/1.0 response as HTTP/1.1 (RFC 9110 §6.2)
1868    /// - save information:
1869    ///   - status code
1870    ///   - reason
1871    ///   - back keep-alive
1872    fn on_response_headers(&mut self, response: &mut GenericHttpStream) {
1873        // Like the request path, response editing only adds or elides — pin
1874        // the entry count so the postcondition can assert "blocks only grow".
1875        let blocks_at_entry = response.blocks.len();
1876
1877        // The response mirror of the request's Content-Length clause: a
1878        // proxy that receives an invalid Content-Length MUST discard the
1879        // response and answer 502 (RFC 9112 §6.3 rule 5). Failing the parse
1880        // here does that: `ConnectionH1::readable` ends the backend stream,
1881        // and `shared::end_stream_decision` answers the client 502 since no
1882        // byte of the response was consumed. Checked before the 204/304/1xx
1883        // override of `body_size`, which leaves the field line in place.
1884        // kawa >= 0.7.2 already fails such a response before calling back
1885        // (CleverCloud/kawa#26) — the same 502 through the same path — so
1886        // this clause is defense in depth against a kawa regression.
1887        if has_non_digit_content_length(response) {
1888            incr!(names::http::BACKEND_CONTENT_LENGTH_INVALID);
1889            warn!(
1890                "{} rejecting response: Content-Length is not 1*DIGIT",
1891                self.log_context()
1892            );
1893            response
1894                .parsing_phase
1895                .error("Content-Length is not 1*DIGIT".into());
1896            return;
1897        }
1898
1899        let buf = &mut response.storage.mut_buffer();
1900
1901        // Captures the response line
1902        if let kawa::StatusLine::Response { code, reason, .. } = &response.detached.status_line {
1903            self.status = Some(*code);
1904            // The standard phrase for the code is borrowed, anything else
1905            // copied: the access log records the wire bytes either way.
1906            self.reason = reason
1907                .data_opt(buf)
1908                .and_then(|data| from_utf8(data).ok())
1909                .map(|data| match standard_reason(*code) {
1910                    Some(phrase) if phrase == data => Cow::Borrowed(phrase),
1911                    _ => Cow::Owned(data.to_owned()),
1912                });
1913            // Post: the captured reason is the wire reason, byte for byte,
1914            // whichever representation carries it.
1915            debug_assert_eq!(
1916                self.reason.as_deref().map(str::as_bytes),
1917                reason.data_opt(buf).filter(|data| from_utf8(data).is_ok()),
1918                "the captured reason must be the backend's own bytes"
1919            );
1920            debug_assert!(
1921                !matches!(&self.reason, Some(Cow::Owned(owned))
1922                    if standard_reason(*code) == Some(owned.as_str())),
1923                "a standard reason must be borrowed, never copied"
1924            );
1925        }
1926
1927        if self.method == Some(Method::Head) {
1928            response.parsing_phase = kawa::ParsingPhase::Terminated;
1929        }
1930
1931        let (is_http10, is_interim) = match &response.detached.status_line {
1932            kawa::StatusLine::Response { version, code, .. } => (
1933                matches!(version, kawa::Version::V10),
1934                (100..200).contains(code),
1935            ),
1936            _ => (false, false),
1937        };
1938
1939        // If found:
1940        // - set keep_alive_backend to false if Connection lists "close",
1941        //   shutting down or not
1942        // - note a "close" and a "keep-alive" option, and Transfer-Encoding
1943        //   (HTTP/1.0)
1944        let mut announces_close = false;
1945        let mut asks_keep_alive = false;
1946        let mut has_transfer_encoding = false;
1947        for block in &mut response.blocks {
1948            match block {
1949                kawa::Block::Header(header) if !header.is_elided() => {
1950                    let key = header.key.data(buf);
1951                    if compare_no_case(key, b"connection") {
1952                        let val = header.val.data(buf);
1953                        let is_close = has_connection_option(val, b"close");
1954                        announces_close |= is_close;
1955                        self.keep_alive_backend &= !is_close;
1956                        asks_keep_alive |= has_connection_option(val, b"keep-alive");
1957                    } else if compare_no_case(key, b"transfer-encoding") {
1958                        has_transfer_encoding = true;
1959                    }
1960                }
1961                _ => {}
1962            }
1963        }
1964
1965        // RFC 9110 §6.2: Sōzu answers in its own version, HTTP/1.1, whatever
1966        // the backend spoke. kawa serialises `Version::V10` as `HTTP/1.0` for
1967        // an H1 client (the H2 converter ignores the version), so only an
1968        // HTTP/1.0 response needs rewriting.
1969        //
1970        // HTTP/1.0 is not persistent unless the response asks to be with a
1971        // `keep-alive` connection option (RFC 9112 §9.3); a close-delimited
1972        // body (RFC 9112 §6.3 rule 8) ends with the close whatever it asked,
1973        // and `Transfer-Encoding` in an HTTP/1.0 message is faulty framing
1974        // that closes the connection after it (RFC 9112 §6.1). Forwarded as
1975        // HTTP/1.1 the client would read it as persistent, so a
1976        // non-persistent one clears `keep_alive_backend`, which lets the
1977        // backend EOF end a close-delimited body
1978        // (`ConnectionH1::terminate_close_delimited`,
1979        // `lib/src/protocol/mux/h1.rs`) and closes an H1 client connection
1980        // after the response (`ConnectionH1::writable`), and announces that
1981        // close (RFC 9112 §9.6).
1982        //
1983        // A shutting-down Sōzu closes the client connection after the
1984        // response too (`ConnectionH1::writable` tests `closing` first), in
1985        // HTTP/1.1 as in HTTP/1.0, and announces it the same way. Either
1986        // close merges the response's `Connection` options into one line
1987        // ending in `close`, keeping every field they nominate hop-by-hop
1988        // (RFC 9110 §7.6.1) and dropping only the `keep-alive` the close
1989        // contradicts. The H2 converter drops that line with every
1990        // connection-specific header.
1991        //
1992        // An interim 1xx says nothing about the connection: its persistence
1993        // belongs to the final response, which runs this callback again, and
1994        // a 101 must keep its `Connection: Upgrade` (RFC 9110 §7.8).
1995        if is_http10 && !is_interim {
1996            let close_delimited = response.parsing_phase == kawa::ParsingPhase::Body
1997                && response.body_size == kawa::BodySize::Empty;
1998            if !asks_keep_alive || close_delimited || has_transfer_encoding {
1999                self.keep_alive_backend = false;
2000            }
2001        }
2002        if !is_interim && (self.closing || (is_http10 && !self.keep_alive_backend)) {
2003            merge_connection_options_into_close(response, false);
2004            announces_close = true;
2005        }
2006        if is_http10 {
2007            if let kawa::StatusLine::Response { version, .. } = &mut response.detached.status_line {
2008                *version = kawa::Version::V11;
2009            }
2010            // Post: the backend's version never reaches the client, and a
2011            // response the backend will not keep announces the close.
2012            debug_assert!(
2013                matches!(
2014                    response.detached.status_line,
2015                    kawa::StatusLine::Response {
2016                        version: kawa::Version::V11,
2017                        ..
2018                    }
2019                ),
2020                "an HTTP/1.0 response must be forwarded as HTTP/1.1"
2021            );
2022            debug_assert!(
2023                self.keep_alive_backend || announces_close,
2024                "a non-persistent HTTP/1.0 response must carry Connection: close"
2025            );
2026        }
2027
2028        // If the sticky_session is set and differs from the one found in the request
2029        // create a "Set-Cookie" header to update the sticky_name value
2030        if let Some(sticky_session) = &self.sticky_session
2031            && self.sticky_session != self.sticky_session_found
2032        {
2033            let blocks_before = response.blocks.len();
2034            let mut cookie_buf =
2035                Vec::with_capacity(self.sticky_name.len() + 1 + sticky_session.len() + 8);
2036            cookie_buf.extend_from_slice(self.sticky_name.as_bytes());
2037            cookie_buf.push(b'=');
2038            cookie_buf.extend_from_slice(sticky_session.as_bytes());
2039            cookie_buf.extend_from_slice(b"; Path=/");
2040            // The cookie value is `name=session; Path=/`, built from the
2041            // proxy-controlled sticky name + session id — assert it is
2042            // well-formed (contains the `=` separator, ends with the
2043            // attribute suffix) and CR/LF-free so it cannot inject an
2044            // extra Set-Cookie / split the response (CWE-113).
2045            debug_assert!(
2046                cookie_buf.contains(&b'='),
2047                "the Set-Cookie value must carry the name=value separator"
2048            );
2049            debug_assert!(
2050                cookie_buf.ends_with(b"; Path=/"),
2051                "the Set-Cookie value must end with the Path attribute"
2052            );
2053            debug_assert!(
2054                is_crlf_free(&cookie_buf),
2055                "the synthesised Set-Cookie value must be CR/LF-free (anti-injection)"
2056            );
2057            response.push_block(kawa::Block::Header(kawa::Pair {
2058                key: kawa::Store::Static(b"Set-Cookie"),
2059                val: kawa::Store::from_vec(cookie_buf),
2060            }));
2061            debug_assert_eq!(
2062                response.blocks.len(),
2063                blocks_before + 1,
2064                "synthesising Set-Cookie must push exactly one block"
2065            );
2066        }
2067
2068        // Create a custom correlation header (defaults to "Sozu-Id", can be
2069        // renamed via the `sozu_id_header` listener config knob).
2070        let blocks_before_sozu_id = response.blocks.len();
2071        response.push_block(kawa::Block::Header(kawa::Pair {
2072            key: correlation_header_name(&self.sozu_id_header),
2073            val: kawa::Store::Shared(self.shared_id(), 0),
2074        }));
2075        debug_assert_eq!(
2076            response.blocks.len(),
2077            blocks_before_sozu_id + 1,
2078            "the Sozu-Id correlation header must be pushed exactly once"
2079        );
2080
2081        // Postcondition: response editing only added or elided headers, so
2082        // the block count never decreased.
2083        debug_assert!(
2084            response.blocks.len() >= blocks_at_entry,
2085            "header editing must never drop a block from the response"
2086        );
2087    }
2088
2089    /// The connection's forwarding values for the current inputs, rendered
2090    /// only when the cached ones were rendered from different inputs — the
2091    /// first request of a connection, or an input changed since.
2092    fn forwarding_hop(&mut self, proto: &str) -> &ForwardingHop {
2093        let (protocol, public_address, session_address) =
2094            (self.protocol, self.public_address, self.session_address);
2095        let cached = self
2096            .forwarding_hop
2097            .as_ref()
2098            .is_some_and(|hop| hop.renders(protocol, public_address, session_address));
2099        if !cached {
2100            self.forwarding_hop = Some(ForwardingHop::render(
2101                protocol,
2102                proto,
2103                public_address,
2104                session_address,
2105            ));
2106        }
2107        let hop = self
2108            .forwarding_hop
2109            .as_ref()
2110            .expect("the forwarding values were just rendered or reused");
2111        // Post: the values handed out were rendered from the current inputs.
2112        debug_assert!(
2113            hop.renders(protocol, public_address, session_address),
2114            "the forwarding values must match the current inputs"
2115        );
2116        hop
2117    }
2118
2119    /// `self.id` rendered, as a shared value for the correlation header.
2120    ///
2121    /// When the request carried no `X-Request-Id`, `on_request_headers`
2122    /// generated one from `self.id` and kept it in `self.x_request_id`: the
2123    /// same 26 bytes, shared with a reference-count bump. A client-supplied
2124    /// `X-Request-Id` is a different value, and `self.id` is then rendered
2125    /// into a copy of its own, because the correlation header always carries
2126    /// Sōzu's id, whatever the client sent.
2127    fn shared_id(&self) -> Rc<[u8]> {
2128        let rendered = render_ulid(self.id);
2129        let shared = match &self.x_request_id {
2130            Some(generated) if generated.as_bytes() == rendered => {
2131                Rc::<[u8]>::from(generated.clone())
2132            }
2133            _ => Rc::from(&rendered[..]),
2134        };
2135        debug_assert_eq!(
2136            &*shared, &rendered,
2137            "the correlation header must carry this request's own id"
2138        );
2139        shared
2140    }
2141
2142    /// Police the trailer fields that the last `kawa::h1::parse` appended
2143    /// to the trailer section of a chunked H1 request.
2144    ///
2145    /// - A field in [`TRAILER_SPOOF_VECTOR_HEADERS`] (client attribution,
2146    ///   sozu-proxy/sozu#1689) or in [`TRAILER_FORBIDDEN_FIELDS`] (framing,
2147    ///   routing, request modifiers, authentication, content processing and
2148    ///   connection-specific fields, RFC 9110 §6.5.1, sozu-proxy/sozu#1701)
2149    ///   is elided, and counted in `http.trailer.spoof_vector_elided` or
2150    ///   `http.trailer.forbidden_field_elided`.
2151    /// - Every field, elided or not, counts against `max_trailer_fields`,
2152    ///   as `pkawa::handle_trailer` (`lib/src/protocol/mux/pkawa.rs`) counts
2153    ///   an H2 trailer block. The field that goes over the limit marks the
2154    ///   request in error, so the caller's existing parse-error path answers
2155    ///   400 (or cuts a response that already started), and
2156    ///   `http.trailer.field_limit_exceeded` is incremented.
2157    ///
2158    /// kawa's H1 parser has no trailer callback: its `ParsingPhase::Trailers`
2159    /// arm pushes each trailer field as a `Block::Header` after the `Flags`
2160    /// block that carries `end_body` (the last chunk), then closes the
2161    /// section with a `Flags` block carrying `end_header` and `end_stream`.
2162    /// Call this after each `kawa::h1::parse` of a frontend request, with
2163    /// `first_new_block` set to `kawa.blocks.len()` read just before that
2164    /// parse. It walks back from the end of the queue over the closing block
2165    /// and the trailer fields, and stops at the first other block (the
2166    /// `end_body` marker) or at `first_new_block`. It never reaches the
2167    /// header block, which ends with its own `end_header` marker before the
2168    /// chunks.
2169    ///
2170    /// Stopping at `first_new_block` keeps the total walk linear in the
2171    /// number of trailer fields: every field queued before that parse was
2172    /// already examined, and counted in `trailer_fields`, by the call that
2173    /// followed the parse which queued it, so a client that trickles one
2174    /// trailer line per segment while the backend is not writable does not
2175    /// make each call re-walk the whole section. Parsing only appends
2176    /// blocks, and `prepare` only drains them from the front between parses,
2177    /// so the fields past `first_new_block` are exactly the new ones even
2178    /// when earlier trailer fields were already forwarded. That is also why
2179    /// the count lives in `trailer_fields` rather than in the queue.
2180    ///
2181    /// An elided field is skipped by kawa's H1 converter and by the H2 one,
2182    /// so the last chunk and the closing empty line are still written: the
2183    /// chunk framing stays valid even when every trailer field is dropped.
2184    /// A request that is not chunked, or has not reached its trailer
2185    /// section, returns before touching a block, so the walk costs nothing
2186    /// and allocates nothing on the common path.
2187    ///
2188    /// Only requests are filtered: a response trailer travels towards the
2189    /// client, which does not take client attribution from it.
2190    pub fn filter_request_trailers(
2191        &mut self,
2192        kawa: &mut GenericHttpStream,
2193        first_new_block: usize,
2194    ) -> TrailerFilterOutcome {
2195        let mut outcome = TrailerFilterOutcome::default();
2196        if !matches!(kawa.kind, kawa::Kind::Request)
2197            || kawa.body_size != kawa::BodySize::Chunked
2198            || !matches!(
2199                kawa.parsing_phase,
2200                kawa::ParsingPhase::Trailers | kawa::ParsingPhase::Terminated
2201            )
2202        {
2203            return outcome;
2204        }
2205        // Pre: the caller read the length before a parse, which only appends.
2206        debug_assert!(
2207            first_new_block <= kawa.blocks.len(),
2208            "first_new_block must be a block count read before the parse"
2209        );
2210        // Pre: the count never passes the limit, the field that would have
2211        // passed it turned the request into an error, which stops parsing.
2212        debug_assert!(
2213            self.trailer_fields <= self.max_trailer_fields,
2214            "an admitted trailer section fits max_trailer_fields"
2215        );
2216        let first_new_block = first_new_block.min(kawa.blocks.len());
2217        let buf = kawa.storage.buffer();
2218        let mut fields: u32 = 0;
2219        for block in kawa.blocks.range_mut(first_new_block..).rev() {
2220            match block {
2221                kawa::Block::Flags(kawa::Flags {
2222                    end_body: false,
2223                    end_chunk: false,
2224                    end_header: true,
2225                    end_stream: true,
2226                }) => {}
2227                kawa::Block::Header(pair) => {
2228                    fields = fields.saturating_add(1);
2229                    if pair.is_elided() {
2230                        continue;
2231                    }
2232                    let key = pair.key.data(buf);
2233                    if is_trailer_spoof_vector(key) {
2234                        pair.elide();
2235                        outcome.spoof_vectors += 1;
2236                        incr!(names::http::TRAILER_SPOOF_VECTOR_ELIDED);
2237                    } else if is_trailer_forbidden_field(key) {
2238                        pair.elide();
2239                        outcome.forbidden += 1;
2240                        incr!(names::http::TRAILER_FORBIDDEN_FIELD_ELIDED);
2241                    }
2242                    // Post: no spoof-vector or forbidden field survives.
2243                    debug_assert!(
2244                        pair.is_elided()
2245                            || !(is_trailer_spoof_vector(pair.key.data(buf))
2246                                || is_trailer_forbidden_field(pair.key.data(buf))),
2247                        "a spoof-vector or forbidden trailer field must be elided"
2248                    );
2249                }
2250                _ => break,
2251            }
2252        }
2253        self.trailer_fields = self.trailer_fields.saturating_add(fields);
2254        if self.trailer_fields > self.max_trailer_fields {
2255            outcome.over_limit = true;
2256            incr!(names::http::TRAILER_FIELD_LIMIT_EXCEEDED);
2257            kawa.parsing_phase
2258                .error("trailer section exceeds max_trailer_fields".into());
2259        }
2260        // Post: only blocks appended by the last parse are touched, and a
2261        // section over the limit never leaves the request forwardable.
2262        debug_assert!(
2263            outcome.spoof_vectors + outcome.forbidden <= kawa.blocks.len() - first_new_block,
2264            "only blocks appended by the last parse are elided"
2265        );
2266        debug_assert!(
2267            !outcome.over_limit || kawa.is_error(),
2268            "a trailer section over the limit marks the request in error"
2269        );
2270        outcome
2271    }
2272
2273    /// Prepare this context for the next request of a keep-alive connection,
2274    /// which `request_id` identifies.
2275    ///
2276    /// The request id is request-scoped like the rest of what this clears:
2277    /// `Sozu-Id`, a generated `X-Request-Id`, `%REQUEST_ID` and the access
2278    /// log's `request_id` all read `self.id`, so keeping it would give every
2279    /// request of the connection the first one's id. The caller mints it —
2280    /// `ConnectionH1` from `Context::next_request_id`
2281    /// (`lib/src/protocol/mux/mod.rs`), the same source that numbers H2
2282    /// streams — because this struct owns neither a clock nor an RNG.
2283    pub fn reset(&mut self, request_id: Ulid) {
2284        // Snapshot the connection-scoped identity + TLS/listener fields that
2285        // reset() must NOT touch (set once at handshake, reused across every
2286        // keep-alive request). Cheap to copy — all `Copy`. Read only inside
2287        // the postcondition `debug_assert!`s below, so dead code in release.
2288        let session_id_before = self.session_id;
2289        let id_before = self.id;
2290        debug_assert_ne!(
2291            request_id, id_before,
2292            "the next request of a connection needs an id of its own"
2293        );
2294        let strict_sni_before = self.strict_sni_binding;
2295        let elide_before = self.elide_x_real_ip;
2296        let send_before = self.send_x_real_ip;
2297        let forwarded_headers_before = self.forwarded_headers;
2298        let max_trailer_fields_before = self.max_trailer_fields;
2299        let tls_version_before = self.tls_version;
2300        let tls_cipher_before = self.tls_cipher;
2301        let tls_alpn_before = self.tls_alpn;
2302        let forwarding_hop_before = self
2303            .forwarding_hop
2304            .as_ref()
2305            .map(|hop| Rc::as_ptr(&hop.port));
2306
2307        self.id = request_id;
2308        self.keep_alive_backend = true;
2309        self.keep_alive_frontend = true;
2310        self.sticky_session_found = None;
2311        // The sticky-session answer is request-scoped (#1822):
2312        // `backend_from_request` writes it only when the request's frontend
2313        // sticks, and `on_response_headers` answers with a `Set-Cookie` for
2314        // any value left here, so keeping it would hand a request to a
2315        // frontend that does not stick the previous request's cookie.
2316        self.sticky_session = None;
2317        self.affinity_key = None;
2318        self.cluster_metrics_incarnation = Default::default();
2319        self.method = None;
2320        self.authority = None;
2321        self.path = None;
2322        self.status = None;
2323        self.reason = None;
2324        self.user_agent = None;
2325        self.x_request_id = None;
2326        self.xff_chain = None;
2327        self.redirect_location = None;
2328        self.www_authenticate = None;
2329        self.original_authority = None;
2330        self.headers_response.clear();
2331        self.trailer_fields = 0;
2332        self.backends_unavailable = false;
2333        // Note: tls_server_name, tls_version, tls_cipher, tls_alpn,
2334        // strict_sni_binding, elide_x_real_ip, send_x_real_ip,
2335        // forwarded_headers and max_trailer_fields are
2336        // connection-scoped — set once at handshake completion and reused
2337        // across every keep-alive request, so reset() intentionally leaves
2338        // them in place. So is forwarding_hop, rendered from connection-
2339        // scoped inputs only.
2340
2341        // Post: request-scoped state is fully cleared (a stale value here
2342        // would leak across pipelined requests on the same connection).
2343        debug_assert!(
2344            self.method.is_none()
2345                && self.sticky_session.is_none()
2346                && self.authority.is_none()
2347                && self.path.is_none()
2348                && self.status.is_none()
2349                && self.x_request_id.is_none()
2350                && self.headers_response.is_empty()
2351                && self.trailer_fields == 0
2352                && !self.backends_unavailable,
2353            "reset() must clear all request-scoped state"
2354        );
2355        debug_assert!(
2356            self.keep_alive_backend && self.keep_alive_frontend,
2357            "reset() must restore keep-alive to its optimistic default"
2358        );
2359        // Post: connection-scoped identity + TLS/listener knobs are untouched.
2360        debug_assert_eq!(
2361            self.session_id, session_id_before,
2362            "reset() must preserve the connection session id"
2363        );
2364        debug_assert!(
2365            self.id == request_id && self.id != id_before,
2366            "reset() must give the next request its own id"
2367        );
2368        debug_assert!(
2369            self.strict_sni_binding == strict_sni_before
2370                && self.elide_x_real_ip == elide_before
2371                && self.send_x_real_ip == send_before
2372                && self.forwarded_headers == forwarded_headers_before
2373                && self.max_trailer_fields == max_trailer_fields_before
2374                && self.tls_version == tls_version_before
2375                && self.tls_cipher == tls_cipher_before
2376                && self.tls_alpn == tls_alpn_before,
2377            "reset() must preserve connection-scoped TLS/listener knobs"
2378        );
2379        debug_assert_eq!(
2380            self.forwarding_hop
2381                .as_ref()
2382                .map(|hop| Rc::as_ptr(&hop.port)),
2383            forwarding_hop_before,
2384            "reset() must keep the connection's forwarding values"
2385        );
2386    }
2387
2388    pub fn extract_route(&self) -> Result<(&str, &str, &Method), RetrieveClusterError> {
2389        let given_method = self.method.as_ref().ok_or(RetrieveClusterError::NoMethod)?;
2390        let given_authority = self
2391            .authority
2392            .as_deref()
2393            .ok_or(RetrieveClusterError::NoHost)?;
2394        let given_path = self.path.as_deref().ok_or(RetrieveClusterError::NoPath)?;
2395
2396        // Post: the triple is returned in (authority, path, method) order —
2397        // pin it against a future field-swap regression that would route to
2398        // the wrong cluster. (Both fields can be empty strings on the wire,
2399        // so we assert the mapping, not non-emptiness.)
2400        debug_assert!(
2401            std::ptr::eq(given_authority, self.authority.as_deref().unwrap())
2402                && std::ptr::eq(given_path, self.path.as_deref().unwrap()),
2403            "extract_route must return (authority, path, method) in order"
2404        );
2405        Ok((given_authority, given_path, given_method))
2406    }
2407
2408    pub fn get_route(&self) -> String {
2409        if let Some(method) = &self.method {
2410            if let Some(authority) = &self.authority {
2411                if let Some(path) = &self.path {
2412                    return format!("{method} {authority}{path}");
2413                }
2414                return format!("{method} {authority}");
2415            }
2416            return format!("{method}");
2417        }
2418        String::new()
2419    }
2420
2421    pub fn websocket_context(&self) -> WebSocketContext {
2422        WebSocketContext::Http {
2423            method: self.method.clone(),
2424            authority: self.authority.clone(),
2425            path: self.path.clone(),
2426            reason: self.reason.as_deref().map(ToOwned::to_owned),
2427            status: self.status,
2428        }
2429    }
2430
2431    pub fn log_context(&self) -> LogContext<'_> {
2432        let ctx = LogContext {
2433            session_id: self.session_id,
2434            request_id: Some(self.id),
2435            cluster_id: self.cluster_id.as_deref(),
2436            backend_id: self.backend_id.as_deref(),
2437        };
2438        // The access-log bracket `[session req cluster backend]` is keyed on
2439        // session + request id; both must always be present so the log line
2440        // is correlatable. cluster/backend are legitimately absent before
2441        // routing, so they are not asserted here.
2442        debug_assert!(
2443            ctx.request_id.is_some(),
2444            "log_context must always carry the request id for correlation"
2445        );
2446        ctx
2447    }
2448}
2449
2450#[cfg(test)]
2451mod tests {
2452    use super::*;
2453    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
2454
2455    /// Helper to create a minimal HttpContext for testing.
2456    fn make_context() -> HttpContext {
2457        HttpContext::new(
2458            Ulid::generate(),
2459            Ulid::generate(),
2460            Protocol::HTTP,
2461            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
2462            Some(SocketAddr::new(
2463                IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
2464                54321,
2465            )),
2466            "SERVERID".to_owned(),
2467            "Sozu-Id".to_owned(),
2468            false,
2469            false,
2470        )
2471    }
2472
2473    // ── sozu_id_header ──────────────────────────────────────────────────
2474
2475    #[test]
2476    fn test_sozu_id_header_default_name_stored_on_context() {
2477        // The make_context helper uses the documented default "Sozu-Id" to
2478        // match the trait default on `L7ListenerHandler::get_sozu_id_header`.
2479        let ctx = make_context();
2480        assert_eq!(ctx.sozu_id_header, "Sozu-Id");
2481    }
2482
2483    #[test]
2484    fn test_sozu_id_header_custom_name_stored_on_context() {
2485        // Operator-provided rename is carried verbatim onto the HttpContext.
2486        let ctx = HttpContext::new(
2487            Ulid::generate(),
2488            Ulid::generate(),
2489            Protocol::HTTP,
2490            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
2491            None,
2492            "SERVERID".to_owned(),
2493            "X-Edge-Id".to_owned(),
2494            false,
2495            false,
2496        );
2497        assert_eq!(ctx.sozu_id_header, "X-Edge-Id");
2498    }
2499
2500    // ── extract_route ──────────────────────────────────────────────────
2501
2502    #[test]
2503    fn test_extract_route_all_present() {
2504        let mut ctx = make_context();
2505        ctx.method = Some(Method::Get);
2506        ctx.authority = Some("example.com".to_owned());
2507        ctx.path = Some("/index.html".to_owned());
2508
2509        let (authority, path, method) = ctx.extract_route().unwrap();
2510        assert_eq!(authority, "example.com");
2511        assert_eq!(path, "/index.html");
2512        assert_eq!(method, &Method::Get);
2513    }
2514
2515    #[test]
2516    fn test_extract_route_no_method() {
2517        let mut ctx = make_context();
2518        ctx.authority = Some("example.com".to_owned());
2519        ctx.path = Some("/".to_owned());
2520
2521        let err = ctx.extract_route().unwrap_err();
2522        assert!(matches!(err, RetrieveClusterError::NoMethod));
2523    }
2524
2525    #[test]
2526    fn test_extract_route_no_host() {
2527        let mut ctx = make_context();
2528        ctx.method = Some(Method::Get);
2529        ctx.path = Some("/".to_owned());
2530
2531        let err = ctx.extract_route().unwrap_err();
2532        assert!(matches!(err, RetrieveClusterError::NoHost));
2533    }
2534
2535    #[test]
2536    fn test_extract_route_no_path() {
2537        let mut ctx = make_context();
2538        ctx.method = Some(Method::Get);
2539        ctx.authority = Some("example.com".to_owned());
2540
2541        let err = ctx.extract_route().unwrap_err();
2542        assert!(matches!(err, RetrieveClusterError::NoPath));
2543    }
2544
2545    // ── get_route ──────────────────────────────────────────────────────
2546
2547    #[test]
2548    fn test_get_route_all_present() {
2549        let mut ctx = make_context();
2550        ctx.method = Some(Method::Get);
2551        ctx.authority = Some("example.com".to_owned());
2552        ctx.path = Some("/api/v1".to_owned());
2553
2554        assert_eq!(ctx.get_route(), "GET example.com/api/v1");
2555    }
2556
2557    #[test]
2558    fn test_get_route_method_and_authority_only() {
2559        let mut ctx = make_context();
2560        ctx.method = Some(Method::Post);
2561        ctx.authority = Some("example.com".to_owned());
2562
2563        assert_eq!(ctx.get_route(), "POST example.com");
2564    }
2565
2566    #[test]
2567    fn test_get_route_method_only() {
2568        let mut ctx = make_context();
2569        ctx.method = Some(Method::Delete);
2570
2571        assert_eq!(ctx.get_route(), "DELETE");
2572    }
2573
2574    #[test]
2575    fn test_get_route_empty() {
2576        let ctx = make_context();
2577        assert_eq!(ctx.get_route(), "");
2578    }
2579
2580    // ── reset ──────────────────────────────────────────────────────────
2581
2582    #[test]
2583    fn test_reset_clears_request_response_state() {
2584        let mut ctx = make_context();
2585        ctx.keep_alive_backend = false;
2586        ctx.keep_alive_frontend = false;
2587        ctx.sticky_session_found = Some("abc123".to_owned());
2588        ctx.method = Some(Method::Post);
2589        ctx.authority = Some("example.com".to_owned());
2590        ctx.path = Some("/upload".to_owned());
2591        ctx.status = Some(200);
2592        ctx.reason = Some(Cow::Owned("Fine".to_owned()));
2593        ctx.user_agent = Some("curl/7.81".to_owned());
2594        ctx.x_request_id = Some(Rc::from("client-xrid-123"));
2595        ctx.xff_chain = Some("203.0.113.5, 198.51.100.10".to_owned());
2596        ctx.redirect_location = Some("https://example.com/".to_owned());
2597        ctx.www_authenticate = Some("Basic realm=\"sozu\"".to_owned());
2598        ctx.original_authority = Some("old.example.com".to_owned());
2599        ctx.headers_response.push(HeaderEditSnapshot {
2600            key: b"X-Cache".to_vec(),
2601            val: b"HIT".to_vec(),
2602            mode: HeaderEditMode::Append,
2603        });
2604
2605        ctx.reset(Ulid::generate());
2606
2607        assert!(ctx.keep_alive_backend);
2608        assert!(ctx.keep_alive_frontend);
2609        assert!(ctx.sticky_session_found.is_none());
2610        assert!(ctx.method.is_none());
2611        assert!(ctx.authority.is_none());
2612        assert!(ctx.path.is_none());
2613        assert!(ctx.status.is_none());
2614        assert!(ctx.reason.is_none());
2615        assert!(ctx.user_agent.is_none());
2616        assert!(ctx.x_request_id.is_none());
2617        assert!(ctx.xff_chain.is_none());
2618        // The four stash slots written by the routing layer must clear
2619        // between pipelined H1 requests; otherwise a future code path that
2620        // emits a 301 / 401 default-answer without re-routing would
2621        // inherit a stale Location / WWW-Authenticate from a prior request,
2622        // or the backend would receive a stale X-Forwarded-Host or a stale
2623        // response-side header edit.
2624        assert!(ctx.redirect_location.is_none());
2625        assert!(ctx.www_authenticate.is_none());
2626        assert!(ctx.original_authority.is_none());
2627        assert!(ctx.headers_response.is_empty());
2628    }
2629
2630    #[test]
2631    fn test_reset_preserves_tls_metadata() {
2632        // TLS metadata is connection-scoped (set once at handshake, reused
2633        // across every keep-alive request) — reset() must leave it intact
2634        // so the access log of the second request still carries it.
2635        let mut ctx = make_context();
2636        ctx.tls_server_name = Some("example.com".to_owned());
2637        ctx.tls_version = Some("TLSv1.3");
2638        ctx.tls_cipher = Some("TLS_AES_128_GCM_SHA256");
2639        ctx.tls_alpn = Some("h2");
2640        ctx.strict_sni_binding = false;
2641
2642        ctx.reset(Ulid::generate());
2643
2644        assert_eq!(ctx.tls_server_name.as_deref(), Some("example.com"));
2645        assert_eq!(ctx.tls_version, Some("TLSv1.3"));
2646        assert_eq!(ctx.tls_cipher, Some("TLS_AES_128_GCM_SHA256"));
2647        assert_eq!(ctx.tls_alpn, Some("h2"));
2648        assert!(!ctx.strict_sni_binding);
2649    }
2650
2651    #[test]
2652    fn test_reset_preserves_connection_state() {
2653        let mut ctx = make_context();
2654        ctx.closing = true;
2655        ctx.cluster_id = Some("cluster-1".into());
2656        ctx.backend_id = Some("backend-1".into());
2657        // `sticky_session` is request-scoped, not connection state (#1822):
2658        // `reset_clears_the_sticky_session_answer_of_the_previous_request`
2659        // pins that it is cleared.
2660
2661        let original_id = ctx.id;
2662        let original_protocol = ctx.protocol;
2663        let original_public_address = ctx.public_address;
2664
2665        let next_id = Ulid::generate();
2666        ctx.reset(next_id);
2667
2668        // Connection-level state is preserved
2669        assert!(ctx.closing);
2670        assert_eq!(ctx.cluster_id.as_deref(), Some("cluster-1"));
2671        assert_eq!(ctx.backend_id.as_deref(), Some("backend-1"));
2672        // The request id is request-scoped: a keep-alive connection's next
2673        // request must not inherit the previous one's.
2674        assert_ne!(ctx.id, original_id);
2675        assert_eq!(ctx.id, next_id);
2676        assert_eq!(ctx.protocol, original_protocol);
2677        assert_eq!(ctx.public_address, original_public_address);
2678    }
2679
2680    /// The response a context's editor builds for `bytes`, serialised.
2681    fn edited_response(ctx: &mut HttpContext, bytes: &[u8]) -> String {
2682        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
2683        let mut kawa: GenericHttpStream = kawa::Kawa::new(
2684            kawa::Kind::Response,
2685            kawa::Buffer::new(
2686                pool.checkout()
2687                    .expect("the test pool must hand out a buffer"),
2688            ),
2689        );
2690        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
2691        kawa.storage.fill(bytes.len());
2692        kawa::h1::parse(&mut kawa, ctx);
2693        assert!(!kawa.is_error(), "premise: the response must parse");
2694        serialized_request(&mut kawa)
2695    }
2696
2697    /// #1822: the sticky-session cookie `backend_from_request` chose for one
2698    /// request of a keep-alive connection is not the next request's. That
2699    /// call writes `sticky_session` only when the request's frontend sticks,
2700    /// so a request to a frontend that does not stick, following one to a
2701    /// frontend that does, used to answer with the previous request's
2702    /// `Set-Cookie`, naming a backend of another cluster that frontend
2703    /// never asked for and cannot use.
2704    ///
2705    /// TO SEE THIS RED: drop `self.sticky_session = None;` from
2706    /// `HttpContext::reset`.
2707    #[test]
2708    fn reset_clears_the_sticky_session_answer_of_the_previous_request() {
2709        const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nb";
2710        let mut ctx = make_context();
2711        // Request A, to a sticky frontend whose client sent no cookie: the
2712        // router answers with the chosen backend's sticky id.
2713        ctx.sticky_session = Some("sticky-a".to_owned());
2714        let first = edited_response(&mut ctx, RESPONSE);
2715        assert!(
2716            first.contains("Set-Cookie: SERVERID=sticky-a; Path=/\r\n"),
2717            "premise: the sticky request is answered with its cookie, got {first:?}"
2718        );
2719
2720        // Request B, on the same connection, to a frontend that does not
2721        // stick: the router leaves `sticky_session` as `reset` left it.
2722        ctx.reset(Ulid::generate());
2723        assert_eq!(
2724            ctx.sticky_session, None,
2725            "reset must clear the previous request's sticky-session answer"
2726        );
2727        let second = edited_response(&mut ctx, RESPONSE);
2728        assert!(
2729            !second.to_ascii_lowercase().contains("set-cookie"),
2730            "a request to a frontend that does not stick must not carry the \
2731             previous request's cookie, got {second:?}"
2732        );
2733    }
2734
2735    // ── write_forwarded_for_by (RFC 7239 §6 IP-literal bracketing) ──────
2736    //
2737    // Locks in the IPv6 bracket+quote contract that matches HAProxy
2738    // (`_7239_print_ip6` in `src/http_ext.c`) and prevents regression to
2739    // the ambiguous `for="2001:db8::1:8080"` form flagged in issue #1254.
2740
2741    use std::net::Ipv6Addr;
2742
2743    fn render_for_by(peer: SocketAddr, public_ip: IpAddr) -> String {
2744        let mut buf = Vec::new();
2745        write_forwarded_for_by(&mut buf, peer.ip(), peer.port(), public_ip);
2746        String::from_utf8(buf).expect("forwarded fragment must be ASCII")
2747    }
2748
2749    fn render_suffix(proto: &str, peer: SocketAddr, public_ip: IpAddr) -> String {
2750        let mut buf = Vec::new();
2751        write_forwarded_suffix(&mut buf, proto, peer.ip(), peer.port(), public_ip);
2752        String::from_utf8(buf).expect("forwarded fragment must be ASCII")
2753    }
2754
2755    #[test]
2756    fn test_forwarded_ipv4_peer_ipv4_by_unchanged() {
2757        // IPv4 must keep the pre-fix wire format: unquoted `by=`, bare IPv4 in
2758        // `for=`. Sōzu emits `for=` quoted unconditionally (HAProxy emits it
2759        // quoted only when a port is attached; we always attach a port).
2760        let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), 54321);
2761        let public_ip = IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1));
2762        assert_eq!(
2763            render_for_by(peer, public_ip),
2764            r#";for="203.0.113.7:54321";by=198.51.100.1"#,
2765        );
2766    }
2767
2768    #[test]
2769    fn test_forwarded_ipv6_peer_brackets_disambiguate_port() {
2770        // Regression for issue #1254: without brackets, `for="2001:db8::1:8080"`
2771        // is ambiguous (could parse as `::1` + port `8080` or `:1:8080` literal).
2772        let peer = SocketAddr::new(IpAddr::V6("2001:db8::1".parse::<Ipv6Addr>().unwrap()), 8080);
2773        let public_ip = IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1));
2774        assert_eq!(
2775            render_for_by(peer, public_ip),
2776            r#";for="[2001:db8::1]:8080";by=198.51.100.1"#,
2777        );
2778    }
2779
2780    #[test]
2781    fn test_forwarded_ipv6_public_address_bracketed_and_quoted() {
2782        // `by=` carries no port but RFC 7239 §6 still requires the IP-literal
2783        // brackets for IPv6, and the value must be quoted because the literal
2784        // contains `:`.
2785        let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), 54321);
2786        let public_ip = IpAddr::V6("2001:db8::2".parse::<Ipv6Addr>().unwrap());
2787        assert_eq!(
2788            render_for_by(peer, public_ip),
2789            r#";for="203.0.113.7:54321";by="[2001:db8::2]""#,
2790        );
2791    }
2792
2793    #[test]
2794    fn test_forwarded_ipv6_peer_and_public_both_bracketed() {
2795        let peer = SocketAddr::new(IpAddr::V6("2001:db8::1".parse::<Ipv6Addr>().unwrap()), 8080);
2796        let public_ip = IpAddr::V6("2001:db8::2".parse::<Ipv6Addr>().unwrap());
2797        assert_eq!(
2798            render_for_by(peer, public_ip),
2799            r#";for="[2001:db8::1]:8080";by="[2001:db8::2]""#,
2800        );
2801    }
2802
2803    #[test]
2804    fn test_forwarded_suffix_prepends_proto_and_brackets_ipv6() {
2805        // The suffix form is what we append when an inbound `Forwarded`
2806        // header already exists. Same bracketing contract must hold.
2807        let peer = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 443);
2808        let public_ip = IpAddr::V6("fe80::1".parse::<Ipv6Addr>().unwrap());
2809        assert_eq!(
2810            render_suffix("https", peer, public_ip),
2811            r#", proto=https;for="[::1]:443";by="[fe80::1]""#,
2812        );
2813    }
2814
2815    // ── traceparent / opentelemetry helpers ─────────────────────────────
2816
2817    #[cfg(feature = "opentelemetry")]
2818    mod otel {
2819        use super::super::*;
2820
2821        #[test]
2822        fn test_parse_hex_valid() {
2823            let (val, rest) = parse_hex::<4>(b"abcd1234").unwrap();
2824            assert_eq!(&val, b"abcd");
2825            assert_eq!(rest, b"1234");
2826        }
2827
2828        #[test]
2829        fn test_parse_hex_exact_length() {
2830            let (val, rest) = parse_hex::<8>(b"01234567").unwrap();
2831            assert_eq!(&val, b"01234567");
2832            assert!(rest.is_empty());
2833        }
2834
2835        #[test]
2836        fn test_parse_hex_too_short() {
2837            assert!(parse_hex::<4>(b"ab").is_none());
2838        }
2839
2840        #[test]
2841        fn test_parse_hex_rejects_non_hex() {
2842            assert!(parse_hex::<4>(b"ghij").is_none());
2843        }
2844
2845        #[test]
2846        fn test_parse_hex_rejects_uppercase_is_ok() {
2847            // Uppercase hex digits are valid
2848            let (val, _) = parse_hex::<4>(b"ABCD").unwrap();
2849            assert_eq!(&val, b"ABCD");
2850        }
2851
2852        #[test]
2853        fn test_skip_separator_valid() {
2854            let rest = skip_separator(b"-hello").unwrap();
2855            assert_eq!(rest, b"hello");
2856        }
2857
2858        #[test]
2859        fn test_skip_separator_wrong_char() {
2860            assert!(skip_separator(b"+hello").is_none());
2861        }
2862
2863        #[test]
2864        fn test_skip_separator_empty() {
2865            assert!(skip_separator(b"").is_none());
2866        }
2867
2868        #[test]
2869        fn test_build_traceparent_format() {
2870            let trace_id: [u8; 32] = *b"4bf92f3577b34da6a3ce929d0e0e4736";
2871            let parent_id: [u8; 16] = *b"00f067aa0ba902b7";
2872
2873            let result = build_traceparent(&trace_id, &parent_id);
2874            assert_eq!(
2875                &result,
2876                b"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
2877            );
2878        }
2879
2880        #[test]
2881        fn test_build_traceparent_length() {
2882            let trace_id = [b'a'; 32];
2883            let parent_id = [b'b'; 16];
2884            let result = build_traceparent(&trace_id, &parent_id);
2885            // Format: "00-" (3) + trace_id (32) + "-" (1) + parent_id (16) + "-01" (3) = 55
2886            assert_eq!(result.len(), 55);
2887        }
2888
2889        #[test]
2890        fn test_parse_traceparent_valid() {
2891            let input = b"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
2892            let store = kawa::Store::Static(input);
2893            let (trace_id, parent_id) = parse_traceparent(&store, input).unwrap();
2894            assert_eq!(&trace_id, b"4bf92f3577b34da6a3ce929d0e0e4736");
2895            assert_eq!(&parent_id, b"00f067aa0ba902b7");
2896        }
2897
2898        #[test]
2899        fn test_parse_traceparent_sampled_flag_zero() {
2900            let input = b"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00";
2901            let store = kawa::Store::Static(input);
2902            let result = parse_traceparent(&store, input);
2903            assert!(result.is_some());
2904        }
2905
2906        #[test]
2907        fn test_parse_traceparent_wrong_version() {
2908            let input = b"01-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
2909            let store = kawa::Store::Static(input);
2910            assert!(parse_traceparent(&store, input).is_none());
2911        }
2912
2913        #[test]
2914        fn test_parse_traceparent_too_short() {
2915            let input = b"00-4bf9";
2916            let store = kawa::Store::Static(input);
2917            assert!(parse_traceparent(&store, input).is_none());
2918        }
2919
2920        #[test]
2921        fn test_parse_traceparent_trailing_data() {
2922            // Extra characters after the trace-flags should cause rejection
2923            let input = b"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01-extra";
2924            let store = kawa::Store::Static(input);
2925            assert!(parse_traceparent(&store, input).is_none());
2926        }
2927
2928        #[test]
2929        fn test_parse_traceparent_missing_separator() {
2930            let input = b"004bf92f3577b34da6a3ce929d0e0e473600f067aa0ba902b701";
2931            let store = kawa::Store::Static(input);
2932            assert!(parse_traceparent(&store, input).is_none());
2933        }
2934
2935        #[test]
2936        fn test_parse_build_roundtrip() {
2937            let trace_id: [u8; 32] = *b"4bf92f3577b34da6a3ce929d0e0e4736";
2938            let parent_id: [u8; 16] = *b"00f067aa0ba902b7";
2939
2940            // Build a traceparent from known IDs
2941            let built = build_traceparent(&trace_id, &parent_id);
2942
2943            // Verify that the built value matches the expected static string,
2944            // then parse that static string back to confirm roundtrip.
2945            let expected: &[u8] = b"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
2946            assert_eq!(&built[..], expected);
2947
2948            let store = kawa::Store::Static(expected);
2949            let (parsed_trace_id, parsed_parent_id) = parse_traceparent(&store, expected).unwrap();
2950
2951            assert_eq!(parsed_trace_id, trace_id);
2952            assert_eq!(parsed_parent_id, parent_id);
2953        }
2954    }
2955
2956    // ── Content-Length field value: 1*DIGIT ────────────────────────────
2957
2958    /// Parse `bytes` as a `kind` message through the real parser, and so
2959    /// through `on_request_headers` / `on_response_headers`, on a fresh
2960    /// context — the way `ConnectionH1::readable` drives it.
2961    fn parse_framed(
2962        pool: &mut crate::pool::Pool,
2963        kind: kawa::Kind,
2964        bytes: &[u8],
2965    ) -> GenericHttpStream {
2966        let mut kawa: GenericHttpStream = kawa::Kawa::new(
2967            kind,
2968            kawa::Buffer::new(
2969                pool.checkout()
2970                    .expect("the test pool must hand out a buffer"),
2971            ),
2972        );
2973        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
2974        kawa.storage.fill(bytes.len());
2975        kawa::h1::parse(&mut kawa, &mut make_context());
2976        kawa
2977    }
2978
2979    /// RFC 9110 §8.6: `Content-Length = 1*DIGIT`, and a sender MUST NOT
2980    /// forward a message whose Content-Length does not match that grammar.
2981    /// kawa 0.7.1 read the value with `usize::from_str`, which also accepts
2982    /// one leading `+`: `Content-Length: +5` framed a 5-byte body and was
2983    /// forwarded verbatim, handing any backend that refuses or re-reads
2984    /// that spelling a body it would take for the next request (CWE-444).
2985    ///
2986    /// kawa >= 0.7.2 refuses every row here itself (`Invalid Content-Length
2987    /// field value`, CleverCloud/kawa#26) before `on_request_headers` runs;
2988    /// under kawa 0.7.1 the `+` rows reached it. The test pins the outcome —
2989    /// the request is refused — whichever layer refuses it, and that the
2990    /// Sōzu clause neither relaxes nor depends on kawa's refusal.
2991    ///
2992    /// TO SEE THIS RED: deleting the Content-Length clause of
2993    /// `on_request_headers` alone no longer turns it red, since kawa >= 0.7.2
2994    /// refuses every row first. Either see the clause's own predicate fail in
2995    /// `the_content_length_helper_judges_every_non_digit_value`, or pin
2996    /// kawa 0.7.1 (`kawa = { version = "=0.7.1", default-features = false }`
2997    /// in the root `Cargo.toml`, then `cargo update -p kawa --precise 0.7.1`)
2998    /// and delete the clause: the `plus`, `plus-zero` and
2999    /// `plus-then-canonical` rows then parse clean. Under that pin the
3000    /// `canonical-then-plus` assertion fails even with the clause present:
3001    /// that refusal is kawa's alone, since kawa 0.7.1 elided the `+5` line.
3002    #[test]
3003    fn a_request_content_length_that_is_not_only_digits_is_rejected() {
3004        let cases: [(&str, &[u8]); 13] = [
3005            ("plus", b"+5"),
3006            ("plus-zero", b"+0"),
3007            ("minus-zero", b"-0"),
3008            ("plus-alone", b"+"),
3009            ("list-of-equal-values", b"5, 5"),
3010            ("list-without-space", b"5,5"),
3011            ("inner-space", b"5 5"),
3012            ("hexadecimal", b"0x5"),
3013            ("decimal-point", b"5.0"),
3014            ("overflow", b"99999999999999999999999"),
3015            ("arabic-indic-digit", "\u{0665}".as_bytes()),
3016            ("fullwidth-digit", "\u{FF15}".as_bytes()),
3017            ("empty", b""),
3018        ];
3019        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3020        for (label, value) in cases {
3021            let request = [
3022                &b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: "[..],
3023                value,
3024                b"\r\n\r\nHello",
3025            ]
3026            .concat();
3027            let kawa = parse_framed(&mut pool, kawa::Kind::Request, &request);
3028            assert!(
3029                kawa.is_error(),
3030                "{label}: a Content-Length that is not 1*DIGIT must be refused, got {:?} with {:?}",
3031                kawa.parsing_phase,
3032                kawa.body_size
3033            );
3034        }
3035
3036        // kawa elides a second Content-Length equal to the first, so the
3037        // FIRST line is the one that survives to be forwarded — and so the
3038        // one the guard judges.
3039        let request = b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: +5\r\nContent-Length: 5\r\n\r\nHello";
3040        let kawa = parse_framed(&mut pool, kawa::Kind::Request, request);
3041        assert!(
3042            kawa.is_error(),
3043            "plus-then-canonical: the surviving `+5` must be refused, got {:?}",
3044            kawa.parsing_phase
3045        );
3046        // Hand the pool's only buffer back for the next parse.
3047        drop(kawa);
3048
3049        // The mirror order. kawa 0.7.1 parsed the second line as 5, elided
3050        // it as equal to the first and forwarded only `5`. kawa >= 0.7.2
3051        // judges every line against `1*DIGIT` before comparing it, so the
3052        // message carries an invalid Content-Length field line and is
3053        // refused (RFC 9112 §6.3 rule 5) — the same verdict as the order
3054        // above, no longer one that depends on which line comes first.
3055        let request = b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\nContent-Length: +5\r\n\r\nHello";
3056        let kawa = parse_framed(&mut pool, kawa::Kind::Request, request);
3057        assert!(
3058            kawa.is_error(),
3059            "canonical-then-plus: a `+5` field line must be refused, got {:?}",
3060            kawa.parsing_phase
3061        );
3062    }
3063
3064    /// `has_non_digit_content_length` on its own, over hand-built header
3065    /// blocks. kawa refuses `-0`, `0x5`, `5 5` and the empty value before
3066    /// `on_request_headers` runs — and, since 0.7.2, `+5` too — so the
3067    /// parse-driven tests above prove kawa's refusal, not the helper's: a
3068    /// helper narrowed to a leading `+`, or not called at all, would still
3069    /// pass them. This test is the one that pins the helper's own `1*DIGIT`
3070    /// contract, and that an elided line is not judged.
3071    ///
3072    /// TO SEE THIS RED, narrow the helper's predicate to
3073    /// `val.first() == Some(&b'+')`: the `-0`, `0x5`, `5 5` and empty rows
3074    /// then read `false`.
3075    #[test]
3076    fn the_content_length_helper_judges_every_non_digit_value() {
3077        fn content_length(val: &'static [u8], elided: bool) -> GenericHttpStream {
3078            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3079            let mut kawa: GenericHttpStream = kawa::Kawa::new(
3080                kawa::Kind::Request,
3081                kawa::Buffer::new(
3082                    pool.checkout()
3083                        .expect("the test pool must hand out a buffer"),
3084                ),
3085            );
3086            let mut pair = kawa::Pair {
3087                key: kawa::Store::Static(b"Content-Length"),
3088                val: kawa::Store::Static(val),
3089            };
3090            if elided {
3091                pair.elide();
3092            }
3093            kawa.push_block(kawa::Block::Header(pair));
3094            kawa
3095        }
3096        let cases: [(&str, &'static [u8], bool, bool); 7] = [
3097            ("minus-zero", b"-0", false, true),
3098            ("hexadecimal", b"0x5", false, true),
3099            ("inner-space", b"5 5", false, true),
3100            ("empty", b"", false, true),
3101            ("leading-zeros", b"005", false, false),
3102            ("zero", b"0", false, false),
3103            ("elided-plus", b"+5", true, false),
3104        ];
3105        for (label, val, elided, expected) in cases {
3106            assert_eq!(
3107                has_non_digit_content_length(&content_length(val, elided)),
3108                expected,
3109                "{label}"
3110            );
3111        }
3112    }
3113
3114    /// Non-regression for the Content-Length clause of `on_request_headers`:
3115    /// it judges only the Content-Length that is forwarded.
3116    ///
3117    /// - `005` matches `1*DIGIT`: it is legal and forwarded as sent, exactly
3118    ///   as the H2 path (`pkawa::write_regular_header`) forwards it.
3119    /// - `+5` beside `Transfer-Encoding: chunked`: kawa elided the
3120    ///   Content-Length (RFC 9110 §6.3 — Transfer-Encoding overrides it)
3121    ///   without reading its value, so nothing non-canonical is forwarded.
3122    ///
3123    /// `5` then `+5` is no longer a row here: kawa 0.7.1 elided the second
3124    /// line as equal to the first and forwarded `5`, while kawa >= 0.7.2
3125    /// refuses the `+5` line itself, so that request is now pinned as
3126    /// refused in `a_request_content_length_that_is_not_only_digits_is_rejected`.
3127    #[test]
3128    fn a_forwarded_content_length_is_only_digits() {
3129        let cases: [(&str, &[u8], &str); 3] = [
3130            (
3131                "canonical",
3132                b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\n\r\nHello",
3133                "Content-Length: 5\r\n",
3134            ),
3135            (
3136                "leading-zeros",
3137                b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 005\r\n\r\nHello",
3138                "Content-Length: 005\r\n",
3139            ),
3140            (
3141                "plus-beside-chunked",
3142                b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: +5\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\n\r\n",
3143                "Transfer-Encoding: chunked\r\n",
3144            ),
3145        ];
3146        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3147        for (label, request, expected) in cases {
3148            let mut kawa = parse_framed(&mut pool, kawa::Kind::Request, request);
3149            assert!(
3150                !kawa.is_error(),
3151                "{label}: must be accepted, got {:?}",
3152                kawa.parsing_phase
3153            );
3154            let forwarded = serialized_request(&mut kawa);
3155            assert!(
3156                forwarded.contains(expected),
3157                "{label}: expected {expected:?} in {forwarded:?}"
3158            );
3159            assert!(
3160                !forwarded.contains('+'),
3161                "{label}: a signed Content-Length must never be forwarded: {forwarded:?}"
3162            );
3163            assert_eq!(
3164                forwarded.matches("Content-Length").count(),
3165                usize::from(label != "plus-beside-chunked"),
3166                "{label}: exactly the framing Content-Length is forwarded: {forwarded:?}"
3167            );
3168        }
3169    }
3170
3171    /// The response mirror: a backend's `Content-Length: +5` fails the
3172    /// response parse, which the mux answers with a 502 before any byte of
3173    /// it reaches the client (`shared::end_stream_decision`).
3174    ///
3175    /// kawa >= 0.7.2 already fails such a response itself
3176    /// (CleverCloud/kawa#26), before `on_response_headers` runs.
3177    ///
3178    /// TO SEE THIS RED: deleting the Content-Length clause of
3179    /// `on_response_headers` alone no longer turns it red. Pin
3180    /// kawa 0.7.1 (`kawa = { version = "=0.7.1", default-features = false }`
3181    /// in the root `Cargo.toml`, then `cargo update -p kawa --precise 0.7.1`)
3182    /// and delete the clause; the helper's own predicate is pinned by
3183    /// `the_content_length_helper_judges_every_non_digit_value`.
3184    #[test]
3185    fn a_response_content_length_that_is_not_only_digits_is_rejected() {
3186        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3187        for status in ["200 OK", "204 No Content", "304 Not Modified"] {
3188            let response = format!("HTTP/1.1 {status}\r\nContent-Length: +5\r\n\r\nHello");
3189            let kawa = parse_framed(&mut pool, kawa::Kind::Response, response.as_bytes());
3190            assert!(
3191                kawa.is_error(),
3192                "{status}: a signed Content-Length must fail the response, got {:?}",
3193                kawa.parsing_phase
3194            );
3195        }
3196        let kawa = parse_framed(
3197            &mut pool,
3198            kawa::Kind::Response,
3199            b"HTTP/1.1 200 OK\r\nContent-Length: 005\r\n\r\nHello",
3200        );
3201        assert!(
3202            !kawa.is_error(),
3203            "a 1*DIGIT Content-Length must still be accepted, got {:?}",
3204            kawa.parsing_phase
3205        );
3206    }
3207
3208    // ── header-editing allocations ─────────────────────────────────────
3209
3210    /// A bare keep-alive request: no forwarding header, no `X-Request-Id`,
3211    /// no `User-Agent`, so every header the editor adds is synthesised.
3212    const BARE_REQUEST: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
3213
3214    /// Parse `request` through the real parser, and so through
3215    /// `on_request_headers`, into `kawa`, and return the heap allocations
3216    /// the parse made.
3217    ///
3218    /// `kawa` is cleared first, the way the H1 keep-alive path clears a
3219    /// stream's front kawa between two requests, so a caller that parses on
3220    /// the same kawa twice measures the second parse without the growth of
3221    /// kawa's own block queue.
3222    fn allocations_of_request_parse(
3223        ctx: &mut HttpContext,
3224        kawa: &mut GenericHttpStream,
3225        request: &[u8],
3226    ) -> usize {
3227        use crate::test_allocations::allocations;
3228        kawa.clear();
3229        kawa.storage.clear();
3230        kawa.storage.space()[..request.len()].copy_from_slice(request);
3231        kawa.storage.fill(request.len());
3232        let before = allocations();
3233        kawa::h1::parse(kawa, ctx);
3234        let parsed = allocations() - before;
3235        assert!(kawa.is_main_phase(), "premise: the request must parse");
3236        parsed
3237    }
3238
3239    /// A kawa on its own pool buffer, its block queue warmed by one parse on
3240    /// a throwaway context so later measurements see only the editor.
3241    fn warm_request_kawa(pool: &mut crate::pool::Pool) -> GenericHttpStream {
3242        let mut kawa: GenericHttpStream = kawa::Kawa::new(
3243            kawa::Kind::Request,
3244            kawa::Buffer::new(
3245                pool.checkout()
3246                    .expect("the test pool must hand out a buffer"),
3247            ),
3248        );
3249        allocations_of_request_parse(&mut make_context(), &mut kawa, BARE_REQUEST);
3250        kawa
3251    }
3252
3253    /// The bytes `kawa` would send upstream, serialised the way
3254    /// `ConnectionH1::writable` serialises them.
3255    fn serialized_request(kawa: &mut GenericHttpStream) -> String {
3256        kawa.prepare(&mut kawa::h1::BlockConverter);
3257        let buffer = kawa.storage.buffer();
3258        let bytes = kawa
3259            .out
3260            .iter()
3261            .flat_map(|block| match block {
3262                kawa::OutBlock::Store(store) => store.data(buffer).to_vec(),
3263                kawa::OutBlock::Delimiter => Vec::new(),
3264            })
3265            .collect::<Vec<u8>>();
3266        String::from_utf8(bytes).expect("the serialised request must be UTF-8")
3267    }
3268
3269    /// The `traceparent` line `on_request_headers` synthesises for a request
3270    /// that carried none, rendered from the ids it recorded on `ctx.otel`;
3271    /// empty without the `opentelemetry` feature, which adds no header.
3272    fn synthesised_traceparent_line(ctx: &HttpContext) -> String {
3273        #[cfg(feature = "opentelemetry")]
3274        {
3275            let otel = ctx
3276                .otel
3277                .as_ref()
3278                .expect("on_request_headers records the trace context it forwards");
3279            let value = build_traceparent(&otel.trace_id, &otel.span_id);
3280            format!(
3281                "traceparent: {}\r\n",
3282                from_utf8(&value).expect("a traceparent is ASCII hex")
3283            )
3284        }
3285        #[cfg(not(feature = "opentelemetry"))]
3286        {
3287            let _ = ctx;
3288            String::new()
3289        }
3290    }
3291
3292    /// Heap operations of the `traceparent` header `on_request_headers`
3293    /// synthesises under the `opentelemetry` feature: its value is built on
3294    /// the stack (`build_traceparent`) and copied once
3295    /// (`kawa::Store::from_slice`), the same one exact copy as every other
3296    /// synthesised header. Without the feature there is no such header.
3297    const SYNTHESISED_TRACEPARENT: usize = if cfg!(feature = "opentelemetry") {
3298        1
3299    } else {
3300        0
3301    };
3302
3303    /// The first request of a connection: every forwarding header is
3304    /// synthesised from the connection's forwarding values, which one
3305    /// scratch renders and each copies out once; the request id is rendered
3306    /// once. The next requests reuse the values
3307    /// (`keep_alive_requests_reuse_the_connection_forwarding_hop`).
3308    ///
3309    /// The seven: `authority` and `path` captured for routing and the access
3310    /// log (2), the scratch (1), the `X-Forwarded-For` hop, the `Forwarded`
3311    /// element and the `X-Forwarded-Port` value (3), and the one rendering of
3312    /// the request id that `X-Request-Id`, the access log and the `Sozu-Id`
3313    /// value share (1). The default `Sozu-Id` name is a `'static` literal
3314    /// (0). Under the `opentelemetry` feature, the synthesised `traceparent`
3315    /// adds its one copy ([`SYNTHESISED_TRACEPARENT`]).
3316    ///
3317    /// TO SEE THIS RED, either of:
3318    /// - in `on_request_headers`, allocate a scratch for every request again
3319    ///   to extend the client chains in, as the recipe of
3320    ///   `a_client_chain_costs_one_exact_copy_per_extended_header` does.
3321    ///   Measured: `left: 8, right: 7`.
3322    /// - render the id per header again: `kawa::Store::from_string(
3323    ///   self.id.to_string())` for the `Sozu-Id` value, `from_string(
3324    ///   self.sozu_id_header.clone())` for its name, and a `self.id
3325    ///   .to_string()` copied for `X-Request-Id`. Measured: `left: 10,
3326    ///   right: 7`.
3327    #[test]
3328    fn a_bare_request_costs_one_scratch_and_one_copy_per_forwarding_header() {
3329        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3330        let mut kawa = warm_request_kawa(&mut pool);
3331        let mut ctx = make_context();
3332
3333        let first = allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3334
3335        assert_eq!(
3336            first,
3337            7 + SYNTHESISED_TRACEPARENT,
3338            "a bare request's header editing must cost one scratch, one exact \
3339             copy per synthesised forwarding header and one rendering of the \
3340             request id"
3341        );
3342    }
3343
3344    /// Parse `response` through the real parser, and so through
3345    /// `on_response_headers`, into a warm response kawa; return the heap
3346    /// allocations the parse made and the serialised response.
3347    fn response_parse(ctx: &mut HttpContext, response: &[u8]) -> (usize, String) {
3348        use crate::test_allocations::allocations;
3349        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3350        let mut kawa: GenericHttpStream = kawa::Kawa::new(
3351            kawa::Kind::Response,
3352            kawa::Buffer::new(
3353                pool.checkout()
3354                    .expect("the test pool must hand out a buffer"),
3355            ),
3356        );
3357        let mut measured = 0;
3358        // The first parse warms kawa's block queue on a throwaway context.
3359        for context in [&mut make_context(), ctx] {
3360            kawa.clear();
3361            kawa.storage.clear();
3362            kawa.storage.space()[..response.len()].copy_from_slice(response);
3363            kawa.storage.fill(response.len());
3364            let before = allocations();
3365            kawa::h1::parse(&mut kawa, context);
3366            measured = allocations() - before;
3367            assert!(kawa.is_main_phase(), "premise: the response must parse");
3368        }
3369        kawa.prepare(&mut kawa::h1::BlockConverter);
3370        let buffer = kawa.storage.buffer();
3371        let bytes = kawa
3372            .out
3373            .iter()
3374            .flat_map(|block| match block {
3375                kawa::OutBlock::Store(store) => store.data(buffer).to_vec(),
3376                kawa::OutBlock::Delimiter => Vec::new(),
3377            })
3378            .collect::<Vec<u8>>();
3379        (
3380            measured,
3381            String::from_utf8(bytes).expect("the serialised response must be UTF-8"),
3382        )
3383    }
3384
3385    /// The response's `Sozu-Id` shares the rendering the request generated
3386    /// for `X-Request-Id`, and still carries Sōzu's own id when the client
3387    /// chose its `X-Request-Id`.
3388    ///
3389    /// Nothing is left to allocate on the generated path: the standard `OK`
3390    /// reason is captured for the access log as a `'static` phrase
3391    /// (`standard_reason`).
3392    ///
3393    /// TO SEE THIS RED, either of:
3394    /// - in `on_response_headers`, push the correlation header as `key:
3395    ///   kawa::Store::from_string(self.sozu_id_header.clone())` and `val:
3396    ///   kawa::Store::from_string(self.id.to_string())`. Measured: `left: 2,
3397    ///   right: 0`.
3398    /// - capture every reason as a copy again, `.map(|reason|
3399    ///   Cow::Owned(reason.to_owned()))`. Measured: `left: 1, right: 0`.
3400    #[test]
3401    fn a_response_shares_the_request_id_rendering() {
3402        const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
3403        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3404        let mut kawa = warm_request_kawa(&mut pool);
3405
3406        let mut generated = make_context();
3407        allocations_of_request_parse(&mut generated, &mut kawa, BARE_REQUEST);
3408        let (allocations, response) = response_parse(&mut generated, RESPONSE);
3409        let id = generated.id.to_string();
3410        assert_eq!(
3411            response,
3412            format!("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nSozu-Id: {id}\r\n\r\n")
3413        );
3414        assert_eq!(
3415            allocations, 0,
3416            "the response must share the id rendering the request generated \
3417             and borrow its standard reason"
3418        );
3419
3420        let mut client_chosen = make_context();
3421        allocations_of_request_parse(
3422            &mut client_chosen,
3423            &mut kawa,
3424            b"GET / HTTP/1.1\r\nHost: example.com\r\nX-Request-Id: client-chosen\r\n\r\n",
3425        );
3426        let (_, response) = response_parse(&mut client_chosen, RESPONSE);
3427        assert_eq!(
3428            response,
3429            format!(
3430                "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nSozu-Id: {}\r\n\r\n",
3431                client_chosen.id
3432            ),
3433            "the correlation header carries Sōzu's id, not the client's"
3434        );
3435    }
3436
3437    /// The reason of a response is captured for the access log as the
3438    /// `'static` phrase RFC 9110 §15 registers for its code when the backend
3439    /// sent exactly that phrase, and as a verbatim copy otherwise — a
3440    /// different phrase, a different case, an older name or an unregistered
3441    /// code. The forwarded bytes come from kawa's status line either way and
3442    /// stay byte-exact across the keep-alive requests of a connection.
3443    ///
3444    /// TO SEE THIS RED: capture every reason as a copy again, `.map(|reason|
3445    /// Cow::Owned(reason.to_owned()))` in `on_response_headers`. Measured:
3446    /// `left: 1, right: 0` on the first, standard, response.
3447    #[test]
3448    fn a_response_reason_is_borrowed_when_standard_and_copied_otherwise() {
3449        let cases: [(&[u8], &str, usize); 8] = [
3450            (b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", "OK", 0),
3451            (b"HTTP/1.1 200 Fine\r\nContent-Length: 0\r\n\r\n", "Fine", 1),
3452            (b"HTTP/1.1 200 ok\r\nContent-Length: 0\r\n\r\n", "ok", 1),
3453            (
3454                b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n",
3455                "Not Found",
3456                0,
3457            ),
3458            (
3459                b"HTTP/1.1 413 Payload Too Large\r\nContent-Length: 0\r\n\r\n",
3460                "Payload Too Large",
3461                1,
3462            ),
3463            (
3464                b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
3465                "Service Unavailable",
3466                0,
3467            ),
3468            (
3469                b"HTTP/1.1 599 Network Connect Timeout Error\r\nContent-Length: 0\r\n\r\n",
3470                "Network Connect Timeout Error",
3471                1,
3472            ),
3473            (
3474                b"HTTP/1.1 302 Found\r\nContent-Length: 0\r\n\r\n",
3475                "Found",
3476                0,
3477            ),
3478        ];
3479        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3480        let mut kawa = warm_request_kawa(&mut pool);
3481        let mut ctx = make_context();
3482
3483        for (response, reason, cost) in cases {
3484            allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3485            let (allocations, serialized) = response_parse(&mut ctx, response);
3486            let raw = from_utf8(response).expect("the response literal is ASCII");
3487            let headers_end = raw.len() - "\r\n".len();
3488            assert_eq!(
3489                serialized,
3490                format!("{}Sozu-Id: {}\r\n\r\n", &raw[..headers_end], ctx.id),
3491                "the forwarded response of {raw:?} is byte-exact"
3492            );
3493            assert_eq!(
3494                ctx.reason.as_deref(),
3495                Some(reason),
3496                "the access log records the reason of {raw:?} verbatim"
3497            );
3498            assert_eq!(
3499                allocations, cost,
3500                "capturing the reason of {raw:?} costs {cost} allocation(s)"
3501            );
3502            ctx.reset(Ulid::generate());
3503        }
3504    }
3505
3506    /// `render_ulid` agrees with `Ulid`'s own `Display` at both ends of the
3507    /// 128-bit range and on generated ids.
3508    #[test]
3509    fn render_ulid_matches_the_display_rendering() {
3510        let ids = [
3511            Ulid::from(0u128),
3512            Ulid::from(u128::MAX),
3513            Ulid::from(1u128 << 127),
3514            Ulid::generate(),
3515            Ulid::generate(),
3516            Ulid::generate(),
3517        ];
3518        for id in ids {
3519            assert_eq!(
3520                from_utf8(&render_ulid(id)).expect("a rendered ULID is ASCII"),
3521                id.to_string()
3522            );
3523        }
3524        assert_eq!(render_ulid(Ulid::from(u128::MAX))[0], b'7');
3525    }
3526
3527    /// Byte-exact output of the header editing, for the synthesised headers
3528    /// and for the chains a client supplied, on the first request of a
3529    /// connection and on the next ones after `reset`: each request carries
3530    /// its own id, never the one of the request before it.
3531    #[test]
3532    fn header_editing_output_is_byte_exact_across_keep_alive_requests() {
3533        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3534        let mut kawa = warm_request_kawa(&mut pool);
3535        let mut ctx = HttpContext::new(
3536            Ulid::generate(),
3537            Ulid::generate(),
3538            Protocol::HTTP,
3539            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
3540            Some(SocketAddr::new(
3541                IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
3542                54321,
3543            )),
3544            "SERVERID".to_owned(),
3545            "X-Edge-Id".to_owned(),
3546            false,
3547            true,
3548        );
3549        let mut seen = Vec::new();
3550
3551        for request in 0..3 {
3552            let id = ctx.id.to_string();
3553            assert!(
3554                !seen.contains(&id),
3555                "request {request} of the connection reuses the id of an earlier one: {id}"
3556            );
3557            seen.push(id.clone());
3558            allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3559            let traceparent = synthesised_traceparent_line(&ctx);
3560            assert_eq!(
3561                serialized_request(&mut kawa),
3562                format!(
3563                    "GET / HTTP/1.1\r\nHost: example.com\r\n\
3564                     X-Forwarded-For: 10.0.0.1\r\n\
3565                     Forwarded: proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
3566                     X-Real-IP: 10.0.0.1\r\n\
3567                     {traceparent}\
3568                     X-Forwarded-Port: 8080\r\n\
3569                     X-Forwarded-Proto: http\r\n\
3570                     X-Request-Id: {id}\r\n\
3571                     X-Edge-Id: {id}\r\n\r\n"
3572                ),
3573                "request {request} of the connection"
3574            );
3575            assert_eq!(ctx.x_request_id.as_deref(), Some(id.as_str()));
3576            ctx.reset(Ulid::generate());
3577        }
3578
3579        let id = ctx.id.to_string();
3580        assert!(
3581            !seen.contains(&id),
3582            "the request with a client X-Request-Id reuses an earlier id: {id}"
3583        );
3584        allocations_of_request_parse(
3585            &mut ctx,
3586            &mut kawa,
3587            b"GET / HTTP/1.1\r\nHost: example.com\r\n\
3588              X-Forwarded-For: 192.0.2.7\r\n\
3589              Forwarded: for=192.0.2.7\r\n\
3590              X-Request-Id: client-chosen\r\n\r\n",
3591        );
3592        let traceparent = synthesised_traceparent_line(&ctx);
3593        assert_eq!(
3594            serialized_request(&mut kawa),
3595            format!(
3596                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3597                 X-Forwarded-For: 192.0.2.7, 10.0.0.1\r\n\
3598                 Forwarded: for=192.0.2.7, proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
3599                 X-Request-Id: client-chosen\r\n\
3600                 X-Real-IP: 10.0.0.1\r\n\
3601                 {traceparent}\
3602                 X-Forwarded-Port: 8080\r\n\
3603                 X-Forwarded-Proto: http\r\n\
3604                 X-Edge-Id: {id}\r\n\r\n"
3605            ),
3606            "client chains are extended, a client X-Request-Id is kept verbatim"
3607        );
3608        assert_eq!(ctx.x_request_id.as_deref(), Some("client-chosen"));
3609    }
3610
3611    // ── connection-scoped forwarding hop ───────────────────────────────
3612
3613    /// A context whose forwarding inputs the caller chooses, with the
3614    /// default correlation header name.
3615    fn forwarding_context(
3616        protocol: Protocol,
3617        public_address: SocketAddr,
3618        session_address: SocketAddr,
3619        send_x_real_ip: bool,
3620    ) -> HttpContext {
3621        HttpContext::new(
3622            Ulid::generate(),
3623            Ulid::generate(),
3624            protocol,
3625            public_address,
3626            Some(session_address),
3627            "SERVERID".to_owned(),
3628            "Sozu-Id".to_owned(),
3629            false,
3630            send_x_real_ip,
3631        )
3632    }
3633
3634    /// The second and later requests of a keep-alive connection reuse the
3635    /// forwarding values the first one rendered: `X-Forwarded-For`,
3636    /// `Forwarded`, `X-Real-IP` and `X-Forwarded-Port` then cost no heap
3637    /// operation at all.
3638    ///
3639    /// The three left: `authority` and `path` captured for routing and the
3640    /// access log (2), and the one rendering of the request id (1). Under
3641    /// the `opentelemetry` feature, the synthesised `traceparent` adds its
3642    /// one copy ([`SYNTHESISED_TRACEPARENT`]).
3643    ///
3644    /// TO SEE THIS RED: in `HttpContext::forwarding_hop`, rebuild the hop on
3645    /// every call instead of returning the cached one whose inputs match.
3646    /// Measured: `left: 7, right: 3` (the scratch and the three values).
3647    #[test]
3648    fn keep_alive_requests_reuse_the_connection_forwarding_hop() {
3649        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3650        let mut kawa = warm_request_kawa(&mut pool);
3651        let mut ctx = forwarding_context(
3652            Protocol::HTTP,
3653            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
3654            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 54321),
3655            true,
3656        );
3657
3658        let first = allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3659        assert_eq!(
3660            first,
3661            7 + SYNTHESISED_TRACEPARENT,
3662            "the first request renders the connection's forwarding values once, \
3663             X-Real-IP sharing the X-Forwarded-For rendering"
3664        );
3665        for request in 1..4 {
3666            ctx.reset(Ulid::generate());
3667            let later = allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3668            assert_eq!(
3669                later,
3670                3 + SYNTHESISED_TRACEPARENT,
3671                "request {request} of the connection must reuse the forwarding values"
3672            );
3673        }
3674    }
3675
3676    /// A client-supplied chain is extended with one exact-size copy per
3677    /// header, the appended hop coming from the connection's rendering.
3678    ///
3679    /// The six on a later request: `authority` and `path` (2), the
3680    /// `xff_chain` snapshot for the access log (1), the extended
3681    /// `X-Forwarded-For` and `Forwarded` (2), and the request id (1).
3682    ///
3683    /// TO SEE THIS RED: in `on_request_headers`, extend the client chains in
3684    /// a `Vec::with_capacity(128)` scratch shared by both again, each copied
3685    /// out with `kawa::Store::from_slice`. Measured: `left: 7, right: 6`, the
3686    /// scratch.
3687    #[test]
3688    fn a_client_chain_costs_one_exact_copy_per_extended_header() {
3689        const CLIENT_CHAINS: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
3690              X-Forwarded-For: 192.0.2.7\r\n\
3691              Forwarded: for=192.0.2.7\r\n\r\n";
3692        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3693        let mut kawa = warm_request_kawa(&mut pool);
3694        let mut ctx = make_context();
3695
3696        allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3697        ctx.reset(Ulid::generate());
3698        let later = allocations_of_request_parse(&mut ctx, &mut kawa, CLIENT_CHAINS);
3699        assert_eq!(
3700            later,
3701            6 + SYNTHESISED_TRACEPARENT,
3702            "a client chain costs one exact copy per extended header"
3703        );
3704    }
3705
3706    /// The bytes forwarded on a keep-alive connection that alternates bare
3707    /// requests with requests carrying their own chains: a client chain is
3708    /// extended, never kept for the next request, and the IPv6 literals
3709    /// keep their RFC 7239 §6 brackets and quotes.
3710    #[test]
3711    fn a_client_chain_never_leaks_into_the_next_request() {
3712        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3713        let mut kawa = warm_request_kawa(&mut pool);
3714        let mut ctx = forwarding_context(
3715            Protocol::HTTPS,
3716            SocketAddr::new(
3717                IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
3718                443,
3719            ),
3720            SocketAddr::new(
3721                IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 1, 2)),
3722                50000,
3723            ),
3724            true,
3725        );
3726        let bare = |id: &str, traceparent: &str| {
3727            format!(
3728                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3729                 X-Forwarded-For: 2001:db8::1:2\r\n\
3730                 Forwarded: proto=https;for=\"[2001:db8::1:2]:50000\";by=\"[2001:db8::1]\"\r\n\
3731                 X-Real-IP: 2001:db8::1:2\r\n\
3732                 {traceparent}\
3733                 X-Forwarded-Port: 443\r\n\
3734                 X-Forwarded-Proto: https\r\n\
3735                 X-Request-Id: {id}\r\n\
3736                 Sozu-Id: {id}\r\n\r\n"
3737            )
3738        };
3739
3740        for round in 0..2 {
3741            let id = ctx.id.to_string();
3742            allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3743            let traceparent = synthesised_traceparent_line(&ctx);
3744            assert_eq!(
3745                serialized_request(&mut kawa),
3746                bare(&id, &traceparent),
3747                "bare request of round {round}"
3748            );
3749            assert_eq!(ctx.xff_chain, None);
3750            ctx.reset(Ulid::generate());
3751
3752            let id = ctx.id.to_string();
3753            allocations_of_request_parse(
3754                &mut ctx,
3755                &mut kawa,
3756                b"GET / HTTP/1.1\r\nHost: example.com\r\n\
3757                  X-Forwarded-For: 192.0.2.7, 198.51.100.3\r\n\
3758                  Forwarded: for=192.0.2.7;proto=http\r\n\r\n",
3759            );
3760            let traceparent = synthesised_traceparent_line(&ctx);
3761            assert_eq!(
3762                serialized_request(&mut kawa),
3763                format!(
3764                    "GET / HTTP/1.1\r\nHost: example.com\r\n\
3765                     X-Forwarded-For: 192.0.2.7, 198.51.100.3, 2001:db8::1:2\r\n\
3766                     Forwarded: for=192.0.2.7;proto=http, \
3767                     proto=https;for=\"[2001:db8::1:2]:50000\";by=\"[2001:db8::1]\"\r\n\
3768                     X-Real-IP: 2001:db8::1:2\r\n\
3769                     {traceparent}\
3770                     X-Forwarded-Port: 443\r\n\
3771                     X-Forwarded-Proto: https\r\n\
3772                     X-Request-Id: {id}\r\n\
3773                     Sozu-Id: {id}\r\n\r\n"
3774                ),
3775                "request with client chains of round {round}"
3776            );
3777            assert_eq!(ctx.xff_chain.as_deref(), Some("192.0.2.7, 198.51.100.3"));
3778            ctx.reset(Ulid::generate());
3779        }
3780    }
3781
3782    /// Changing an input of the forwarding values between two requests —
3783    /// the peer, the public address or the protocol — renders them again:
3784    /// a value is reused only for the inputs it was rendered from.
3785    #[test]
3786    fn a_changed_forwarding_input_renders_the_hop_again() {
3787        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3788        let mut kawa = warm_request_kawa(&mut pool);
3789        let mut ctx = forwarding_context(
3790            Protocol::HTTP,
3791            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
3792            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 54321),
3793            true,
3794        );
3795        let expected = |ctx: &HttpContext, peer: &str, forwarded: &str, port: u16, proto: &str| {
3796            let id = ctx.id.to_string();
3797            let traceparent = synthesised_traceparent_line(ctx);
3798            format!(
3799                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3800                 X-Forwarded-For: {peer}\r\n\
3801                 Forwarded: {forwarded}\r\n\
3802                 X-Real-IP: {peer}\r\n\
3803                 {traceparent}\
3804                 X-Forwarded-Port: {port}\r\n\
3805                 X-Forwarded-Proto: {proto}\r\n\
3806                 X-Request-Id: {id}\r\n\
3807                 Sozu-Id: {id}\r\n\r\n"
3808            )
3809        };
3810
3811        allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3812        assert_eq!(
3813            serialized_request(&mut kawa),
3814            expected(
3815                &ctx,
3816                "10.0.0.1",
3817                "proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1",
3818                8080,
3819                "http"
3820            )
3821        );
3822
3823        ctx.reset(Ulid::generate());
3824        ctx.session_address = Some(SocketAddr::new(
3825            IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)),
3826            40000,
3827        ));
3828        allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3829        assert_eq!(
3830            serialized_request(&mut kawa),
3831            expected(
3832                &ctx,
3833                "10.0.0.2",
3834                "proto=http;for=\"10.0.0.2:40000\";by=127.0.0.1",
3835                8080,
3836                "http"
3837            ),
3838            "a new peer is rendered"
3839        );
3840
3841        ctx.reset(Ulid::generate());
3842        ctx.public_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), 8443);
3843        allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3844        assert_eq!(
3845            serialized_request(&mut kawa),
3846            expected(
3847                &ctx,
3848                "10.0.0.2",
3849                "proto=http;for=\"10.0.0.2:40000\";by=192.0.2.1",
3850                8443,
3851                "http"
3852            ),
3853            "a new public address is rendered"
3854        );
3855
3856        // The protocol alone changes: the addresses are the ones the cached
3857        // values were rendered from, so only the protocol can invalidate
3858        // them — a stale `Forwarded` would still say `proto=http`.
3859        ctx.reset(Ulid::generate());
3860        ctx.protocol = Protocol::HTTPS;
3861        allocations_of_request_parse(&mut ctx, &mut kawa, BARE_REQUEST);
3862        assert_eq!(
3863            serialized_request(&mut kawa),
3864            expected(
3865                &ctx,
3866                "10.0.0.2",
3867                "proto=https;for=\"10.0.0.2:40000\";by=192.0.2.1",
3868                8443,
3869                "https"
3870            ),
3871            "a new protocol is rendered"
3872        );
3873    }
3874
3875    // ── forwarding header family (`forwarded_headers`) ─────────────────
3876
3877    /// A request carrying a client chain of every forwarding header the
3878    /// editor manages: two `X-Forwarded-For` lines, a `Forwarded` chain,
3879    /// `X-Forwarded-Proto`, `X-Forwarded-Port` and `X-Forwarded-Host`.
3880    const CLIENT_CHAIN_REQUEST: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
3881        X-Forwarded-For: 198.51.100.1\r\n\
3882        X-Forwarded-For: 192.0.2.7\r\n\
3883        Forwarded: for=192.0.2.7\r\n\
3884        X-Forwarded-Proto: https\r\n\
3885        X-Forwarded-Port: 443\r\n\
3886        X-Forwarded-Host: public.example\r\n\
3887        X-Real-IP: 203.0.113.9\r\n\
3888        X-Request-Id: client-chosen\r\n\r\n";
3889
3890    /// The request `on_request_headers` forwards for `request` on a
3891    /// connection from `10.0.0.1:54321` to `127.0.0.1:8080` whose listener
3892    /// has the forwarding header family `mode` and `send_x_real_ip` on,
3893    /// with the context left for the caller to inspect.
3894    fn forwarded_in_mode(mode: ForwardedHeaders, request: &[u8]) -> (String, HttpContext) {
3895        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
3896        let mut kawa = warm_request_kawa(&mut pool);
3897        let mut ctx = forwarding_context(
3898            Protocol::HTTP,
3899            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
3900            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 54321),
3901            true,
3902        );
3903        ctx.forwarded_headers = mode;
3904        allocations_of_request_parse(&mut ctx, &mut kawa, request);
3905        (serialized_request(&mut kawa), ctx)
3906    }
3907
3908    /// `both`, the default, is the historical behaviour, pinned byte for
3909    /// byte: a bare request gains the whole `X-Forwarded-*` family and a
3910    /// `Forwarded` element; a client chain is extended (the last
3911    /// `X-Forwarded-For` line and the last `Forwarded` line), and a client
3912    /// `X-Forwarded-Proto`, `-Port` or `-Host` is kept as sent.
3913    #[test]
3914    fn forwarded_headers_both_adds_both_families_and_extends_client_chains() {
3915        let (bare, ctx) = forwarded_in_mode(ForwardedHeaders::Both, BARE_REQUEST);
3916        let id = ctx.id.to_string();
3917        let traceparent = synthesised_traceparent_line(&ctx);
3918        assert_eq!(
3919            bare,
3920            format!(
3921                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3922                 X-Forwarded-For: 10.0.0.1\r\n\
3923                 Forwarded: proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
3924                 X-Real-IP: 10.0.0.1\r\n\
3925                 {traceparent}\
3926                 X-Forwarded-Port: 8080\r\n\
3927                 X-Forwarded-Proto: http\r\n\
3928                 X-Request-Id: {id}\r\n\
3929                 Sozu-Id: {id}\r\n\r\n"
3930            )
3931        );
3932
3933        let (chained, ctx) = forwarded_in_mode(ForwardedHeaders::Both, CLIENT_CHAIN_REQUEST);
3934        let id = ctx.id.to_string();
3935        let traceparent = synthesised_traceparent_line(&ctx);
3936        assert_eq!(
3937            chained,
3938            format!(
3939                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3940                 X-Forwarded-For: 198.51.100.1\r\n\
3941                 X-Forwarded-For: 192.0.2.7, 10.0.0.1\r\n\
3942                 Forwarded: for=192.0.2.7, proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
3943                 X-Forwarded-Proto: https\r\n\
3944                 X-Forwarded-Port: 443\r\n\
3945                 X-Forwarded-Host: public.example\r\n\
3946                 X-Real-IP: 203.0.113.9\r\n\
3947                 X-Request-Id: client-chosen\r\n\
3948                 X-Real-IP: 10.0.0.1\r\n\
3949                 {traceparent}\
3950                 Sozu-Id: {id}\r\n\r\n"
3951            )
3952        );
3953        assert_eq!(ctx.xff_chain.as_deref(), Some("192.0.2.7"));
3954    }
3955
3956    /// `x_forwarded` handles the `X-Forwarded-*` family exactly as `both`
3957    /// does and adds no `Forwarded`: a client `Forwarded` chain passes
3958    /// through as sent, neither extended nor removed.
3959    ///
3960    /// TO SEE THIS RED: in `HttpContext::on_request_headers`, drop the
3961    /// `emits_rfc7239` guard on the `Forwarded` extension and synthesis.
3962    #[test]
3963    fn forwarded_headers_x_forwarded_adds_only_the_x_forwarded_family() {
3964        let (bare, ctx) = forwarded_in_mode(ForwardedHeaders::XForwarded, BARE_REQUEST);
3965        let id = ctx.id.to_string();
3966        let traceparent = synthesised_traceparent_line(&ctx);
3967        assert_eq!(
3968            bare,
3969            format!(
3970                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3971                 X-Forwarded-For: 10.0.0.1\r\n\
3972                 X-Real-IP: 10.0.0.1\r\n\
3973                 {traceparent}\
3974                 X-Forwarded-Port: 8080\r\n\
3975                 X-Forwarded-Proto: http\r\n\
3976                 X-Request-Id: {id}\r\n\
3977                 Sozu-Id: {id}\r\n\r\n"
3978            )
3979        );
3980
3981        let (chained, ctx) = forwarded_in_mode(ForwardedHeaders::XForwarded, CLIENT_CHAIN_REQUEST);
3982        let id = ctx.id.to_string();
3983        let traceparent = synthesised_traceparent_line(&ctx);
3984        assert_eq!(
3985            chained,
3986            format!(
3987                "GET / HTTP/1.1\r\nHost: example.com\r\n\
3988                 X-Forwarded-For: 198.51.100.1\r\n\
3989                 X-Forwarded-For: 192.0.2.7, 10.0.0.1\r\n\
3990                 Forwarded: for=192.0.2.7\r\n\
3991                 X-Forwarded-Proto: https\r\n\
3992                 X-Forwarded-Port: 443\r\n\
3993                 X-Forwarded-Host: public.example\r\n\
3994                 X-Real-IP: 203.0.113.9\r\n\
3995                 X-Request-Id: client-chosen\r\n\
3996                 X-Real-IP: 10.0.0.1\r\n\
3997                 {traceparent}\
3998                 Sozu-Id: {id}\r\n\r\n"
3999            )
4000        );
4001        assert_eq!(ctx.xff_chain.as_deref(), Some("192.0.2.7"));
4002    }
4003
4004    /// `rfc7239` adds only `Forwarded`: its own element is appended to the
4005    /// last client `Forwarded` line (RFC 7239 §4) or synthesised, and every
4006    /// client `X-Forwarded-For`, `-Proto`, `-Port` and `-Host` line is
4007    /// removed, so the backend sees one forwarding family whose last element
4008    /// Sōzu wrote. The client `X-Forwarded-For` is still recorded for the
4009    /// access log, and `X-Real-IP` is untouched by the mode.
4010    ///
4011    /// TO SEE THIS RED: in `HttpContext::on_request_headers`, drop the
4012    /// `strips_x_forwarded` branch that elides the client headers.
4013    #[test]
4014    fn forwarded_headers_rfc7239_adds_only_forwarded_and_strips_client_x_forwarded() {
4015        let (bare, ctx) = forwarded_in_mode(ForwardedHeaders::Rfc7239, BARE_REQUEST);
4016        let id = ctx.id.to_string();
4017        let traceparent = synthesised_traceparent_line(&ctx);
4018        assert_eq!(
4019            bare,
4020            format!(
4021                "GET / HTTP/1.1\r\nHost: example.com\r\n\
4022                 Forwarded: proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
4023                 X-Real-IP: 10.0.0.1\r\n\
4024                 {traceparent}\
4025                 X-Request-Id: {id}\r\n\
4026                 Sozu-Id: {id}\r\n\r\n"
4027            )
4028        );
4029
4030        let (chained, ctx) = forwarded_in_mode(ForwardedHeaders::Rfc7239, CLIENT_CHAIN_REQUEST);
4031        let id = ctx.id.to_string();
4032        let traceparent = synthesised_traceparent_line(&ctx);
4033        assert_eq!(
4034            chained,
4035            format!(
4036                "GET / HTTP/1.1\r\nHost: example.com\r\n\
4037                 Forwarded: for=192.0.2.7, proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1\r\n\
4038                 X-Real-IP: 203.0.113.9\r\n\
4039                 X-Request-Id: client-chosen\r\n\
4040                 X-Real-IP: 10.0.0.1\r\n\
4041                 {traceparent}\
4042                 Sozu-Id: {id}\r\n\r\n"
4043            )
4044        );
4045        assert_eq!(ctx.xff_chain.as_deref(), Some("192.0.2.7"));
4046    }
4047
4048    /// `none` adds no forwarding header of either family and leaves every
4049    /// client one as sent, while `X-Real-IP` — governed by `send_x_real_ip`
4050    /// alone — is still injected.
4051    ///
4052    /// TO SEE THIS RED: in `HttpContext::on_request_headers`, drop the
4053    /// `emits_x_forwarded` guard on the `X-Forwarded-*` synthesis.
4054    #[test]
4055    fn forwarded_headers_none_adds_nothing_and_passes_client_headers_through() {
4056        let (bare, ctx) = forwarded_in_mode(ForwardedHeaders::None, BARE_REQUEST);
4057        let id = ctx.id.to_string();
4058        let traceparent = synthesised_traceparent_line(&ctx);
4059        assert_eq!(
4060            bare,
4061            format!(
4062                "GET / HTTP/1.1\r\nHost: example.com\r\n\
4063                 X-Real-IP: 10.0.0.1\r\n\
4064                 {traceparent}\
4065                 X-Request-Id: {id}\r\n\
4066                 Sozu-Id: {id}\r\n\r\n"
4067            )
4068        );
4069
4070        let (chained, ctx) = forwarded_in_mode(ForwardedHeaders::None, CLIENT_CHAIN_REQUEST);
4071        let id = ctx.id.to_string();
4072        let traceparent = synthesised_traceparent_line(&ctx);
4073        assert_eq!(
4074            chained,
4075            format!(
4076                "GET / HTTP/1.1\r\nHost: example.com\r\n\
4077                 X-Forwarded-For: 198.51.100.1\r\n\
4078                 X-Forwarded-For: 192.0.2.7\r\n\
4079                 Forwarded: for=192.0.2.7\r\n\
4080                 X-Forwarded-Proto: https\r\n\
4081                 X-Forwarded-Port: 443\r\n\
4082                 X-Forwarded-Host: public.example\r\n\
4083                 X-Real-IP: 203.0.113.9\r\n\
4084                 X-Request-Id: client-chosen\r\n\
4085                 X-Real-IP: 10.0.0.1\r\n\
4086                 {traceparent}\
4087                 Sozu-Id: {id}\r\n\r\n"
4088            )
4089        );
4090        assert_eq!(ctx.xff_chain.as_deref(), Some("192.0.2.7"));
4091    }
4092
4093    /// A client `Forwarded` value is only extended when it is a well-formed
4094    /// RFC 7239 §4 list. An unclosed quoted-string would otherwise swallow
4095    /// Sōzu's element into the client's value
4096    /// (`for="6.6.6.6, proto=http;for="10.0.0.1:54321";by=127.0.0.1`), and
4097    /// the backend would read neither the client's element nor Sōzu's.
4098    /// In `both` and `rfc7239` — the modes that extend the chain — a
4099    /// malformed line is elided (RFC 7239 §4 lets a proxy remove the field),
4100    /// so the element Sōzu writes is always well-formed and last; a
4101    /// well-formed earlier line is extended instead. `x_forwarded` and
4102    /// `none` do not touch `Forwarded`, so the line passes through as sent.
4103    ///
4104    /// TO SEE THIS RED: in `HttpContext::on_request_headers`, drop the
4105    /// `is_valid_forwarded` check on the client `Forwarded` line.
4106    #[test]
4107    fn a_malformed_client_forwarded_is_elided_where_the_chain_is_extended() {
4108        const MALFORMED: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
4109            Forwarded: for=\"6.6.6.6\r\n\r\n";
4110        const VALID_THEN_MALFORMED: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
4111            Forwarded: for=192.0.2.7\r\n\
4112            Forwarded: for=\"6.6.6.6\r\n\r\n";
4113        let forwarded_lines = |out: &str| -> Vec<String> {
4114            out.split("\r\n")
4115                .filter_map(|line| line.strip_prefix("Forwarded: "))
4116                .map(ToOwned::to_owned)
4117                .collect()
4118        };
4119        let hop = "proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1";
4120
4121        for mode in [ForwardedHeaders::Both, ForwardedHeaders::Rfc7239] {
4122            let (out, _) = forwarded_in_mode(mode, MALFORMED);
4123            assert_eq!(
4124                forwarded_lines(&out),
4125                vec![hop.to_owned()],
4126                "{mode:?}: {out}"
4127            );
4128            let (out, _) = forwarded_in_mode(mode, VALID_THEN_MALFORMED);
4129            let lines = forwarded_lines(&out);
4130            assert_eq!(
4131                lines,
4132                vec![format!("for=192.0.2.7, {hop}")],
4133                "{mode:?}: {out}"
4134            );
4135            assert!(
4136                lines.iter().all(|line| is_valid_forwarded(line.as_bytes())),
4137                "{mode:?}: the forwarded chain must be well-formed: {lines:?}"
4138            );
4139        }
4140        for mode in [ForwardedHeaders::XForwarded, ForwardedHeaders::None] {
4141            let (out, _) = forwarded_in_mode(mode, MALFORMED);
4142            assert_eq!(
4143                forwarded_lines(&out),
4144                vec!["for=\"6.6.6.6".to_owned()],
4145                "{mode:?} passes a client Forwarded through as sent: {out}"
4146            );
4147        }
4148    }
4149
4150    /// Read a process-local counter by raw key, treating an absent key as 0.
4151    /// `dump_local_proxy_metrics` does not drain, and `METRICS` is
4152    /// thread-local, so a test reads its own increments only.
4153    fn proxy_counter(key: &str) -> i64 {
4154        use sozu_command::proto::command::filtered_metrics::Inner;
4155        crate::metrics::METRICS.with(|metrics| {
4156            metrics
4157                .borrow_mut()
4158                .dump_local_proxy_metrics()
4159                .get(key)
4160                .and_then(|fm| fm.inner.as_ref())
4161                .and_then(|inner| match inner {
4162                    Inner::Count(v) => Some(*v),
4163                    _ => None,
4164                })
4165                .unwrap_or(0)
4166        })
4167    }
4168
4169    /// Each client `Forwarded` line Sōzu removes as malformed is counted
4170    /// once in `http.forwarded_malformed_elided`, so an operator behind a
4171    /// non-conforming upstream load balancer sees the chain being dropped
4172    /// without enabling `debug` logs. A well-formed line, and every line in
4173    /// `x_forwarded` and `none` (which pass `Forwarded` through), counts
4174    /// nothing.
4175    ///
4176    /// TO SEE THIS RED: in `HttpContext::on_request_headers`, drop the
4177    /// `incr!(names::http::FORWARDED_MALFORMED_ELIDED)` on the elision path.
4178    #[test]
4179    fn a_malformed_client_forwarded_elision_is_counted_once_per_line() {
4180        const MALFORMED: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
4181            Forwarded: for=\"6.6.6.6\r\n\r\n";
4182        const TWO_MALFORMED_AROUND_VALID: &[u8] = b"GET / HTTP/1.1\r\nHost: example.com\r\n\
4183            Forwarded: for=1.2.3.4:80\r\n\
4184            Forwarded: for=192.0.2.7\r\n\
4185            Forwarded: for=2001:db8::1\r\n\r\n";
4186        let elided = || proxy_counter(names::http::FORWARDED_MALFORMED_ELIDED);
4187
4188        for mode in [ForwardedHeaders::Both, ForwardedHeaders::Rfc7239] {
4189            let before = elided();
4190            forwarded_in_mode(mode, CLIENT_CHAIN_REQUEST);
4191            assert_eq!(elided(), before, "{mode:?}: a valid line is not counted");
4192
4193            let before = elided();
4194            forwarded_in_mode(mode, MALFORMED);
4195            assert_eq!(elided(), before + 1, "{mode:?}: one malformed line");
4196
4197            let before = elided();
4198            forwarded_in_mode(mode, TWO_MALFORMED_AROUND_VALID);
4199            assert_eq!(
4200                elided(),
4201                before + 2,
4202                "{mode:?}: each removed line counts once, the valid one not"
4203            );
4204        }
4205        for mode in [ForwardedHeaders::XForwarded, ForwardedHeaders::None] {
4206            let before = elided();
4207            forwarded_in_mode(mode, MALFORMED);
4208            forwarded_in_mode(mode, TWO_MALFORMED_AROUND_VALID);
4209            assert_eq!(
4210                elided(),
4211                before,
4212                "{mode:?} passes a client Forwarded through and counts nothing"
4213            );
4214        }
4215    }
4216
4217    /// `is_valid_forwarded` accepts the RFC 7239 §4 grammar — a list of
4218    /// `;`-separated `token=value` pairs, `value` a token or a quoted-string
4219    /// (RFC 9110 §5.6.4) — and refuses anything that would make a parser
4220    /// misread the element appended after it.
4221    #[test]
4222    fn is_valid_forwarded_follows_the_rfc7239_grammar() {
4223        for valid in [
4224            &b""[..],
4225            b"for=192.0.2.7",
4226            b"For=\"[2001:db8:cafe::17]:4711\"",
4227            b"for=192.0.2.60;proto=http;by=203.0.113.43",
4228            b"for=192.0.2.43, for=198.51.100.17",
4229            b"for=192.0.2.43,for=\"[2001:db8:cafe::17]\",for=unknown",
4230            b"for=_gazonk",
4231            b"for=\"\\\"quoted\\\"\"",
4232            b"for=a;",
4233            b"for=a ; by=b",
4234            b"for=a, , for=b",
4235            b"proto=http;for=\"10.0.0.1:54321\";by=127.0.0.1",
4236        ] {
4237            assert!(
4238                is_valid_forwarded(valid),
4239                "{:?} is valid",
4240                String::from_utf8_lossy(valid)
4241            );
4242        }
4243        for invalid in [
4244            &b"for=\"6.6.6.6"[..],
4245            b"for=\"6.6.6.6\\\"",
4246            b"for=",
4247            b"=192.0.2.7",
4248            b"for",
4249            b"for=a b",
4250            b"for=a\"b\"",
4251            b"for=\"a\"b",
4252            b"for = a",
4253            b"for=a\x01",
4254            b"for=\"a\x7f\"",
4255        ] {
4256            assert!(
4257                !is_valid_forwarded(invalid),
4258                "{:?} is invalid",
4259                String::from_utf8_lossy(invalid)
4260            );
4261        }
4262    }
4263
4264    /// The mode is listener-scoped: `reset` carries it to the next request
4265    /// of a keep-alive connection.
4266    #[test]
4267    fn forwarded_headers_survives_reset() {
4268        let mut ctx = make_context();
4269        ctx.forwarded_headers = ForwardedHeaders::Rfc7239;
4270        ctx.reset(Ulid::generate());
4271        assert_eq!(ctx.forwarded_headers, ForwardedHeaders::Rfc7239);
4272    }
4273
4274    // ── request framing: no Content-Length, no Transfer-Encoding ───────
4275
4276    /// Parse `bytes` as a `kind` message through the real parser and a fresh
4277    /// context, as `ConnectionH1::readable` does.
4278    fn parse_message(
4279        pool: &mut crate::pool::Pool,
4280        kind: kawa::Kind,
4281        bytes: &[u8],
4282    ) -> GenericHttpStream {
4283        let mut kawa: GenericHttpStream = kawa::Kawa::new(
4284            kind,
4285            kawa::Buffer::new(
4286                pool.checkout()
4287                    .expect("the test pool must hand out a buffer"),
4288            ),
4289        );
4290        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
4291        kawa.storage.fill(bytes.len());
4292        kawa::h1::parse(&mut kawa, &mut make_context());
4293        kawa
4294    }
4295
4296    /// The flag block kawa pushes right after the header callback, which
4297    /// every converter reads to decide whether the message ended.
4298    fn header_end_flags(kawa: &GenericHttpStream) -> kawa::Flags {
4299        kawa.blocks
4300            .iter()
4301            .find_map(|block| match block {
4302                kawa::Block::Flags(flags) if flags.end_header => Some(flags.clone()),
4303                _ => None,
4304            })
4305            .expect("a parsed message carries its end-of-headers flags")
4306    }
4307
4308    /// RFC 9112 §6.3 rule 7: a request with neither `Content-Length` nor
4309    /// `Transfer-Encoding` has a zero-length body, whatever its method or
4310    /// version. kawa 0.7.1 read such a request as close-delimited
4311    /// (`ParsingPhase::Body` on `BodySize::Empty`, whose arm takes every
4312    /// byte left in the buffer), so without `on_request_headers` ending it,
4313    /// the request pipelined behind it became a `Block::Chunk` of its body
4314    /// and was forwarded raw to the first request's backend (CWE-444).
4315    /// kawa >= 0.7.2 ends it itself (CleverCloud/kawa#27); the test pins the
4316    /// outcome, whichever layer ends it.
4317    ///
4318    /// TO SEE THIS RED: deleting the `ParsingPhase::Body` +
4319    /// `BodySize::Empty` branch of `on_request_headers` alone no longer
4320    /// turns it red, since kawa >= 0.7.2 terminates the request before that
4321    /// branch is reached. Pin
4322    /// kawa 0.7.1 (`kawa = { version = "=0.7.1", default-features = false }`
4323    /// in the root `Cargo.toml`, then `cargo update -p kawa --precise 0.7.1`)
4324    /// and delete the branch: every case then fails the first assertion,
4325    /// still in `Body`.
4326    #[test]
4327    fn a_request_without_length_ends_after_its_headers() {
4328        const NEXT: &[u8] = b"GET /next HTTP/1.1\r\nHost: example.com\r\n\r\n";
4329        let heads: [&[u8]; 7] = [
4330            b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n",
4331            b"HEAD / HTTP/1.1\r\nHost: example.com\r\n\r\n",
4332            b"DELETE /item HTTP/1.1\r\nHost: example.com\r\n\r\n",
4333            b"POST /form HTTP/1.1\r\nHost: example.com\r\n\r\n",
4334            b"POST /form HTTP/1.0\r\nHost: example.com\r\n\r\n",
4335            b"PUT /item HTTP/1.1\r\nHost: example.com\r\nExpect: 100-continue\r\n\r\n",
4336            b"GET /chat HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
4337        ];
4338        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4339        for head in heads {
4340            let label = String::from_utf8_lossy(head.split(|b| *b == b'\r').next().unwrap());
4341            let bytes = [head, NEXT].concat();
4342            let kawa = parse_message(&mut pool, kawa::Kind::Request, &bytes);
4343            assert!(
4344                kawa.is_terminated(),
4345                "{label}: must end after its headers, got {:?}",
4346                kawa.parsing_phase
4347            );
4348            let flags = header_end_flags(&kawa);
4349            assert!(
4350                flags.end_stream,
4351                "{label}: the end-of-headers flags must end the stream"
4352            );
4353            assert!(
4354                !kawa
4355                    .blocks
4356                    .iter()
4357                    .any(|block| matches!(block, kawa::Block::Chunk(_))),
4358                "{label}: no byte may be read as its body"
4359            );
4360            assert_eq!(
4361                kawa.storage.unparsed_data(),
4362                NEXT,
4363                "{label}: the pipelined request must stay unparsed for its own turn"
4364            );
4365        }
4366    }
4367
4368    /// The guard reads a missing length, not a present one: a request that
4369    /// declares its body keeps it, and a close-delimited RESPONSE (RFC 9112
4370    /// §6.3 rule 8) keeps reading until the close.
4371    #[test]
4372    fn a_framed_request_and_an_unframed_response_keep_their_bodies() {
4373        let mut pool = crate::pool::Pool::with_capacity(1, 3, 4096);
4374
4375        let length = parse_message(
4376            &mut pool,
4377            kawa::Kind::Request,
4378            b"POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 10\r\n\r\nHello",
4379        );
4380        assert_eq!(length.parsing_phase, kawa::ParsingPhase::Body);
4381        assert_eq!(length.expects, 5, "half of the declared body is still due");
4382
4383        let chunked = parse_message(
4384            &mut pool,
4385            kawa::Kind::Request,
4386            b"POST / HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n",
4387        );
4388        assert!(
4389            matches!(chunked.parsing_phase, kawa::ParsingPhase::Chunks { .. }),
4390            "a chunked request waits for its last chunk, got {:?}",
4391            chunked.parsing_phase
4392        );
4393
4394        let response = parse_message(
4395            &mut pool,
4396            kawa::Kind::Response,
4397            b"HTTP/1.1 200 OK\r\n\r\nclose-delimited",
4398        );
4399        assert_eq!(
4400            response.parsing_phase,
4401            kawa::ParsingPhase::Body,
4402            "a response without length is delimited by the close"
4403        );
4404        assert!(!header_end_flags(&response).end_stream);
4405        assert!(
4406            response
4407                .blocks
4408                .iter()
4409                .any(|block| matches!(block, kawa::Block::Chunk(_))),
4410            "the close-delimited response body is still read"
4411        );
4412    }
4413
4414    /// `pkawa::handle_header` calls the same `on_headers` for an H2 request
4415    /// BEFORE it resolves the framing: `body_size` is still `Empty` there
4416    /// for a request whose DATA frames follow, and it upgrades that to
4417    /// chunked afterwards unless the callback terminated the message. The
4418    /// guard keys on the `ParsingPhase::Body` kawa's H1 parser set for a
4419    /// close-delimited request up to kawa 0.7.1, which the H2 path never
4420    /// sets before the callback, so an H2 request is left for pkawa to frame.
4421    /// This test calls the callback directly, so it is unaffected by
4422    /// kawa >= 0.7.2 ending an H1 request itself.
4423    ///
4424    /// TO SEE THIS RED, key the guard on `body_size == BodySize::Empty`
4425    /// alone: this request is then terminated.
4426    #[test]
4427    fn an_h2_request_is_left_for_pkawa_to_frame() {
4428        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4429        let mut kawa: GenericHttpStream = kawa::Kawa::new(
4430            kawa::Kind::Request,
4431            kawa::Buffer::new(
4432                pool.checkout()
4433                    .expect("the test pool must hand out a buffer"),
4434            ),
4435        );
4436        kawa.detached.status_line = kawa::StatusLine::Request {
4437            version: kawa::Version::V20,
4438            method: kawa::Store::Static(b"POST"),
4439            uri: kawa::Store::Static(b"/upload"),
4440            authority: kawa::Store::Static(b"example.com"),
4441            path: kawa::Store::Static(b"/upload"),
4442        };
4443        let phase_before = kawa.parsing_phase;
4444        assert_eq!(kawa.body_size, kawa::BodySize::Empty, "premise");
4445
4446        kawa::h1::ParserCallbacks::on_headers(&mut make_context(), &mut kawa);
4447
4448        assert!(!kawa.is_terminated(), "an H2 request body may still follow");
4449        assert_eq!(kawa.parsing_phase, phase_before);
4450    }
4451
4452    /// A backend response parsed by kawa's H1 parser on a context the test
4453    /// keeps, so it can read what `on_response_headers` recorded.
4454    fn parse_response(
4455        pool: &mut crate::pool::Pool,
4456        bytes: &[u8],
4457    ) -> (GenericHttpStream, HttpContext) {
4458        let mut ctx = make_context();
4459        let mut kawa: GenericHttpStream = kawa::Kawa::new(
4460            kawa::Kind::Response,
4461            kawa::Buffer::new(
4462                pool.checkout()
4463                    .expect("the test pool must hand out a buffer"),
4464            ),
4465        );
4466        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
4467        kawa.storage.fill(bytes.len());
4468        kawa::h1::parse(&mut kawa, &mut ctx);
4469        assert!(!kawa.is_error(), "premise: the response must parse");
4470        (kawa, ctx)
4471    }
4472
4473    /// RFC 9110 §6.2: Sōzu answers its client in its own HTTP version, not
4474    /// the backend's. An HTTP/1.0 response that asked to persist
4475    /// (`Connection: keep-alive`, RFC 9112 §9.3) with a `Content-Length` is
4476    /// forwarded as HTTP/1.1 and keeps both connections.
4477    ///
4478    /// TO SEE THIS RED, delete the `Version::V10` branch of
4479    /// `HttpContext::on_response_headers`: the status line keeps `HTTP/1.0`.
4480    #[test]
4481    fn an_http10_keep_alive_response_is_forwarded_as_http11() {
4482        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4483        let (mut kawa, ctx) = parse_response(
4484            &mut pool,
4485            b"HTTP/1.0 200 OK\r\nContent-Length: 5\r\nConnection: keep-alive\r\n\r\nhello",
4486        );
4487        assert!(kawa.is_terminated(), "premise: the whole body was parsed");
4488        assert!(
4489            ctx.keep_alive_backend,
4490            "a keep-alive HTTP/1.0 response keeps the backend connection"
4491        );
4492        let wire = serialized_request(&mut kawa);
4493        assert!(
4494            wire.starts_with("HTTP/1.1 200 OK\r\n"),
4495            "the status line carries Sōzu's version, got {wire:?}"
4496        );
4497        assert!(
4498            !wire.contains("HTTP/1.0"),
4499            "the backend's version must not reach the client, got {wire:?}"
4500        );
4501        assert!(
4502            !wire.to_ascii_lowercase().contains("connection: close"),
4503            "a persistent response must not announce a close, got {wire:?}"
4504        );
4505        assert!(
4506            wire.ends_with("\r\n\r\nhello"),
4507            "the body follows, got {wire:?}"
4508        );
4509    }
4510
4511    /// An HTTP/1.0 response without `Connection: keep-alive` is not
4512    /// persistent (RFC 9112 §9.3): the backend closes after it. Forwarded as
4513    /// HTTP/1.1, the client would take it for persistent, so the response
4514    /// carries `Connection: close` (RFC 9112 §9.6) and `keep_alive_backend`
4515    /// is cleared, which closes the client connection after the response
4516    /// (`ConnectionH1::writable`) and lets the backend EOF end a
4517    /// close-delimited body (`ConnectionH1::terminate_close_delimited`).
4518    ///
4519    /// TO SEE THIS RED, delete the `Version::V10` branch of
4520    /// `HttpContext::on_response_headers`: the status line keeps `HTTP/1.0`
4521    /// and `keep_alive_backend` stays set.
4522    #[test]
4523    fn an_http10_response_without_keep_alive_is_forwarded_as_http11_with_close() {
4524        let cases: [(&str, &[u8], &str); 3] = [
4525            ("close-delimited", b"HTTP/1.0 200 OK\r\n\r\nabcd", "abcd"),
4526            (
4527                "content-length",
4528                b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\n\r\nabcd",
4529                "abcd",
4530            ),
4531            (
4532                // A close-delimited body ends only with the close (RFC 9112
4533                // §6.3 rule 8), whatever the backend claimed.
4534                "keep-alive but close-delimited",
4535                b"HTTP/1.0 200 OK\r\nConnection: keep-alive\r\n\r\nabcd",
4536                "abcd",
4537            ),
4538        ];
4539        for (label, bytes, body) in cases {
4540            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4541            let (mut kawa, ctx) = parse_response(&mut pool, bytes);
4542            assert!(
4543                !ctx.keep_alive_backend,
4544                "{label}: a non-persistent HTTP/1.0 response closes the backend"
4545            );
4546            let wire = serialized_request(&mut kawa);
4547            assert!(
4548                wire.starts_with("HTTP/1.1 200 OK\r\n"),
4549                "{label}: the status line carries Sōzu's version, got {wire:?}"
4550            );
4551            let (head, forwarded_body) = wire
4552                .split_once("\r\n\r\n")
4553                .expect("the response head ends with an empty line");
4554            let connection: Vec<&str> = head
4555                .split("\r\n")
4556                .filter(|line| line.to_ascii_lowercase().starts_with("connection:"))
4557                .collect();
4558            assert_eq!(
4559                connection,
4560                ["Connection: close"],
4561                "{label}: exactly one Connection: close announces the close, got {wire:?}"
4562            );
4563            assert_eq!(forwarded_body, body, "{label}: the body is forwarded");
4564        }
4565    }
4566
4567    /// The `Connection` lines of a serialised response head.
4568    fn connection_lines(wire: &str) -> Vec<&str> {
4569        let (head, _) = wire
4570            .split_once("\r\n\r\n")
4571            .expect("the response head ends with an empty line");
4572        head.split("\r\n")
4573            .filter(|line| line.to_ascii_lowercase().starts_with("connection:"))
4574            .collect()
4575    }
4576
4577    /// `close` is a connection option like any other (RFC 9110 §7.6.1): a
4578    /// list that carries it closes the backend connection, whatever else it
4579    /// lists and in whichever `Connection` line, for HTTP/1.1 as for HTTP/1.0.
4580    ///
4581    /// TO SEE THIS RED, compare the whole `Connection` value with `close` in
4582    /// `HttpContext::on_response_headers` instead of each of its options.
4583    #[test]
4584    fn a_close_option_in_a_connection_list_closes_the_backend() {
4585        let cases: [(&str, &[u8]); 3] = [
4586            (
4587                "HTTP/1.0 list",
4588                b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive, close\r\n\r\nabcd",
4589            ),
4590            (
4591                "HTTP/1.1 list",
4592                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive, Close \r\n\r\nabcd",
4593            ),
4594            (
4595                "HTTP/1.1 second line",
4596                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: x-custom\r\nConnection: close\r\n\r\nabcd",
4597            ),
4598        ];
4599        for (label, bytes) in cases {
4600            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4601            let (mut kawa, ctx) = parse_response(&mut pool, bytes);
4602            assert!(
4603                !ctx.keep_alive_backend,
4604                "{label}: a close option closes the backend"
4605            );
4606            let wire = serialized_request(&mut kawa);
4607            assert!(
4608                connection_lines(&wire)
4609                    .iter()
4610                    .any(|line| line.to_ascii_lowercase().contains("close")),
4611                "{label}: the close is still announced, got {wire:?}"
4612            );
4613        }
4614    }
4615
4616    /// A client request parsed by kawa's H1 parser on a context the test
4617    /// keeps, so it can read what `on_request_headers` recorded.
4618    fn parse_request(
4619        pool: &mut crate::pool::Pool,
4620        bytes: &[u8],
4621    ) -> (GenericHttpStream, HttpContext) {
4622        let mut ctx = make_context();
4623        let mut kawa: GenericHttpStream = kawa::Kawa::new(
4624            kawa::Kind::Request,
4625            kawa::Buffer::new(
4626                pool.checkout()
4627                    .expect("the test pool must hand out a buffer"),
4628            ),
4629        );
4630        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
4631        kawa.storage.fill(bytes.len());
4632        kawa::h1::parse(&mut kawa, &mut ctx);
4633        assert!(!kawa.is_error(), "premise: the request must parse");
4634        (kawa, ctx)
4635    }
4636
4637    /// `close` is a connection option of the request too (RFC 9110 §7.6.1),
4638    /// and a client that sends it asks for the connection to close after the
4639    /// response (RFC 9112 §9.6): a list that carries it clears
4640    /// `keep_alive_frontend`, whatever else it lists and in whichever
4641    /// `Connection` line, as `on_response_headers` already does for the
4642    /// backend. A token that merely starts with `close` is not the option,
4643    /// and the forwarded `Connection` lines are left as the client sent them.
4644    ///
4645    /// TO SEE THIS RED, compare the whole field value with `close` in
4646    /// `HttpContext::on_request_headers` instead of matching a list token:
4647    /// the three list cases keep the client connection.
4648    #[test]
4649    fn a_close_option_in_a_request_connection_list_closes_the_client() {
4650        let cases: [(&str, &[u8], bool); 7] = [
4651            (
4652                "single close",
4653                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n",
4654                false,
4655            ),
4656            (
4657                "keep-alive then close",
4658                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: keep-alive, close\r\n\r\n",
4659                false,
4660            ),
4661            (
4662                "close then TE",
4663                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close, TE\r\nTE: trailers\r\n\r\n",
4664                false,
4665            ),
4666            (
4667                "close listed in the second line",
4668                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: keep-alive\r\nConnection: TE, Close \r\nTE: trailers\r\n\r\n",
4669                false,
4670            ),
4671            (
4672                "closed is not close",
4673                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: closed\r\n\r\n",
4674                true,
4675            ),
4676            (
4677                "keep-alive only",
4678                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: keep-alive\r\n\r\n",
4679                true,
4680            ),
4681            (
4682                "no Connection",
4683                b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n",
4684                true,
4685            ),
4686        ];
4687        for (label, bytes, keeps) in cases {
4688            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4689            let (mut kawa, ctx) = parse_request(&mut pool, bytes);
4690            assert_eq!(
4691                ctx.keep_alive_frontend, keeps,
4692                "{label}: keep_alive_frontend"
4693            );
4694            assert!(
4695                ctx.keep_alive_backend,
4696                "{label}: the request never decides the backend's persistence"
4697            );
4698            let sent: Vec<String> = connection_lines(&String::from_utf8_lossy(bytes))
4699                .iter()
4700                .map(|line| line.to_ascii_lowercase().replace(' ', ""))
4701                .collect();
4702            let wire = serialized_request(&mut kawa);
4703            let forwarded: Vec<String> = connection_lines(&wire)
4704                .iter()
4705                .map(|line| line.to_ascii_lowercase().replace(' ', ""))
4706                .collect();
4707            assert_eq!(
4708                forwarded, sent,
4709                "{label}: the Connection lines are forwarded as sent, got {wire:?}"
4710            );
4711        }
4712    }
4713
4714    /// Removing a `Connection` header would turn the fields it nominates as
4715    /// hop-by-hop into end-to-end ones (RFC 9110 §7.6.1). A non-persistent
4716    /// HTTP/1.0 response keeps its other options and gains `close`; only
4717    /// `keep-alive`, which the close contradicts, is dropped.
4718    ///
4719    /// TO SEE THIS RED, elide the backend's `Connection` headers and push a
4720    /// bare `Connection: close` instead of extending the option list.
4721    #[test]
4722    fn a_non_persistent_http10_response_keeps_its_connection_options() {
4723        let cases: [(&str, &[u8], &str); 3] = [
4724            (
4725                "nominated field",
4726                b"HTTP/1.0 200 OK\r\nConnection: x-custom\r\nX-Custom: 1\r\n\r\nabcd",
4727                "Connection: x-custom, close",
4728            ),
4729            (
4730                "keep-alive dropped",
4731                b"HTTP/1.0 200 OK\r\nConnection: keep-alive, x-custom\r\n\r\nabcd",
4732                "Connection: x-custom, close",
4733            ),
4734            (
4735                "two lines merged",
4736                b"HTTP/1.0 200 OK\r\nConnection: x-one\r\nConnection: x-two\r\n\r\nabcd",
4737                "Connection: x-one, x-two, close",
4738            ),
4739        ];
4740        for (label, bytes, want) in cases {
4741            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4742            let (mut kawa, ctx) = parse_response(&mut pool, bytes);
4743            assert!(!ctx.keep_alive_backend, "{label}: premise, not persistent");
4744            let wire = serialized_request(&mut kawa);
4745            assert_eq!(
4746                connection_lines(&wire),
4747                [want],
4748                "{label}: the options survive beside close, got {wire:?}"
4749            );
4750        }
4751    }
4752
4753    /// RFC 9112 §6.1: an HTTP/1.0 message with `Transfer-Encoding` has
4754    /// faulty framing and its connection closes after it, `keep-alive` or
4755    /// not.
4756    ///
4757    /// TO SEE THIS RED, drop the `Transfer-Encoding` clause from the HTTP/1.0
4758    /// persistence test of `HttpContext::on_response_headers`.
4759    #[test]
4760    fn a_chunked_http10_response_closes_the_backend() {
4761        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4762        let (mut kawa, ctx) = parse_response(
4763            &mut pool,
4764            b"HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n4\r\nabcd\r\n0\r\n\r\n",
4765        );
4766        assert!(kawa.is_terminated(), "premise: the chunked body ended");
4767        assert!(
4768            !ctx.keep_alive_backend,
4769            "a chunked HTTP/1.0 response closes the backend"
4770        );
4771        let wire = serialized_request(&mut kawa);
4772        assert_eq!(
4773            connection_lines(&wire),
4774            ["Connection: close"],
4775            "got {wire:?}"
4776        );
4777    }
4778
4779    /// An interim response says nothing about the connection: a 101 from an
4780    /// HTTP/1.0 backend keeps its `Connection: Upgrade`, without which the
4781    /// client does not switch protocols (RFC 9110 §7.8).
4782    ///
4783    /// TO SEE THIS RED, apply the HTTP/1.0 persistence rewrite of
4784    /// `HttpContext::on_response_headers` to 1xx responses too.
4785    #[test]
4786    fn an_http10_101_keeps_its_connection_upgrade() {
4787        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4788        let (mut kawa, ctx) = parse_response(
4789            &mut pool,
4790            b"HTTP/1.0 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
4791        );
4792        assert!(
4793            ctx.keep_alive_backend,
4794            "a 101 leaves keep_alive_backend alone"
4795        );
4796        let wire = serialized_request(&mut kawa);
4797        assert!(
4798            wire.starts_with("HTTP/1.1 101 Switching Protocols\r\n"),
4799            "got {wire:?}"
4800        );
4801        assert_eq!(
4802            connection_lines(&wire),
4803            ["Connection: Upgrade"],
4804            "the upgrade must survive, got {wire:?}"
4805        );
4806    }
4807
4808    /// A 100 from an HTTP/1.0 backend leaves the persistence of the final
4809    /// response to that response: the backend context outlives the interim
4810    /// response, so a 100 that cleared `keep_alive_backend` would close a
4811    /// persistent final response.
4812    ///
4813    /// TO SEE THIS RED, apply the HTTP/1.0 persistence rewrite of
4814    /// `HttpContext::on_response_headers` to 1xx responses too.
4815    #[test]
4816    fn an_http10_100_leaves_persistence_to_the_final_response() {
4817        let mut pool = crate::pool::Pool::with_capacity(1, 2, 4096);
4818        let mut ctx = make_context();
4819        for (bytes, label) in [
4820            (&b"HTTP/1.0 100 Continue\r\n\r\n"[..], "100"),
4821            (
4822                &b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive\r\n\r\nabcd"[..],
4823                "final",
4824            ),
4825        ] {
4826            let mut kawa: GenericHttpStream = kawa::Kawa::new(
4827                kawa::Kind::Response,
4828                kawa::Buffer::new(
4829                    pool.checkout()
4830                        .expect("the test pool must hand out a buffer"),
4831                ),
4832            );
4833            kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
4834            kawa.storage.fill(bytes.len());
4835            kawa::h1::parse(&mut kawa, &mut ctx);
4836            assert!(kawa.is_terminated(), "{label}: premise, parsed whole");
4837            assert!(
4838                ctx.keep_alive_backend,
4839                "{label}: a 100 then a keep-alive response keep the backend"
4840            );
4841            let wire = serialized_request(&mut kawa);
4842            assert!(
4843                connection_lines(&wire)
4844                    .iter()
4845                    .all(|line| !line.to_ascii_lowercase().contains("close")),
4846                "{label}: no close is announced, got {wire:?}"
4847            );
4848        }
4849    }
4850
4851    /// A message parsed by kawa's H1 parser on a context that is shutting
4852    /// down (`HttpContext::closing`, set by `Mux::shutting_down_inner`,
4853    /// `lib/src/protocol/mux/mod.rs`), which the test keeps.
4854    fn parse_closing(
4855        pool: &mut crate::pool::Pool,
4856        kind: kawa::Kind,
4857        bytes: &[u8],
4858    ) -> (GenericHttpStream, HttpContext) {
4859        let mut ctx = make_context();
4860        ctx.closing = true;
4861        let mut kawa: GenericHttpStream = kawa::Kawa::new(
4862            kind,
4863            kawa::Buffer::new(
4864                pool.checkout()
4865                    .expect("the test pool must hand out a buffer"),
4866            ),
4867        );
4868        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
4869        kawa.storage.fill(bytes.len());
4870        kawa::h1::parse(&mut kawa, &mut ctx);
4871        assert!(!kawa.is_error(), "premise: the message must parse");
4872        (kawa, ctx)
4873    }
4874
4875    /// The head of a serialised message, lower-cased, one entry per line.
4876    fn head_lines(wire: &str) -> Vec<String> {
4877        let (head, _) = wire
4878            .split_once("\r\n\r\n")
4879            .expect("the head ends with an empty line");
4880        head.split("\r\n")
4881            .map(|line| line.to_ascii_lowercase())
4882            .collect()
4883    }
4884
4885    /// While Sōzu shuts down, the request it forwards announces the close
4886    /// (RFC 9112 §9.6) without dropping the other options the client listed:
4887    /// removing one would turn the field it nominates as hop-by-hop into an
4888    /// end-to-end one (RFC 9110 §7.6.1). The options are merged into one
4889    /// line ending in a single `close`, `keep-alive` is dropped as the close
4890    /// contradicts it, and a client `close` still clears
4891    /// `keep_alive_frontend`.
4892    ///
4893    /// TO SEE THIS RED, restore the whole-value overwrite of the `Connection`
4894    /// field under `self.closing` in `HttpContext::on_request_headers`.
4895    #[test]
4896    fn a_shutting_down_request_keeps_its_connection_options() {
4897        let cases: [(&str, &[u8], &str, bool); 6] = [
4898            (
4899                "no Connection",
4900                b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n",
4901                "Connection: close",
4902                true,
4903            ),
4904            (
4905                "nominated field",
4906                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: x-custom\r\nX-Custom: 1\r\n\r\n",
4907                "Connection: x-custom, close",
4908                true,
4909            ),
4910            (
4911                "keep-alive dropped",
4912                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: keep-alive, x-custom\r\n\r\n",
4913                "Connection: x-custom, close",
4914                true,
4915            ),
4916            (
4917                "single close not duplicated",
4918                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n",
4919                "Connection: close",
4920                false,
4921            ),
4922            (
4923                "two lines merged",
4924                b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: x-one\r\nConnection: TE, Close \r\nTE: trailers\r\n\r\n",
4925                "Connection: x-one, TE, close",
4926                false,
4927            ),
4928            (
4929                "HTTP/1.0",
4930                b"GET / HTTP/1.0\r\nHost: example.com\r\nConnection: keep-alive, x-custom\r\n\r\n",
4931                "Connection: x-custom, close",
4932                true,
4933            ),
4934        ];
4935        for (label, bytes, want, keeps) in cases {
4936            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4937            let (mut kawa, ctx) = parse_closing(&mut pool, kawa::Kind::Request, bytes);
4938            assert_eq!(
4939                ctx.keep_alive_frontend, keeps,
4940                "{label}: a client close is still recorded"
4941            );
4942            let wire = serialized_request(&mut kawa);
4943            assert_eq!(
4944                connection_lines(&wire),
4945                [want],
4946                "{label}: the options survive beside one close, got {wire:?}"
4947            );
4948        }
4949    }
4950
4951    /// A shutting-down Sōzu refuses a protocol upgrade: it closes the client
4952    /// connection as soon as the response is written
4953    /// (`ConnectionH1::writable`, `lib/src/protocol/mux/h1.rs`, tests
4954    /// `closing` before it handles a 101), so a 101 would hand the client a
4955    /// switched protocol on a connection that is already closing. The
4956    /// `upgrade` option is dropped with the `Upgrade` field it nominates
4957    /// (RFC 9110 §7.6.1), so the backend answers in HTTP/1.1, which it may
4958    /// always do (RFC 9110 §7.8), and the client retries the upgrade on a
4959    /// new connection. The other options are kept.
4960    ///
4961    /// TO SEE THIS RED, keep the `upgrade` option or the `Upgrade` field in
4962    /// the shutdown merge of `HttpContext::on_request_headers`.
4963    #[test]
4964    fn a_shutting_down_request_does_not_upgrade() {
4965        let cases: [(&str, &[u8], &str); 2] = [
4966            (
4967                "websocket",
4968                b"GET /chat HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
4969                "Connection: close",
4970            ),
4971            (
4972                "upgrade among other options",
4973                b"GET /chat HTTP/1.1\r\nHost: example.com\r\nConnection: keep-alive, Upgrade, x-custom\r\nUpgrade: websocket\r\nX-Custom: 1\r\n\r\n",
4974                "Connection: x-custom, close",
4975            ),
4976        ];
4977        for (label, bytes, want) in cases {
4978            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
4979            let (mut kawa, _ctx) = parse_closing(&mut pool, kawa::Kind::Request, bytes);
4980            let wire = serialized_request(&mut kawa);
4981            assert_eq!(
4982                connection_lines(&wire),
4983                [want],
4984                "{label}: the upgrade is not requested, got {wire:?}"
4985            );
4986            assert!(
4987                !head_lines(&wire)
4988                    .iter()
4989                    .any(|line| line.starts_with("upgrade:")),
4990                "{label}: the Upgrade field is not forwarded, got {wire:?}"
4991            );
4992        }
4993    }
4994
4995    /// While Sōzu shuts down, the final response it forwards announces the
4996    /// close (RFC 9112 §9.6), in HTTP/1.1 as in HTTP/1.0, and keeps the
4997    /// options the backend listed (RFC 9110 §7.6.1) apart from the
4998    /// `keep-alive` the close contradicts. The backend's own `close` is
4999    /// still recorded in `keep_alive_backend`, so that connection is not
5000    /// pooled.
5001    ///
5002    /// TO SEE THIS RED, restore the whole-value overwrite of the `Connection`
5003    /// field under `self.closing` in `HttpContext::on_response_headers`.
5004    #[test]
5005    fn a_shutting_down_response_keeps_its_connection_options() {
5006        let cases: [(&str, &[u8], &str, bool); 6] = [
5007            (
5008                "no Connection",
5009                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nabcd",
5010                "Connection: close",
5011                true,
5012            ),
5013            (
5014                "nominated field",
5015                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: x-custom\r\nX-Custom: 1\r\n\r\nabcd",
5016                "Connection: x-custom, close",
5017                true,
5018            ),
5019            (
5020                "keep-alive dropped",
5021                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive, x-custom\r\n\r\nabcd",
5022                "Connection: x-custom, close",
5023                true,
5024            ),
5025            (
5026                "single close not duplicated",
5027                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\nabcd",
5028                "Connection: close",
5029                false,
5030            ),
5031            (
5032                "two lines merged",
5033                b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: x-one\r\nConnection: x-two, Close \r\n\r\nabcd",
5034                "Connection: x-one, x-two, close",
5035                false,
5036            ),
5037            (
5038                "persistent HTTP/1.0",
5039                b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive, x-custom\r\n\r\nabcd",
5040                "Connection: x-custom, close",
5041                true,
5042            ),
5043        ];
5044        for (label, bytes, want, keeps) in cases {
5045            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5046            let (mut kawa, ctx) = parse_closing(&mut pool, kawa::Kind::Response, bytes);
5047            assert_eq!(
5048                ctx.keep_alive_backend, keeps,
5049                "{label}: the backend's close is still recorded"
5050            );
5051            let wire = serialized_request(&mut kawa);
5052            assert!(wire.starts_with("HTTP/1.1 200 OK\r\n"), "got {wire:?}");
5053            assert_eq!(
5054                connection_lines(&wire),
5055                [want],
5056                "{label}: the options survive beside one close, got {wire:?}"
5057            );
5058        }
5059    }
5060
5061    /// An interim response says nothing about the connection, shutdown or
5062    /// not: the final response carries the close, and a 101 keeps its
5063    /// `Connection: Upgrade` (RFC 9110 §7.8).
5064    ///
5065    /// TO SEE THIS RED, apply the shutdown merge of
5066    /// `HttpContext::on_response_headers` to 1xx responses too.
5067    #[test]
5068    fn a_shutting_down_interim_response_is_left_alone() {
5069        let cases: [(&str, &[u8], &[&str]); 2] = [
5070            (
5071                "101",
5072                b"HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
5073                &["Connection: Upgrade"],
5074            ),
5075            ("100", b"HTTP/1.1 100 Continue\r\n\r\n", &[]),
5076        ];
5077        for (label, bytes, want) in cases {
5078            let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5079            let (mut kawa, ctx) = parse_closing(&mut pool, kawa::Kind::Response, bytes);
5080            assert!(
5081                ctx.keep_alive_backend,
5082                "{label}: an interim response leaves keep_alive_backend alone"
5083            );
5084            let wire = serialized_request(&mut kawa);
5085            assert_eq!(
5086                connection_lines(&wire),
5087                want,
5088                "{label}: the interim Connection lines are forwarded as sent, got {wire:?}"
5089            );
5090        }
5091    }
5092
5093    // ── chunked request trailers: elision and field bound ──────────────
5094
5095    /// Filter the trailer fields of `kawa` appended after `first_new_block`
5096    /// with a fresh context, and return how many spoof vectors were elided.
5097    fn elide_spoof_vectors(kawa: &mut GenericHttpStream, first_new_block: usize) -> usize {
5098        make_context()
5099            .filter_request_trailers(kawa, first_new_block)
5100            .spoof_vectors
5101    }
5102
5103    /// Head and body of a chunked request, up to and including the
5104    /// last-chunk line, so a test can send its trailer section separately.
5105    const CHUNKED_HEAD: &[u8] = b"POST /upload HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\n";
5106
5107    /// A trailer section carrying every spoof-vector name, in mixed case,
5108    /// with the forged value `6.6.6.6`, plus one legitimate field.
5109    const SPOOF_TRAILERS: &[u8] = b"X-Forwarded-For: 6.6.6.6\r\nforwarded: for=6.6.6.6\r\nX-REAL-IP: 6.6.6.6\r\nX-Request-Id: 6.6.6.6\r\nX-Forwarded-Proto: 6.6.6.6\r\nx-forwarded-port: 6.6.6.6\r\nX-Forwarded-Host: 6.6.6.6\r\nGrpc-Status: 0\r\n\r\n";
5110
5111    /// RFC 9110 §6.5.1, sozu-proxy/sozu#1689: an H1 chunked request cannot
5112    /// carry `X-Forwarded-For`, `Forwarded`, `X-Real-IP` or the rest of
5113    /// `TRAILER_SPOOF_VECTOR_HEADERS` in its trailer section, whatever the
5114    /// case of the name, while a legitimate trailer and the chunk framing
5115    /// survive: last-chunk, the surviving field, then the empty line.
5116    ///
5117    /// TO SEE THIS RED: make `HttpContext::filter_request_trailers` return
5118    /// before its walk.
5119    #[test]
5120    fn a_chunked_request_trailer_section_loses_its_spoof_vector_fields() {
5121        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5122        let bytes = [CHUNKED_HEAD, SPOOF_TRAILERS].concat();
5123        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, &bytes);
5124        assert!(kawa.is_terminated(), "the trailer section ends the request");
5125
5126        assert_eq!(
5127            elide_spoof_vectors(&mut kawa, 0),
5128            TRAILER_SPOOF_VECTOR_HEADERS.len(),
5129            "every spoof-vector trailer field is elided once"
5130        );
5131        assert_eq!(
5132            elide_spoof_vectors(&mut kawa, 0),
5133            0,
5134            "a second walk finds nothing left to elide"
5135        );
5136        let wire = serialized_request(&mut kawa);
5137        assert!(
5138            !wire.contains("6.6.6.6"),
5139            "no spoofed trailer may reach the backend, got {wire:?}"
5140        );
5141        assert!(
5142            wire.ends_with("\r\n\r\n5\r\nHello\r\n0\r\nGrpc-Status: 0\r\n\r\n"),
5143            "the legitimate trailer and the chunk framing survive, got {wire:?}"
5144        );
5145    }
5146
5147    /// The trailer section may arrive after the last-chunk line was already
5148    /// forwarded, so the `end_body` marker is no longer in the block queue
5149    /// when the trailer fields are parsed. They must still be elided, and a
5150    /// request whose every trailer field is dropped still ends with the
5151    /// empty line that closes its (now empty) trailer section.
5152    #[test]
5153    fn a_trailer_section_parsed_after_the_last_chunk_was_forwarded_is_filtered() {
5154        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5155        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, CHUNKED_HEAD);
5156        assert_eq!(kawa.parsing_phase, kawa::ParsingPhase::Trailers);
5157        assert_eq!(elide_spoof_vectors(&mut kawa, 0), 0);
5158        let head = serialized_request(&mut kawa);
5159        assert!(
5160            kawa.blocks.is_empty(),
5161            "the last-chunk marker was forwarded"
5162        );
5163
5164        const ONLY_SPOOF: &[u8] = b"X-Forwarded-For: 6.6.6.6\r\nForwarded: for=6.6.6.6\r\n\r\n";
5165        kawa.storage.space()[..ONLY_SPOOF.len()].copy_from_slice(ONLY_SPOOF);
5166        kawa.storage.fill(ONLY_SPOOF.len());
5167        let before = kawa.blocks.len();
5168        kawa::h1::parse(&mut kawa, &mut make_context());
5169        assert!(kawa.is_terminated(), "the trailer section ends the request");
5170        assert_eq!(elide_spoof_vectors(&mut kawa, before), 2);
5171        let wire = serialized_request(&mut kawa);
5172        assert!(
5173            wire.starts_with(&head) && !wire.contains("6.6.6.6"),
5174            "no spoofed trailer may reach the backend, got {wire:?}"
5175        );
5176        assert!(
5177            wire.ends_with("0\r\n\r\n"),
5178            "the empty trailer section is still closed, got {wire:?}"
5179        );
5180    }
5181
5182    /// The elision is confined to a chunked request's trailer section: the
5183    /// header block of a request, chunked or not, and a response trailer
5184    /// are left to their own paths, and the walk returns before touching a
5185    /// block.
5186    #[test]
5187    fn trailer_elision_leaves_headers_and_responses_alone() {
5188        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5189        let cases: [(kawa::Kind, &[u8]); 3] = [
5190            (
5191                kawa::Kind::Request,
5192                b"POST / HTTP/1.1\r\nHost: example.com\r\nX-Request-Id: keep\r\nContent-Length: 2\r\n\r\nok",
5193            ),
5194            (
5195                kawa::Kind::Request,
5196                b"POST / HTTP/1.1\r\nHost: example.com\r\nX-Request-Id: keep\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nok\r\n",
5197            ),
5198            (
5199                kawa::Kind::Response,
5200                b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nX-Request-Id: keep\r\n\r\n",
5201            ),
5202        ];
5203        for (kind, bytes) in cases {
5204            let mut kawa = parse_message(&mut pool, kind, bytes);
5205            assert_eq!(
5206                elide_spoof_vectors(&mut kawa, 0),
5207                0,
5208                "{kind:?} {:?}: nothing to elide",
5209                kawa.parsing_phase
5210            );
5211            let wire = serialized_request(&mut kawa);
5212            assert!(
5213                wire.contains("X-Request-Id: keep\r\n"),
5214                "{kind:?}: the field survives, got {wire:?}"
5215            );
5216        }
5217    }
5218
5219    /// Append `bytes` to `kawa`, parse them, and filter the trailer fields
5220    /// that parse appended with `context`, as `ConnectionH1::readable` does.
5221    fn feed_and_filter(
5222        context: &mut HttpContext,
5223        kawa: &mut GenericHttpStream,
5224        bytes: &[u8],
5225    ) -> TrailerFilterOutcome {
5226        kawa.storage.space()[..bytes.len()].copy_from_slice(bytes);
5227        kawa.storage.fill(bytes.len());
5228        let before = kawa.blocks.len();
5229        kawa::h1::parse(kawa, &mut make_context());
5230        context.filter_request_trailers(kawa, before)
5231    }
5232
5233    /// [`feed_and_filter`] with a fresh context, returning how many spoof
5234    /// vectors were elided.
5235    fn feed_and_elide(kawa: &mut GenericHttpStream, bytes: &[u8]) -> usize {
5236        feed_and_filter(&mut make_context(), kawa, bytes).spoof_vectors
5237    }
5238
5239    /// The trailer section itself may be split across reads, and the first
5240    /// part forwarded before the second arrives: `X-Forwarded-For` comes in
5241    /// one segment and is drained towards the backend, `Forwarded` in the
5242    /// next. Both must be elided, and the legitimate field and the closing
5243    /// empty line of the section survive.
5244    ///
5245    /// TO SEE THIS RED: make `HttpContext::filter_request_trailers` return
5246    /// before its walk.
5247    #[test]
5248    fn a_trailer_section_split_across_reads_is_filtered_in_each_part() {
5249        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5250        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, CHUNKED_HEAD);
5251        assert_eq!(kawa.parsing_phase, kawa::ParsingPhase::Trailers);
5252
5253        assert_eq!(
5254            feed_and_elide(&mut kawa, b"X-Forwarded-For: 6.6.6.6\r\n"),
5255            1,
5256            "the first trailer part is filtered on its own"
5257        );
5258        assert_eq!(kawa.parsing_phase, kawa::ParsingPhase::Trailers);
5259        let first = serialized_request(&mut kawa);
5260        assert!(kawa.blocks.is_empty(), "the first part was forwarded");
5261
5262        assert_eq!(
5263            feed_and_elide(
5264                &mut kawa,
5265                b"Forwarded: for=6.6.6.6\r\nGrpc-Status: 0\r\n\r\n"
5266            ),
5267            1,
5268            "the second trailer part is filtered on its own"
5269        );
5270        assert!(kawa.is_terminated(), "the trailer section ends the request");
5271        let wire = serialized_request(&mut kawa);
5272        assert!(
5273            wire.starts_with(&first) && !wire.contains("6.6.6.6"),
5274            "no spoofed trailer may reach the backend, got {wire:?}"
5275        );
5276        assert!(
5277            wire.ends_with("0\r\nGrpc-Status: 0\r\n\r\n"),
5278            "the legitimate trailer and the closing empty line survive, got {wire:?}"
5279        );
5280    }
5281
5282    /// Each call walks only the blocks the last parse appended, so a client
5283    /// trickling one trailer line per segment while the backend is not
5284    /// writable costs a linear walk in total, not a quadratic one. A field
5285    /// queued before the parse is left for the call that followed the parse
5286    /// which queued it: here it was deliberately never filtered, and the
5287    /// bounded walk must not reach it.
5288    ///
5289    /// TO SEE THIS RED: ignore `first_new_block` in
5290    /// `HttpContext::filter_request_trailers` and walk the whole queue
5291    /// (`range_mut(0..)`); the call then also elides the older field.
5292    #[test]
5293    fn a_trailer_walk_stops_at_the_blocks_queued_before_the_parse() {
5294        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5295        let head = [CHUNKED_HEAD, b"X-Forwarded-For: 6.6.6.6\r\n"].concat();
5296        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, &head);
5297        assert_eq!(kawa.parsing_phase, kawa::ParsingPhase::Trailers);
5298        let queued = kawa.blocks.len();
5299
5300        // Trickle the rest of the section one line per read, never
5301        // draining: each call must see exactly the one new field.
5302        for line in [
5303            &b"Forwarded: for=6.6.6.6\r\n"[..],
5304            b"X-Real-IP: 6.6.6.6\r\n",
5305            b"Grpc-Status: 0\r\n",
5306        ] {
5307            let spoofed = usize::from(!line.starts_with(b"Grpc"));
5308            assert_eq!(
5309                feed_and_elide(&mut kawa, line),
5310                spoofed,
5311                "only the field this read appended is examined"
5312            );
5313        }
5314        assert_eq!(feed_and_elide(&mut kawa, b"\r\n"), 0);
5315        assert!(kawa.is_terminated(), "the trailer section ends the request");
5316
5317        let buf = kawa.storage.buffer();
5318        let older = kawa
5319            .blocks
5320            .range(..queued)
5321            .filter_map(|block| match block {
5322                kawa::Block::Header(pair) if !pair.is_elided() => Some(pair.key.data(buf)),
5323                _ => None,
5324            })
5325            .collect::<Vec<_>>();
5326        assert!(
5327            older
5328                .iter()
5329                .any(|key| key.eq_ignore_ascii_case(b"X-Forwarded-For")),
5330            "a block queued before the parse is outside the walk, got {older:?}"
5331        );
5332    }
5333
5334    /// A trailer section carrying one field of each RFC 9110 §6.5.1
5335    /// category in mixed case, with the forged value `6.6.6.6`, plus two
5336    /// legitimate fields that must survive.
5337    const FORBIDDEN_TRAILERS: &[u8] = b"Content-Length: 6666\r\nTRANSFER-ENCODING: 6.6.6.6\r\nHost: 6.6.6.6\r\nExpect: 6.6.6.6\r\nIf-Match: 6.6.6.6\r\nauthorization: 6.6.6.6\r\nCookie: 6.6.6.6\r\nContent-Type: 6.6.6.6\r\nTrailer: 6.6.6.6\r\nConnection: 6.6.6.6\r\nGrpc-Status: 0\r\nX-Checksum: abc\r\n\r\n";
5338
5339    /// RFC 9110 §6.5.1, sozu-proxy/sozu#1701: an H1 chunked request cannot
5340    /// carry a framing, routing, request-modifier, authentication, content
5341    /// processing or connection-specific field in its trailer section,
5342    /// whatever the case of the name; legitimate trailers and the chunk
5343    /// framing survive.
5344    ///
5345    /// TO SEE THIS RED: drop the `is_trailer_forbidden_field` branch of
5346    /// `HttpContext::filter_request_trailers`.
5347    #[test]
5348    fn a_chunked_request_trailer_section_loses_its_forbidden_fields() {
5349        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5350        let bytes = [CHUNKED_HEAD, FORBIDDEN_TRAILERS].concat();
5351        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, &bytes);
5352        assert!(kawa.is_terminated(), "the trailer section ends the request");
5353
5354        let mut context = make_context();
5355        let outcome = context.filter_request_trailers(&mut kawa, 0);
5356        assert_eq!(
5357            outcome,
5358            TrailerFilterOutcome {
5359                spoof_vectors: 0,
5360                forbidden: 10,
5361                over_limit: false,
5362            },
5363            "every forbidden trailer field is elided once"
5364        );
5365        assert_eq!(context.trailer_fields, 12, "every field is counted");
5366        assert!(!kawa.is_error(), "a section within the bound is admitted");
5367        let wire = serialized_request(&mut kawa);
5368        // The forged `Content-Length` is matched on its whole field line: a
5369        // bare `6666` also occurs in the random `traceparent` and request ids.
5370        assert!(
5371            !wire.contains("6.6.6.6")
5372                && !wire
5373                    .to_ascii_lowercase()
5374                    .contains("\r\ncontent-length: 6666\r\n"),
5375            "no forbidden trailer may reach the backend, got {wire:?}"
5376        );
5377        assert!(
5378            wire.ends_with("\r\n\r\n5\r\nHello\r\n0\r\nGrpc-Status: 0\r\nX-Checksum: abc\r\n\r\n"),
5379            "the legitimate trailers and the chunk framing survive, got {wire:?}"
5380        );
5381    }
5382
5383    /// Every name of `TRAILER_FORBIDDEN_FIELDS` is recognised in upper case
5384    /// too, and no spoof vector is double-listed in it, so each elided field
5385    /// lands in exactly one metric.
5386    #[test]
5387    fn trailer_forbidden_fields_are_matched_without_case_and_disjoint_from_spoof_vectors() {
5388        for name in TRAILER_FORBIDDEN_FIELDS {
5389            assert!(
5390                is_trailer_forbidden_field(&name.to_ascii_uppercase()),
5391                "{:?} is matched in upper case",
5392                String::from_utf8_lossy(name)
5393            );
5394            assert!(
5395                !is_trailer_spoof_vector(name),
5396                "{:?} is in both lists",
5397                String::from_utf8_lossy(name)
5398            );
5399        }
5400        for legit in [
5401            &b"grpc-status"[..],
5402            b"grpc-message",
5403            b"x-checksum",
5404            b"digest",
5405        ] {
5406            assert!(!is_trailer_forbidden_field(legit));
5407        }
5408    }
5409
5410    /// Build a trailer section of `count` distinct legitimate fields.
5411    fn numbered_trailers(count: usize) -> Vec<u8> {
5412        let mut bytes = Vec::new();
5413        for i in 0..count {
5414            bytes.extend_from_slice(format!("X-T{i}: {i}\r\n").as_bytes());
5415        }
5416        bytes.extend_from_slice(b"\r\n");
5417        bytes
5418    }
5419
5420    /// sozu-proxy/sozu#1701: a trailer section may carry up to
5421    /// `max_trailer_fields` fields, the bound `h2_max_header_fields` puts on
5422    /// an H2 trailer block; one more marks the request in error, so the
5423    /// caller answers 400 before any of it is forwarded. Elided fields count
5424    /// too, as they do on the H2 path.
5425    ///
5426    /// TO SEE THIS RED: drop the `max_trailer_fields` comparison of
5427    /// `HttpContext::filter_request_trailers`.
5428    #[test]
5429    fn a_trailer_section_over_the_field_bound_is_an_error() {
5430        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5431        for (count, over) in [(4, false), (5, true)] {
5432            let bytes = [CHUNKED_HEAD, &numbered_trailers(count)].concat();
5433            let mut kawa = parse_message(&mut pool, kawa::Kind::Request, &bytes);
5434            assert!(kawa.is_terminated());
5435            let mut context = make_context();
5436            context.max_trailer_fields = 4;
5437            let outcome = context.filter_request_trailers(&mut kawa, 0);
5438            assert_eq!(
5439                outcome.over_limit, over,
5440                "{count} fields against a bound of 4"
5441            );
5442            assert_eq!(kawa.is_error(), over, "{count} fields against a bound of 4");
5443        }
5444
5445        // An elided field still counts.
5446        let bytes = [
5447            CHUNKED_HEAD,
5448            b"Host: a\r\nHost: b\r\nX-Forwarded-For: c\r\n\r\n".as_slice(),
5449        ]
5450        .concat();
5451        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, &bytes);
5452        let mut context = make_context();
5453        context.max_trailer_fields = 2;
5454        let outcome = context.filter_request_trailers(&mut kawa, 0);
5455        assert!(outcome.over_limit && kawa.is_error(), "elided fields count");
5456    }
5457
5458    /// The bound covers the whole trailer section, not one read: a client
5459    /// trickling one field per segment, each forwarded before the next
5460    /// arrives, is still refused once the running count passes the bound,
5461    /// and `reset` gives the next pipelined request a budget of its own.
5462    ///
5463    /// TO SEE THIS RED: count only the fields of the current walk (compare
5464    /// `fields` instead of `self.trailer_fields` to the bound).
5465    #[test]
5466    fn the_trailer_field_bound_spans_reads_and_resets_per_request() {
5467        let mut pool = crate::pool::Pool::with_capacity(1, 1, 4096);
5468        let mut kawa = parse_message(&mut pool, kawa::Kind::Request, CHUNKED_HEAD);
5469        let mut context = make_context();
5470        context.max_trailer_fields = 2;
5471        assert_eq!(
5472            context.filter_request_trailers(&mut kawa, 0),
5473            Default::default()
5474        );
5475        for (i, line) in [&b"X-A: 1\r\n"[..], b"X-B: 2\r\n"].into_iter().enumerate() {
5476            let outcome = feed_and_filter(&mut context, &mut kawa, line);
5477            assert!(!outcome.over_limit && !kawa.is_error(), "field {i} fits");
5478            serialized_request(&mut kawa);
5479            assert!(kawa.blocks.is_empty(), "field {i} was forwarded");
5480        }
5481        let outcome = feed_and_filter(&mut context, &mut kawa, b"X-C: 3\r\n");
5482        assert!(
5483            outcome.over_limit && kawa.is_error(),
5484            "the third field goes over"
5485        );
5486
5487        context.reset(Ulid::generate());
5488        assert_eq!(context.trailer_fields, 0, "reset clears the running count");
5489        assert_eq!(context.max_trailer_fields, 2, "reset keeps the bound");
5490    }
5491}