diff --git a/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs b/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs index 01e46f52..d172e4fd 100644 --- a/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs +++ b/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs @@ -12,6 +12,13 @@ use syn::parse_macro_input; #[proc_macro_derive(DatabaseEntry, attributes(cultcache))] pub fn derive_database_entry(input: TokenStream) -> TokenStream { let input = parse_macro_input!(input as DeriveInput); + match database_entry_codec(input) { + Ok(tokens) => tokens.into(), + Err(error) => error.to_compile_error().into(), + } +} + +fn database_entry_codec(input: DeriveInput) -> syn::Result { let ident = input.ident.clone(); let mut entry_type: Option = None; let mut schema_name: Option = None; @@ -34,36 +41,28 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { Err(meta.error("unsupported cultcache attribute")) } }); - if let Err(error) = parse_result { - return error.to_compile_error().into(); - } + parse_result?; } let Some(entry_type) = entry_type else { - return syn::Error::new_spanned( + return Err(syn::Error::new_spanned( ident, "DatabaseEntry derive requires #[cultcache(type = \"...\")]", - ) - .to_compile_error() - .into(); + )); }; let schema_name = schema_name.unwrap_or_else(|| LitStr::new(&ident.to_string(), ident.span())); - let slots = match named_slots(input.data, &ident, "DatabaseEntry") { - Ok(slots) => slots, - Err(error) => return error.to_compile_error().into(), - }; + let slots = named_slots(input.data, &ident, "DatabaseEntry")?; let codec = slot_codec(&ident, &slots); - quote! { + Ok(quote! { impl ::cultcache_rs::DatabaseEntry for #ident { const TYPE: &'static str = #entry_type; const SCHEMA_NAME: &'static str = #schema_name; } #codec - } - .into() + }) } /// A named struct whose fields carry `#[cultcache(key = N)]`: the same positional @@ -72,18 +71,21 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { #[proc_macro_derive(CultSlots, attributes(cultcache))] pub fn derive_cult_slots(input: TokenStream) -> TokenStream { let input = parse_macro_input!(input as DeriveInput); - let ident = input.ident.clone(); - if !input.generics.params.is_empty() { - return syn::Error::new_spanned(ident, "CultSlots derive does not support generics") - .to_compile_error() - .into(); - } - match named_slots(input.data, &ident, "CultSlots") { - Ok(slots) => slot_codec(&ident, &slots).into(), + match cult_slots_codec(input) { + Ok(tokens) => tokens.into(), Err(error) => error.to_compile_error().into(), } } +fn cult_slots_codec(input: DeriveInput) -> syn::Result { + let ident = input.ident.clone(); + if !input.generics.params.is_empty() { + return Err(syn::Error::new_spanned(ident, "CultSlots derive does not support generics")); + } + let slots = named_slots(input.data, &ident, "CultSlots")?; + Ok(slot_codec(&ident, &slots)) +} + /// An enum written as `[key, armSlots]`. Every variant carries `#[cultcache(key = N)]` /// and is a unit (`[N, []]`) or a newtype over a type that serializes as an array. /// A key the enum does not declare is a decode error; there is no catch-all arm. @@ -96,6 +98,11 @@ pub fn derive_cult_union(input: TokenStream) -> TokenStream { } } +/// The largest array16 length: a slot record is written as a dense array, so a key above it would +/// make every encode allocate and write an unbounded run of nils. Every runtime and the C# catalog +/// refuse the same domain. +const MAX_SLOT_KEY: usize = 65535; + fn named_slots(data: Data, ident: &syn::Ident, owner: &str) -> syn::Result> { let fields = match data { Data::Struct(data) => match data.fields { @@ -135,7 +142,14 @@ fn named_slots(data: Data, ident: &syn::Ident, owner: &str) -> syn::Result MAX_SLOT_KEY { + return Err(syn::Error::new_spanned( + &value, + format!("{owner} slot key {slot} is outside 0..={MAX_SLOT_KEY}"), + )); + } + key = Some(slot); Ok(()) } else if meta.path.is_ident("default") { default = true; @@ -261,12 +275,14 @@ fn slot_codec(ident: &syn::Ident, slots: &[CultCacheSlot]) -> proc_macro2::Token }); quote! { - impl ::cultcache_rs::SlotRecord for #ident {} + impl ::cultcache_rs::SlotRecord for #ident { + fn __slots_array() {} + } impl ::serde::Serialize for #ident { - fn serialize(&self, serializer: S) -> ::std::result::Result + fn serialize<__S>(&self, serializer: __S) -> ::std::result::Result<__S::Ok, __S::Error> where - S: ::serde::Serializer, + __S: ::serde::Serializer, { use ::serde::ser::SerializeTuple; let mut tuple = serializer.serialize_tuple(#max_slot + 1)?; @@ -276,22 +292,22 @@ fn slot_codec(ident: &syn::Ident, slots: &[CultCacheSlot]) -> proc_macro2::Token } impl<'de> ::serde::Deserialize<'de> for #ident { - fn deserialize(deserializer: D) -> ::std::result::Result + fn deserialize<__D>(deserializer: __D) -> ::std::result::Result where - D: ::serde::Deserializer<'de>, + __D: ::serde::Deserializer<'de>, { - struct CultCacheVisitor; + struct __CultCacheVisitor; - impl<'de> ::serde::de::Visitor<'de> for CultCacheVisitor { + impl<'de> ::serde::de::Visitor<'de> for __CultCacheVisitor { type Value = #ident; fn expecting(&self, formatter: &mut ::std::fmt::Formatter) -> ::std::fmt::Result { formatter.write_str(concat!("a CultCache array for ", stringify!(#ident))) } - fn visit_seq(self, mut seq: A) -> ::std::result::Result + fn visit_seq<__A>(self, mut seq: __A) -> ::std::result::Result where - A: ::serde::de::SeqAccess<'de>, + __A: ::serde::de::SeqAccess<'de>, { #(#option_declarations)* let mut index = 0usize; @@ -309,7 +325,7 @@ fn slot_codec(ident: &syn::Ident, slots: &[CultCacheSlot]) -> proc_macro2::Token } } - deserializer.deserialize_tuple(#max_slot + 1, CultCacheVisitor) + deserializer.deserialize_tuple(#max_slot + 1, __CultCacheVisitor) } } } @@ -424,6 +440,7 @@ fn union_codec(input: DeriveInput) -> syn::Result { let unknown_message = format!("unknown arm key {{}} for union {enum_name}; declared keys: {declared}"); let arity_message = format!("union {enum_name} is not a [key, slots] pair"); let non_integer_message = format!("union {enum_name} has a non-integer arm key; declared keys: {declared}"); + let nil_slots_message = format!("union {enum_name} arm {{}} has nil slots; declared keys: {declared}"); let out_of_range_message = format!("union {enum_name} has an arm key outside Int32; declared keys: {declared}"); let payload_checks = arms.iter().filter_map(|arm| arm.payload.as_ref()).map(|ty| { quote! { assert_slot_record::<#ty>(); } @@ -455,19 +472,37 @@ fn union_codec(input: DeriveInput) -> syn::Result { if let Some(ty) = &arm.payload { quote! { #key => { - let arm = seq - .next_element::<#ty>()? - .ok_or_else(|| ::serde::de::Error::custom(#arity_message))?; - #ident::#variant(arm) + // A nil in the slots position is not an empty record. + match seq.next_element::<::std::option::Option<#ty>>()? { + ::std::option::Option::Some(::std::option::Option::Some(arm)) => #ident::#variant(arm), + ::std::option::Option::Some(::std::option::Option::None) => { + return ::std::result::Result::Err(::serde::de::Error::custom( + format!(#nil_slots_message, #key), + )); + } + ::std::option::Option::None => { + return ::std::result::Result::Err(::serde::de::Error::custom(#arity_message)); + } + } } } } else { quote! { #key => { - seq.next_element::<[(); 0]>() + match seq + .next_element::<::std::option::Option<[(); 0]>>() .map_err(|_| ::serde::de::Error::custom(#arity_message))? - .ok_or_else(|| ::serde::de::Error::custom(#arity_message))?; - #ident::#variant + { + ::std::option::Option::Some(::std::option::Option::Some(_)) => #ident::#variant, + ::std::option::Option::Some(::std::option::Option::None) => { + return ::std::result::Result::Err(::serde::de::Error::custom( + format!(#nil_slots_message, #key), + )); + } + ::std::option::Option::None => { + return ::std::result::Result::Err(::serde::de::Error::custom(#arity_message)); + } + } } } } @@ -483,9 +518,9 @@ fn union_codec(input: DeriveInput) -> syn::Result { }; impl ::serde::Serialize for #ident { - fn serialize(&self, serializer: S) -> ::std::result::Result + fn serialize<__S>(&self, serializer: __S) -> ::std::result::Result<__S::Ok, __S::Error> where - S: ::serde::Serializer, + __S: ::serde::Serializer, { use ::serde::ser::SerializeTuple; let mut tuple = serializer.serialize_tuple(2)?; @@ -497,128 +532,137 @@ fn union_codec(input: DeriveInput) -> syn::Result { } impl<'de> ::serde::Deserialize<'de> for #ident { - fn deserialize(deserializer: D) -> ::std::result::Result + fn deserialize<__D>(deserializer: __D) -> ::std::result::Result where - D: ::serde::Deserializer<'de>, + __D: ::serde::Deserializer<'de>, { // The reference classifies the key before the arm: an integer in Int32, an // integer outside it, or not an integer. No input value is kept or echoed. - enum ArmKey { + enum __ArmKey { Int(i64), OutOfRange, NonInteger, } - struct ArmKeyProbe(ArmKey); - struct ArmKeyVisitor; + struct __ArmKeyProbe(__ArmKey); + struct __ArmKeyVisitor; - impl<'de> ::serde::de::Visitor<'de> for ArmKeyVisitor { - type Value = ArmKeyProbe; + impl<'de> ::serde::de::Visitor<'de> for __ArmKeyVisitor { + type Value = __ArmKeyProbe; fn expecting(&self, formatter: &mut ::std::fmt::Formatter) -> ::std::fmt::Result { formatter.write_str("an integer arm key") } - fn visit_i64(self, value: i64) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::Int(value))) + fn visit_i64<__E>(self, value: i64) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::Int(value))) } - fn visit_u64(self, value: u64) -> ::std::result::Result { - Ok(ArmKeyProbe(match i64::try_from(value) { - Ok(value) => ArmKey::Int(value), - Err(_) => ArmKey::OutOfRange, + fn visit_u64<__E>(self, value: u64) -> ::std::result::Result { + Ok(__ArmKeyProbe(match i64::try_from(value) { + Ok(value) => __ArmKey::Int(value), + Err(_) => __ArmKey::OutOfRange, })) } - fn visit_bool(self, _: bool) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_bool<__E>(self, _: bool) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_f64(self, _: f64) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_f64<__E>(self, _: f64) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_str(self, _: &str) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_str<__E>(self, _: &str) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_bytes(self, _: &[u8]) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_bytes<__E>(self, _: &[u8]) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_unit(self) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_unit<__E>(self) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_none(self) -> ::std::result::Result { - Ok(ArmKeyProbe(ArmKey::NonInteger)) + fn visit_none<__E>(self) -> ::std::result::Result { + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_some(self, deserializer: D) -> ::std::result::Result + fn visit_some<__D>(self, deserializer: __D) -> ::std::result::Result where - D: ::serde::Deserializer<'de>, + __D: ::serde::Deserializer<'de>, { <::serde::de::IgnoredAny as ::serde::Deserialize>::deserialize(deserializer)?; - Ok(ArmKeyProbe(ArmKey::NonInteger)) + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_seq(self, mut seq: A) -> ::std::result::Result + // A MessagePack ext value (fixext, timestamp, ext8) reaches the visitor here. + fn visit_newtype_struct<__D>(self, deserializer: __D) -> ::std::result::Result where - A: ::serde::de::SeqAccess<'de>, + __D: ::serde::Deserializer<'de>, + { + <::serde::de::IgnoredAny as ::serde::Deserialize>::deserialize(deserializer)?; + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) + } + + fn visit_seq<__A>(self, mut seq: __A) -> ::std::result::Result + where + __A: ::serde::de::SeqAccess<'de>, { while seq.next_element::<::serde::de::IgnoredAny>()?.is_some() {} - Ok(ArmKeyProbe(ArmKey::NonInteger)) + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } - fn visit_map(self, mut map: A) -> ::std::result::Result + fn visit_map<__A>(self, mut map: __A) -> ::std::result::Result where - A: ::serde::de::MapAccess<'de>, + __A: ::serde::de::MapAccess<'de>, { while map.next_entry::<::serde::de::IgnoredAny, ::serde::de::IgnoredAny>()?.is_some() {} - Ok(ArmKeyProbe(ArmKey::NonInteger)) + Ok(__ArmKeyProbe(__ArmKey::NonInteger)) } } - impl<'de> ::serde::Deserialize<'de> for ArmKeyProbe { - fn deserialize(deserializer: D) -> ::std::result::Result + impl<'de> ::serde::Deserialize<'de> for __ArmKeyProbe { + fn deserialize<__D>(deserializer: __D) -> ::std::result::Result where - D: ::serde::Deserializer<'de>, + __D: ::serde::Deserializer<'de>, { - deserializer.deserialize_any(ArmKeyVisitor) + deserializer.deserialize_any(__ArmKeyVisitor) } } - struct CultUnionVisitor; + struct __CultUnionVisitor; - impl<'de> ::serde::de::Visitor<'de> for CultUnionVisitor { + impl<'de> ::serde::de::Visitor<'de> for __CultUnionVisitor { type Value = #ident; fn expecting(&self, formatter: &mut ::std::fmt::Formatter) -> ::std::fmt::Result { formatter.write_str(concat!("a [key, slots] pair for union ", stringify!(#ident))) } - fn visit_seq(self, mut seq: A) -> ::std::result::Result + fn visit_seq<__A>(self, mut seq: __A) -> ::std::result::Result where - A: ::serde::de::SeqAccess<'de>, + __A: ::serde::de::SeqAccess<'de>, { if seq.size_hint().is_some_and(|len| len != 2) { return ::std::result::Result::Err(::serde::de::Error::custom(#arity_message)); } let key = match seq - .next_element::() + .next_element::<__ArmKeyProbe>() .map_err(|_| ::serde::de::Error::custom(#arity_message))? .ok_or_else(|| ::serde::de::Error::custom(#arity_message))? .0 { - ArmKey::Int(wide) => match i32::try_from(wide) { + __ArmKey::Int(wide) => match i32::try_from(wide) { Ok(key) => key, Err(_) => { return ::std::result::Result::Err(::serde::de::Error::custom(#out_of_range_message)); } }, - ArmKey::OutOfRange => { + __ArmKey::OutOfRange => { return ::std::result::Result::Err(::serde::de::Error::custom(#out_of_range_message)); } - ArmKey::NonInteger => { + __ArmKey::NonInteger => { return ::std::result::Result::Err(::serde::de::Error::custom(#non_integer_message)); } }; @@ -637,7 +681,7 @@ fn union_codec(input: DeriveInput) -> syn::Result { } } - deserializer.deserialize_tuple(2, CultUnionVisitor) + deserializer.deserialize_tuple(2, __CultUnionVisitor) } } }) @@ -738,4 +782,54 @@ mod tests { let message = union_error("enum E { #[cultcache(key = \"1\")] A(Arm) }"); assert!(message.contains("integer literal"), "{message}"); } + fn entry_result(source: &str) -> syn::Result { + database_entry_codec(syn::parse_str(source).expect("the fixture parses")) + } + + fn slots_result(source: &str) -> syn::Result { + cult_slots_codec(syn::parse_str(source).expect("the fixture parses")) + } + + // Each attribute names a thing once; the second would silently win otherwise. + #[test] + fn a_repeated_type_or_schema_attribute_is_refused_for_that_reason() { + for source in [ + "#[cultcache(type = \"a.one\")] #[cultcache(type = \"a.two\")] struct S { #[cultcache(key = 0)] id: String }", + "#[cultcache(type = \"a.one\", type = \"a.two\")] struct S { #[cultcache(key = 0)] id: String }", + "#[cultcache(type = \"a.one\", schema = \"s.one\")] #[cultcache(schema = \"s.two\")] struct S { #[cultcache(key = 0)] id: String }", + "#[cultcache(type = \"a.one\", schema = \"s.one\", schema = \"s.two\")] struct S { #[cultcache(key = 0)] id: String }", + ] { + let message = entry_result(source).expect_err(source).to_string(); + assert!(message.contains("repeated `type`") || message.contains("repeated `schema`"), "{source}: {message}"); + } + entry_result("#[cultcache(type = \"a.one\", schema = \"s.one\")] struct S { #[cultcache(key = 0)] id: String }") + .expect("one type and one schema are accepted"); + } + + #[test] + fn a_generic_slots_struct_and_a_missing_type_are_refused_for_their_reasons() { + let message = slots_result("struct S { #[cultcache(key = 0)] id: T }").unwrap_err().to_string(); + assert!(message.contains("does not support generics"), "{message}"); + let message = entry_result("struct S { #[cultcache(key = 0)] id: String }").unwrap_err().to_string(); + assert!(message.contains("requires #[cultcache(type"), "{message}"); + } + + // A slot record is a dense array: keys stop at the largest array16 length. + #[test] + fn the_slot_key_domain_ends_at_65535() { + slots_error_free("struct S { #[cultcache(key = 0)] a: String, #[cultcache(key = 65535)] b: String }"); + for source in [ + "struct S { #[cultcache(key = 65536)] a: String }", + "struct S { #[cultcache(key = 4294967296)] a: String }", + "struct S { #[cultcache(key = 0x10000)] a: String }", + ] { + let message = slots_error(source); + assert!(message.contains("outside 0..=65535"), "{source}: {message}"); + } + } + + fn slots_error_free(source: &str) { + let input: DeriveInput = syn::parse_str(source).expect("the fixture parses"); + named_slots(input.data, &input.ident, "CultSlots").expect("the fixture is accepted"); + } } diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 0423c224..5332e8a4 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -99,6 +99,14 @@ //! enum Bad { #[cultcache(key = 4)] A(Option) } //! ``` //! +//! A hand-written `SlotRecord` impl, which only the derives provide: +//! +//! ```compile_fail,E0046 +//! use cultcache_rs::SlotRecord; +//! struct Sneaky(String); +//! impl SlotRecord for Sneaky {} +//! ``` +//! //! A CultSlots field without a key, a repeated slot, a repeated key attribute, and a generic struct: //! //! ```compile_fail @@ -163,7 +171,11 @@ pub trait DatabaseEntry: Serialize + DeserializeOwned + Clone + Send + 'static { label = "a CultUnion arm is a newtype over a #[derive(CultSlots)] struct", note = "a String, an Option or any other type is not written as an array of slots" )] -pub trait SlotRecord {} +pub trait SlotRecord { + /// Implemented only by the derives, which write the positional array the contract names. + #[doc(hidden)] + fn __slots_array(); +} pub trait CultCacheRegistry { fn register_entries(&self, cache: &mut CultCache) -> Result<()>; @@ -484,6 +496,124 @@ fn describe_undecodable(error: &anyhow::Error) -> String { parts.join(": ") } +/// The deepest a document payload may nest, on read and on write, in every CultCache runtime: every +/// MessagePack array and map counts, read by a deserializer or skipped past it, and the outermost +/// array is depth 1. A payload of this depth is accepted and one container deeper is refused. It is +/// the C# reference's `CultMessagePackSecurity.MaximumPayloadDepth`. +pub const MAX_PAYLOAD_DEPTH: usize = 256; + +/// A payload nests more containers than [`MAX_PAYLOAD_DEPTH`]. It names the depth reached and the +/// limit, never payload bytes, and can be recovered from the `anyhow::Error` a read or write returns. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PayloadTooDeep { + pub depth: usize, + pub limit: usize, +} + +impl std::fmt::Display for PayloadTooDeep { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "payload nests {} MessagePack containers deep; the limit is {}", + self.depth, self.limit + ) + } +} + +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. +pub fn check_payload_depth(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(()); + }; + position = position.saturating_add(advance); + if let Some(children) = children { + if depth == MAX_PAYLOAD_DEPTH { + return Err(PayloadTooDeep { depth: MAX_PAYLOAD_DEPTH + 1, limit: MAX_PAYLOAD_DEPTH }); + } + if children > 0 { + remaining[depth] = children; + depth += 1; + continue; + } + } + // One element is complete; close every container it was the last element of. + while depth > 0 { + remaining[depth - 1] -= 1; + if remaining[depth - 1] != 0 { + break; + } + depth -= 1; + } + if depth == 0 { + return Ok(()); + } + } + Ok(()) +} + +/// The element at `position`: how many bytes to advance past it (its whole length for a scalar, its +/// header for a container) and, for a container, how many child elements follow. `None` when the +/// payload ends inside the header. +fn element_header(bytes: &[u8], position: usize) -> Option<(usize, Option)> { + // The big-endian length after the code byte. + let length = |width: usize| -> Option { + let field = bytes.get(position + 1..position + 1 + width)?; + Some(field.iter().fold(0usize, |length, byte| (length << 8) | usize::from(*byte))) + }; + let code = bytes[position]; + Some(match code { + 0x00..=0x7f | 0xe0..=0xff | 0xc0..=0xc3 => (1, None), + 0x80..=0x8f => (1, Some(2 * usize::from(code & 0x0f))), + 0x90..=0x9f => (1, Some(usize::from(code & 0x0f))), + 0xa0..=0xbf => (1 + usize::from(code & 0x1f), None), + 0xc4 | 0xd9 => (2usize.saturating_add(length(1)?), None), + 0xc5 | 0xda => (3usize.saturating_add(length(2)?), None), + 0xc6 | 0xdb => (5usize.saturating_add(length(4)?), None), + 0xc7 => (3usize.saturating_add(length(1)?), None), + 0xc8 => (4usize.saturating_add(length(2)?), None), + 0xc9 => (6usize.saturating_add(length(4)?), None), + 0xca | 0xce | 0xd2 => (5, None), + 0xcb | 0xcf | 0xd3 => (9, None), + 0xcc | 0xd0 => (2, None), + 0xcd | 0xd1 | 0xd4 => (3, None), + 0xd5 => (4, None), + 0xd6 => (6, None), + 0xd7 => (10, None), + 0xd8 => (18, None), + 0xdc => (3, Some(length(2)?)), + 0xdd => (5, Some(length(4)?)), + 0xde => (3, Some(length(2)?.saturating_mul(2))), + 0xdf => (5, Some(length(4)?.saturating_mul(2))), + }) +} + +/// The one decode of a document payload: the nesting bound first, then the deserializer, whose own +/// 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. +pub fn decode_payload(payload: &[u8]) -> Result { + check_payload_depth(payload)?; + rmp_serde::from_slice(payload).map_err(opaque_decode_error) +} + +/// 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(|| { + format!( + "refused CultCache envelope {:?} at key {:?}", + entry.r#type, entry.key + ) + }) +} + /// A decoder failure as an error that keeps only its kind. Every decode of stored bytes /// goes through this or [`describe_undecodable`], so no error text quotes a stored value. fn opaque_decode_error(error: rmp_serde::decode::Error) -> anyhow::Error { @@ -2420,7 +2550,7 @@ impl CultCache { let Some(entry) = self.entries.get(&entry_id_parts(T::TYPE, key)) else { return Ok(None); }; - let payload = rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { + let payload = decode_payload(&entry.payload).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2453,7 +2583,7 @@ impl CultCache { if entry.r#type != T::TYPE { continue; } - values.push(rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { + values.push(decode_payload(&entry.payload).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2474,7 +2604,7 @@ impl CultCache { } values.push(( entry.key.clone(), - rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { + decode_payload(&entry.payload).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2556,7 +2686,7 @@ impl CultCache { key: String, payload: Vec, ) -> Result<(CultCacheEnvelope, T)> { - let parsed: T = rmp_serde::from_slice(&payload).map_err(opaque_decode_error).with_context(|| { + let parsed: T = decode_payload(&payload).with_context(|| { format!( "failed to validate CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2599,6 +2729,7 @@ impl CultCache { entry.r#type )); } + require_payload_depth(entry)?; let Some(home) = self.home_index(&entry.r#type)? else { continue; }; @@ -2643,7 +2774,7 @@ impl CultCache { /// claiming compile-time knowledge of its payload schema. This is the /// CultMesh/schema-registry boundary: callers own payload validation before /// admission; CultCache still enforces registered type, identity, timestamp, - /// routing, and persistence. + /// the payload nesting bound ([`MAX_PAYLOAD_DEPTH`]), routing, and persistence. pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> { if !self.definitions.contains_key(&entry.r#type) { return Err(anyhow!( @@ -2663,6 +2794,7 @@ impl CultCache { entry.r#type )); } + require_payload_depth(&entry)?; if let Some(home) = self.home_index(&entry.r#type)? { self.stores[home].store.push(&entry)?; } @@ -2692,7 +2824,7 @@ impl CultCache { )); } - rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { + decode_payload(&entry.payload).with_context(|| { format!( "failed to validate CultCache envelope {:?} at key {:?} as {}", T::TYPE, diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index 1f0cdf31..da2b1a08 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -4,6 +4,11 @@ use std::path::PathBuf; use cultcache_rs::CultCache; +use cultcache_rs::CacheBackingStore; +use cultcache_rs::CultCacheEnvelope; +use cultcache_rs::MAX_PAYLOAD_DEPTH; +use cultcache_rs::PayloadTooDeep; +use cultcache_rs::check_payload_depth; use cultcache_rs::CultSlots; use cultcache_rs::CultUnion; use cultcache_rs::DatabaseEntry; @@ -19,6 +24,14 @@ enum Node { Use(UseArm), #[cultcache(key = 5)] Image(ImageArm), + #[cultcache(key = 6)] + Wrap(WrapArm), +} + +#[derive(Clone, Debug, PartialEq, CultSlots)] +struct WrapArm { + #[cultcache(key = 0)] + inner: Option>, } #[derive(Clone, Debug, PartialEq, CultSlots)] @@ -74,6 +87,10 @@ enum Outcome { Completed(CompletedArm), #[cultcache(key = 2)] Denied, + #[cultcache(key = -2147483648)] + Floor, + #[cultcache(key = 2147483647)] + Ceiling, } #[derive(Clone, Debug, PartialEq, CultSlots)] @@ -243,6 +260,15 @@ fn union_refusal_vectors() { assert_eq!((refused, read_only), (5, 4)); } +// A nil where the arm's slots belong is not an empty record, for an arm with slots and a unit arm. +#[test] +fn nil_slots_are_refused_naming_the_union_and_the_key() { + for (bytes, key) in [(vec![0x92, 0x01, 0xc0], "1"), (vec![0x92, 0x02, 0xc0], "2")] { + let message = rmp_serde::from_slice::(&bytes).unwrap_err().to_string(); + assert!(message.contains("Outcome") && message.contains(&format!("arm {key} has nil slots")), "{message}"); + } +} + #[test] fn an_unknown_arm_lists_the_declared_keys_and_never_echoes_the_payload() { // Shaped like a real arm, an array of strings, under an undeclared key. @@ -393,7 +419,11 @@ mod arm_keys { #[test] fn a_key_that_is_not_an_integer_is_its_own_class_and_never_echoed() { - let keys: [&[u8]; 7] = [ + let keys: [&[u8]; 10] = [ + // A MessagePack ext: fixext 1, the timestamp extension and ext 8. + &[0xd4, 0x01, 0x00], + &[0xd6, 0xff, 0x00, 0x00, 0x00, 0x00], + &[0xc7, 0x02, 0x05, b'C', b'A'], &[0xc0], &[0xc3], &[0xcb, 0x3f, 0xf0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], @@ -467,3 +497,449 @@ mod serde_enum_bytes { assert_eq!(rmp_serde::from_slice::(AT_BASE).unwrap(), value); } } + +// Names a derive's generated code must not capture: the generic parameters and the local items of the +// expansion, used as user type names. +mod generated_names { + use cultcache_rs::CultSlots; + use cultcache_rs::CultUnion; + use cultcache_rs::DatabaseEntry; + + #[derive(Clone, Debug, PartialEq, CultSlots)] + struct D { + #[cultcache(key = 0)] + name: String, + } + + #[derive(Clone, Debug, PartialEq, CultSlots)] + struct S { + #[cultcache(key = 0)] + count: i32, + } + + #[derive(Clone, Debug, PartialEq, CultSlots)] + struct ArmKey { + #[cultcache(key = 0)] + id: i32, + } + + #[derive(Clone, Debug, PartialEq, CultUnion)] + enum A { + #[cultcache(key = 1)] + D(D), + #[cultcache(key = 2)] + S(S), + #[cultcache(key = 3)] + ArmKey(ArmKey), + #[cultcache(key = 4)] + E, + } + + #[derive(Clone, Debug, PartialEq, CultUnion)] + enum CultUnionVisitor { + #[cultcache(key = 1)] + A(ArmKey), + } + + #[derive(Clone, Debug, PartialEq, DatabaseEntry)] + #[cultcache(type = "tests.names")] + struct E { + #[cultcache(key = 0)] + union: A, + #[cultcache(key = 1)] + other: CultUnionVisitor, + } + + #[test] + fn types_named_like_generated_generics_and_locals_round_trip() { + let value = E { + union: A::D(D { name: "n".into() }), + other: CultUnionVisitor::A(ArmKey { id: 9 }), + }; + let bytes = rmp_serde::to_vec(&value).unwrap(); + assert_eq!(rmp_serde::from_slice::(&bytes).unwrap(), value); + for union in [A::S(S { count: 2 }), A::ArmKey(ArmKey { id: 3 }), A::E] { + let bytes = rmp_serde::to_vec(&union).unwrap(); + assert_eq!(rmp_serde::from_slice::(&bytes).unwrap(), union); + } + } +} + +// The reference's CultUnionGuard classes, from bounds-outcomes.msgpack; depth entries go through the +// cache because the nesting rule lives in its one decode. +const CANARY: &str = "CANARY-union-golden-4c2a"; + +fn envelope(key: &str, payload: &[u8]) -> CultCacheEnvelope { + CultCacheEnvelope { + key: key.into(), + r#type: UnionSurface::TYPE.into(), + payload: payload.to_vec(), + stored_at: "2026-01-01T00:00:00Z".into(), + schema_id: None, + } +} + +// 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. +fn through_cache(payload: &[u8]) -> Result { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("foreign.cc"); + SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", payload)).unwrap(); + 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(); + match cache.get::("k") { + Ok(value) => Ok(value.expect("the envelope was stored")), + Err(error) => Err(format!("{error:#}")), + } +} + +#[test] +fn bounds_vectors() { + let entries: Vec = rmp_serde::from_slice(&vector("bounds-outcomes.msgpack")).unwrap(); + assert_eq!(entries[0], serde_json::json!(["limit", 256])); + assert_eq!(MAX_PAYLOAD_DEPTH, 256); + let (mut accepted, mut refused, mut read_only) = (0, 0, 0); + for entry in &entries[1..] { + let fields: Vec<&str> = entry.as_array().unwrap().iter().filter_map(|field| field.as_str()).collect(); + let (file, outcome, union, detail) = (fields[0], fields[1], fields[2], fields[3]); + let bytes = vector(file); + let result = if union == "UnionSurface" { through_cache(&bytes) } else { decode(&bytes) }; + match outcome { + "accepted" => { + accepted += 1; + let value = result.unwrap_or_else(|message| panic!("{file}: {message}")); + assert_eq!(rmp_serde::to_vec(&value).unwrap(), vector(detail), "{file}"); + match file { + "arm-key-int32-min.msgpack" => assert_eq!(value.outcomes.last(), Some(&Outcome::Floor)), + "arm-key-int32-max.msgpack" | "arm-key-wide-coding.msgpack" => { + assert_eq!(value.outcomes.last(), Some(&Outcome::Ceiling)) + } + _ => {} + } + } + "read-only" => { + read_only += 1; + assert_eq!(result.unwrap_or_else(|message| panic!("{file}: {message}")), constructed(), "{file}"); + } + "refused" => { + refused += 1; + let message = result.expect_err(file); + // The cache path names the entry type in its context; the union names itself. + let names = if union == "UnionSurface" { UnionSurface::TYPE } else { union }; + assert!(message.contains(names), "{file}: {message}"); + let class = match detail { + "key-out-of-range" => "outside Int32", + "key-not-integer" => "non-integer", + "nil-slots" => "arm 1 has nil slots", + "depth" => "the limit is 256", + other => panic!("{file}: unknown class {other}"), + }; + assert!(message.contains(class), "{file}: {message}"); + assert!(!message.contains(CANARY), "{file} echoed a value: {message}"); + } + other => panic!("{file}: unknown outcome {other}"), + } + } + assert_eq!((accepted, refused, read_only), (4, 7, 1)); +} + +fn wrap_chain(wraps: usize, leaf: Node) -> UnionSurface { + let mut node = leaf; + for _ in 0..wraps { + node = Node::Wrap(WrapArm { inner: Some(std::boxed::Box::new(node)) }); + } + UnionSurface { root: Some(node), ..constructed() } +} + +// Each Wrap adds a pair and a slots array; the surface array is depth 1, so a Box leaf (pair, slots, +// children) ends at 2 * (wraps + 1) + 2 and a Text leaf at 2 * (wraps + 1) + 1. +#[test] +fn the_constructed_chains_write_the_vector_bytes() { + let at_limit = wrap_chain(126, boxed("leaf", vec![])); + assert_eq!(rmp_serde::to_vec(&at_limit).unwrap(), vector("depth-at-limit.msgpack")); + let past_limit = wrap_chain(127, text("leaf", CANARY)); + assert_eq!(rmp_serde::to_vec(&past_limit).unwrap(), vector("depth-past-limit.msgpack")); +} + +#[test] +fn the_canary_is_in_the_committed_bytes() { + let bytes = vector("depth-past-limit.msgpack"); + assert!(bytes.windows(CANARY.len()).any(|window| window == CANARY.as_bytes())); +} + +#[test] +fn write_refused_past_limit() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("depth.cc"); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + + let at_limit = wrap_chain(126, boxed("leaf", vec![])); + assert_eq!(cache.put("at", &at_limit).unwrap(), at_limit); + let stored = cache.snapshot_envelopes(); + assert_eq!(stored.len(), 1); + assert_eq!(stored[0].payload, vector("depth-at-limit.msgpack")); + let file_before = std::fs::read(&path).unwrap(); + + let error = cache.put("past", &wrap_chain(127, text("leaf", CANARY))).unwrap_err(); + assert_eq!(error.downcast_ref::(), Some(&PayloadTooDeep { depth: 257, limit: 256 })); + assert!(!format!("{error:#}").contains(CANARY), "{error:#}"); + assert_eq!(cache.get::("past").unwrap(), None); + assert_eq!(cache.snapshot_envelopes(), stored); + assert_eq!(std::fs::read(&path).unwrap(), file_before); +} + +// Every door an envelope enters the store by holds the one bound: the typed put above, and the raw +// doors a peer's bytes arrive through. Three nestings: union arms and maps +// written by the production encoder, and the arrays of the committed vector. +#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] +struct MapNest { + inner: Option>, +} + +#[derive(Clone, Debug, PartialEq, DatabaseEntry)] +#[cultcache(type = "tests.map_doc", schema = "tests.map_doc.v1")] +struct MapDoc { + #[cultcache(key = 0)] + root: Option, +} + +// The document array is depth 1 and each MapNest is one map. +fn map_doc(maps: usize) -> MapDoc { + let mut nest = MapNest { inner: None }; + for _ in 1..maps { + nest = MapNest { inner: Some(std::boxed::Box::new(nest)) }; + } + MapDoc { root: Some(nest) } +} + +#[test] +fn maps_count_one_level_each_on_every_write_door() { + let at_limit = map_doc(MAX_PAYLOAD_DEPTH - 1); + let past_limit = map_doc(MAX_PAYLOAD_DEPTH); + let at_bytes = rmp_serde::to_vec_named(&at_limit).unwrap(); + let past_bytes = rmp_serde::to_vec_named(&past_limit).unwrap(); + assert_eq!(check_payload_depth(&at_bytes), Ok(())); + assert_eq!(check_payload_depth(&past_bytes), Err(TOO_DEEP)); + + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + let (prepared, parsed) = cache.prepare_entry_named("at", &at_limit).unwrap(); + assert_eq!((prepared.payload.as_slice(), &parsed), (at_bytes.as_slice(), &at_limit)); + cache.put_prepared_batch(vec![prepared]).unwrap(); + assert_eq!(cache.get::("at").unwrap(), Some(at_limit)); + let error = cache.prepare_entry_named("past", &past_limit).unwrap_err(); + assert_eq!(error.downcast_ref::(), Some(&TOO_DEEP)); +} + +fn raw(type_name: &str, key: &str, payload: Vec) -> CultCacheEnvelope { + CultCacheEnvelope { r#type: type_name.into(), ..envelope(key, &payload) } +} + +#[test] +fn the_raw_doors_hold_the_bound_for_arms_maps_and_arrays() { + let arms_at = rmp_serde::to_vec(&wrap_chain(126, boxed("leaf", vec![]))).unwrap(); + let arms_past = rmp_serde::to_vec(&wrap_chain(127, text("leaf", CANARY))).unwrap(); + let maps_at = rmp_serde::to_vec_named(&map_doc(MAX_PAYLOAD_DEPTH - 1)).unwrap(); + let maps_past = rmp_serde::to_vec_named(&map_doc(MAX_PAYLOAD_DEPTH)).unwrap(); + let arrays_at = vector("depth-at-limit.msgpack"); + let arrays_past = vector("depth-past-limit.msgpack"); + let cases = [ + (UnionSurface::TYPE, "arms", arms_at, arms_past), + (MapDoc::TYPE, "maps", maps_at, maps_past), + (UnionSurface::TYPE, "arrays", arrays_at, arrays_past), + ]; + for (type_name, name, at, past) in cases { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("doors.cc"); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + cache.register_entry_type::().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + cache.put_raw_envelope(raw(type_name, "at-raw", at.clone())).unwrap(); + cache.put_prepared_batch(vec![raw(type_name, "at-batch", at.clone())]).unwrap(); + let stored = cache.snapshot_envelopes(); + let file_before = std::fs::read(&path).unwrap(); + + let refusals = [ + cache.put_raw_envelope(raw(type_name, "past-raw", past.clone())).unwrap_err(), + // The valid entry staged beside it must not land either. + cache + .put_prepared_batch(vec![raw(type_name, "ok-staged", at.clone()), raw(type_name, "past-batch", past.clone())]) + .unwrap_err(), + ]; + for error in refusals { + assert_eq!(error.downcast_ref::(), Some(&TOO_DEEP), "{name}"); + let message = format!("{error:#}"); + assert!(message.contains(type_name) && !message.contains(CANARY), "{name}: {message}"); + } + assert_eq!(cache.snapshot_envelopes(), stored, "{name}"); + assert_eq!(std::fs::read(&path).unwrap(), file_before, "{name}"); + } +} + +// A debug build on a 2 MiB thread carries 405 containers of this document; the bound keeps every +// accepted payload below that, and every refusal is an error, not an abort. +#[test] +fn stack_budget_debug_2mib() { + std::thread::Builder::new() + .stack_size(2 * 1024 * 1024) + .spawn(|| { + assert_eq!( + through_cache(&vector("depth-at-limit.msgpack")).unwrap(), + wrap_chain(126, boxed("leaf", vec![])) + ); + let refused = through_cache(&vector("depth-past-limit.msgpack")).unwrap_err(); + assert!(refused.contains("the limit is 256"), "{refused}"); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + let value = wrap_chain(126, boxed("leaf", vec![])); + assert_eq!(cache.put("at", &value).unwrap(), value); + assert_eq!(cache.get::("at").unwrap(), Some(value)); + let mut deep = vec![0x91u8; 100_000]; + deep.push(0xc0); + let refused = through_cache(&deep).unwrap_err(); + assert!(refused.contains("the limit is 256"), "{refused}"); + }) + .unwrap() + .join() + .unwrap(); +} + +// The walk is iterative: 100000 nested arrays are refused on a 256 KiB stack. +#[test] +fn the_walk_does_not_recurse() { + std::thread::Builder::new() + .stack_size(256 * 1024) + .spawn(|| { + let mut deep = vec![0x91u8; 100_000]; + deep.push(0xc0); + assert_eq!(check_payload_depth(&deep), Err(PayloadTooDeep { depth: 257, limit: 256 })); + }) + .unwrap() + .join() + .unwrap(); +} + +fn nested(open: &[u8], levels: usize, leaf: &[u8]) -> Vec { + let mut bytes = open.repeat(levels); + bytes.extend_from_slice(leaf); + bytes +} + +const TOO_DEEP: PayloadTooDeep = PayloadTooDeep { depth: 257, limit: 256 }; + +// Every container header width counts one level, and the bound sits between 256 and 257 for each. +#[test] +fn every_container_header_counts_one_level() { + let openers: [(&str, &[u8]); 6] = [ + ("fixarray", &[0x91]), + ("array16", &[0xdc, 0x00, 0x01]), + ("array32", &[0xdd, 0x00, 0x00, 0x00, 0x01]), + ("fixmap", &[0x81, 0x01]), + ("map16", &[0xde, 0x00, 0x01, 0x01]), + ("map32", &[0xdf, 0x00, 0x00, 0x00, 0x01, 0x01]), + ]; + for (name, open) in openers { + assert_eq!(check_payload_depth(&nested(open, 256, &[0xc0])), Ok(()), "{name} at the limit"); + assert_eq!(check_payload_depth(&nested(open, 257, &[0xc0])), Err(TOO_DEEP), "{name} past the limit"); + } + // An empty container still counts the level it sits on. + assert_eq!(check_payload_depth(&nested(&[0x91], 255, &[0x90])), Ok(())); + assert_eq!(check_payload_depth(&nested(&[0x91], 256, &[0x90])), Err(TOO_DEEP)); + assert_eq!(check_payload_depth(&nested(&[0x91], 255, &[0x80])), Ok(())); + assert_eq!(check_payload_depth(&nested(&[0x91], 256, &[0x80])), Err(TOO_DEEP)); +} + +// Every scalar and container header is followed by the chain that tests the bound. The skipped bytes +// are filled with the array marker, so a skip that stops short reads them as containers and a skip +// that runs long eats the chain: either way the chain's depth stops reading as 256 and 257. +#[test] +fn every_element_is_skipped_by_its_length_and_counted_by_its_children() { + const FILL: u8 = 0x91; + let filled = |head: &[u8], bytes: usize| -> Vec { [head.to_vec(), vec![FILL; bytes]].concat() }; + let long = |code: u8, width: usize, length: usize, extra: usize| -> Vec { + let size = length.to_be_bytes(); + filled(&[&[code][..], &size[size.len() - width..]].concat(), length + extra) + }; + let mut elements: Vec> = vec![ + vec![0x00], + vec![0xff], + vec![0xc0], + vec![0xc2], + vec![0xc3], + filled(&[0xcc], 1), + filled(&[0xcd], 2), + filled(&[0xce], 4), + filled(&[0xcf], 8), + filled(&[0xd0], 1), + filled(&[0xd1], 2), + filled(&[0xd2], 4), + filled(&[0xd3], 8), + filled(&[0xca], 4), + filled(&[0xcb], 8), + filled(&[0xa3], 3), + filled(&[0xbf], 31), + filled(&[0xd4], 2), + filled(&[0xd5], 3), + filled(&[0xd6], 5), + filled(&[0xd7], 9), + filled(&[0xd8], 17), + ]; + for (code, width) in [(0xd9, 1), (0xc4, 1), (0xda, 2), (0xc5, 2), (0xdb, 4), (0xc6, 4)] { + for length in [3, 200, 256, 65536] { + if length < (1usize << (8 * width)) { + elements.push(long(code, width, length, 0)); + } + } + } + // An ext carries a type byte after its length. + for (code, width) in [(0xc7, 1), (0xc8, 2), (0xc9, 4)] { + for length in [3, 200, 256, 65536] { + if length < (1usize << (8 * width)) { + elements.push(long(code, width, length, 1)); + } + } + } + // Containers whose children are scalars, so a wrong child count shifts everything after them. + elements.push(vec![0x92, 0x00, 0x00]); + elements.push(vec![0x93, 0x00, 0x00, 0x00]); + elements.push(vec![0x82, 0x00, 0x00, 0x00, 0x00]); + elements.push(vec![0x83, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]); + elements.push([vec![0xdc, 0x01, 0x00], vec![0x00; 256]].concat()); + elements.push([vec![0xdd, 0x00, 0x00, 0x00, 0x03], vec![0x00; 3]].concat()); + elements.push([vec![0xde, 0x00, 0x03], vec![0x00; 6]].concat()); + elements.push([vec![0xde, 0x01, 0x00], vec![0x00; 512]].concat()); + elements.push([vec![0xdf, 0x00, 0x00, 0x00, 0x03], vec![0x00; 6]].concat()); + elements.push(vec![0xdc, 0x00, 0x00]); + elements.push(vec![0xdf, 0x00, 0x00, 0x00, 0x00]); + + let with_chain = |levels: usize| { + let mut bytes = vec![0xdd]; + bytes.extend_from_slice(&u32::try_from(elements.len() + 1).unwrap().to_be_bytes()); + for element in &elements { + bytes.extend_from_slice(element); + } + bytes.extend(nested(&[0x91], levels, &[0xc0])); + bytes + }; + assert_eq!(check_payload_depth(&with_chain(255)), Ok(())); + assert_eq!(check_payload_depth(&with_chain(256)), Err(TOO_DEEP)); +} + +// An ending cut short is the decoder's to refuse, never a panic or a depth error. +#[test] +fn a_payload_cut_short_is_not_a_depth_error() { + for bytes in [&[0xdc, 0x00][..], &[0xdd, 0, 0], &[0xd9], &[0xc5, 0x00], &[0xde], &[0xdf, 0, 0, 0], &[0x91], &[]] { + assert_eq!(check_payload_depth(bytes), Ok(()), "{bytes:02x?}"); + } + assert_eq!(check_payload_depth(&[0xdb, 0xff, 0xff, 0xff, 0xff, b'a']), Ok(())); +} + +#[test] +fn the_refusal_names_depth_and_limit_only() { + assert_eq!(TOO_DEEP.to_string(), "payload nests 257 MessagePack containers deep; the limit is 256"); +} diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 1af0d906..6965e2d5 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -1733,6 +1733,114 @@ mod tests { Ok(()) } + #[derive(serde::Serialize)] + struct Nest { + inner: Option>, + } + + // `levels` maps deep, written by the serializer the cache itself uses. + fn nest(levels: usize) -> Vec { + let mut value = Nest { inner: None }; + for _ in 1..levels { + value = Nest { inner: Some(Box::new(value)) }; + } + rmp_serde::to_vec_named(&value).unwrap() + } + + fn peer_document(key: &str, payload: Vec) -> cultnet_rs::CultNetRawDocumentRecord { + cultnet_rs::CultNetRawDocumentRecord { + schema_id: "cultmesh.test.note.v0".to_string(), + record_key: key.to_string(), + stored_at: "2026-07-03T00:00:00Z".to_string(), + payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + } + } + + // Serves `documents` as the catalog's snapshot until the returned flag is set. + fn serve_snapshot( + documents: Vec, + ) -> Result<( + SocketAddr, + std::sync::Arc, + thread::JoinHandle>, + )> { + let socket = UdpSocket::bind("127.0.0.1:0")?; + let target = socket.local_addr()?; + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let server_stop = stop.clone(); + let server = thread::spawn(move || -> Result<()> { + let mut server = CultMeshRudpDocumentServer::new( + socket, + |_receipt: CultMeshRudpRawDocumentReceipt| Ok(()), + move |_query: &CultMeshRudpSnapshotQuery| Ok(documents.clone()), + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), + )?; + while !server_stop.load(std::sync::atomic::Ordering::SeqCst) { + server.poll_once()?; + thread::sleep(Duration::from_millis(1)); + } + Ok(()) + }); + Ok((target, stop, server)) + } + + fn pull( + node: &mut CultMeshNode, + documents: Vec, + ) -> Result { + let (target, stop, server) = serve_snapshot(documents)?; + let pulled = node.pull_rudp_catalog_snapshot(CultMeshRudpSnapshotOptions { + target, + runtime_id: "pull-test".into(), + message_id: String::new(), + connect_timeout: Duration::from_millis(500), + response_timeout: Duration::from_secs(2), + poll_interval: Duration::from_millis(2), + resend_delay_ms: 5, + schema_ids: None, + record_keys: None, + }); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + server.join().expect("server thread should not panic")?; + pulled + } + + #[test] + fn a_pulled_peer_payload_nested_past_the_bound_is_refused_and_nothing_is_persisted() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cultmesh.cc"); + let mut node = CultMesh::create_node(&store_path, TestDocuments, Default::default())?; + + let limit = cultcache_rs::MAX_PAYLOAD_DEPTH; + let pulled = pull( + &mut node, + vec![peer_document("a", nest(limit)), peer_document("b", nest(limit))], + )?; + assert_eq!(pulled, 2); + node.flush()?; + let held = std::fs::read(&store_path)?; + + let error = pull(&mut node, vec![peer_document("past", nest(limit + 1))]).unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&cultcache_rs::PayloadTooDeep { depth: limit + 1, limit }) + ); + node.flush()?; + assert_eq!(std::fs::read(&store_path)?, held); + let reloaded = CultMesh::create_node(&store_path, TestDocuments, Default::default())?; + let mut keys: Vec<_> = + reloaded.cache().snapshot().into_iter().map(|entry| entry.key).collect(); + keys.sort(); + assert_eq!(keys, vec!["a".to_string(), "b".to_string()]); + Ok(()) + } + #[test] fn stream_catalog_negotiates_and_publishes_latest_shared_memory_frame() -> Result<()> { let mut catalog = CultMesh::create_stream_catalog(); diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index 1808c62f..6d9a2d77 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -604,11 +604,7 @@ pub fn decode_cultnet_message_from_slice( bytes: &[u8], contract: CultNetWireContract, ) -> Result { - if contract == CultNetWireContract::CultNetSchemaV0 { - let wire_value: rmpv::Value = rmp_serde::from_slice(bytes)?; - return parse_cultnet_message(&wire_value, contract); - } - let wire_value: rmpv::Value = rmp_serde::from_slice(bytes)?; + let wire_value: rmpv::Value = cultcache_rs::decode_payload(bytes)?; parse_cultnet_message(&wire_value, contract) } diff --git a/packages/cultnet-rs/src/media_stream_wire.rs b/packages/cultnet-rs/src/media_stream_wire.rs index 5c89a8c3..6b3d0d85 100644 --- a/packages/cultnet-rs/src/media_stream_wire.rs +++ b/packages/cultnet-rs/src/media_stream_wire.rs @@ -211,31 +211,31 @@ pub fn decode_media_wire_record(payload: &[u8]) -> Result { let record: GameCultMediaVideoAccessUnitRecord = - rmp_serde::from_slice(&document.payload)?; + cultcache_rs::decode_payload(&document.payload)?; validate_video_record(&record)?; GameCultMediaWireRecord::Video(record) } GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA => { let record: GameCultMediaVideoParityShardRecord = - rmp_serde::from_slice(&document.payload)?; + cultcache_rs::decode_payload(&document.payload)?; validate_video_parity_record(&record)?; GameCultMediaWireRecord::VideoParity(record) } GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA => { let record: GameCultMediaAudioPacketRecord = - rmp_serde::from_slice(&document.payload)?; + cultcache_rs::decode_payload(&document.payload)?; validate_audio_record(&record)?; GameCultMediaWireRecord::Audio(record) } GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA => { let record: GameCultMediaAudioParityShardRecord = - rmp_serde::from_slice(&document.payload)?; + cultcache_rs::decode_payload(&document.payload)?; validate_audio_parity_record(&record)?; GameCultMediaWireRecord::AudioParity(record) } GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA => { let record: GameCultMediaReceiverFeedbackRecord = - rmp_serde::from_slice(&document.payload)?; + cultcache_rs::decode_payload(&document.payload)?; validate_feedback_record(&record)?; GameCultMediaWireRecord::Feedback(record) } diff --git a/packages/cultnet-rs/src/provider_session.rs b/packages/cultnet-rs/src/provider_session.rs index 9965b0ef..2584120f 100644 --- a/packages/cultnet-rs/src/provider_session.rs +++ b/packages/cultnet-rs/src/provider_session.rs @@ -325,7 +325,7 @@ pub fn decode_provider_connect_evidence( "Provider Connect evidence must be a non-empty MessagePack map" )); } - let evidence: CultMeshProviderConnectEvidenceWire = rmp_serde::from_slice(payload)?; + let evidence: CultMeshProviderConnectEvidenceWire = cultcache_rs::decode_payload(payload)?; evidence.validate()?; Ok(evidence) } @@ -338,7 +338,7 @@ where let bytes = STANDARD .decode(payload) .map_err(|error| anyhow!("Provider-session payload must be MessagePack base64: {error}"))?; - let value: T = rmp_serde::from_slice(&bytes)?; + let value: T = cultcache_rs::decode_payload(&bytes)?; value.validate()?; Ok(value) } @@ -353,7 +353,7 @@ pub fn decode_provider_command_document( document.schema_id )); } - let command: CultMeshProviderCommandWire = rmp_serde::from_slice(&document.payload)?; + let command: CultMeshProviderCommandWire = cultcache_rs::decode_payload(&document.payload)?; command.validate()?; Ok(command) } diff --git a/packages/cultnet-rs/src/replication.rs b/packages/cultnet-rs/src/replication.rs index 7073f1a9..7b070fbd 100644 --- a/packages/cultnet-rs/src/replication.rs +++ b/packages/cultnet-rs/src/replication.rs @@ -522,7 +522,7 @@ impl CultNetDocumentRegistry { envelope: &CultCacheEnvelope, ) -> Result> { let binding = self.require_binding(&envelope.r#type)?; - let payload: Value = rmp_serde::from_slice(&envelope.payload).with_context(|| { + let payload: Value = cultcache_rs::decode_payload(&envelope.payload).with_context(|| { format!( "failed to decode CultCache envelope {:?} at {:?} as generic CultNet payload", envelope.r#type, envelope.key diff --git a/packages/cultnet-rs/src/service_identity.rs b/packages/cultnet-rs/src/service_identity.rs index 6f3e58d8..3d37adfa 100644 --- a/packages/cultnet-rs/src/service_identity.rs +++ b/packages/cultnet-rs/src/service_identity.rs @@ -306,7 +306,7 @@ fn open_service_identity_envelopes( { bail!("service identity store belongs to a different profile or schema"); } - let entry: ServiceIdentityPrivateEntry = rmp_serde::from_slice(&envelope.payload) + let entry: ServiceIdentityPrivateEntry = cultcache_rs::decode_payload(&envelope.payload) .context("service identity private payload is malformed MessagePack")?; validate_private::

