accounts-index

Accounts index evm contract tooling with permissioned writes
Log | Files | Refs

interface.py (5377B)

      1 # standard imports
      2 import logging
      3 import json
      4 import os
      5 
      6 # external imports
      7 from chainlib.eth.tx import (
      8     TxFactory,
      9     TxFormat,
     10 )
     11 from chainlib.eth.contract import (
     12     ABIContractEncoder,
     13     ABIContractDecoder,
     14     ABIContractType,
     15     abi_decode_single,
     16 )
     17 from chainlib.eth.constant import ZERO_ADDRESS
     18 from chainlib.jsonrpc import JSONRPCRequest
     19 from chainlib.eth.error import RequestMismatchException
     20 from hexathon import (
     21     add_0x,
     22     strip_0x,
     23 )
     24 
     25 logg = logging.getLogger()
     26 
     27 
     28 class AccountsIndex(TxFactory):
     29 
     30     def __single_address_method(self, method, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
     31         enc = ABIContractEncoder()
     32         enc.method(method)
     33         enc.typ(ABIContractType.ADDRESS)
     34         enc.address(address)
     35         data = enc.get()
     36         tx = self.template(sender_address, contract_address, use_nonce=True)
     37         tx = self.set_code(tx, data)
     38         tx = self.finalize(tx, tx_format)
     39         return tx
     40 
     41     def add(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
     42         return self.__single_address_method('add', contract_address, sender_address, address, tx_format)
     43 
     44     def remove(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
     45         return self.__single_address_method('remove', contract_address, sender_address, address, tx_format)
     46 
     47     def activate(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
     48         return self.__single_address_method('activate', contract_address, sender_address, address, tx_format)
     49 
     50     def deactivate(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC):
     51         return self.__single_address_method('deactivate', contract_address, sender_address, address, tx_format)
     52 
     53 
     54     def time(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None):
     55         j = JSONRPCRequest(id_generator)
     56         o = j.template()
     57         o['method'] = 'eth_call'
     58         enc = ABIContractEncoder()
     59         enc.method('time')
     60         enc.typ(ABIContractType.ADDRESS)
     61         enc.address(address)
     62         data = add_0x(enc.get())
     63         tx = self.template(sender_address, contract_address)
     64         tx = self.set_code(tx, data)
     65         o['params'].append(self.normalize(tx))
     66         return o
     67 
     68 
     69 
     70     def have(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None):
     71         j = JSONRPCRequest(id_generator)
     72         o = j.template()
     73         o['method'] = 'eth_call'
     74         enc = ABIContractEncoder()
     75         enc.method('have')
     76         enc.typ(ABIContractType.ADDRESS)
     77         enc.address(address)
     78         data = add_0x(enc.get())
     79         tx = self.template(sender_address, contract_address)
     80         tx = self.set_code(tx, data)
     81         o['params'].append(self.normalize(tx))
     82         return o
     83 
     84     def is_active(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None):
     85         j = JSONRPCRequest(id_generator)
     86         o = j.template()
     87         o['method'] = 'eth_call'
     88         enc = ABIContractEncoder()
     89         enc.method('isActive')
     90         enc.typ(ABIContractType.ADDRESS)
     91         enc.address(address)
     92         data = add_0x(enc.get())
     93         tx = self.template(sender_address, contract_address)
     94         tx = self.set_code(tx, data)
     95         o['params'].append(self.normalize(tx))
     96         return o
     97 
     98 
     99     def entry_count(self, contract_address, sender_address=ZERO_ADDRESS, id_generator=None):
    100         j = JSONRPCRequest(id_generator)
    101         o = j.template()
    102         o['method'] = 'eth_call'
    103         enc = ABIContractEncoder()
    104         enc.method('entryCount')
    105         data = add_0x(enc.get())
    106         tx = self.template(sender_address, contract_address)
    107         tx = self.set_code(tx, data)
    108         o['params'].append(self.normalize(tx))
    109         return o
    110 
    111     def count(self, contract_address, sender_address=ZERO_ADDRESS, id_generator=None):
    112         return self.entry_count(contract_address, sender_address=sender_address, id_generator=id_generator)
    113     
    114     def entry(self, contract_address, idx, sender_address=ZERO_ADDRESS, id_generator=None):
    115         j = JSONRPCRequest(id_generator)
    116         o = j.template()
    117         o['method'] = 'eth_call'
    118         enc = ABIContractEncoder()
    119         enc.method('entry')
    120         enc.typ(ABIContractType.UINT256)
    121         enc.uint256(idx)
    122         data = add_0x(enc.get())
    123         tx = self.template(sender_address, contract_address)
    124         tx = self.set_code(tx, data)
    125         o['params'].append(self.normalize(tx))
    126         return o
    127 
    128     @classmethod
    129     def parse_account(self, v):
    130         return abi_decode_single(ABIContractType.ADDRESS, v)
    131 
    132     @classmethod
    133     def parse_entry_count(self, v):
    134         return abi_decode_single(ABIContractType.UINT256, v)
    135 
    136     @classmethod
    137     def parse_have(self, v):
    138         return abi_decode_single(ABIContractType.BOOLEAN, v)
    139 
    140     @classmethod
    141     def parse_add_request(self, v):
    142         v = strip_0x(v)
    143         cursor = 0
    144         enc = ABIContractEncoder()
    145         enc.method('add')
    146         enc.typ(ABIContractType.ADDRESS)
    147         r = enc.get()
    148         l = len(r)
    149         m = v[:l]
    150         if m != r:
    151             raise RequestMismatchException(v)
    152         cursor += l
    153 
    154         dec = ABIContractDecoder()
    155         dec.typ(ABIContractType.ADDRESS)
    156         dec.val(v[cursor:cursor + 64])
    157         r = dec.decode()
    158         return r