fix[codegen]: zero-length dynarray abi_decode validation (vyperlang…

…#4060)

fix an edge case in `abi_decode` dynarray validation. when the child
type is dynamic and the runtime length is zero, the check that the
offset pointer is valid (points within the payload) was skipped.

skipping the offset pointer check is valid any time the runtime
length is nonzero, because the pointer is bounded by the checks in
the recursive runtime loop in `_dynarray_make_setter`. however, it is
invalid to skip the check when the runtime length of the dynarray is
zero, because then the recursive loop does not get run.

the impact of this can be seen in the included test cases, particularly
`test_abi_decode_top_level_head_oob`. although as of eb01136
it is impossible to convince the decoder to *copy* oob data since the
validation is only skipped when the length is zero, a payload can be
crafted which will revert depending on if some value outside of the
buffer is nonzero (i.e. the runtime behavior can be influenced by some
data outside of the payload).

this commit fixes this issue by _unconditionally_ checking that the
offset pointer is valid. note that the check is now always performed,
even when the runtime length is nonzero and therefore the check is
redundant (because, as stated, the checks within the loop already bound
the offset pointer).

a more efficient implementation is possible, since the check only needs
to be run in the case that the runtime length is 0, which theoretically
can be merged into the same basic block with the 0-case in the `repeat`
loop. however, this commit leaves that to future optimizer work; the
optimization here is it just avoids the multiplication when the child
type is dynamic (because the result of the multiplication is always 0).

this commit also fixes another bug in dynarray recursion; the
calculation in `_abi_payload_size` was not correct when the size of the
child type is larger than 32.

misc:
- add additional tests for abi_decode validation.

---------

Co-authored-by: cyberthirst <cyberthirst.eth@gmail.com>
Co-authored-by: Robert Chen <me@robertchen.cc>

tests/functional/builtins/codegen/test_abi_decode.py

@@ -495,8 +495,8 @@ def f(x: Bytes[32 * 3]):
     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
+    # original POC:
+    # assert decoded_y1 != decoded_y2
     """
     c = get_contract(code)
 
@@ -1043,7 +1043,7 @@ def run():
         c.run()
 
 
-def test_abi_decode_extcall_zero_len_array(get_contract):
+def test_abi_decode_extcall_empty_array(get_contract):
     code = """
 @external
 def bar() -> (uint256, uint256):
@@ -1061,6 +1061,59 @@ def run():
     c.run()
 
 
+def test_abi_decode_extcall_complex_empty_dynarray(get_contract):
+    # 5th word of the payload points to the last word of the payload
+    # which is considered the length of the Point.y array
+    # because the length is 0, the decoding should succeed
+    code = """
+struct Point:
+    x: uint256
+    y: DynArray[uint256, 2]
+    z: uint256
+
+@external
+def bar() -> (uint256, uint256, uint256, uint256, uint256, uint256):
+    return 32, 1, 32, 1, 64, 0
+
+interface A:
+    def bar() -> DynArray[Point, 2]: nonpayable
+
+@external
+def run():
+    x: DynArray[Point, 2] = extcall A(self).bar()
+    assert len(x) == 1 and len(x[0].y) == 0
+    """
+    c = get_contract(code)
+
+    c.run()
+
+
+def test_abi_decode_extcall_complex_empty_dynarray2(tx_failed, get_contract):
+    # top-level head points 1B over the runtime buffer end
+    # thus the decoding should fail although the length is 0
+    code = """
+struct Point:
+    x: uint256
+    y: DynArray[uint256, 2]
+    z: uint256
+
+@external
+def bar() -> (uint256, uint256):
+    return 33, 0
+
+interface A:
+    def bar() -> DynArray[Point, 2]: nonpayable
+
+@external
+def run():
+    x: DynArray[Point, 2] = extcall A(self).bar()
+    """
+    c = get_contract(code)
+
+    with tx_failed():
+        c.run()
+
+
 def test_abi_decode_extcall_zero_len_array2(get_contract):
     code = """
 @external
@@ -1080,3 +1133,193 @@ def run() -> uint256:
     length = c.run()
 
     assert length == 0
+
+
+def test_abi_decode_top_level_head_oob(tx_failed, get_contract):
+    code = """
+@external
+def run(x: Bytes[256], y: uint256):
+    player_lost: bool = empty(bool)
+
+    if y == 1:
+        player_lost = True
+
+    decoded: DynArray[Bytes[1], 2] = empty(DynArray[Bytes[1], 2])
+    decoded = _abi_decode(x, DynArray[Bytes[1], 2])
+    """
+    c = get_contract(code)
+
+    # head points over the buffer end
+    payload = (0x0100, *_replicate(0x00, 7))
+
+    data = _abi_payload_from_tuple(payload)
+
+    with tx_failed():
+        c.run(data, 1)
+
+    with tx_failed():
+        c.run(data, 0)
+
+
+def test_abi_decode_dynarray_complex_insufficient_data(env, tx_failed, get_contract):
+    code = """
+struct Point:
+    x: uint256
+    y: uint256
+
+@external
+def run(x: Bytes[32 * 8]):
+    y: Bytes[32 * 8] = x
+    decoded_y1: DynArray[Point, 3] = _abi_decode(y, DynArray[Point, 3])
+    """
+    c = get_contract(code)
+
+    # runtime buffer has insufficient size - we decode 3 points, but provide only
+    # 3 * 32B of payload
+    payload = (0x20, 0x03, *_replicate(0x03, 3))
+
+    data = _abi_payload_from_tuple(payload)
+
+    with tx_failed():
+        c.run(data)
+
+
+def test_abi_decode_dynarray_complex2(env, tx_failed, get_contract):
+    # point head to the 1st 0x01 word (ie the length)
+    # but size of the point is 3 * 32B, thus we'd decode 2B over the buffer end
+    code = """
+struct Point:
+    x: uint256
+    y: uint256
+    z: uint256
+
+
+@external
+def run(x: Bytes[32 * 8]):
+    y: Bytes[32 * 11] = x
+    decoded_y1: DynArray[Point, 2] = _abi_decode(y, DynArray[Point, 2])
+    """
+    c = get_contract(code)
+
+    payload = (
+        0xC0,  # points to the 1st 0x01 word (ie the length)
+        *_replicate(0x03, 5),
+        *_replicate(0x01, 2),
+    )
+
+    data = _abi_payload_from_tuple(payload)
+
+    with tx_failed():
+        c.run(data)
+
+
+def test_abi_decode_complex_empty_dynarray(env, tx_failed, get_contract):
+    # point head to the last word of the payload
+    # this will be the length, but because it's set to 0, the decoding should succeed
+    code = """
+struct Point:
+    x: uint256
+    y: DynArray[uint256, 2]
+    z: uint256
+
+
+@external
+def run(x: Bytes[32 * 16]):
+    y: Bytes[32 * 16] = x
+    decoded_y1: DynArray[Point, 2] = _abi_decode(y, DynArray[Point, 2])
+    assert len(decoded_y1) == 1 and len(decoded_y1[0].y) == 0
+    """
+    c = get_contract(code)
+
+    payload = (
+        0x20,
+        0x01,
+        0x20,
+        0x01,
+        0xA0,  # points to the last word of the payload
+        0x04,
+        0x02,
+        0x02,
+        0x00,  # length is 0, so decoding should succeed
+    )
+
+    data = _abi_payload_from_tuple(payload)
+
+    c.run(data)
+
+
+def test_abi_decode_complex_arithmetic_overflow(tx_failed, get_contract):
+    # inner head roundtrips due to arithmetic overflow
+    code = """
+struct Point:
+    x: uint256
+    y: DynArray[uint256, 2]
+    z: uint256
+
+
+@external
+def run(x: Bytes[32 * 16]):
+    y: Bytes[32 * 16] = x
+    decoded_y1: DynArray[Point, 2] = _abi_decode(y, DynArray[Point, 2])
+    """
+    c = get_contract(code)
+
+    payload = (
+        0x20,
+        0x01,
+        0x20,
+        0x01,  # both Point.x and Point.y length
+        2**256 - 0x20,  # points to the "previous" word of the payload
+        0x04,
+        0x02,
+        0x02,
+        0x00,
+    )
+
+    data = _abi_payload_from_tuple(payload)
+
+    with tx_failed():
+        c.run(data)
+
+
+def test_abi_decode_empty_toplevel_dynarray(get_contract):
+    code = """
+@external
+def run(x: Bytes[2 * 32 + 3 * 32  + 3 * 32 * 4]):
+    y: Bytes[2 * 32 + 3 * 32 + 3 * 32 * 4] = x
+    assert len(y) == 2 * 32
+    decoded_y1: DynArray[DynArray[uint256, 3], 3] = _abi_decode(
+        y,
+        DynArray[DynArray[uint256, 3], 3]
+    )
+    assert len(decoded_y1) == 0
+    """
+    c = get_contract(code)
+
+    payload = (0x20, 0x00)  # DynArray head, DynArray length
+
+    data = _abi_payload_from_tuple(payload)
+
+    c.run(data)
+
+
+def test_abi_decode_invalid_toplevel_dynarray_head(tx_failed, get_contract):
+    # head points 1B over the bounds of the runtime buffer
+    code = """
+@external
+def run(x: Bytes[2 * 32 + 3 * 32  + 3 * 32 * 4]):
+    y: Bytes[2 * 32 + 3 * 32 + 3 * 32 * 4] = x
+    decoded_y1: DynArray[DynArray[uint256, 3], 3] = _abi_decode(
+        y,
+        DynArray[DynArray[uint256, 3], 3]
+    )
+    """
+    c = get_contract(code)
+
+    # head points 1B over the bounds of the runtime buffer
+    payload = (0x21, 0x00)  # DynArray head, DynArray length
+
+    data = _abi_payload_from_tuple(payload)
+
+    with tx_failed():
+        c.run(data)

vyper/codegen/core.py

@@ -889,11 +889,17 @@ def _dirty_read_risk(ir_node):
 def _abi_payload_size(ir_node):
     SCALE = ir_node.location.word_scale
     assert SCALE == 32  # we must be in some byte-addressable region, like memory
-
     OFFSET = DYNAMIC_ARRAY_OVERHEAD * SCALE
 
     if isinstance(ir_node.typ, DArrayT):
-        return ["add", OFFSET, ["mul", get_dyn_array_count(ir_node), SCALE]]
+        # the amount of size each value occupies in static section
+        # (the amount of size it occupies in the dynamic section is handled in
+        # make_setter recursion)
+        item_size = ir_node.typ.value_type.abi_type.static_size()
+        if item_size == 0:
+            # manual optimization; the mload cannot currently be optimized out
+            return ["add", OFFSET, 0]
+        return ["add", OFFSET, ["mul", get_dyn_array_count(ir_node), item_size]]
 
     if isinstance(ir_node.typ, _BytestringT):
         return ["add", OFFSET, get_bytearray_length(ir_node)]
@@ -1175,14 +1181,17 @@ def clamp_dyn_array(ir_node, hi=None):
 
     assert (hi is not None) == _dirty_read_risk(ir_node)
 
-    # if the subtype is dynamic, the check will be performed in the recursion
-    if hi is not None and not t.abi_type.subtyp.is_dynamic():
+    if hi is not None:
         assert t.count < 2**64  # sanity check
 
         # note: this add does not risk arithmetic overflow because
         # length is bounded by count * elemsize.
         item_end = add_ofst(ir_node, _abi_payload_size(ir_node))
 
+        # if the subtype is dynamic, the length check is performed in
+        # the recursion, UNLESS the count is zero. here we perform the
+        # check all the time, but it could maybe be optimized out in the
+        # make_setter loop (in the common case that runtime count > 0).
         len_check = ["seq", ["assert", ["le", item_end, hi]], len_check]
 
     return IRnode.from_list(len_check, error_msg=f"{ir_node.typ} bounds check")