cic-cli/cic/ext/eth/__init__.py

251 lines
8.2 KiB
Python
Raw Normal View History

# standard imports
import logging
2021-10-10 13:48:32 +02:00
import copy
# external imports
2021-10-10 12:50:18 +02:00
from chainlib.chain import ChainSpec
2021-10-10 15:37:26 +02:00
from chainlib.eth.tx import (
TxFormat,
TxFactory,
2021-10-11 08:57:03 +02:00
Tx,
receipt,
2021-10-10 15:37:26 +02:00
)
from chainlib.eth.connection import RPCConnection
from chainlib.eth.contract import (
ABIContractEncoder,
ABIContractType
)
2021-10-11 08:57:03 +02:00
from chainlib.eth.gas import OverrideGasOracle
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.address import (
is_address,
to_checksum_address,
)
from hexathon import add_0x
2021-10-10 12:50:18 +02:00
from eth_token_index import TokenUniqueSymbolIndex
2021-10-10 14:06:40 +02:00
from eth_address_declarator import Declarator
2021-10-11 08:57:03 +02:00
from eth_address_declarator.declarator import AddressDeclarator
2021-10-12 19:39:18 +02:00
from giftable_erc20_token import GiftableToken
# local imports
from cic.ext.eth.rpc import parse_adapter
2021-10-12 08:39:20 +02:00
from cic.extension import Extension
logg = logging.getLogger(__name__)
2021-10-12 08:39:20 +02:00
class CICEth(Extension):
2021-10-12 08:39:20 +02:00
def __init__(self, chain_spec, resources, proof, signer=None, rpc=None, outputs_writer=None, fee_oracle=None):
2021-10-10 13:48:32 +02:00
"""resources will be modified
"""
2021-10-12 08:39:20 +02:00
super(CICEth, self).__init__(chain_spec, resources, proof, signer=signer, rpc=rpc, outputs_writer=outputs_writer)
2021-10-10 12:50:18 +02:00
self.fee_oracle = fee_oracle
2021-10-10 13:48:32 +02:00
self.tx_format = TxFormat.RAW_ARGS
self.outputs_writer = outputs_writer
2021-10-10 13:48:32 +02:00
if self.rpc != None:
self.tx_format = TxFormat.JSONRPC
elif self.signer != None:
self.tx_format = TxFormat.RLP_SIGNED
def __detect_arg_type(self, v):
typ = None
try:
int(v, 10)
typ = ABIContractType.UINT256
except TypeError:
pass
if typ == None:
try:
vv = strip_0x(v)
if is_address(vv):
typ = ABIContractType.ADDRESS
else:
typ = ABIContractType.BYTES32
except ValueError:
pass
if typ == None:
try:
v.encode('utf-8')
typ = ABIContractType.STRING
except ValueError:
pass
if typ == None:
raise ValueError('cannot automatically determine type for value {}'.format(v))
logg.info('argument {} parsed as abi contract type {}'.format(typ.value))
return typ
def __order_args(self):
args = [
self.token_details['name'],
self.token_details['symbol'],
self.token_details['precision'],
]
args_types = [
ABIContractType.STRING.value,
ABIContractType.STRING.value,
ABIContractType.UINT256.value,
]
for i, x in enumerate(self.token_details['extra']):
args.append(x)
typ = None
if self.token_details['extra_types'] != None:
typ = self.token_details['extra_types'][i]
else:
typ = self.__detect_arg_type(x)
2021-10-10 19:59:52 +02:00
args_types.append(typ)
positions = self.token_details['positions']
if positions == None:
positions = list(range(len(args)))
return (args, args_types, positions)
2021-10-10 15:37:26 +02:00
def add_outputs(self, k, v):
logg.debug('adding outputs {} {}'.format(k, v))
self.outputs.append((k, v))
def get_outputs(self):
return self.outputs
def process_token(self, writer=None):
if writer == None:
writer = self.outputs_writer
(args, args_types, positions) = self.__order_args()
2021-10-10 15:37:26 +02:00
enc = ABIContractEncoder()
for i in positions:
getattr(enc, args_types[i])(args[i])
2021-10-12 08:39:20 +02:00
code = enc.get()
if self.token_code != None:
code = self.token_code + code
2021-10-10 15:37:26 +02:00
logg.debug('resource {}'.format(self.resources))
signer_address = add_0x(to_checksum_address(self.resources['token']['key_account']))
nonce_oracle = None
if self.rpc != None:
nonce_oracle = RPCNonceOracle(signer_address, conn=self.rpc)
2021-10-11 08:57:03 +02:00
c = TxFactory(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=self.fee_oracle)
2021-10-10 15:37:26 +02:00
tx = c.template(signer_address, None, use_nonce=True)
tx = c.set_code(tx, code)
o = c.finalize(tx, self.tx_format)
r = None
if self.rpc != None:
r = self.rpc.do(o[1])
o = self.rpc.wait(r)
o = Tx.src_normalize(o)
self.token_address = o['contract_address']
2021-10-10 15:37:26 +02:00
elif self.signer != None:
r = o[1]
2021-10-10 12:50:18 +02:00
2021-10-10 15:37:26 +02:00
if r == None:
r = code
writer.write('token', r.encode('utf-8'))
2021-10-12 19:39:18 +02:00
writer.write('token_address', self.token_address.encode('utf-8'))
self.add_outputs('token', r.encode('utf-8'))
2021-10-12 19:39:18 +02:00
if self.token_details['supply'] > 0:
c = GiftableToken(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=self.fee_oracle)
o = c.mint_to(self.token_address, self.resources['token']['key_account'], self.resources['token']['key_account'], self.token_details['supply'])
r = None
if self.rpc != None:
r = self.rpc.do(o[1])
self.rpc.wait(r)
elif self.signer != None:
r = o[1]
else:
r = o
writer.write('token_supply', r.encode('utf-8'))
2021-10-10 15:37:26 +02:00
return r
2021-10-10 12:50:18 +02:00
def process_token_index(self, writer=None):
if writer == None:
writer = self.outputs_writer
2021-10-12 08:39:20 +02:00
signer_address = add_0x(to_checksum_address(self.resources['token_index']['key_account']))
contract_address = add_0x(to_checksum_address(self.resources['token_index']['reference']))
2021-10-11 08:57:03 +02:00
gas_oracle = OverrideGasOracle(limit=TokenUniqueSymbolIndex.gas(), conn=self.rpc)
nonce_oracle = None
if self.rpc != None:
2021-10-12 08:39:20 +02:00
nonce_oracle = RPCNonceOracle(add_0x(signer_address), conn=self.rpc)
2021-10-11 08:57:03 +02:00
c = TokenUniqueSymbolIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
2021-10-10 21:48:16 +02:00
2021-10-10 13:20:00 +02:00
o = c.register(contract_address, signer_address, self.token_address, tx_format=self.tx_format)
2021-10-10 14:00:47 +02:00
r = None
2021-10-10 13:20:00 +02:00
if self.rpc != None:
2021-10-10 13:48:32 +02:00
r = self.rpc.do(o[1])
2021-10-12 08:39:20 +02:00
self.rpc.wait(r)
2021-10-10 13:20:00 +02:00
elif self.signer != None:
2021-10-10 14:00:47 +02:00
r = o[1]
2021-10-10 13:20:00 +02:00
else:
2021-10-10 14:00:47 +02:00
r = o
2021-10-10 13:48:32 +02:00
writer.write('token_index', r.encode('utf-8'))
self.add_outputs('token_index', r)
2021-10-10 14:00:47 +02:00
return r
2021-10-10 13:48:32 +02:00
2021-10-10 12:50:18 +02:00
def process_address_declarator(self, writer=None):
if writer == None:
writer = self.outputs_writer
2021-10-12 08:39:20 +02:00
signer_address = add_0x(to_checksum_address(self.resources['address_declarator']['key_account']))
contract_address = add_0x(to_checksum_address(self.resources['address_declarator']['reference']))
2021-10-11 08:57:03 +02:00
gas_oracle = OverrideGasOracle(limit=AddressDeclarator.gas(), conn=self.rpc)
nonce_oracle = None
if self.rpc != None:
nonce_oracle = RPCNonceOracle(signer_address, conn=self.rpc)
2021-10-11 08:57:03 +02:00
c = Declarator(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
2021-10-10 14:06:40 +02:00
results = []
2021-10-11 19:02:42 +02:00
(main_proof, all_proofs) = self.proof.get()
for proof in all_proofs:
logg.debug('proof {} '.format(proof))
k = 'address_declarator_' + proof
2021-10-10 14:06:40 +02:00
o = c.add_declaration(contract_address, signer_address, self.token_address, proof, tx_format=self.tx_format)
r = None
2021-10-10 14:06:40 +02:00
if self.rpc != None:
r = self.rpc.do(o[1])
2021-10-12 08:39:20 +02:00
self.rpc.wait(r)
2021-10-10 14:06:40 +02:00
elif self.signer != None:
r = o[1]
2021-10-10 14:06:40 +02:00
else:
r = o
self.add_outputs(k, r)
results.append(r)
2021-10-12 08:39:20 +02:00
v = r.encode('utf-8')
if writer != None:
2021-10-12 08:39:20 +02:00
writer.write(k, v)
return results
2021-10-10 14:06:40 +02:00
2021-10-12 08:39:20 +02:00
def prepare_extension(self):
super(CICEth, self).prepare_extension()
2021-10-10 14:06:40 +02:00
2021-10-12 08:39:20 +02:00
if self.token_address != None:
self.token_address = add_0x(to_checksum_address(self.token_address))
2021-10-10 11:33:30 +02:00
def new(chain_spec, resources, proof, signer_hint=None, rpc=None, outputs_writer=None):
return CICEth(chain_spec, resources, proof, signer=signer_hint, rpc=rpc, outputs_writer=outputs_writer)