(&entry)?; if entry.protector_kind != platform_protector_kind() || entry.assurance != platform_assurance() @@ -741,6 +741,34 @@ mod tests { Ok(()) } + #[test] + fn a_credential_payload_nested_past_the_bound_is_refused_by_the_cache_walk() -> Result<()> { + let limit = cultcache_rs::MAX_PAYLOAD_DEPTH; + let credential = |levels: usize| -> Result> { + let envelope = CultCacheEnvelope { + key: IdunnServiceIdentity::PRIVATE_KEY.into(), + r#type: IdunnServiceIdentity::PRIVATE_TYPE.into(), + payload: [vec![0x91u8; levels], vec![0xc0]].concat(), + stored_at: "2026-10-11T00:00:00Z".into(), + schema_id: Some(IdunnServiceIdentity::PRIVATE_SCHEMA.into()), + }; + Ok(rmp_serde::to_vec(&vec![envelope])?) + }; + let past = open_service_identity_credential_bytes::(&credential(limit + 1)?) + .err() + .expect("a payload past the bound is refused"); + assert_eq!( + past.downcast_ref::(), + Some(&cultcache_rs::PayloadTooDeep { depth: limit + 1, limit }) + ); + // At the bound the walk lets the payload through to the decoder, which refuses its shape. + let at = open_service_identity_credential_bytes::(&credential(limit)?) + .err() + .expect("the payload is not a credential"); + assert!(at.downcast_ref::().is_none()); + Ok(()) + } + #[test] fn immutable_credential_reader_opens_without_a_path_and_retains_signing_authority() -> Result<()> { diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 2f4572d9..cac4d9dd 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -1,6 +1,7 @@ use anyhow::Result; use chrono::Duration; use chrono::Utc; +use cultcache_rs::CacheBackingStore; use cultcache_rs::CultCache; use cultcache_rs::DatabaseEntry; use cultcache_rs::SingleFileMessagePackBackingStore; @@ -1778,6 +1779,49 @@ fn document_registry_advertises_mutation_contracts_with_bindings() -> Result<()> Ok(()) } +#[test] +fn a_stored_payload_nested_past_the_bound_is_not_replicated_as_a_document() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = temp.path().join("foreign.msgpack"); + let limit = cultcache_rs::MAX_PAYLOAD_DEPTH; + let envelope = |key: &str, levels: usize| cultcache_rs::CultCacheEnvelope { + key: key.to_string(), + r#type: GhostlightAgentStateFixture::TYPE.to_string(), + payload: [vec![0x91u8; levels], vec![0xc0]].concat(), + stored_at: "2026-10-11T00:00:00Z".to_string(), + schema_id: None, + }; + let mut registry = CultNetDocumentRegistry::new(); + registry.register(CultNetDocumentBinding::for_entry::( + "ghostlight.agent_state.v0".to_string(), + )); + // A store written by something that does not hold the bound. + let mut backing = SingleFileMessagePackBackingStore::new(&store); + backing.push(&envelope("at", limit))?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?; + cache.pull_all_backing_stores()?; + + let at_limit = registry.create_snapshot_response(&cache, "at", None, None)?; + let CultNetMessage::SnapshotResponse { documents, .. } = at_limit else { + panic!("expected a snapshot response"); + }; + 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 error = registry.create_snapshot_response(&cache, "past", None, None).unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&cultcache_rs::PayloadTooDeep { depth: limit + 1, limit }) + ); + Ok(()) +} + #[test] fn document_registry_replicates_typed_cultcache_state() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/packages/cultnet-rs/tests/payload_depth.rs b/packages/cultnet-rs/tests/payload_depth.rs new file mode 100644 index 00000000..d810e2de --- /dev/null +++ b/packages/cultnet-rs/tests/payload_depth.rs @@ -0,0 +1,378 @@ +//! Every cultnet-rs decoder of a wire payload that can carry free-form or skipped content holds the +//! shared 256 nesting bound through `cultcache_rs::decode_payload`: 256 is accepted and 257 is refused +//! with the typed refusal, for arrays, maps and union arms, and a 2000-deep payload is an error on a +//! 2 MiB stack, not an abort. + +use anyhow::{Error, Result}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use cultcache_rs::{CultSlots, CultUnion, MAX_PAYLOAD_DEPTH, PayloadTooDeep}; +use cultnet_rs::*; +use serde::Serialize; + +const CANARY: &str = "canary-secret-payload-text"; +const TOO_DEEP: PayloadTooDeep = PayloadTooDeep { depth: MAX_PAYLOAD_DEPTH + 1, limit: MAX_PAYLOAD_DEPTH }; + +#[derive(Clone, Debug, PartialEq, CultUnion)] +enum Node { + #[cultcache(key = 0)] + Box(BoxArm), + #[cultcache(key = 1)] + Text(TextArm), + #[cultcache(key = 6)] + Wrap(WrapArm), +} + +#[derive(Clone, Debug, PartialEq, CultSlots)] +struct BoxArm { + #[cultcache(key = 0)] + id: String, + #[cultcache(key = 1)] + children: Vec, +} + +#[derive(Clone, Debug, PartialEq, CultSlots)] +struct TextArm { + #[cultcache(key = 0)] + id: String, + #[cultcache(key = 1)] + content: String, +} + +#[derive(Clone, Debug, PartialEq, CultSlots)] +struct WrapArm { + #[cultcache(key = 0)] + inner: Option>, +} + +#[derive(Serialize)] +struct MapNest { + inner: Option>, +} + +/// One value `depth` containers deep, from the production encoder. +fn arrays(depth: usize) -> Vec { + let mut value = rmpv::Value::String(CANARY.into()); + for _ in 0..depth { + value = rmpv::Value::Array(vec![value]); + } + rmp_serde::to_vec(&value).unwrap() +} + +fn maps(depth: usize) -> Vec { + let mut nest = MapNest { inner: None }; + for _ in 1..depth { + nest = MapNest { inner: Some(std::boxed::Box::new(nest)) }; + } + rmp_serde::to_vec_named(&nest).unwrap() +} + +// Each Wrap adds a pair and a slots array: a Text leaf is 2 * wraps + 2 deep, a Box leaf 2 * wraps + 3. +fn arms(depth: usize) -> Vec { + let (mut node, wraps) = if depth % 2 == 0 { + (Node::Text(TextArm { id: "leaf".into(), content: CANARY.into() }), (depth - 2) / 2) + } else { + (Node::Box(BoxArm { id: "leaf".into(), children: vec![] }), (depth - 3) / 2) + }; + for _ in 0..wraps { + node = Node::Wrap(WrapArm { inner: Some(std::boxed::Box::new(node)) }); + } + rmp_serde::to_vec(&node).unwrap() +} + +type Nesting = fn(usize) -> Vec; +const NESTINGS: [(&str, Nesting); 3] = [("arrays", arrays), ("maps", maps), ("arms", arms)]; + +/// The hole at 2000 containers, from marker bytes: the encoders recurse too. +fn deep_marker_payload() -> Vec { + let mut bytes = vec![0x91; 2000]; + bytes.push(0xc0); + bytes +} + +/// Swaps the trailing nil of a document for `nested`: the free field is its last. +fn replace_trailing_nil(mut document: Vec, nested: &[u8]) -> Vec { + assert_eq!(document.pop(), Some(0xc0)); + document.extend_from_slice(nested); + document +} + +/// Adds one `x` entry holding `nested` to a MessagePack map: the unknown key a named-map record skips. +fn add_unknown_entry(mut map: Vec, nested: &[u8]) -> Vec { + match map[0] { + header @ 0x80..=0x8e => map[0] = header + 1, + 0xde => { + let count = u16::from_be_bytes([map[1], map[2]]) + 1; + map[1..3].copy_from_slice(&count.to_be_bytes()); + } + other => panic!("not a small map: {other:#x}"), + } + map.extend_from_slice(&[0xa1, b'x']); + map.extend_from_slice(nested); + map +} + +/// Adds one trailing slot to a positional record array: the unknown slot the derive skips. +fn add_trailing_slot(mut array: Vec, nested: &[u8]) -> Vec { + match array[0] { + header @ 0x90..=0x9e => array[0] = header + 1, + 0x9f => array.splice(0..1, [0xdc, 0x00, 0x10]).for_each(drop), + 0xdc => { + let count = u16::from_be_bytes([array[1], array[2]]) + 1; + array[1..3].copy_from_slice(&count.to_be_bytes()); + } + other => panic!("not a small array: {other:#x}"), + } + array.extend_from_slice(nested); + array +} + +type Decode = fn(&[u8]) -> Result<()>; +/// A payload whose free or skipped part is `nested`, deep by `1 + depth(nested)`. +type Frame = Box Vec>; + +struct Decoder { + name: &'static str, + frame: Frame, + decode: Decode, +} + +fn command_document(payload: Vec) -> CultNetRawDocumentRecord { + CultNetRawDocumentRecord { + schema_id: CULTMESH_PROVIDER_COMMAND_SCHEMA.to_string(), + record_key: "command-1".to_string(), + stored_at: "unix:1000".to_string(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + } +} + +fn command_with_nil() -> Vec { + rmp_serde::to_vec_named(&CultMeshProviderCommandWire { + command_id: "command-1".into(), + command_kind: "reload".into(), + provider_id: "aetheria".into(), + service_instance_id: "aetheria-1".into(), + payload: rmpv::Value::Nil, + }) + .unwrap() +} + +fn receipt_with_nil() -> Vec { + rmp_serde::to_vec_named(&CultMeshProviderCommandReceiptWire { + receipt_id: "receipt-1".into(), + command_id: "command-1".into(), + command_kind: "reload".into(), + provider_id: "aetheria".into(), + service_instance_id: "aetheria-1".into(), + state: CultMeshProviderReceiptStateWire::Applied, + completed_at_utc: "2026-01-01T00:00:00Z".into(), + result: Some(rmpv::Value::Nil), + error: None, + }) + .unwrap() +} + +fn evidence_bytes() -> Vec { + encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire { + client_session_id: "client-1".into(), + session_token: None, + }) + .unwrap() +} + +fn hello_bytes() -> Vec { + let hello = CultNetMessage::Hello { + runtime_id: "peer".into(), + runtime_kind: "rust-worker".into(), + agent_id: None, + role: None, + display_name: None, + supported_document_types: None, + supported_mutation_contracts: None, + supported_message_versions: None, + transport_profiles: None, + supports_schema_catalog: None, + }; + encode_cultnet_message_to_vec(&hello, CultNetWireContract::CultNetSchemaV0).unwrap() +} + +fn media_records() -> Vec { + vec![ + GameCultMediaWireRecord::Video(GameCultMediaVideoAccessUnitRecord { + stream_id: "s".into(), + session_id: "x".into(), + frame_id: 7, + codec: "h264".into(), + pts_ticks: 90_000, + duration_ticks: 3_000, + timebase_num: 1, + timebase_den: 90_000, + keyframe: true, + dependency_frame_id: None, + deadline_ticks: 108_000, + chunk_index: 0, + chunk_count: 1, + payload: vec![1, 2, 3], + }), + GameCultMediaWireRecord::VideoParity(GameCultMediaVideoParityShardRecord { + stream_id: "s".into(), + session_id: "x".into(), + frame_id: 1, + codec: "h264".into(), + pts_ticks: 0, + duration_ticks: 1, + timebase_num: 1, + timebase_den: 90_000, + keyframe: false, + dependency_frame_id: Some(0), + deadline_ticks: 1, + chunk_count: 4, + fec_scheme: "rs-gf256-v1".into(), + block_index: 0, + block_count: 1, + block_data_start: 0, + block_data_count: 4, + parity_index: 1, + parity_count: 2, + shard_payload_bytes: 3, + last_chunk_payload_bytes: 2, + payload: vec![1, 2, 3], + }), + GameCultMediaWireRecord::AudioParity(GameCultMediaAudioParityShardRecord { + stream_id: "s".into(), + session_id: "x".into(), + codec: "opus".into(), + fec_scheme: "rs-gf256-v1".into(), + base_packet_id: 8, + base_pts_ticks: 0, + packet_duration_ticks: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 4_000, + data_shard_count: 4, + parity_index: 0, + parity_shard_count: 2, + shard_payload_bytes: 3, + payload: vec![4, 5, 6], + }), + GameCultMediaWireRecord::Audio(GameCultMediaAudioPacketRecord { + stream_id: "s".into(), + session_id: "x".into(), + packet_id: 9, + codec: "opus".into(), + pts_ticks: 0, + duration_ticks: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 1, + payload: vec![4, 5, 6], + }), + GameCultMediaWireRecord::Feedback(GameCultMediaReceiverFeedbackRecord { + stream_id: "s".into(), + session_id: "x".into(), + receiver_id: "obs".into(), + highest_decodable_frame_id: Some(41), + missing_frame_ids: vec![42], + late_frame_ids: vec![43], + requested_keyframe: true, + jitter_us: 500, + decode_queue_us: 2_000, + observed_at: "unix:1000".into(), + missing_video_chunk_keys: vec!["42:1".into()], + }), + ] +} + +/// A valid media wire message whose record carries one more, unknown slot holding `nested`. +fn media_wire_with_unknown_slot(record: &GameCultMediaWireRecord, nested: &[u8]) -> Vec { + let provenance = + MediaWireProvenance { stored_at: "unix:1000", runtime_id: "peer", role: "media", producer: "peer" }; + let wire = encode_media_wire_record(record, provenance).unwrap(); + let message = decode_cultnet_message_from_slice(&wire, CultNetWireContract::CultNetSchemaV0).unwrap(); + let CultNetMessage::DocumentPutRaw { message_id, mut document } = message else { unreachable!() }; + document.payload = add_trailing_slot(document.payload, nested); + let message = CultNetMessage::DocumentPutRaw { message_id, document }; + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap() +} + +const MEDIA_NAMES: [&str; 5] = [ + "media video record", + "media video parity record", + "media audio parity record", + "media audio record", + "media feedback record", +]; + +fn decoders() -> Vec { + vec![ + Decoder { + name: "provider command document (a free Value field)", + frame: Box::new(|nested| replace_trailing_nil(command_with_nil(), nested)), + decode: |payload| decode_provider_command_document(&command_document(payload.to_vec())).map(drop), + }, + Decoder { + name: "provider session payload (a receipt's free Value field)", + frame: Box::new(|nested| replace_trailing_nil(receipt_with_nil(), nested)), + decode: |payload| { + decode_provider_session_payload::(&STANDARD.encode(payload)) + .map(drop) + }, + }, + Decoder { + name: "provider connect evidence (an unknown key a named map skips)", + frame: Box::new(|nested| add_unknown_entry(evidence_bytes(), nested)), + decode: |payload| decode_provider_connect_evidence(payload).map(drop), + }, + Decoder { + name: "CultNet message (a free rmpv::Value tree)", + frame: Box::new(|nested| add_unknown_entry(hello_bytes(), nested)), + decode: |payload| decode_cultnet_message_from_slice(payload, CultNetWireContract::CultNetSchemaV0).map(drop), + }, + ] + .into_iter() + .chain(media_records().into_iter().zip(MEDIA_NAMES).map(|(record, name)| Decoder { + name, + frame: Box::new(move |nested| media_wire_with_unknown_slot(&record, nested)), + decode: |wire| decode_media_wire_record(wire).map(drop), + })) + .collect() +} + +fn refusal(error: &Error) -> Option { + error.downcast_ref::().copied() +} + +// The document is one container itself, so the free part nests 255 at the limit and 256 past it. +#[test] +fn every_payload_decoder_accepts_256_and_refuses_257_for_arrays_maps_and_arms() { + for decoder in decoders() { + for (kind, nesting) in NESTINGS { + let at = (decoder.frame)(&nesting(MAX_PAYLOAD_DEPTH - 1)); + let past = (decoder.frame)(&nesting(MAX_PAYLOAD_DEPTH)); + (decoder.decode)(&at).unwrap_or_else(|error| panic!("{} {kind} at the limit: {error:#}", decoder.name)); + let error = (decoder.decode)(&past).expect_err(&format!("{} {kind} past the limit", decoder.name)); + assert_eq!(refusal(&error), Some(TOO_DEEP), "{} {kind}: {error:#}", decoder.name); + assert!(!format!("{error:#}").contains(CANARY), "{} {kind} echoed the payload", decoder.name); + } + } +} + +#[test] +fn a_2000_deep_payload_is_a_typed_refusal_on_a_2_mib_stack() { + let handle = std::thread::Builder::new() + .stack_size(2 * 1024 * 1024) + .spawn(|| { + for decoder in decoders() { + let deep = (decoder.frame)(&deep_marker_payload()); + let error = (decoder.decode)(&deep).expect_err(decoder.name); + assert_eq!(refusal(&error), Some(TOO_DEEP), "{}: {error:#}", decoder.name); + } + }) + .unwrap(); + handle.join().expect("a decoder aborted or panicked on a 2 MiB stack"); +}