pardarsh_core/
interchange.rs

1//! Canonical serialization, content hashing, export and import.
2//!
3//! The interchange format is a JSON [`Bundle`]: records with their full (or
4//! viewer-filtered) histories. Hashes are SHA-256 over the *canonical JSON*
5//! form: UTF-8, object keys sorted by code point, no insignificant
6//! whitespace, numbers and strings as produced by `serde_json`.
7//!
8//! Imported bundles are untrusted: size limits, format/version checks, hash
9//! verification, origin policy, history replay and full validation are all
10//! applied before anything is written.
11
12use std::collections::BTreeSet;
13
14use serde::{Deserialize, Serialize};
15use serde_json::Value;
16
17use crate::access::{Viewer, view_history, view_record};
18use crate::error::{Error, Result};
19use crate::event::Event;
20use crate::id::{Origin, RecordId};
21use crate::model::{Record, Timestamp};
22use crate::projection::project;
23use crate::provenance::ContentHash;
24use crate::repository::Repository;
25use crate::schema::{BUNDLE_FORMAT_VERSION, ENVELOPE_SCHEMA_VERSION, SchemaVersion};
26use crate::storage::Store;
27
28pub const BUNDLE_FORMAT: &str = "pardarsh.bundle";
29
30/// Serialize to canonical JSON bytes.
31pub fn canonical_json<T: Serialize>(value: &T) -> Result<Vec<u8>> {
32    let v = serde_json::to_value(value)?;
33    let mut out = Vec::new();
34    write_canonical(&v, &mut out)?;
35    Ok(out)
36}
37
38fn write_canonical(v: &Value, out: &mut Vec<u8>) -> Result<()> {
39    match v {
40        Value::Object(map) => {
41            let mut keys: Vec<&String> = map.keys().collect();
42            keys.sort();
43            out.push(b'{');
44            for (i, k) in keys.into_iter().enumerate() {
45                if i > 0 {
46                    out.push(b',');
47                }
48                serde_json::to_writer(&mut *out, k)?;
49                out.push(b':');
50                write_canonical(&map[k], out)?;
51            }
52            out.push(b'}');
53        }
54        Value::Array(items) => {
55            out.push(b'[');
56            for (i, item) in items.iter().enumerate() {
57                if i > 0 {
58                    out.push(b',');
59                }
60                write_canonical(item, out)?;
61            }
62            out.push(b']');
63        }
64        scalar => serde_json::to_writer(&mut *out, scalar)?,
65    }
66    Ok(())
67}
68
69/// SHA-256 of the canonical JSON form.
70pub fn content_hash<T: Serialize>(value: &T) -> Result<ContentHash> {
71    Ok(ContentHash::sha256(&canonical_json(value)?))
72}
73
74/// Whether a bundle carries complete histories or a viewer-filtered view.
75#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
76#[serde(rename_all = "snake_case")]
77pub enum Completeness {
78    /// Full histories; records can be re-derived and the bundle imported.
79    Complete,
80    /// Restricted events/sources removed for a viewer; for reading only.
81    Filtered,
82}
83
84#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
85pub struct BundleEntry {
86    pub record: Record,
87    pub history: Vec<Event>,
88    pub record_hash: ContentHash,
89}
90
91#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
92pub struct Bundle {
93    pub format: String,
94    pub format_version: SchemaVersion,
95    pub envelope_version: SchemaVersion,
96    pub origin: Origin,
97    pub exported_at: Timestamp,
98    pub completeness: Completeness,
99    pub entries: Vec<BundleEntry>,
100    /// Hash of the canonical bundle with this field set to `null`.
101    #[serde(default)]
102    pub bundle_hash: Option<ContentHash>,
103}
104
105impl Bundle {
106    fn compute_hash(&self) -> Result<ContentHash> {
107        let mut copy = self.clone();
108        copy.bundle_hash = None;
109        content_hash(&copy)
110    }
111
112    pub fn to_json_pretty(&self) -> Result<String> {
113        Ok(serde_json::to_string_pretty(self)?)
114    }
115}
116
117/// Export records (and their histories) as `viewer` may see them. Records
118/// the viewer cannot see are silently omitted.
119pub fn export<S: Store>(repo: &Repository<S>, ids: &[RecordId], viewer: &Viewer) -> Result<Bundle> {
120    let mut entries = Vec::new();
121    let mut seen = BTreeSet::new();
122    for id in ids {
123        if !seen.insert(id.clone()) {
124            continue;
125        }
126        let Some(record) = repo.find(id)? else { continue };
127        let Some(record) = view_record(&record, viewer) else { continue };
128        let history = view_history(&repo.history(id)?, viewer);
129        let record_hash = content_hash(&record)?;
130        entries.push(BundleEntry { record, history, record_hash });
131    }
132    let mut bundle = Bundle {
133        format: BUNDLE_FORMAT.to_string(),
134        format_version: BUNDLE_FORMAT_VERSION,
135        envelope_version: ENVELOPE_SCHEMA_VERSION,
136        origin: repo.store().origin().clone(),
137        exported_at: repo.now(),
138        completeness: if viewer.privileged { Completeness::Complete } else { Completeness::Filtered },
139        entries,
140        bundle_hash: None,
141    };
142    bundle.bundle_hash = Some(bundle.compute_hash()?);
143    Ok(bundle)
144}
145
146/// Rules for accepting a bundle from elsewhere.
147#[derive(Clone, Debug)]
148pub struct ImportPolicy {
149    pub max_bytes: usize,
150    pub max_records: usize,
151    pub max_events_per_record: usize,
152    /// Accept records minted by the importing store's own origin
153    /// (e.g. restoring a backup). Off by default: a remote party must not be
154    /// able to inject records that look locally authored.
155    pub allow_own_origin: bool,
156    /// If set, only records from these origins are accepted.
157    pub trusted_origins: Option<BTreeSet<Origin>>,
158}
159
160impl Default for ImportPolicy {
161    fn default() -> Self {
162        Self {
163            max_bytes: 16 * 1024 * 1024,
164            max_records: 10_000,
165            max_events_per_record: 10_000,
166            allow_own_origin: false,
167            trusted_origins: None,
168        }
169    }
170}
171
172/// Parse and verify an untrusted bundle without writing anything.
173pub fn parse_bundle(bytes: &[u8], policy: &ImportPolicy) -> Result<Bundle> {
174    if bytes.len() > policy.max_bytes {
175        return Err(Error::Import(format!("bundle is {} bytes; limit is {}", bytes.len(), policy.max_bytes)));
176    }
177    let bundle: Bundle = serde_json::from_slice(bytes).map_err(|e| Error::Import(format!("malformed bundle: {e}")))?;
178    verify_bundle(&bundle, policy)?;
179    Ok(bundle)
180}
181
182pub fn verify_bundle(bundle: &Bundle, policy: &ImportPolicy) -> Result<()> {
183    let err = |m: String| Err(Error::Import(m));
184    if bundle.format != BUNDLE_FORMAT {
185        return err(format!("unknown format {:?}", bundle.format));
186    }
187    if !BUNDLE_FORMAT_VERSION.can_read(&bundle.format_version) {
188        return err(format!("unsupported bundle version {}", bundle.format_version));
189    }
190    if !ENVELOPE_SCHEMA_VERSION.can_read(&bundle.envelope_version) {
191        return err(format!("unsupported envelope version {}", bundle.envelope_version));
192    }
193    if bundle.entries.len() > policy.max_records {
194        return err(format!("{} records exceed the limit of {}", bundle.entries.len(), policy.max_records));
195    }
196    match &bundle.bundle_hash {
197        Some(h) if *h == bundle.compute_hash()? => {}
198        Some(_) => return err("bundle hash does not match its contents".into()),
199        None => return err("bundle hash is missing".into()),
200    }
201    for entry in &bundle.entries {
202        let id = &entry.record.id;
203        if entry.history.len() > policy.max_events_per_record {
204            return err(format!("{id}: too many events"));
205        }
206        if content_hash(&entry.record)? != entry.record_hash {
207            return err(format!("{id}: record hash does not match"));
208        }
209        if entry.history.iter().any(|e| &e.record_id != id) {
210            return err(format!("{id}: history contains events of another record"));
211        }
212        if bundle.completeness == Completeness::Complete {
213            let projected = project(&entry.history).map_err(|e| Error::Import(format!("{id}: {e}")))?;
214            if projected.as_ref() != Some(&entry.record) {
215                return err(format!("{id}: record does not match its history"));
216            }
217        }
218    }
219    Ok(())
220}
221
222#[derive(Clone, Debug, Default, PartialEq, Eq)]
223pub struct ImportReport {
224    pub records: usize,
225    pub events_written: usize,
226    pub events_already_present: usize,
227}
228
229/// Import a verified, complete bundle. Events already present (same event id
230/// at the same position) are skipped, so re-importing is idempotent and a
231/// newer export of the same records only appends what is new.
232pub fn import<S: Store>(repo: &mut Repository<S>, bundle: &Bundle, policy: &ImportPolicy) -> Result<ImportReport> {
233    verify_bundle(bundle, policy)?;
234    if bundle.completeness != Completeness::Complete {
235        return Err(Error::Import("filtered bundles cannot be imported".into()));
236    }
237    let own = repo.store().origin().clone();
238    for entry in &bundle.entries {
239        let origin = entry.record.id.origin();
240        if origin == &own && !policy.allow_own_origin {
241            return Err(Error::Import(format!("{}: records of the local origin are not accepted", entry.record.id)));
242        }
243        if let Some(trusted) = &policy.trusted_origins {
244            if !trusted.contains(origin) {
245                return Err(Error::Import(format!("{}: origin {origin} is not trusted", entry.record.id)));
246            }
247        }
248    }
249    let mut report = ImportReport::default();
250    for entry in &bundle.entries {
251        let existing = repo.history(&entry.record.id)?;
252        for (i, local) in existing.iter().enumerate() {
253            if entry.history.get(i).map(|e| e.id) != Some(local.id) {
254                return Err(Error::Import(format!(
255                    "{}: local history diverges from the bundle at sequence {}",
256                    entry.record.id, local.sequence
257                )));
258            }
259        }
260        report.events_already_present += existing.len();
261        for event in entry.history.iter().skip(existing.len()) {
262            repo.ingest(event.clone())?;
263            report.events_written += 1;
264        }
265        report.records += 1;
266    }
267    Ok(report)
268}
pardarsh_core/interchange.rs

pardarsh_core/
interchange.rs
