fix: cap integer products, concatenation, nested collections and % formatting in the restricted evaluator

docs/io/security.md

@@ -115,12 +115,15 @@ action failure.
 
 ## Arithmetic is magnitude-bounded
 
-The restricted evaluator allows arithmetic, but `**` and `*` are capped so that a tiny
-expression cannot exhaust CPU or memory: an exponentiation whose result would be enormous
-(`9 ** 9 ** 9`, a ~370-million-digit number) and a sequence replication that would allocate a
-huge object (`[0] * 20000000`) are both rejected before they run. Ordinary scalar arithmetic
-(`x * 2`, `x ** 2`) is unaffected. Under `trusted=True` the full Python evaluator is used, so
-these caps do not apply.
+The restricted evaluator allows arithmetic, but caps the operators that can build a huge object
+from a tiny expression, so it cannot exhaust CPU or memory. An exponentiation or an integer
+product whose result would be enormous (`9 ** 9 ** 9`, a ~370-million-digit number), a
+sequence replication or concatenation that would allocate a huge object (`[0] * 20000000`), a
+list, tuple, set or dict whose nested items add up to a huge value (`[[0] * 1000] * 2000`,
+cheap to build but slow to compare, hash or print), and `%` string formatting, whose field
+width sizes the result (`'%20000000d' % 0`), are rejected before they run. Ordinary arithmetic
+and collections (`x * 2`, `x ** 2`, `x + 1`, `x % 2`, `[a, b]`) are unaffected. Under
+`trusted=True` the full Python evaluator is used, so these caps do not apply.
 
 ## External resource references (`src`)
 

docs/releases/3.2.2.md

@@ -2,12 +2,31 @@
 
 *Not released yet*
 
-python-statemachine 3.2.2 is a bugfix release that also adds Python 3.15 support. Most of its
+python-statemachine 3.2.2 is a bugfix release that also adds Python 3.15 support and closes
+denial-of-service gaps in the restricted expression evaluator. Most of its
 fixes close gaps between what the documentation promised and what the engine did: events
 declared inside a `State.Compound` body, `delay` and `internal` declared on an `Event`, internal
 events and the order of the event queues, callbacks spread over nested state bodies, and the
 errors Graphviz reports when it cannot render a diagram.
 
+## Security fixes in 3.2.2
+
+### Unbounded arithmetic in the restricted evaluator (GHSA-9r2q-5523-wc47, GHSA-rfcg-j6wv-c8cr, GHSA-wq35-qxpw-w5jc)
+
+The magnitude caps added in 3.2.1 (GHSA-r8gj-366q-cgvj) covered `**` and sequence replication
+with `*`, but other operators could still build a huge object from a small untrusted document
+loaded with the default `trusted=False`: a product of two integers, a sequence concatenation
+with `+`, nested collections that repeat a reference (`[[0] * 1000] * 2000` is cheap to build
+but walks two million items when compared, hashed or printed), and `%` string formatting, whose
+field width sizes the result. The restricted evaluator now caps the integer product, and caps
+every sequence or collection it builds by the size of its nested items counted per
+reference, and it rejects `%` on `str` and `bytes` operands. Numeric `%` and ordinary arithmetic are
+unaffected, and `trusted=True` still uses full Python without these caps. See
+[](../io/security.md).
+
+Reported by [@manus-pi](https://github.com/manus-pi) (GHSA-9r2q-5523-wc47,
+GHSA-rfcg-j6wv-c8cr) and [@manus-use](https://github.com/manus-use) (GHSA-wq35-qxpw-w5jc).
+
 ## Python compatibility in 3.2.2
 
 Added Python 3.15 on the test matrix. python-statemachine 3.2.2 supports Python 3.10, 3.11, 3.12,
@@ -63,6 +82,9 @@ fail after upgrading:
   {ref}`release-3.2.2-name-collisions`.
 - A state reached through its parent (`sm.door.shut`) now reports its real `is_active` and
   `parent`. See {ref}`release-3.2.2-nested-proxies`.
+- `%` string formatting (`'%d items' % n`) in an expression of a document loaded with
+  `trusted=False` is now rejected and routed to `error.execution` like any failing expression.
+  Use `trusted=True` for documents that need it.
 - The diagram paths that run `dot` (the `svg` format, `_repr_svg_()`, the command line and the
   Sphinx directive) raise `statemachine.contrib.diagram.GraphvizError` where they raised pydot's
   `AssertionError` when `dot` failed. A missing `dot` raises `GraphvizNotFoundError`, which is

statemachine/spec_parser.py

@@ -204,18 +204,22 @@ def decorated(*args, **kwargs):
 
 def build_collection(factory, item_exprs: "list[Callable]") -> Callable:
     def decorated(*args, **kwargs):
-        return factory(item(*args, **kwargs) for item in item_exprs)
+        items = [item(*args, **kwargs) for item in item_exprs]
+        _check_value_size(items, "collection")
+        return factory(items)
 
     decorated.__name__ = f"{factory.__name__}(...)"
     return decorated
 
 
 def build_dict(key_exprs: "list[Callable]", value_exprs: "list[Callable]") -> Callable:
     def decorated(*args, **kwargs):
-        return {
-            key(*args, **kwargs): value(*args, **kwargs)
+        items = [
+            (key(*args, **kwargs), value(*args, **kwargs))
             for key, value in zip(key_exprs, value_exprs, strict=True)
-        }
+        ]
+        _check_value_size(items, "dict")
+        return dict(items)
 
     decorated.__name__ = "dict(...)"
     return decorated
@@ -383,11 +387,51 @@ def parse_expr(expr: str, variable_hook: Callable) -> Callable:
     ast.NotEq: build_custom_operator(operator.ne),
 }
 
-# Result-size caps for the two operators that can blow up cheaply. Ordinary scalar
-# arithmetic (``x * 2``, ``x ** 2``) is well under these; the caps only reject the
-# denial-of-service forms (``9**9**9`` bignum, ``[0]*20000000`` sequence replication).
+# Result-size caps for the values an expression can build cheaply. Ordinary arithmetic and
+# collections are well under these; the caps only reject the denial-of-service forms (giant
+# bignums, giant or deeply shared sequences and collections).
 _MAX_POW_RESULT_BITS = 4096
-_MAX_SEQUENCE_LEN = 1_000_000
+_MAX_VALUE_SIZE = 1_000_000
+
+_TEXT_TYPES = (str, bytes, bytearray)
+_SEQUENCE_TYPES = (*_TEXT_TYPES, list, tuple)
+_CONTAINER_TYPES = (list, tuple, set, frozenset)
+
+
+def _value_size(value) -> int:
+    """Size of ``value`` counting a shared reference every time it appears.
+
+    ``[[0] * 1000] * 1000`` holds a thousand references to one list, so it is cheap to build,
+    but comparing, hashing or printing it walks a million elements. That walked size is what
+    is measured here, in elements, characters or bytes, memoized by identity so the
+    measurement itself stays cheap.
+    """
+    memo: "dict[int, int]" = {}
+
+    def size(item) -> int:
+        key = id(item)
+        if key in memo:
+            return memo[key]
+        memo[key] = 1
+        if isinstance(item, _TEXT_TYPES):
+            total = len(item)
+        elif isinstance(item, int):
+            total = item.bit_length() // 8
+        elif isinstance(item, dict):
+            total = sum(size(k) + size(v) for k, v in item.items())
+        elif isinstance(item, _CONTAINER_TYPES):
+            total = sum(size(element) for element in item)
+        else:
+            total = 1
+        memo[key] = max(total, 1)
+        return memo[key]
+
+    return size(value)
+
+
+def _check_value_size(value, what: str):
+    if _value_size(value) > _MAX_VALUE_SIZE:
+        raise ValueError(f"{what} is too large for the restricted evaluator")
 
 
 def _guarded_pow(base, exp):
@@ -405,29 +449,52 @@ def _guarded_pow(base, exp):
 
 
 def _guarded_mul(a, b):
-    """``*`` with a sequence-replication cap (GHSA-r8gj-366q-cgvj).
+    """``*`` with a result-size cap (GHSA-r8gj-366q-cgvj).
 
     ``seq * n`` (list/str/bytes/tuple times an int) can allocate an enormous object from a
