fix[codegen]: fix _abi_decode buffer overflow

tests/functional/builtins/codegen/test_abi_decode.py

@@ -4,6 +4,7 @@
 from eth.codecs import abi
 
 from vyper.exceptions import ArgumentException, StackTooDeep, StructureException
+from vyper.utils import method_id
 
 TEST_ADDR = "0x" + b"".join(chr(i).encode("utf-8") for i in range(20)).hex()
 
@@ -467,3 +468,77 @@
 @pytest.mark.parametrize("bad_code,exception", FAIL_LIST)
 def test_abi_decode_length_mismatch(get_contract, assert_compile_failed, bad_code, exception):
     assert_compile_failed(lambda: get_contract(bad_code), exception)
+
+
+def test_abi_decode_arithmetic_overflow(w3, tx_failed, get_contract):
+    # test based on GHSA-9p8r-4xp4-gw5w:
+    # https://github.com/vyperlang/vyper/security/advisories/GHSA-9p8r-4xp4-gw5w#advisory-comment-91841
+    # note: doesn't even reach the assert but reverts internally on the clamp in getelemptr
+    code = """
+@external
+def f(x: Bytes[32 * 3]):
+    a: Bytes[32] = b"foo"
+    y: Bytes[32 * 3] = x
+
+    decoded_y1: Bytes[32] = _abi_decode(y, Bytes[32])
+    a = b"bar"
+    decoded_y2: Bytes[32] = _abi_decode(y, Bytes[32])
+
+    assert decoded_y1 != decoded_y2
+    """
+    c = get_contract(code)
+    data = method_id("f(bytes)")
+    data += (0x20).to_bytes(32, "big") # tuple head
+    data += (0x60).to_bytes(32, "big") # parent array length
+    # parent payload - this word will be considered as the head of the abi-encoded inner array
+    # and it will be added to base ptr leading to an arithmetic overflow
+    data += (2**256 - 0x60).to_bytes(32, "big")
+    with tx_failed():
+        w3.eth.send_transaction({"to": c.address, "data": data})
+
+
+
+def test_abi_decode_oob_due_to_invalid_head(w3, tx_failed, get_contract):
+    code = """
+@external
+def f(x: Bytes[32 * 5]):
+    y: Bytes[32 * 5] = x
+    a: Bytes[32] = b"a"
+    decoded_y1: DynArray[uint256, 3] = _abi_decode(y, DynArray[uint256, 3])
+    a = b"aaaa"
+    decoded_y1 = _abi_decode(y, DynArray[uint256, 3])
+    """
+    c = get_contract(code)
+    data = method_id("f(bytes)")
+    data += (0x20).to_bytes(32, "big")  # tuple head
+    data += (0xA0).to_bytes(32, "big")  # parent array length
+    # head should be 20 and thus the decoding func would decode 1 byte
+    # over the end of the input data
+    # _getelemptr_abi_helper will revert due to clamping
+    data += (0x21).to_bytes(32, "big")  # invalid inner array head (1 byte over)
+    # we don't want to revert on invalid length, so set this to 0
+    # the first byte of payload will be considered as the length
+    data += (0x00).to_bytes(32, "big")
+    data += (0x01).to_bytes(1, "big")   # will be considered as the length=1
+    data += (0x00).to_bytes(31, "big")
+    data += (0x03).to_bytes(32, "big") * 2
+    with tx_failed():
+        w3.eth.send_transaction({"to": c.address, "data": data})
+
+
+def test_abi_decode_oob_due_to_invalid_head(tx_failed, get_contract):
+    code = """
+@external
+def bar() -> (uint256, uint256, uint256):
+    return (480, 0, 0)
+
+interface A:
+    def bar() -> String[32]: nonpayable
+
+@external
+def foo():
+    x:String[32] = extcall A(self).bar()
+    """
+    c = get_contract(code)
+    with tx_failed():
+        c.foo()

vyper/builtins/functions.py

@@ -14,6 +14,7 @@
     add_ofst,
     bytes_data_ptr,
     calculate_type_for_external_return,
+    check_buffer_overflow_ir,
     check_external_call,
     clamp,
     clamp2,
@@ -230,18 +231,6 @@ def build_IR(self, expr, context):
 ADHOC_SLICE_NODE_MACROS = ["~calldata", "~selfcode", "~extcode"]
 
 
-# make sure we don't overrun the source buffer, checking for overflow:
-# valid inputs satisfy:
-#   `assert !(start+length > src_len || start+length < start`
-def _make_slice_bounds_check(start, length, src_len):
-    with start.cache_when_complex("start") as (b1, start):
-        with add_ofst(start, length).cache_when_complex("end") as (b2, end):
-            arithmetic_overflow = ["lt", end, start]
-            buffer_oob = ["gt", end, src_len]
-            ok = ["iszero", ["or", arithmetic_overflow, buffer_oob]]
-            return b1.resolve(b2.resolve(["assert", ok]))
-
-
 def _build_adhoc_slice_node(sub: IRnode, start: IRnode, length: IRnode, context: Context) -> IRnode:
     assert length.is_literal, "typechecker failed"
     assert isinstance(length.value, int)  # mypy hint
