[WIP] Add manual symbolic calldata support

package/version

@@ -1 +1 @@
-0.1.122
+0.1.123

pyproject.toml

@@ -4,15 +4,15 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "kontrol"
-version = "0.1.122"
+version = "0.1.123"
 description = "Foundry integration for KEVM"
 authors = [
     "Runtime Verification, Inc. <[email protected]>",
 ]
 
 [tool.poetry.dependencies]
 python = "^3.10"
-kevm-pyk = { git = "https://github.com/runtimeverification/evm-semantics.git", tag = "v1.0.422", subdirectory = "kevm-pyk" }
+kevm-pyk = { git = "https://github.com/runtimeverification/evm-semantics.git", branch = "symbolic-calldata", subdirectory = "kevm-pyk" }
 
 [tool.poetry.group.dev.dependencies]
 autoflake = "*"

src/kontrol/__init__.py

@@ -5,4 +5,4 @@
 if TYPE_CHECKING:
     from typing import Final
 
-VERSION: Final = '0.1.122'
+VERSION: Final = '0.1.123'

src/kontrol/kdist/cheatcodes.md

@@ -345,6 +345,7 @@ The `<output>` cell will be updated with the value of the address generated from
          <k> #call_foundry SELECTOR ARGS => . ... </k>
          <output> _ => #bufStrict(32, #addrFromPrivateKey(#unparseDataBytes(ARGS))) </output>
       requires SELECTOR ==Int selector ( "addr(uint256)" )
+      [preserves-definedness]
 ```
 
 #### `load` - Loads a storage slot from an address.
@@ -896,6 +897,7 @@ The `ECDSASign` function returns the signed data in [r,s,v] form, which we conve
          <k> #call_foundry SELECTOR ARGS => . ... </k>
          <output> _ => #sign(#range(ARGS, 32, 32),#range(ARGS,0,32)) </output>
       requires SELECTOR ==Int selector ( "sign(uint256,bytes32)" )
+      [preserves-definedness]
 ```
 
 Otherwise, throw an error for any other call to the Foundry contract.
@@ -943,6 +945,7 @@ Utils
            <code> _ => #if #asWord(CODE) ==Int 0 #then .Bytes #else CODE #fi </code>
            ...
          </account>
+    [preserves-definedness]
 ```
 
 - `#returnNonce ACCTID` takes the nonce of a given account and places it on the `<output>` cell.
@@ -997,6 +1000,7 @@ Utils
            <storage> STORAGE => STORAGE [ LOC <- VALUE ] </storage>
              ...
          </account>
+    [preserves-definedness]
 ```
 
 `#setSymbolicStorage ACCTID` takes a given account and makes its storage fully symbolic.

src/kontrol/prove.py

@@ -14,7 +14,7 @@
 from pyk.prelude.collections import map_empty, map_of
 from pyk.prelude.k import GENERATED_TOP_CELL
 from pyk.prelude.kbool import FALSE, TRUE, notBool
-from pyk.prelude.kint import intToken
+from pyk.prelude.kint import eqInt, intToken, leInt
 from pyk.prelude.ml import mlEqualsTrue
 from pyk.prelude.string import stringToken
 from pyk.proof.proof import Proof
