eth-address-index

signed metadata declarations for ethereum addresses
Log | Files | Refs

interface.py (5325B)

      1 # standard imports
      2 import logging
      3 import json
      4 import os
      5 
      6 # external imports
      7 from hexathon import (
      8         strip_0x,
      9         add_0x,
     10         )
     11 from chainlib.eth.tx import (
     12         TxFormat,
     13         TxFactory,
     14         )
     15 from chainlib.eth.contract import (
     16         ABIContractEncoder,
     17         ABIContractType,
     18         abi_decode_single,
     19         )
     20 from chainlib.jsonrpc import JSONRPCRequest
     21 from chainlib.eth.constant import ZERO_ADDRESS
     22 from chainlib.block import BlockSpec
     23 from chainlib.eth.jsonrpc import to_blockheight_param
     24 
     25 logg = logging.getLogger(__name__)
     26 
     27 
     28 class Declarator(TxFactory):
     29 
     30     def add_declaration(self, contract_address, sender_address, subject_address, proof, topic=None, tx_format=TxFormat.JSONRPC):
     31         enc = ABIContractEncoder()
     32         enc.method('addDeclaration')
     33         enc.typ(ABIContractType.ADDRESS)
     34         enc.typ(ABIContractType.BYTES32)
     35         if topic != None:
     36             enc.typ(ABIContractType.BYTES32)
     37         enc.address(subject_address)
     38         enc.bytes32(proof)
     39         if topic != None:
     40             enc.bytes32(topic)
     41         data = enc.get()
     42         tx = self.template(sender_address, contract_address, use_nonce=True)
     43         tx = self.set_code(tx, data)
     44         tx = self.finalize(tx, tx_format)
     45         return tx
     46 
     47 
     48     def declarator_count(self, contract_address, subject_address, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST):
     49         j = JSONRPCRequest(id_generator)
     50         o = j.template()
     51         o['method'] = 'eth_call'
     52         enc = ABIContractEncoder()
     53         enc.method('declaratorCount')
     54         enc.typ(ABIContractType.ADDRESS)
     55         enc.address(subject_address)
     56         data = add_0x(enc.get())
     57         tx = self.template(sender_address, contract_address)
     58         tx = self.set_code(tx, data)
     59         o['params'].append(self.normalize(tx))
     60         height = to_blockheight_param(height)
     61         o['params'].append(height)
     62         o = j.finalize(o)
     63         return o
     64 
     65 
     66     def declaration(self, contract_address, declarator_address, subject_address, topic=None, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST):
     67         j = JSONRPCRequest(id_generator)
     68         o = j.template()
     69         o['method'] = 'eth_call'
     70         enc = ABIContractEncoder()
     71         enc.method('declaration')
     72         enc.typ(ABIContractType.ADDRESS)
     73         enc.typ(ABIContractType.ADDRESS)
     74         if topic != None:
     75             enc.typ(ABIContractType.BYTES32)
     76         enc.address(declarator_address)
     77         enc.address(subject_address)
     78         if topic != None:
     79             enc.bytes32(topic)
     80         data = add_0x(enc.get())
     81         tx = self.template(sender_address, contract_address)
     82         tx = self.set_code(tx, data)
     83         o['params'].append(self.normalize(tx))
     84         height = to_blockheight_param(height)
     85         o['params'].append(height)
     86         o = j.finalize(o)
     87         return o
     88 
     89 
     90     def declaration_address_at(self, contract_address, declarator_address, idx, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST):
     91         j = JSONRPCRequest(id_generator)
     92         o = j.template()
     93         o['method'] = 'eth_call'
     94         enc = ABIContractEncoder()
     95         enc.method('declarationAddressAt')
     96         enc.typ(ABIContractType.ADDRESS)
     97         enc.typ(ABIContractType.UINT256)
     98         enc.address(declarator_address)
     99         enc.uint256(idx)
    100         data = add_0x(enc.get())
    101         tx = self.template(sender_address, contract_address)
    102         tx = self.set_code(tx, data)
    103         o['params'].append(self.normalize(tx))
    104         height = to_blockheight_param(height)
    105         o['params'].append(height)
    106         o = j.finalize(o)
    107         return o
    108 
    109 
    110     def declarator_address_at(self, contract_address, subject_address, idx, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST):
    111         j = JSONRPCRequest(id_generator)
    112         o = j.template()
    113         o['method'] = 'eth_call'
    114         enc = ABIContractEncoder()
    115         enc.method('declaratorAddressAt')
    116         enc.typ(ABIContractType.ADDRESS)
    117         enc.typ(ABIContractType.UINT256)
    118         enc.address(subject_address)
    119         enc.uint256(idx)
    120         data = add_0x(enc.get())
    121         tx = self.template(sender_address, contract_address)
    122         tx = self.set_code(tx, data)
    123         o['params'].append(self.normalize(tx))
    124         height = to_blockheight_param(height)
    125         o['params'].append(height)
    126         o = j.finalize(o)
    127         return o
    128 
    129 
    130     @classmethod
    131     def parse_declarator_count(self, v):
    132         return abi_decode_single(ABIContractType.UINT256, v)
    133 
    134     
    135     @classmethod
    136     def parse_declaration(self, v):
    137         cursor = 0
    138         r = []
    139         try:
    140             v = strip_0x(v)
    141         except ValueError:
    142             return r
    143         position = int.from_bytes(bytes.fromhex(v[cursor:cursor+64]), 'big')
    144         cursor += (position * 2)
    145         length = int.from_bytes(bytes.fromhex(v[cursor:cursor+64]), 'big')
    146         cursor += 64
    147         for i in range(length):
    148             r.append(v[cursor:cursor+64])
    149             cursor += 64
    150         return r 
    151 
    152 
    153     @classmethod
    154     def parse_declaration_address_at(self, v):
    155         return abi_decode_single(ABIContractType.ADDRESS, v)
    156 
    157 
    158     @classmethod
    159     def parse_declarator_address_at(self, v):
    160         return abi_decode_single(ABIContractType.ADDRESS, v)