chaintool/chainlib-eth
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: Add generic contract tx/call encoder cli tool

Browse Source
This commit is contained in:
Louis Holbrook 2021-10-18 10:18:20 +00:00
parent 2011ce8bb8
commit 94e016366b
13 changed files with 348 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
chainlib/eth/cli/__init__.py Normal file
View File

@ -0,0 +1,4 @@
from .arg import *
from .config import Config
from .rpc import Rpc
from .wallet import Wallet

9
chainlib/eth/cli/arg.py Normal file
View File

@ -0,0 +1,9 @@
# external imports
from chainlib.cli import (
ArgumentParser,
argflag_std_read,
argflag_std_write,
argflag_std_base,
reset as argflag_reset,
Flag,
)

33
chainlib/eth/cli/config.py Normal file
View File

@ -0,0 +1,33 @@
# standard imports
import os
# external imports
from chainlib.cli import Config as BaseConfig
script_dir = os.path.dirname(os.path.realpath(__file__))
data_dir = os.path.join(script_dir, '..')
class Config(BaseConfig):
"""Convenience constructor to set Ethereum defaults for the chainlib cli config object
"""
default_base_config_dir = os.path.join(data_dir, 'data', 'config')
default_fee_limit = 21000
@classmethod
def from_args(cls, args, arg_flags=0x0f, env=os.environ, extra_args={}, base_config_dir=None, default_config_dir=None, user_config_dir=None, default_fee_limit=None, logger=None, load_callback=None):
super(Config, cls).override_defaults(base_dir=cls.default_base_config_dir)
if default_fee_limit == None:
default_fee_limit = cls.default_fee_limit
config = BaseConfig.from_args(args, arg_flags=arg_flags, env=env, extra_args=extra_args, base_config_dir=base_config_dir, default_config_dir=default_config_dir, user_config_dir=user_config_dir, default_fee_limit=default_fee_limit, logger=logger, load_callback=load_callback)
if not config.get('RPC_DIALECT'):
config.add('default', 'RPC_DIALECT', exists_ok=True)
elif config.get('RPC_DIALECT') not in [
'openethereum',
'default',
]:
raise ValueError('unknown rpc dialect {}'.format(config.get('RPC_DIALECT')))
return config

89
chainlib/eth/cli/encode.py Normal file
View File

@ -0,0 +1,89 @@
# standard imports
import re
import logging
# external imports
from chainlib.eth.contract import (
ABIContractType,
ABIContractEncoder,
)
logg = logging.getLogger(__name__)
class CLIEncoder(ABIContractEncoder):
__re_uint = r'^([uU])[int]*([0-9]+)?$'
__re_bytes = r'^([bB])[ytes]*([0-9]+)?$'
__re_string = r'^([sS])[tring]*$'
__translations = [
'to_uint',
'to_bytes',
'to_string',
]
def __init__(self, signature=None):
super(CLIEncoder, self).__init__()
self.signature = signature
if signature != None:
self.method(signature)
def to_uint(self, typ):
s = None
a = None
m = re.match(self.__re_uint, typ)
if m == None:
return None
n = m.group(2)
if m.group(2) == None:
n = 256
s = 'UINT256'.format(m.group(2))
a = getattr(ABIContractType, s)
return (s, a)
def to_bytes(self, typ):
s = None
a = None
m = re.match(self.__re_bytes, typ)
if m == None:
return None
n = m.group(2)
if n == None:
n = 32
s = 'BYTES{}'.format(n)
a = getattr(ABIContractType, s)
return (s, a)
def to_string(self, typ):
m = re.match(self.__re_string, typ)
if m == None:
return None
s = 'STRING'
a = getattr(ABIContractType, s)
return (s, a)
def translate_type(self, typ):
r = None
for tr in self.__translations:
r = getattr(self, tr)(typ)
if r != None:
break
if r == None:
raise ValueError('no translation for type {}'.format(typ))
logg.debug('type {} translated to {}'.format(typ, r[0]))
return r[1]
def add_from(self, arg):
logg.debug('arg {}'.format(arg))
(typ, val) = arg.split(':', maxsplit=1)
real_typ = self.translate_type(typ)
if self.signature != None:
self.typ(real_typ)
fn = getattr(self, real_typ.value)
fn(val)

56
chainlib/eth/cli.py → chainlib/eth/cli/rpc.py
View File

