Initial commit

tests/test_cli.py

@@ -0,0 +1,88 @@
+# standard imports
+import unittest
+import logging
+import os
+
+# external imports
+from hexathon import strip_0x
+
+# local imports
+from funga.eth.signer import EIP155Signer
+from funga.eth.keystore.dict import DictKeystore
+from funga.eth.cli.handle import SignRequestHandler
+from funga.eth.transaction import EIP155Transaction
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+script_dir = os.path.dirname(os.path.realpath(__file__))
+data_dir = os.path.join(script_dir, 'testdata')
+
+class TestCli(unittest.TestCase):
+
+    def setUp(self):
+        #pk = bytes.fromhex('5087503f0a9cc35b38665955eb830c63f778453dd11b8fa5bd04bc41fd2cc6d6')
+        #pk_getter = pkGetter(pk)
+        self.keystore = DictKeystore()
+        SignRequestHandler.keystore = self.keystore
+        self.signer = EIP155Signer(self.keystore)
+        SignRequestHandler.signer = self.signer
+        self.handler = SignRequestHandler()
+
+
+    def test_new_account(self):
+        q = {
+                'id': 0,
+                'method': 'personal_newAccount',
+                'params': [''],
+                }
+        (rpc_id, result) = self.handler.process_input(q)
+        self.assertTrue(self.keystore.get(result))
+
+
+    def test_sign_tx(self):
+        keystore_file = os.path.join(data_dir, 'UTC--2021-01-08T18-37-01.187235289Z--00a329c0648769a73afac7f9381e08fb43dbea72')
+        sender = self.keystore.import_keystore_file(keystore_file)
+        tx_hexs = {
+            'nonce': '0x',
+            'from': sender,
+            'gasPrice': "0x04a817c800",
+            'gas': "0x5208",
+            'to': '0x3535353535353535353535353535353535353535',
+            'value': "0x03e8",
+            'data': "0xdeadbeef",
+            'chainId': 8995,
+        }
+        tx = EIP155Transaction(tx_hexs, 42, 8995)
+        tx_s = tx.serialize()
+
+        # TODO: move to serialization wrapper for tests
+        tx_s['chainId'] = tx_s['v']
+        tx_s['from'] = sender
+
+        # eth_signTransaction wraps personal_signTransaction, so here we test both already
+        q = {
+                'id': 0,
+                'method': 'eth_signTransaction',
+                'params': [tx_s],
+                }
+        (rpc_id, result) = self.handler.process_input(q)
+        logg.debug('result {}'.format(result))
+
+        self.assertEqual(strip_0x(result), 'f86c2a8504a817c8008252089435353535353535353535353535353535353535358203e884deadbeef82466aa0b7c1bbf52f736ada30fe253c7484176f44d6fd097a9720dc85ae5bbc7f060e54a07afee2563b0cf6d00333df51cc62b0d13c63108b2bce54ce2ad24e26ce7b4f25')
+
+    def test_sign_msg(self):
+        keystore_file = os.path.join(data_dir, 'UTC--2021-01-08T18-37-01.187235289Z--00a329c0648769a73afac7f9381e08fb43dbea72')
+        sender = self.keystore.import_keystore_file(keystore_file)
+        q = {
+                'id': 0,
+                'method': 'eth_sign',
+                'params': [sender, '0xdeadbeef'],
+                }
+        (rpc_id, result) = self.handler.process_input(q)
+        logg.debug('result msg {}'.format(result))
+        self.assertEqual(strip_0x(result), '50320dda75190a121b7b5979de66edadafd02bdfbe4f6d49552e79c01410d2464aae35e385c0e5b61663ff7b44ef65fa0ac7ad8a57472cf405db399b9dba3e1600')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_keystore_dict.py

@@ -0,0 +1,63 @@
+#!/usr/bin/python
+
+# standard imports
+import unittest
+import logging
+import base64
+import os
+
+# external imports
+from hexathon import (
+    strip_0x,
+    add_0x,
+    )
+
+# local imports
+from funga.error import UnknownAccountError
+from funga.eth.keystore.dict import DictKeystore
+from funga.eth.signer import EIP155Signer
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+script_dir = os.path.realpath(os.path.dirname(__file__))
+
+
+class TestDict(unittest.TestCase):
+
+    address_hex = None
+    db = None
+
+    def setUp(self):
+        self.db = DictKeystore()
+
+        keystore_filepath = os.path.join(script_dir, 'testdata', 'UTC--2021-01-08T18-37-01.187235289Z--00a329c0648769a73afac7f9381e08fb43dbea72')
+
+        address_hex = self.db.import_keystore_file(keystore_filepath, '')
+        self.address_hex = add_0x(address_hex)
+
+
+    def tearDown(self): 
+        pass
+
+
+    def test_get_key(self):
+        logg.debug('getting {}'.format(strip_0x(self.address_hex)))
+        pk = self.db.get(strip_0x(self.address_hex), '')
+
+        self.assertEqual(self.address_hex.lower(), '0x00a329c0648769a73afac7f9381e08fb43dbea72')
+
+        bogus_account = os.urandom(20).hex()
+        with self.assertRaises(UnknownAccountError):
+           self.db.get(bogus_account, '')
+
+    
+    def test_sign_message(self):
+        s = EIP155Signer(self.db)
+        z = s.sign_ethereum_message(strip_0x(self.address_hex), b'foo')
+        logg.debug('zzz {}'.format(str(z)))
+
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_keystore_reference.py