@@ -254,7 +243,7 @@ def _build_adhoc_slice_node(sub: IRnode, start: IRnode, length: IRnode, context:
     if sub.value == "~calldata":
         node = [
             "seq",
-            _make_slice_bounds_check(start, length, "calldatasize"),
+            check_buffer_overflow_ir(start, length, "calldatasize"),
             ["mstore", np, length],
             ["calldatacopy", np + 32, start, length],
             np,
@@ -264,7 +253,7 @@ def _build_adhoc_slice_node(sub: IRnode, start: IRnode, length: IRnode, context:
     elif sub.value == "~selfcode":
         node = [
             "seq",
-            _make_slice_bounds_check(start, length, "codesize"),
+            check_buffer_overflow_ir(start, length, "codesize"),
             ["mstore", np, length],
             ["codecopy", np + 32, start, length],
             np,
@@ -279,7 +268,7 @@ def _build_adhoc_slice_node(sub: IRnode, start: IRnode, length: IRnode, context:
             sub.args[0],
             [
                 "seq",
-                _make_slice_bounds_check(start, length, ["extcodesize", "_extcode_address"]),
+                check_buffer_overflow_ir(start, length, ["extcodesize", "_extcode_address"]),
                 ["mstore", np, length],
                 ["extcodecopy", "_extcode_address", np + 32, start, length],
                 np,
@@ -452,7 +441,7 @@ def build_IR(self, expr, args, kwargs, context):
 
             ret = [
                 "seq",
-                _make_slice_bounds_check(start, length, src_len),
+                check_buffer_overflow_ir(start, length, src_len),
                 do_copy,
                 ["mstore", dst, length],  # set length
                 dst,  # return pointer to dst

vyper/codegen/core.py

@@ -445,8 +445,9 @@ def _mul(x, y):
 
 
 # Resolve pointer locations for ABI-encoded data
-def _getelemptr_abi_helper(parent, member_t, ofst, clamp=True):
+def _getelemptr_abi_helper(parent, member_t, ofst, clamp_=True):
     member_abi_t = member_t.abi_type
+    parent_abi_t = parent.typ.abi_type
 
     # ABI encoding has length word and then pretends length is not there
     # e.g. [[1,2]] is encoded as 0x01 <len> 0x20 <inner array ofst> <encode(inner array)>
@@ -457,10 +458,21 @@ def _getelemptr_abi_helper(parent, member_t, ofst, clamp=True):
     ofst_ir = add_ofst(parent, ofst)
 
     if member_abi_t.is_dynamic():
+        abi_ofst = unwrap_location(ofst_ir)
         # double dereference, according to ABI spec
         # TODO optimize special case: first dynamic item
         # offset is statically known.
-        ofst_ir = add_ofst(parent, unwrap_location(ofst_ir))
+        ofst_ir = add_ofst(parent, abi_ofst)
+
+        if parent.location == MEMORY:  # TODO: replace with utility function
+            with abi_ofst.cache_when_complex("abi_ofst") as (b1, abi_ofst):
+                bound = parent_abi_t.size_bound()
+                ofst_ir = [
+                    "seq",
+                    check_buffer_overflow_ir(abi_ofst, member_abi_t.size_bound(), bound),
+                    add_ofst(parent, abi_ofst),
+                ]
+                ofst_ir = b1.resolve(ofst_ir)
 
     return IRnode.from_list(
         ofst_ir,
@@ -1230,3 +1242,15 @@ def clamp2(lo, arg, hi, signed):
         LE = "sle" if signed else "le"
         ret = ["seq", ["assert", ["and", [GE, arg, lo], [LE, arg, hi]]], arg]
         return IRnode.from_list(b1.resolve(ret), typ=arg.typ)
+
+
+# make sure we don't overrun the source buffer, checking for overflow:
+# valid inputs satisfy:
+#   `assert !(start+length > src_len || start+length < start)`
+def check_buffer_overflow_ir(start, length, src_len):
+    with start.cache_when_complex("start") as (b1, start):
+        with add_ofst(start, length).cache_when_complex("end") as (b2, end):
+            arithmetic_overflow = ["lt", end, start]
+            buffer_oob = ["gt", end, src_len]
+            ok = ["iszero", ["or", arithmetic_overflow, buffer_oob]]
+            return b1.resolve(b2.resolve(["assert", ok]))