chore(fw): raise an exception if an opcode recieves an invalid keyword argument (#1739)

* chore(fw): raise an exception if an opcode recieves an invalid keyword argument

* Add kwargs validation in Opcode.__call__ method in opcode.py
* Add test coverage in test_vm.py for validation behavior

* fix(tests): Incorrect kwargs usages

* fix(vm): tox, mypy

* fix(tests): kwargs fix

* fix: mypy

* fix(tests): Fix coverage

* Update docs/CHANGELOG.md

---------

Co-authored-by: Mario Vega <[email protected]>

Author: richardgreg

docs/CHANGELOG.md

@@ -28,6 +28,7 @@ Users can select any of the artifacts depending on their testing needs for their
 #### 🔀 Refactoring
 
 - 🔀 Move `TransactionType` enum from test file to proper module location in `ethereum_test_types.transaction_types` for better code organization and reusability.
+- ✨ Opcode classes now validate keyword arguments and raise `ValueError` with clear error messages.
 
 #### `fill`
 

src/ethereum_test_vm/opcode.py

@@ -87,7 +87,7 @@ class Opcode(Bytecode):
     data_portion_length: int
     data_portion_formatter: Optional[Callable[[Any], bytes]]
     stack_properties_modifier: Optional[Callable[[Any], tuple[int, int, int, int]]]
-    kwargs: List[str] | None
+    kwargs: List[str]
     kwargs_defaults: KW_ARGS_DEFAULTS_TYPE
     unchecked_stack: bool = False
 
@@ -137,7 +137,10 @@ def __new__(
             obj.data_portion_formatter = data_portion_formatter
             obj.stack_properties_modifier = stack_properties_modifier
             obj.unchecked_stack = unchecked_stack
-            obj.kwargs = kwargs
+            if kwargs is None:
+                obj.kwargs = []
+            else:
+                obj.kwargs = kwargs
             obj.kwargs_defaults = kwargs_defaults
             return obj
         raise TypeError("Opcode constructor '__new__' didn't return an instance!")
@@ -254,6 +257,16 @@ def __call__(
 
         if self.kwargs is not None and len(kwargs) > 0:
             assert len(args) == 0, f"Cannot mix positional and keyword arguments {args} {kwargs}"
+
+            # Validate that all provided kwargs are valid
+            invalid_kwargs = set(kwargs.keys()) - set(self.kwargs)
+            if invalid_kwargs:
+                raise ValueError(
+                    f"Invalid keyword argument(s) {list(invalid_kwargs)} for opcode "
+                    f"{self._name_}. Valid arguments are: {self.kwargs}"
+                    f"Valid arguments are: {self.kwargs}"
+                )
+
             for kw in self.kwargs:
                 args.append(kwargs[kw] if kw in kwargs else self.kwargs_defaults.get(kw, 0))
 

src/ethereum_test_vm/tests/test_vm.py

@@ -431,3 +431,38 @@ def test_bytecode_concatenation_with_bytes():
     assert code.max_stack_height == base.max_stack_height
     assert code.min_stack_height == base.min_stack_height
     assert code.terminating == base.terminating
+
+
+def test_opcode_kwargs_validation():
+    """Test that invalid keyword arguments raise ValueError."""
+    # Test valid kwargs work
+    Op.MSTORE(offset=0, value=1)
+    Op.CALL(gas=1, address=2, value=3, args_offset=4, args_size=5, ret_offset=6, ret_size=7)
+
+    # Test invalid kwargs raise ValueError
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['offest'\] for opcode MSTORE"
+    ):
+        Op.MSTORE(offest=0, value=1)  # codespell:ignore offest
+
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['wrong_arg'\] for opcode MSTORE"
+    ):
+        Op.MSTORE(offset=0, value=1, wrong_arg=2)
+
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['addres'\] for opcode CALL"
+    ):
+        Op.CALL(
+            gas=1,
+            addres=2,  # codespell:ignore
+            value=3,
+            args_offset=4,
+            args_size=5,
+            ret_offset=6,
+            ret_size=7,
+        )
+
+    # Test multiple invalid kwargs
+    with pytest.raises(ValueError, match=r"Invalid keyword argument\(s\).*for opcode MSTORE"):
+        Op.MSTORE(offest=0, valu=1, extra=2)  # codespell:ignore offest,valu

