Add an optional max_size parameter to serialize to prevent certain attacks where size can blow up

README.md

@@ -8,6 +8,8 @@ See docs/clvm.org or https://chialisp.com/ for more info.
 Testing
 =======
 
+    $ python3 -m venv venv
+    $ . ./venv/bin/activate
     $ pip install -e '.[dev]'
-    $ py.test tests
+    $ pytest tests
 

clvm/more_ops.py

@@ -242,7 +242,9 @@ def op_point_add(items: _T_SExp) -> typing.Tuple[int, _T_SExp]:
         if _.pair:
             raise EvalError("point_add on list", _)
         try:
-            p += G1Element.from_bytes(_.as_atom())
+            atom = _.as_atom()
+            assert atom is not None
+            p += G1Element.from_bytes(atom)
             cost += POINT_ADD_COST_PER_ARG
         except Exception as ex:
             raise EvalError("point_add expects blob, got %s: %s" % (_, ex), items)

clvm/serialize.py

@@ -17,6 +17,7 @@
 import io
 import typing
 
+from typing import Optional
 from .read_cache_lookup import ReadCacheLookup
 from .object_cache import ObjectCache, treehash, serialized_length
 
@@ -26,6 +27,8 @@
 BACK_REFERENCE = 0xFE
 CONS_BOX_MARKER = 0xFF
 
+MAX_SAFE_BYTES = 2_000_000
+
 T = typing.TypeVar("T")
 _T_CLVMStorage = typing.TypeVar("_T_CLVMStorage", bound=CLVMStorage)
 
@@ -153,9 +156,14 @@ def atom_to_byte_iterator(as_atom: bytes) -> typing.Iterator[bytes]:
 
 
 def sexp_to_stream(
-    sexp: CLVMStorage, f: typing.BinaryIO, *, allow_backrefs: bool = False
+    sexp: CLVMStorage, f: typing.BinaryIO, *, allow_backrefs: bool = False, max_size: Optional[int] = None
 ) -> None:
+    remaining = max_size
     for b in sexp_to_byte_iterator(sexp, allow_backrefs=allow_backrefs):
+        if remaining is not None:
+            remaining -= len(b)
+            if remaining < 0:
+                raise ValueError("SExp exceeds maximum size")
         f.write(b)
 
 

tests/serialize_test.py

@@ -3,13 +3,16 @@
 import unittest
 from typing import Optional
 
+import pytest
+
 from clvm import to_sexp_f
 from clvm.SExp import CastableType, SExp
 from clvm.serialize import (
     _atom_from_stream,
     sexp_from_stream,
     sexp_buffer_from_stream,
     atom_to_byte_iterator,
+    sexp_to_stream,
 )
 
 
@@ -54,13 +57,17 @@ class SerializeTest(unittest.TestCase):
     def check_serde(self, s: CastableType) -> bytes:
         v = to_sexp_f(s)
         b = v.as_bin()
+        f = io.BytesIO()
         v1 = sexp_from_stream(io.BytesIO(b), to_sexp_f)
         if v != v1:
             print("%s: %d %r %s" % (v, len(b), b, v1))
             breakpoint()
             b = v.as_bin()
             v1 = sexp_from_stream(io.BytesIO(b), to_sexp_f)
         self.assertEqual(v, v1)
+        sexp_to_stream(v1, f)
+        length = len(f.getvalue())
+        assert f.getbuffer() == b
         # this copies the bytes that represent a single s-expression, just to
         # know where the message ends. It doesn't build a python representaion
         # of it
@@ -85,6 +92,11 @@ def check_serde(self, s: CastableType) -> bytes:
             self.assertEqual(v2, s)
             b3 = v2.as_bin()
             self.assertEqual(b, b3)
+        with pytest.raises(ValueError):
+            f = io.BytesIO()
+            sexp_to_stream(v1, f, max_size=length-1)
+        f = io.BytesIO()
+        sexp_to_stream(v1, f, max_size=length)
         return b2
 
     def test_zero(self) -> None: