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

cultcache-py: the payload depth count takes every mapping, sequence and set; a codec's RecursionError is no longer relabelled as a depth refusal

Browse source 
The tree walk counted only list, tuple and dict, so a codec whose tree used deque,
mappingproxy or another abc container was unbounded. Every Mapping, Sequence and Set
that is not text or bytes now counts. RecursionError from a user encode or decode,
and from the JSON decoder, reaches the caller as itself with its traceback; only
msgpack's StackError (raised by depth alone) is mapped to the depth refusal.
All commits build.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-11 17:47:55 +02:00
commit 7f6649f9a1
2 changed files with 72 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

36
packages/cultcache-py/src/cultcache_py/documents.py
View file

@ -2,8 +2,9 @@
import json
from dataclasses import MISSING, dataclass, field
from collections.abc import Mapping, Sequence, Set
from itertools import chain
from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar
from typing import Any, Callable, Generic, Iterable, TypeVar
from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
@ -24,12 +25,13 @@ MAX_SLOT_KEY = 65535
try:
from msgpack.exceptions import StackError as _MsgpackStackError
_NESTING_FAULTS: tuple[type[BaseException], ...] = (RecursionError, _MsgpackStackError)
_STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
except ModuleNotFoundError:
_NESTING_FAULTS = (RecursionError,)
_STACK_FAULTS = ()
_CONTAINERS = (list, tuple, dict)
_CONTAINERS = (Mapping, Sequence, Set)
_SCALAR_SEQUENCES = (str, bytes, bytearray, memoryview)
class CultCachePayloadDepthError(ValueError):
@ -44,15 +46,19 @@ class CultCachePayloadDepthError(ValueError):
def _refuse_past_depth_limit(tree: Any, type_name: str, direction: str) -> None:
"""The one nesting count. Walks a payload tree iteratively (lists, tuples and dicts are the containers), refusing
the first container past `MAX_PAYLOAD_DEPTH`."""
stack = [(tree, 1)] if isinstance(tree, _CONTAINERS) else []
"""The one nesting count. Walks a payload tree iteratively (every mapping, sequence and set that is not text or
bytes is a container), refusing the first container past `MAX_PAYLOAD_DEPTH`."""
stack = [(tree, 1)] if _is_container(tree) else []
while stack:
node, depth = stack.pop()
if depth > MAX_PAYLOAD_DEPTH:
raise CultCachePayloadDepthError(type_name, direction)
children = chain(node.keys(), node.values()) if isinstance(node, dict) else node
stack.extend((child, depth + 1) for child in children if isinstance(child, _CONTAINERS))
children = chain(node.keys(), node.values()) if isinstance(node, Mapping) else node
stack.extend((child, depth + 1) for child in children if _is_container(child))
def _is_container(node: Any) -> bool:
return isinstance(node, _CONTAINERS) and not isinstance(node, _SCALAR_SEQUENCES)
def _default_encode(value: Any) -> Any:
@ -166,23 +172,17 @@ class DocumentDefinition(Generic[T]):
def encode_payload(self, value: T) -> bytes:
if self.validate:
self.validate(value)
try:
tree = self.encode(value)
except RecursionError:
raise CultCachePayloadDepthError(self.type, "write") from None
tree = self.encode(value)
_refuse_past_depth_limit(tree, self.type, "write")
return self.payload_encoder(tree)
def decode_payload(self, payload: bytes) -> T:
try:
tree = self.payload_decoder(payload)
except _NESTING_FAULTS:
except _STACK_FAULTS:
raise CultCachePayloadDepthError(self.type, "read") from None
_refuse_past_depth_limit(tree, self.type, "read")
try:
value = self.decode(tree)
except RecursionError:
raise CultCachePayloadDepthError(self.type, "read") from None
value = self.decode(tree)
if self.validate:
self.validate(value)
return value

68
packages/cultcache-py/tests/test_union_parity.py
View file

