eth-offline

EVM token minter frontend for offline issuance using ERC712 structured signatures.
Log | Files | Refs

test_basic.py (13984B)

      1 # standard imports
      2 import logging
      3 import os
      4 import unittest
      5 
      6 # external imports
      7 from chainlib.eth.tx import TxFactory
      8 from chainlib.eth.tx import TxFormat
      9 from chainlib.eth.unittest.ethtester import EthTesterCase
     10 from chainlib.connection import RPCConnection
     11 from chainlib.eth.nonce import RPCNonceOracle
     12 from chainlib.eth.address import to_checksum_address
     13 from chainlib.eth.tx import receipt
     14 from chainlib.eth.gas import Gas
     15 from chainlib.eth.gas import OverrideGasOracle
     16 from chainlib.eth.contract import ABIContractEncoder
     17 from chainlib.eth.contract import ABIContractType
     18 from chainlib.jsonrpc import JSONRPCRequest
     19 from chainlib.error import JSONRPCException
     20 from hexathon import add_0x
     21 from hexathon import strip_0x
     22 from hexathon import same as same_hex
     23 from funga.eth.message import to_validator_message
     24 from eth_erc712 import EIP712Domain
     25 from eth_erc712 import EIP712DomainEncoder
     26 
     27 script_dir = os.path.realpath(os.path.dirname(__file__))
     28 data_dir = os.path.join(script_dir, '..', 'eth_offline', 'data')
     29 
     30 logging.basicConfig(level=logging.DEBUG)
     31 logg = logging.getLogger(__name__)
     32 
     33 hash_of_foo = '2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae'
     34 hash_of_bar = 'fcde2b2edba56bf408601fb721fe9b5c338d10ee429ea04fae5511b68fbf8fb9'
     35 
     36 
     37 class TestOfflineEth(EthTesterCase):
     38 
     39     def setUp(self):
     40         super(TestOfflineEth, self).setUp()
     41         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     42         gas_oracle = OverrideGasOracle(limit=3000000)
     43 
     44         fp = os.path.join(data_dir, 'Offline.bin')
     45         f = open(fp, 'r')
     46         bytecode = f.read()
     47         f.close()
     48 
     49         c = Gas(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     50         self.conn = RPCConnection.connect(self.chain_spec, 'default')
     51 
     52         data = self.chain_spec.chain_id().to_bytes(32, byteorder='big').hex()
     53         (tx_hash, o) = c.create(self.accounts[0], None, 0, data=bytecode + data)
     54         r = self.conn.do(o)
     55         o = receipt(r)
     56         r = self.conn.do(o)
     57         self.assertEqual(r['status'], 1)
     58         self.address = to_checksum_address(r['contract_address'])
     59         self.contract_block_hash = r['block_hash']
     60         logg.debug('smart contract published with hash {} address {}'.format(r, self.address))
     61 
     62 
     63     def test_validator(self):
     64         n = os.urandom(12)
     65         c = TxFactory(self.chain_spec)
     66         j = JSONRPCRequest()
     67         o = j.template()
     68         o['method'] = 'eth_call'
     69         enc = ABIContractEncoder()
     70         enc.method('isOfflineValid')
     71         enc.typ(ABIContractType.ADDRESS)
     72         enc.typ(ABIContractType.BYTES)
     73         enc.address(strip_0x(self.accounts[0]))
     74         enc.bytes(n.hex() + strip_0x(self.accounts[1]) + '666f6f')
     75         data = add_0x(enc.get())
     76         tx = c.template(self.accounts[0], self.address)
     77         tx = c.set_code(tx, data)
     78         o['params'].append(c.normalize(tx))
     79         o['params'].append('latest')
     80         o = j.finalize(o)
     81         r = self.rpc.do(o)
     82         self.assertEqual(int(r, 16), 0)
     83 
     84         v = os.urandom(64)
     85         c = TxFactory(self.chain_spec)
     86         j = JSONRPCRequest()
     87         o = j.template()
     88         o['method'] = 'eth_call'
     89         enc = ABIContractEncoder()
     90         enc.method('isOfflineValid')
     91         enc.typ(ABIContractType.ADDRESS)
     92         enc.typ(ABIContractType.BYTES)
     93         enc.address(strip_0x(self.accounts[2]))
     94         enc.bytes(n.hex() + strip_0x(self.accounts[1]) + v.hex())
     95         data = add_0x(enc.get())
     96         tx = c.template(self.accounts[0], self.address)
     97         tx = c.set_code(tx, data)
     98         o['params'].append(c.normalize(tx))
     99         o['params'].append('latest')
    100         o = j.finalize(o)
    101         r = self.rpc.do(o)
    102         self.assertEqual(int(r, 16), 0)
    103 
    104         n = b'\x00' * 12
    105         c = TxFactory(self.chain_spec)
    106         j = JSONRPCRequest()
    107         o = j.template()
    108         o['method'] = 'eth_call'
    109         enc = ABIContractEncoder()
    110         enc.method('isOfflineValid')
    111         enc.typ(ABIContractType.ADDRESS)
    112         enc.typ(ABIContractType.BYTES)
    113         enc.address(strip_0x(self.accounts[0]))
    114         enc.bytes(n.hex() + strip_0x(self.accounts[1]) + v.hex())
    115         data = add_0x(enc.get())
    116         tx = c.template(self.accounts[0], self.address)
    117         tx = c.set_code(tx, data)
    118         o['params'].append(c.normalize(tx))
    119         o['params'].append('latest')
    120         o = j.finalize(o)
    121         r = self.rpc.do(o)
    122         self.assertEqual(int(r, 16), 0)
    123 
    124         n = os.urandom(12)
    125         c = TxFactory(self.chain_spec)
    126         j = JSONRPCRequest()
    127         o = j.template()
    128         o['method'] = 'eth_call'
    129         enc = ABIContractEncoder()
    130         enc.method('isOfflineValid')
    131         enc.typ(ABIContractType.ADDRESS)
    132         enc.typ(ABIContractType.BYTES)
    133         enc.address(strip_0x(self.accounts[0]))
    134         enc.bytes(n.hex() + strip_0x(self.accounts[1]) + v.hex())
    135         data = add_0x(enc.get())
    136         tx = c.template(self.accounts[0], self.address)
    137         tx = c.set_code(tx, data)
    138         o['params'].append(c.normalize(tx))
    139         o['params'].append('latest')
    140         o = j.finalize(o)
    141         r = self.rpc.do(o)
    142         r = strip_0x(r)
    143         self.assertEqual(int(r, 16), 1)
    144 
    145 
    146     def test_ok_verify(self):
    147         beneficiary_bin = bytes.fromhex(strip_0x(self.accounts[2]))
    148         msg_nonce = os.urandom(12)
    149         msg_bin = os.urandom(64)
    150         msg_data = msg_nonce + beneficiary_bin + msg_bin
    151 
    152         sig = self.signer.sign_validator_message(self.accounts[0], self.address, msg_data)
    153         sig = sig[:64] + (sig[64] + 27).to_bytes(1, byteorder='big')
    154         logg.debug('message is {} ({})  signed by {}'.format(msg_data.hex(), len(msg_data), self.accounts[0]))
    155 
    156         c = TxFactory(self.chain_spec)
    157         j = JSONRPCRequest()
    158         o = j.template()
    159         o['method'] = 'eth_call'
    160         enc = ABIContractEncoder()
    161         enc.method('verifyOfflineRequest')
    162         enc.typ(ABIContractType.BYTES)
    163         enc.typ(ABIContractType.BYTES)
    164         enc.bytes(msg_data.hex())
    165         enc.bytes(sig.hex())
    166         data = add_0x(enc.get())
    167         tx = c.template(self.accounts[0], self.address)
    168         tx = c.set_code(tx, data)
    169         o['params'].append(c.normalize(tx))
    170         o['params'].append('latest')
    171         o = j.finalize(o)
    172 
    173         
    174         r = self.rpc.do(o)
    175         r = strip_0x(r)
    176 
    177         self.assertEqual(int(r, 16), 1)
    178  
    179 
    180     def test_verify_fail_owner(self):
    181         beneficiary_bin = bytes.fromhex(strip_0x(self.accounts[2]))
    182         msg_nonce = os.urandom(12)
    183         msg_bin = os.urandom(64)
    184         msg_data = msg_nonce + beneficiary_bin + msg_bin
    185 
    186         sig = self.signer.sign_validator_message(self.accounts[1], self.address, msg_data)
    187         sig = sig[:64] + (sig[64] + 27).to_bytes(1, byteorder='big')
    188         logg.debug('message is {} signed by {}'.format(msg_data.hex(), self.accounts[1]))
    189 
    190         c = TxFactory(self.chain_spec)
    191         j = JSONRPCRequest()
    192         o = j.template()
    193         o['method'] = 'eth_call'
    194         enc = ABIContractEncoder()
    195         enc.method('verifyOfflineRequest')
    196         enc.typ(ABIContractType.BYTES)
    197         enc.typ(ABIContractType.BYTES)
    198         enc.bytes(msg_data.hex())
    199         enc.bytes(sig.hex())
    200         data = add_0x(enc.get())
    201         tx = c.template(self.accounts[0], self.address)
    202         tx = c.set_code(tx, data)
    203         o['params'].append(c.normalize(tx))
    204         o['params'].append('latest')
    205         o = j.finalize(o)
    206         r = self.rpc.do(o)
    207         self.assertEqual(int(r, 16), 0)
    208 
    209 
    210     def test_execute(self):
    211         beneficiary_bin = bytes.fromhex(strip_0x(self.accounts[2]))
    212         msg_nonce = os.urandom(12)
    213         msg_bin = os.urandom(64)
    214         msg_data = msg_nonce + beneficiary_bin + msg_bin
    215 
    216         sig = self.signer.sign_validator_message(self.accounts[0], self.address, msg_data)
    217         sig = sig[:64] + (sig[64] + 27).to_bytes(1, byteorder='big')
    218         logg.debug('message is {} signed by {}'.format(msg_data.hex(), self.accounts[1]))
    219 
    220         c = TxFactory(self.chain_spec)
    221         j = JSONRPCRequest()
    222         o = j.template()
    223         o['method'] = 'eth_call'
    224         enc = ABIContractEncoder()
    225         enc.method('executedTimes')
    226         data = add_0x(enc.get())
    227         tx = c.template(self.accounts[0], self.address)
    228         tx = c.set_code(tx, data)
    229         o['params'].append(c.normalize(tx))
    230         o['params'].append('latest')
    231         o = j.finalize(o)
    232         r = self.rpc.do(o)
    233         r = strip_0x(r)
    234         self.assertEqual(int(r, 16), 0)
    235 
    236         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
    237         c = TxFactory(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
    238         enc = ABIContractEncoder()
    239         enc.method('executeOfflineRequest')
    240         enc.typ(ABIContractType.BYTES)
    241         enc.typ(ABIContractType.BYTES)
    242         enc.typ(ABIContractType.UINT8)
    243         enc.bytes(msg_data.hex())
    244         enc.bytes(sig.hex())
    245         enc.uintn(0, 8)
    246         data = enc.get()
    247         tx = c.template(self.accounts[0], self.address, use_nonce=True)
    248         tx = c.set_code(tx, data)
    249         (tx_hash, o) = c.finalize(tx, TxFormat.JSONRPC)
    250         self.rpc.do(o)
    251         o = receipt(tx_hash)
    252         r = self.rpc.do(o)
    253         self.assertEqual(r['status'], 1)
    254 
    255         c = TxFactory(self.chain_spec)
    256         j = JSONRPCRequest()
    257         o = j.template()
    258         o['method'] = 'eth_call'
    259         enc = ABIContractEncoder()
    260         enc.method('executedTimes')
    261         data = add_0x(enc.get())
    262         tx = c.template(self.accounts[0], self.address)
    263         tx = c.set_code(tx, data)
    264         o['params'].append(c.normalize(tx))
    265         o['params'].append('latest')
    266         o = j.finalize(o)
    267         r = self.rpc.do(o)
    268         r = strip_0x(r)
    269         self.assertEqual(int(r, 16), 1)
    270 
    271 
    272     # TODO: check if uint96 etc works with packed data
    273     def test_from_struct(self):
    274         instruction_nonce = os.urandom(16)
    275         enc_instruction = ABIContractEncoder()
    276         # bytes12 does not work with this pyevm in struct encoding 
    277         #enc_instruction.typ(ABIContractType.BYTES32)
    278         enc_instruction.typ(ABIContractType.BYTES32)
    279         enc_instruction.typ(ABIContractType.UINT256)
    280         enc_instruction.typ(ABIContractType.ADDRESS)
    281         #enc_instruction.typ_literal('uint96')
    282         #enc_instruction.bytesn(instruction_nonce.hex(), 12)
    283         #enc_instruction.bytes32(hash_of_bar)
    284         enc_instruction.bytes32(hash_of_foo)
    285         enc_instruction.uint256(42)
    286         enc_instruction.address(self.accounts[2])
    287         #enc_instruction.uintn(int(instruction_nonce.hex(), 16), 96)
    288 
    289         c = TxFactory(self.chain_spec)
    290         j = JSONRPCRequest()
    291         o = j.template()
    292         o['method'] = 'eth_call'
    293         enc = ABIContractEncoder()
    294         enc.method('toBytes')
    295         enc.typ(enc_instruction)
    296         enc.tuple(enc_instruction)
    297         data = strip_0x(enc.get())
    298         data = data[:8] + data[8+64:]
    299         data = add_0x(data)
    300         tx = c.template(self.accounts[0], self.address)
    301         tx = c.set_code(tx, data)
    302         o['params'].append(c.normalize(tx))
    303         o['params'].append('latest')
    304         o = j.finalize(o)
    305         r = self.rpc.do(o)
    306         r = strip_0x(r)
    307         print(r)
    308 
    309 
    310     def test_execute_erc712(self):
    311         c = TxFactory(self.chain_spec)
    312         j = JSONRPCRequest()
    313         o = j.template()
    314         o['method'] = 'eth_call'
    315         enc = ABIContractEncoder()
    316         enc.method('eip712DomainSeparator')
    317         data = add_0x(enc.get())
    318         tx = c.template(self.accounts[0], self.address)
    319         tx = c.set_code(tx, data)
    320         o['params'].append(c.normalize(tx))
    321         o['params'].append('latest')
    322         o = j.finalize(o)
    323         r = self.rpc.do(o)
    324         r = strip_0x(r)
    325 
    326         domain = EIP712Domain(
    327                 name='Offline signing test',
    328                 version='0',
    329                 chain_id=self.chain_spec.chain_id(),
    330                 verifying_contract=self.address,
    331                 #salt=self.contract_block_hash,
    332                 salt='0x00',
    333                 )
    334 
    335         self.assertEqual(r, domain.get_contents().hex())
    336 
    337         c = TxFactory(self.chain_spec)
    338         j = JSONRPCRequest()
    339         o = j.template()
    340         o['method'] = 'eth_call'
    341         enc = ABIContractEncoder()
    342         enc.method('eip712TypeHash')
    343         data = add_0x(enc.get())
    344         tx = c.template(self.accounts[0], self.address)
    345         tx = c.set_code(tx, data)
    346         o['params'].append(c.normalize(tx))
    347         o['params'].append('latest')
    348         o = j.finalize(o)
    349         r = self.rpc.do(o)
    350         r = strip_0x(r)
    351 
    352         subdomain = os.urandom(32)
    353         enc = EIP712DomainEncoder('Instruction', domain)
    354         enc.add('domain', ABIContractType.BYTES32, subdomain)
    355         enc.add('value', ABIContractType.UINT256, 42)
    356         enc.add('beneficiary', ABIContractType.ADDRESS, self.accounts[2])
    357         
    358         self.assertEqual(r, enc.get_type_hash().hex())
    359 
    360         msg_data = enc.get_typed_data()
    361         sig = self.signer.sign_typed_message(self.accounts[0], enc.get_domain(), enc.get_hash())
    362         sig = sig[:64] + (sig[64] + 27).to_bytes(1, byteorder='big')
    363         logg.debug('message is:\n{}\nsigned by {}'.format(enc, self.accounts[0]))
    364 
    365         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
    366         c = TxFactory(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
    367         enc = ABIContractEncoder()
    368         enc.method('executeOfflineRequest')
    369         enc.typ(ABIContractType.BYTES)
    370         enc.typ(ABIContractType.BYTES)
    371         enc.typ(ABIContractType.UINT8)
    372         enc.bytes(msg_data.hex())
    373         enc.bytes(sig.hex())
    374         enc.uintn(1, 8)
    375         data = enc.get()
    376         tx = c.template(self.accounts[0], self.address, use_nonce=True)
    377         tx = c.set_code(tx, data)
    378         (tx_hash, o) = c.finalize(tx, TxFormat.JSONRPC)
    379         self.rpc.do(o)
    380         o = receipt(tx_hash)
    381         r = self.rpc.do(o)
    382         self.assertEqual(r['status'], 1)
    383    
    384 
    385 if __name__ == '__main__':
    386     unittest.main()