Skip to content

Instantly share code, notes, and snippets.

@alpinevm
Created October 13, 2023 01:58
Show Gist options
  • Select an option

  • Save alpinevm/72b51101d9863d7a9edb81e93cc072b7 to your computer and use it in GitHub Desktop.

Select an option

Save alpinevm/72b51101d9863d7a9edb81e93cc072b7 to your computer and use it in GitHub Desktop.
Invoke and use Foundry Anvil from a python class
"""
python3 -m pip install web3 requests
"""
import subprocess
import socket
from web3 import Web3, HTTPProvider
from web3.types import Wei
import time
import requests
import traceback
class AnvilInstance:
def __init__(self, blockNumber: int, forkUrl: str, hideAnvilOutput: bool = True, autoMineOnTxn: bool = True):
self.blockNumber = blockNumber
self.forkUrl = forkUrl
self.defaultHost = "127.0.0.1"
self.defaultPort = str(self._find_free_port())
self.anvilInstance = subprocess.Popen(
["anvil",
"--no-mining",
"--fork-url", self.forkUrl,
"--fork-block-number", str(self.blockNumber),
"--steps-tracing",
"--host", self.defaultHost,
"--port", self.defaultPort
],
stdout=subprocess.DEVNULL if hideAnvilOutput else None,
stderr=subprocess.DEVNULL if hideAnvilOutput else None
)
self.wait_till_live()
self.evm_setAutomine(autoMineOnTxn)
def wait_till_live(self):
while True:
try:
Web3(HTTPProvider(self.getUrl())).eth.block_number
break
except:
pass
time.sleep(.01)
def _find_free_port(self):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('', 0)) # Bind to a port assigned by the kernel
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1] # Return the port number assigned
def getUrl(self):
return f"http://{self.defaultHost}:{self.defaultPort}"
def kill(self):
self.anvilInstance.kill()
# Set the balance of the account (spoof)
def anvil_setBalance(self, account: str, balance: Wei):
return requests.post(self.getUrl(), json={
"jsonrpc": "2.0",
"method": "anvil_setBalance",
"params": [account, balance.to_bytes(32, 'big').hex()],
"id": 1
}).json()
def evm_setAutomine(self, autoMine: bool):
return requests.post(self.getUrl(), json={
"jsonrpc": "2.0",
"method": "evm_setAutomine",
"params": [autoMine],
"id": 1
}).json()
def anvil_mine(self):
return requests.post(self.getUrl(), json={
"jsonrpc": "2.0",
"method": "anvil_mine",
"params": [],
"id": 1
}).json()
def debug_traceTransaction(self, txn_hash: str):
return requests.post(self.getUrl(), json={
"method": "debug_traceTransaction",
"params": [txn_hash, {}],
"jsonrpc": "2.0",
"id": 1,
}).json()
def setBalanceCheck():
instance = None
try:
instance = AnvilInstance(
blockNumber=17687153,
forkUrl="https://mainnet.gateway.tenderly.co"
)
simProvider = Web3(HTTPProvider(instance.getUrl()))
wallet = simProvider.to_checksum_address("0x3c1e3d527103694885f5DeFE21bf8f871CBC8133")
print("before", simProvider.eth.get_balance(wallet))
instance.anvil_setBalance(
wallet,
Wei(10**18)
)
print("after", simProvider.eth.get_balance(wallet))
except:
print(traceback.format_exc())
pass
if (instance is not None):
instance.kill()
if __name__ == "__main__":
setBalanceCheck()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment