CVE-2022-24787 (Python, published 2022-04-04T18:15Z)
Description: value: Vyper is a Pythonic Smart Contract Language for the Ethereum Virtual Machine. In version 0.3.1 and prior, bytestrings can have dirty bytes in them, resulting in the word-for-word comparisons giving incorrect results. Even without dirty nonzero bytes, two bytestrings can compare to equal if one ends with `"\x00"` because there is no comparison of the length. A patch is available and expected to be part of the 0.3.2 release. There are currently no known workarounds.
Severity: MEDIUM (CVSS v3 base score: 7.5)
Weakness: CWE-697 \u2014 Incorrect Comparison
The product compares two entities in a security-relevant context, but the comparison is incorrect, which may lead to resultant weaknesses.
Repository: https://github.com/vyperlang/vyper
Files changed:
tests/parser/functions/test_slice.py
tests/parser/types/test_string.py
vyper/codegen/core.py
vyper/codegen/expr.py
vyper/codegen/keccak256_helper.py
Commit message: Merge pull request from GHSA-7vrm-3jc8-5wwm

* add more tests for string comparison

explicitly test the codepath with <= 32 bytes

* refactor keccak256 helper a bit

* fix bytestring equality

existing bytestring equality checks do not check length equality or for
dirty bytes.

Vulnerable code:
def test_string_equality(get_contract_with_gas_estimation):
    code = """
_compA: String[100]
_compB: String[100]
    compA: String[100] = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy dog"
    return compA == compB
    compA: String[100] = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy hog"
    return compA == compB
    compA: String[100] = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy hog"
    return compA != compB
    compA: String[100] = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy dog"
    return compA != compB
    return "The quick brown fox jumps over the lazy dog" == \
    "The quick brown fox jumps over the lazy dog"
    return "The quick brown fox jumps over the lazy dog" == \
    "The quick brown fox jumps over the lazy hog"
    return "The quick brown fox jumps over the lazy dog" != \
    "The quick brown fox jumps over the lazy hog"
    return "The quick brown fox jumps over the lazy dog" != \
    "The quick brown fox jumps over the lazy dog"
    self._compA = "The quick brown fox jumps over the lazy dog"
    self._compB = "The quick brown fox jumps over the lazy dog"
    return self._compA == self._compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    self._compB = "The quick brown fox jumps over the lazy hog"
    return self._compA == self._compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    self._compB = "The quick brown fox jumps over the lazy hog"
    return self._compA != self._compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    self._compB = "The quick brown fox jumps over the lazy dog"
    return self._compA != self._compB
def string_compare_equal(str1: String[100], str2: String[100]) -> bool:
def string_compare_not_equal(str1: String[100], str2: String[100]) -> bool:
def compare_passed_storage_equal(str: String[100]) -> bool:
    self._compA = "The quick brown fox jumps over the lazy dog"
    return self._compA == str
def compare_passed_storage_not_equal(str: String[100]) -> bool:
    self._compA = "The quick brown fox jumps over the lazy dog"
    return self._compA != str
    self._compA = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy dog"
    return self._compA == compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy hog"
    return self._compA == compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy hog"
    return self._compA != compB
    self._compA = "The quick brown fox jumps over the lazy dog"
    compB: String[100] = "The quick brown fox jumps over the lazy dog"
    return self._compA != compB
    a = "The quick brown fox jumps over the lazy dog"
    b = "The quick brown fox jumps over the lazy hog"
    ) as (b2, length,), dst.cache_when_complex("dst") as (b3, dst):
    LOAD,
    bytes_data_ptr,
            length_mismatch = left.typ.maxlen != right.typ.maxlen
            left_over_32 = left.typ.maxlen > 32
            right_over_32 = right.typ.maxlen > 32

            if length_mismatch or left_over_32 or right_over_32:
                left_keccak = keccak256_helper(self.expr, left, self.context)
                right_keccak = keccak256_helper(self.expr, right, self.context)

                if op == "eq" or op == "ne":
                    return IRnode.from_list([op, left_keccak, right_keccak], typ="bool")

                else:
                    return

                def load_bytearray(side):
                    return LOAD(bytes_data_ptr(side))

                return IRnode.from_list(
                    # CMC 2022-03-24 TODO investigate this.
                    [op, load_bytearray(left), load_bytearray(right)],
                    typ="bool",
                )
from vyper.codegen.core import ensure_in_memory
def keccak256_helper(expr, ir_arg, context):
    sub = ir_arg  # TODO get rid of useless variable
    _check_byteslike(sub.typ, expr)
    if isinstance(sub, bytes):
        return IRnode.from_list(bytes_to_int(keccak256(sub)), typ=BaseType("bytes32"))
    if is_base_type(sub.typ, "bytes32"):
                ["mstore", MemoryPositions.FREE_VAR_SPACE, sub],
    sub = ensure_in_memory(sub, context)

    return IRnode.from_list(
        [
            "with",
            "_buf",
            sub,
            ["sha3", ["add", "_buf", 32], ["mload", "_buf"]],
        ],
        typ=BaseType("bytes32"),
        annotation="keccak256",
        add_gas_estimate=_gas_bound(ceil(sub.typ.maxlen / 32)),
    )

Fixed code:
def test_slice_equality(get_contract):
    # test for equality with dirty bytes
    code = """
