219 lines
6.0 KiB
Python
219 lines
6.0 KiB
Python
# standard imports
|
|
import os
|
|
import logging
|
|
|
|
# external imports
|
|
import eth_tester
|
|
import coincurve
|
|
from chainlib.connection import (
|
|
RPCConnection,
|
|
error_parser,
|
|
)
|
|
from chainlib.eth.address import (
|
|
to_checksum_address,
|
|
)
|
|
from chainlib.jsonrpc import (
|
|
jsonrpc_response,
|
|
jsonrpc_error,
|
|
jsonrpc_result,
|
|
)
|
|
from hexathon import (
|
|
unpad,
|
|
add_0x,
|
|
strip_0x,
|
|
)
|
|
|
|
from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
|
|
from crypto_dev_signer.encoding import private_key_to_address
|
|
|
|
|
|
logg = logging.getLogger().getChild(__name__)
|
|
|
|
test_pk = bytes.fromhex('5087503f0a9cc35b38665955eb830c63f778453dd11b8fa5bd04bc41fd2cc6d6')
|
|
|
|
|
|
class EthTesterSigner(eth_tester.EthereumTester):
|
|
|
|
def __init__(self, backend, keystore):
|
|
super(EthTesterSigner, self).__init__(backend)
|
|
logg.debug('accounts {}'.format(self.get_accounts()))
|
|
|
|
self.keystore = keystore
|
|
self.backend = backend
|
|
self.backend.add_account(test_pk)
|
|
for pk in self.backend.account_keys:
|
|
pubk = pk.public_key
|
|
address = pubk.to_checksum_address()
|
|
logg.debug('test keystore have pk {} pubk {} addr {}'.format(pk, pk.public_key, address))
|
|
self.keystore.import_raw_key(pk._raw_key)
|
|
|
|
|
|
def new_account(self):
|
|
pk = os.urandom(32)
|
|
address = self.keystore.import_raw_key(pk)
|
|
checksum_address = add_0x(to_checksum_address(address))
|
|
self.backend.add_account(pk)
|
|
return checksum_address
|
|
|
|
|
|
class TestRPCConnection(RPCConnection):
|
|
|
|
def __init__(self, location, backend, signer):
|
|
super(TestRPCConnection, self).__init__(location)
|
|
self.backend = backend
|
|
self.signer = signer
|
|
|
|
|
|
def do(self, o, error_parser=error_parser):
|
|
logg.debug('testrpc do {}'.format(o))
|
|
m = getattr(self, o['method'])
|
|
if m == None:
|
|
raise ValueError('unhandled method {}'.format(o['method']))
|
|
r = None
|
|
try:
|
|
result = m(o['params'])
|
|
logg.debug('result {}'.format(result))
|
|
r = jsonrpc_response(o['id'], result)
|
|
except Exception as e:
|
|
logg.exception(e)
|
|
r = jsonrpc_error(o['id'], message=str(e))
|
|
return jsonrpc_result(r, error_parser)
|
|
|
|
|
|
def eth_blockNumber(self, p):
|
|
block = self.backend.get_block_by_number('latest')
|
|
return block['number']
|
|
|
|
|
|
def eth_getBlockByNumber(self, p):
|
|
b = bytes.fromhex(strip_0x(p[0]))
|
|
n = int.from_bytes(b, 'big')
|
|
block = self.backend.get_block_by_number(n)
|
|
return block
|
|
|
|
|
|
def eth_getBlockByHash(self, p):
|
|
block = self.backend.get_block_by_hash(p[0])
|
|
return block
|
|
|
|
|
|
def eth_getTransactionByBlock(self, p):
|
|
block = self.eth_getBlockByHash(p)
|
|
try:
|
|
tx_index = int(p[1], 16)
|
|
except TypeError:
|
|
tx_index = int(p[1])
|
|
tx_hash = block['transactions'][tx_index]
|
|
tx = self.eth_getTransactionByHash([tx_hash])
|
|
return tx
|
|
|
|
def eth_getBalance(self, p):
|
|
balance = self.backend.get_balance(p[0])
|
|
hx = balance.to_bytes(32, 'big').hex()
|
|
return add_0x(unpad(hx))
|
|
|
|
|
|
def eth_getTransactionCount(self, p):
|
|
nonce = self.backend.get_nonce(p[0])
|
|
hx = nonce.to_bytes(4, 'big').hex()
|
|
return add_0x(unpad(hx))
|
|
|
|
|
|
def eth_getTransactionByHash(self, p):
|
|
tx = self.backend.get_transaction_by_hash(p[0])
|
|
return tx
|
|
|
|
|
|
def eth_getTransactionByBlockHashAndIndex(self, p):
|
|
#logg.debug('p {}'.format(p))
|
|
#block = self.eth_getBlockByHash(p[0])
|
|
#tx = block.transactions[p[1]]
|
|
#return eth_getTransactionByHash(tx[0])
|
|
return self.eth_getTransactionByBlock(p)
|
|
|
|
|
|
def eth_getTransactionReceipt(self, p):
|
|
rcpt = self.backend.get_transaction_receipt(p[0])
|
|
if rcpt.get('block_number') == None:
|
|
rcpt['block_number'] = rcpt['blockNumber']
|
|
else:
|
|
rcpt['blockNumber'] = rcpt['block_number']
|
|
return rcpt
|
|
|
|
|
|
def eth_getCode(self, p):
|
|
r = self.backend.get_code(p[0])
|
|
return r
|
|
|
|
|
|
def eth_call(self, p):
|
|
tx_ethtester = to_ethtester_call(p[0])
|
|
r = self.backend.call(tx_ethtester)
|
|
return r
|
|
|
|
|
|
def eth_gasPrice(self, p):
|
|
return hex(1000000000)
|
|
|
|
|
|
def personal_newAccount(self, passphrase):
|
|
a = self.backend.new_account()
|
|
return a
|
|
|
|
|
|
def eth_sign(self, p):
|
|
r = self.signer.sign_ethereum_message(strip_0x(p[0]), strip_0x(p[1]))
|
|
return r
|
|
|
|
|
|
def eth_sendRawTransaction(self, p):
|
|
r = self.backend.send_raw_transaction(p[0])
|
|
return r
|
|
|
|
|
|
def eth_signTransaction(self, p):
|
|
raise NotImplementedError('needs transaction deserializer for EIP155Transaction')
|
|
tx_dict = p[0]
|
|
tx = EIP155Transaction(tx_dict, tx_dict['nonce'], tx_dict['chainId'])
|
|
passphrase = p[1]
|
|
r = self.signer.sign_transaction_to_rlp(tx, passphrase)
|
|
return r
|
|
|
|
|
|
def __verify_signer(self, tx, passphrase=''):
|
|
pk_bytes = self.backend.keystore.get(tx.sender)
|
|
pk = coincurve.PrivateKey(secret=pk_bytes)
|
|
result_address = private_key_to_address(pk)
|
|
assert strip_0x(result_address) == strip_0x(tx.sender)
|
|
|
|
|
|
def sign_transaction(self, tx, passphrase=''):
|
|
self.__verify_signer(tx, passphrase)
|
|
return self.signer.sign_transaction(tx, passphrase)
|
|
|
|
|
|
def sign_transaction_to_rlp(self, tx, passphrase=''):
|
|
self.__verify_signer(tx, passphrase)
|
|
return self.signer.sign_transaction_to_rlp(tx, passphrase)
|
|
|
|
|
|
def disconnect(self):
|
|
pass
|
|
|
|
|
|
def to_ethtester_call(tx):
|
|
if tx['gas'] == '':
|
|
tx['gas'] = '0x00'
|
|
|
|
if tx['gasPrice'] == '':
|
|
tx['gasPrice'] = '0x00'
|
|
|
|
tx = {
|
|
'to': tx['to'],
|
|
'from': tx['from'],
|
|
'gas': int(tx['gas'], 16),
|
|
'gas_price': int(tx['gasPrice'], 16),
|
|
'data': tx['data'],
|
|
}
|
|
return tx
|