hyphae_engine/proof/
verify.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use std::{
4    fs::{File, Metadata, OpenOptions},
5    io::{Read, Write},
6    path::Path,
7    time::Duration,
8    time::Instant,
9};
10
11use hyphae_query::{BoundedQueryError, Record, execute_with_byte_limit};
12use hyphae_storage::{SnapshotError, load_snapshot_with_timeout};
13
14use super::{
15    MAX_RESULT_PROOF_BYTES, ProofAnchor, ProofError, ProvenOperation, ProvenResult, ResultProof,
16    VerificationLimits, VerificationReport, decode_proof, encode_proof,
17};
18use crate::decode_document;
19
20const PROOF_READ_BUFFER_BYTES: usize = 64 * 1024;
21
22/// Writes a canonical result proof to a new file and synchronizes it.
23///
24/// Existing paths are never replaced.
25///
26/// # Errors
27///
28/// Returns a proof encoding, path, create, write, or synchronization error.
29pub fn write_result_proof(path: impl AsRef<Path>, proof: &ResultProof) -> Result<(), ProofError> {
30    let encoded = encode_proof(proof)?;
31    let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
32    file.write_all(&encoded)?;
33    file.sync_all()?;
34    Ok(())
35}
36
37/// Reads and verifies one canonical result-proof file under a byte limit.
38///
39/// # Errors
40///
41/// Returns an I/O, resource-limit, framing, canonicality, checksum, or digest
42/// error.
43pub fn read_result_proof(
44    path: impl AsRef<Path>,
45    maximum_bytes: u64,
46) -> Result<ResultProof, ProofError> {
47    let mut no_deadline = || Ok(());
48    decode_proof(&read_result_proof_bytes(
49        path,
50        maximum_bytes,
51        &mut no_deadline,
52    )?)
53}
54
55/// Verifies a result proof completely offline against a trusted anchor and
56/// canonical snapshot witness.
57///
58/// # Errors
59///
60/// Returns an error for any proof or snapshot corruption, wrong anchor,
61/// resource exhaustion, document failure, timeout, or replay mismatch. No
62/// partial result is accepted.
63pub fn verify_result_proof(
64    proof_path: impl AsRef<Path>,
65    snapshot_path: impl AsRef<Path>,
66    expected_anchor_digest: [u8; 32],
67    limits: &VerificationLimits,
68) -> Result<VerificationReport, ProofError> {
69    let started = Instant::now();
70    let mut check_read_deadline = || check_timeout(started, limits);
71    let proof = decode_proof(&read_result_proof_bytes(
72        proof_path,
73        limits.proof_bytes,
74        &mut check_read_deadline,
75    )?)?;
76    check_timeout(started, limits)?;
77
78    let anchor_digest = proof.anchor_digest();
79    if anchor_digest != expected_anchor_digest {
80        return Err(ProofError::AnchorMismatch);
81    }
82
83    let snapshot = match load_snapshot_with_timeout(
84        snapshot_path,
85        &limits.snapshot,
86        remaining_timeout(started, limits)?,
87    ) {
88        Err(error) if error.is_timeout() => return Err(ProofError::TimedOut),
89        Err(error) => return Err(error.into()),
90        Ok(snapshot) => snapshot,
91    };
92    check_timeout(started, limits)?;
93    if ProofAnchor::from_snapshot(&snapshot.info) != *proof.anchor() {
94        return Err(ProofError::SnapshotAnchorMismatch);
95    }
96
97    let mut records = Vec::with_capacity(snapshot.entries.len());
98    for entry in snapshot.entries {
99        check_timeout(started, limits)?;
100        records.push(Record {
101            key: entry.key,
102            value: decode_document(&entry.value)?,
103        });
104    }
105
106    let verified_result = match (proof.operation(), proof.result()) {
107        (ProvenOperation::Get { key }, ProvenResult::Get(expected)) => {
108            let actual = records
109                .binary_search_by(|record| record.key.as_slice().cmp(key))
110                .ok()
111                .map(|index| records[index].clone());
112            if &actual != expected {
113                return Err(ProofError::ReexecutionMismatch);
114            }
115            ProvenResult::Get(actual)
116        }
117        (ProvenOperation::Query(query), ProvenResult::Query(expected)) => {
118            let query_limits = hyphae_query::ExecutionLimits {
119                timeout: remaining_timeout(started, limits)?.min(limits.query.timeout),
120                ..limits.query.clone()
121            };
122            let actual = execute_with_byte_limit(
123                &[records.as_slice()],
124                query,
125                &query_limits,
126                limits.snapshot.decoded_bytes,
127            )
128            .map_err(|source| match source {
129                BoundedQueryError::Query(source) => ProofError::from(source),
130                BoundedQueryError::RecordDocument(source) => ProofError::from(source),
131                BoundedQueryError::ScannedByteBudgetExceeded { maximum } => {
132                    ProofError::from(SnapshotError::DecodedBytesLimitExceeded { maximum })
133                }
134            })?;
135            if &actual != expected {
136                return Err(ProofError::ReexecutionMismatch);
137            }
138            ProvenResult::Query(actual)
139        }
140        _ => return Err(ProofError::OperationResultMismatch),
141    };
142    check_timeout(started, limits)?;
143
144    Ok(VerificationReport {
145        anchor: proof.anchor().clone(),
146        anchor_digest,
147        proof_digest: proof.proof_digest(),
148        result: verified_result,
149    })
150}
151
152fn read_result_proof_bytes(
153    path: impl AsRef<Path>,
154    maximum_bytes: u64,
155    check_deadline: &mut impl FnMut() -> Result<(), ProofError>,
156) -> Result<Vec<u8>, ProofError> {
157    check_deadline()?;
158    let path = path.as_ref();
159    let path_metadata = std::fs::metadata(path)?;
160    check_deadline()?;
161    ensure_regular_proof_file(&path_metadata)?;
162
163    let file = File::open(path)?;
164    check_deadline()?;
165    let initial_metadata = file.metadata()?;
166    check_deadline()?;
167    ensure_regular_proof_file(&initial_metadata)?;
168
169    read_open_result_proof(
170        file,
171        &initial_metadata,
172        maximum_bytes.min(MAX_RESULT_PROOF_BYTES),
173        check_deadline,
174    )
175}
176
177fn read_open_result_proof(
178    mut file: File,
179    initial_metadata: &Metadata,
180    maximum_bytes: u64,
181    check_deadline: &mut impl FnMut() -> Result<(), ProofError>,
182) -> Result<Vec<u8>, ProofError> {
183    let initial_length = initial_metadata.len();
184    if initial_length > maximum_bytes {
185        return Err(ProofError::ProofLimitExceeded {
186            actual: initial_length,
187            maximum: maximum_bytes,
188        });
189    }
190    let capacity = usize::try_from(initial_length).map_err(|_| ProofError::LengthOverflow)?;
191    let mut encoded = Vec::with_capacity(capacity);
192    let mut remaining = maximum_bytes
193        .checked_add(1)
194        .ok_or(ProofError::LengthOverflow)?;
195    let mut buffer = vec![0_u8; PROOF_READ_BUFFER_BYTES];
196    while remaining > 0 {
197        check_deadline()?;
198        let read_length = usize::try_from(remaining.min(PROOF_READ_BUFFER_BYTES as u64))
199            .map_err(|_| ProofError::LengthOverflow)?;
200        let read = file.read(&mut buffer[..read_length])?;
201        check_deadline()?;
202        if read == 0 {
203            break;
204        }
205        encoded.extend_from_slice(&buffer[..read]);
206        remaining = remaining
207            .checked_sub(u64::try_from(read).map_err(|_| ProofError::LengthOverflow)?)
208            .ok_or(ProofError::LengthOverflow)?;
209    }
210
211    let final_metadata = file.metadata()?;
212    check_deadline()?;
213    ensure_regular_proof_file(&final_metadata)?;
214    let actual = u64::try_from(encoded.len()).map_err(|_| ProofError::LengthOverflow)?;
215    let observed = actual.max(final_metadata.len());
216    if observed > maximum_bytes {
217        return Err(ProofError::ProofLimitExceeded {
218            actual: observed,
219            maximum: maximum_bytes,
220        });
221    }
222    if actual != initial_length || final_metadata.len() != initial_length {
223        return Err(ProofError::Invalid {
224            reason: "proof changed while being read",
225        });
226    }
227    Ok(encoded)
228}
229
230fn ensure_regular_proof_file(metadata: &Metadata) -> Result<(), ProofError> {
231    if metadata.is_file() {
232        Ok(())
233    } else {
234        Err(ProofError::Invalid {
235            reason: "proof path is not a regular file",
236        })
237    }
238}
239
240fn remaining_timeout(
241    started: Instant,
242    limits: &VerificationLimits,
243) -> Result<Duration, ProofError> {
244    let remaining = limits
245        .timeout
246        .checked_sub(started.elapsed())
247        .ok_or(ProofError::TimedOut)?;
248    if remaining.is_zero() {
249        Err(ProofError::TimedOut)
250    } else {
251        Ok(remaining)
252    }
253}
254
255fn check_timeout(started: Instant, limits: &VerificationLimits) -> Result<(), ProofError> {
256    if started.elapsed() >= limits.timeout {
257        Err(ProofError::TimedOut)
258    } else {
259        Ok(())
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use std::{collections::BTreeMap, error::Error, fs, io::Write as _, path::PathBuf};
266
267    use hyphae_query::{ExecutionLimits, Filter, Query, Record, Value};
268    use uuid::Uuid;
269
270    use super::{
271        PROOF_READ_BUFFER_BYTES, VerificationLimits, read_open_result_proof, read_result_proof,
272        verify_result_proof, write_result_proof,
273    };
274    use crate::{HyphaeEngine, MAX_RESULT_PROOF_BYTES, ProofError, ProvenResult};
275
276    struct TestDirectory {
277        path: PathBuf,
278    }
279
280    impl TestDirectory {
281        fn create() -> Result<Self, Box<dyn Error>> {
282            let path = std::env::temp_dir()
283                .join(format!("hyphae-proof-rehashed-tamper-{}", Uuid::now_v7()));
284            fs::create_dir_all(&path)?;
285            Ok(Self { path })
286        }
287    }
288
289    impl Drop for TestDirectory {
290        fn drop(&mut self) {
291            let _ignored = fs::remove_dir_all(&self.path);
292        }
293    }
294
295    #[test]
296    fn result_proof_reader_enforces_the_canonical_hard_limit() -> Result<(), Box<dyn Error>> {
297        let temporary = TestDirectory::create()?;
298        let proof_path = temporary.path.join("oversized.hyproof");
299        let file = fs::File::create(&proof_path)?;
300        file.set_len(MAX_RESULT_PROOF_BYTES + 1)?;
301        drop(file);
302
303        assert!(matches!(
304            read_result_proof(&proof_path, u64::MAX),
305            Err(ProofError::ProofLimitExceeded {
306                actual,
307                maximum: MAX_RESULT_PROOF_BYTES,
308            }) if actual == MAX_RESULT_PROOF_BYTES + 1
309        ));
310        Ok(())
311    }
312
313    #[test]
314    fn result_proof_reader_detects_same_handle_growth() -> Result<(), Box<dyn Error>> {
315        let temporary = TestDirectory::create()?;
316        let proof_path = temporary.path.join("growing.hyproof");
317        fs::write(&proof_path, b"initial")?;
318        let file = fs::File::open(&proof_path)?;
319        let initial_metadata = file.metadata()?;
320        let mut writer = fs::OpenOptions::new().append(true).open(&proof_path)?;
321        writer.write_all(b"-growth")?;
322        writer.sync_all()?;
323        drop(writer);
324
325        let mut no_deadline = || Ok(());
326        assert!(matches!(
327            read_open_result_proof(file, &initial_metadata, 1024, &mut no_deadline),
328            Err(ProofError::Invalid {
329                reason: "proof changed while being read",
330            })
331        ));
332        Ok(())
333    }
334
335    #[test]
336    fn result_proof_reader_checks_deadline_between_chunks() -> Result<(), Box<dyn Error>> {
337        let temporary = TestDirectory::create()?;
338        let proof_path = temporary.path.join("timed.hyproof");
339        fs::write(&proof_path, vec![0_u8; PROOF_READ_BUFFER_BYTES * 2])?;
340        let file = fs::File::open(&proof_path)?;
341        let initial_metadata = file.metadata()?;
342        let mut checks = 0_u8;
343        let mut deadline = || {
344            checks += 1;
345            if checks == 2 {
346                Err(ProofError::TimedOut)
347            } else {
348                Ok(())
349            }
350        };
351
352        assert!(matches!(
353            read_open_result_proof(
354                file,
355                &initial_metadata,
356                MAX_RESULT_PROOF_BYTES,
357                &mut deadline,
358            ),
359            Err(ProofError::TimedOut)
360        ));
361        assert_eq!(checks, 2);
362        Ok(())
363    }
364
365    #[test]
366    fn self_consistently_rehashed_result_edits_are_rejected() -> Result<(), Box<dyn Error>> {
367        let temporary = TestDirectory::create()?;
368        let mut opened = HyphaeEngine::open(temporary.path.join("data"))?;
369        opened.engine.put_records(
370            Uuid::now_v7(),
371            &[record(b"a", 1), record(b"b", 2), record(b"c", 3)],
372        )?;
373        let artifact = opened.engine.query_with_proof(
374            &Query {
375                filter: Filter::MatchAll,
376                sort: Vec::new(),
377                cursor: None,
378                limit: 3,
379                aggregation: None,
380            },
381            &ExecutionLimits::default(),
382        )?;
383
384        for (name, mutation) in [
385            ("delete", 1_u8),
386            ("insert", 2_u8),
387            ("reorder", 3_u8),
388            ("edit", 4_u8),
389        ] {
390            let mut forged = artifact.proof.clone();
391            let ProvenResult::Query(result) = &mut forged.result else {
392                return Err(ProofError::OperationResultMismatch.into());
393            };
394            match mutation {
395                1 => {
396                    result.rows.remove(0);
397                }
398                2 => result.rows.push(result.rows[0].clone()),
399                3 => result.rows.swap(0, 1),
400                4 => result.rows[0].value = Value::Integer(999),
401                _ => return Err(ProofError::OperationResultMismatch.into()),
402            }
403            let proof_path = temporary.path.join(format!("{name}.hyproof"));
404            write_result_proof(&proof_path, &forged)?;
405            assert!(matches!(
406                verify_result_proof(
407                    &proof_path,
408                    &artifact.snapshot.path,
409                    artifact.proof.anchor_digest(),
410                    &VerificationLimits::default(),
411                ),
412                Err(ProofError::ReexecutionMismatch)
413            ));
414        }
415        Ok(())
416    }
417
418    fn record(key: &[u8], score: i64) -> Record {
419        Record::new(
420            key,
421            Value::Object(BTreeMap::from([(
422                "score".to_owned(),
423                Value::Integer(score),
424            )])),
425        )
426    }
427}
hyphae_engine/proof/verify.rs

hyphae_engine/proof/
verify.rs