@@ -0,0 +1,64 @@
+#!/usr/bin/python
+
+# standard imports
+import unittest
+import logging
+import base64
+import os
+
+# external imports
+import psycopg2
+from psycopg2 import sql
+from cryptography.fernet import Fernet, InvalidToken
+
+# local imports
+from funga.eth.keystore.sql import SQLKeystore
+from funga.error import UnknownAccountError
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+
+class TestDatabase(unittest.TestCase):
+
+    conn = None
+    cur = None
+    symkey = None
+    address_hex = None
+    db = None
+
+    def setUp(self):
+        logg.debug('setup')
+        # arbitrary value
+        symkey_hex = 'E92431CAEE69313A7BE9E443C4ABEED9BF8157E9A13553B4D5D6E7D51B5021D9'
+        self.symkey = bytes.fromhex(symkey_hex)
+        self.address_hex = '9FA61f0E52A5C51b43f0d32404625BC436bb7041'
+
+        kw = {
+                'symmetric_key': self.symkey,
+                }
+        self.db = SQLKeystore('postgres+psycopg2://postgres@localhost:5432/signer_test', **kw)
+        self.address_hex = self.db.new('foo')
+        #self.address_hex = add_0x(address_hex)
+
+
+    def tearDown(self):
+        self.db.db_session.execute('DROP INDEX ethereum_address_idx;')
+        self.db.db_session.execute('DROP TABLE ethereum;')
+        self.db.db_session.commit()
+
+
+
+    def test_get_key(self):
+        logg.debug('getting {}'.format(self.address_hex))
+        self.db.get(self.address_hex, 'foo')
+        with self.assertRaises(InvalidToken):
+           self.db.get(self.address_hex, 'bar')
+
+        bogus_account = '0x' + os.urandom(20).hex()
+        with self.assertRaises(UnknownAccountError):
+           self.db.get(bogus_account, 'bar')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_sign.py

@@ -0,0 +1,99 @@
+# standard imports
+import unittest
+import logging
+import json
+
+from rlp import encode as rlp_encode
+
+from funga.eth.signer import EIP155Signer
+from funga.eth.transaction import EIP155Transaction
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+
+tx_ints = {
+    'nonce': 0,
+    'from': "0xEB014f8c8B418Db6b45774c326A0E64C78914dC0",
+    'gasPrice': "20000000000",
+    'gas': "21000",
+    'to': '0x3535353535353535353535353535353535353535',
+    'value': "1000",
+    'data': "deadbeef",
+}
+
+tx_hexs = {
+    'nonce': '0x0',
+    'from': "0xEB014f8c8B418Db6b45774c326A0E64C78914dC0",
+    'gasPrice': "0x4a817c800",
+    'gas': "0x5208",
+    'to': '0x3535353535353535353535353535353535353535',
+    'value': "0x3e8",
+    'data': "deadbeef",
+}
+
+class pkGetter:
+
+    def __init__(self, pk):
+        self.pk = pk
+
+    def get(self, address, password=None):
+        return self.pk
+
+
+class TestSign(unittest.TestCase):
+
+    pk = None
+    nonce = -1
+    pk_getter = None
+
+
+    def getNonce(self):
+        self.nonce += 1
+        return self.nonce
+
+
+    def setUp(self):
+        self.pk = bytes.fromhex('5087503f0a9cc35b38665955eb830c63f778453dd11b8fa5bd04bc41fd2cc6d6')
+        self.pk_getter = pkGetter(self.pk)
+
+
+    def tearDown(self):
+        logg.info('teardown empty')
+
+
+
+    # TODO: verify rlp tx output
+    def test_serialize_transaction(self):
+        t = EIP155Transaction(tx_ints, 0)
+        self.assertRegex(t.__class__.__name__, "Transaction")
+        s = t.serialize()
+        self.assertDictEqual(s, {'nonce': '0x', 'gasPrice': '0x04a817c800', 'gas': '0x5208', 'to': '0x3535353535353535353535353535353535353535', 'value': '0x03e8', 'data': '0xdeadbeef', 'v': '0x01', 'r': '0x', 's': '0x'})
+        r = t.rlp_serialize()
+        self.assertEqual(r.hex(), 'ea808504a817c8008252089435353535353535353535353535353535353535358203e884deadbeef018080')
+
+        t = EIP155Transaction(tx_hexs, 0)
+        self.assertRegex(t.__class__.__name__, "Transaction")
+        s = t.serialize()
+        #o = json.loads(s)
+        self.assertDictEqual(s, {'nonce': '0x', 'gasPrice': '0x04a817c800', 'gas': '0x5208', 'to': '0x3535353535353535353535353535353535353535', 'value': '0x03e8', 'data': '0xdeadbeef', 'v': '0x01', 'r': '0x', 's': '0x'})
+        r = t.rlp_serialize()
+        self.assertEqual(r.hex(), 'ea808504a817c8008252089435353535353535353535353535353535353535358203e884deadbeef018080')
+
+
+
+    def test_sign_transaction(self):
+        t = EIP155Transaction(tx_ints, 461, 8995)
+        s = EIP155Signer(self.pk_getter)
+        z = s.sign_transaction(t)
+
+
+    def test_sign_message(self):
+        s = EIP155Signer(self.pk_getter)
+        z = s.sign_ethereum_message(tx_ints['from'], '666f6f')
+        z = s.sign_ethereum_message(tx_ints['from'], b'foo')
+        logg.debug('zzz {}'.format(str(z)))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_socket.py

@@ -0,0 +1,15 @@
+# standard imports
+import unittest
+import logging
+
+logg = logging.getLogger(__name__)
+
+
+class SocketTest(unittest.TestCase):
+
+    def test_placeholder_warning(self):
+        logg.warning('socket tests are missing! :/')
+
+
+if __name__ == '__main__':
+    unittest.main()