Skip to content

test_worst_storage_access_cold()

Documentation for tests/zkevm/test_worst_stateful_opcodes.py::test_worst_storage_access_cold@64f949d0.

Generate fixtures for these test cases for Prague with:

fill -v tests/zkevm/test_worst_stateful_opcodes.py::test_worst_storage_access_cold --fork Prague

Test running a block with as many cold storage slot accesses as possible.

Source code in tests/zkevm/test_worst_stateful_opcodes.py
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
@pytest.mark.valid_from("Cancun")
@pytest.mark.parametrize(
    "storage_action",
    [
        pytest.param(StorageAction.READ, id="SSLOAD"),
        pytest.param(StorageAction.WRITE_SAME_VALUE, id="SSTORE same value"),
        pytest.param(StorageAction.WRITE_NEW_VALUE, id="SSTORE new value"),
    ],
)
@pytest.mark.parametrize(
    "absent_slots",
    [
        True,
        False,
    ],
)
@pytest.mark.slow()
def test_worst_storage_access_cold(
    blockchain_test: BlockchainTestFiller,
    pre: Alloc,
    fork: Fork,
    storage_action: StorageAction,
    absent_slots: bool,
):
    """Test running a block with as many cold storage slot accesses as possible."""
    env = Environment(gas_limit=100_000_000_000)
    gas_costs = fork.gas_costs()
    attack_gas_limit = Environment().gas_limit

    cost = gas_costs.G_COLD_SLOAD  # All accesses are always cold
    if storage_action == StorageAction.WRITE_NEW_VALUE:
        if not absent_slots:
            cost += gas_costs.G_STORAGE_RESET
        else:
            cost += gas_costs.G_STORAGE_SET
    elif storage_action == StorageAction.WRITE_SAME_VALUE:
        if absent_slots:
            cost += gas_costs.G_STORAGE_SET
        else:
            cost += gas_costs.G_WARM_SLOAD
    elif storage_action == StorageAction.READ:
        cost += gas_costs.G_WARM_SLOAD

    intrinsic_gas_cost_calc = fork.transaction_intrinsic_cost_calculator()
    num_target_slots = (attack_gas_limit - intrinsic_gas_cost_calc()) // cost

    blocks = []

    # Contract code
    execution_code_body = Bytecode()
    if storage_action == StorageAction.WRITE_SAME_VALUE:
        # All the storage slots in the contract are initialized to their index.
        # That is, storage slot `i` is initialized to `i`.
        execution_code_body = Op.SSTORE(Op.DUP1, Op.DUP1)
    elif storage_action == StorageAction.WRITE_NEW_VALUE:
        # The new value 2^256-1 is guaranteed to be different from the initial value.
        execution_code_body = Op.SSTORE(Op.DUP2, Op.NOT(0))
    elif storage_action == StorageAction.READ:
        execution_code_body = Op.POP(Op.SLOAD(Op.DUP1))

    execution_code = Op.PUSH4(num_target_slots) + While(
        body=execution_code_body,
        condition=Op.PUSH1(1) + Op.SWAP1 + Op.SUB + Op.DUP1 + Op.ISZERO + Op.ISZERO,
    )
    execution_code_address = pre.deploy_contract(code=execution_code)

    # Contract creation
    slots_init = Bytecode()
    if not absent_slots:
        slots_init = Op.PUSH4(num_target_slots) + While(
            body=Op.SSTORE(Op.DUP1, Op.DUP1),
            condition=Op.PUSH1(1) + Op.SWAP1 + Op.SUB + Op.DUP1 + Op.ISZERO + Op.ISZERO,
        )

    # To create the contract, we apply the slots_init code to initialize the storage slots
    # (int the case of absent_slots=False) and then copy the execution code to the contract.
    creation_code = (
        slots_init
        + Op.EXTCODECOPY(
            address=execution_code_address,
            dest_offset=0,
            offset=0,
            size=Op.EXTCODESIZE(execution_code_address),
        )
        + Op.RETURN(0, Op.MSIZE)
    )
    sender_addr = pre.fund_eoa()
    setup_tx = Transaction(
        to=None,
        gas_limit=env.gas_limit,
        data=creation_code,
        sender=sender_addr,
    )
    blocks.append(Block(txs=[setup_tx]))

    contract_address = compute_create_address(address=sender_addr, nonce=0)

    op_tx = Transaction(
        to=contract_address,
        gas_limit=attack_gas_limit,
        sender=pre.fund_eoa(),
    )
    blocks.append(Block(txs=[op_tx]))

    blockchain_test(
        genesis_environment=env,
        pre=pre,
        post={},
        blocks=blocks,
        exclude_full_post_state_in_output=True,
    )

Parametrized Test Cases

The interactive table below is also available as a standalone page.

Test ID (Abbreviated) absent_slots storage_action
...fork_Cancun-blockchain_test-absent_slots_True-SSLOAD True 1
...fork_Cancun-blockchain_test-absent_slots_True-SSTORE same value True 2
...fork_Cancun-blockchain_test-absent_slots_True-SSTORE new value True 3
...fork_Cancun-blockchain_test-absent_slots_False-SSLOAD False 1
...fork_Cancun-blockchain_test-absent_slots_False-SSTORE same value False 2
...fork_Cancun-blockchain_test-absent_slots_False-SSTORE new value False 3
...fork_Prague-blockchain_test-absent_slots_True-SSLOAD True 1
...fork_Prague-blockchain_test-absent_slots_True-SSTORE same value True 2
...fork_Prague-blockchain_test-absent_slots_True-SSTORE new value True 3
...fork_Prague-blockchain_test-absent_slots_False-SSLOAD False 1
...fork_Prague-blockchain_test-absent_slots_False-SSTORE same value False 2
...fork_Prague-blockchain_test-absent_slots_False-SSTORE new value False 3
...fork_Osaka-blockchain_test-absent_slots_True-SSLOAD True 1
...fork_Osaka-blockchain_test-absent_slots_True-SSTORE same value True 2
...fork_Osaka-blockchain_test-absent_slots_True-SSTORE new value True 3
...fork_Osaka-blockchain_test-absent_slots_False-SSLOAD False 1
...fork_Osaka-blockchain_test-absent_slots_False-SSTORE same value False 2
...fork_Osaka-blockchain_test-absent_slots_False-SSTORE new value False 3