transaction.py (4693B)
1 # standard imports 2 import logging 3 import binascii 4 import re 5 6 # external imports 7 #from rlp import encode as rlp_encode 8 from hexathon import ( 9 strip_0x, 10 add_0x, 11 int_to_minbytes, 12 ) 13 14 # local imports 15 from funga.eth.encoding import chain_id_to_v 16 #from funga.eth.rlp import rlp_encode 17 import rlp 18 19 logg = logging.getLogger(__name__) 20 21 rlp_encode = rlp.encode 22 23 class Transaction: 24 25 def rlp_serialize(self): 26 raise NotImplementedError 27 28 def serialize(self): 29 raise NotImplementedError 30 31 32 class EIP155Transaction: 33 34 def __init__(self, tx, nonce_in, chainId_in=1): 35 to = b'' 36 data = b'' 37 if tx.get('to') != None: 38 to = bytes.fromhex(strip_0x(tx['to'], allow_empty=True)) 39 if tx.get('data') != None: 40 data = bytes.fromhex(strip_0x(tx['data'], allow_empty=True)) 41 42 gas_price = None 43 start_gas = None 44 value = None 45 nonce = None 46 chainId = None 47 48 # TODO: go directly from hex to bytes 49 try: 50 gas_price = int(tx['gasPrice']) 51 byts = ((gas_price.bit_length()-1)/8)+1 52 gas_price = gas_price.to_bytes(int(byts), 'big') 53 except ValueError: 54 gas_price = bytes.fromhex(strip_0x(tx['gasPrice'], allow_empty=True)) 55 56 try: 57 start_gas = int(tx['gas']) 58 byts = ((start_gas.bit_length()-1)/8)+1 59 start_gas = start_gas.to_bytes(int(byts), 'big') 60 except ValueError: 61 start_gas = bytes.fromhex(strip_0x(tx['gas'], allow_empty=True)) 62 63 try: 64 value = int(tx['value']) 65 byts = ((value.bit_length()-1)/8)+1 66 value = value.to_bytes(int(byts), 'big') 67 except ValueError: 68 value = bytes.fromhex(strip_0x(tx['value'], allow_empty=True)) 69 70 try: 71 nonce = int(nonce_in) 72 byts = ((nonce.bit_length()-1)/8)+1 73 nonce = nonce.to_bytes(int(byts), 'big') 74 except ValueError: 75 nonce = bytes.fromhex(strip_0x(nonce_in, allow_empty=True)) 76 77 try: 78 chainId = int(chainId_in) 79 byts = ((chainId.bit_length()-1)/8)+1 80 chainId = chainId.to_bytes(int(byts), 'big') 81 except ValueError: 82 chainId = bytes.fromhex(strip_0x(chainId_in, allow_empty=True)) 83 84 self.nonce = nonce 85 self.gas_price = gas_price 86 self.start_gas = start_gas 87 self.to = to 88 self.value = value 89 self.data = data 90 self.v = chainId 91 self.r = b'' 92 self.s = b'' 93 self.sender = strip_0x(tx['from']) 94 95 96 def canonical_order(self): 97 s = [ 98 self.nonce, 99 self.gas_price, 100 self.start_gas, 101 self.to, 102 self.value, 103 self.data, 104 self.v, 105 self.r, 106 self.s, 107 ] 108 109 return s 110 111 112 def bytes_serialize(self): 113 s = self.canonical_order() 114 b = b'' 115 for e in s: 116 b += e 117 return b 118 119 120 def rlp_serialize(self): 121 s = self.canonical_order() 122 return rlp_encode(s) 123 124 125 def serialize(self): 126 tx = { 127 'nonce': add_0x(self.nonce.hex(), allow_empty=True), 128 'gasPrice': add_0x(self.gas_price.hex()), 129 'gas': add_0x(self.start_gas.hex()), 130 'value': add_0x(self.value.hex(), allow_empty=True), 131 'data': add_0x(self.data.hex(), allow_empty=True), 132 'v': add_0x(self.v.hex(), allow_empty=True), 133 'r': add_0x(self.r.hex(), allow_empty=True), 134 's': add_0x(self.s.hex(), allow_empty=True), 135 } 136 if self.to == None or len(self.to) == 0: 137 tx['to'] = None 138 else: 139 tx['to'] = add_0x(self.to.hex()) 140 141 if tx['data'] == '': 142 tx['data'] = '0x' 143 144 if tx['value'] == '': 145 tx['value'] = '0x00' 146 147 if tx['nonce'] == '': 148 tx['nonce'] = '0x00' 149 150 return tx 151 152 153 def apply_signature(self, chain_id, signature, v=None): 154 if len(self.r + self.s) > 0: 155 raise AttributeError('signature already set') 156 if len(signature) < 65: 157 raise ValueError('invalid signature length') 158 if v == None: 159 v = chain_id_to_v(chain_id, signature) 160 self.v = int_to_minbytes(v) 161 self.r = signature[:32] 162 self.s = signature[32:64] 163 164 for i in range(len(self.r)): 165 if self.r[i] > 0: 166 self.r = self.r[i:] 167 break 168 169 for i in range(len(self.s)): 170 if self.s[i] > 0: 171 self.s = self.s[i:] 172 break