tests/cancun/eip1153_tstore/test_tstorage_create_contexts.py

@@ -155,7 +155,16 @@ def creator_contract_code(  # noqa: D102
             + Op.TSTORE(1, 0x0200)
             + Op.TSTORE(2, 0x0300)
             + Op.CALLDATACOPY(0, 0, Op.CALLDATASIZE)
-            + Op.SSTORE(4, Op.CALL(address=opcode(size=Op.CALLDATASIZE, salt=create2_salt)))
+            + Op.SSTORE(
+                4,
+                Op.CALL(
+                    address=(
+                        opcode(size=Op.CALLDATASIZE, salt=create2_salt)
+                        if opcode == Op.CREATE2
+                        else opcode(size=Op.CALLDATASIZE)
+                    )
+                ),
+            )
             # Save the state of transient storage following call to storage; the transient
             # storage should not have been overwritten
             + Op.SSTORE(0, Op.TLOAD(0))

tests/cancun/eip4844_blobs/test_blobhash_opcode_contexts.py

@@ -132,7 +132,11 @@ def deploy_contract(
                 create_opcode = Op.CREATE if self == BlobhashContext.CREATE else Op.CREATE2
                 create_bytecode = Op.EXTCODECOPY(
                     address=initcode_address, dest_offset=0, offset=0, size=len(initcode)
-                ) + Op.POP(create_opcode(value=0, offset=0, size=len(initcode), salt=0))
+                ) + Op.POP(
+                    create_opcode(value=0, offset=0, size=len(initcode), salt=0)
+                    if create_opcode == Op.CREATE2
+                    else create_opcode(value=0, offset=0, size=len(initcode))
+                )
                 return pre.deploy_contract(create_bytecode)
 
             case _:

tests/cancun/eip6780_selfdestruct/test_reentrancy_selfdestruct_revert.py

@@ -53,7 +53,14 @@ def executor_contract_bytecode(
 ) -> Bytecode:
     """Contract code that performs a selfdestruct call then revert."""
     return (
-        Op.SSTORE(1, first_suicide(address=selfdestruct_contract_address, value=0))
+        Op.SSTORE(
+            1,
+            (
+                first_suicide(address=selfdestruct_contract_address, value=0)
+                if first_suicide in [Op.CALL, Op.CALLCODE]
+                else first_suicide(address=selfdestruct_contract_address)
+            ),
+        )
         + Op.SSTORE(2, Op.CALL(address=revert_contract_address))
         + Op.RETURNDATACOPY(0, 0, Op.RETURNDATASIZE())
         + Op.SSTORE(3, Op.MLOAD(0))
@@ -95,7 +102,11 @@ def revert_contract_bytecode(
     selfdestruct_contract_address: Address,
 ) -> Bytecode:
     """Contract code that performs a call and then reverts."""
-    call_op = second_suicide(address=selfdestruct_contract_address, value=100)
+    call_op = (
+        second_suicide(address=selfdestruct_contract_address, value=100)
+        if second_suicide in [Op.CALL, Op.CALLCODE]
+        else second_suicide(address=selfdestruct_contract_address)
+    )
     return Op.MSTORE(0, Op.ADD(15, call_op)) + Op.REVERT(0, 32)
 
 

tests/constantinople/eip1014_create2/test_create_returndata.py

@@ -49,7 +49,6 @@ def test_create2_return_data(
     slot_begin_memory_after_create = 8
 
     # CREATE2 Initcode
-    create2_salt = 1
     return_data_in_create = 0xFFFAFB
     initcode = Op.MSTORE(0, return_data_in_create) + return_type_in_create(0, 32)
     call_return_data_value = 0x1122334455667788991011121314151617181920212223242526272829303132
@@ -79,7 +78,7 @@ def test_create2_return_data(
         + Op.SSTORE(slot_return_data_hash_before_create, Op.SHA3(0, call_return_size))
         #
         #
-        + create_type(offset=0x100, size=Op.CALLDATASIZE(), salt=create2_salt)
+        + create_type(offset=0x100, size=Op.CALLDATASIZE())
         + Op.SSTORE(slot_returndatasize_after_create, Op.RETURNDATASIZE())
         + Op.RETURNDATACOPY(0x300, 0, Op.RETURNDATASIZE())
         + Op.SSTORE(slot_returndatacopy_after_create, Op.MLOAD(0x300))

tests/frontier/create/test_create_one_byte.py

@@ -50,7 +50,7 @@ def test_create_one_byte(
     # make a subcontract that deploys code, because deploy 0xef eats ALL gas
     create_contract = pre.deploy_contract(
         code=Op.MSTORE(0, Op.CALLDATALOAD(0))
-        + Op.MSTORE(32, create_opcode(offset=32 - initcode_length, salt=0, size=initcode_length))
+        + Op.MSTORE(32, create_opcode(offset=32 - initcode_length, size=initcode_length))
         + Op.RETURN(32, 32)
     )
     code = pre.deploy_contract(

tests/frontier/scenarios/scenarios/call_combinations.py

@@ -93,13 +93,23 @@ def _generate_one_call_scenario(self, first_call: Opcode) -> Scenario:
 
         scenario_contract = pre.deploy_contract(
             code=Op.MSTORE(32, 1122334455)
-            + first_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=operation_contract,
-                args_offset=32,
-                args_size=40,
-                ret_size=32,
-                value=balance.first_call_value,
+            + (
+                first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_offset=32,
+                    args_size=40,
+                    ret_size=32,
+                    value=balance.first_call_value,
+                )
+                if first_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_offset=32,
+                    args_size=40,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.scenario_contract_balance,
@@ -230,23 +240,41 @@ def _compute_callvalue() -> int:
         )
         sub_contract = pre.deploy_contract(
             code=Op.MSTORE(32, 1122334455)
-            + second_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=operation_contract,
-                args_size=40,
-                args_offset=32,
-                ret_size=32,
-                value=second_call_value,
+            + (
+                second_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_size=40,
+                    args_offset=32,
+                    ret_size=32,
+                    value=second_call_value,
+                )
+                if second_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else second_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_size=40,
+                    args_offset=32,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.sub_contract_balance,
         )
         scenario_contract = pre.deploy_contract(
-            code=first_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=sub_contract,
-                ret_size=32,
-                value=balance.first_call_value,
+            code=(
+                first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=sub_contract,
+                    ret_size=32,
+                    value=balance.first_call_value,
+                )
+                if first_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=sub_contract,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.scenario_contract_balance,

tests/frontier/scenarios/scenarios/create_combinations.py

@@ -109,14 +109,25 @@ def _compute_selfbalance() -> int:
                 + Op.MSTORE(32, create(balance.create_value, 0, deploy_code_size, *salt))
                 + Op.MSTORE(0, 0)
                 + Op.MSTORE(64, 1122334455)
-                + call(
-                    gas=Op.SUB(Op.GAS, keep_gas),
-                    address=Op.MLOAD(32),
-                    args_offset=64,
-                    args_size=40,
-                    ret_offset=0,
-                    ret_size=32,
-                    value=balance.call_value,
+                + (
+                    call(
+                        gas=Op.SUB(Op.GAS, keep_gas),
+                        address=Op.MLOAD(32),
+                        args_offset=64,
+                        args_size=40,
+                        ret_offset=0,
+                        ret_size=32,
+                        value=balance.call_value,
+                    )
+                    if call not in [Op.DELEGATECALL, Op.STATICCALL]
+                    else call(
+                        gas=Op.SUB(Op.GAS, keep_gas),
+                        address=Op.MLOAD(32),
+                        args_offset=64,
+                        args_size=40,
+                        ret_offset=0,
+                        ret_size=32,
+                    )
                 )
                 + Op.RETURN(0, 32),
             )