Replaying an upstream transaction using Ganache fork mode

replay.py

#!/usr/bin/env python3
'''
To run this, install ganache-cli and web3.py:

    $ npm install -g ganache-cli
    $ pip install --user web3

Usage:

    $ ./replay.py <RPC URL> <txhash>

Example:

    $ ./replay.py https://rinkeby.infura.io/v3/XXX 0xad5f8957fd0615c6ee3c08283080046050a739c313ffe069cad476e7452e3389
'''

import json, random, socket, sys, time
from subprocess import Popen, PIPE, STDOUT
from web3 import Web3, HTTPProvider
from web3.middleware import geth_poa_middleware

upstream = sys.argv[1]
txhash = sys.argv[2]

w3 = Web3(HTTPProvider(upstream))

# only necessary for POA networks
w3.middleware_onion.inject(geth_poa_middleware, layer=0)

tx = w3.eth.getTransaction(txhash)

port = random.randint(20000, 30000)
ganache = Popen([
    'ganache-cli',
    '-f', f'{upstream}@{tx.blockNumber - 1}',
    '-u', tx['from'],
    '-p', str(port),
    # '--debug',   # enable for (not very detailed) EVM trace
], stdout=PIPE, stderr=STDOUT)

while True:
    try:
        s = socket.create_connection(('127.0.0.1', port))
        s.close()
        break
    except ConnectionRefusedError:
        time.sleep(0.1)

w3 = Web3(HTTPProvider(f'http://localhost:{port}'))

try:
    txhash = w3.eth.sendTransaction({
        'to': tx.to,
        'from': tx['from'],
        'nonce': tx.nonce,
        'gas': tx.gas,
        'gasPrice': tx.gasPrice,
        'value': tx.value,
        'data': tx.input
    })
except Exception as exc:
    data = exc.args[0]['data']
    print('Error occured:')
    print(json.dumps(data, indent=4))
    print()
    txhash = next(iter(data.keys()))

print('Replayed transaction hash: ', txhash)

# Let's ensure that we can indeed use the debug API, hence we could also
# use `truffle debug <txhash>` on the forked node for source-level tracing.
w3.manager.request_blocking('debug_traceTransaction', [txhash, {}])

time.sleep(1)
ganache.terminate()

out, err = ganache.communicate()
print()
print('Ganache log:')
print()
print(out.decode())