Skip to main content

sozu_command_lib/
secret.rs

1//! Protected health-check header values and their keyed fingerprints.
2//!
3//! A health-check header value is a credential more often than not
4//! (`Authorization`). It lives in memory only, in a [`SecretValue`](crate::secret::SecretValue) that is
5//! neither `Clone`, `Serialize`, `Display` nor value-revealing `Debug`, and
6//! whose buffer is zeroed when it drops. Anything that has to show or compare
7//! a value shows the header name plus a
8//! [`HeaderFingerprint`](crate::secret::HeaderFingerprint): a keyed HMAC-SHA256
9//! under a [`FingerprintKey`](crate::secret::FingerprintKey) the process
10//! generates for itself.
11//!
12//! Zeroing on drop covers the buffer this module owns. It does not reach what
13//! the allocator, the kernel, a socket buffer or swap may already have copied.
14
15use std::{
16    collections::HashMap,
17    fmt,
18    io::{ErrorKind, Read},
19    str::FromStr,
20    sync::Arc,
21};
22
23use hmac::{Hmac, KeyInit, Mac};
24use prost::{
25    DecodeError, Message,
26    bytes::{Buf, BufMut},
27    encoding::{
28        DecodeContext, WireType, bytes, check_wire_type, decode_key, decode_varint, encode_varint,
29        message, skip_field, string,
30    },
31};
32use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _, ser::Error as _};
33use sha2::Sha256;
34pub use zeroize::{Zeroize, Zeroizing};
35
36use crate::{
37    config::{
38        check_health_check_header_value, validate_health_check_config,
39        validate_health_check_header_values,
40    },
41    proto::command::{
42        HealthCheckConfig, HealthCheckHeader, Request, SetProtectedHealthCheck,
43        request::RequestType,
44    },
45};
46
47type HmacSha256 = Hmac<Sha256>;
48
49/// Algorithm label of a [`HeaderFingerprint`] and of its MAC input.
50const FINGERPRINT_ALGORITHM: &str = "sozu-hc-hmac-sha256";
51/// Format version of a [`HeaderFingerprint`] and of its MAC input.
52const FINGERPRINT_VERSION: &str = "v1";
53/// Bytes of the HMAC key.
54const KEY_LEN: usize = 32;
55/// Bytes of the public epoch that names a key without revealing it.
56const EPOCH_LEN: usize = 16;
57/// Bytes of an HMAC-SHA256 tag; the fingerprint carries all of them.
58const TAG_LEN: usize = 32;
59
60/// `serialize_with` of the generated ingress value: it is never serialized.
61pub fn refuse_to_serialize<S: Serializer>(
62    _value: &[u8],
63    _serializer: S,
64) -> Result<S::Ok, S::Error> {
65    Err(S::Error::custom(
66        "a protected health check header value is never serialized",
67    ))
68}
69
70/// `deserialize_with` of the generated ingress value: it is never read back
71/// from a serialized form.
72pub fn refuse_to_deserialize<'de, D: Deserializer<'de>>(
73    _deserializer: D,
74) -> Result<Vec<u8>, D::Error> {
75    Err(D::Error::custom(
76        "a protected health check header value is never deserialized",
77    ))
78}
79
80/// Protected ingress/IPC carrier of one health-check header value: the clear
81/// bytes travel in `SetProtectedHealthCheck` from the operator to the main
82/// process and are moved out into a [`SecretValue`] at once. It is the
83/// `HealthCheckHeaderValue` message of `command.proto` (`name = 1`,
84/// `value = 2`, both required) with a hand-written [`Message`] impl, mapped to
85/// the generated path by `command/build.rs`.
86///
87/// prost's generated decode replaces a populated `bytes` field with `clear`,
88/// `reserve`, `put`, so a reallocation frees the old buffer before any `Drop`
89/// runs. This impl wipes the value before it is replaced, cleared, or kept
90/// after a failed decode, and reads the new value straight from the input
91/// buffer into the field, with no intermediate copy.
92#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize)]
93pub struct HealthCheckHeaderValue {
94    pub name: String,
95    /// Never serialized, never read back: serde refuses it in both directions
96    /// instead of rendering it, so it can never be saved or printed as JSON.
97    #[serde(
98        serialize_with = "refuse_to_serialize",
99        deserialize_with = "refuse_to_deserialize"
100    )]
101    pub value: Vec<u8>,
102}
103
104impl HealthCheckHeaderValue {
105    /// Zero the whole allocation behind the value, spare capacity included,
106    /// and empty it.
107    pub(crate) fn wipe(&mut self) {
108        self.value.zeroize();
109    }
110
111    /// Replace the value with the length-delimited bytes at `buf`. The old
112    /// value is wiped before the buffer can grow, so a reallocation frees only
113    /// zeroes, and nothing is accepted before the wire type and the length
114    /// are known to be valid.
115    fn merge_value(&mut self, wire_type: WireType, buf: &mut impl Buf) -> Result<(), DecodeError> {
116        check_wire_type(WireType::LengthDelimited, wire_type)?;
117        let len = decode_varint(buf)?;
118        if len > buf.remaining() as u64 {
119            return Err(underflow(len));
120        }
121        let len = len as usize;
122        self.wipe();
123        self.value.reserve(len);
124        self.value.resize(len, 0);
125        buf.copy_to_slice(&mut self.value);
126        Ok(())
127    }
128
129    /// The provided `Message::merge` loop: every field until `buf` is spent.
130    fn merge_fields(&mut self, buf: &mut impl Buf) -> Result<(), DecodeError> {
131        let ctx = DecodeContext::default();
132        while buf.has_remaining() {
133            let (tag, wire_type) = decode_key(buf)?;
134            self.merge_field(tag, wire_type, buf, ctx.clone())?;
135        }
136        Ok(())
137    }
138
139    /// Wipe the value when `result` is an error: a refused fragment leaves no
140    /// accepted secret behind in the carrier.
141    fn wipe_on_error<T>(&mut self, result: Result<T, DecodeError>) -> Result<T, DecodeError> {
142        if result.is_err() {
143            self.wipe();
144        }
145        result
146    }
147}
148
149/// The error prost raises when a length delimiter exceeds the input, built by
150/// prost itself (its error kinds are private) from a value-less buffer that
151/// declares `len` bytes. Nothing is copied or accepted on the way.
152fn underflow(len: u64) -> DecodeError {
153    let mut declared = Vec::new();
154    encode_varint(len, &mut declared);
155    match bytes::merge(
156        WireType::LengthDelimited,
157        &mut Vec::<u8>::new(),
158        &mut declared.as_slice(),
159        DecodeContext::default(),
160    ) {
161        Err(error) => error,
162        Ok(()) => unreachable!("a length delimiter without its payload always underflows"),
163    }
164}
165
166impl Message for HealthCheckHeaderValue {
167    fn encode_raw(&self, buf: &mut impl BufMut) {
168        string::encode(1, &self.name, buf);
169        bytes::encode(2, &self.value, buf);
170    }
171
172    fn merge_field(
173        &mut self,
174        tag: u32,
175        wire_type: WireType,
176        buf: &mut impl Buf,
177        ctx: DecodeContext,
178    ) -> Result<(), DecodeError> {
179        const STRUCT_NAME: &str = "HealthCheckHeaderValue";
180        let result = match tag {
181            1 => string::merge(wire_type, &mut self.name, buf, ctx).map_err(|mut error| {
182                error.push(STRUCT_NAME, "name");
183                error
184            }),
185            2 => self.merge_value(wire_type, buf).map_err(|mut error| {
186                error.push(STRUCT_NAME, "value");
187                error
188            }),
189            _ => skip_field(wire_type, tag, buf, ctx),
190        };
191        self.wipe_on_error(result)
192    }
193
194    fn encoded_len(&self) -> usize {
195        string::encoded_len(1, &self.name) + bytes::encoded_len(2, &self.value)
196    }
197
198    fn clear(&mut self) {
199        self.name.clear();
200        self.wipe();
201    }
202
203    // The provided `merge` and `merge_length_delimited` can fail between two
204    // fields (a bad key, a delimiter overrun), outside `merge_field`: these
205    // overrides wipe on those paths too.
206    fn merge(&mut self, mut buf: impl Buf) -> Result<(), DecodeError> {
207        let result = self.merge_fields(&mut buf);
208        self.wipe_on_error(result)
209    }
210
211    fn merge_length_delimited(&mut self, mut buf: impl Buf) -> Result<(), DecodeError> {
212        let result = message::merge(
213            WireType::LengthDelimited,
214            self,
215            &mut buf,
216            DecodeContext::default(),
217        );
218        self.wipe_on_error(result)
219    }
220}
221
222/// A carrier owns clear bytes: whichever way it ends, by a refused order, a
223/// finished encode or a plain drop, its buffer is zeroed first. `Clone` is
224/// derived, and every clone is a carrier with this same `Drop`.
225impl Drop for HealthCheckHeaderValue {
226    fn drop(&mut self) {
227        self.wipe();
228    }
229}
230
231/// Why an ingress value was refused. Static on purpose: neither the value nor
232/// anything derived from it, its length included, appears in a diagnostic.
233#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
234pub enum SecretValueError {
235    #[error("health check header value must be valid UTF-8")]
236    NotUtf8,
237    /// A per-value rule of `crate::config::check_health_check_header_value`.
238    #[error("{0}")]
239    Invalid(&'static str),
240}
241
242/// A health-check header value held in memory only.
243///
244/// There is no `Clone`, `Serialize`, `Deserialize` or `Display`, and `Debug`
245/// prints a placeholder. The single way to read the bytes is
246/// [`SecretValue::expose_secret_bytes`]; call it only for the fingerprint HMAC
247/// and for the controlled in-memory IPC and wire builders.
248pub struct SecretValue(Zeroizing<Vec<u8>>);
249
250impl SecretValue {
251    /// Move the value out of a generated ingress message.
252    ///
253    /// The allocation moves with [`std::mem::take`], so no copy is made and
254    /// the generated field is left empty whether or not the value is accepted.
255    /// The value must be UTF-8 and meet the per-value rules every health-check
256    /// header meets (no forbidden control byte, no leading or trailing SP or
257    /// HTAB); the name is the caller's to validate.
258    pub fn take_from_ingress(
259        message: &mut HealthCheckHeaderValue,
260    ) -> Result<Self, SecretValueError> {
261        // Owned by a zeroizing wrapper before any check, so a refused value is
262        // wiped when the error path drops it.
263        let bytes = Zeroizing::new(std::mem::take(&mut message.value));
264        let text = std::str::from_utf8(&bytes).map_err(|_| SecretValueError::NotUtf8)?;
265        check_health_check_header_value(text).map_err(SecretValueError::Invalid)?;
266        Ok(Self(bytes))
267    }
268
269    /// The clear bytes. The name is the warning: the only legitimate callers
270    /// are the fingerprint HMAC and the controlled in-memory IPC and wire
271    /// builders that carry the value onward.
272    pub fn expose_secret_bytes(&self) -> &[u8] {
273        &self.0
274    }
275}
276
277impl fmt::Debug for SecretValue {
278    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
279        f.debug_struct("SecretValue")
280            .field("value", &format_args!("<protected>"))
281            .finish()
282    }
283}
284
285/// Why a header value could not be read from its source. Static: neither the
286/// bytes read nor any length of them appears in a diagnostic.
287#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
288pub enum HeaderValueReadError {
289    #[error("could not read the health check header value: {0}")]
290    Io(ErrorKind),
291    #[error("health check header value is longer than the allowed maximum")]
292    TooLong,
293}
294
295/// Read one health-check header value from `source` (a file or stdin) into a
296/// protected ingress carrier.
297///
298/// At most `max_len + 3` bytes are read: the value, a possible terminal CRLF
299/// and one byte that proves an overflow. The bytes are kept exactly, except one
300/// terminal LF and the CR right before it, which a text editor or `echo` adds.
301/// Nothing else is trimmed, and an empty value is a value. The value is
302/// validated later, where the carrier is turned into a [`SecretValue`].
303///
304/// The read buffer is allocated once at its final size, so no earlier,
305/// unwiped copy is left behind by a reallocation, and it is zeroed on every
306/// return. Neither the bytes nor any length of them appears in an error.
307pub fn read_header_value(
308    name: String,
309    mut source: impl Read,
310    max_len: usize,
311) -> Result<HealthCheckHeaderValue, HeaderValueReadError> {
312    let capacity = max_len
313        .checked_add(3)
314        .ok_or(HeaderValueReadError::TooLong)?;
315    let mut buffer = Zeroizing::new(vec![0u8; capacity]);
316    let mut filled = 0;
317    while filled < capacity {
318        match source.read(&mut buffer[filled..]) {
319            Ok(0) => break,
320            Ok(read) => filled += read,
321            Err(error) if error.kind() == ErrorKind::Interrupted => {}
322            Err(error) => return Err(HeaderValueReadError::Io(error.kind())),
323        }
324    }
325    let mut end = filled;
326    if end > 0 && buffer[end - 1] == b'\n' {
327        end -= 1;
328        if end > 0 && buffer[end - 1] == b'\r' {
329            end -= 1;
330        }
331    }
332    if end > max_len {
333        return Err(HeaderValueReadError::TooLong);
334    }
335    Ok(HealthCheckHeaderValue {
336        name,
337        value: buffer[..end].to_vec(),
338    })
339}
340
341/// The HMAC-SHA256 of the concatenation of `parts`, as a still-open MAC.
342fn hmac_sha256_mac(key: &[u8], parts: &[&[u8]]) -> HmacSha256 {
343    // HMAC takes a key of any length: this cannot fail.
344    let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts a key of any length");
345    for part in parts {
346        mac.update(part);
347    }
348    mac
349}
350
351/// The process-local secret that keys [`HeaderFingerprint`]s, with the public
352/// epoch that names it.
353///
354/// The key is random, lives in memory only and is zeroed on drop. There is no
355/// `Clone`, no serde and `Debug` prints the epoch only. The epoch is not a
356/// proof of anything: two keys may share one, and a fingerprint only verifies
357/// when the key itself reproduces its tag.
358pub struct FingerprintKey {
359    key: Zeroizing<[u8; KEY_LEN]>,
360    epoch: [u8; EPOCH_LEN],
361}
362
363impl FingerprintKey {
364    /// A fresh key and an independently drawn epoch from the OS generator.
365    pub fn generate() -> Result<Self, getrandom::Error> {
366        let mut key = Zeroizing::new([0u8; KEY_LEN]);
367        getrandom::fill(key.as_mut_slice())?;
368        let mut epoch = [0u8; EPOCH_LEN];
369        getrandom::fill(&mut epoch)?;
370        Ok(Self { key, epoch })
371    }
372
373    /// Rebuild a key that has to survive a hot upgrade, or a known test key.
374    pub fn from_parts(key: Zeroizing<[u8; KEY_LEN]>, epoch: [u8; EPOCH_LEN]) -> Self {
375        Self { key, epoch }
376    }
377
378    /// The MAC of one header under this key. The input is unambiguous: every
379    /// variable-length field carries its 8-byte big-endian length, and the
380    /// algorithm, version and epoch are part of it.
381    fn mac(&self, name: &str, value: &SecretValue) -> HmacSha256 {
382        let domain = format!("{FINGERPRINT_ALGORITHM}:{FINGERPRINT_VERSION}");
383        let name = name.to_ascii_lowercase();
384        let value = value.expose_secret_bytes();
385        let domain_len = (domain.len() as u64).to_be_bytes();
386        let name_len = (name.len() as u64).to_be_bytes();
387        let value_len = (value.len() as u64).to_be_bytes();
388        hmac_sha256_mac(
389            self.key.as_slice(),
390            &[
391                &domain_len,
392                domain.as_bytes(),
393                &self.epoch,
394                &name_len,
395                name.as_bytes(),
396                &value_len,
397                value,
398            ],
399        )
400    }
401
402    /// The fingerprint of `value` as the header `name` (case-insensitive).
403    pub fn fingerprint(&self, name: &str, value: &SecretValue) -> HeaderFingerprint {
404        HeaderFingerprint {
405            epoch: self.epoch,
406            tag: self.mac(name, value).finalize().into_bytes().into(),
407        }
408    }
409
410    /// Whether `fingerprint` was issued by this very key for this header and
411    /// value. A matching epoch is necessary, never sufficient: the full tag is
412    /// checked in constant time.
413    pub fn verify(&self, name: &str, value: &SecretValue, fingerprint: &HeaderFingerprint) -> bool {
414        fingerprint.epoch == self.epoch
415            && self.mac(name, value).verify_slice(&fingerprint.tag).is_ok()
416    }
417}
418
419impl fmt::Debug for FingerprintKey {
420    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
421        f.debug_struct("FingerprintKey")
422            .field("epoch", &hex::encode(self.epoch))
423            .field("key", &format_args!("<protected>"))
424            .finish()
425    }
426}
427
428/// Why a fingerprint string was refused. Static: it never echoes the input.
429#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
430pub enum FingerprintParseError {
431    #[error("fingerprint must look like 'algorithm:version:epoch:tag'")]
432    Malformed,
433    #[error("unknown fingerprint algorithm")]
434    Algorithm,
435    #[error("unsupported fingerprint version")]
436    Version,
437    #[error("fingerprint epoch must be 32 lowercase hex digits")]
438    Epoch,
439    #[error("fingerprint tag must be 64 lowercase hex digits")]
440    Tag,
441}
442
443/// A keyed fingerprint of a header value:
444/// `sozu-hc-hmac-sha256:v1:<epoch-32-hex>:<full-tag-64-hex>`. The tag is the
445/// whole 256-bit HMAC, never truncated. The same name and value under one
446/// epoch give the same fingerprint, so equality is visible; without the key it
447/// reveals no plaintext.
448#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
449pub struct HeaderFingerprint {
450    epoch: [u8; EPOCH_LEN],
451    tag: [u8; TAG_LEN],
452}
453
454impl fmt::Display for HeaderFingerprint {
455    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
456        write!(
457            f,
458            "{FINGERPRINT_ALGORITHM}:{FINGERPRINT_VERSION}:{}:{}",
459            hex::encode(self.epoch),
460            hex::encode(self.tag)
461        )
462    }
463}
464
465/// Exactly `N` bytes from exactly `2 * N` lowercase hex digits.
466fn parse_lowercase_hex<const N: usize>(text: &str) -> Option<[u8; N]> {
467    if text.len() != 2 * N || !text.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) {
468        return None;
469    }
470    let mut out = [0u8; N];
471    hex::decode_to_slice(text, &mut out).ok()?;
472    Some(out)
473}
474
475impl FromStr for HeaderFingerprint {
476    type Err = FingerprintParseError;
477
478    fn from_str(text: &str) -> Result<Self, Self::Err> {
479        let mut parts = text.split(':');
480        let (Some(algorithm), Some(version), Some(epoch), Some(tag), None) = (
481            parts.next(),
482            parts.next(),
483            parts.next(),
484            parts.next(),
485            parts.next(),
486        ) else {
487            return Err(FingerprintParseError::Malformed);
488        };
489        if algorithm != FINGERPRINT_ALGORITHM {
490            return Err(FingerprintParseError::Algorithm);
491        }
492        if version != FINGERPRINT_VERSION {
493            return Err(FingerprintParseError::Version);
494        }
495        Ok(Self {
496            epoch: parse_lowercase_hex(epoch).ok_or(FingerprintParseError::Epoch)?,
497            tag: parse_lowercase_hex(tag).ok_or(FingerprintParseError::Tag)?,
498        })
499    }
500}
501
502/// Longest header name [`ProtectedHeader`]'s `Debug` prints.
503const DEBUG_NAME_LEN: usize = 64;
504
505/// One protected header: its public name, its public keyed fingerprint and a
506/// shared handle on the value. Cloning clones the `Arc`, never the bytes.
507#[derive(Clone)]
508pub struct ProtectedHeader {
509    name: String,
510    fingerprint: HeaderFingerprint,
511    value: Arc<SecretValue>,
512}
513
514impl ProtectedHeader {
515    /// Fingerprint `value` under `key` as header `name`, which the caller has
516    /// already validated.
517    pub fn new(key: &FingerprintKey, name: String, value: SecretValue) -> Self {
518        let fingerprint = key.fingerprint(&name, &value);
519        Self {
520            name,
521            fingerprint,
522            value: Arc::new(value),
523        }
524    }
525
526    /// A header whose fingerprint another process issued and this one trusts:
527    /// the worker side of an order the main process already fingerprinted.
528    pub fn from_trusted(name: String, fingerprint: HeaderFingerprint, value: SecretValue) -> Self {
529        Self {
530            name,
531            fingerprint,
532            value: Arc::new(value),
533        }
534    }
535
536    pub fn name(&self) -> &str {
537        &self.name
538    }
539
540    pub fn fingerprint(&self) -> &HeaderFingerprint {
541        &self.fingerprint
542    }
543
544    pub fn value(&self) -> &Arc<SecretValue> {
545        &self.value
546    }
547}
548
549impl fmt::Debug for ProtectedHeader {
550    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
551        let name: String = self.name.chars().take(DEBUG_NAME_LEN).collect();
552        f.debug_struct("ProtectedHeader")
553            .field("name_truncated", &(name.len() < self.name.len()))
554            .field("name", &name)
555            .field("fingerprint", &self.fingerprint.to_string())
556            .field("value", &format_args!("<protected>"))
557            .finish()
558    }
559}
560
561/// An immutable, ordered set of [`ProtectedHeader`]s behind one `Arc`. Cloning
562/// clones that handle; the headers and their values are shared.
563#[derive(Clone)]
564pub struct ProtectedHeaderSet(Arc<[ProtectedHeader]>);
565
566impl ProtectedHeaderSet {
567    /// The headers, in the order given.
568    pub fn new(headers: Vec<ProtectedHeader>) -> Self {
569        Self(Arc::from(headers))
570    }
571
572    pub fn headers(&self) -> &[ProtectedHeader] {
573        &self.0
574    }
575
576    pub fn len(&self) -> usize {
577        self.0.len()
578    }
579
580    pub fn is_empty(&self) -> bool {
581        self.0.is_empty()
582    }
583
584    /// Whether this set holds the value behind every header of `metadata`, in
585    /// order: same count, same names, same fingerprints. A configuration that
586    /// names headers is only as good as the set that satisfies it.
587    pub fn satisfies(&self, metadata: &[HealthCheckHeader]) -> bool {
588        self.0.len() == metadata.len()
589            && self.0.iter().zip(metadata).all(|(held, meta)| {
590                held.name == meta.name && held.fingerprint.to_string() == meta.fingerprint
591            })
592    }
593
594    /// Whether both handles share one allocation.
595    pub fn ptr_eq(&self, other: &Self) -> bool {
596        Arc::ptr_eq(&self.0, &other.0)
597    }
598}
599
600impl fmt::Debug for ProtectedHeaderSet {
601    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
602        f.debug_struct("ProtectedHeaderSet")
603            .field("headers", &self.0.len())
604            .finish()
605    }
606}
607
608/// Why a protected health-check order was refused. Static on purpose: neither
609/// a value, nor any length of one, appears in a diagnostic.
610#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
611pub enum ProtectedHealthCheckError {
612    /// A rule of `validate_health_check_config` or
613    /// `validate_health_check_header_values`.
614    #[error("{0}")]
615    Invalid(&'static str),
616    #[error(transparent)]
617    Value(#[from] SecretValueError),
618    #[error("health check header fingerprint does not match its value")]
619    FingerprintMismatch,
620    #[error("health check header fingerprint is missing")]
621    FingerprintMissing,
622}
623
624/// A validated protected health-check order: the cluster, its configuration
625/// with a trusted fingerprint on every header, and the held values.
626pub struct ProtectedHealthCheck {
627    cluster_id: String,
628    config: HealthCheckConfig,
629    headers: ProtectedHeaderSet,
630}
631
632impl ProtectedHealthCheck {
633    /// An operator order: validate every name and value, then fingerprint each
634    /// value under `key`. A fingerprint the client supplied is only accepted
635    /// when it is the one `key` computes. Nothing is built unless all of it is
636    /// valid; the carriers are emptied of the values that are accepted.
637    pub fn from_ingress(
638        key: &FingerprintKey,
639        order: &mut SetProtectedHealthCheck,
640    ) -> Result<Self, ProtectedHealthCheckError> {
641        let held = Self::held_values(order)?;
642        let mut config = order.config.to_owned();
643        let mut headers = Vec::with_capacity(held.len());
644        for (meta, (name, value)) in config.headers.iter_mut().zip(held) {
645            let header = ProtectedHeader::new(key, name, value);
646            if !meta.fingerprint.is_empty() {
647                let claimed: HeaderFingerprint = meta.fingerprint.parse().map_err(|_| {
648                    ProtectedHealthCheckError::Invalid(
649                        "health check header fingerprint is malformed",
650                    )
651                })?;
652                if &claimed != header.fingerprint() {
653                    return Err(ProtectedHealthCheckError::FingerprintMismatch);
654                }
655            }
656            meta.fingerprint = header.fingerprint().to_string();
657            headers.push(header);
658        }
659        Ok(Self {
660            cluster_id: order.cluster_id.to_owned(),
661            config,
662            headers: ProtectedHeaderSet::new(headers),
663        })
664    }
665
666    /// An order the main process built: every header already carries the
667    /// fingerprint it issued, which is trusted as it is.
668    pub fn from_trusted(
669        order: &mut SetProtectedHealthCheck,
670    ) -> Result<Self, ProtectedHealthCheckError> {
671        let held = Self::held_values(order)?;
672        let mut headers = Vec::with_capacity(held.len());
673        for (meta, (name, value)) in order.config.headers.iter().zip(held) {
674            if meta.fingerprint.is_empty() {
675                return Err(ProtectedHealthCheckError::FingerprintMissing);
676            }
677            let fingerprint = meta.fingerprint.parse().map_err(|_| {
678                ProtectedHealthCheckError::Invalid("health check header fingerprint is malformed")
679            })?;
680            headers.push(ProtectedHeader::from_trusted(name, fingerprint, value));
681        }
682        Ok(Self {
683            cluster_id: order.cluster_id.to_owned(),
684            config: order.config.to_owned(),
685            headers: ProtectedHeaderSet::new(headers),
686        })
687    }
688
689    /// Validate `order` as a whole, then move every value out of its carriers.
690    fn held_values(
691        order: &mut SetProtectedHealthCheck,
692    ) -> Result<Vec<(String, SecretValue)>, ProtectedHealthCheckError> {
693        validate_health_check_config(&order.config).map_err(ProtectedHealthCheckError::Invalid)?;
694        validate_health_check_header_values(&order.config, &order.values)
695            .map_err(ProtectedHealthCheckError::Invalid)?;
696        order
697            .values
698            .iter_mut()
699            .map(|carrier| {
700                let value = SecretValue::take_from_ingress(carrier)?;
701                Ok((carrier.name.to_owned(), value))
702            })
703            .collect()
704    }
705
706    /// The order of a cluster from its parts: the main process's own
707    /// configuration, every header of which carries the fingerprint it issued,
708    /// and the held values behind them. The set must satisfy the headers.
709    pub fn from_parts(
710        cluster_id: String,
711        config: HealthCheckConfig,
712        headers: ProtectedHeaderSet,
713    ) -> Result<Self, ProtectedHealthCheckError> {
714        if !headers.satisfies(&config.headers) {
715            return Err(ProtectedHealthCheckError::Invalid(
716                "health check header values do not match the header metadata",
717            ));
718        }
719        Ok(Self {
720            cluster_id,
721            config,
722            headers,
723        })
724    }
725
726    pub fn cluster_id(&self) -> &str {
727        &self.cluster_id
728    }
729
730    /// The configuration, headers carrying names and fingerprints only.
731    pub fn config(&self) -> &HealthCheckConfig {
732        &self.config
733    }
734
735    pub fn headers(&self) -> &ProtectedHeaderSet {
736        &self.headers
737    }
738
739    /// Whether the held values are the ones `metadata` names.
740    pub fn satisfies(&self, metadata: &[HealthCheckHeader]) -> bool {
741        self.headers.satisfies(metadata)
742    }
743
744    /// The order a worker receives: trusted fingerprints in place and one
745    /// freshly copied carrier per value. Each carrier wipes itself on drop.
746    pub fn to_order(&self) -> SetProtectedHealthCheck {
747        SetProtectedHealthCheck {
748            cluster_id: self.cluster_id.to_owned(),
749            config: self.config.to_owned(),
750            values: self
751                .headers
752                .headers()
753                .iter()
754                .map(|header| HealthCheckHeaderValue {
755                    name: header.name().to_owned(),
756                    value: header.value().expose_secret_bytes().to_vec(),
757                })
758                .collect(),
759        }
760    }
761
762    /// The same order as a request, for the caller to encode and drop.
763    pub fn to_request(&self) -> Request {
764        RequestType::SetProtectedHealthCheck(self.to_order()).into()
765    }
766}
767
768impl fmt::Debug for ProtectedHealthCheck {
769    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
770        f.debug_struct("ProtectedHealthCheck")
771            .field("cluster_id", &self.cluster_id)
772            .field("headers", &self.config.headers)
773            .finish()
774    }
775}
776
777/// Opaque identity of one proposal. It carries no header, so a task may
778/// persist it next to a cluster id; the set it names lives in memory only.
779#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
780pub struct ProtectedHeaderVersion(u64);
781
782impl ProtectedHeaderVersion {
783    /// The version as it travels in a protected health check order.
784    pub fn get(self) -> u64 {
785        self.0
786    }
787
788    /// The version a peer named in an order or an answer. It is only ever
789    /// compared with the versions the peer was given: it grants nothing.
790    pub fn from_wire(version: u64) -> Self {
791        Self(version)
792    }
793}
794
795/// The confirmed clusters a fresh worker is given and the version it stages
796/// them under: names and a number, never a value.
797#[derive(Debug, Clone, PartialEq, Eq)]
798pub struct ConfirmedBootstrap {
799    version: ProtectedHeaderVersion,
800    clusters: Vec<String>,
801}
802
803impl ConfirmedBootstrap {
804    pub fn version(&self) -> ProtectedHeaderVersion {
805        self.version
806    }
807
808    /// The clusters holding a confirmed value, in lexical order.
809    pub fn clusters(&self) -> &[String] {
810        &self.clusters
811    }
812}
813
814/// A full replacement of the protected headers of one cluster.
815pub enum ProtectedHeaderChange {
816    /// These headers, in this order.
817    Replace(ProtectedHeaderSet),
818    /// The cluster stays, with an empty header set.
819    Clear,
820    /// The cluster is absent.
821    Remove,
822}
823
824impl ProtectedHeaderChange {
825    /// `None` is an absent cluster, an empty set a cleared one.
826    fn into_state(self) -> Option<ProtectedHeaderSet> {
827        match self {
828            Self::Replace(set) => Some(set),
829            Self::Clear => Some(ProtectedHeaderSet::new(Vec::new())),
830            Self::Remove => None,
831        }
832    }
833}
834
835/// Every [`ProtectedHeaderVersion`] has been issued.
836#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
837#[error("protected health check header versions are exhausted")]
838pub struct VersionExhausted;
839
840/// What a resolution did.
841#[derive(Debug, Clone, Copy, PartialEq, Eq)]
842pub enum Resolution {
843    Applied,
844    /// The version is unknown, already resolved or superseded: nothing changed.
845    Ignored,
846}
847
848/// An unresolved proposal; `state` is `None` for a removal.
849struct PendingVersion {
850    version: ProtectedHeaderVersion,
851    state: Option<ProtectedHeaderSet>,
852}
853
854#[derive(Default)]
855struct ClusterJournal {
856    /// What every targeted worker acknowledged; `None` is an absent cluster.
857    confirmed: Option<ProtectedHeaderSet>,
858    /// Unresolved proposals, oldest first. The last one is optimistically
859    /// effective.
860    pending: Vec<PendingVersion>,
861}
862
863/// The in-memory protected headers of every cluster, and the unresolved
864/// proposals that may still replace them.
865///
866/// A proposal is effective at once. A rejection drops it, so the newest
867/// remaining proposal, else the confirmed state, is effective: an ancestor's
868/// rejection leaves a live descendant alone. An acceptance confirms the
869/// proposal and supersedes every older one, which can no longer change the
870/// outcome. Resolved proposals are forgotten; versions are never reused, so a
871/// late or repeated resolution finds nothing and changes nothing.
872///
873/// There is no serde, no `Clone` and `Debug` prints counts only: the journal
874/// never reaches a state file, a task snapshot or a log.
875#[derive(Default)]
876pub struct ProtectedHeaderJournal {
877    last_issued: u64,
878    clusters: HashMap<String, ClusterJournal>,
879}
880
881impl ProtectedHeaderJournal {
882    /// Start from a journal that already issued `last_issued` versions.
883    #[cfg(test)]
884    fn with_last_issued(last_issued: u64) -> Self {
885        Self {
886            last_issued,
887            clusters: HashMap::new(),
888        }
889    }
890
891    /// Record `change` as the effective state of `cluster` until it is
892    /// resolved. Nothing is touched when the versions are exhausted.
893    pub fn propose(
894        &mut self,
895        cluster: &str,
896        change: ProtectedHeaderChange,
897    ) -> Result<ProtectedHeaderVersion, VersionExhausted> {
898        let next = self.last_issued.checked_add(1).ok_or(VersionExhausted)?;
899        let version = ProtectedHeaderVersion(next);
900        self.last_issued = next;
901        self.clusters
902            .entry(cluster.to_owned())
903            .or_default()
904            .pending
905            .push(PendingVersion {
906                version,
907                state: change.into_state(),
908            });
909        Ok(version)
910    }
911
912    /// Whether the journal retains a non-empty header set anywhere: confirmed,
913    /// or in any unresolved proposal, for any cluster. A later `Clear` or
914    /// `Remove` hides the headers beneath it without dropping them, so this
915    /// looks at every retained state and not only [`Self::effective`].
916    pub fn holds_values(&self) -> bool {
917        self.clusters.values().any(|journal| {
918            journal
919                .confirmed
920                .iter()
921                .chain(journal.pending.iter().filter_map(|p| p.state.as_ref()))
922                .any(|set| !set.is_empty())
923        })
924    }
925
926    /// What a fresh worker must be given to hold the confirmed values: the
927    /// clusters whose confirmed set is not empty, in lexical order, and the
928    /// version they are all staged under. `None` when no value is held.
929    ///
930    /// Read-only: it proposes, accepts and rejects nothing, and returns no
931    /// byte of a value. A fresh worker store is empty, so the last version the
932    /// journal issued is acceptable for every cluster: it is at least `1` as
933    /// soon as a value was ever held, and no cluster was confirmed under a
934    /// newer one. Refused while any proposal is unresolved, and when a
935    /// confirmed set is not the one `metadata` names.
936    pub fn confirmed_bootstrap<'m>(
937        &self,
938        metadata: impl Fn(&str) -> Option<&'m [HealthCheckHeader]>,
939    ) -> Result<Option<ConfirmedBootstrap>, TransferError> {
940        if self
941            .clusters
942            .values()
943            .any(|cluster| !cluster.pending.is_empty())
944        {
945            return Err(TransferError::Pending);
946        }
947        let mut clusters = Vec::new();
948        for (id, cluster) in &self.clusters {
949            let Some(set) = &cluster.confirmed else {
950                continue;
951            };
952            verify_set(set, metadata(id))?;
953            if !set.is_empty() {
954                clusters.push(id.to_owned());
955            }
956        }
957        if clusters.is_empty() {
958            return Ok(None);
959        }
960        if self.last_issued == 0 {
961            return Err(TransferError::Malformed);
962        }
963        clusters.sort();
964        Ok(Some(ConfirmedBootstrap {
965            version: ProtectedHeaderVersion(self.last_issued),
966            clusters,
967        }))
968    }
969
970    /// The effective protected headers of `cluster`: `None` when it is
971    /// absent, an empty set when it was cleared.
972    pub fn effective(&self, cluster: &str) -> Option<&ProtectedHeaderSet> {
973        let journal = self.clusters.get(cluster)?;
974        journal
975            .pending
976            .last()
977            .map_or(journal.confirmed.as_ref(), |pending| pending.state.as_ref())
978    }
979
980    /// The state `version` proposed for `cluster` while it is unresolved:
981    /// `None` when the version is unknown or resolved, `Some(None)` for a
982    /// removal.
983    pub fn proposed(
984        &self,
985        cluster: &str,
986        version: ProtectedHeaderVersion,
987    ) -> Option<Option<&ProtectedHeaderSet>> {
988        self.clusters
989            .get(cluster)?
990            .pending
991            .iter()
992            .find(|pending| pending.version == version)
993            .map(|pending| pending.state.as_ref())
994    }
995
996    /// Every targeted worker acknowledged `version`. It becomes the confirmed
997    /// state and supersedes the proposals older than it.
998    pub fn accept(&mut self, cluster: &str, version: ProtectedHeaderVersion) -> Resolution {
999        let Some(journal) = self.clusters.get_mut(cluster) else {
1000            return Resolution::Ignored;
1001        };
1002        let Some(index) = journal.pending.iter().position(|p| p.version == version) else {
1003            return Resolution::Ignored;
1004        };
1005        if let Some(accepted) = journal.pending.drain(..=index).next_back() {
1006            journal.confirmed = accepted.state;
1007        }
1008        self.forget_if_absent(cluster);
1009        Resolution::Applied
1010    }
1011
1012    /// `version` failed: drop it and fall back to what is left.
1013    pub fn reject(&mut self, cluster: &str, version: ProtectedHeaderVersion) -> Resolution {
1014        let Some(journal) = self.clusters.get_mut(cluster) else {
1015            return Resolution::Ignored;
1016        };
1017        let Some(index) = journal.pending.iter().position(|p| p.version == version) else {
1018            return Resolution::Ignored;
1019        };
1020        journal.pending.remove(index);
1021        self.forget_if_absent(cluster);
1022        Resolution::Applied
1023    }
1024
1025    /// An absent cluster with nothing pending has no entry: a removal leaves
1026    /// no tombstone, since no resolved version is ever matched again.
1027    fn forget_if_absent(&mut self, cluster: &str) {
1028        if self
1029            .clusters
1030            .get(cluster)
1031            .is_some_and(|j| j.confirmed.is_none() && j.pending.is_empty())
1032        {
1033            self.clusters.remove(cluster);
1034        }
1035    }
1036}
1037
1038impl fmt::Debug for ProtectedHeaderJournal {
1039    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1040        f.debug_struct("ProtectedHeaderJournal")
1041            .field("last_issued", &self.last_issued)
1042            .field("clusters", &self.clusters.len())
1043            .field(
1044                "pending",
1045                &self
1046                    .clusters
1047                    .values()
1048                    .map(|j| j.pending.len())
1049                    .sum::<usize>(),
1050            )
1051            .finish()
1052    }
1053}
1054
1055/// Largest encoded [`ProtectedTransfer`] a main upgrade may carry.
1056pub const PROTECTED_TRANSFER_MAX_BYTES: usize = 64 << 20;
1057/// Layout version of an encoded [`ProtectedTransfer`].
1058const TRANSFER_LAYOUT_VERSION: u8 = 1;
1059/// Most clusters one transfer may name.
1060const TRANSFER_MAX_CLUSTERS: usize = 1 << 16;
1061/// Longest cluster id or header name one transfer may carry.
1062const TRANSFER_MAX_TEXT: usize = 1 << 16;
1063
1064/// Why the confirmed protected state could not be handed to a new main
1065/// process, or an encoded hand-over was refused. Static on purpose: neither a
1066/// value, nor the key, nor any length of one appears in a diagnostic.
1067#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
1068pub enum TransferError {
1069    #[error("an unresolved protected header version cannot be handed over")]
1070    Pending,
1071    #[error("held protected header values have no fingerprint key")]
1072    KeyMissing,
1073    #[error("held protected header state does not match the configuration metadata")]
1074    Mismatch,
1075    #[error("protected header transfer exceeds its maximum size")]
1076    TooLarge,
1077    #[error("protected header transfer is truncated")]
1078    Truncated,
1079    #[error("protected header transfer has trailing bytes")]
1080    Trailing,
1081    #[error("protected header transfer is malformed")]
1082    Malformed,
1083    #[error("protected header transfer buffer is too small")]
1084    BufferTooSmall,
1085}
1086
1087/// The fingerprint key of a [`ProtectedTransfer`], in zeroing storage.
1088struct TransferKey {
1089    key: Zeroizing<[u8; KEY_LEN]>,
1090    epoch: [u8; EPOCH_LEN],
1091}
1092
1093struct TransferHeader {
1094    name: String,
1095    value: Arc<SecretValue>,
1096}
1097
1098/// The confirmed protected state of a main process, as a new main process
1099/// takes it over: the optional fingerprint key, the number of versions issued,
1100/// and the confirmed header set of every cluster, clusters sorted by id and
1101/// headers in their configured order. Pending proposals never travel.
1102///
1103/// There is no serde, no `Clone`, and `Debug` prints counts only. The values
1104/// stay behind their `Arc` handles until [`Self::encode_into`] writes them
1105/// into a caller-owned zeroing buffer; the key is copied into zeroing
1106/// storage. The one reader is [`Self::restore`], which accepts nothing the
1107/// authoritative configuration metadata does not name.
1108pub struct ProtectedTransfer {
1109    key: Option<TransferKey>,
1110    last_issued: u64,
1111    clusters: Vec<(String, Vec<TransferHeader>)>,
1112}
1113
1114/// Whether `set` is the confirmed state the configuration metadata of its
1115/// cluster names: an empty set needs no header named, a held one needs every
1116/// name and full fingerprint, in order.
1117fn verify_set(
1118    set: &ProtectedHeaderSet,
1119    metadata: Option<&[HealthCheckHeader]>,
1120) -> Result<(), TransferError> {
1121    let matches = if set.is_empty() {
1122        metadata.is_none_or(<[HealthCheckHeader]>::is_empty)
1123    } else {
1124        metadata.is_some_and(|metadata| set.satisfies(metadata))
1125    };
1126    if matches {
1127        Ok(())
1128    } else {
1129        Err(TransferError::Mismatch)
1130    }
1131}
1132
1133impl ProtectedTransfer {
1134    /// The confirmed state of `journal` and `key`, or why it cannot be handed
1135    /// over exactly: an unresolved proposal, a held value without a key, or a
1136    /// set the configuration metadata of its cluster does not name.
1137    pub fn export<'m>(
1138        journal: &ProtectedHeaderJournal,
1139        key: Option<&FingerprintKey>,
1140        metadata: impl Fn(&str) -> Option<&'m [HealthCheckHeader]>,
1141    ) -> Result<Self, TransferError> {
1142        if journal
1143            .clusters
1144            .values()
1145            .any(|cluster| !cluster.pending.is_empty())
1146        {
1147            return Err(TransferError::Pending);
1148        }
1149        let mut clusters = Vec::with_capacity(journal.clusters.len());
1150        for (id, cluster) in &journal.clusters {
1151            let Some(set) = &cluster.confirmed else {
1152                continue;
1153            };
1154            verify_set(set, metadata(id))?;
1155            if !set.is_empty() {
1156                let key = key.ok_or(TransferError::KeyMissing)?;
1157                if !set
1158                    .headers()
1159                    .iter()
1160                    .all(|header| key.verify(&header.name, &header.value, &header.fingerprint))
1161                {
1162                    return Err(TransferError::Mismatch);
1163                }
1164            }
1165            clusters.push((
1166                id.to_owned(),
1167                set.headers()
1168                    .iter()
1169                    .map(|header| TransferHeader {
1170                        name: header.name.to_owned(),
1171                        value: Arc::clone(&header.value),
1172                    })
1173                    .collect(),
1174            ));
1175        }
1176        clusters.sort_by(|left, right| left.0.cmp(&right.0));
1177        let key = key.map(|key| {
1178            let mut copy = Zeroizing::new([0u8; KEY_LEN]);
1179            copy.copy_from_slice(key.key.as_slice());
1180            TransferKey {
1181                key: copy,
1182                epoch: key.epoch,
1183            }
1184        });
1185        Ok(Self {
1186            key,
1187            last_issued: journal.last_issued,
1188            clusters,
1189        })
1190    }
1191
1192    /// Rebuild a fresh key and journal from the transfer, and keep a cluster's
1193    /// set only when `metadata`, the authoritative configuration, names it
1194    /// exactly: same names and full HMAC fingerprints under the transferred
1195    /// key, in the same order. An epoch alone is never acceptance.
1196    pub fn restore<'m>(
1197        self,
1198        metadata: impl Fn(&str) -> Option<&'m [HealthCheckHeader]>,
1199    ) -> Result<(Option<FingerprintKey>, ProtectedHeaderJournal), TransferError> {
1200        let Self {
1201            key,
1202            last_issued,
1203            clusters,
1204        } = self;
1205        let key = key.map(|transfer| FingerprintKey::from_parts(transfer.key, transfer.epoch));
1206        let mut restored = HashMap::with_capacity(clusters.len());
1207        for (id, headers) in clusters {
1208            let mut protected = Vec::with_capacity(headers.len());
1209            for header in headers {
1210                let key = key.as_ref().ok_or(TransferError::KeyMissing)?;
1211                protected.push(ProtectedHeader {
1212                    fingerprint: key.fingerprint(&header.name, &header.value),
1213                    name: header.name,
1214                    value: header.value,
1215                });
1216            }
1217            let set = ProtectedHeaderSet::new(protected);
1218            verify_set(&set, metadata(&id))?;
1219            restored.insert(
1220                id,
1221                ClusterJournal {
1222                    confirmed: Some(set),
1223                    pending: Vec::new(),
1224                },
1225            );
1226        }
1227        Ok((
1228            key,
1229            ProtectedHeaderJournal {
1230                last_issued,
1231                clusters: restored,
1232            },
1233        ))
1234    }
1235
1236    /// The exact number of bytes [`Self::encode_into`] appends.
1237    pub fn encoded_len(&self) -> Result<usize, TransferError> {
1238        // Version, key flag, `last_issued` and the cluster count.
1239        let mut total = 2 + 8 + 4;
1240        if self.key.is_some() {
1241            total += KEY_LEN + EPOCH_LEN;
1242        }
1243        for (id, headers) in &self.clusters {
1244            total = total
1245                .checked_add(4 + id.len() + 4)
1246                .ok_or(TransferError::TooLarge)?;
1247            for header in headers {
1248                total = total
1249                    .checked_add(4 + header.name.len() + 4)
1250                    .and_then(|total| total.checked_add(header.value.expose_secret_bytes().len()))
1251                    .ok_or(TransferError::TooLarge)?;
1252            }
1253        }
1254        if total > PROTECTED_TRANSFER_MAX_BYTES {
1255            return Err(TransferError::TooLarge);
1256        }
1257        Ok(total)
1258    }
1259
1260    /// Append the encoding to `out` without ever growing it: the spare
1261    /// capacity must hold [`Self::encoded_len`] bytes, so no unwiped block is
1262    /// left behind by a reallocation.
1263    pub fn encode_into(&self, out: &mut Zeroizing<Vec<u8>>) -> Result<(), TransferError> {
1264        let needed = self.encoded_len()?;
1265        if out.capacity() - out.len() < needed {
1266            return Err(TransferError::BufferTooSmall);
1267        }
1268        let start = out.len();
1269        let length = |len: usize| u32::try_from(len).map_err(|_| TransferError::TooLarge);
1270        out.push(TRANSFER_LAYOUT_VERSION);
1271        match &self.key {
1272            Some(transfer) => {
1273                out.push(1);
1274                out.extend_from_slice(transfer.key.as_slice());
1275                out.extend_from_slice(&transfer.epoch);
1276            }
1277            None => out.push(0),
1278        }
1279        out.extend_from_slice(&self.last_issued.to_be_bytes());
1280        out.extend_from_slice(&length(self.clusters.len())?.to_be_bytes());
1281        for (id, headers) in &self.clusters {
1282            out.extend_from_slice(&length(id.len())?.to_be_bytes());
1283            out.extend_from_slice(id.as_bytes());
1284            out.extend_from_slice(&length(headers.len())?.to_be_bytes());
1285            for header in headers {
1286                let value = header.value.expose_secret_bytes();
1287                out.extend_from_slice(&length(header.name.len())?.to_be_bytes());
1288                out.extend_from_slice(header.name.as_bytes());
1289                out.extend_from_slice(&length(value.len())?.to_be_bytes());
1290                out.extend_from_slice(value);
1291            }
1292        }
1293        debug_assert_eq!(out.len() - start, needed, "encoded_len is exact");
1294        Ok(())
1295    }
1296
1297    /// Read an encoding back. Every count and length is bounded before it is
1298    /// used, text must be UTF-8, values must pass the per-value header rules,
1299    /// clusters must be strictly increasing, and nothing may follow.
1300    pub fn decode(bytes: &[u8]) -> Result<Self, TransferError> {
1301        struct Reader<'a>(&'a [u8]);
1302        impl<'a> Reader<'a> {
1303            fn take(&mut self, len: usize) -> Result<&'a [u8], TransferError> {
1304                if self.0.len() < len {
1305                    return Err(TransferError::Truncated);
1306                }
1307                let (head, tail) = self.0.split_at(len);
1308                self.0 = tail;
1309                Ok(head)
1310            }
1311            fn count(&mut self, maximum: usize) -> Result<usize, TransferError> {
1312                let bytes: [u8; 4] = self
1313                    .take(4)?
1314                    .try_into()
1315                    .map_err(|_| TransferError::Malformed)?;
1316                let count = usize::try_from(u32::from_be_bytes(bytes))
1317                    .map_err(|_| TransferError::TooLarge)?;
1318                if count > maximum {
1319                    return Err(TransferError::TooLarge);
1320                }
1321                Ok(count)
1322            }
1323            fn text(&mut self) -> Result<String, TransferError> {
1324                let len = self.count(TRANSFER_MAX_TEXT)?;
1325                String::from_utf8(self.take(len)?.to_vec()).map_err(|_| TransferError::Malformed)
1326            }
1327        }
1328
1329        if bytes.len() > PROTECTED_TRANSFER_MAX_BYTES {
1330            return Err(TransferError::TooLarge);
1331        }
1332        let mut reader = Reader(bytes);
1333        if reader.take(1)? != [TRANSFER_LAYOUT_VERSION] {
1334            return Err(TransferError::Malformed);
1335        }
1336        let key = match reader.take(1)? {
1337            [0] => None,
1338            [1] => {
1339                let mut key = Zeroizing::new([0u8; KEY_LEN]);
1340                key.copy_from_slice(reader.take(KEY_LEN)?);
1341                let epoch = reader
1342                    .take(EPOCH_LEN)?
1343                    .try_into()
1344                    .map_err(|_| TransferError::Malformed)?;
1345                Some(TransferKey { key, epoch })
1346            }
1347            _ => return Err(TransferError::Malformed),
1348        };
1349        let last_issued = u64::from_be_bytes(
1350            reader
1351                .take(8)?
1352                .try_into()
1353                .map_err(|_| TransferError::Malformed)?,
1354        );
1355        let cluster_count = reader.count(TRANSFER_MAX_CLUSTERS)?;
1356        let mut clusters: Vec<(String, Vec<TransferHeader>)> = Vec::new();
1357        for _ in 0..cluster_count {
1358            let id = reader.text()?;
1359            if clusters.last().is_some_and(|(previous, _)| *previous >= id) {
1360                return Err(TransferError::Malformed);
1361            }
1362            let header_count = reader.count(crate::config::HEALTH_CHECK_MAX_HEADERS)?;
1363            let mut headers = Vec::new();
1364            for _ in 0..header_count {
1365                let name = reader.text()?;
1366                let len = reader.count(crate::config::HEALTH_CHECK_MAX_HEADER_BYTES)?;
1367                let mut carrier = HealthCheckHeaderValue {
1368                    name: name.to_owned(),
1369                    value: reader.take(len)?.to_vec(),
1370                };
1371                let value = SecretValue::take_from_ingress(&mut carrier)
1372                    .map_err(|_| TransferError::Malformed)?;
1373                headers.push(TransferHeader {
1374                    name,
1375                    value: Arc::new(value),
1376                });
1377            }
1378            clusters.push((id, headers));
1379        }
1380        if !reader.0.is_empty() {
1381            return Err(TransferError::Trailing);
1382        }
1383        Ok(Self {
1384            key,
1385            last_issued,
1386            clusters,
1387        })
1388    }
1389}
1390
1391impl fmt::Debug for ProtectedTransfer {
1392    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1393        f.debug_struct("ProtectedTransfer")
1394            .field("key", &self.key.is_some())
1395            .field("clusters", &self.clusters.len())
1396            .finish()
1397    }
1398}
1399
1400#[cfg(test)]
1401mod tests {
1402    use std::str::FromStr;
1403
1404    use zeroize::Zeroizing;
1405
1406    use super::*;
1407    use crate::proto::command::HealthCheckHeaderValue;
1408
1409    /// Test-only marker: it must never appear in a production rendering.
1410    const SENTINEL: &str = "SYNTHETIC-SLICE1-SECRET-4f1c";
1411
1412    /// The raw HMAC-SHA256 primitive, finalized.
1413    fn hmac_sha256(key: &[u8], parts: &[&[u8]]) -> [u8; TAG_LEN] {
1414        hmac_sha256_mac(key, parts).finalize().into_bytes().into()
1415    }
1416
1417    fn ingress(name: &str, value: &[u8]) -> HealthCheckHeaderValue {
1418        HealthCheckHeaderValue {
1419            name: name.to_owned(),
1420            value: value.to_vec(),
1421        }
1422    }
1423
1424    fn secret(value: &[u8]) -> SecretValue {
1425        SecretValue::take_from_ingress(&mut ingress("authorization", value)).unwrap()
1426    }
1427
1428    fn key(byte: u8, epoch_byte: u8) -> FingerprintKey {
1429        FingerprintKey::from_parts(Zeroizing::new([byte; 32]), [epoch_byte; 16])
1430    }
1431
1432    #[test]
1433    fn ingress_serialization_fails_and_never_reaches_the_writer() {
1434        let message = ingress("authorization", SENTINEL.as_bytes());
1435        assert!(serde_json::to_string(&message).is_err());
1436        let mut written = Vec::new();
1437        assert!(serde_json::to_writer(&mut written, &message).is_err());
1438        let written = String::from_utf8_lossy(&written);
1439        assert!(!written.contains(SENTINEL), "{written}");
1440        let bytes = serde_json::to_vec(&message);
1441        assert!(bytes.is_err());
1442    }
1443
1444    #[test]
1445    fn ingress_deserialization_fails_for_every_json_shape() {
1446        let as_array = format!(
1447            r#"{{"name":"authorization","value":{:?}}}"#,
1448            SENTINEL.as_bytes()
1449        );
1450        let as_string = format!(r#"{{"name":"authorization","value":"{SENTINEL}"}}"#);
1451        for json in [as_array, as_string] {
1452            let error = serde_json::from_str::<HealthCheckHeaderValue>(&json)
1453                .expect_err("the ingress value must never be read from JSON");
1454            assert!(!error.to_string().contains(SENTINEL), "{error}");
1455        }
1456    }
1457
1458    #[test]
1459    fn ingress_debug_prints_the_name_and_a_placeholder_only() {
1460        let message = ingress("authorization", SENTINEL.as_bytes());
1461        let rendered = format!("{message:?}");
1462        assert!(rendered.contains("authorization"), "{rendered}");
1463        assert!(rendered.contains("<protected>"), "{rendered}");
1464        assert!(!rendered.contains(SENTINEL), "{rendered}");
1465        assert!(
1466            !rendered.contains(&SENTINEL.len().to_string()),
1467            "{rendered}"
1468        );
1469        let pretty = format!("{message:#?}");
1470        assert!(!pretty.contains(SENTINEL), "{pretty}");
1471        assert!(!pretty.contains(&SENTINEL.len().to_string()), "{pretty}");
1472
1473        let long = ingress(&format!("x-{}", "n".repeat(200)), b"v");
1474        let rendered = format!("{long:?}");
1475        assert!(rendered.contains("name_truncated: true"), "{rendered}");
1476        assert!(!rendered.contains(&"n".repeat(100)), "{rendered}");
1477    }
1478
1479    #[test]
1480    fn secret_debug_never_reveals_the_value() {
1481        let rendered = format!("{:?}", secret(SENTINEL.as_bytes()));
1482        assert!(!rendered.contains(SENTINEL), "{rendered}");
1483        assert!(rendered.contains("<protected>"), "{rendered}");
1484        assert!(
1485            !rendered.contains(&SENTINEL.len().to_string()),
1486            "{rendered}"
1487        );
1488    }
1489
1490    #[test]
1491    fn extraction_moves_the_allocation_and_keeps_the_exact_bytes() {
1492        // Interior HTAB, obs-text and multi-byte UTF-8 are accepted today.
1493        let accepted = "Bearer a\tb \u{e9}\u{4e16}".as_bytes();
1494        let mut message = ingress("authorization", accepted);
1495        let pointer = message.value.as_ptr();
1496        let value = SecretValue::take_from_ingress(&mut message).unwrap();
1497        assert!(message.value.is_empty(), "the generated field is emptied");
1498        assert_eq!(message.name, "authorization", "the name is left in place");
1499        assert_eq!(value.expose_secret_bytes(), accepted);
1500        assert_eq!(value.expose_secret_bytes().as_ptr(), pointer, "moved");
1501
1502        let mut empty = ingress("x-empty", b"");
1503        let value = SecretValue::take_from_ingress(&mut empty).unwrap();
1504        assert_eq!(value.expose_secret_bytes(), b"");
1505    }
1506
1507    #[test]
1508    fn extraction_refuses_with_static_errors_and_still_empties_the_field() {
1509        let sentinel = SENTINEL.as_bytes();
1510        let with = |prefix: &[u8], suffix: &[u8]| [prefix, sentinel, suffix].concat();
1511        let cases: [(Vec<u8>, SecretValueError); 8] = [
1512            (with(b"\xff\xfe", b""), SecretValueError::NotUtf8),
1513            (with(b"", b"\xc3"), SecretValueError::NotUtf8),
1514            (with(b"a\0", b""), SecretValueError::Invalid(CONTROL)),
1515            (with(b"", b"\r\nX: y"), SecretValueError::Invalid(CONTROL)),
1516            (with(b"\x7f", b""), SecretValueError::Invalid(CONTROL)),
1517            (with(b" ", b""), SecretValueError::Invalid(EDGE)),
1518            (with(b"", b" "), SecretValueError::Invalid(EDGE)),
1519            (with(b"\t", b""), SecretValueError::Invalid(EDGE)),
1520        ];
1521        for (bytes, expected) in cases {
1522            let mut message = ingress("authorization", &bytes);
1523            let error = SecretValue::take_from_ingress(&mut message)
1524                .expect_err("the value must be refused");
1525            assert_eq!(error, expected);
1526            assert!(message.value.is_empty(), "refusal still empties the field");
1527            for rendered in [error.to_string(), format!("{error:?}")] {
1528                assert!(!rendered.contains(SENTINEL), "{rendered}");
1529                assert!(!rendered.contains(&bytes.len().to_string()), "{rendered}");
1530            }
1531        }
1532        let trailing_tab = with(b"", b"\t");
1533        let error = SecretValue::take_from_ingress(&mut ingress("a", &trailing_tab));
1534        assert_eq!(error.unwrap_err(), SecretValueError::Invalid(EDGE));
1535    }
1536
1537    const CONTROL: &str = "health check header value must not contain control bytes";
1538    const EDGE: &str = "health check header value must not begin or end with a space or a tab";
1539
1540    #[test]
1541    fn raw_hmac_sha256_matches_rfc_4231_test_case_1() {
1542        let tag = hmac_sha256(&[0x0b; 20], &[b"Hi There"]);
1543        assert_eq!(
1544            hex::encode(tag),
1545            "b0344c61d8db38535ca8afceaf0bf12b881dc200c9833da726e9376c2e32cff7"
1546        );
1547        // Streaming the message in pieces is the same MAC.
1548        assert_eq!(tag, hmac_sha256(&[0x0b; 20], &[b"Hi ", b"There"]));
1549    }
1550
1551    #[test]
1552    fn fingerprint_is_the_full_tag_and_is_stable() {
1553        let key = key(7, 9);
1554        let value = secret(SENTINEL.as_bytes());
1555        let first = key.fingerprint("Authorization", &value);
1556        let second = key.fingerprint("authorization", &value);
1557        assert_eq!(first, second, "the name is lowercased");
1558
1559        let text = first.to_string();
1560        let parts: Vec<&str> = text.split(':').collect();
1561        assert_eq!(parts.len(), 4, "{text}");
1562        assert_eq!(parts[0], "sozu-hc-hmac-sha256");
1563        assert_eq!(parts[1], "v1");
1564        assert_eq!(parts[2], "09".repeat(16));
1565        assert_eq!(parts[3].len(), 64, "full 256-bit tag: {text}");
1566        assert!(
1567            parts[3]
1568                .bytes()
1569                .all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
1570        );
1571        assert!(!text.contains(SENTINEL));
1572
1573        assert_eq!(HeaderFingerprint::from_str(&text).unwrap(), first);
1574        assert!(key.verify("authorization", &value, &first));
1575    }
1576
1577    #[test]
1578    fn fingerprint_depends_on_name_value_key_and_epoch() {
1579        let value = secret(b"one");
1580        let base = key(1, 1).fingerprint("authorization", &value);
1581        assert_ne!(base, key(1, 1).fingerprint("x-other", &value));
1582        assert_ne!(
1583            base,
1584            key(1, 1).fingerprint("authorization", &secret(b"two"))
1585        );
1586        assert_ne!(base, key(2, 1).fingerprint("authorization", &value));
1587        assert_ne!(base, key(1, 2).fingerprint("authorization", &value));
1588        // The epoch is part of the MAC input, not only a label on the output.
1589        let tag = |fingerprint: HeaderFingerprint| {
1590            fingerprint
1591                .to_string()
1592                .rsplit(':')
1593                .next()
1594                .unwrap()
1595                .to_owned()
1596        };
1597        assert_ne!(
1598            tag(base),
1599            tag(key(1, 2).fingerprint("authorization", &value))
1600        );
1601        // Length prefixes keep the name/value boundary unambiguous.
1602        assert_ne!(
1603            key(1, 1).fingerprint("ab", &secret(b"c")),
1604            key(1, 1).fingerprint("a", &secret(b"bc")),
1605        );
1606    }
1607
1608    #[test]
1609    fn verification_needs_the_same_epoch_and_the_same_key() {
1610        let value = secret(SENTINEL.as_bytes());
1611        let issued = key(1, 5).fingerprint("authorization", &value);
1612        assert!(key(1, 5).verify("authorization", &value, &issued));
1613        assert!(!key(1, 5).verify("x-other", &value, &issued));
1614        assert!(!key(1, 5).verify("authorization", &secret(b"other"), &issued));
1615        // Same public epoch, another key: an epoch collision is not acceptance.
1616        assert!(!key(2, 5).verify("authorization", &value, &issued));
1617        // Another epoch, same key.
1618        assert!(!key(1, 6).verify("authorization", &value, &issued));
1619    }
1620
1621    #[test]
1622    fn generated_keys_are_independent() {
1623        let (a, b) = (
1624            FingerprintKey::generate().unwrap(),
1625            FingerprintKey::generate().unwrap(),
1626        );
1627        let value = secret(b"v");
1628        let fingerprint = a.fingerprint("authorization", &value);
1629        assert!(a.verify("authorization", &value, &fingerprint));
1630        assert!(!b.verify("authorization", &value, &fingerprint));
1631        let rendered = format!("{a:?}");
1632        assert!(rendered.contains("<protected>"), "{rendered}");
1633    }
1634
1635    #[test]
1636    fn parsing_rejects_every_malformed_fingerprint() {
1637        let epoch = "ab".repeat(16);
1638        let tag = "cd".repeat(32);
1639        let ok = format!("sozu-hc-hmac-sha256:v1:{epoch}:{tag}");
1640        assert!(HeaderFingerprint::from_str(&ok).is_ok());
1641
1642        let cases = [
1643            (String::new(), FingerprintParseError::Malformed),
1644            (
1645                "sozu-hc-hmac-sha256:v1".to_owned(),
1646                FingerprintParseError::Malformed,
1647            ),
1648            (format!("{ok}:extra"), FingerprintParseError::Malformed),
1649            (
1650                format!("sozu-hc-hmac-sha512:v1:{epoch}:{tag}"),
1651                FingerprintParseError::Algorithm,
1652            ),
1653            (
1654                format!("SOZU-HC-HMAC-SHA256:v1:{epoch}:{tag}"),
1655                FingerprintParseError::Algorithm,
1656            ),
1657            (
1658                format!("sozu-hc-hmac-sha256:v2:{epoch}:{tag}"),
1659                FingerprintParseError::Version,
1660            ),
1661            (
1662                format!("sozu-hc-hmac-sha256:V1:{epoch}:{tag}"),
1663                FingerprintParseError::Version,
1664            ),
1665            (
1666                format!("sozu-hc-hmac-sha256:v1:{}:{tag}", &epoch[2..]),
1667                FingerprintParseError::Epoch,
1668            ),
1669            (
1670                format!("sozu-hc-hmac-sha256:v1:{}:{tag}", "AB".repeat(16)),
1671                FingerprintParseError::Epoch,
1672            ),
1673            (
1674                format!("sozu-hc-hmac-sha256:v1:{}:{tag}", "zz".repeat(16)),
1675                FingerprintParseError::Epoch,
1676            ),
1677            (
1678                format!("sozu-hc-hmac-sha256:v1:{epoch}:{}", &tag[2..]),
1679                FingerprintParseError::Tag,
1680            ),
1681            (
1682                format!("sozu-hc-hmac-sha256:v1:{epoch}:{}", "CD".repeat(32)),
1683                FingerprintParseError::Tag,
1684            ),
1685            (
1686                format!("sozu-hc-hmac-sha256:v1:{epoch}:{}0", tag),
1687                FingerprintParseError::Tag,
1688            ),
1689            (
1690                format!("sozu-hc-hmac-sha256:v1:{epoch}:{}", "\u{e9}".repeat(32)),
1691                FingerprintParseError::Tag,
1692            ),
1693            (
1694                format!("sozu-hc-hmac-sha256:v1:{epoch}:{}", "+".repeat(64)),
1695                FingerprintParseError::Tag,
1696            ),
1697        ];
1698        for (input, expected) in cases {
1699            assert_eq!(
1700                HeaderFingerprint::from_str(&input),
1701                Err(expected),
1702                "{input}"
1703            );
1704        }
1705    }
1706
1707    /// A header whose value is `value`, fingerprinted under the fixed test key.
1708    fn header(name: &str, value: &[u8]) -> ProtectedHeader {
1709        ProtectedHeader::new(&key(3, 4), name.to_owned(), secret(value))
1710    }
1711
1712    fn set(entries: &[(&str, &str)]) -> ProtectedHeaderSet {
1713        ProtectedHeaderSet::new(
1714            entries
1715                .iter()
1716                .map(|(name, value)| header(name, value.as_bytes()))
1717                .collect(),
1718        )
1719    }
1720
1721    fn replace(entries: &[(&str, &str)]) -> ProtectedHeaderChange {
1722        ProtectedHeaderChange::Replace(set(entries))
1723    }
1724
1725    /// `expected` is `None` for an absent cluster, `Some(set)` for the very
1726    /// same `Arc`-backed set (an empty one for a cleared cluster).
1727    fn assert_effective(
1728        journal: &ProtectedHeaderJournal,
1729        expected: Option<&ProtectedHeaderSet>,
1730        context: &str,
1731    ) {
1732        match (journal.effective("c"), expected) {
1733            (None, None) => {}
1734            (Some(actual), Some(expected)) => {
1735                assert!(
1736                    actual.ptr_eq(expected),
1737                    "{context}: another set is effective"
1738                )
1739            }
1740            (actual, expected) => panic!(
1741                "{context}: effective present={} expected present={}",
1742                actual.is_some(),
1743                expected.is_some()
1744            ),
1745        }
1746    }
1747
1748    fn pending_total(journal: &ProtectedHeaderJournal) -> usize {
1749        journal.clusters.values().map(|c| c.pending.len()).sum()
1750    }
1751
1752    /// `old` accepted, then `A` and `B` proposed in that order.
1753    struct Chain {
1754        journal: ProtectedHeaderJournal,
1755        old: ProtectedHeaderSet,
1756        a: ProtectedHeaderSet,
1757        b: ProtectedHeaderSet,
1758        va: ProtectedHeaderVersion,
1759        vb: ProtectedHeaderVersion,
1760    }
1761
1762    fn chain() -> Chain {
1763        let mut journal = ProtectedHeaderJournal::default();
1764        let old = set(&[("authorization", "old")]);
1765        let v_old = journal
1766            .propose("c", ProtectedHeaderChange::Replace(old.clone()))
1767            .unwrap();
1768        assert_eq!(journal.accept("c", v_old), Resolution::Applied);
1769        let a = set(&[("authorization", "a")]);
1770        let b = set(&[("authorization", "b")]);
1771        let va = journal
1772            .propose("c", ProtectedHeaderChange::Replace(a.clone()))
1773            .unwrap();
1774        assert_effective(&journal, Some(&a), "A is optimistically effective");
1775        let vb = journal
1776            .propose("c", ProtectedHeaderChange::Replace(b.clone()))
1777            .unwrap();
1778        assert_effective(&journal, Some(&b), "B is optimistically effective");
1779        Chain {
1780            journal,
1781            old,
1782            a,
1783            b,
1784            va,
1785            vb,
1786        }
1787    }
1788
1789    #[test]
1790    fn arc_handles_clone_without_copying_secret_bytes() {
1791        // `SecretValue: Clone` does not exist; cloning must still compile.
1792        let original = header("authorization", SENTINEL.as_bytes());
1793        let cloned = original.clone();
1794        assert!(Arc::ptr_eq(original.value(), cloned.value()));
1795        assert_eq!(
1796            original.value().expose_secret_bytes().as_ptr(),
1797            cloned.value().expose_secret_bytes().as_ptr(),
1798            "the bytes are shared, not copied"
1799        );
1800        assert_eq!(Arc::strong_count(original.value()), 2);
1801
1802        let handle = ProtectedHeaderSet::new(vec![original]);
1803        let handle_clone = handle.clone();
1804        assert!(handle.ptr_eq(&handle_clone));
1805        assert!(Arc::ptr_eq(
1806            handle.headers()[0].value(),
1807            handle_clone.headers()[0].value()
1808        ));
1809    }
1810
1811    #[test]
1812    fn set_keeps_header_order_and_fingerprints() {
1813        let key = key(3, 4);
1814        let headers = set(&[
1815            ("x-zulu", "one"),
1816            ("authorization", SENTINEL),
1817            ("x-alpha", "one"),
1818        ]);
1819        let names: Vec<&str> = headers.headers().iter().map(|h| h.name()).collect();
1820        assert_eq!(names, ["x-zulu", "authorization", "x-alpha"]);
1821        for header in headers.headers() {
1822            assert!(key.verify(header.name(), header.value(), header.fingerprint()));
1823        }
1824        // Equality inside one epoch is visible through equal fingerprints.
1825        let fingerprints: Vec<&HeaderFingerprint> =
1826            headers.headers().iter().map(|h| h.fingerprint()).collect();
1827        assert_ne!(fingerprints[0], fingerprints[2], "the name is MACed too");
1828        assert_eq!(
1829            header("x-zulu", b"one").fingerprint(),
1830            fingerprints[0],
1831            "same name and value, same epoch, same fingerprint"
1832        );
1833        assert_eq!(headers.len(), 3);
1834        assert!(!headers.is_empty());
1835    }
1836
1837    #[test]
1838    fn protected_debug_prints_no_value_and_no_length() {
1839        let value = format!("{SENTINEL}-{}", "z".repeat(51));
1840        let headers = set(&[("authorization", &value), ("x-other", &value)]);
1841        let mut journal = ProtectedHeaderJournal::default();
1842        journal
1843            .propose("c", ProtectedHeaderChange::Replace(headers.clone()))
1844            .unwrap();
1845        let renderings = [
1846            format!("{:?}", headers.headers()[0]),
1847            format!("{:#?}", headers.headers()[0]),
1848            format!("{headers:?}"),
1849            format!("{headers:#?}"),
1850            format!("{journal:?}"),
1851            format!("{journal:#?}"),
1852        ];
1853        for rendered in renderings {
1854            assert!(!rendered.contains(SENTINEL), "{rendered}");
1855            assert!(!rendered.contains("zzzz"), "{rendered}");
1856            // The fingerprints are hex: take them out before looking for the
1857            // decimal length, which they could contain by chance.
1858            let mut without_hex = rendered.clone();
1859            for header in headers.headers() {
1860                without_hex = without_hex.replace(&header.fingerprint().to_string(), "");
1861            }
1862            assert!(
1863                !without_hex.contains(&value.len().to_string()),
1864                "{rendered}"
1865            );
1866            assert!(!rendered.contains("len"), "{rendered}");
1867        }
1868        let rendered = format!("{:?}", header(&format!("x-{}", "n".repeat(200)), b"v"));
1869        assert!(!rendered.contains(&"n".repeat(100)), "{rendered}");
1870    }
1871
1872    #[test]
1873    fn rejecting_a_then_b_restores_the_base() {
1874        let mut c = chain();
1875        assert_eq!(c.journal.reject("c", c.va), Resolution::Applied);
1876        assert_effective(&c.journal, Some(&c.b), "A rejected, B is still live");
1877        assert_eq!(c.journal.reject("c", c.vb), Resolution::Applied);
1878        assert_effective(&c.journal, Some(&c.old), "both rejected");
1879    }
1880
1881    #[test]
1882    fn rejecting_a_then_accepting_b_keeps_b() {
1883        let mut c = chain();
1884        c.journal.reject("c", c.va);
1885        assert_effective(&c.journal, Some(&c.b), "A rejected, B is still live");
1886        assert_eq!(c.journal.accept("c", c.vb), Resolution::Applied);
1887        assert_effective(&c.journal, Some(&c.b), "B accepted");
1888    }
1889
1890    #[test]
1891    fn rejecting_b_then_accepting_a_keeps_a() {
1892        let mut c = chain();
1893        c.journal.reject("c", c.vb);
1894        assert_effective(&c.journal, Some(&c.a), "B rejected, A is live");
1895        assert_eq!(c.journal.accept("c", c.va), Resolution::Applied);
1896        assert_effective(&c.journal, Some(&c.a), "A accepted");
1897    }
1898
1899    #[test]
1900    fn rejecting_b_then_a_restores_the_base() {
1901        let mut c = chain();
1902        c.journal.reject("c", c.vb);
1903        assert_effective(&c.journal, Some(&c.a), "B rejected, A is live");
1904        c.journal.reject("c", c.va);
1905        assert_effective(&c.journal, Some(&c.old), "both rejected");
1906    }
1907
1908    #[test]
1909    fn accepting_a_then_rejecting_b_keeps_a() {
1910        let mut c = chain();
1911        c.journal.accept("c", c.va);
1912        assert_effective(&c.journal, Some(&c.b), "A accepted, B is still live");
1913        c.journal.reject("c", c.vb);
1914        assert_effective(&c.journal, Some(&c.a), "B rejected over an accepted A");
1915    }
1916
1917    #[test]
1918    fn accepting_b_then_rejecting_a_keeps_b() {
1919        let mut c = chain();
1920        c.journal.accept("c", c.vb);
1921        assert_effective(&c.journal, Some(&c.b), "B accepted");
1922        assert_eq!(c.journal.reject("c", c.va), Resolution::Ignored);
1923        assert_effective(&c.journal, Some(&c.b), "a stale ancestor cannot displace B");
1924    }
1925
1926    #[test]
1927    fn clear_leaves_a_present_empty_set_and_failed_clear_restores_the_predecessor() {
1928        let mut c = chain();
1929        let cleared = c
1930            .journal
1931            .propose("c", ProtectedHeaderChange::Clear)
1932            .unwrap();
1933        let effective = c.journal.effective("c").expect("a cleared cluster remains");
1934        assert!(effective.is_empty());
1935        c.journal.accept("c", cleared);
1936        let effective = c.journal.effective("c").expect("a cleared cluster remains");
1937        assert!(effective.is_empty());
1938        assert!(c.journal.clusters.contains_key("c"), "not a tombstone");
1939        // The older proposals were superseded by the accepted clear.
1940        assert_eq!(pending_total(&c.journal), 0);
1941
1942        let mut c = chain();
1943        let cleared = c
1944            .journal
1945            .propose("c", ProtectedHeaderChange::Clear)
1946            .unwrap();
1947        assert!(c.journal.effective("c").unwrap().is_empty());
1948        c.journal.reject("c", cleared);
1949        assert_effective(&c.journal, Some(&c.b), "failed clear restores B");
1950        c.journal.reject("c", c.vb);
1951        c.journal.reject("c", c.va);
1952        assert_effective(&c.journal, Some(&c.old), "and then the base");
1953    }
1954
1955    #[test]
1956    fn remove_leaves_the_cluster_absent_and_failed_remove_restores_the_predecessor() {
1957        let mut c = chain();
1958        let removed = c
1959            .journal
1960            .propose("c", ProtectedHeaderChange::Remove)
1961            .unwrap();
1962        assert_effective(&c.journal, None, "removal is optimistically effective");
1963        c.journal.accept("c", removed);
1964        assert_effective(&c.journal, None, "removed");
1965        assert!(c.journal.clusters.is_empty(), "no tombstone is kept");
1966
1967        let mut c = chain();
1968        let removed = c
1969            .journal
1970            .propose("c", ProtectedHeaderChange::Remove)
1971            .unwrap();
1972        c.journal.reject("c", removed);
1973        assert_effective(&c.journal, Some(&c.b), "failed removal restores B");
1974    }
1975
1976    #[test]
1977    fn late_completion_after_a_successful_removal_cannot_resurrect() {
1978        for late_accept in [true, false] {
1979            let mut c = chain();
1980            let removed = c
1981                .journal
1982                .propose("c", ProtectedHeaderChange::Remove)
1983                .unwrap();
1984            c.journal.accept("c", removed);
1985            for version in [c.va, c.vb] {
1986                let resolution = if late_accept {
1987                    c.journal.accept("c", version)
1988                } else {
1989                    c.journal.reject("c", version)
1990                };
1991                assert_eq!(resolution, Resolution::Ignored);
1992                assert_effective(&c.journal, None, "no resurrection");
1993            }
1994            assert!(c.journal.clusters.is_empty());
1995        }
1996        // An older proposal accepted before the removal completes does not
1997        // outlive it either.
1998        let mut c = chain();
1999        let removed = c
2000            .journal
2001            .propose("c", ProtectedHeaderChange::Remove)
2002            .unwrap();
2003        c.journal.accept("c", c.va);
2004        assert_effective(&c.journal, None, "the removal is still the head");
2005        c.journal.accept("c", removed);
2006        assert_effective(&c.journal, None, "removed");
2007    }
2008
2009    #[test]
2010    fn duplicate_and_unknown_resolutions_change_nothing() {
2011        let mut c = chain();
2012        c.journal.accept("c", c.vb);
2013        let (last, pending, clusters) = (
2014            c.journal.last_issued,
2015            pending_total(&c.journal),
2016            c.journal.clusters.len(),
2017        );
2018        for (cluster, version) in [("c", c.vb), ("c", c.va), ("other", c.vb)] {
2019            assert_eq!(c.journal.accept(cluster, version), Resolution::Ignored);
2020            assert_eq!(c.journal.reject(cluster, version), Resolution::Ignored);
2021            assert_effective(&c.journal, Some(&c.b), "unchanged");
2022        }
2023        let unknown = ProtectedHeaderVersion(last + 1000);
2024        assert_eq!(c.journal.accept("c", unknown), Resolution::Ignored);
2025        assert_eq!(c.journal.reject("c", unknown), Resolution::Ignored);
2026        assert_eq!(
2027            (
2028                c.journal.last_issued,
2029                pending_total(&c.journal),
2030                c.journal.clusters.len()
2031            ),
2032            (last, pending, clusters)
2033        );
2034        assert_effective(&c.journal, Some(&c.b), "unchanged");
2035
2036        // A rejected id cannot be accepted afterwards.
2037        let mut c = chain();
2038        c.journal.reject("c", c.vb);
2039        assert_eq!(c.journal.accept("c", c.vb), Resolution::Ignored);
2040        assert_eq!(c.journal.reject("c", c.vb), Resolution::Ignored);
2041        assert_effective(&c.journal, Some(&c.a), "B stays rejected");
2042        // An id of another cluster resolves nothing here.
2043        let other = c
2044            .journal
2045            .propose("other", ProtectedHeaderChange::Clear)
2046            .unwrap();
2047        assert_eq!(c.journal.reject("c", other), Resolution::Ignored);
2048        assert!(c.journal.effective("other").unwrap().is_empty());
2049    }
2050
2051    #[test]
2052    fn versions_are_unique_and_increasing_across_clusters_and_resolutions() {
2053        let mut journal = ProtectedHeaderJournal::default();
2054        let mut seen = Vec::new();
2055        for round in 0..4 {
2056            for cluster in ["c", "d"] {
2057                let version = journal.propose(cluster, replace(&[("a", "1")])).unwrap();
2058                seen.push(version);
2059                if round % 2 == 0 {
2060                    journal.reject(cluster, version);
2061                } else {
2062                    journal.accept(cluster, version);
2063                }
2064            }
2065        }
2066        assert!(seen.windows(2).all(|pair| pair[0] < pair[1]), "{seen:?}");
2067    }
2068
2069    #[test]
2070    fn version_exhaustion_refuses_before_any_mutation() {
2071        let mut journal = ProtectedHeaderJournal::with_last_issued(u64::MAX - 1);
2072        let kept = set(&[("authorization", "kept")]);
2073        let last = journal
2074            .propose("c", ProtectedHeaderChange::Replace(kept.clone()))
2075            .unwrap();
2076        assert_eq!(last, ProtectedHeaderVersion(u64::MAX));
2077        let (issued, pending, clusters) = (
2078            journal.last_issued,
2079            pending_total(&journal),
2080            journal.clusters.len(),
2081        );
2082        for (cluster, change) in [
2083            ("c", ProtectedHeaderChange::Remove),
2084            ("fresh", ProtectedHeaderChange::Clear),
2085            ("c", replace(&[("x", "y")])),
2086        ] {
2087            assert_eq!(
2088                journal.propose(cluster, change).unwrap_err(),
2089                VersionExhausted
2090            );
2091            assert_eq!(
2092                (
2093                    journal.last_issued,
2094                    pending_total(&journal),
2095                    journal.clusters.len()
2096                ),
2097                (issued, pending, clusters)
2098            );
2099            assert_effective(&journal, Some(&kept), "head untouched");
2100        }
2101        assert_eq!(journal.accept("c", last), Resolution::Applied);
2102        assert_effective(&journal, Some(&kept), "the last id still resolves");
2103    }
2104
2105    #[test]
2106    fn compaction_drops_every_resolved_chain_and_keeps_the_effective_handle() {
2107        let mut c = chain();
2108        c.journal.accept("c", c.vb);
2109        assert_eq!(pending_total(&c.journal), 0, "A is superseded by B");
2110        assert_effective(&c.journal, Some(&c.b), "exact Arc preserved");
2111        // Only the effective set (confirmed) and its two handles remain: the
2112        // fixture `b`, the journal; `a` and `old` are no longer held by it.
2113        assert_eq!(Arc::strong_count(&c.b.0), 2);
2114        assert_eq!(Arc::strong_count(&c.a.0), 1);
2115        assert_eq!(Arc::strong_count(&c.old.0), 1);
2116
2117        let mut c = chain();
2118        c.journal.reject("c", c.va);
2119        c.journal.reject("c", c.vb);
2120        assert_eq!(pending_total(&c.journal), 0);
2121        assert_effective(&c.journal, Some(&c.old), "exact Arc preserved");
2122        assert_eq!(Arc::strong_count(&c.a.0), 1, "payload of A released");
2123        assert_eq!(Arc::strong_count(&c.b.0), 1, "payload of B released");
2124
2125        // Churn does not grow the store.
2126        let mut journal = ProtectedHeaderJournal::default();
2127        for round in 0..100 {
2128            let version = journal.propose("c", replace(&[("a", "1")])).unwrap();
2129            if round % 3 == 0 {
2130                journal.reject("c", version);
2131            } else {
2132                journal.accept("c", version);
2133            }
2134            assert_eq!(pending_total(&journal), 0);
2135            assert!(journal.clusters.len() <= 1);
2136        }
2137        let removal = journal.propose("c", ProtectedHeaderChange::Remove).unwrap();
2138        journal.accept("c", removal);
2139        assert!(journal.clusters.is_empty());
2140    }
2141
2142    #[test]
2143    fn parse_errors_are_static() {
2144        let error = HeaderFingerprint::from_str(&format!("x:{SENTINEL}")).unwrap_err();
2145        assert!(!error.to_string().contains(SENTINEL));
2146        assert!(!format!("{error:?}").contains(SENTINEL));
2147    }
2148
2149    // ---- protected health-check orders -------------------------------------
2150
2151    use crate::proto::command::{
2152        HealthCheckConfig, HealthCheckHeader, HealthCheckMode, SetProtectedHealthCheck,
2153    };
2154
2155    fn order(headers: &[(&str, &str)]) -> SetProtectedHealthCheck {
2156        SetProtectedHealthCheck {
2157            cluster_id: "app".to_owned(),
2158            config: HealthCheckConfig {
2159                uri: "/health".to_owned(),
2160                interval: 10,
2161                timeout: 5,
2162                healthy_threshold: 3,
2163                unhealthy_threshold: 3,
2164                mode: HealthCheckMode::Http as i32,
2165                headers: headers
2166                    .iter()
2167                    .map(|(name, _)| HealthCheckHeader {
2168                        name: (*name).to_owned(),
2169                        fingerprint: String::new(),
2170                    })
2171                    .collect(),
2172                ..Default::default()
2173            },
2174            values: headers
2175                .iter()
2176                .map(|(name, value)| ingress(name, value.as_bytes()))
2177                .collect(),
2178        }
2179    }
2180
2181    #[test]
2182    fn ingress_order_gets_server_fingerprints_and_moves_its_values_out() {
2183        let key = key(7, 8);
2184        let mut incoming = order(&[("Authorization", SENTINEL), ("X-Second", "two")]);
2185        let protected = ProtectedHealthCheck::from_ingress(&key, &mut incoming).unwrap();
2186
2187        assert_eq!(protected.cluster_id(), "app");
2188        let set = protected.headers();
2189        assert_eq!(
2190            set.headers().iter().map(|h| h.name()).collect::<Vec<_>>(),
2191            ["Authorization", "X-Second"],
2192            "order is kept"
2193        );
2194        for (header, meta) in set.headers().iter().zip(&protected.config().headers) {
2195            assert_eq!(meta.name, header.name());
2196            assert_eq!(meta.fingerprint, header.fingerprint().to_string());
2197            assert!(key.verify(header.name(), header.value(), header.fingerprint()));
2198        }
2199        assert_eq!(
2200            set.headers()[0].value().expose_secret_bytes(),
2201            SENTINEL.as_bytes()
2202        );
2203        assert!(
2204            incoming
2205                .values
2206                .iter()
2207                .all(|carrier| carrier.value.is_empty()),
2208            "the clear bytes moved out of the ingress carriers"
2209        );
2210        assert!(protected.satisfies(&protected.config().headers));
2211    }
2212
2213    #[test]
2214    fn ingress_order_refusals_name_no_value_and_a_spoofed_fingerprint_is_one() {
2215        let key = key(7, 8);
2216        let genuine = {
2217            let mut incoming = order(&[("Authorization", SENTINEL)]);
2218            ProtectedHealthCheck::from_ingress(&key, &mut incoming)
2219                .unwrap()
2220                .config()
2221                .headers[0]
2222                .fingerprint
2223                .to_owned()
2224        };
2225
2226        // A client fingerprint that is the genuine one is accepted; any other is refused.
2227        let mut same = order(&[("Authorization", SENTINEL)]);
2228        same.config.headers[0].fingerprint = genuine.to_owned();
2229        assert!(ProtectedHealthCheck::from_ingress(&key, &mut same).is_ok());
2230        let mut spoofed = order(&[("Authorization", "another")]);
2231        spoofed.config.headers[0].fingerprint = genuine;
2232        assert_eq!(
2233            ProtectedHealthCheck::from_ingress(&key, &mut spoofed).err(),
2234            Some(ProtectedHealthCheckError::FingerprintMismatch)
2235        );
2236
2237        let mut malformed = order(&[("Authorization", SENTINEL)]);
2238        malformed.config.headers[0].fingerprint = "junk".to_owned();
2239        assert!(ProtectedHealthCheck::from_ingress(&key, &mut malformed).is_err());
2240
2241        let mut short = order(&[("a", SENTINEL), ("b", "2")]);
2242        short.values.pop();
2243        let mut swapped = order(&[("a", SENTINEL), ("b", "2")]);
2244        swapped.values.swap(0, 1);
2245        let mut renamed = order(&[("a", SENTINEL)]);
2246        renamed.values[0].name = "A-Different".to_owned();
2247        let mut bad_value = order(&[("a", SENTINEL)]);
2248        bad_value.values[0].value = format!(" {SENTINEL}").into_bytes();
2249        let mut bad_utf8 = order(&[("a", SENTINEL)]);
2250        bad_utf8.values[0].value = vec![0xff];
2251        let mut bad_name = order(&[("Host", SENTINEL)]);
2252        let mut tcp = order(&[("a", SENTINEL)]);
2253        tcp.config.mode = HealthCheckMode::Tcp as i32;
2254        bad_name.values[0].name = "Host".to_owned();
2255        for (what, mut incoming) in [
2256            ("short", short),
2257            ("swapped", swapped),
2258            ("renamed", renamed),
2259            ("bad value", bad_value),
2260            ("bad utf8", bad_utf8),
2261            ("reserved name", bad_name),
2262            ("tcp mode", tcp),
2263        ] {
2264            let error = ProtectedHealthCheck::from_ingress(&key, &mut incoming)
2265                .err()
2266                .unwrap_or_else(|| panic!("{what} must be refused"));
2267            let rendered = format!("{error} {error:?}");
2268            assert!(!rendered.contains(SENTINEL), "{what}: {rendered}");
2269        }
2270    }
2271
2272    #[test]
2273    fn trusted_order_round_trips_and_demands_a_fingerprint_per_header() {
2274        let key = key(7, 8);
2275        let mut incoming = order(&[("Authorization", SENTINEL), ("X-Second", "two")]);
2276        let main = ProtectedHealthCheck::from_ingress(&key, &mut incoming).unwrap();
2277
2278        let mut forwarded = main.to_order();
2279        let worker = ProtectedHealthCheck::from_trusted(&mut forwarded).unwrap();
2280        assert_eq!(worker.config(), main.config());
2281        for (a, b) in worker
2282            .headers()
2283            .headers()
2284            .iter()
2285            .zip(main.headers().headers())
2286        {
2287            assert_eq!(a.name(), b.name());
2288            assert_eq!(a.fingerprint(), b.fingerprint());
2289            assert_eq!(
2290                a.value().expose_secret_bytes(),
2291                b.value().expose_secret_bytes()
2292            );
2293        }
2294        assert!(
2295            forwarded
2296                .values
2297                .iter()
2298                .all(|carrier| carrier.value.is_empty())
2299        );
2300
2301        // A worker never takes an unfingerprinted order as trusted.
2302        let mut bare = order(&[("Authorization", SENTINEL)]);
2303        assert_eq!(
2304            ProtectedHealthCheck::from_trusted(&mut bare).err(),
2305            Some(ProtectedHealthCheckError::FingerprintMissing)
2306        );
2307    }
2308
2309    #[test]
2310    fn a_header_set_satisfies_only_the_metadata_it_was_built_from() {
2311        let key = key(7, 8);
2312        let mut incoming = order(&[("Authorization", SENTINEL), ("X-Second", "two")]);
2313        let protected = ProtectedHealthCheck::from_ingress(&key, &mut incoming).unwrap();
2314        let metadata = protected.config().headers.to_owned();
2315        assert!(protected.headers().satisfies(&metadata));
2316
2317        let mut reordered = metadata.to_owned();
2318        reordered.reverse();
2319        let mut other_value = metadata.to_owned();
2320        other_value[0].fingerprint = {
2321            let mut other = order(&[("Authorization", "another")]);
2322            ProtectedHealthCheck::from_ingress(&key, &mut other)
2323                .unwrap()
2324                .config()
2325                .headers[0]
2326                .fingerprint
2327                .to_owned()
2328        };
2329        let mut missing = metadata.to_owned();
2330        missing.pop();
2331        for (what, headers) in [
2332            ("reordered", reordered),
2333            ("other value", other_value),
2334            ("fewer", missing),
2335        ] {
2336            assert!(!protected.headers().satisfies(&headers), "{what}");
2337        }
2338        // No metadata is satisfied by no set, and by an empty one.
2339        assert!(ProtectedHeaderSet::new(Vec::new()).satisfies(&[]));
2340        assert!(!ProtectedHeaderSet::new(Vec::new()).satisfies(&metadata));
2341    }
2342
2343    #[test]
2344    fn a_dropped_ingress_carrier_wipes_its_whole_buffer() {
2345        let mut carrier = ingress("authorization", SENTINEL.as_bytes());
2346        carrier.value.reserve(256);
2347        let (pointer, capacity) = (carrier.value.as_ptr(), carrier.value.capacity());
2348        carrier.wipe();
2349        assert!(carrier.value.is_empty());
2350        // SAFETY: `wipe` zeroes the owned allocation in place and the carrier
2351        // still owns it, so all `capacity` bytes are initialised and live.
2352        let bytes = unsafe { std::slice::from_raw_parts(pointer, capacity) };
2353        assert!(bytes.iter().all(|&b| b == 0), "the buffer was not wiped");
2354    }
2355
2356    /// A reader that hands out at most `step` bytes per call.
2357    struct Trickle<'a> {
2358        data: &'a [u8],
2359        step: usize,
2360    }
2361
2362    impl Read for Trickle<'_> {
2363        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
2364            let n = self.step.min(buf.len()).min(self.data.len());
2365            buf[..n].copy_from_slice(&self.data[..n]);
2366            self.data = &self.data[n..];
2367            Ok(n)
2368        }
2369    }
2370
2371    /// Some bytes, then a failure carrying the sentinel in its message.
2372    struct Failing(bool);
2373
2374    impl Read for Failing {
2375        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
2376            if self.0 {
2377                return Err(std::io::Error::other(SENTINEL));
2378            }
2379            self.0 = true;
2380            buf[..4].copy_from_slice(b"abcd");
2381            Ok(4)
2382        }
2383    }
2384
2385    fn read_value(bytes: &[u8], max: usize) -> Result<Vec<u8>, HeaderValueReadError> {
2386        read_header_value("authorization".to_owned(), bytes, max).map(|mut carrier| {
2387            let value = carrier.value.to_vec();
2388            carrier.wipe();
2389            value
2390        })
2391    }
2392
2393    #[test]
2394    fn the_reader_removes_exactly_one_terminal_line_ending() {
2395        let cases: [(&[u8], &[u8]); 14] = [
2396            (b"", b""),
2397            (b"abc", b"abc"),
2398            (b"abc\n", b"abc"),
2399            (b"abc\r\n", b"abc"),
2400            (b"\n", b""),
2401            (b"\r\n", b""),
2402            (b"abc\r", b"abc\r"),
2403            (b"a\nb\n", b"a\nb"),
2404            (b"a\rb\r\n", b"a\rb"),
2405            (b"abc\n\n", b"abc\n"),
2406            (b"abc\r\n\r\n", b"abc\r\n"),
2407            (b"abc\n\r\n", b"abc\n"),
2408            (b"abc \n", b"abc "),
2409            (b"\tabc\t\r\n", b"\tabc\t"),
2410        ];
2411        for (input, expected) in cases {
2412            assert_eq!(read_value(input, 64).as_deref(), Ok(expected), "{input:?}");
2413        }
2414    }
2415
2416    #[test]
2417    fn the_reader_applies_the_maximum_after_the_line_ending_is_removed() {
2418        let max = 16;
2419        let value = vec![b'v'; max];
2420        for terminator in [&b""[..], b"\n", b"\r\n"] {
2421            let input = [value.as_slice(), terminator].concat();
2422            assert_eq!(read_value(&input, max).as_deref(), Ok(value.as_slice()));
2423        }
2424        let over = vec![b'v'; max + 1];
2425        for terminator in [&b""[..], b"\n", b"\r\n"] {
2426            let input = [over.as_slice(), terminator].concat();
2427            assert_eq!(
2428                read_value(&input, max),
2429                Err(HeaderValueReadError::TooLong),
2430                "terminator {terminator:?}"
2431            );
2432        }
2433        // Content that fills the whole allowance and then more.
2434        let flood = vec![b'v'; max + 64];
2435        assert_eq!(read_value(&flood, max), Err(HeaderValueReadError::TooLong));
2436        let doubled = [value.as_slice(), b"\r\n\r\n"].concat();
2437        assert_eq!(
2438            read_value(&doubled, max),
2439            Err(HeaderValueReadError::TooLong)
2440        );
2441    }
2442
2443    #[test]
2444    fn the_reader_reads_at_most_the_maximum_plus_three_bytes() {
2445        let max = 16;
2446        let mut source = std::io::Cursor::new(vec![b'v'; 4096]);
2447        let result = read_header_value("authorization".to_owned(), &mut source, max);
2448        assert_eq!(result.unwrap_err(), HeaderValueReadError::TooLong);
2449        assert_eq!(
2450            source.position(),
2451            (max + 3) as u64,
2452            "the source is not drained"
2453        );
2454    }
2455
2456    #[test]
2457    fn the_reader_assembles_short_reads_and_keeps_the_name() {
2458        let input = b"Bearer a\tb\r\n";
2459        for step in [1, 2, 3, 5] {
2460            let source = Trickle { data: input, step };
2461            let carrier = read_header_value("X-Token".to_owned(), source, 32).unwrap();
2462            assert_eq!(carrier.name, "X-Token");
2463            assert_eq!(carrier.value, b"Bearer a\tb");
2464        }
2465    }
2466
2467    #[test]
2468    fn the_reader_reports_an_io_error_without_the_bytes_or_their_length() {
2469        let error = read_header_value("authorization".to_owned(), Failing(false), 32)
2470            .expect_err("the source fails");
2471        assert_eq!(error, HeaderValueReadError::Io(ErrorKind::Other));
2472        for rendered in [error.to_string(), format!("{error:?}")] {
2473            assert!(!rendered.contains(SENTINEL), "{rendered}");
2474            assert!(!rendered.contains("abcd"), "{rendered}");
2475            assert!(!rendered.contains('4'), "{rendered}");
2476        }
2477        let too_long = read_value(&[b'v'; 40], 8).unwrap_err();
2478        for rendered in [too_long.to_string(), format!("{too_long:?}")] {
2479            assert!(
2480                !rendered.contains("40") && !rendered.contains('8'),
2481                "{rendered}"
2482            );
2483        }
2484    }
2485
2486    #[test]
2487    fn the_reader_leaves_validation_to_the_value_validators() {
2488        // UTF-8, control bytes and edge blanks are not the reader's to refuse.
2489        let mut carrier = read_header_value("a".to_owned(), &b"\xff\0 \r\n"[..], 16).unwrap();
2490        assert_eq!(carrier.value, b"\xff\0 ");
2491        assert!(SecretValue::take_from_ingress(&mut carrier).is_err());
2492        assert!(carrier.value.is_empty(), "the refused bytes are gone");
2493    }
2494
2495    #[test]
2496    fn the_journal_reports_whether_it_retains_any_value() {
2497        let mut journal = ProtectedHeaderJournal::default();
2498        assert!(!journal.holds_values(), "a new journal holds nothing");
2499
2500        // A pending non-empty set is retained, then dropped by its rejection.
2501        let pending = journal
2502            .propose("c", replace(&[("authorization", "a")]))
2503            .unwrap();
2504        assert!(journal.holds_values());
2505        journal.reject("c", pending);
2506        assert!(!journal.holds_values());
2507
2508        // A confirmed non-empty set is retained, in any cluster.
2509        let confirmed = journal
2510            .propose("c", replace(&[("authorization", "a")]))
2511            .unwrap();
2512        journal.accept("c", confirmed);
2513        assert!(journal.holds_values());
2514        let other = journal.propose("d", ProtectedHeaderChange::Clear).unwrap();
2515        journal.accept("d", other);
2516        assert!(
2517            journal.holds_values(),
2518            "another cluster's empty set hides nothing"
2519        );
2520    }
2521
2522    #[test]
2523    fn a_pending_clear_or_remove_still_retains_the_confirmed_values() {
2524        for hide in [ProtectedHeaderChange::Clear, ProtectedHeaderChange::Remove] {
2525            let mut journal = ProtectedHeaderJournal::default();
2526            let confirmed = journal
2527                .propose("c", replace(&[("authorization", "a")]))
2528                .unwrap();
2529            journal.accept("c", confirmed);
2530            let hiding = journal.propose("c", hide).unwrap();
2531            assert!(
2532                journal.holds_values(),
2533                "the confirmed bytes survive until the pending version resolves"
2534            );
2535            // Accepting it supersedes the confirmed state: nothing is left.
2536            let mut accepted = ProtectedHeaderJournal::default();
2537            let base = accepted
2538                .propose("c", replace(&[("authorization", "a")]))
2539                .unwrap();
2540            accepted.accept("c", base);
2541            let version = accepted.propose("c", ProtectedHeaderChange::Clear).unwrap();
2542            accepted.accept("c", version);
2543            assert!(
2544                !accepted.holds_values(),
2545                "an accepted clear drops the values"
2546            );
2547            // Rejecting it uncovers the confirmed state, which is still held.
2548            journal.reject("c", hiding);
2549            assert!(journal.holds_values());
2550        }
2551    }
2552
2553    #[test]
2554    fn an_accepted_remove_leaves_nothing_retained() {
2555        let mut journal = ProtectedHeaderJournal::default();
2556        let confirmed = journal
2557            .propose("c", replace(&[("authorization", "a")]))
2558            .unwrap();
2559        journal.accept("c", confirmed);
2560        let removed = journal.propose("c", ProtectedHeaderChange::Remove).unwrap();
2561        assert!(
2562            journal.holds_values(),
2563            "pending remove above confirmed values"
2564        );
2565        journal.accept("c", removed);
2566        assert!(!journal.holds_values());
2567    }
2568
2569    #[test]
2570    fn a_pending_stack_retains_a_non_empty_ancestor_hidden_by_a_later_clear() {
2571        let mut journal = ProtectedHeaderJournal::default();
2572        let a = journal
2573            .propose("c", replace(&[("authorization", "a")]))
2574            .unwrap();
2575        let b = journal.propose("c", ProtectedHeaderChange::Clear).unwrap();
2576        let c = journal.propose("c", ProtectedHeaderChange::Remove).unwrap();
2577        assert!(
2578            journal.effective("c").is_none(),
2579            "the newest pending version is effective"
2580        );
2581        assert!(journal.holds_values(), "ancestor A is hidden but retained");
2582        journal.reject("c", c);
2583        assert!(journal.holds_values());
2584        journal.reject("c", b);
2585        assert!(journal.holds_values(), "A is effective again");
2586        journal.reject("c", a);
2587        assert!(!journal.holds_values());
2588    }
2589
2590    #[test]
2591    fn two_pending_non_empty_versions_are_released_one_by_one() {
2592        let mut journal = ProtectedHeaderJournal::default();
2593        let a = journal
2594            .propose("c", replace(&[("authorization", "a")]))
2595            .unwrap();
2596        let b = journal
2597            .propose("c", replace(&[("authorization", "b")]))
2598            .unwrap();
2599        journal.reject("c", b);
2600        assert!(journal.holds_values(), "A remains pending");
2601        journal.reject("c", a);
2602        assert!(!journal.holds_values());
2603
2604        // Accepting the newer supersedes the older, which then retains nothing.
2605        let a = journal
2606            .propose("c", replace(&[("authorization", "a")]))
2607            .unwrap();
2608        let b = journal
2609            .propose("c", replace(&[("authorization", "b")]))
2610            .unwrap();
2611        journal.accept("c", b);
2612        assert!(journal.holds_values(), "B is confirmed");
2613        assert_eq!(journal.accept("c", a), Resolution::Ignored);
2614        assert!(journal.holds_values());
2615
2616        // Accepting the older keeps the newer pending.
2617        let mut journal = ProtectedHeaderJournal::default();
2618        let a = journal
2619            .propose("c", replace(&[("authorization", "a")]))
2620            .unwrap();
2621        let _b = journal
2622            .propose("c", replace(&[("authorization", "b")]))
2623            .unwrap();
2624        journal.accept("c", a);
2625        assert!(journal.holds_values());
2626    }
2627
2628    #[test]
2629    fn a_secret_held_by_any_cluster_is_found_among_empty_ones() {
2630        let mut journal = ProtectedHeaderJournal::default();
2631        for name in ["a", "b", "c"] {
2632            let version = journal.propose(name, ProtectedHeaderChange::Clear).unwrap();
2633            journal.accept(name, version);
2634        }
2635        assert!(!journal.holds_values(), "only empty sets");
2636        let pending = journal.propose("b", ProtectedHeaderChange::Clear).unwrap();
2637        assert!(!journal.holds_values(), "empty confirmed, empty pending");
2638        journal
2639            .propose("z", replace(&[("authorization", "x")]))
2640            .unwrap();
2641        assert!(journal.holds_values());
2642        journal.reject("b", pending);
2643    }
2644
2645    // ---- ProtectedTransfer ------------------------------------------------
2646
2647    const TRANSFER_SENTINEL: &str = "Bearer SYNTHETIC-TRANSFER-SECRET-91ab";
2648
2649    type Metadata = HashMap<String, Vec<HealthCheckHeader>>;
2650
2651    fn named(name: &str, value: &str, key: &FingerprintKey) -> ProtectedHeader {
2652        ProtectedHeader::new(key, name.to_owned(), secret(value.as_bytes()))
2653    }
2654
2655    /// Confirm `clusters` (a `None` header list is a cleared cluster) and
2656    /// return the journal with the metadata a configuration would name.
2657    fn confirmed(
2658        key: &FingerprintKey,
2659        clusters: &[(&str, &[(&str, &str)])],
2660    ) -> (ProtectedHeaderJournal, Metadata) {
2661        let mut journal = ProtectedHeaderJournal::default();
2662        let mut metadata = Metadata::new();
2663        for (cluster, headers) in clusters {
2664            let set = ProtectedHeaderSet::new(
2665                headers
2666                    .iter()
2667                    .map(|(name, value)| named(name, value, key))
2668                    .collect(),
2669            );
2670            metadata.insert(
2671                (*cluster).to_owned(),
2672                set.headers()
2673                    .iter()
2674                    .map(|header| HealthCheckHeader {
2675                        name: header.name().to_owned(),
2676                        fingerprint: header.fingerprint().to_string(),
2677                    })
2678                    .collect(),
2679            );
2680            let version = journal
2681                .propose(cluster, ProtectedHeaderChange::Replace(set))
2682                .unwrap();
2683            journal.accept(cluster, version);
2684        }
2685        (journal, metadata)
2686    }
2687
2688    fn lookup<'m>(metadata: &'m Metadata) -> impl Fn(&str) -> Option<&'m [HealthCheckHeader]> {
2689        |cluster| metadata.get(cluster).map(Vec::as_slice)
2690    }
2691
2692    fn encode(transfer: &ProtectedTransfer) -> Zeroizing<Vec<u8>> {
2693        let mut out = Zeroizing::new(Vec::with_capacity(transfer.encoded_len().unwrap()));
2694        transfer.encode_into(&mut out).unwrap();
2695        out
2696    }
2697
2698    /// The confirmed state `sample` exports: two secret-bearing clusters and
2699    /// an explicitly empty one, built again for each caller.
2700    fn sample_state() -> (ProtectedHeaderJournal, Metadata, FingerprintKey) {
2701        let key = key(7, 9);
2702        let (journal, metadata) = confirmed(
2703            &key,
2704            &[
2705                (
2706                    "b",
2707                    &[("X-One", "one"), ("Authorization", TRANSFER_SENTINEL)],
2708                ),
2709                ("a", &[("Authorization", "alpha")]),
2710                ("empty", &[]),
2711            ],
2712        );
2713        (journal, metadata, key)
2714    }
2715
2716    fn sample() -> (ProtectedTransfer, Metadata, FingerprintKey) {
2717        let (journal, metadata, key) = sample_state();
2718        let transfer = ProtectedTransfer::export(&journal, Some(&key), lookup(&metadata)).unwrap();
2719        (transfer, metadata, key)
2720    }
2721
2722    #[test]
2723    fn a_confirmed_transfer_round_trips_key_counter_order_and_empty_sets() {
2724        let (transfer, metadata, key) = sample();
2725        let encoded = encode(&transfer);
2726        assert_eq!(encoded.len(), transfer.encoded_len().unwrap());
2727
2728        let decoded = ProtectedTransfer::decode(&encoded).unwrap();
2729        let (restored_key, journal) = decoded.restore(lookup(&metadata)).unwrap();
2730        let restored_key = restored_key.expect("the key travels");
2731        assert_eq!(restored_key.epoch, key.epoch);
2732        assert_eq!(restored_key.key.as_slice(), key.key.as_slice());
2733        assert_eq!(journal.last_issued, 3, "versions issued are carried");
2734
2735        let held = journal.effective("b").unwrap();
2736        let names: Vec<_> = held.headers().iter().map(ProtectedHeader::name).collect();
2737        assert_eq!(names, ["X-One", "Authorization"], "header order is kept");
2738        assert_eq!(
2739            held.headers()[1].value().expose_secret_bytes(),
2740            TRANSFER_SENTINEL.as_bytes()
2741        );
2742        assert!(held.satisfies(&metadata["b"]));
2743        assert!(journal.effective("empty").unwrap().is_empty());
2744        assert!(journal.effective("absent").is_none());
2745        let mut journal = journal;
2746        assert_eq!(
2747            journal
2748                .propose("a", ProtectedHeaderChange::Clear)
2749                .unwrap()
2750                .get(),
2751            4,
2752            "the next version continues the counter"
2753        );
2754    }
2755
2756    #[test]
2757    fn a_runtime_without_a_key_transfers_only_its_counter() {
2758        let journal = ProtectedHeaderJournal::with_last_issued(5);
2759        let metadata = Metadata::new();
2760        let transfer = ProtectedTransfer::export(&journal, None, lookup(&metadata)).unwrap();
2761        let decoded = ProtectedTransfer::decode(&encode(&transfer)).unwrap();
2762        let (key, journal) = decoded.restore(lookup(&metadata)).unwrap();
2763        assert!(key.is_none());
2764        assert_eq!(journal.last_issued, 5);
2765    }
2766
2767    #[test]
2768    fn an_unresolved_version_is_never_exported() {
2769        let key = key(1, 2);
2770        let (mut journal, metadata) = confirmed(&key, &[("a", &[("Authorization", "alpha")])]);
2771        journal
2772            .propose("a", ProtectedHeaderChange::Clear)
2773            .expect("a pending proposal");
2774        assert_eq!(
2775            ProtectedTransfer::export(&journal, Some(&key), lookup(&metadata)).err(),
2776            Some(TransferError::Pending)
2777        );
2778    }
2779
2780    #[test]
2781    fn metadata_that_does_not_name_the_held_set_exactly_is_refused_both_ways() {
2782        let (transfer, metadata, _) = sample();
2783        let encoded = encode(&transfer);
2784        let wrong = |edit: &dyn Fn(&mut Metadata)| {
2785            let mut changed = metadata.clone();
2786            edit(&mut changed);
2787            changed
2788        };
2789        let cases = [
2790            wrong(&|m| {
2791                m.remove("a");
2792            }),
2793            wrong(&|m| m.get_mut("a").unwrap()[0].name = "Other".to_owned()),
2794            wrong(&|m| m.get_mut("a").unwrap()[0].fingerprint.push('0')),
2795            wrong(&|m| m.get_mut("b").unwrap().reverse()),
2796            wrong(&|m| m.get_mut("b").unwrap().truncate(1)),
2797            wrong(&|m| {
2798                m.insert(
2799                    "empty".to_owned(),
2800                    vec![HealthCheckHeader {
2801                        name: "X".to_owned(),
2802                        fingerprint: String::new(),
2803                    }],
2804                );
2805            }),
2806        ];
2807        for changed in &cases {
2808            let decoded = ProtectedTransfer::decode(&encoded).unwrap();
2809            assert_eq!(
2810                decoded.restore(lookup(changed)).err(),
2811                Some(TransferError::Mismatch)
2812            );
2813            // Every case edits metadata the exported journal names, so the
2814            // journal is the whole sample state, not a subset of it.
2815            let (journal, _, key) = sample_state();
2816            assert_eq!(
2817                ProtectedTransfer::export(&journal, Some(&key), lookup(changed)).err(),
2818                Some(TransferError::Mismatch)
2819            );
2820        }
2821    }
2822
2823    #[test]
2824    fn a_tampered_value_or_a_foreign_key_with_the_same_epoch_fails_the_hmac() {
2825        let (transfer, metadata, _) = sample();
2826        let mut encoded = encode(&transfer).to_vec();
2827        // The last byte belongs to the sorted trailing empty cluster, so
2828        // flipping it corrupts the structure instead of a value. `alpha` is
2829        // the one value of cluster `a` and occurs exactly once in the
2830        // encoding; flipping its first byte (`a` -> `` ` ``) keeps a valid
2831        // value and a decodable layout, and only the HMAC can refuse it.
2832        let value = b"alpha";
2833        let at: Vec<usize> = encoded
2834            .windows(value.len())
2835            .enumerate()
2836            .filter(|(_, window)| *window == value)
2837            .map(|(offset, _)| offset)
2838            .collect();
2839        assert_eq!(at.len(), 1, "the value is a unique preimage in the bytes");
2840        encoded[at[0]] ^= 1;
2841        let decoded = ProtectedTransfer::decode(&encoded).unwrap();
2842        assert_eq!(
2843            decoded.restore(lookup(&metadata)).err(),
2844            Some(TransferError::Mismatch)
2845        );
2846
2847        // Same epoch, different key bytes: the epoch alone is not acceptance.
2848        // Layout: version at 0, key flag at 1, then the key from offset 2, so
2849        // offset 2 is the first key byte and the epoch starts at 2 + KEY_LEN.
2850        let (transfer, metadata, _) = sample();
2851        let mut encoded = encode(&transfer).to_vec();
2852        encoded[2] ^= 1;
2853        let decoded = ProtectedTransfer::decode(&encoded).unwrap();
2854        assert_eq!(
2855            decoded.restore(lookup(&metadata)).err(),
2856            Some(TransferError::Mismatch)
2857        );
2858    }
2859
2860    #[test]
2861    fn every_truncation_and_any_trailing_byte_is_refused() {
2862        let (transfer, _, _) = sample();
2863        let encoded = encode(&transfer);
2864        for cut in 0..encoded.len() {
2865            assert!(
2866                ProtectedTransfer::decode(&encoded[..cut]).is_err(),
2867                "a {cut}-byte prefix was accepted"
2868            );
2869        }
2870        assert_eq!(
2871            ProtectedTransfer::decode(&encoded[..encoded.len() - 1]).err(),
2872            Some(TransferError::Truncated)
2873        );
2874        let mut longer = encoded.to_vec();
2875        longer.push(0);
2876        assert_eq!(
2877            ProtectedTransfer::decode(&longer).err(),
2878            Some(TransferError::Trailing)
2879        );
2880    }
2881
2882    #[test]
2883    fn oversized_counts_lengths_and_blocks_are_refused_before_allocating() {
2884        let prefix = |clusters: u32| {
2885            let mut bytes = vec![TRANSFER_LAYOUT_VERSION, 0];
2886            bytes.extend_from_slice(&0u64.to_be_bytes());
2887            bytes.extend_from_slice(&clusters.to_be_bytes());
2888            bytes
2889        };
2890        let too_many_clusters = prefix(TRANSFER_MAX_CLUSTERS as u32 + 1);
2891        assert_eq!(
2892            ProtectedTransfer::decode(&too_many_clusters).err(),
2893            Some(TransferError::TooLarge)
2894        );
2895        let mut long_id = prefix(1);
2896        long_id.extend_from_slice(&(TRANSFER_MAX_TEXT as u32 + 1).to_be_bytes());
2897        assert_eq!(
2898            ProtectedTransfer::decode(&long_id).err(),
2899            Some(TransferError::TooLarge)
2900        );
2901        let mut many_headers = prefix(1);
2902        many_headers.extend_from_slice(&1u32.to_be_bytes());
2903        many_headers.push(b'a');
2904        many_headers
2905            .extend_from_slice(&(crate::config::HEALTH_CHECK_MAX_HEADERS as u32 + 1).to_be_bytes());
2906        assert_eq!(
2907            ProtectedTransfer::decode(&many_headers).err(),
2908            Some(TransferError::TooLarge)
2909        );
2910        let mut long_value = prefix(1);
2911        long_value.extend_from_slice(&1u32.to_be_bytes());
2912        long_value.push(b'a');
2913        long_value.extend_from_slice(&1u32.to_be_bytes());
2914        long_value.extend_from_slice(&1u32.to_be_bytes());
2915        long_value.push(b'n');
2916        long_value.extend_from_slice(
2917            &(crate::config::HEALTH_CHECK_MAX_HEADER_BYTES as u32 + 1).to_be_bytes(),
2918        );
2919        assert_eq!(
2920            ProtectedTransfer::decode(&long_value).err(),
2921            Some(TransferError::TooLarge)
2922        );
2923        let block = vec![0u8; PROTECTED_TRANSFER_MAX_BYTES + 1];
2924        assert_eq!(
2925            ProtectedTransfer::decode(&block).err(),
2926            Some(TransferError::TooLarge)
2927        );
2928    }
2929
2930    #[test]
2931    fn malformed_text_order_flags_and_values_are_refused() {
2932        let (transfer, _, _) = sample();
2933        let encoded = encode(&transfer).to_vec();
2934        // Flag byte other than 0 or 1.
2935        let mut flag = encoded.clone();
2936        flag[1] = 2;
2937        assert_eq!(
2938            ProtectedTransfer::decode(&flag).err(),
2939            Some(TransferError::Malformed)
2940        );
2941        // Layout version.
2942        let mut version = encoded.clone();
2943        version[0] = 9;
2944        assert_eq!(
2945            ProtectedTransfer::decode(&version).err(),
2946            Some(TransferError::Malformed)
2947        );
2948        // A cluster id that is not UTF-8: first id byte sits after
2949        // version, flag, key, epoch, counter and count and its own length.
2950        let mut utf8 = encoded.clone();
2951        utf8[2 + KEY_LEN + EPOCH_LEN + 8 + 4 + 4] = 0xff;
2952        assert_eq!(
2953            ProtectedTransfer::decode(&utf8).err(),
2954            Some(TransferError::Malformed)
2955        );
2956        // Clusters out of order (or duplicated): "a" then "a".
2957        let mut twice = vec![TRANSFER_LAYOUT_VERSION, 0];
2958        twice.extend_from_slice(&0u64.to_be_bytes());
2959        twice.extend_from_slice(&2u32.to_be_bytes());
2960        for _ in 0..2 {
2961            twice.extend_from_slice(&1u32.to_be_bytes());
2962            twice.push(b'a');
2963            twice.extend_from_slice(&0u32.to_be_bytes());
2964        }
2965        assert_eq!(
2966            ProtectedTransfer::decode(&twice).err(),
2967            Some(TransferError::Malformed)
2968        );
2969        // A value with a forbidden control byte.
2970        let mut control = vec![TRANSFER_LAYOUT_VERSION, 0];
2971        control.extend_from_slice(&0u64.to_be_bytes());
2972        control.extend_from_slice(&1u32.to_be_bytes());
2973        control.extend_from_slice(&1u32.to_be_bytes());
2974        control.push(b'a');
2975        control.extend_from_slice(&1u32.to_be_bytes());
2976        control.extend_from_slice(&1u32.to_be_bytes());
2977        control.push(b'n');
2978        control.extend_from_slice(&1u32.to_be_bytes());
2979        control.push(0);
2980        assert_eq!(
2981            ProtectedTransfer::decode(&control).err(),
2982            Some(TransferError::Malformed)
2983        );
2984    }
2985
2986    #[test]
2987    fn encoding_never_grows_its_buffer_and_nothing_prints_a_value_or_the_key() {
2988        let (transfer, _, _) = sample();
2989        let mut small = Zeroizing::new(Vec::with_capacity(transfer.encoded_len().unwrap() - 1));
2990        let capacity = small.capacity();
2991        assert_eq!(
2992            transfer.encode_into(&mut small).err(),
2993            Some(TransferError::BufferTooSmall)
2994        );
2995        assert_eq!((small.len(), small.capacity()), (0, capacity));
2996
2997        let rendered = format!("{transfer:?}");
2998        assert!(!rendered.contains(TRANSFER_SENTINEL), "{rendered}");
2999        assert!(!rendered.contains(&hex::encode([7u8; 32])), "{rendered}");
3000        for error in [
3001            TransferError::Pending,
3002            TransferError::KeyMissing,
3003            TransferError::Mismatch,
3004            TransferError::TooLarge,
3005            TransferError::Truncated,
3006            TransferError::Trailing,
3007            TransferError::Malformed,
3008            TransferError::BufferTooSmall,
3009        ] {
3010            assert!(!error.to_string().contains(TRANSFER_SENTINEL));
3011        }
3012    }
3013}