@ -1,22 +1,8 @@
# standard imports
import os
import logging
# external imports
from chainlib.cli import (
ArgumentParser,
argflag_std_read,
argflag_std_write,
argflag_std_base,
Config as BaseConfig,
Wallet as BaseWallet,
Rpc as BaseRpc, Flag,
)
from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
from chainlib.cli import Rpc as BaseRpc
from chainlib.eth.connection import EthHTTPConnection
# local imports
from chainlib.eth.address import AddressChecksum
from chainlib.eth.connection import EthHTTPConnection
from chainlib.eth.gas import (
OverrideGasOracle,
RPCGasOracle,
@ -26,20 +12,6 @@ from chainlib.eth.nonce import (
RPCNonceOracle,
)
logg = logging.getLogger(__name__)
script_dir = os.path.dirname(os.path.realpath(__file__))
class Wallet(BaseWallet):
"""Convenience constructor to set Ethereum defaults for chainlib cli Wallet object
:param checksummer: Address checksummer object
:type checksummer: Implementation of chainlib.eth.address.AddressChecksum
"""
def __init__(self, checksummer=AddressChecksum):
super(Wallet, self).__init__(EIP155Signer, checksummer=checksummer)
class Rpc(BaseRpc):
"""Convenience constructor to set Ethereum defaults for chainlib cli Rpc object
@ -96,27 +68,3 @@ class Rpc(BaseRpc):
def get_gas_oracle(self):
return self.get_fee_oracle()
class Config(BaseConfig):
"""Convenience constructor to set Ethereum defaults for the chainlib cli config object
"""
default_base_config_dir = os.path.join(script_dir, 'data', 'config')
default_fee_limit = 21000
@classmethod
def from_args(cls, args, arg_flags=0x0f, env=os.environ, extra_args={}, base_config_dir=None, default_config_dir=None, user_config_dir=None, default_fee_limit=None, logger=None, load_callback=None):
super(Config, cls).override_defaults(base_dir=cls.default_base_config_dir)
if default_fee_limit == None:
default_fee_limit = cls.default_fee_limit
config = BaseConfig.from_args(args, arg_flags=arg_flags, env=env, extra_args=extra_args, base_config_dir=base_config_dir, default_config_dir=default_config_dir, user_config_dir=user_config_dir, default_fee_limit=default_fee_limit, logger=logger, load_callback=load_callback)
if not config.get('RPC_DIALECT'):
config.add('default', 'RPC_DIALECT', exists_ok=True)
elif config.get('RPC_DIALECT') not in [
'openethereum',
'default',
]:
raise ValueError('unknown rpc dialect {}'.format(config.get('RPC_DIALECT')))
return config

19
chainlib/eth/cli/wallet.py Normal file
View File

@ -0,0 +1,19 @@
# external imports
from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
from chainlib.cli import Wallet as BaseWallet
# local imports
from chainlib.eth.address import AddressChecksum
class Wallet(BaseWallet):
"""Convenience constructor to set Ethereum defaults for chainlib cli Wallet object
:param checksummer: Address checksummer object
:type checksummer: Implementation of chainlib.eth.address.AddressChecksum
"""
def __init__(self, checksummer=AddressChecksum):
super(Wallet, self).__init__(EIP155Signer, checksummer=checksummer)

152
chainlib/eth/runnable/encode.py Normal file
View File

@ -0,0 +1,152 @@
# SPDX-License-Identifier: GPL-3.0-or-later
# standard imports
import io
import sys
import os
import json
import argparse
import logging
import urllib
import sha3
# external imports
import chainlib.eth.cli
from chainlib.eth.cli.encode import CLIEncoder
from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
from crypto_dev_signer.keystore.dict import DictKeystore
from hexathon import (
add_0x,
strip_0x,
)
# local imports
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.eth.address import to_checksum
from chainlib.eth.connection import EthHTTPConnection
from chainlib.jsonrpc import (
JSONRPCRequest,
IntSequenceGenerator,
)
from chainlib.eth.nonce import (
RPCNonceOracle,
OverrideNonceOracle,
)
from chainlib.eth.gas import (
RPCGasOracle,
OverrideGasOracle,
)
from chainlib.eth.tx import (
TxFactory,
TxFormat,
raw,
)
from chainlib.error import SignerMissingException
from chainlib.chain import ChainSpec
from chainlib.eth.runnable.util import decode_for_puny_humans
from chainlib.eth.jsonrpc import to_blockheight_param
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
script_dir = os.path.dirname(os.path.realpath(__file__))
config_dir = os.path.join(script_dir, '..', 'data', 'config')
arg_flags = chainlib.eth.cli.argflag_std_write | chainlib.eth.cli.Flag.EXEC
argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
argparser.add_argument('--signature', type=str, help='Method signature to encode')
argparser.add_argument('contract_args', type=str, nargs='*', help='arguments to encode')
args = argparser.parse_args()
extra_args = {
'signature': None,
'contract_args': None,
}
config = chainlib.eth.cli.Config.from_args(args, arg_flags, extra_args=extra_args, default_config_dir=config_dir)
if not config.get('_EXEC_ADDRESS'):
argparser.error('exec address (-e) must be defined')
block_all = args.ww
block_last = args.w or block_all
wallet = chainlib.eth.cli.Wallet(EIP155Signer)
wallet.from_config(config)
rpc = chainlib.eth.cli.Rpc(wallet=wallet)
conn = rpc.connect_by_config(config)
send = config.true('_RPC_SEND')
chain_spec = None
try:
chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
except AttributeError:
pass
def main():
signer_address = ZERO_ADDRESS
signer = None
try:
signer = rpc.get_signer()
signer_address = rpc.get_signer_address()
except SignerMissingException:
pass
code = '0x'
cli_encoder = CLIEncoder(signature=config.get('_SIGNATURE'))
for arg in config.get('_CONTRACT_ARGS'):
cli_encoder.add_from(arg)
code += cli_encoder.get()
if not config.get('_SIGNATURE'):
print(strip_0x(code))
return
if signer == None:
c = TxFactory(chain_spec)
j = JSONRPCRequest(id_generator=rpc.id_generator)
o = j.template()
o['method'] = 'eth_call'
o['params'].append({
'to': exec_address,
'from': signer_address,
'value': '0x00',
'gas': add_0x(int.to_bytes(8000000, 8, byteorder='big').hex()), # TODO: better get of network gas limit
'gasPrice': '0x01',
'data': add_0x(code),
})
height = to_blockheight_param(config.get('_HEIGHT'))
o['params'].append(height)
o = j.finalize(o)
r = conn.do(r)
try:
print(strip_0x(r))
return
except ValueError:
sys.stderr.write('query returned an empty value ({})\n'.format(r))
sys.exit(1)
if chain_spec == None:
raise ValueError('chain spec must be specified')
c = TxFactory(chain_spec, signer=signer, gas_oracle=rpc.get_gas_oracle(), nonce_oracle=rpc.get_nonce_oracle())
tx = c.template(signer_address, config.get('_EXEC_ADDRESS'), use_nonce=True)
tx = c.set_code(tx, code)
tx_format = TxFormat.JSONRPC
if config.get('_RAW'):
tx_format = TxFormat.RLP_SIGNED
(tx_hash_hex, o) = c.finalize(tx, tx_format=tx_format)
if send:
r = conn.do(r)
print(r)
else:
if config.get('_RAW'):
o = strip_0x(o)
print(o)
if __name__ == '__main__':
main()

4
chainlib/eth/runnable/gas.py
View File

@ -80,7 +80,7 @@ def main():
if logg.isEnabledFor(logging.DEBUG):
try:
sender_balance = balance(signer_address, rpc.id_generator)
recipient_balance = balance(recipient, rpc.id_generator)
recipient_balance = balance(add_0x(recipient), rpc.id_generator)
logg.debug('sender {} balance before: {}'.format(signer_address, sender_balance))
logg.debug('recipient {} balance before: {}'.format(recipient, recipient_balance))
except urllib.error.URLError:
@ -94,7 +94,7 @@ def main():
r = conn.wait(tx_hash_hex)
if logg.isEnabledFor(logging.DEBUG):
sender_balance = balance(signer_address, rpc.id_generator)
recipient_balance = balance(recipient, rpc.id_generator)
recipient_balance = balance(add_0x(recipient), rpc.id_generator)
logg.debug('sender {} balance after: {}'.format(signer_address, sender_balance))
logg.debug('recipient {} balance after: {}'.format(recipient, recipient_balance))
if r['status'] == 0:

18
chainlib/eth/runnable/raw.py
View File

@ -50,6 +50,7 @@ config_dir = os.path.join(script_dir, '..', 'data', 'config')
arg_flags = chainlib.eth.cli.argflag_std_write | chainlib.eth.cli.Flag.EXEC
argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
argparser.add_argument('--deploy', action='store_true', help='Deploy data as contract')
argparser.add_positional('data', type=str, help='Transaction data')
args = argparser.parse_args()
config = chainlib.eth.cli.Config.from_args(args, arg_flags, default_config_dir=config_dir)
@ -65,9 +66,6 @@ conn = rpc.connect_by_config(config)
send = config.true('_RPC_SEND')
if config.get('_EXEC_ADDRESS') != None:
send = False
chain_spec = None
try:
chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
@ -83,11 +81,15 @@ def main():
except SignerMissingException:
pass
if config.get('_EXEC_ADDRESS') != None or args.deploy:
exec_address = None
if config.get('_EXEC_ADDRESS') != None:
exec_address = add_0x(to_checksum(config.get('_EXEC_ADDRESS')))
if not args.u and exec_address != add_0x(exec_address):
#if not args.u and exec_address != add_0x(exec_address):
if not args.u and exec_address != exec_address:
raise ValueError('invalid checksum address')
if signer_address == None:
j = JSONRPCRequest(id_generator=rpc.id_generator)
o = j.template()
o['method'] = 'eth_call'
@ -106,11 +108,10 @@ def main():
try:
print(strip_0x(r))
except ValueError:
sys.stderr.write('query returned an empty value\n')
sys.stderr.write('query returned an empty value ({})\n'.format(r))
sys.exit(1)
return
if signer_address != None:
else:
if chain_spec == None:
raise ValueError('chain spec must be specified')
g = TxFactory(chain_spec, signer=rpc.get_signer(), gas_oracle=rpc.get_gas_oracle(), nonce_oracle=rpc.get_nonce_oracle())
@ -124,8 +125,9 @@ def main():
r = conn.do(o)
print(r)
else:
if config.get('_RAW'):
o = strip_0x(o)
print(o)
print(tx_hash_hex)
else:
o = raw(args.data, id_generator=rpc.id_generator)

1
chainlib/eth/tx.py
View File

@ -167,7 +167,6 @@ def __unpack_raw(tx_raw_bytes, chain_id=1):
s[32-len(d[8]):] = d[8]
logg.debug('vb {}'.format(vb))
sig = b''.join([r, s, bytes([vb])])
#so = KeyAPI.Signature(signature_bytes=sig)
h = sha3.keccak_256()
h.update(rlp_encode(d))

5
chainlib/eth/unittest/ethtester.py
View File

@ -26,7 +26,8 @@ from chainlib.connection import (
from chainlib.eth.address import to_checksum_address
from chainlib.chain import ChainSpec
logg = logging.getLogger(__name__)
#logg = logging.getLogger(__name__)
logg = logging.getLogger()
test_address = bytes.fromhex('Eb3907eCad74a0013c259D5874AE7f22DcBcC95C')
@ -63,6 +64,7 @@ class EthTesterCase(unittest.TestCase):
self.helper = eth_tester_instance
self.backend = self.helper.backend
self.rpc = TestRPCConnection(None, eth_tester_instance, self.signer)
for a in self.keystore.list():
self.accounts.append(add_0x(to_checksum_address(a)))
@ -75,6 +77,5 @@ class EthTesterCase(unittest.TestCase):
RPCConnection.register_location('custom', self.chain_spec, tag='signer', exist_ok=True)
def tearDown(self):
pass

2
requirements.txt
View File

@ -1,4 +1,4 @@
crypto-dev-signer>=0.4.15a4,<=0.4.15
crypto-dev-signer>=0.4.15rc2,<=0.4.15
pysha3==1.0.2
hexathon~=0.0.1a8
websocket-client==0.57.0

2
setup.cfg
View File

@ -32,6 +32,7 @@ packages =
chainlib.eth.runnable
chainlib.eth.pytest
chainlib.eth.unittest
chainlib.eth.cli
[options.entry_points]
console_scripts =
@ -41,6 +42,7 @@ console_scripts =
eth-raw = chainlib.eth.runnable.raw:main
eth-get = chainlib.eth.runnable.get:main
eth-decode = chainlib.eth.runnable.decode:main
eth-encode = chainlib.eth.runnable.encode:main
eth-info = chainlib.eth.runnable.info:main
eth-nonce = chainlib.eth.runnable.count:main
eth = chainlib.eth.runnable.info:main