laser_wire/filter/
headers.rs

1use super::eval::{HeaderNeed, HeaderRef, HeaderValueRef};
2use crate::headers::CONTENT_TYPE;
3
4/// Decode native Iggy header TLVs without changing unknown value kinds. An
5/// entry under a non-text key is stepped over undecoded. A malformed value
6/// under a text key faults the whole record. Entries keep their order, and
7/// the evaluator reads the last entry of a repeated key.
8pub fn decode(bytes: &[u8]) -> Option<Vec<HeaderRef<'_>>> {
9    let mut headers = Vec::new();
10    decode_into(bytes, &mut headers)?;
11    Some(headers)
12}
13
14/// [`decode_for`] into the caller's buffer, so a scan reuses one allocation
15/// across records. The buffer is cleared first. On `Some` it holds the
16/// decoded entries, on `None` (a malformed block) it holds nothing.
17pub fn decode_for_into<'a>(
18    need: HeaderNeed,
19    bytes: &'a [u8],
20    headers: &mut Vec<HeaderRef<'a>>,
21) -> Option<()> {
22    headers.clear();
23    match need {
24        HeaderNeed::None => Some(()),
25        HeaderNeed::ContentType => {
26            headers.extend(content_type_one(bytes)?);
27            Some(())
28        }
29        HeaderNeed::All => decode_into(bytes, headers),
30    }
31}
32
33fn decode_into<'a>(mut bytes: &'a [u8], headers: &mut Vec<HeaderRef<'a>>) -> Option<()> {
34    while !bytes.is_empty() {
35        let (key_kind, key) = field(&mut bytes)?;
36        let (value_kind, value) = field(&mut bytes)?;
37        if key_kind != 2 {
38            continue;
39        }
40        let key = std::str::from_utf8(key).ok()?;
41        let value = scalar(value_kind, value)?;
42        headers.push(HeaderRef { key, value });
43    }
44    Some(())
45}
46
47/// The headers a filter needs before evaluating, decoded from the raw block:
48/// nothing, only `agdx.ct`, or every entry. `None` when the block's structure
49/// does not parse, which the caller reports as a malformed record.
50pub enum DecodedHeaders<'a> {
51    Empty,
52    ContentType(Option<HeaderRef<'a>>),
53    All(Vec<HeaderRef<'a>>),
54}
55
56impl<'a> std::ops::Deref for DecodedHeaders<'a> {
57    type Target = [HeaderRef<'a>];
58    fn deref(&self) -> &Self::Target {
59        match self {
60            Self::Empty => &[],
61            Self::ContentType(header) => header.as_slice(),
62            Self::All(headers) => headers,
63        }
64    }
65}
66
67pub fn decode_for(need: HeaderNeed, bytes: &[u8]) -> Option<DecodedHeaders<'_>> {
68    match need {
69        HeaderNeed::None => Some(DecodedHeaders::Empty),
70        HeaderNeed::ContentType => content_type_one(bytes).map(DecodedHeaders::ContentType),
71        HeaderNeed::All => decode(bytes).map(DecodedHeaders::All),
72    }
73}
74
75/// Only the `agdx.ct` header, for a filter that reads no other header. Every
76/// entry is stepped over, so a block whose structure does not parse still
77/// faults the record, but no other key or value is decoded or checked.
78pub fn content_type(bytes: &[u8]) -> Option<Vec<HeaderRef<'static>>> {
79    content_type_one(bytes).map(|header| header.into_iter().collect())
80}
81
82fn content_type_one(mut bytes: &[u8]) -> Option<Option<HeaderRef<'static>>> {
83    let mut found = None;
84    while !bytes.is_empty() {
85        let (key_kind, key) = field(&mut bytes)?;
86        let (value_kind, value) = field(&mut bytes)?;
87        if key_kind == 2 && key == CONTENT_TYPE.as_bytes() {
88            // Only an integer names a codec. Any other value declares nothing.
89            found = match scalar(value_kind, value)? {
90                HeaderValueRef::Uint(code) => Some(HeaderValueRef::Uint(code)),
91                HeaderValueRef::Int(code) => Some(HeaderValueRef::Int(code)),
92                _ => None,
93            }
94            .map(|value| HeaderRef {
95                key: CONTENT_TYPE,
96                value,
97            });
98        }
99    }
100    Some(found)
101}
102
103fn field<'a>(bytes: &mut &'a [u8]) -> Option<(u8, &'a [u8])> {
104    let kind = *bytes.first()?;
105    let length = u32::from_le_bytes(bytes.get(1..5)?.try_into().ok()?) as usize;
106    if kind == 0 || !(1..=255).contains(&length) {
107        return None;
108    }
109    let value = bytes.get(5..5 + length)?;
110    *bytes = &bytes[5 + length..];
111    Some((kind, value))
112}
113
114fn scalar(kind: u8, value: &[u8]) -> Option<HeaderValueRef<'_>> {
115    Some(match kind {
116        2 => HeaderValueRef::String(std::str::from_utf8(value).ok()?),
117        3 => match value {
118            [0] => HeaderValueRef::Bool(false),
119            [1] => HeaderValueRef::Bool(true),
120            _ => return None,
121        },
122        4 => HeaderValueRef::Int(i64::from(i8::from_le_bytes(value.try_into().ok()?))),
123        5 => HeaderValueRef::Int(i64::from(i16::from_le_bytes(value.try_into().ok()?))),
124        6 => HeaderValueRef::Int(i64::from(i32::from_le_bytes(value.try_into().ok()?))),
125        7 => HeaderValueRef::Int(i64::from_le_bytes(value.try_into().ok()?)),
126        // 128-bit integers compare as raw bytes, but only at their width.
127        8 | 13 if value.len() != 16 => return None,
128        9 => HeaderValueRef::Uint(u64::from(u8::from_le_bytes(value.try_into().ok()?))),
129        10 => HeaderValueRef::Uint(u64::from(u16::from_le_bytes(value.try_into().ok()?))),
130        11 => HeaderValueRef::Uint(u64::from(u32::from_le_bytes(value.try_into().ok()?))),
131        12 => HeaderValueRef::Uint(u64::from_le_bytes(value.try_into().ok()?)),
132        14 => HeaderValueRef::Float(f64::from(f32::from_le_bytes(value.try_into().ok()?))),
133        15 => HeaderValueRef::Float(f64::from_le_bytes(value.try_into().ok()?)),
134        _ => HeaderValueRef::Raw(value),
135    })
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    fn entry(key: &str, kind: u8, value: &[u8]) -> Vec<u8> {
143        let mut bytes = vec![2];
144        bytes.extend_from_slice(&(key.len() as u32).to_le_bytes());
145        bytes.extend_from_slice(key.as_bytes());
146        bytes.push(kind);
147        bytes.extend_from_slice(&(value.len() as u32).to_le_bytes());
148        bytes.extend_from_slice(value);
149        bytes
150    }
151
152    #[test]
153    fn given_unknown_and_known_headers_when_decoded_then_should_preserve_both() {
154        let mut bytes = entry("future", 99, b"opaque");
155        bytes.extend(entry("priority", 9, &[2]));
156        let headers = decode(&bytes).expect("valid headers");
157        assert_eq!(headers[0].value, HeaderValueRef::Raw(b"opaque"));
158        assert_eq!(headers[1].value, HeaderValueRef::Uint(2));
159    }
160
161    #[test]
162    fn given_many_headers_when_only_the_content_type_is_read_then_should_decode_just_that_one() {
163        let mut bytes = entry("frostline.unit", 2, b"reefer");
164        bytes.extend(entry("agdx.ct", 9, &[1]));
165        bytes.extend(entry("agdx.ct", 9, &[3]));
166        bytes.extend(entry("odd", 3, &[2]));
167        let headers = content_type(&bytes).expect("the structure parses");
168        assert_eq!(headers.len(), 1);
169        assert_eq!(
170            (headers[0].key, headers[0].value),
171            ("agdx.ct", HeaderValueRef::Uint(3))
172        );
173        assert!(
174            content_type(&entry("frostline.unit", 2, b"reefer"))
175                .expect("parses")
176                .is_empty()
177        );
178        let mut truncated = entry("agdx.ct", 9, &[1]);
179        truncated.truncate(truncated.len() - 1);
180        assert!(content_type(&truncated).is_none());
181        assert!(content_type(&entry("agdx.ct", 9, &[1, 2])).is_none());
182    }
183
184    // A scan decodes one record's headers into the same buffer each time, so
185    // the capacity is kept, every need fills it like the fresh decode would,
186    // and a malformed block leaves nothing of the previous record behind.
187    #[test]
188    fn given_a_reused_buffer_when_each_need_is_decoded_then_should_match_the_fresh_decode() {
189        let mut block = entry("priority", 9, &[2]);
190        block.extend(entry("agdx.ct", 9, &[3]));
191        block.extend(entry("future", 99, b"opaque"));
192        let mut buffer = Vec::new();
193        decode_for_into(HeaderNeed::All, &block, &mut buffer).expect("parses");
194        let fresh = decode(&block).expect("parses");
195        assert_eq!(buffer.len(), fresh.len());
196        assert!(
197            buffer
198                .iter()
199                .zip(&fresh)
200                .all(|(reused, fresh)| reused.key == fresh.key && reused.value == fresh.value)
201        );
202        let capacity = buffer.capacity();
203        decode_for_into(HeaderNeed::ContentType, &block, &mut buffer).expect("parses");
204        assert_eq!(buffer.len(), 1);
205        assert_eq!(buffer[0].value, HeaderValueRef::Uint(3));
206        decode_for_into(HeaderNeed::None, &block, &mut buffer).expect("parses");
207        assert!(buffer.is_empty());
208        assert_eq!(
209            buffer.capacity(),
210            capacity,
211            "the buffer keeps its allocation"
212        );
213        let mut truncated = entry("agdx.ct", 9, &[1]);
214        truncated.truncate(truncated.len() - 1);
215        decode_for_into(HeaderNeed::All, &block, &mut buffer).expect("parses");
216        assert!(decode_for_into(HeaderNeed::All, &truncated, &mut buffer).is_none());
217        assert!(buffer.is_empty(), "malformed input leaves nothing behind");
218    }
219
220    #[test]
221    fn given_malformed_known_headers_when_decoded_then_should_fault_the_whole_block() {
222        for (kind, value) in [
223            (11, vec![1]),
224            (3, vec![2]),
225            (8, vec![0; 15]),
226            (2, vec![255]),
227        ] {
228            let mut bytes = entry("priority", 9, &[2]);
229            bytes.extend(entry("broken", kind, &value));
230            assert!(decode(&bytes).is_none());
231        }
232    }
233}
laser_wire/filter/headers.rs

laser_wire/filter/
headers.rs
