erc20-transfer-authorization

Simple approval escrow for ERC20 spending
Info | Log | Files | Refs

transfer_authorization.py (11076B)


      1 # Author:	Louis Holbrook <dev@holbrook.no> 0826EDA1702D1E87C6E2875121D2E7BB88C2A746
      2 # SPDX-License-Identifier:	GPL-3.0-or-later
      3 # File-version: 1
      4 # Description: Python interface to abi and bin files for faucet contracts
      5 
      6 # standard imports
      7 import logging
      8 import json
      9 import os
     10 
     11 # external imports
     12 from chainlib.eth.tx import (
     13         TxFactory,
     14         TxFormat,
     15         )
     16 from chainlib.eth.constant import ZERO_ADDRESS
     17 from chainlib.eth.contract import (
     18         ABIContractEncoder,
     19         ABIContractDecoder,
     20         ABIContractType,
     21         abi_decode_single,
     22         )
     23 from chainlib.jsonrpc import JSONRPCRequest
     24 from hexathon import (
     25         add_0x,
     26         strip_0x,
     27         )
     28 
     29 logg = logging.getLogger().getChild(__name__)
     30 
     31 moddir = os.path.dirname(__file__)
     32 datadir = os.path.join(moddir, 'data')
     33 
     34 
     35 class Request:
     36 
     37     def __init__(self, sender, recipient, value, token):
     38         self.sender = sender
     39         self.recipient = recipient
     40         self.value = value
     41         self.token = token
     42         self.serial = 0
     43         self.yay = 0
     44         self.nay = 0
     45         self.result = 0
     46 
     47 
     48     @classmethod
     49     def create(cls, sender, recipient, value, token, *args):
     50         o = Request(sender, recipient, value, token)
     51         if len(args) > 0:
     52             o.serial = args[0]
     53             o.yay = args[1]
     54             o.nsy = args[2]
     55             o.result = args[3]
     56         return o
     57 
     58 
     59     def is_accepted(self):
     60         return self.result & 2 > 0
     61 
     62 
     63     def is_rejected(self):
     64         return self.result & 4 > 0
     65 
     66 
     67     def is_transferred(self):
     68         return self.result & 24 == 8
     69 
     70 
     71     def is_voting(self):
     72         return self.result & 7 == 1
     73 
     74 
     75     def is_executed(self):
     76         return self.result & 8 == 8
     77 
     78 
     79     def __str__(self):
     80         return "{} {} {} {}".format(self.sender, self.recipient, self.value, self.result)
     81 
     82 
     83 
     84 class TransferAuthorization(TxFactory):
     85 
     86     __abi = None
     87     __bytecode = None
     88 
     89 
     90     def __init__(self, *args, **kwargs):
     91         super(TransferAuthorization, self).__init__(*args, **kwargs)
     92 
     93 
     94     def count(self, *args, **kwargs):
     95         return self.call_noarg('count', *args, **kwargs)
     96 
     97 
     98     def quorum_threshold(self, *args, **kwargs):
     99         return self.call_noarg('quorum', *args, **kwargs)
    100 
    101 
    102     def veto_threshold(self, *args, **kwargs):
    103         return self.call_noarg('vetoThreshold', *args, **kwargs)
    104 
    105 
    106     def signer_count(self, *args, **kwargs):
    107         return self.call_noarg('signerCount', *args, **kwargs)
    108 
    109 
    110     def last_serial(self, *args, **kwargs):
    111         return self.call_noarg('lastSerial', *args, **kwargs)
    112 
    113 
    114     def next_serial(self, *args, **kwargs):
    115         return self.call_noarg('nextSerial', *args, **kwargs)
    116 
    117 
    118     @staticmethod
    119     def abi():
    120         if TransferAuthorization.__abi == None:
    121             f = open(os.path.join(datadir, 'ERC20TransferAuthorization.json'), 'r')
    122             TransferAuthorization.__abi = json.load(f)
    123             f.close()
    124         return TransferAuthorization.__abi
    125 
    126 
    127     @staticmethod
    128     def bytecode():
    129         if TransferAuthorization.__bytecode == None:
    130             f = open(os.path.join(datadir, 'ERC20TransferAuthorization.bin'))
    131             TransferAuthorization.__bytecode = f.read()
    132             f.close()
    133         return TransferAuthorization.__bytecode
    134 
    135 
    136     @staticmethod
    137     def gas(code=None):
    138         return 2800000
    139 
    140 
    141     def __single_address_method(self, method, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
    142         enc = ABIContractEncoder()
    143         enc.method(method)
    144         enc.typ(ABIContractType.ADDRESS)
    145         enc.address(address)
    146         data = enc.get()
    147         tx = self.template(sender_address, contract_address, use_nonce=True)
    148         tx = self.set_code(tx, data)
    149         tx = self.finalize(tx, tx_format)        
    150         return tx
    151 
    152 
    153     def add_writer(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
    154         return self.__single_address_method('addWriter', contract_address, sender_address, address, tx_format)
    155 
    156 
    157     def delete_writer(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
    158         return self.__single_address_method('deleteWriter', contract_address, sender_address, address, tx_format)
    159 
    160 
    161     def create_request(self, contract_address, sender_address, sender, recipient, token, value, tx_format=TxFormat.JSONRPC):
    162         enc = ABIContractEncoder()
    163         enc.method('createRequest')
    164         enc.typ(ABIContractType.ADDRESS)
    165         enc.typ(ABIContractType.ADDRESS)
    166         enc.typ(ABIContractType.ADDRESS)
    167         enc.typ(ABIContractType.UINT256)
    168         enc.address(sender)
    169         enc.address(recipient)
    170         enc.address(token)
    171         enc.uint256(value)
    172         data = enc.get()
    173         tx = self.template(sender_address, contract_address, use_nonce=True)
    174         tx = self.set_code(tx, data)
    175         tx = self.finalize(tx, tx_format)        
    176         return tx
    177 
    178 
    179 
    180     def set_thresholds(self, contract_address, sender_address, quorum_threshold, veto_threshold, tx_format=TxFormat.JSONRPC):
    181         enc = ABIContractEncoder()
    182         enc.method('setThresholds')
    183         enc.typ(ABIContractType.UINT32)
    184         enc.typ(ABIContractType.UINT32)
    185         enc.uintn(quorum_threshold, 32)
    186         enc.uintn(veto_threshold, 32)
    187         data = enc.get()
    188         tx = self.template(sender_address, contract_address, use_nonce=True)
    189         tx = self.set_code(tx, data)
    190         tx = self.finalize(tx, tx_format)        
    191         return tx
    192 
    193 
    194     def requests(self, contract_address, idx, sender_address=ZERO_ADDRESS, id_generator=None):
    195         j = JSONRPCRequest(id_generator)
    196         o = j.template()
    197         o['method'] = 'eth_call'
    198         enc = ABIContractEncoder()
    199         enc.method('requests')
    200         enc.typ(ABIContractType.UINT32)
    201         enc.uintn(idx, 32)
    202         data = add_0x(enc.get())
    203         tx = self.template(sender_address, contract_address)
    204         tx = self.set_code(tx, data)
    205         o['params'].append(self.normalize(tx))
    206         o['params'].append('latest')
    207         o = j.finalize(o)
    208         return o
    209 
    210 
    211     def yay(self, contract_address, sender_address, serial, tx_format=TxFormat.JSONRPC):
    212         enc = ABIContractEncoder()
    213         enc.method('yay')
    214         enc.typ(ABIContractType.UINT32)
    215         enc.uintn(serial, 32)
    216         data = enc.get()
    217         tx = self.template(sender_address, contract_address, use_nonce=True)
    218         tx = self.set_code(tx, data)
    219         tx = self.finalize(tx, tx_format)        
    220         return tx
    221 
    222 
    223     def nay(self, contract_address, sender_address, serial, tx_format=TxFormat.JSONRPC):
    224         enc = ABIContractEncoder()
    225         enc.method('nay')
    226         enc.typ(ABIContractType.UINT32)
    227         enc.uintn(serial, 32)
    228         data = enc.get()
    229         tx = self.template(sender_address, contract_address, use_nonce=True)
    230         tx = self.set_code(tx, data)
    231         tx = self.finalize(tx, tx_format)        
    232         return tx
    233 
    234 
    235     def constructor(self, sender_address):
    236         code = TransferAuthorization.bytecode()
    237         tx = self.template(sender_address, None, use_nonce=True)
    238         tx = self.set_code(tx, code)
    239         return self.build(tx)
    240 
    241 
    242     def writers(self, contract_address, signer_address, sender_address=ZERO_ADDRESS, id_generator=None):
    243         j = JSONRPCRequest(id_generator)
    244         o = j.template()
    245         o['method'] = 'eth_call'
    246         enc = ABIContractEncoder()
    247         enc.method('signers')
    248         enc.typ(ABIContractType.ADDRESS)
    249         enc.address(signer_address)
    250         data = add_0x(enc.get())
    251         tx = self.template(sender_address, contract_address)
    252         tx = self.set_code(tx, data)
    253         o['params'].append(self.normalize(tx))
    254         o['params'].append('latest')
    255         o = j.finalize(o)
    256         return o
    257 
    258 
    259     def check_result(self, contract_address, sender_address, serial, id_generator=None, tx_format=TxFormat.JSONRPC):
    260         enc = ABIContractEncoder()
    261         enc.method('checkResult')
    262         enc.typ(ABIContractType.UINT32)
    263         enc.uintn(serial, 32)
    264         data = enc.get()
    265         tx = self.template(sender_address, contract_address, use_nonce=True)
    266         tx = self.set_code(tx, data)
    267         tx = self.finalize(tx, tx_format)        
    268         return tx
    269 
    270 
    271     def execute_request(self, contract_address, sender_address, serial, id_generator=None, tx_format=TxFormat.JSONRPC):
    272         enc = ABIContractEncoder()
    273         enc.method('executeRequest')
    274         enc.typ(ABIContractType.UINT32)
    275         enc.uintn(serial, 32)
    276         data = enc.get()
    277         tx = self.template(sender_address, contract_address, use_nonce=True)
    278         tx = self.set_code(tx, data)
    279         tx = self.finalize(tx, tx_format)        
    280         return tx
    281 
    282 
    283     def is_writer(self, contract_address, signer_address, sender_address=ZERO_ADDRESS, id_generator=None):
    284         return self.writers(contract_address, signer_address, sender_address)
    285 
    286 
    287     @classmethod
    288     def parse_signers(self, v):
    289         return abi_decode_single(ABIContractType.BOOLEAN, v)
    290 
    291 
    292     @classmethod
    293     def parse_count(self, v):
    294         return abi_decode_single(ABIContractType.UINT32, v)
    295 
    296 
    297     @classmethod
    298     def parse_create_request_request(self, v):
    299         v = strip_0x(v)
    300         cursor = 0
    301         enc = ABIContractEncoder()
    302         enc.method('createRequest')
    303         enc.typ(ABIContractType.ADDRESS)
    304         enc.typ(ABIContractType.ADDRESS)
    305         enc.typ(ABIContractType.ADDRESS)
    306         enc.typ(ABIContractType.UINT256)
    307         r = enc.get()
    308         l = len(r)
    309         m = v[:l]
    310         if m != r:
    311             logg.error('method mismatch, expected {}, got {}'.format(r, m))
    312             raise RequestMismatchException(v)
    313         cursor += l
    314 
    315         dec = ABIContractDecoder()
    316         dec.typ(ABIContractType.ADDRESS)
    317         dec.typ(ABIContractType.ADDRESS)
    318         dec.typ(ABIContractType.ADDRESS)
    319         dec.typ(ABIContractType.UINT256)
    320         dec.val(v[cursor:cursor+64])
    321         cursor += 64
    322         dec.val(v[cursor:cursor+64])
    323         cursor += 64
    324         dec.val(v[cursor:cursor+64])
    325         cursor += 64
    326         dec.val(v[cursor:cursor+64])
    327         r = dec.decode()
    328         return r 
    329 
    330 
    331     @classmethod
    332     def parse_request(self, v):
    333         cursor = 0
    334         v = strip_0x(v)
    335         d = ABIContractDecoder()
    336         d.typ(ABIContractType.UINT256)
    337         d.typ(ABIContractType.ADDRESS)
    338         d.typ(ABIContractType.ADDRESS)
    339         d.typ(ABIContractType.ADDRESS)
    340         d.typ(ABIContractType.UINT32)
    341         d.typ(ABIContractType.UINT32)
    342         d.typ(ABIContractType.UINT32)
    343         d.typ(ABIContractType.UINT32)
    344         d.val(v[cursor:cursor+64])
    345         cursor += 64
    346         d.val(v[cursor:cursor+64])
    347         cursor += 64 
    348         d.val(v[cursor:cursor+64])
    349         cursor += 64 
    350         d.val(v[cursor:cursor+64])
    351         cursor += 64 
    352         d.val(v[cursor:cursor+64])
    353         cursor += 64
    354         d.val(v[cursor:cursor+64])
    355         cursor += 64
    356         d.val(v[cursor:cursor+64])
    357         cursor += 64
    358         d.val(v[cursor:cursor+64])
    359         cursor += 64
    360         r = d.decode()
    361         return Request.create(r[1], r[2], r[0], r[3], r[4], r[5], r[6], r[7])