166 lines
4.5 KiB
Python
166 lines
4.5 KiB
Python
# standard imports
|
|
import logging
|
|
import binascii
|
|
import re
|
|
|
|
# external imports
|
|
#from rlp import encode as rlp_encode
|
|
from hexathon import (
|
|
strip_0x,
|
|
add_0x,
|
|
int_to_minbytes,
|
|
)
|
|
|
|
# local imports
|
|
from crypto_dev_signer.eth.encoding import chain_id_to_v
|
|
from crypto_dev_signer.eth.rlp import rlp_encode
|
|
|
|
logg = logging.getLogger().getChild(__name__)
|
|
|
|
|
|
class Transaction:
|
|
|
|
def rlp_serialize(self):
|
|
raise NotImplementedError
|
|
|
|
def serialize(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class EIP155Transaction:
|
|
|
|
def __init__(self, tx, nonce_in, chainId_in=1):
|
|
to = b''
|
|
data = b''
|
|
if tx.get('to') != None:
|
|
to = bytes.fromhex(strip_0x(tx['to'], allow_empty=True))
|
|
if tx.get('data') != None:
|
|
data = bytes.fromhex(strip_0x(tx['data'], allow_empty=True))
|
|
|
|
gas_price = None
|
|
start_gas = None
|
|
value = None
|
|
nonce = None
|
|
chainId = None
|
|
|
|
# TODO: go directly from hex to bytes
|
|
try:
|
|
gas_price = int(tx['gasPrice'])
|
|
byts = ((gas_price.bit_length()-1)/8)+1
|
|
gas_price = gas_price.to_bytes(int(byts), 'big')
|
|
except ValueError:
|
|
gas_price = bytes.fromhex(strip_0x(tx['gasPrice'], allow_empty=True))
|
|
|
|
try:
|
|
start_gas = int(tx['gas'])
|
|
byts = ((start_gas.bit_length()-1)/8)+1
|
|
start_gas = start_gas.to_bytes(int(byts), 'big')
|
|
except ValueError:
|
|
start_gas = bytes.fromhex(strip_0x(tx['gas'], allow_empty=True))
|
|
|
|
try:
|
|
value = int(tx['value'])
|
|
byts = ((value.bit_length()-1)/8)+1
|
|
value = value.to_bytes(int(byts), 'big')
|
|
except ValueError:
|
|
value = bytes.fromhex(strip_0x(tx['value'], allow_empty=True))
|
|
|
|
try:
|
|
nonce = int(nonce_in)
|
|
byts = ((nonce.bit_length()-1)/8)+1
|
|
nonce = nonce.to_bytes(int(byts), 'big')
|
|
except ValueError:
|
|
nonce = bytes.fromhex(strip_0x(nonce_in, allow_empty=True))
|
|
|
|
try:
|
|
chainId = int(chainId_in)
|
|
byts = ((chainId.bit_length()-1)/8)+1
|
|
chainId = chainId.to_bytes(int(byts), 'big')
|
|
except ValueError:
|
|
chainId = bytes.fromhex(strip_0x(chainId_in, allow_empty=True))
|
|
|
|
self.nonce = nonce
|
|
self.gas_price = gas_price
|
|
self.start_gas = start_gas
|
|
self.to = to
|
|
self.value = value
|
|
self.data = data
|
|
self.v = chainId
|
|
self.r = b''
|
|
self.s = b''
|
|
self.sender = strip_0x(tx['from'])
|
|
|
|
|
|
def __canonical_order(self):
|
|
s = [
|
|
self.nonce,
|
|
self.gas_price,
|
|
self.start_gas,
|
|
self.to,
|
|
self.value,
|
|
self.data,
|
|
self.v,
|
|
self.r,
|
|
self.s,
|
|
]
|
|
|
|
return s
|
|
|
|
|
|
def bytes_serialize(self):
|
|
s = self.__canonical_order()
|
|
b = b''
|
|
for e in s:
|
|
b += e
|
|
return b
|
|
|
|
|
|
def rlp_serialize(self):
|
|
s = self.__canonical_order()
|
|
return rlp_encode(s)
|
|
|
|
|
|
def serialize(self):
|
|
tx = {
|
|
'nonce': add_0x(self.nonce.hex(), allow_empty=True),
|
|
'gasPrice': add_0x(self.gas_price.hex()),
|
|
'gas': add_0x(self.start_gas.hex()),
|
|
'to': add_0x(self.to.hex()),
|
|
'value': add_0x(self.value.hex(), allow_empty=True),
|
|
'data': add_0x(self.data.hex(), allow_empty=True),
|
|
'v': add_0x(self.v.hex(), allow_empty=True),
|
|
'r': add_0x(self.r.hex(), allow_empty=True),
|
|
's': add_0x(self.s.hex(), allow_empty=True),
|
|
}
|
|
if tx['data'] == '':
|
|
tx['data'] = '0x'
|
|
|
|
if tx['value'] == '':
|
|
tx['value'] = '0x00'
|
|
|
|
if tx['nonce'] == '':
|
|
tx['nonce'] = '0x00'
|
|
|
|
return tx
|
|
|
|
|
|
def apply_signature(self, chain_id, signature):
|
|
if len(self.r + self.s) > 0:
|
|
raise AttributeError('signature already set')
|
|
if len(signature) < 65:
|
|
raise ValueError('invalid signature length')
|
|
v = chain_id_to_v(chain_id, signature)
|
|
self.v = int_to_minbytes(v)
|
|
self.r = signature[:32]
|
|
self.s = signature[32:64]
|
|
|
|
for i in range(len(self.r)):
|
|
if self.r[i] > 0:
|
|
self.r = self.r[i:]
|
|
break
|
|
|
|
for i in range(len(self.s)):
|
|
if self.s[i] > 0:
|
|
self.s = self.s[i:]
|
|
break
|