@pytest.mark.repricing
@pytest.mark.parametrize("cache_strategy", [CacheStrategy.NO_CACHE])
@pytest.mark.parametrize(
"opcode,value_sent,account_mode,overhead_baseline", account_access_params()
)
def test_account_access(
benchmark_test: BenchmarkTestFiller,
pre: Alloc,
fork: Fork,
opcode: Op,
value_sent: int,
gas_benchmark_value: int,
fixed_opcode_count: int | None,
account_mode: AccountMode,
overhead_baseline: bool,
cache_strategy: CacheStrategy,
verified_accounts: dict,
) -> None:
"""Benchmark account access with caching strategies."""
account_creator = AccountCreator(account_mode)
address_source = account_creator.address_source(Op.CALLDATALOAD(0))
increment_op = address_source.next_op()
cache_op = (
Op.POP(
Op.BALANCE(
address=address_source.address_op(),
# Gas accounting
address_warm=False,
)
)
if cache_strategy == CacheStrategy.CACHE_TX
else Bytecode()
)
access_warm = cache_strategy == CacheStrategy.CACHE_TX
setup_code = address_source.setup
if opcode == Op.EXTCODECOPY:
copy_size = 1024
copy_dest = address_source.memory_size
attack_call = opcode(
address=address_source.address_op(),
dest_offset=copy_dest,
size=copy_size,
# Gas accounting
data_size=copy_size,
address_warm=access_warm,
)
# Expand memory during setup so the loop cost is constant.
setup_code += Op.MSTORE8(
copy_dest + copy_size - 1,
0,
# Gas accounting
old_memory_size=address_source.memory_size,
new_memory_size=copy_dest + copy_size,
)
elif opcode in (Op.CALL, Op.CALLCODE):
attack_call = Op.POP(
opcode(
address=address_source.address_op(),
value=value_sent,
# Gas accounting
address_warm=access_warm,
value_transfer=value_sent > 0,
account_new=(
opcode == Op.CALL
and value_sent > 0
and account_mode == AccountMode.NON_EXISTING_ACCOUNT
),
)
)
else:
# BALANCE, STATICCALL, DELEGATECALL, EXTCODESIZE, EXTCODEHASH
attack_call = Op.POP(
opcode(
address=address_source.address_op(),
# Gas accounting
address_warm=access_warm,
)
)
setup_code += Op.ADD(1, Op.CALLDATALOAD(32)) + Op.CALLDATALOAD(0)
loop_code = While(
body=cache_op + attack_call + increment_op,
condition=Op.PUSH1(1) + Op.ADD + Op.DUP1 + Op.DUP3 + Op.GT,
)
call_operations = (Op.CALL, Op.CALLCODE, Op.DELEGATECALL, Op.STATICCALL)
executes_contract_code = (
opcode in call_operations and account_creator.has_execution_code
)
iterating_subcall: Bytecode = (
account_creator.execution_code if executes_contract_code else Op.STOP
)
attack_code = IteratingBytecode(
setup=setup_code,
iterating=loop_code,
iterating_subcall=iterating_subcall,
)
# Calldata generator for each transaction of the iterating bytecode.
def calldata(iteration_count: int, start_iteration: int) -> bytes:
index_end = start_iteration + iteration_count - 1
return Hash(start_iteration) + Hash(index_end)
run_code = attack_code
target_opcode = opcode
if overhead_baseline:
keccak_op = Op.POP(address_source.address_op())
if cache_strategy == CacheStrategy.CACHE_TX:
keccak_op = keccak_op * 2
run_code = IteratingBytecode(
setup=address_source.setup
+ Op.ADD(1, Op.CALLDATALOAD(32))
+ Op.CALLDATALOAD(0),
iterating=While(
body=keccak_op + increment_op,
condition=Op.PUSH1(1) + Op.ADD + Op.DUP1 + Op.DUP3 + Op.GT,
),
)
target_opcode = Op.SHA3
total_iterations = None
if fixed_opcode_count is not None:
total_iterations = int(fixed_opcode_count * 1000)
elif overhead_baseline:
total_iterations = sum(
attack_code.tx_iterations_by_gas_limit(
fork=fork,
gas_limit=gas_benchmark_value,
calldata=calldata,
)
)
attack_address = pre.deploy_contract(code=run_code, balance=10**21)
post: dict = {}
cache_txs = []
with TestPhaseManager.execution():
attack_sender = pre.fund_eoa()
if total_iterations is not None:
attack_txs = list(
run_code.transactions_by_total_iteration_count(
fork=fork,
total_iterations=total_iterations,
sender=attack_sender,
to=attack_address,
calldata=calldata,
)
)
else:
attack_txs = list(
run_code.transactions_by_gas_limit(
fork=fork,
gas_limit=gas_benchmark_value,
sender=attack_sender,
to=attack_address,
calldata=calldata,
)
)
if not overhead_baseline and attack_txs:
count = 1 + max(
int.from_bytes(bytes(tx.data)[32:64], "big") for tx in attack_txs
)
account_creator.register_targets(
pre,
count,
verified_accounts=verified_accounts,
label=account_mode.name,
)
if cache_strategy == CacheStrategy.CACHE_PREVIOUS_BLOCK:
with TestPhaseManager.setup():
cache_sender = pre.fund_eoa()
for tx in attack_txs:
cache_txs.append(
Transaction(
gas_limit=tx.gas_limit,
data=tx.data,
to=attack_address,
sender=cache_sender,
)
)
blocks = build_cache_strategy_blocks(cache_strategy, attack_txs, cache_txs)
benchmark_test(
pre=pre,
post=post,
blocks=blocks,
target_opcode=target_opcode,
skip_gas_used_validation=True,
verifications=[ReceiptStatusExpected(), NoTraceErrors()],
)