@@ -368,7 +368,7 @@ def _method_to_cfg(
         use_init_code = True
 
     elif isinstance(method, Contract.Method):
-        calldata = method.calldata_cell(contract)
+        calldata, is_calldata_symbolic = method.calldata_cell(contract)
         callvalue = method.callvalue_cell
         program = KEVM.bin_runtime(KApply(f'contract_{contract.name_with_path}'))
         use_init_code = False
@@ -378,6 +378,7 @@ def _method_to_cfg(
         contract.name_with_path,
         program=program,
         calldata=calldata,
+        is_calldata_symbolic=is_calldata_symbolic,
         callvalue=callvalue,
         use_gas=use_gas,
         summary_entries=summary_entries,
@@ -468,6 +469,10 @@ def _update_cterm_from_node(cterm: CTerm, node: KCFG.Node, contract_name: str) -
     new_init_cterm = CTerm(set_cell(new_init_cterm.config, 'GAS_CELL', gas_cell), [])
     new_init_cterm = CTerm(set_cell(new_init_cterm.config, 'CALLGAS_CELL', callgas_cell), [])
 
+    # adding constraints from the initial cterm
+    for constraint in cterm.constraints:
+        new_init_cterm = new_init_cterm.add_constraint(constraint)
+
     return new_init_cterm
 
 
@@ -528,6 +533,7 @@ def _init_cterm(
     *,
     setup_cterm: CTerm | None = None,
     calldata: KInner | None = None,
+    is_calldata_symbolic: bool = False,
     callvalue: KInner | None = None,
     summary_entries: Iterable[SummaryEntry] | None = None,
 ) -> CTerm:
@@ -594,6 +600,14 @@ def _init_cterm(
 
     if calldata is not None:
         init_subst['CALLDATA_CELL'] = calldata
+        constraints = []
+
+        if is_calldata_symbolic:
+            # TODO(palina): EXPERIMENTAL: if calldata is symbolic,
+            # add assumptions that correspond to the test w/`BYTES_DATA` having symbolic length, and `bytes[]` containing
+            # 10 elements, each 600 bytes long
+            structured_calldata_symbolic_data_constraints = structured_calldata_symbolic_data_assumptions()
+            constraints.extend(structured_calldata_symbolic_data_constraints)
 
     if callvalue is not None:
         init_subst['CALLVALUE_CELL'] = callvalue
@@ -614,6 +628,52 @@ def _init_cterm(
         return init_cterm
 
 
+def structured_calldata_symbolic_data_assumptions() -> list[KApply]:
+    constraints = []
+
+    target_constraint = mlEqualsTrue(KEVM.range_address(KVariable('target', 'Int')))
+    sender_constraint = mlEqualsTrue(KEVM.range_address(KVariable('sender', 'Int')))
+    l2_output_constraint = mlEqualsTrue(KEVM.range_uint('256', KVariable('l2OutputIndex', 'Int')))
+    constraints.append(l2_output_constraint)
+
+    constraints.append(target_constraint)
+    constraints.append(sender_constraint)
+
+    length_bytes_data_equality = mlEqualsTrue(
+        eqInt(KEVM.size_bytes(KVariable('BYTES_DATA', 'Bytes')), KVariable('BYTES_SIZE', 'Int'))
+    )
+
+    # Size of the symbolic bytes is limited to 1Gb
+    length_bytes_data_constraint = mlEqualsTrue(leInt(KVariable('BYTES_SIZE', 'Int'), intToken("1073741824")))
+
+    constraints.append(length_bytes_data_equality)
+    constraints.append(length_bytes_data_constraint)
+
+    length_bytes_arr_constraint_1 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_1', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_1)
+    length_bytes_arr_constraint_2 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_2', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_2)
+    length_bytes_arr_constraint_3 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_3', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_3)
+    length_bytes_arr_constraint_4 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_4', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_4)
+    length_bytes_arr_constraint_5 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_5', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_5)
+    length_bytes_arr_constraint_6 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_6', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_6)
+    length_bytes_arr_constraint_7 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_7', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_7)
+    length_bytes_arr_constraint_8 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_8', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_8)
+    length_bytes_arr_constraint_9 = mlEqualsTrue(eqInt(KEVM.size_bytes(KVariable('BYTES_9', 'Bytes')), intToken("600")))
+    constraints.append(length_bytes_arr_constraint_9)
+    length_bytes_arr_constraint_10 = mlEqualsTrue(
+        eqInt(KEVM.size_bytes(KVariable('BYTES_10', 'Bytes')), intToken("600"))
+    )
+    constraints.append(length_bytes_arr_constraint_10)
+    return constraints
+
+
 def _final_cterm(
     empty_config: KInner, contract_name: str, *, failing: bool, is_test: bool = True, use_init_code: bool = False
 ) -> CTerm:

src/kontrol/solc_to_k.py

@@ -72,6 +72,10 @@ class Input:
     components: tuple[Input, ...] = ()
     idx: int = 0
 
+    @cached_property
+    def arg_name(self) -> str:
+        return f'V{self.idx}_{self.name.replace("-", "_")}'
+
     @staticmethod
     def from_dict(input: dict, idx: int = 0) -> Input:
         name = input['name']
@@ -82,30 +86,39 @@ def from_dict(input: dict, idx: int = 0) -> Input:
             return Input(name, type, idx=idx)
 
     @staticmethod
-    def arg_name(input: Input) -> str:
-        return f'V{input.idx}_{input.name.replace("-", "_")}'
-
-    @staticmethod
-    def _make_single_type(input: Input) -> KApply:
-        input_name = Input.arg_name(input)
-        input_type = input.type
-        return KEVM.abi_type(input_type, KVariable(input_name))
+    def _make_single_type(input: Input) -> (KApply, bool):
+        if input.type == 'bytes':
+            # getting the type
+            element_type = KEVM.abi_type(input.type, KVariable(input.arg_name))
+            return (KEVM.abi_dynamic_bytes_array(element_type), True)
+        elif input.type.endswith('[]'):
+            element_type = KEVM.abi_type(input.type[0 : input.type.index('[')], KVariable(input.arg_name))
+            if element_type == 'bytes':
+                return (KEVM.abi_dynamic_bytes_array(element_type), True)
+            else:
+                return (KEVM.abi_dynamic_array(element_type), True)
+        else:
+            input_name = input.arg_name
+            input_type = input.type
+            return (KEVM.abi_type(input_type, KVariable(input_name)), False)
 
     @staticmethod
-    def _make_complex_type(components: Iterable[Input]) -> KApply:
+    def _make_complex_type(components: Iterable[Input]) -> (KApply, bool):
+        # TODO(palina): add support for static arrays
         """
         recursively unwrap components in arguments of complex types and convert them to KEVM types
         """
+        has_dynamic_type = False
         abi_types: list[KInner] = []
         for comp in components:
             # nested tuple, unwrap its components
             if comp.type == 'tuple':
-                tuple = Input._make_complex_type(comp.components)
+                tuple, has_dynamic_type = Input._make_complex_type(comp.components)
                 abi_type = tuple
             else:
-                abi_type = Input._make_single_type(comp)
+                abi_type, has_dynamic_type = Input._make_single_type(comp)
             abi_types.append(abi_type)
-        return KEVM.abi_tuple(abi_types)
+        return (KEVM.abi_tuple(abi_types), has_dynamic_type)
 
     @staticmethod
     def _unwrap_components(components: dict, i: int = 0) -> list[Input]:
@@ -124,7 +137,7 @@ def _unwrap_components(components: dict, i: int = 0) -> list[Input]:
             i += 1
         return comps
 
-    def to_abi(self) -> KApply:
+    def to_abi(self) -> (KApply, bool):
         if self.type == 'tuple':
             return Input._make_complex_type(self.components)
         else:
@@ -282,7 +295,7 @@ def flat_inputs(self) -> tuple[Input, ...]:
 
         @cached_property
         def arg_names(self) -> Iterable[str]:
-            return tuple(Input.arg_name(input) for input in self.flat_inputs)
+            return tuple(input.arg_name for input in self.flat_inputs)
 
         @cached_property
         def arg_types(self) -> Iterable[str]:
@@ -351,7 +364,7 @@ def rule(self, contract: KInner, application_label: KLabel, contract_name: str)
             args: list[KInner] = []
             conjuncts: list[KInner] = []
             for input in self.inputs:
-                abi_type = input.to_abi()
+                abi_type, _ = input.to_abi()
                 args.append(abi_type)
                 rps = _range_predicates(abi_type)
                 for rp in rps:
@@ -374,8 +387,20 @@ def callvalue_cell(self) -> KInner:
                 else abstract_term_safely(KVariable('_###CALLVALUE###_'), base_name='CALLVALUE')
             )
 
-        def calldata_cell(self, contract: Contract) -> KInner:
-            return KApply(contract.klabel_method, [KApply(contract.klabel), self.application])
+        def calldata_cell(self, contract: Contract) -> (KInner, bool):
+            has_dynamic_types = False
+            args: list[KInner] = []
+
+            for input in self.inputs:
+                abi_type, input_has_dynamic_types = input.to_abi()
+                # if any of the inputs contain dynamic types, make it symbolic
+                has_dynamic_types = has_dynamic_types or input_has_dynamic_types
+                args.append(abi_type)
+
+            if has_dynamic_types:
+                return (KEVM.abi_symbolic_calldata(self.name, args), True)
+            else:
+                return (KApply(contract.klabel_method, [KApply(contract.klabel), self.application]), False)
 
         @cached_property
         def application(self) -> KInner:
@@ -827,6 +852,10 @@ def _range_predicates(abi: KApply) -> list[KInner | None]:
     if abi.label.name == 'abi_type_tuple':
         if type(abi.args[0]) is KApply:
             rp += _range_collection_predicates(abi.args[0])
+    elif abi.label.name == 'abi_dynamic_array':
+        # TODO(palina): support other arrays
+        type_label = "bytes"
+        rp.append(_range_predicate(single(abi.args), type_label))
     else:
         type_label = abi.label.name.removeprefix('abi_type_')
         rp.append(_range_predicate(single(abi.args), type_label))