lace_sans_io/records/
validator.rs

1//! Incremental Lace-010 validation of canonical record bytes in arbitrary
2//! fragments. Only the canonical non-payload prefix is buffered; Blob data
3//! passes straight through the active BLAKE3 contexts.
4
5use std::fmt;
6
7use b64a::{b64a_decode_into, b64a_encoded_len};
8use schnorr_blake3::{Mark, VerificationKey};
9
10use crate::hash::{RecordHash, RecordKind};
11use crate::typed_text::ByValue;
12use crate::util::{
13    MAX_BLOB_SIZE, MAX_FIELD_LINE_BYTES_EXCLUDING_LF, MAX_PLEX_FIELDS, MAX_RECORD_METADATA_BYTES,
14    RECORD_GLYPH, RECORD_GLYPH_LINE_PREFIX, validate_application_field_line, validate_field_line,
15};
16
17/// Error type for record validation failures.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum RecordError {
20    InvalidFormat(String),
21    HashMismatch { expected: String, computed: String },
22    MarkInvalid,
23    UnknownType(String),
24    MissingField(String),
25}
26
27impl fmt::Display for RecordError {
28    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
29        match self {
30            Self::InvalidFormat(message) => write!(formatter, "invalid record format: {message}"),
31            Self::HashMismatch { expected, computed } => {
32                write!(
33                    formatter,
34                    "hash mismatch: expected {expected}, got {computed}"
35                )
36            }
37            Self::MarkInvalid => formatter.write_str("mark verification failed"),
38            Self::UnknownType(kind) => write!(formatter, "unknown record type: {kind}"),
39            Self::MissingField(field) => write!(formatter, "missing field: {field}"),
40        }
41    }
42}
43
44impl std::error::Error for RecordError {}
45
46fn invalid(message: impl Into<String>) -> RecordError {
47    RecordError::InvalidFormat(message.into())
48}
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51enum Phase {
52    /// A record glyph line; `Some` names the only kind allowed here.
53    Glyph(Option<RecordKind>),
54    By,
55    Mark,
56    PlexFields,
57    DataLength,
58    Blank,
59    Body,
60}
61
62/// Validates exactly one canonical record supplied in arbitrary fragments.
63///
64/// `feed` returns how many bytes it consumed. Once the record is complete it
65/// consumes nothing more, so bytes that follow the record stay with the caller.
66/// Errors are terminal: later input cannot turn the validator into a record.
67pub struct RecordValidator {
68    prefix: Vec<u8>,
69    line_start: usize,
70    phase: Phase,
71    /// Declared glyph hashes, outermost first, each with a digest context fed
72    /// every byte after its glyph line. At most three; kept on the heap so the
73    /// validator stays small on constrained stacks.
74    digests: Vec<(RecordHash, blake3::Hasher)>,
75    previous_name: Option<(usize, usize)>,
76    field_count: usize,
77    by: Option<VerificationKey>,
78    mark: Option<Mark>,
79    data_length: usize,
80    data_received: usize,
81    complete: bool,
82    failure: Option<RecordError>,
83}
84
85impl Default for RecordValidator {
86    fn default() -> Self {
87        Self::new()
88    }
89}
90
91impl fmt::Debug for RecordValidator {
92    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
93        formatter
94            .debug_struct("RecordValidator")
95            .field("phase", &self.phase)
96            .field("prefix_len", &self.prefix.len())
97            .field("data_received", &self.data_received)
98            .field("complete", &self.complete)
99            .field("failure", &self.failure)
100            .finish()
101    }
102}
103
104impl RecordValidator {
105    pub fn new() -> Self {
106        Self {
107            prefix: Vec::new(),
108            line_start: 0,
109            phase: Phase::Glyph(None),
110            digests: Vec::new(),
111            previous_name: None,
112            field_count: 0,
113            by: None,
114            mark: None,
115            data_length: 0,
116            data_received: 0,
117            complete: false,
118            failure: None,
119        }
120    }
121
122    /// Consume a prefix of `input` and return its length. Stops at the
123    /// record's end; the remaining input is not part of this record.
124    pub fn feed(&mut self, input: &[u8]) -> Result<usize, RecordError> {
125        if let Some(failure) = &self.failure {
126            return Err(failure.clone());
127        }
128        let result = self.consume(input);
129        if let Err(failure) = &result {
130            self.failure = Some(failure.clone());
131        }
132        result
133    }
134
135    /// The exact canonical length once the complete prefix has been read.
136    /// It is not evidence that the data, hashes or Mark are valid.
137    pub fn canonical_len(&self) -> Option<usize> {
138        (self.phase == Phase::Body).then(|| self.prefix.len() + self.data_length)
139    }
140
141    /// The canonical bytes before the Blob data received so far: the complete
142    /// prefix once [`canonical_len`](Self::canonical_len) is known.
143    pub fn prefix(&self) -> &[u8] {
144        &self.prefix
145    }
146
147    /// Whether the record has been completely received and verified.
148    pub fn is_complete(&self) -> bool {
149        self.complete
150    }
151
152    /// The verified header of a complete record.
153    pub fn finish(mut self) -> Result<ValidatedHeader, RecordError> {
154        self.take_header()
155    }
156
157    /// The verified header of a complete record, taking its prefix; only
158    /// [`reset`](Self::reset) may follow.
159    pub(crate) fn take_header(&mut self) -> Result<ValidatedHeader, RecordError> {
160        if let Some(failure) = &self.failure {
161            return Err(failure.clone());
162        }
163        if !self.complete {
164            return Err(invalid("truncated record"));
165        }
166        Ok(ValidatedHeader {
167            hash: self.digests[0].0,
168            canonical_len: self.prefix.len() + self.data_length,
169            prefix: std::mem::take(&mut self.prefix),
170        })
171    }
172
173    /// Start over for another record, keeping the digest contexts' allocation.
174    pub(crate) fn reset(&mut self) {
175        let mut digests = std::mem::take(&mut self.digests);
176        digests.clear();
177        *self = Self {
178            digests,
179            ..Self::new()
180        };
181    }
182
183    fn consume(&mut self, input: &[u8]) -> Result<usize, RecordError> {
184        let mut consumed = 0;
185        while !self.complete && consumed < input.len() {
186            let rest = &input[consumed..];
187            let count = if self.phase == Phase::Body {
188                let count = (self.data_length - self.data_received).min(rest.len());
189                self.hash(&rest[..count]);
190                self.data_received += count;
191                count
192            } else {
193                self.consume_line_bytes(rest)?
194            };
195            consumed += count;
196            if self.phase == Phase::Body && self.data_received == self.data_length {
197                self.verify()?;
198                self.complete = true;
199            }
200        }
201        Ok(consumed)
202    }
203
204    /// Consume header bytes up to and including the next LF.
205    fn consume_line_bytes(&mut self, input: &[u8]) -> Result<usize, RecordError> {
206        let (count, has_lf) = match memchr::memchr(b'\n', input) {
207            Some(lf) => (lf + 1, true),
208            None => (input.len(), false),
209        };
210        let line_len = self.prefix.len() - self.line_start + count - usize::from(has_lf);
211        if line_len > MAX_FIELD_LINE_BYTES_EXCLUDING_LF {
212            return Err(invalid("field line too long"));
213        }
214        let len = self.prefix.len() + count;
215        if len > MAX_RECORD_METADATA_BYTES {
216            return Err(invalid("record metadata too long"));
217        }
218        // Grow by doubling, never past the largest prefix.
219        if len > self.prefix.capacity() {
220            let capacity = (self.prefix.capacity() * 2).clamp(len, MAX_RECORD_METADATA_BYTES);
221            self.prefix.reserve_exact(capacity - self.prefix.len());
222        }
223        let bytes = &input[..count];
224        self.hash(bytes);
225        self.prefix.extend_from_slice(bytes);
226        if has_lf {
227            self.finish_line()?;
228            self.line_start = self.prefix.len();
229        }
230        Ok(count)
231    }
232
233    fn hash(&mut self, bytes: &[u8]) {
234        for (_, context) in &mut self.digests {
235            context.update(bytes);
236        }
237    }
238
239    fn finish_line(&mut self) -> Result<(), RecordError> {
240        let start = self.line_start;
241        let end = self.prefix.len() - 1;
242        let line = std::str::from_utf8(&self.prefix[start..end])
243            .map_err(|_| invalid("record header line is not UTF-8"))?;
244        match self.phase {
245            Phase::Glyph(expected) => {
246                let hash = parse_glyph(line)?;
247                if expected.is_some_and(|kind| kind != hash.kind()) {
248                    return Err(invalid("a Seal must embed a Plex"));
249                }
250                self.phase = match hash.kind() {
251                    RecordKind::Blob => Phase::DataLength,
252                    RecordKind::Plex => Phase::PlexFields,
253                    RecordKind::Seal => Phase::By,
254                };
255                self.open(hash);
256            }
257            Phase::By => {
258                let value = structural_value(line, "By")?;
259                let (_, key) = ByValue::parse_with_key(value)
260                    .map_err(|error| invalid(format!("invalid By value: {error}")))?;
261                self.by = Some(key);
262                self.phase = Phase::Mark;
263            }
264            Phase::Mark => {
265                self.mark = Some(parse_mark(structural_value(line, "Mark")?)?);
266                self.phase = Phase::Glyph(Some(RecordKind::Plex));
267            }
268            Phase::PlexFields => {
269                if line.as_bytes().starts_with(RECORD_GLYPH_LINE_PREFIX) {
270                    let hash = parse_glyph(line)?;
271                    if hash.kind() != RecordKind::Blob {
272                        return Err(invalid("a Plex must embed a Blob"));
273                    }
274                    if self.field_count == 0 {
275                        return Err(invalid("Plex has no application fields"));
276                    }
277                    self.phase = Phase::DataLength;
278                    self.open(hash);
279                } else {
280                    let (name, _) = validate_application_field_line(line).map_err(|error| {
281                        invalid(format!("invalid Plex application field: {error}"))
282                    })?;
283                    if self.field_count == MAX_PLEX_FIELDS {
284                        return Err(invalid("too many Plex fields"));
285                    }
286                    if self
287                        .previous_name
288                        .is_some_and(|(previous_start, previous_end)| {
289                            name.as_bytes() < &self.prefix[previous_start..previous_end]
290                        })
291                    {
292                        return Err(invalid("Plex fields are out of order"));
293                    }
294                    self.previous_name = Some((start, start + name.len()));
295                    self.field_count += 1;
296                }
297            }
298            Phase::DataLength => {
299                let value = structural_value(line, "Data-Length")?;
300                if !value.bytes().all(|byte| byte.is_ascii_digit())
301                    || (value.len() > 1 && value.starts_with('0'))
302                {
303                    return Err(invalid("invalid Data-Length"));
304                }
305                self.data_length = value
306                    .parse::<usize>()
307                    .ok()
308                    .filter(|&length| length <= MAX_BLOB_SIZE)
309                    .ok_or_else(|| invalid("Blob data exceeds maximum"))?;
310                self.phase = Phase::Blank;
311            }
312            Phase::Blank => {
313                if !line.is_empty() {
314                    return Err(invalid("missing Blob blank line"));
315                }
316                self.phase = Phase::Body;
317            }
318            Phase::Body => unreachable!("body bytes are not header lines"),
319        }
320        Ok(())
321    }
322
323    /// Record a glyph and start its digest after the glyph line.
324    fn open(&mut self, hash: RecordHash) {
325        if self.digests.is_empty() {
326            self.digests.reserve_exact(3);
327        }
328        self.digests.push((hash, blake3::Hasher::new()));
329    }
330
331    fn verify(&mut self) -> Result<(), RecordError> {
332        for (expected, context) in &self.digests {
333            let computed = RecordHash::from_digest(expected.kind(), context.finalize().into());
334            if computed != *expected {
335                return Err(RecordError::HashMismatch {
336                    expected: expected.to_string(),
337                    computed: computed.to_string(),
338                });
339            }
340        }
341        if let (Some(key), Some(mark)) = (self.by, self.mark) {
342            let plex = self.digests[1].0;
343            if !key.verify_digest(plex.digest(), mark) {
344                return Err(RecordError::MarkInvalid);
345            }
346        }
347        Ok(())
348    }
349}
350
351/// The verified header of one complete record: outer hash, canonical length
352/// and the canonical non-payload prefix, whose length is the data offset.
353#[derive(Debug, Clone, PartialEq, Eq)]
354pub struct ValidatedHeader {
355    hash: RecordHash,
356    canonical_len: usize,
357    prefix: Vec<u8>,
358}
359
360impl ValidatedHeader {
361    pub fn hash(&self) -> RecordHash {
362        self.hash
363    }
364
365    pub fn canonical_len(&self) -> usize {
366        self.canonical_len
367    }
368
369    /// The canonical bytes before the Blob data.
370    pub fn prefix(&self) -> &[u8] {
371        &self.prefix
372    }
373}
374
375fn parse_glyph(line: &str) -> Result<RecordHash, RecordError> {
376    let (name, value) = validate_field_line(line)
377        .map_err(|error| invalid(format!("invalid record glyph line: {error}")))?;
378    if name != RECORD_GLYPH {
379        return Err(invalid("invalid record glyph name"));
380    }
381    RecordHash::parse(value).map_err(|error| invalid(format!("invalid record hash: {error}")))
382}
383
384fn structural_value<'a>(line: &'a str, expected: &str) -> Result<&'a str, RecordError> {
385    let (name, value) = validate_field_line(line)
386        .map_err(|error| invalid(format!("invalid {expected} line: {error}")))?;
387    if name != expected {
388        return Err(RecordError::MissingField(expected.into()));
389    }
390    Ok(value)
391}
392
393fn parse_mark(raw: &str) -> Result<Mark, RecordError> {
394    if raw.len() != b64a_encoded_len(64) {
395        return Err(invalid(format!(
396            "Mark must encode 64 bytes, got length {}",
397            raw.len()
398        )));
399    }
400    let mut bytes = [0u8; 64];
401    b64a_decode_into(raw, &mut bytes)
402        .map_err(|error| invalid(format!("invalid Mark B64A: {error}")))?;
403    Ok(Mark::from_bytes(bytes))
404}
405
406#[cfg(test)]
407#[path = "validator_tests.rs"]
408mod tests;
lace_sans_io/records/validator.rs

lace_sans_io/records/
validator.rs
