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

367 lines
11 KiB
Python
Raw Normal View History

# standard imports
2021-10-10 13:48:32 +02:00
import copy
2021-10-15 12:48:14 +02:00
import json
import logging
# external imports
2021-10-10 12:50:18 +02:00
from chainlib.chain import ChainSpec
from chainlib.eth.address import is_address, to_checksum_address
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.tx import Tx, TxFactory, TxFormat, receipt
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
from eth_token_index import TokenUniqueSymbolIndex
2021-10-12 19:39:18 +02:00
from giftable_erc20_token import GiftableToken
from hexathon import add_0x, strip_0x
# 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):
def __init__(
self,
chain_spec,
resources,
proof,
signer=None,
rpc=None,
outputs_writer=None,
fee_oracle=None,
):
2021-11-29 15:39:16 +01:00
"""Implementation for the eth extension.
The state of the resources instance will be modified.
:param chain_spec: Chain Spec that extension will operate for
:type chain_spec: chainlib.chain.ChainSpec
:param resources: Chain application resources to deploy or interface with
:type resources: dict
:param proof: Proof object to publish
:type proof: cic.proof.Proof
:param signer: Signer capable of generating signatures for chain aplication deployments
:type signer: funga.signer.Signer
:param rpc: RPC adapter capable of submitting and querying the chain network node
:type rpc: chainlib.connection.RPCConnection
:param outputs_writer: Writer interface receiving the output of the processor
:type outputs_writer: cic.writers.OutputWriter
2021-11-29 15:39:16 +01:00
:param fee_oracle: Fee oracle required by signer
:type fee_oracle: chainlib.fee.FeeOracle
2021-10-10 13:48:32 +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
if self.rpc is not None:
2021-10-10 13:48:32 +02:00
self.tx_format = TxFormat.JSONRPC
elif self.signer is not None:
2021-10-10 13:48:32 +02:00
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 is None:
try:
vv = strip_0x(v)
if is_address(vv):
typ = ABIContractType.ADDRESS
else:
typ = ABIContractType.BYTES32
except ValueError:
pass
if typ is None:
try:
v.encode("utf-8")
typ = ABIContractType.STRING
except ValueError:
pass
if typ is None:
raise ValueError(
f"cannot automatically determine type for value {v}"
)
logg.info(f"argument {v} parsed as abi contract type {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"] is not 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 is 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):
2021-11-29 19:15:10 +01:00
"""Adds given key/value pair to outputs array.
:param k: Output key
:type k: str
:param v: Output value
:param v: bytes or str
"""
logg.debug(f"adding outputs {k} {v}")
self.outputs.append((k, v))
def get_outputs(self):
2021-11-29 19:15:10 +01:00
"""Get wrapper for outputs captured from processing.
:rtype: list of tuples
:return: Captured outputs
"""
return self.outputs
def process_token(self, writer=None):
2021-11-29 19:15:10 +01:00
"""Deploy token, and optionally mint token supply to token deployer account.
:param writer: Writer interface receiving the output of the processor step
:type writer: cic.writers.OutputWriter
2021-11-29 19:15:10 +01:00
"""
if writer is 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 is not None:
2021-10-12 08:39:20 +02:00
code = self.token_code + code
2021-10-10 15:37:26 +02:00
logg.debug(f"resource {self.resources}")
signer_address = add_0x(
to_checksum_address(self.resources["token"]["key_account"])
)
nonce_oracle = None
if self.rpc is not 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)
2021-10-15 12:48:14 +02:00
token_address_tx = None
2021-10-10 15:37:26 +02:00
r = None
if self.rpc is not None:
2021-10-10 15:37:26 +02:00
r = self.rpc.do(o[1])
2021-10-15 12:48:14 +02:00
token_address_tx = r
o = self.rpc.wait(r)
o = Tx.src_normalize(o)
self.token_address = o["contract_address"]
elif self.signer is not None:
2021-10-10 15:37:26 +02:00
r = o[1]
2021-10-15 12:48:14 +02:00
token_address_tx = r
2021-10-10 12:50:18 +02:00
if r is None:
2021-10-10 15:37:26 +02:00
r = code
writer.write("token", r.encode("utf-8"))
writer.write("token_address", self.token_address.encode("utf-8"))
self.add_outputs("token", r)
if int(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"],
)
2021-10-12 19:39:18 +02:00
r = None
if self.rpc is not None:
2021-10-12 19:39:18 +02:00
r = self.rpc.do(o[1])
self.rpc.wait(r)
writer.write("token_supply", r.encode("utf-8"))
elif self.signer is not None:
2021-10-12 19:39:18 +02:00
r = o[1]
writer.write(
"token_supply", json.dumps(r, separators=(",", ":")).encode("utf-8")
)
2021-10-12 19:39:18 +02:00
else:
r = o
writer.write("token_supply", r.encode("utf-8"))
2021-10-15 12:48:14 +02:00
return token_address_tx
2021-10-10 12:50:18 +02:00
def process_token_index(self, writer=None):
2021-11-29 19:15:10 +01:00
"""Register deployed token with token index.
:param writer: Writer interface receiving the output of the processor step
:type writer: cic.writers.OutputWriter
2021-11-29 19:15:10 +01:00
"""
if writer is None:
writer = self.outputs_writer
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 is not None:
2021-10-12 08:39:20 +02:00
nonce_oracle = RPCNonceOracle(add_0x(signer_address), conn=self.rpc)
c = TokenUniqueSymbolIndex(
self.chain_spec,
signer=self.signer,
nonce_oracle=nonce_oracle,
gas_oracle=gas_oracle,
)
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
if self.rpc is not 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)
elif self.signer is not 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
def process_address_declarator(self, writer=None):
2021-11-29 19:15:10 +01:00
"""Register token proofs with address declarator.
:param writer: Writer interface receiving the output of the processor step
:type writer: cic.writers.OutputWriter
2021-11-29 19:15:10 +01:00
"""
if writer is None:
writer = self.outputs_writer
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 is not None:
nonce_oracle = RPCNonceOracle(signer_address, conn=self.rpc)
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 = []
# (main_proof, all_proofs) = self.proof.get()
# for proof in all_proofs:
# logg.debug('proof {} '.format(proof))
(k, v) = self.proof.root()
fk = "address_declarator_" + k
o = c.add_declaration(
contract_address,
signer_address,
self.token_address,
k,
tx_format=self.tx_format,
)
r = None
if self.rpc is not None:
r = self.rpc.do(o[1])
self.rpc.wait(r)
elif self.signer is not None:
r = o[1]
else:
r = o
self.add_outputs(fk, r)
results.append(r)
v = r.encode("utf-8")
if writer is not None:
writer.write(fk, v)
return results
2021-10-10 14:06:40 +02:00
2021-10-12 08:39:20 +02:00
def prepare_extension(self):
"""Sets token address for extension if defined in settings."""
2021-10-12 08:39:20 +02:00
super(CICEth, self).prepare_extension()
2021-10-10 14:06:40 +02:00
if self.token_address is not None:
2021-10-12 08:39:20 +02:00
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):
2021-11-29 19:15:10 +01:00
"""Convenience function to enable object instantiation through predictable module symbol
2021-11-30 08:14:57 +01:00
See CICEth constructor for details.
2021-11-29 19:15:10 +01:00
"""
return CICEth(
chain_spec,
resources,
proof,
signer=signer_hint,
rpc=rpc,
outputs_writer=outputs_writer,
)