@external
def assert_eq() -> bool:
    dirty_bytes: String[4] = "abcd"
    dirty_bytes = slice(dirty_bytes, 0, 3)
    clean_bytes: String[4] = "abc"
    return dirty_bytes == clean_bytes
    """

    c = get_contract(code)
    assert c.assert_eq()


def test_slice_inequality(get_contract):
    # test for equality with dirty bytes
    code = """
@external
def assert_ne() -> bool:
    dirty_bytes: String[4] = "abcd"
    dirty_bytes = slice(dirty_bytes, 0, 3)
    clean_bytes: String[4] = "abcd"
    return dirty_bytes != clean_bytes
    """

    c = get_contract(code)
    assert c.assert_ne()


string_equality_tests = [
    (
        100,
        "The quick brown fox jumps over the lazy dog",
        "The quick brown fox jumps over the lazy hog",
    ),
    # check <= 32 codepath
    (32, "abc", "abc\0"),
    (32, "abc", "abc\1"),  # use a_init dirty bytes
    (32, "abc\2", "abc"),  # use b_init dirty bytes
    (32, "", "\0"),
    (32, "", "\1"),
    (33, "", "\1"),
    (33, "", "\0"),
]


@pytest.mark.parametrize("len_,a,b", string_equality_tests)
def test_string_equality(get_contract_with_gas_estimation, len_, a, b):
    # fixtures to initialize strings with dirty bytes
    a_init = "\\1" * len_
    b_init = "\\2" * len_
    string1 = a.encode("unicode_escape").decode("utf-8")
    string2 = b.encode("unicode_escape").decode("utf-8")
    code = f"""
a: String[{len_}]
b: String[{len_}]
    a: String[{len_}] = "{a_init}"
    b: String[{len_}] = "{b_init}"
    a = "{string1}"
    b = "{string1}"
    return a == b
    a: String[{len_}] = "{a_init}"
    b: String[{len_}] = "{b_init}"
    a = "{string1}"
    b = "{string2}"
    return a == b
    a: String[{len_}] = "{a_init}"
    b: String[{len_}] = "{b_init}"
    a = "{string1}"
    b = "{string2}"
    return a != b
    a: String[{len_}] = "{a_init}"
    b: String[{len_}] = "{b_init}"
    a = "{string1}"
    b = "{string1}"
    return a != b
    return "{string1}" == "{string1}"
    return "{string1}" == "{string2}"
    return "{string1}" != "{string2}"
    return "{string1}" != "{string1}"
    self.a = "{a_init}"
    self.b = "{b_init}"
    self.a = "{string1}"
    self.b = "{string1}"
    return self.a == self.b
    self.a = "{a_init}"
    self.b = "{b_init}"
    self.a = "{string1}"
    self.b = "{string2}"
    return self.a == self.b
    self.a = "{a_init}"
    self.b = "{b_init}"
    self.a = "{string1}"
    self.b = "{string2}"
    return self.a != self.b
    self.a = "{a_init}"
    self.b = "{b_init}"
    self.a = "{string1}"
    self.b = "{string1}"
    return self.a != self.b
def string_compare_equal(str1: String[{len_}], str2: String[{len_}]) -> bool:
def string_compare_not_equal(str1: String[{len_}], str2: String[{len_}]) -> bool:
def compare_passed_storage_equal(str_: String[{len_}]) -> bool:
    self.a = "{a_init}"
    self.a = "{string1}"
    return self.a == str_
def compare_passed_storage_not_equal(str_: String[{len_}]) -> bool:
    self.a = "{a_init}"
    self.a = "{string1}"
    return self.a != str_
    self.a = "{a_init}"
    b: String[{len_}] = "{b_init}"
    self.a = "{string1}"
    b = "{string1}"
    return self.a == b
    self.a = "{a_init}"
    b: String[{len_}] = "{b_init}"
    self.a = "{string1}"
    b = "{string2}"
    return self.a == b
    self.a = "{a_init}"
    b: String[{len_}] = "{b_init}"
    self.a = "{string1}"
    b = "{string2}"
    return self.a != b
    self.a = "{a_init}"
    b: String[{len_}] = "{b_init}"
    self.a = "{string1}"
    b = "{string1}"
    return self.a != b
    ) as (b2, length), dst.cache_when_complex("dst") as (b3, dst):
            left_keccak = keccak256_helper(self.expr, left, self.context)
            right_keccak = keccak256_helper(self.expr, right, self.context)
            if op not in ("eq", "ne"):
                return  # raises
                # use hash even for Bytes[N<=32], because there could be dirty
                # bytes past the bytes data.
                return IRnode.from_list([op, left_keccak, right_keccak], typ="bool")
from vyper.codegen.core import bytes_data_ptr, ensure_in_memory, get_bytearray_length
def keccak256_helper(expr, to_hash, context):
    _check_byteslike(to_hash.typ, expr)
    if isinstance(to_hash, bytes):
        return IRnode.from_list(bytes_to_int(keccak256(to_hash)), typ=BaseType("bytes32"))
    if is_base_type(to_hash.typ, "bytes32"):
                ["mstore", MemoryPositions.FREE_VAR_SPACE, to_hash],
    to_hash = ensure_in_memory(to_hash, context)

    with to_hash.cache_when_complex("buf") as (b1, to_hash):
        data = bytes_data_ptr(to_hash)
        len_ = get_bytearray_length(to_hash)
        return b1.resolve(
            IRnode.from_list(
                ["sha3", data, len_],
                typ="bytes32",
                annotation="keccak256",
                add_gas_estimate=_gas_bound(ceil(to_hash.typ.maxlen / 32)),
            )
        )