511 lines
14 KiB
Python
511 lines
14 KiB
Python
# standard imports
|
|
import enum
|
|
import re
|
|
import logging
|
|
|
|
# external imports
|
|
from hexathon import (
|
|
strip_0x,
|
|
pad,
|
|
)
|
|
|
|
# local imports
|
|
from chainlib.hash import keccak256_string_to_hex
|
|
from chainlib.block import BlockSpec
|
|
from chainlib.jsonrpc import JSONRPCRequest
|
|
from .address import to_checksum_address
|
|
|
|
logg = logging.getLogger(__name__)
|
|
|
|
|
|
re_method = r'^[a-zA-Z0-9_]+$'
|
|
|
|
class ABIContractType(enum.Enum):
|
|
"""Data types used by ABI encoders
|
|
"""
|
|
BYTES32 = 'bytes32'
|
|
BYTES4 = 'bytes4'
|
|
UINT256 = 'uint256'
|
|
ADDRESS = 'address'
|
|
STRING = 'string'
|
|
BOOLEAN = 'bool'
|
|
|
|
dynamic_contract_types = [
|
|
ABIContractType.STRING,
|
|
]
|
|
|
|
|
|
class ABIContract:
|
|
"""Base class for Ethereum smart contract encoder
|
|
"""
|
|
def __init__(self):
|
|
self.types = []
|
|
self.contents = []
|
|
|
|
|
|
class ABIMethodEncoder(ABIContract):
|
|
"""Generate ABI method signatures from method signature string.
|
|
"""
|
|
def __init__(self):
|
|
super(ABIMethodEncoder, self).__init__()
|
|
self.method_name = None
|
|
self.method_contents = []
|
|
|
|
|
|
def method(self, m):
|
|
"""Set method name.
|
|
|
|
:param m: Method name
|
|
:type m: str
|
|
:raises ValueError: Invalid method name
|
|
"""
|
|
if re.match(re_method, m) == None:
|
|
raise ValueError('Invalid method {}, must match regular expression {}'.format(re_method))
|
|
self.method_name = m
|
|
self.__log_method()
|
|
|
|
|
|
def get_method(self):
|
|
"""Return currently set method signature string.
|
|
|
|
:rtype: str
|
|
:returns: Method signature
|
|
"""
|
|
if self.method_name == None:
|
|
return ''
|
|
return '{}({})'.format(self.method_name, ','.join(self.method_contents))
|
|
|
|
|
|
def typ(self, v):
|
|
"""Add argument type to argument vector.
|
|
|
|
Method name must be set before this is called.
|
|
|
|
:param v: Type to add
|
|
:type v: chainlib.eth.contract.ABIContractType
|
|
:raises AttributeError: Type set before method name
|
|
:raises TypeError: Invalid type
|
|
"""
|
|
if self.method_name == None:
|
|
raise AttributeError('method name must be set before adding types')
|
|
if not isinstance(v, ABIContractType):
|
|
raise TypeError('method type not valid; expected {}, got {}'.format(type(ABIContractType).__name__, type(v).__name__))
|
|
self.method_contents.append(v.value)
|
|
self.__log_method()
|
|
|
|
|
|
def __log_method(self):
|
|
logg.debug('method set to {}'.format(self.get_method()))
|
|
|
|
|
|
|
|
class ABIContractDecoder(ABIContract):
|
|
"""Decode serialized ABI contract input data to corresponding python primitives.
|
|
"""
|
|
|
|
def typ(self, v):
|
|
"""Add type to argument array to parse input against.
|
|
|
|
:param v: Type
|
|
:type v: chainlib.eth.contract.ABIContractType
|
|
:raises TypeError: Invalid type
|
|
"""
|
|
if not isinstance(v, ABIContractType):
|
|
raise TypeError('method type not valid; expected {}, got {}'.format(type(ABIContractType).__name__, type(v).__name__))
|
|
self.types.append(v.value)
|
|
self.__log_typ()
|
|
|
|
|
|
def val(self, v):
|
|
"""Add value to value array.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
"""
|
|
self.contents.append(v)
|
|
logg.debug('content is now {}'.format(self.contents))
|
|
|
|
|
|
def uint256(self, v):
|
|
"""Parse value as uint256.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
:rtype: int
|
|
:returns: Int value
|
|
"""
|
|
return int(v, 16)
|
|
|
|
|
|
def bytes32(self, v):
|
|
"""Parse value as bytes32.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
:rtype: str
|
|
:returns: Value, in hex
|
|
"""
|
|
return v
|
|
|
|
|
|
def bool(self, v):
|
|
"""Parse value as bool.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
:rtype: bool
|
|
:returns: Value
|
|
"""
|
|
return bool(self.uint256(v))
|
|
|
|
|
|
def boolean(self, v):
|
|
"""Alias of chainlib.eth.contract.ABIContractDecoder.bool
|
|
"""
|
|
return bool(self.uint256(v))
|
|
|
|
|
|
def address(self, v):
|
|
"""Parse value as address.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
:rtype: str
|
|
:returns: Value. in hex
|
|
"""
|
|
a = strip_0x(v)[64-40:]
|
|
return to_checksum_address(a)
|
|
|
|
|
|
def string(self, v):
|
|
"""Parse value as string.
|
|
|
|
:param v: Value, in hex
|
|
:type v: str
|
|
:rtype: str
|
|
:returns: Value
|
|
"""
|
|
s = strip_0x(v)
|
|
b = bytes.fromhex(s)
|
|
cursor = 0
|
|
offset = int.from_bytes(b[cursor:cursor+32], 'big')
|
|
cursor += 32
|
|
length = int.from_bytes(b[cursor:cursor+32], 'big')
|
|
cursor += 32
|
|
content = b[cursor:cursor+length]
|
|
logg.debug('parsing string offset {} length {} content {}'.format(offset, length, content))
|
|
return content.decode('utf-8')
|
|
|
|
|
|
def __log_typ(self):
|
|
logg.debug('types set to ({})'.format(','.join(self.types)))
|
|
|
|
|
|
def decode(self):
|
|
"""Apply decoder on value array using argument type array.
|
|
|
|
:rtype: list
|
|
:returns: List of decoded values
|
|
"""
|
|
r = []
|
|
logg.debug('contents {}'.format(self.contents))
|
|
for i in range(len(self.types)):
|
|
m = getattr(self, self.types[i])
|
|
s = self.contents[i]
|
|
r.append(m(s))
|
|
return r
|
|
|
|
|
|
def get(self):
|
|
"""Alias of chainlib.eth.contract.ABIContractDecoder.decode
|
|
"""
|
|
return self.decode()
|
|
|
|
|
|
def __str__(self):
|
|
return self.decode()
|
|
|
|
|
|
class ABIContractLogDecoder(ABIMethodEncoder, ABIContractDecoder):
|
|
"""Decoder utils for log entries of an Ethereum network transaction receipt.
|
|
|
|
Uses chainlib.eth.contract.ABIContractDecoder.decode to render output from template.
|
|
"""
|
|
def __init__(self):
|
|
super(ABIContractLogDecoder, self).__init__()
|
|
self.method_name = None
|
|
self.indexed_content = []
|
|
|
|
|
|
def topic(self, event):
|
|
"""Set topic to match.
|
|
|
|
:param event: Topic name
|
|
:type event: str
|
|
"""
|
|
self.method(event)
|
|
|
|
|
|
def get_method_signature(self):
|
|
"""Generate topic signature from set topic.
|
|
|
|
:rtype: str
|
|
:returns: Topic signature, in hex
|
|
"""
|
|
s = self.get_method()
|
|
return keccak256_string_to_hex(s)
|
|
|
|
|
|
def typ(self, v):
|
|
"""Add type to event argument array.
|
|
|
|
:param v: Type
|
|
:type v: chainlib.eth.contract.ABIContractType
|
|
"""
|
|
super(ABIContractLogDecoder, self).typ(v)
|
|
self.types.append(v.value)
|
|
|
|
|
|
def apply(self, topics, data):
|
|
"""Set log entry data to parse.
|
|
|
|
After set, self.decode can be used to render the output.
|
|
|
|
:param topics: The topics array of the receipt, list of hex
|
|
:type topics: list
|
|
:param data: Non-indexed data, in hex
|
|
:type data: str
|
|
:raises ValueError: Topic of input does not match topic set in parser
|
|
"""
|
|
t = self.get_method_signature()
|
|
if topics[0] != t:
|
|
raise ValueError('topic mismatch')
|
|
for i in range(len(topics) - 1):
|
|
self.contents.append(topics[i+1])
|
|
self.contents += data
|
|
|
|
|
|
class ABIContractEncoder(ABIMethodEncoder):
|
|
|
|
def __log_latest(self, v):
|
|
l = len(self.types) - 1
|
|
logg.debug('Encoder added {} -> {} ({})'.format(v, self.contents[l], self.types[l].value))
|
|
|
|
|
|
def uint256(self, v):
|
|
"""Encode value to uint256 and add to input value vector.
|
|
|
|
:param v: Integer value
|
|
:type v: int
|
|
"""
|
|
v = int(v)
|
|
b = v.to_bytes(32, 'big')
|
|
self.contents.append(b.hex())
|
|
self.types.append(ABIContractType.UINT256)
|
|
self.__log_latest(v)
|
|
|
|
|
|
def bool(self, v):
|
|
"""Alias of chainlib.eth.contract.ABIContractEncoder.boolean.
|
|
"""
|
|
return self.boolean(v)
|
|
|
|
|
|
def boolean(self, v):
|
|
"""Encode value to boolean and add to input value vector.
|
|
|
|
:param v: Trueish or falsish value
|
|
:type v: any
|
|
:rtype: See chainlib.eth.contract.ABIContractEncoder.uint256
|
|
:returns: See chainlib.eth.contract.ABIContractEncoder.uint256
|
|
"""
|
|
if bool(v):
|
|
return self.uint256(1)
|
|
return self.uint256(0)
|
|
|
|
|
|
def address(self, v):
|
|
"""Encode value to address and add to input value vector.
|
|
|
|
:param v: Ethereum address, in hex
|
|
:type v: str
|
|
"""
|
|
self.bytes_fixed(32, v, 20)
|
|
self.types.append(ABIContractType.ADDRESS)
|
|
self.__log_latest(v)
|
|
|
|
|
|
def bytes32(self, v):
|
|
"""Encode value to bytes32 and add to input value vector.
|
|
|
|
:param v: Bytes, in hex
|
|
:type v: str
|
|
"""
|
|
self.bytes_fixed(32, v)
|
|
self.types.append(ABIContractType.BYTES32)
|
|
self.__log_latest(v)
|
|
|
|
|
|
def bytes4(self, v):
|
|
"""Encode value to bytes4 and add to input value vector.
|
|
|
|
:param v: Bytes, in hex
|
|
:type v: str
|
|
"""
|
|
self.bytes_fixed(4, v)
|
|
self.types.append(ABIContractType.BYTES4)
|
|
self.__log_latest(v)
|
|
|
|
|
|
|
|
def string(self, v):
|
|
"""Encode value to string and add to input value vector.
|
|
|
|
:param v: String input
|
|
:type v: str
|
|
"""
|
|
b = v.encode('utf-8')
|
|
l = len(b)
|
|
contents = l.to_bytes(32, 'big')
|
|
contents += b
|
|
padlen = 32 - (l % 32)
|
|
contents += padlen * b'\x00'
|
|
self.bytes_fixed(len(contents), contents)
|
|
self.types.append(ABIContractType.STRING)
|
|
self.__log_latest(v)
|
|
return contents
|
|
|
|
|
|
def bytes_fixed(self, mx, v, exact=0):
|
|
"""Add arbirary length byte data to value vector.
|
|
|
|
:param mx: Max length of input data.
|
|
:type mx: int
|
|
:param v: Byte input, hex or bytes
|
|
:type v: str | bytes
|
|
:param exact: Fail parsing if input does not translate to given byte length.
|
|
:type exact: int
|
|
:raises ValueError: Input length or input format mismatch.
|
|
"""
|
|
typ = type(v).__name__
|
|
if typ == 'str':
|
|
v = strip_0x(v)
|
|
l = len(v)
|
|
if exact > 0 and l != exact * 2:
|
|
raise ValueError('value wrong size; expected {}, got {})'.format(mx, l))
|
|
if l > mx * 2:
|
|
raise ValueError('value too long ({})'.format(l))
|
|
v = pad(v, mx)
|
|
elif typ == 'bytes':
|
|
l = len(v)
|
|
if exact > 0 and l != exact:
|
|
raise ValueError('value wrong size; expected {}, got {})'.format(mx, l))
|
|
b = bytearray(mx)
|
|
b[mx-l:] = v
|
|
v = pad(b.hex(), mx)
|
|
else:
|
|
raise ValueError('invalid input {}'.format(typ))
|
|
self.contents.append(v.ljust(64, '0'))
|
|
|
|
|
|
def get_method_signature(self):
|
|
"""Return abi encoded signature of currently set method.
|
|
"""
|
|
s = self.get_method()
|
|
if s == '':
|
|
return s
|
|
return keccak256_string_to_hex(s)[:8]
|
|
|
|
|
|
def get_contents(self):
|
|
"""Encode value array.
|
|
|
|
:rtype: str
|
|
:returns: ABI encoded values, in hex
|
|
"""
|
|
direct_contents = ''
|
|
pointer_contents = ''
|
|
l = len(self.types)
|
|
pointer_cursor = 32 * l
|
|
for i in range(l):
|
|
if self.types[i] in dynamic_contract_types:
|
|
content_length = len(self.contents[i])
|
|
pointer_contents += self.contents[i]
|
|
direct_contents += pointer_cursor.to_bytes(32, 'big').hex()
|
|
pointer_cursor += int(content_length / 2)
|
|
else:
|
|
direct_contents += self.contents[i]
|
|
s = ''.join(direct_contents + pointer_contents)
|
|
for i in range(0, len(s), 64):
|
|
l = len(s) - i
|
|
if l > 64:
|
|
l = 64
|
|
logg.debug('code word {} {}'.format(int(i / 64), s[i:i+64]))
|
|
return s
|
|
|
|
|
|
def get(self):
|
|
"""Alias of chainlib.eth.contract.ABIContractEncoder.encode
|
|
"""
|
|
return self.encode()
|
|
|
|
|
|
def encode(self):
|
|
"""Encode method and value array.
|
|
|
|
The data generated by this method is the literal data used as input to contract calls or transactions.
|
|
|
|
:rtype: str
|
|
:returns: ABI encoded contract input data, in hex
|
|
"""
|
|
m = self.get_method_signature()
|
|
c = self.get_contents()
|
|
return m + c
|
|
|
|
|
|
def __str__(self):
|
|
return self.encode()
|
|
|
|
|
|
|
|
def abi_decode_single(typ, v):
|
|
"""Convenience function to decode a single ABI encoded value against a given type.
|
|
|
|
:param typ: Type to parse value as
|
|
:type typ: chainlib.eth.contract.ABIContractEncoder
|
|
:param v: Value to parse, in hex
|
|
:type v: str
|
|
"""
|
|
d = ABIContractDecoder()
|
|
d.typ(typ)
|
|
d.val(v)
|
|
r = d.decode()
|
|
return r[0]
|
|
|
|
|
|
def code(address, block_spec=BlockSpec.LATEST, id_generator=None):
|
|
"""Generate json-rpc query to retrieve code stored at an Ethereum address.
|
|
|
|
:param address: Address to use for query, in hex
|
|
:type address: str
|
|
:param block_spec: Block height spec
|
|
:type block_spec: chainlib.block.BlockSpec
|
|
:param id_generator: json-rpc id generator
|
|
:type id_generator: chainlib.jsonrpc.JSONRPCIdGenerator
|
|
:rtype: dict
|
|
:returns: rpc query object
|
|
"""
|
|
block_height = None
|
|
if block_spec == BlockSpec.LATEST:
|
|
block_height = 'latest'
|
|
elif block_spec == BlockSpec.PENDING:
|
|
block_height = 'pending'
|
|
else:
|
|
block_height = int(block_spec)
|
|
j = JSONRPCRequest(id_generator)
|
|
o = j.template()
|
|
o['method'] = 'eth_getCode'
|
|
o['params'].append(address)
|
|
o['params'].append(block_height)
|
|
return j.finalize(o)
|