@ -1,8 +1,11 @@
from __future__ import annotations
import collections
import dataclasses
import shutil
import tempfile
import traceback
import types
import unittest
from pathlib import Path
@ -495,30 +498,28 @@ class BoundsVectorTests(unittest.TestCase):
surface_document.decode_payload(huge)
def test_a_deep_value_is_refused_by_the_depth_error_not_a_recursion_error(self) -> None:
deep = {**_surface_value(), "root": _wrap_chain(5000, UnionArm("Box", {"id": "leaf", "children": []}))}
tree = define_document_type("tests.deep_tree", encode=lambda value: value)
with self.assertRaises(CultCachePayloadDepthError) as caught:
surface_document.encode_payload(deep)
tree.encode_payload(_nest(5000))
self.assertIn("refused on write", str(caught.exception))
document = define_document_type("tests.json_tree")
with self.assertRaises(CultCachePayloadDepthError) as read:
document.decode_payload(b"[" * 100000)
document.decode_payload(b"[" * 300 + b"]" * 300)
self.assertIn("refused on read", str(read.exception))
def test_a_codec_that_exhausts_the_interpreter_stack_is_the_depth_refusal_on_each_side(self) -> None:
def test_a_codec_that_exhausts_the_interpreter_stack_keeps_its_own_recursion_error_on_each_side(self) -> None:
def boom(value: object) -> object:
return boom(value)
document = define_document_type("tests.boom", encode=boom, decode=boom)
with self.assertRaises(CultCachePayloadDepthError) as write:
document.encode_payload("x")
self.assertEqual(
str(write.exception), "tests.boom payload nests 257 containers deep; the limit is 256 (refused on write)."
)
with self.assertRaises(CultCachePayloadDepthError) as read:
document.decode_payload(b'"x"')
self.assertEqual(
str(read.exception), "tests.boom payload nests 257 containers deep; the limit is 256 (refused on read)."
)
for run in (lambda: document.encode_payload("x"), lambda: document.decode_payload(b'"x"')):
try:
run()
except RecursionError as caught:
self.assertNotIsInstance(caught, CultCachePayloadDepthError)
self.assertIn("boom", [frame.name for frame in traceback.extract_tb(caught.__traceback__)])
else:
self.fail("the codec's RecursionError was swallowed")
with self.assertRaises(CultCachePayloadDepthError) as stack:
surface_document.decode_payload(bytes([0x91]) * 2000 + bytes([0xC0]))
self.assertEqual(
@ -526,6 +527,22 @@ class BoundsVectorTests(unittest.TestCase):
"tests.union_surface payload nests 257 containers deep; the limit is 256 (refused on read).",
)
def test_a_json_payload_past_the_interpreters_own_stack_keeps_the_decoders_recursion_error(self) -> None:
document = define_document_type("tests.json_tree")
with self.assertRaises(RecursionError) as caught:
document.decode_payload(b"[" * 100000)
self.assertNotIsInstance(caught.exception, CultCachePayloadDepthError)
def test_a_refusal_reports_the_depth_where_the_walk_stopped_whatever_the_payloads_real_depth(self) -> None:
document = define_document_type("tests.json_tree", encode=lambda value: value)
for depth in (257, 300, 900):
with self.assertRaises(CultCachePayloadDepthError) as write:
document.encode_payload(_nest(depth))
self.assertEqual((write.exception.depth, write.exception.limit), (257, 256), depth)
with self.assertRaises(CultCachePayloadDepthError) as read:
document.decode_payload(b"[" * depth + b"]" * depth)
self.assertEqual((read.exception.depth, read.exception.limit), (257, 256), depth)
def test_the_validator_sees_the_value_on_write_and_the_decoded_value_on_read(self) -> None:
seen: list[object] = []
document = define_document_type("tests.validated", validate=seen.append)
@ -575,6 +592,29 @@ class PayloadDepthRuleTests(unittest.TestCase):
with self.assertRaises(CultCachePayloadDepthError):
document.encode_payload({_as_tuple(_nest(256)): 1})
def test_every_mapping_sequence_and_set_counts_not_only_list_tuple_and_dict(self) -> None:
wrappers = {
"deque": lambda inner: collections.deque([inner]),
"mappingproxy": lambda inner: types.MappingProxyType({"k": inner}),
"frozenset": lambda inner: frozenset([inner]),
"userlist": lambda inner: collections.UserList([inner]),
}
for name, wrap in wrappers.items():
document = define_document_type(f"tests.{name}", encode=lambda value: value, payload_encoder=lambda tree: b"ok")
at_limit: object = "leaf"
for _ in range(256):
at_limit = wrap(at_limit)
self.assertEqual(document.encode_payload(at_limit), b"ok", name)
with self.assertRaises(CultCachePayloadDepthError, msg=name):
document.encode_payload(wrap(at_limit))
decoding = define_document_type(f"tests.{name}.read", payload_decoder=lambda payload, w=wrap: w(at_limit))
with self.assertRaises(CultCachePayloadDepthError, msg=name):
decoding.decode_payload(b"x")
def test_text_and_bytes_are_leaves_however_long(self) -> None:
document = define_document_type("tests.leaves", encode=lambda value: value, payload_encoder=lambda tree: b"ok")
self.assertEqual(document.encode_payload([["x" * 100000, b"y" * 100000, bytearray(5), memoryview(b"z")]]), b"ok")
def test_a_scalar_payload_has_no_depth(self) -> None:
document = define_document_type("tests.scalar", encode=lambda value: value)
for scalar in ("x", 5, None):