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

@@ -26,6 +26,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 +155,12 @@ 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: int = MAX_SAFE_BYTES
 ) -> None:
     for b in sexp_to_byte_iterator(sexp, allow_backrefs=allow_backrefs):
+        max_size -= len(b)
+        if max_size < 0:
+            raise ValueError("SExp exceeds maximum size")
         f.write(b)
 
 

tests/serialize_test.py

@@ -3,13 +3,17 @@
 import unittest
 from typing import Optional
 
+import pytest
+
 from clvm import to_sexp_f
 from clvm.SExp import CastableType, SExp
 from clvm.serialize import (
+    MAX_SAFE_BYTES,
     _atom_from_stream,
     sexp_from_stream,
     sexp_buffer_from_stream,
     atom_to_byte_iterator,
+    sexp_to_stream,
 )
 
 
@@ -54,13 +58,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 +93,11 @@ def check_serde(self, s: CastableType) -> bytes:
             self.assertEqual(v2, s)
             b3 = v2.as_bin()
             self.assertEqual(b, b3)
+        with pytest.raises(ValueError, match="SExp exceeds maximum size"):
+            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:
@@ -134,10 +147,18 @@ def test_very_long_blobs(self) -> None:
             count = size // len(TEXT)
             text = TEXT * count
             assert len(text) < size
-            self.check_serde(text)
+            if len(text) >= MAX_SAFE_BYTES:
+                with pytest.raises(ValueError, match="SExp exceeds maximum size"):
+                    self.check_serde(text)
+            else:
+                self.check_serde(text)
             text = TEXT * (count + 1)
             assert len(text) > size
-            self.check_serde(text)
+            if len(text) >= MAX_SAFE_BYTES:
+                with pytest.raises(ValueError, match="SExp exceeds maximum size"):
+                    self.check_serde(text)
+            else:
+                self.check_serde(text)
 
     def test_very_deep_tree(self) -> None:
         blob = b"a"
@@ -210,17 +231,16 @@ def make_bomb(depth: int) -> SExp:
         self.assertEqual(len(b10_2), 75)
 
         bomb_20 = make_bomb(20)
-        b20_1 = bomb_20.as_bin(allow_backrefs=False)
+        with pytest.raises(ValueError, match="SExp exceeds maximum size"):
+            bomb_20.as_bin(allow_backrefs=False)
         b20_2 = bomb_20.as_bin(allow_backrefs=True)
-        self.assertEqual(len(b20_1), 48234495)
         self.assertEqual(len(b20_2), 105)
 
         bomb_30 = make_bomb(30)
-        # do not uncomment the next line unless you want to run out of memory
-        # b30_1 = bomb_30.as_bin(allow_backrefs=False)
+        with pytest.raises(ValueError, match="SExp exceeds maximum size"):
+            bomb_30.as_bin(allow_backrefs=False)
         b30_2 = bomb_30.as_bin(allow_backrefs=True)
 
-        # self.assertEqual(len(b30_1), 1)
         self.assertEqual(len(b30_2), 135)
 
     def test_specific_tree(self) -> None: