GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge hands/union-bounds-rs: one 256 nesting bound at every Rust payload door, cache, mesh network puts and cultnet decoders, held to the C# bounds vectors

Browse source 
Soul passes cultlib-gaps:verdict:cut-union-bounds-rs.s1, .s2 and .s3.
Low residuals deferred to follow_ups depth-vectors-nest-every-container
and rust-past-limit-record-handling; encode-stack-safety stays open.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-11 19:01:38 +02:00
commit c028c51102
11 changed files with 1367 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

272
packages/cultcache-rs/cultcache-rs-derive/src/lib.rs
View file

@ -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<proc_macro2::TokenStream> {
let ident = input.ident.clone();
let mut entry_type: Option<LitStr> = None;
let mut schema_name: Option<LitStr> = 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<proc_macro2::TokenStream> {
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<Vec<CultCacheSlot>> {
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<Vec<C
if meta.path.is_ident("key") {
once(&key, &meta, "key")?;
let value: LitInt = meta.value()?.parse()?;
key = Some(value.base10_parse()?);
let slot: usize = value.base10_parse()?;
if slot > 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<S>(&self, serializer: S) -> ::std::result::Result<S::Ok, S::Error>
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<D>(deserializer: D) -> ::std::result::Result<Self, D::Error>
fn deserialize<__D>(deserializer: __D) -> ::std::result::Result<Self, __D::Error>
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<A>(self, mut seq: A) -> ::std::result::Result<Self::Value, A::Error>
fn visit_seq<__A>(self, mut seq: __A) -> ::std::result::Result<Self::Value, __A::Error>
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<proc_macro2::TokenStream> {
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<proc_macro2::TokenStream> {
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<proc_macro2::TokenStream> {
};
impl ::serde::Serialize for #ident {
fn serialize<S>(&self, serializer: S) -> ::std::result::Result<S::Ok, S::Error>
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<proc_macro2::TokenStream> {
}
impl<'de> ::serde::Deserialize<'de> for #ident {
fn deserialize<D>(deserializer: D) -> ::std::result::Result<Self, D::Error>
fn deserialize<__D>(deserializer: __D) -> ::std::result::Result<Self, __D::Error>
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<E>(self, value: i64) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::Int(value)))
fn visit_i64<__E>(self, value: i64) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::Int(value)))
}
fn visit_u64<E>(self, value: u64) -> ::std::result::Result<Self::Value, E> {
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<Self::Value, __E> {
Ok(__ArmKeyProbe(match i64::try_from(value) {
Ok(value) => __ArmKey::Int(value),
Err(_) => __ArmKey::OutOfRange,
}))
}
fn visit_bool<E>(self, _: bool) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_bool<__E>(self, _: bool) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_f64<E>(self, _: f64) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_f64<__E>(self, _: f64) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_str<E>(self, _: &str) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_str<__E>(self, _: &str) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_bytes<E>(self, _: &[u8]) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_bytes<__E>(self, _: &[u8]) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_unit<E>(self) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_unit<__E>(self) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_none<E>(self) -> ::std::result::Result<Self::Value, E> {
Ok(ArmKeyProbe(ArmKey::NonInteger))
fn visit_none<__E>(self) -> ::std::result::Result<Self::Value, __E> {
Ok(__ArmKeyProbe(__ArmKey::NonInteger))
}
fn visit_some<D>(self, deserializer: D) -> ::std::result::Result<Self::Value, D::Error>
fn visit_some<__D>(self, deserializer: __D) -> ::std::result::Result<Self::Value, __D::Error>
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<A>(self, mut seq: A) -> ::std::result::Result<Self::Value, A::Error>
// A MessagePack ext value (fixext, timestamp, ext8) reaches the visitor here.
fn visit_newtype_struct<__D>(self, deserializer: __D) -> ::std::result::Result<Self::Value, __D::Error>
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<Self::Value, __A::Error>
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<A>(self, mut map: A) -> ::std::result::Result<Self::Value, A::Error>
fn visit_map<__A>(self, mut map: __A) -> ::std::result::Result<Self::Value, __A::Error>
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<D>(deserializer: D) -> ::std::result::Result<Self, D::Error>
impl<'de> ::serde::Deserialize<'de> for __ArmKeyProbe {
fn deserialize<__D>(deserializer: __D) -> ::std::result::Result<Self, __D::Error>
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<A>(self, mut seq: A) -> ::std::result::Result<Self::Value, A::Error>
fn visit_seq<__A>(self, mut seq: __A) -> ::std::result::Result<Self::Value, __A::Error>
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::<ArmKeyProbe>()
.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<proc_macro2::TokenStream> {
}
}
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<proc_macro2::TokenStream> {
database_entry_codec(syn::parse_str(source).expect("the fixture parses"))
}
fn slots_result(source: &str) -> syn::Result<proc_macro2::TokenStream> {
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<T> { #[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");
}
}

146
packages/cultcache-rs/src/lib.rs
View file

@ -99,6 +99,14 @@
//! enum Bad { #[cultcache(key = 4)] A(Option<Arm>) }
//! ```
//!
//! 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<usize>)> {
// The big-endian length after the code byte.
let length = |width: usize| -> Option<usize> {
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<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<T> {
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<u8>,
) -> 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,

478
packages/cultcache-rs/tests/union_parity.rs
View file

@ -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<std::boxed::Box<Node>>,
}
#[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::<Outcome>(&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::<PlainEnums>(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::<E>(&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::<A>(&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<UnionSurface, String> {
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::<UnionSurface>().unwrap();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
cache.pull_all_backing_stores().unwrap();
match cache.get::<UnionSurface>("k") {
Ok(value) => Ok(value.expect("the envelope was stored")),
Err(error) => Err(format!("{error:#}")),
}
}
#[test]
fn bounds_vectors() {
let entries: Vec<serde_json::Value> = 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::<UnionSurface>().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::<PayloadTooDeep>(), Some(&PayloadTooDeep { depth: 257, limit: 256 }));
assert!(!format!("{error:#}").contains(CANARY), "{error:#}");
assert_eq!(cache.get::<UnionSurface>("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<std::boxed::Box<MapNest>>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.map_doc", schema = "tests.map_doc.v1")]
struct MapDoc {
#[cultcache(key = 0)]
root: Option<MapNest>,
}
// 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::<MapDoc>().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::<MapDoc>("at").unwrap(), Some(at_limit));
let error = cache.prepare_entry_named("past", &past_limit).unwrap_err();
assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP));
}
fn raw(type_name: &str, key: &str, payload: Vec<u8>) -> 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::<UnionSurface>().unwrap();
cache.register_entry_type::<MapDoc>().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::<PayloadTooDeep>(), 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::<UnionSurface>().unwrap();
let value = wrap_chain(126, boxed("leaf", vec![]));
assert_eq!(cache.put("at", &value).unwrap(), value);
assert_eq!(cache.get::<UnionSurface>("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<u8> {
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<u8> { [head.to_vec(), vec![FILL; bytes]].concat() };
let long = |code: u8, width: usize, length: usize, extra: usize| -> Vec<u8> {
let size = length.to_be_bytes();
filled(&[&[code][..], &size[size.len() - width..]].concat(), length + extra)
};
let mut elements: Vec<Vec<u8>> = 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");
}

108
packages/cultmesh-rs/src/lib.rs
View file

@ -1733,6 +1733,114 @@ mod tests {
Ok(())
}
#[derive(serde::Serialize)]
struct Nest {
inner: Option<Box<Nest>>,
}
// `levels` maps deep, written by the serializer the cache itself uses.
fn nest(levels: usize) -> Vec<u8> {
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<u8>) -> 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<cultnet_rs::CultNetRawDocumentRecord>,
) -> Result<(
SocketAddr,
std::sync::Arc<std::sync::atomic::AtomicBool>,
thread::JoinHandle<Result<()>>,
)> {
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<cultnet_rs::CultNetRawDocumentRecord>,
) -> Result<usize> {
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::<cultcache_rs::PayloadTooDeep>(),
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();

6
packages/cultnet-rs/src/contracts.rs
View file

@ -604,11 +604,7 @@ pub fn decode_cultnet_message_from_slice(
bytes: &[u8],
contract: CultNetWireContract,
) -> Result<CultNetMessage> {
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)
}

10
packages/cultnet-rs/src/media_stream_wire.rs
View file

@ -211,31 +211,31 @@ pub fn decode_media_wire_record(payload: &[u8]) -> Result<GameCultMediaWireRecor
let record = match document.schema_id.as_str() {
GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA => {
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)
}

6
packages/cultnet-rs/src/provider_session.rs
View file

@ -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)
}

2
packages/cultnet-rs/src/replication.rs
View file

@ -522,7 +522,7 @@ impl CultNetDocumentRegistry {
envelope: &CultCacheEnvelope,
) -> Result<CultNetDocumentRecord<Value>> {
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

30
packages/cultnet-rs/src/service_identity.rs
View file

@ -306,7 +306,7 @@ fn open_service_identity_envelopes<P: ServiceIdentityProfile>(
{
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::<P>(&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<Vec<u8>> {
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::<IdunnServiceIdentity>(&credential(limit + 1)?)
.err()
.expect("a payload past the bound is refused");
assert_eq!(
past.downcast_ref::<cultcache_rs::PayloadTooDeep>(),
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::<IdunnServiceIdentity>(&credential(limit)?)
.err()
.expect("the payload is not a credential");
assert!(at.downcast_ref::<cultcache_rs::PayloadTooDeep>().is_none());
Ok(())
}
#[test]
fn immutable_credential_reader_opens_without_a_path_and_retains_signing_authority() -> Result<()>
{

44
packages/cultnet-rs/tests/cultnet.rs
View file

@ -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::<GhostlightAgentStateFixture>(
"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::<GhostlightAgentStateFixture>()?;
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::<GhostlightAgentStateFixture>()?;
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::<cultcache_rs::PayloadTooDeep>(),
Some(&cultcache_rs::PayloadTooDeep { depth: limit + 1, limit })
);
Ok(())
}
#[test]
fn document_registry_replicates_typed_cultcache_state() -> Result<()> {
let temp = tempfile::tempdir()?;

378
packages/cultnet-rs/tests/payload_depth.rs Normal file
View file

@ -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<Node>,
}
#[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<std::boxed::Box<Node>>,
}
#[derive(Serialize)]
struct MapNest {
inner: Option<std::boxed::Box<MapNest>>,
}
/// One value `depth` containers deep, from the production encoder.
fn arrays(depth: usize) -> Vec<u8> {
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<u8> {
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<u8> {
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<u8>;
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<u8> {
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<u8>, nested: &[u8]) -> Vec<u8> {
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<u8>, nested: &[u8]) -> Vec<u8> {
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<u8>, nested: &[u8]) -> Vec<u8> {
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<dyn Fn(&[u8]) -> Vec<u8>>;
struct Decoder {
name: &'static str,
frame: Frame,
decode: Decode,
}
fn command_document(payload: Vec<u8>) -> 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<u8> {
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<u8> {
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<u8> {
encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: "client-1".into(),
session_token: None,
})
.unwrap()
}
fn hello_bytes() -> Vec<u8> {
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<GameCultMediaWireRecord> {
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<u8> {
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<Decoder> {
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::<CultMeshProviderCommandReceiptWire>(&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<PayloadTooDeep> {
error.downcast_ref::<PayloadTooDeep>().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");
}