-    12-character expression (``[0]*20000000``). The resulting length is checked *before*
-    allocating. Scalar numeric multiplication is unaffected.
+    12-character expression (``[0]*20000000``), and repeatedly squaring an ``int`` doubles its
+    size every time. The result size is checked *before* allocating. Other operands (floats)
+    are passed straight through.
     """
+    if isinstance(a, int) and isinstance(b, int):
+        if a.bit_length() + b.bit_length() > _MAX_POW_RESULT_BITS:
+            raise ValueError("'*' result is too large for the restricted evaluator")
     for seq, n in ((a, b), (b, a)):
-        if isinstance(seq, (str, bytes, bytearray, list, tuple)) and isinstance(n, int):
-            if n > 0 and n * len(seq) > _MAX_SEQUENCE_LEN:
+        if isinstance(seq, _SEQUENCE_TYPES) and isinstance(n, int):
+            if n > 0 and n * _value_size(seq) > _MAX_VALUE_SIZE:
                 raise ValueError("'*' repetition is too large for the restricted evaluator")
     return operator.mul(a, b)
 
 
-# ``**`` and ``*`` stay available for ordinary arithmetic but are wrapped so a tiny untrusted
-# expression cannot exhaust CPU/memory (GHSA-r8gj-366q-cgvj). ``trusted=True`` uses the full
-# Python evaluator instead, without these caps.
+def _guarded_add(a, b):
+    """``+`` with a sequence-length cap: repeated ``x + x`` doubles a sequence every time."""
+    sizes = [_value_size(x) for x in (a, b) if isinstance(x, _SEQUENCE_TYPES)]
+    if len(sizes) == 2 and sum(sizes) > _MAX_VALUE_SIZE:
+        raise ValueError("'+' result is too large for the restricted evaluator")
+    return operator.add(a, b)
+
+
+def _guarded_mod(a, b):
+    """``%`` restricted to numbers.
+
+    printf-style formatting (``'%2000000000d' % 0``) allocates whatever field width the
+    expression asks for, so it is rejected on ``str``/``bytes`` operands.
+    """
+    if isinstance(a, _TEXT_TYPES):
+        raise ValueError("'%' string formatting is not allowed by the restricted evaluator")
+    return operator.mod(a, b)
+
+
+# ``**``, ``*``, ``+`` and ``%`` stay available for ordinary arithmetic but are wrapped so a
+# tiny untrusted expression cannot exhaust CPU/memory (GHSA-r8gj-366q-cgvj). ``trusted=True``
+# uses the full Python evaluator instead, without these caps.
 binary_operators = {
-    ast.Add: operator.add,
+    ast.Add: _guarded_add,
     ast.Sub: operator.sub,
     ast.Mult: _guarded_mul,
     ast.Div: operator.truediv,
     ast.FloorDiv: operator.floordiv,
-    ast.Mod: operator.mod,
+    ast.Mod: _guarded_mod,
     ast.Pow: _guarded_pow,
 }
 

tests/io/test_security.py

@@ -600,10 +600,22 @@ def _dos_expr_scxml(expr: str) -> str:
 
 @pytest.mark.parametrize("fmt", FORMATS)
 class TestArithmeticDoS:
-    """``**`` and ``*`` are magnitude-capped in the restricted evaluator: the denial-of-service
-    forms are rejected, but ordinary scalar arithmetic keeps working (GHSA-r8gj-366q-cgvj)."""
-
-    @pytest.mark.parametrize("expr", ["9**9**9", "2**100000", "[0]*20000000", "'a'*20000000"])
+    """``**``, ``*``, ``+`` and ``%`` are magnitude-capped in the restricted evaluator: the
+    denial-of-service forms are rejected, but ordinary scalar arithmetic keeps working
+    (GHSA-r8gj-366q-cgvj)."""
+
+    @pytest.mark.parametrize(
+        "expr",
+        [
+            "9**9**9",
+            "2**100000",
+            "[0]*20000000",
+            "'a'*20000000",
+            "3**2048 * 3**2048",
+            "[0]*999999 + [0]*999999",
+            "'%20000000d' % 0",
+        ],
+    )
     def test_dos_expr_rejected(self, fmt, expr):
         # The cap raises while evaluating the expression, surfacing as error.execution.
         sm = _run_exec(
@@ -615,12 +627,12 @@ def test_dos_expr_rejected(self, fmt, expr):
 
     def test_scalar_arithmetic_preserved(self, fmt):
         sm = _run_exec(
-            _dos_expr_scxml("x * 2 + 3 ** 2"),
-            [{"assign": {"location": "x", "expr": "x * 2 + 3 ** 2"}}],
+            _dos_expr_scxml("x * 2 + 3 ** 2 + 10 % 3"),
+            [{"assign": {"location": "x", "expr": "x * 2 + 3 ** 2 + 10 % 3"}}],
             fmt,
         )
         assert "failed" not in _config(sm)
-        assert sm.model.x == 9  # 0 * 2 + 3 ** 2
+        assert sm.model.x == 10  # 0 * 2 + 3 ** 2 + 10 % 3
 
 
 @pytest.mark.parametrize("fmt", FORMATS)

tests/test_spec_parser_values.py

@@ -86,12 +86,35 @@
 
 
 class TestArithmeticMagnitudeCaps:
-    """``**`` and ``*`` stay usable but cannot blow up: the denial-of-service forms raise,
-    while ordinary and edge-case arithmetic pass through (GHSA-r8gj-366q-cgvj)."""
+    """``**``, ``*``, ``+`` and ``%`` stay usable but cannot blow up: the denial-of-service
+    forms raise, while ordinary and edge-case arithmetic pass through (GHSA-r8gj-366q-cgvj)."""
 
-    @pytest.mark.parametrize("expr", ["9**9**9", "2**100000", "[0]*20000000", "'a'*20000000"])
+    @pytest.mark.parametrize(
+        "expr",
+        [
+            "9**9**9",
+            "2**100000",
+            "[0]*20000000",
+            "'a'*20000000",
+            "3**2048 * 3**2048",
+            "[0]*999999 + [0]*999999",
+            "'a'*999999 + 'a'*999999",
+            "[[0]*1000]*1001",
+            "[[0]*1000]*500 + [[0]*1000]*501",
+            "[('a'*999999,)]*2",
+            "[3**2048]*3000",
+            "{((0,)*1000,)*1000, ((1,)*1000,)*1000}",
+            "{1: [[0]*1000]*600, 2: [[0]*1000]*600}",
+            "[p, p]",
+        ],
+    )
     def test_dos_forms_rejected(self, expr):
         with pytest.raises(ValueError, match="too large"):
+            parse_expr(expr, hook_from({"p": [[0] * 1000] * 600}))()
+
+    @pytest.mark.parametrize("expr", ["'%20000000d' % 0", "b'%d' % 1"])
+    def test_string_formatting_rejected(self, expr):
+        with pytest.raises(ValueError, match="string formatting is not allowed"):
             parse_expr(expr, hook_from({}))()
 
     @pytest.mark.parametrize(
@@ -106,10 +129,29 @@
             ("[0] * 0", []),  # non-positive count: cap check skipped
             ("'ab' * 2", "abab"),  # small str replication
             ("3 * 4", 12),  # scalar multiply: neither operand a sequence
+            ("2.5 * 2", 5.0),  # float operand: int cap check skipped
+            ("3**2048 * 3", 3**2049),  # int product under the cap
+            ("[1] + [2]", [1, 2]),  # small sequence concatenation
+            ("'a' + 'b'", "ab"),  # small str concatenation
+            ("1 + 2", 3),  # scalar add: neither operand a sequence
+            ("10 % 3", 1),  # numeric modulo
+            ("[[0] * 2] * 2", [[0, 0], [0, 0]]),  # small nested replication
+            ("[[], 1.5, None]", [[], 1.5, None]),  # empty and scalar items
+            ("{1: (2, 'a'), 3: {4: 5}}", {1: (2, "a"), 3: {4: 5}}),  # nested dict
+            ("{(1, 2)}", {(1, 2)}),  # set literal
+            ("[d, d]", [{1: 2}, {1: 2}]),  # shared reference counted per appearance
         ],
     )
     def test_bounded_and_edge_cases_pass(self, expr, expected):
-        assert parse_expr(expr, hook_from({}))() == expected
+        assert parse_expr(expr, hook_from({"d": {1: 2}}))() == expected
+
+    def test_sets_are_measured_by_their_items(self):
+        assert parse_expr("[s, s]", hook_from({"s": frozenset({1, 2})}))() == [
+            frozenset({1, 2}),
+            frozenset({1, 2}),
+        ]
+        with pytest.raises(ValueError, match="too large"):
+            parse_expr("[s, s]", hook_from({"s": set(range(600_000))}))()
 
 
 class TestRejectedExpressions: