limiter.py (3409B)
1 # standard imports 2 import logging 3 import os 4 import enum 5 6 # external imports 7 from chainlib.eth.constant import ZERO_ADDRESS 8 from chainlib.eth.constant import ZERO_CONTENT 9 from chainlib.eth.contract import ( 10 ABIContractEncoder, 11 ABIContractDecoder, 12 ABIContractType, 13 abi_decode_single, 14 ) 15 from chainlib.eth.jsonrpc import to_blockheight_param 16 from chainlib.eth.error import RequestMismatchException 17 from chainlib.eth.tx import ( 18 TxFactory, 19 TxFormat, 20 ) 21 from chainlib.jsonrpc import JSONRPCRequest 22 from chainlib.block import BlockSpec 23 from hexathon import ( 24 add_0x, 25 strip_0x, 26 ) 27 from chainlib.eth.cli.encode import CLIEncoder 28 29 # local imports 30 from erc20_limiter.data import data_dir 31 32 logg = logging.getLogger() 33 34 35 class Limiter(TxFactory): 36 37 __abi = None 38 __bytecode = None 39 40 def constructor(self, sender_address, tx_format=TxFormat.JSONRPC, version=None): 41 code = self.cargs(version=version) 42 tx = self.template(sender_address, None, use_nonce=True) 43 tx = self.set_code(tx, code) 44 return self.finalize(tx, tx_format) 45 46 47 @staticmethod 48 def cargs(version=None): 49 code = Limiter.bytecode(version=version) 50 enc = ABIContractEncoder() 51 args = enc.get() 52 code += args 53 logg.debug('constructor code: ' + args) 54 return code 55 56 57 @staticmethod 58 def gas(code=None): 59 return 4000000 60 61 62 @staticmethod 63 def abi(): 64 if Limiter.__abi == None: 65 f = open(os.path.join(data_dir, 'Limiter.json'), 'r') 66 Limiter.__abi = json.load(f) 67 f.close() 68 return Limiter.__abi 69 70 71 @staticmethod 72 def bytecode(version=None): 73 if Limiter.__bytecode == None: 74 f = open(os.path.join(data_dir, 'Limiter.bin')) 75 Limiter.__bytecode = f.read() 76 f.close() 77 return Limiter.__bytecode 78 79 80 def set_limit(self, contract_address, sender_address, token_address, limit, holder_address=None, tx_format=TxFormat.JSONRPC, id_generator=None): 81 enc = ABIContractEncoder() 82 if holder_address == None: 83 enc.method('setLimit') 84 else: 85 enc.method('setLimitFor') 86 enc.typ(ABIContractType.ADDRESS) 87 if holder_address != None: 88 enc.typ(ABIContractType.ADDRESS) 89 enc.typ(ABIContractType.UINT256) 90 enc.address(token_address) 91 if holder_address != None: 92 enc.address(holder_address) 93 enc.uint256(limit) 94 data = add_0x(enc.get()) 95 tx = self.template(sender_address, contract_address, use_nonce=True) 96 tx = self.set_code(tx, data) 97 tx = self.finalize(tx, tx_format, id_generator=id_generator) 98 return tx 99 100 101 def limit_of(self, contract_address, token_address, holder_address, sender_address=ZERO_ADDRESS, id_generator=None): 102 j = JSONRPCRequest(id_generator) 103 o = j.template() 104 o['method'] = 'eth_call' 105 enc = ABIContractEncoder() 106 enc.method('limitOf') 107 enc.typ(ABIContractType.ADDRESS) 108 enc.typ(ABIContractType.ADDRESS) 109 enc.address(token_address) 110 enc.address(holder_address) 111 data = add_0x(enc.get()) 112 tx = self.template(sender_address, contract_address) 113 tx = self.set_code(tx, data) 114 o['params'].append(self.normalize(tx)) 115 o['params'].append('latest') 116 o = j.finalize(o) 117 return o 118 119