funga-eth

Ethereum implementation of the funga keystore and signer
Log | Files | Refs | README | LICENSE

commit d50489f0ba01d927bf16465e3add64bb1029027a
parent 30c82b9cd1ea912ff872e699509a8fcac094405a
Author: lash <dev@holbrook.no>
Date:   Tue, 25 Jan 2022 09:02:00 +0000

Merge remote-tracking branch 'origin/master' into lash/message-signer

Diffstat:
Mfunga/eth/keystore/keyfile.py | 86+++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------
Mfunga/eth/runnable/keyfile.py | 2++
Mtests/test_cli.py | 53+++++++++++++++++++++++++++++++++--------------------
Atests/test_pbkdf2.py | 59+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Atests/testdata/UTC--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a | 2++
5 files changed, 152 insertions(+), 50 deletions(-)

diff --git a/funga/eth/keystore/keyfile.py b/funga/eth/keystore/keyfile.py @@ -13,28 +13,35 @@ import sha3 # local imports from funga.error import ( - DecryptError, - KeyfileError, - ) + DecryptError, + KeyfileError, +) from funga.eth.encoding import private_key_to_address logg = logging.getLogger(__name__) algo_keywords = [ 'aes-128-ctr', - ] +] hash_keywords = [ - 'scrypt' - ] + 'scrypt', + 'pbkdf2' +] -default_kdfparams = { +default_scrypt_kdfparams = { 'dklen': 32, 'n': 1 << 18, 'p': 1, 'r': 8, 'salt': os.urandom(32).hex(), - } +} +default_pbkdf2_kdfparams = { + 'c': 100000, + 'dklen': 32, + 'prf': 'sha256', + 'salt': os.urandom(32).hex(), +} def to_mac(mac_key, ciphertext_bytes): h = sha3.keccak_256() @@ -46,18 +53,32 @@ def to_mac(mac_key, ciphertext_bytes): class Hashes: @staticmethod - def from_scrypt(kdfparams=default_kdfparams, passphrase=''): - dklen = int(kdfparams['dklen']) - n = int(kdfparams['n']) - p = int(kdfparams['p']) - r = int(kdfparams['r']) + def from_scrypt(kdfparams=default_scrypt_kdfparams, passphrase=''): + dklen = int(kdfparams['dklen']) + n = int(kdfparams['n']) + p = int(kdfparams['p']) + r = int(kdfparams['r']) salt = bytes.fromhex(kdfparams['salt']) - return hashlib.scrypt(passphrase.encode('utf-8'), salt=salt,n=n, p=p, r=r, maxmem=1024*1024*1024, dklen=dklen) + return hashlib.scrypt(passphrase.encode('utf-8'), salt=salt, n=n, p=p, r=r, maxmem=1024 * 1024 * 1024, + dklen=dklen) - -class Ciphers: + @staticmethod + def from_pbkdf2(kdfparams=default_pbkdf2_kdfparams, passphrase=''): + if kdfparams['prf'] == 'hmac-sha256': + kdfparams['prf'].replace('hmac-sha256','sha256') + + derived_key = hashlib.pbkdf2_hmac( + hash_name='sha256', + password=passphrase.encode('utf-8'), + salt=bytes.fromhex(kdfparams['salt']), + iterations=int(kdfparams['c']), + dklen=int(kdfparams['dklen']) + ) + return derived_key + +class Ciphers: aes_128_block_size = 1 << 7 aes_iv_len = 16 @@ -68,7 +89,6 @@ class Ciphers: plaintext = cipher.decrypt(ciphertext) return plaintext - @staticmethod def encrypt_aes_128_ctr(plaintext, encryption_key, iv): ctr = Counter.new(Ciphers.aes_128_block_size, initial_value=iv) @@ -77,11 +97,19 @@ class Ciphers: return ciphertext -def to_dict(private_key_bytes, passphrase=''): - +def to_dict(private_key_bytes, kdf='scrypt', passphrase=''): private_key = coincurve.PrivateKey(secret=private_key_bytes) - encryption_key = Hashes.from_scrypt(passphrase=passphrase) + if kdf == 'scrypt': + encryption_key = Hashes.from_scrypt(passphrase=passphrase) + kdfparams = default_scrypt_kdfparams + + elif kdf == 'pbkdf2': + encryption_key = Hashes.from_pbkdf2(passphrase=passphrase) + kdfparams = pbkdf2_kdfparams + + else: + raise NotImplementedError("KDF not implemented: {0}".format(kdf)) address_hex = private_key_to_address(private_key) iv_bytes = os.urandom(Ciphers.aes_iv_len) @@ -95,11 +123,11 @@ def to_dict(private_key_bytes, passphrase=''): 'ciphertext': ciphertext_bytes.hex(), 'cipherparams': { 'iv': iv_bytes.hex(), - }, - 'kdf': 'scrypt', - 'kdfparams': default_kdfparams, + }, + 'kdf': kdf, + 'kdfparams': kdfparams, 'mac': mac.hex(), - } + } uu = uuid.uuid1() o = { @@ -107,12 +135,11 @@ def to_dict(private_key_bytes, passphrase=''): 'version': 3, 'crypto': crypto_dict, 'id': str(uu), - } + } return o def from_dict(o, passphrase=''): - cipher = o['crypto']['cipher'] if cipher not in algo_keywords: raise NotImplementedError('cipher "{}" not implemented'.format(cipher)) @@ -121,19 +148,19 @@ def from_dict(o, passphrase=''): if kdf not in hash_keywords: raise NotImplementedError('kdf "{}" not implemented'.format(kdf)) - m = getattr(Hashes, 'from_{}'.format(kdf.replace('-', '_'))) + m = getattr(Hashes, 'from_{}'.format(kdf.replace('-', '_'))) decryption_key = m(o['crypto']['kdfparams'], passphrase) control_mac = bytes.fromhex(o['crypto']['mac']) iv_bytes = bytes.fromhex(o['crypto']['cipherparams']['iv']) iv = int.from_bytes(iv_bytes, "big") ciphertext_bytes = bytes.fromhex(o['crypto']['ciphertext']) - + # check mac calculated_mac = to_mac(decryption_key[16:], ciphertext_bytes) if control_mac != calculated_mac: raise DecryptError('mac mismatch when decrypting passphrase') - + m = getattr(Ciphers, 'decrypt_{}'.format(cipher.replace('-', '_'))) try: @@ -145,7 +172,6 @@ def from_dict(o, passphrase=''): def from_file(filepath, passphrase=''): - f = open(filepath, 'r') try: o = json.load(f) diff --git a/funga/eth/runnable/keyfile.py b/funga/eth/runnable/keyfile.py @@ -16,6 +16,8 @@ from funga.eth.keystore.keyfile import ( from_file, to_dict, ) + + from funga.eth.encoding import ( private_key_to_address, private_key_from_bytes, diff --git a/tests/test_cli.py b/tests/test_cli.py @@ -5,10 +5,18 @@ import os # external imports from hexathon import strip_0x +from pathlib import Path + +import sys +path_root = Path('/home/vincent/ida/grassroots/funga-eth/funga/eth/keystore') +sys.path.append(str(path_root)) +print(sys.path) # 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 @@ -18,30 +26,30 @@ 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) + # 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': [''], - } + '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') + 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', @@ -62,26 +70,31 @@ class TestCli(unittest.TestCase): # eth_signTransaction wraps personal_signTransaction, so here we test both already q = { - 'id': 0, - 'method': 'eth_signTransaction', - 'params': [tx_s], - } + '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') + 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') + 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'], - } + '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') + self.assertEqual(strip_0x(result), + '50320dda75190a121b7b5979de66edadafd02bdfbe4f6d49552e79c01410d2464aae35e385c0e5b61663ff7b44ef65fa0ac7ad8a57472cf405db399b9dba3e1600') if __name__ == '__main__': diff --git a/tests/test_pbkdf2.py b/tests/test_pbkdf2.py @@ -0,0 +1,59 @@ +#!/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--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a') + + address_hex = self.db.import_keystore_file(keystore_filepath, 'test') + 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(), '0xb8df77e1b4fa142e83bf9706f66fd76ad2a564f8') + + 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() diff --git a/tests/testdata/UTC--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a b/tests/testdata/UTC--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a @@ -0,0 +1 @@ +{"id":"cc47ad90-71a0-7fbe-0224-63326e27263a","version":3,"crypto":{"cipher":"aes-128-ctr","cipherparams":{"iv":"7bff67c888a9878a88e8548a4598322d"},"ciphertext":"0cb0e3c69d224d0a645f2784b64f507e5aecdc7bb8a7ea31963d25e6b8020ccf","kdf":"pbkdf2","kdfparams":{"c":10240,"dklen":32,"prf":"hmac-sha256","salt":"02f8b51b07a66a357c2d812952e6bee70fccc2e6a55e7cbd5c22d97d32fa8873"},"mac":"bb45aaabdb9fbbbde89631444ac39f8d76107381f16591799664274fd5d8c5bb"},"address":"b8df77e1b4fa142e83bf9706f66fd76ad2a564f8","name":"","meta":"{}"} +\ No newline at end of file