From 363a8d6fe5b0e09330fe5a2f26bcdb80238e5dd5 Mon Sep 17 00:00:00 2001 From: Libby Date: Thu, 12 Nov 2026 05:43:55 +0000 Subject: [PATCH] Rust: a payload is one MessagePack value (bytes-after-rs) (#131) --- docs/cultcache-union-parity-map.md | 2 +- packages/cultcache-rs/CHANGELOG.md | 2 + packages/cultcache-rs/src/lib.rs | 137 +++++++++++++--- packages/cultcache-rs/tests/slot_readonly.rs | 52 +++++- packages/cultcache-rs/tests/union_parity.rs | 162 ++++++++++++++++++- packages/cultnet-rs/CHANGELOG.md | 1 + packages/cultnet-rs/tests/cultnet.rs | 10 +- packages/cultnet-rs/tests/decode_door.rs | 10 ++ 8 files changed, 346 insertions(+), 30 deletions(-) diff --git a/docs/cultcache-union-parity-map.md b/docs/cultcache-union-parity-map.md index 368be16a..1442abdc 100644 --- a/docs/cultcache-union-parity-map.md +++ b/docs/cultcache-union-parity-map.md @@ -259,7 +259,7 @@ landed in `ead5260c`, Python in `5618d8dd`, Rust in `c028c511`. TypeScript is no | Runtime | Owner of the decision (constant and walker) | Doors routed through it | |---|---|---| | C# | `CultPayloadDepth.Check` (`src/GameCult.Caching.MessagePack/CultPayloadDepth.cs`), limit `CultMessagePackSecurity.MaximumPayloadDepth`. MessagePack's own graph counter is set to the same value as a backstop only. | Write: `CultDocumentMessagePackSerialization.SerializeUntyped`. Read: `DeserializeUntyped`. The cache store (`SerializePayload`/`DeserializePayload`), the directory store, `CultMesh` and single-file documents all call those two; C# has no separate raw-envelope or prepared-batch door. | -| Rust | `cultcache_rs::check_payload_depth` and `MAX_PAYLOAD_DEPTH` (`packages/cultcache-rs/src/lib.rs`). | Read: `decode_payload` (get, list, `validate_envelope`/`put_envelope`, `finish_prepared_entry`). Write: `put`/`prepare_entry`/`prepare_entry_named` reach it through `finish_prepared_entry`, which decodes what it just encoded. Raw envelope: `put_raw_envelope`, the door CultMesh network puts use (`cultmesh-rs` `lib.rs`, two call sites). Prepared batch: `put_prepared_batch` via `require_payload_depth`. cultnet-rs: every free-form or skippable wire payload decodes through `cultcache_rs::decode_payload` (message decode in `contracts.rs`, `provider_session`, `media_stream_wire`, `replication`, `service_identity`). | +| Rust | `check_record_payload` (private; depth, then no bytes after the first value) over `first_value_end`, and `MAX_PAYLOAD_DEPTH`; `cultcache_rs::check_payload_depth` is the depth-only probe for outside callers (`packages/cultcache-rs/src/lib.rs`). | Read: `decode_payload` (get, list, `validate_envelope`/`put_envelope`, `finish_prepared_entry`). Write: `put`/`prepare_entry`/`prepare_entry_named` reach it through `finish_prepared_entry`, which decodes what it just encoded. Raw envelope: `put_raw_envelope`, the door CultMesh network puts use (`cultmesh-rs` `lib.rs`, two call sites). Prepared batch: `put_prepared_batch` via `require_record_payload`. Load: `pull_all_backing_stores` and `load_soa` run `require_record_payload` on every record, so a store holding a payload past the bound (or with bytes after it) is refused at load, and the cache then refuses every write and removal (`CultCacheLoadRefused`) until a load succeeds. cultnet-rs: every free-form or skippable wire payload decodes through `cultcache_rs::decode_payload` (message decode in `contracts.rs`, `provider_session`, `media_stream_wire`, `replication`, `service_identity`). | | Python | `_refuse_past_depth_limit` and `MAX_PAYLOAD_DEPTH` (`packages/cultcache-py/src/cultcache_py/documents.py`); a msgpack `StackError` on read maps to the same error. | `DocumentDefinition.encode_payload` (write) and `decode_payload` (read). `CultCache.put`, `put_envelope`, `put_envelopes` and load-time hydration go through them; there is no raw-envelope door. cultmesh-py round trips call the same pair. | - **Doors deliberately not routed.** cultnet-rs `idunn_lifecycle_brake`, `provider_health_contracts` and diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 353facea..4e97588c 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -15,6 +15,8 @@ Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0). - `SchemaConflictError` gains the public field `unread_slots: Option`; code that builds it with a struct literal must set it (`None`). - A unit arm with slots (`[N, [extra...]]`) decodes as the unit arm and marks the record, where the derive refused it. - The registration map's value changes from the schema name to a private struct holding the schema name and the unread-slots check. +- A cache whose last `pull_all_backing_stores` was refused (a tail, a payload past the nesting bound, a duplicate unique-index value, a retype conflict, a record from the wrong store) refuses every `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch`, `update` and `delete` with `CultCacheLoadRefused` until a pull succeeds. Before, it stayed writable over an empty image, and the next batch write replaced the store with only the new records. +- `decode_payload`, `put_raw_envelope` and prepared batches refuse a payload with bytes after its first MessagePack value (`payload has {n} bytes after its first MessagePack value`), by count; the tail is never walked or quoted. They used to read the first value and keep or drop the tail, so a record with a stray tail was served and rewritten without it. A store holding such a record (single-file, redb, `load_soa`) is refused at load, naming the type, the key and the count, and stays unreadable until the record is repaired; the same check now runs on every record at load, as the C# and Python runtimes do at open. The nesting bound is part of that check, so a stored payload nested past it is refused at load too, where it used to load and refuse on read. - A single-file store written as `cultcache.store.v3` is read and written: a store the C# runtime marked v3 (it holds a record with an element id: an owned list element with a `[CultElementId]` member) keeps its header here, and is carried through every rewrite with its ids; this runtime never drops the mark. A runtime from before this change refuses a v3 store by name. Rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading. - A write over a file the reader refuses (a zero-byte file, a truncated store, a stamp this runtime does not read) refuses with `CultCacheStoreUnreadable` and leaves the file as it is. A whole-store write used to replace such a file. This reverses 0.3.1, where `push_all` replaced a file with no readable stamp (healing it); the stamp rule now has one owner, the reader every write asks under the lock, and it accepts v1 and v3 (0.3.1 refused v3). - `#[cultcache(index)]` means unique, as in C# `RefuseIndexSharing`: within one type and one alias no two records hold one non-blank value. A put, batch, push or load that would make two records hold a value is refused, naming both keys and the alias, and a store that already holds a duplicate does not open. A grouping index is no longer expressible; a consumer that relied on shared values must drop the `index` and look up by another way. diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 59ffa1af..522e226e 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -600,14 +600,22 @@ impl std::error::Error for PayloadTooDeep {} /// Walks the MessagePack headers of one payload without recursion, counting every array and map and /// refusing the first container past [`MAX_PAYLOAD_DEPTH`]. Scalars are skipped by length and nothing -/// is allocated per element. A payload that ends early is the decoder's to refuse. +/// is allocated per element. Only the first value is walked; a payload that ends early is the +/// decoder's to refuse. A depth-only probe: [`decode_payload`] also refuses bytes after the value. pub fn check_payload_depth(bytes: &[u8]) -> std::result::Result<(), PayloadTooDeep> { + first_value_end(bytes).map(|_| ()) +} + +/// Where the first MessagePack value of `bytes` ends, walking only that value (its tail is never +/// read). `None` when the payload ends inside a header; a final scalar cut short ends past the +/// payload's length, which is not bytes after the value either. Both are the decoder's to refuse. +fn first_value_end(bytes: &[u8]) -> std::result::Result, PayloadTooDeep> { let mut remaining = [0usize; MAX_PAYLOAD_DEPTH]; let mut depth = 0usize; let mut position = 0usize; while position < bytes.len() { let Some((advance, children)) = element_header(bytes, position) else { - return Ok(()); + return Ok(None); }; position = position.saturating_add(advance); if let Some(children) = children { @@ -629,7 +637,18 @@ pub fn check_payload_depth(bytes: &[u8]) -> std::result::Result<(), PayloadTooDe depth -= 1; } if depth == 0 { - return Ok(()); + return Ok(Some(position)); + } + } + Ok(None) +} + +/// The one rule for "is this payload one value": the nesting bound, then no bytes after the first +/// value. The count is all it reports; the tail is never walked or quoted. +fn check_record_payload(bytes: &[u8]) -> Result<()> { + if let Some(end) = first_value_end(bytes)? { + if end < bytes.len() { + bail!("payload has {} bytes after its first MessagePack value", bytes.len() - end); } } Ok(()) @@ -784,7 +803,7 @@ fn decode_grown(bytes: &[u8]) -> std::result::Re grown(&mut reader, |grown| T::deserialize(grown)) } -/// The one decode of a document payload: the nesting bound first, then [`decode_grown`], whose +/// The one decode of a document payload: the payload check first (nesting bound, one value), then [`decode_grown`], whose /// rmp-serde depth counter (1024) is only a backstop. A failure keeps its kind and never quotes the /// bytes. Readers of stored envelope payloads outside this crate decode through it too. /// @@ -796,7 +815,7 @@ fn decode_grown(bytes: &[u8]) -> std::result::Re /// verified on x86_64-pc-windows-msvc by hand (`decode_stack`, revision 10a39dac, 13 tests, debug and release); a /// `stacker` bump or a change to the stack-size tests repeats it. pub fn decode_payload(payload: &[u8]) -> Result { - check_payload_depth(payload)?; + check_record_payload(payload)?; decode_grown(payload).map_err(opaque_decode_error) } @@ -918,10 +937,11 @@ impl Drop for UnreadScope { } } -/// [`decode_payload`] with the mark: the same depth check and grown decode, inside a scope the derives' slot -/// visitors note into. A payload past the nesting limit is refused before the decode, never marked. +/// [`decode_payload`] with the mark: the same payload check and grown decode, inside a scope the derives' slot +/// visitors note into. A payload past the nesting limit or with bytes after its value is refused before the +/// decode, never marked. pub fn decode_payload_marked(payload: &[u8]) -> Result<(T, Option)> { - check_payload_depth(payload)?; + check_record_payload(payload)?; let scope = UnreadScope::open(); let value = decode_grown(payload).map_err(opaque_decode_error)?; Ok((value, scope.mark())) @@ -948,10 +968,10 @@ struct Definition { unread_slots: fn(&[u8]) -> Result>, } -/// The nesting bound for an envelope admitted without a typed decode, from the same walk that -/// [`decode_payload`] runs first. Names the entry's type and key, never the payload. -fn require_payload_depth(entry: &CultCacheEnvelope) -> Result<()> { - check_payload_depth(&entry.payload).with_context(|| { +/// The payload check for an envelope admitted without a typed decode, the one [`decode_payload`] +/// runs first. Names the entry's type and key, never the payload. +fn require_record_payload(entry: &CultCacheEnvelope) -> Result<()> { + check_record_payload(&entry.payload).with_context(|| { format!( "refused CultCache envelope {:?} at key {:?}", entry.r#type, entry.key @@ -2917,8 +2937,30 @@ pub struct CultCache { /// Records a store holds that no registered type claims: (store index, the envelope as the store read it). Derived from the last /// pull and replaced with it; written nowhere else. It decides only refusals at its keys. foreign: Vec<(usize, CultCacheEnvelope)>, + /// Why the last pull of the backing stores was refused; `None` once a pull succeeds. While set, every write and removal is refused. + load_refused: Option, } +/// A write or removal refused because the cache's last load of its backing stores was refused. The cache holds no +/// image of those stores, so any write would persist an image that lacks the records it never read. Repair the store +/// named by `reason` (which names a type, a key and a clause, never a payload) and load again. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultCacheLoadRefused { + pub reason: String, +} + +impl std::fmt::Display for CultCacheLoadRefused { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "CultCache refuses writes and removals: its load of the backing stores failed ({}); repair the store and load again", + self.reason + ) + } +} + +impl std::error::Error for CultCacheLoadRefused {} + /// A record a store holds that no registered type claims: carried, never served. #[derive(Clone, Debug, PartialEq, Eq)] pub struct ForeignRecord { @@ -2980,6 +3022,7 @@ impl CultCache { projectors: BTreeMap::new(), stores: Vec::new(), foreign: Vec::new(), + load_refused: None, } } @@ -3116,12 +3159,39 @@ impl CultCache { /// Reads every store into a staged image and replaces the cache's view /// only when all of them loaded, so a refused load admits nothing. pub fn pull_all_backing_stores(&mut self) -> Result<()> { + match self.stage_pull() { + Ok((loaded, foreign, holders)) => { + self.entries = loaded; + self.foreign = foreign; + self.index_holders = holders; + self.load_refused = None; + Ok(()) + } + Err(error) => { + self.load_refused = Some(format!("{error:#}")); + Err(error) + } + } + } + + /// The one gate of every write and removal: a cache whose load was refused holds no image of its stores. + fn require_loaded(&self) -> Result<()> { + match &self.load_refused { + Some(reason) => Err(CultCacheLoadRefused { reason: reason.clone() }.into()), + None => Ok(()), + } + } + + /// The image every store holds, judged whole, installed nowhere. + #[allow(clippy::type_complexity)] + fn stage_pull(&self) -> Result<(BTreeMap<(String, String), CultCacheEnvelope>, Vec<(usize, CultCacheEnvelope)>, IndexHolders)> { let mut loaded = BTreeMap::new(); let mut foreign = Vec::new(); let known_types: BTreeSet = self.definitions.keys().cloned().collect(); let schema_name_definitions = self.schema_name_definitions.clone(); for (index, registration) in self.stores.iter().enumerate() { for mut entry in registration.store.pull_all()? { + require_record_payload(&entry)?; let Some(canonical_type) = resolve_registered_type( &known_types, &schema_name_definitions, @@ -3154,10 +3224,7 @@ impl CultCache { } self.refuse_second_globals(&loaded)?; let holders = self.judge_image(&loaded)?; - self.entries = loaded; - self.foreign = foreign; - self.index_holders = holders; - Ok(()) + Ok((loaded, foreign, holders)) } /// The records the last pull found in a store that no registered type claims, sorted by key (UTF-16 order). The cache does not @@ -3383,6 +3450,7 @@ impl CultCache { } pub fn put(&mut self, key: impl Into, value: &T) -> Result { + self.require_loaded()?; let (entry, parsed) = self.prepare_entry(key, value)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?; @@ -3460,6 +3528,7 @@ impl CultCache { } pub fn put_prepared_batch(&mut self, staged: Vec) -> Result<()> { + self.require_loaded()?; if staged.is_empty() { return Err(anyhow!("CultCache batch must contain at least one entry")); } @@ -3484,7 +3553,7 @@ impl CultCache { entry.r#type )); } - require_payload_depth(entry)?; + require_record_payload(entry)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_claimed_id(entry)?; let Some(home) = self.home_index(&entry.r#type)? else { @@ -3527,6 +3596,7 @@ impl CultCache { } pub fn put_envelope(&mut self, entry: CultCacheEnvelope) -> Result { + self.require_loaded()?; let parsed = self.validate_envelope::(&entry)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_claimed_id(&entry)?; @@ -3547,6 +3617,7 @@ impl CultCache { /// admission; CultCache still enforces registered type, identity, timestamp, /// the payload nesting bound ([`MAX_PAYLOAD_DEPTH`]), routing, and persistence. pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> { + self.require_loaded()?; if !self.definitions.contains_key(&entry.r#type) { return Err(anyhow!( "No entry type registered for CultCache envelope type {:?}", @@ -3565,7 +3636,7 @@ impl CultCache { entry.r#type )); } - require_payload_depth(&entry)?; + require_record_payload(&entry)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_claimed_id(&entry)?; self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?; @@ -3621,6 +3692,7 @@ impl CultCache { } pub fn delete(&mut self, key: &str) -> Result { + self.require_loaded()?; self.require_entry_type::()?; self.refuse_unread_slots(T::TYPE, key)?; let id = entry_id_parts(T::TYPE, key); @@ -3662,6 +3734,7 @@ impl CultCache { entry.r#type )); } + require_record_payload(&entry)?; self.home_index(&entry.r#type)?; self.refuse_claimed_id(&entry)?; self.refuse_retype(&loaded, &self.foreign, &entry)?; @@ -6296,6 +6369,34 @@ mod tests { Ok(()) } + #[test] + fn put_raw_envelope_refuses_bytes_after_the_value() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("tail.cc"); + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + let tailed = CultCacheEnvelope { + key: "app".to_string(), + r#type: Settings::TYPE.to_string(), + payload: vec![0x91, 0x01, 0xc0], + stored_at: now_utc_second(), + schema_id: None, + }; + let error = format!("{:#}", cache.put_raw_envelope(tailed.clone()).unwrap_err()); + assert!(error.contains("payload has 1 bytes after its first MessagePack value"), "{error}"); + assert!(error.contains("app") && error.contains(Settings::TYPE), "{error}"); + assert!(cache.snapshot().is_empty()); + assert!(!store_path.exists() || SingleFileMessagePackBackingStore::new(&store_path).pull_all()?.is_empty()); + let error = format!("{:#}", cache.put_prepared_batch(vec![tailed.clone()]).unwrap_err()); + assert!(error.contains("payload has 1 bytes after its first MessagePack value"), "{error}"); + assert!(cache.snapshot().is_empty()); + // The same bytes without the tail are admitted: the refusal is the count, not the shape. + cache.put_raw_envelope(CultCacheEnvelope { payload: vec![0x91, 0x01], ..tailed })?; + assert_eq!(cache.snapshot().len(), 1); + Ok(()) + } + #[test] fn soa_snapshot_round_trips_registered_entries() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/packages/cultcache-rs/tests/slot_readonly.rs b/packages/cultcache-rs/tests/slot_readonly.rs index 5bffcd9a..0031e372 100644 --- a/packages/cultcache-rs/tests/slot_readonly.rs +++ b/packages/cultcache-rs/tests/slot_readonly.rs @@ -481,9 +481,11 @@ fn load_soa_refuses_a_same_length_replacement_of_a_marked_record() { let (mut cache, _dir, _) = held_marked_and_clean(); let all = cache.snapshot_envelopes(); let marked = all.iter().find(|held| held.key == "marked").unwrap(); - // The clean record's bytes padded to the marked record's length: same length, different bytes. - let mut same_length = vec![0xc0; marked.payload.len()]; - same_length[..2].copy_from_slice(&[0x91, 0xa1]); + // One value (a bin16) of the marked record's length: same length, different bytes. A value followed by padding + // would be refused at admission as bytes after the value, before the read-only check is reached. + let length = marked.payload.len(); + let mut same_length = vec![0u8; length]; + same_length[..3].copy_from_slice(&[0xc5, ((length - 3) >> 8) as u8, (length - 3) as u8]); assert_eq!(same_length.len(), marked.payload.len()); let mut replaced = all.clone(); replaced.iter_mut().find(|held| held.key == "marked").unwrap().payload = same_length; @@ -491,3 +493,47 @@ fn load_soa_refuses_a_same_length_replacement_of_a_marked_record() { assert_refused_with(&error, "marked", &mark("Composition", 3, 2, 1)); assert_eq!(cache.snapshot_envelopes(), all); } + +// The payload check runs before the mark scope opens: a tail on a slot-bearing payload is refused by count, never marked. +#[test] +fn a_tail_on_a_slot_payload_is_refused_before_any_mark() { + let mut tailed = vector("extra-slot-record.msgpack"); + tailed.push(0xc0); + let error = decode_payload_marked::(&tailed).unwrap_err(); + assert!(format!("{error:#}").contains("payload has 1 bytes after its first MessagePack value"), "{error:#}"); + // Nothing leaked: the next marked decode finds exactly its own mark. + let found = decode_payload_marked::(&vector("extra-slot-record.msgpack")).unwrap().1.unwrap(); + assert_eq!(found.occurrences, 1); +} + +// A store holding marked records and one with a tail is refused at load; the cache then refuses every write with the load +// refusal (not a per-key mark conflict), and the store file is untouched. +#[test] +fn a_store_with_marked_records_and_a_tail_is_refused_and_its_cache_writes_nothing() { + use cultcache_rs::CacheBackingStore; + use cultcache_rs::CultCacheLoadRefused; + let (mut held, _dir, path) = held_marked_and_clean(); + let mut tailed = vector("union-surface.msgpack"); + tailed.push(0xc0); + SingleFileMessagePackBackingStore::new(&path).push(&envelope("tailed", tailed)).unwrap(); + let before = std::fs::read(&path).unwrap(); + // The cache that held the store re-reads it and is refused. + assert!(format!("{:#}", held.pull_all_backing_stores().unwrap_err()).contains("bytes after")); + let fresh_value = constructed(); + let mut fresh = registered(); + fresh.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + assert!(fresh.pull_all_backing_stores().is_err()); + for cache in [&mut held, &mut fresh] { + let staged = cache.prepare_entry("new", &fresh_value).unwrap().0; + let errors = [ + cache.put("new", &fresh_value).unwrap_err(), + cache.put_prepared_batch(vec![staged.clone()]).unwrap_err(), + cache.put_raw_envelope(staged).unwrap_err(), + cache.delete::("new").unwrap_err(), + ]; + for error in errors { + assert!(error.downcast_ref::().is_some(), "{error:#}"); + } + } + assert_eq!(std::fs::read(&path).unwrap(), before); +} diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index c51bd3d8..67f2b389 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -4,9 +4,13 @@ use cultcache_rs::CultCache; use cultcache_rs::CacheBackingStore; use cultcache_rs::CultCacheEnvelope; +use cultcache_rs::CultCacheLoadRefused; +use cultcache_rs::CultCacheSoaTable; use cultcache_rs::MAX_PAYLOAD_DEPTH; use cultcache_rs::PayloadTooDeep; use cultcache_rs::check_payload_depth; +use cultcache_rs::decode_payload; +use cultcache_rs::RedbMessagePackBackingStore; use cultcache_rs::DatabaseEntry; use cultcache_rs::SingleFileMessagePackBackingStore; @@ -464,7 +468,7 @@ fn envelope(key: &str, payload: &[u8]) -> CultCacheEnvelope { } // A payload past the bound cannot be admitted through the cache, so a store file written behind its -// back stands in for one a peer or an older runtime left. +// back stands in for one a peer or an older runtime left; load or read refuses it. fn through_cache(payload: &[u8]) -> Result { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("foreign.cc"); @@ -472,7 +476,10 @@ fn through_cache(payload: &[u8]) -> Result { let mut cache = CultCache::new(); cache.register_entry_type::().unwrap(); cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); - cache.pull_all_backing_stores().unwrap(); + // The payload check runs at load, so a payload past the bound is refused there, before any read. + if let Err(error) = cache.pull_all_backing_stores() { + return Err(format!("{error:#}")); + } match cache.get::("k") { Ok(value) => Ok(value.expect("the envelope was stored")), Err(error) => Err(format!("{error:#}")), @@ -529,6 +536,157 @@ fn bounds_vectors() { assert_eq!((accepted, refused, read_only), (4, 7, 1)); } +// A payload is one MessagePack value (bytes-after-outcomes.msgpack): every door refuses a tail by +// its count, names the clause, never the depth limit (the tail is not walked) and never a value. +#[test] +fn bytes_after_vectors() { + let rows: Vec = rmp_serde::from_slice(&vector("bytes-after-outcomes.msgpack")).unwrap(); + assert_eq!(rows.len(), 4); + let surface = vector("union-surface.msgpack"); + for row in &rows { + let fields = row.as_array().unwrap(); + let (file, outcome, union, detail) = (fields[0].as_str().unwrap(), fields[1].as_str().unwrap(), fields[2].as_str().unwrap(), fields[4].as_str().unwrap()); + assert_eq!((outcome, union, detail), ("refused", "UnionSurface", "bytes-after"), "{file}"); + let bytes = vector(file); + assert_eq!(&bytes[..surface.len()], &surface[..], "{file}"); + let clause = format!("payload has {} bytes after its first MessagePack value", bytes.len() - surface.len()); + let check = |door: &str, message: &str| { + assert!(message.contains(&clause), "{file} {door}: {message}"); + assert!(!message.contains("the limit is"), "{file} {door} walked the tail: {message}"); + assert!(!message.contains("CANARY-bytes-after-7d1e"), "{file} {door} echoed the tail: {message}"); + assert!(!message.contains(CANARY), "{file} {door} echoed a value: {message}"); + }; + let error = decode_payload::(&bytes).unwrap_err(); + assert!(error.downcast_ref::().is_none(), "{file}"); + check("decode_payload", &format!("{error:#}")); + // A store written behind the cache's back is refused at load, naming the type, the key and the count. + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("foreign.cc"); + SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", &bytes)).unwrap(); + let mut single = CultCache::new(); + single.register_entry_type::().unwrap(); + single.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + let message = format!("{:#}", single.pull_all_backing_stores().unwrap_err()); + check("single-file load", &message); + assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}"); + let redb_path = temp.path().join("foreign.redb"); + RedbMessagePackBackingStore::new(&redb_path).unwrap().push(&envelope("k", &bytes)).unwrap(); + let mut redb = CultCache::new(); + redb.register_entry_type::().unwrap(); + redb.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb_path).unwrap()).unwrap(); + let message = format!("{:#}", redb.pull_all_backing_stores().unwrap_err()); + check("redb load", &message); + assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}"); + let mut soa = CultCache::new(); + soa.register_entry_type::().unwrap(); + let message = format!("{:#}", soa.load_soa(CultCacheSoaTable::from_envelopes(vec![envelope("k", &bytes)])).unwrap_err()); + check("load_soa", &message); + assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}"); + assert!(soa.snapshot().is_empty(), "{file}"); + // Raw admission refuses and stages nothing. + let mut raw = CultCache::new(); + raw.register_entry_type::().unwrap(); + let message = format!("{:#}", raw.put_raw_envelope(envelope("k", &bytes)).unwrap_err()); + check("put_raw_envelope", &message); + assert!(raw.snapshot().is_empty(), "{file}"); + } +} + +// Depth is found inside the first value, so a tail never hides it: a payload nested past the bound +// with a tail after it is the depth refusal, not bytes after. +#[test] +fn depth_beats_a_tail() { + let mut deep = vec![0x91u8; MAX_PAYLOAD_DEPTH + 1]; + deep.push(0xc0); + deep.push(0xc0); + let error = decode_payload::(&deep).unwrap_err(); + assert_eq!(error.downcast_ref::(), Some(&TOO_DEEP), "{error:#}"); +} + +// Depth is refused where the payload is loaded: a store holding one nested past the bound is not read. +#[test] +fn depth_is_refused_at_load() { + let mut deep = vec![0x91u8; MAX_PAYLOAD_DEPTH + 1]; + deep.push(0xc0); + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("deep.cc"); + SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", &deep)).unwrap(); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + let error = cache.pull_all_backing_stores().unwrap_err(); + assert_eq!(error.downcast_ref::(), Some(&TOO_DEEP), "{error:#}"); + assert!(cache.snapshot().is_empty()); +} + +// A cache whose load was refused refuses every write and removal and leaves the store as it was: a +// later write would otherwise persist the empty image over the records it never read. +#[test] +fn a_refused_load_refuses_every_write() { + let tail = { + let mut bytes = vector("union-surface.msgpack"); + bytes.push(0xc0); + bytes + }; + let good = envelope("good", &vector("union-surface.msgpack")); + let bad = envelope("bad", &tail); + for redb in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join(if redb { "seed.redb" } else { "seed.cc" }); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + if redb { + let mut store = RedbMessagePackBackingStore::new(&path).unwrap(); + store.push(&good).unwrap(); + store.push(&bad).unwrap(); + drop(store); + cache.add_generic_backing_store(RedbMessagePackBackingStore::new(&path).unwrap()).unwrap(); + } else { + let mut store = SingleFileMessagePackBackingStore::new(&path); + store.push(&good).unwrap(); + store.push(&bad).unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + } + let load = cache.pull_all_backing_stores().unwrap_err(); + assert!(format!("{load:#}").contains("bytes after"), "{load:#}"); + let before = std::fs::read(&path).unwrap(); + let value = constructed(); + let staged = cache.prepare_entry("new", &value).unwrap().0; + let results: Vec<(&str, anyhow::Error)> = vec![ + ("put", cache.put("new", &value).unwrap_err()), + ("put_prepared_batch", cache.put_prepared_batch(vec![staged.clone()]).unwrap_err()), + ("put_envelope", cache.put_envelope::(staged.clone()).unwrap_err()), + ("put_raw_envelope", cache.put_raw_envelope(staged.clone()).unwrap_err()), + ("update", cache.update::("good", |current| current.unwrap_or_else(constructed)).unwrap_err()), + ("delete", cache.delete::("good").unwrap_err()), + ("delete a key the image lacks", cache.delete::("absent").unwrap_err()), + ]; + for (door, error) in results { + let refusal = error.downcast_ref::().unwrap_or_else(|| panic!("{door} redb={redb}: {error:#}")); + assert!(refusal.to_string().contains("bytes after"), "{door}: {refusal}"); + } + assert_eq!(std::fs::read(&path).unwrap(), before, "redb={redb}: the store changed"); + assert!(cache.snapshot().is_empty()); + // Repairing the store and loading again lifts it. + if redb { + RedbMessagePackBackingStore::new(&path).unwrap().delete(&bad).unwrap(); + } else { + SingleFileMessagePackBackingStore::new(&path).delete(&bad).unwrap(); + } + cache.pull_all_backing_stores().unwrap(); + cache.put("new", &value).unwrap(); + } +} + +// A payload that ends early is not bytes after the value: the decoder refuses it, not the check. +#[test] +fn a_truncated_payload_is_the_decoders_not_bytes_after() { + let mut truncated = vec![0x91, 0xd9, 0x40]; + truncated.extend_from_slice(b"short"); + let message = format!("{:#}", decode_payload::(&truncated).unwrap_err()); + assert!(!message.contains("bytes after"), "{message}"); +} + fn wrap_chain(wraps: usize, leaf: Node) -> UnionSurface { let mut node = leaf; for _ in 0..wraps { diff --git a/packages/cultnet-rs/CHANGELOG.md b/packages/cultnet-rs/CHANGELOG.md index 384395ec..1a01dbe1 100644 --- a/packages/cultnet-rs/CHANGELOG.md +++ b/packages/cultnet-rs/CHANGELOG.md @@ -15,5 +15,6 @@ only in git history. ### Breaking +- A CultNet frame with bytes after its message is refused, by count (through `cultcache_rs::decode_payload`); the decoder used to read the message and drop the tail. No producer emits one. Behaviour change: the next release is a minor bump (0.x breaking). - `CultNetDocumentRegistry::register` returns `Result<&mut Self>` where it returned `&mut Self`. A binding that claims a document type or a schema id another binding already holds is refused with `cultcache_rs::SchemaConflictError` (naming the new binding's schema id and both document types) and nothing is bound; binding a type to the schema id it already holds still replaces that binding. Callers chain `register(...)?`. - `CultMeshCdnChunkRef` gains a public `id` field, the decimal text of `offset` (the reference's element id, slot 4). A manifest written before ids reads `""`; encoding writes the slot. diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 64c57aaf..7bc84957 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -1810,12 +1810,10 @@ fn a_replicated_payload_nested_past_the_bound_is_refused_typed_on_apply() -> Res }; assert_eq!(documents.len(), 1); - backing.push(&envelope("past", limit + 1))?; - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?; - cache.pull_all_backing_stores()?; - let past = registry.create_raw_snapshot_response(&cache, "past", None, Some(&["past".to_string()]))?; + // The sender cannot load a payload past the bound, so the frame is the at-limit one with its bytes swapped. + let mut past = at_limit.clone(); + let CultNetMessage::SnapshotResponseRaw { documents, .. } = &mut past else { unreachable!() }; + documents[0].payload = envelope("past", limit + 1).payload; let mut target = CultCache::new(); target.register_entry_type::()?; let error = registry diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index fd5bab6a..c55a6404 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -190,6 +190,16 @@ fn message_past_the_bound_is_refused_typed() { }); } +// A frame is one message: cultcache's payload rule refuses bytes after it, by count. +#[test] +fn frame_with_bytes_after_is_refused() { + let mut bytes = hello_with(&nested(0)); + assert!(decode_message(bytes.clone(), BIG).is_ok()); + bytes.push(0xc0); + let error = format!("{:#}", decode_message(bytes, BIG).unwrap_err()); + assert!(error.contains("1 bytes after its first MessagePack value"), "{error}"); +} + fn raw_v1_bytes(header_unknown: Option<&[u8]>) -> Vec { let mut header: Vec<(rmpv::Value, rmpv::Value)> = vec![ ("schemaId".into(), "s".into()),