commit cb974126cdfb13a55e3eb641e5700c0cab90ceb6 parent 7f8a3628ce14d6387f7a875c1b22e3247220cc8c Author: nolash <dev@holbrook.no> Date: Sun, 10 Oct 2021 09:41:56 +0200 WIP refactor eth to separate module, rename to funga Diffstat:
35 files changed, 699 insertions(+), 440 deletions(-)
diff --git a/crypto_dev_signer/common.py b/crypto_dev_signer/common.py @@ -1,11 +0,0 @@ -def strip_hex_prefix(hx): - if hx[:2] == '0x': - return hx[2:] - return hx - -def add_hex_prefix(hx): - if len(hx) < 2: - return hx - if hx[:2] != '0x': - return '0x' + hx - return hx diff --git a/crypto_dev_signer/encoding.py b/crypto_dev_signer/encoding.py @@ -1,81 +0,0 @@ -# standard imports -import logging - -# external imports -import coincurve -import sha3 -from hexathon import ( - strip_0x, - uniform, - ) - -logg = logging.getLogger(__name__) - - -def private_key_from_bytes(b): - return coincurve.PrivateKey(secret=b) - - -def public_key_bytes_to_address(pubk_bytes, result_format='hex'): - h = sha3.keccak_256() - logg.debug('public key bytes {}'.format(pubk_bytes.hex())) - h.update(pubk_bytes[1:]) - z = h.digest()[12:] - if result_format == 'hex': - return to_checksum_address(z[:20].hex()) - elif result_format == 'bytes': - return z[:20] - raise ValueError('invalid result format "{}"'.format(result_format)) - - -def public_key_to_address(pubk, result_format='hex'): - pubk_bytes = pubk.format(compressed=False) - return public_key_bytes_to_address(pubk_bytes, result_format='hex') - - -def private_key_to_address(pk, result_format='hex'): - pubk = coincurve.PublicKey.from_secret(pk.secret) - #logg.debug('secret {} '.format(pk.secret.hex())) - return public_key_to_address(pubk, result_format) - - -def is_address(address_hex): - try: - address_hex = strip_0x(address_hex) - except ValueError: - return False - return len(address_hex) == 40 - - -def is_checksum_address(address_hex): - hx = None - try: - hx = to_checksum(address_hex) - except ValueError: - return False - return hx == address_hex - - -def to_checksum_address(address_hex): - address_hex = strip_0x(address_hex) - address_hex = uniform(address_hex) - if len(address_hex) != 40: - raise ValueError('Invalid address length') - h = sha3.keccak_256() - h.update(address_hex.encode('utf-8')) - z = h.digest() - - checksum_address_hex = '0x' - - for (i, c) in enumerate(address_hex): - if c in '1234567890': - checksum_address_hex += c - elif c in 'abcdef': - if z[int(i / 2)] & (0x80 >> ((i % 2) * 4)) > 1: - checksum_address_hex += c.upper() - else: - checksum_address_hex += c - - return checksum_address_hex - -to_checksum = to_checksum_address diff --git a/crypto_dev_signer/eth/encoding.py b/crypto_dev_signer/eth/encoding.py @@ -1,8 +0,0 @@ -ethereum_recid_modifier = 35 - -def chain_id_to_v(chain_id, signature): - v = signature[64] - return (chain_id * 2) + ethereum_recid_modifier + v - -def chainv_to_v(chain_id, v): - return v - ethereum_recid_modifier - (chain_id * 2) diff --git a/crypto_dev_signer/eth/signer/__init__.py b/crypto_dev_signer/eth/signer/__init__.py @@ -1 +0,0 @@ -from crypto_dev_signer.eth.signer.defaultsigner import ReferenceSigner, Signer diff --git a/crypto_dev_signer/eth/signer/defaultsigner.py b/crypto_dev_signer/eth/signer/defaultsigner.py @@ -1,90 +0,0 @@ -# standard imports -import logging - -# external imports -import sha3 -import coincurve -from hexathon import int_to_minbytes - -# local imports -from crypto_dev_signer.eth.encoding import chain_id_to_v - -logg = logging.getLogger(__name__) - - -class Signer: - - - def __init__(self, keyGetter): - self.keyGetter = keyGetter - - - def sign_transaction(self, tx, password=None): - return NotImplementedError - - -class ReferenceSigner(Signer): - - - def __init__(self, keyGetter): - super(ReferenceSigner, self).__init__(keyGetter) - - - def sign_transaction(self, tx, password=None): - s = tx.rlp_serialize() - h = sha3.keccak_256() - h.update(s) - message_to_sign = h.digest() - z = self.sign_pure(tx.sender, message_to_sign, password) - - return z - - - def sign_transaction_to_rlp(self, tx, password=None): - chain_id = int.from_bytes(tx.v, byteorder='big') - sig = self.sign_transaction(tx, password) - tx.apply_signature(chain_id, sig) - return tx.rlp_serialize() - - - def sign_transaction_to_wire(self, tx, password=None): - return self.sign_transaction_to_rlp(tx, password=password) - - - def sign_ethereum_message(self, address, message, password=None): - - #k = keys.PrivateKey(self.keyGetter.get(address, password)) - #z = keys.ecdsa_sign(message_hash=g, private_key=k) - if type(message).__name__ == 'str': - logg.debug('signing message in "str" format: {}'.format(message)) - #z = k.sign_msg(bytes.fromhex(message)) - message = bytes.fromhex(message) - elif type(message).__name__ == 'bytes': - logg.debug('signing message in "bytes" format: {}'.format(message.hex())) - #z = k.sign_msg(message) - else: - logg.debug('unhandled format {}'.format(type(message).__name__)) - raise ValueError('message must be type str or bytes, received {}'.format(type(message).__name__)) - - ethereumed_message_header = b'\x19' + 'Ethereum Signed Message:\n{}'.format(len(message)).encode('utf-8') - h = sha3.keccak_256() - h.update(ethereumed_message_header + message) - message_to_sign = h.digest() - - z = self.sign_pure(address, message_to_sign, password) - return z - - - # TODO: generic sign should be moved to non-eth context - def sign_pure(self, address, message, password=None): - pk = coincurve.PrivateKey(secret=self.keyGetter.get(address, password)) - z = pk.sign_recoverable(hasher=None, message=message) - return z - - - def sign_message(self, address, message, password=None, dialect='eth'): - if dialect == None: - return self.sign_pure(address, message, password=password) - elif dialect == 'eth': - return self.sign_ethereum_message(address, message, password=password) - raise ValueError('Unknown message sign dialect "{}"'.format(dialect)) diff --git a/crypto_dev_signer/keystore/dict.py b/crypto_dev_signer/keystore/dict.py @@ -1,44 +0,0 @@ -# standard imports -import logging - -# external imports -from hexathon import ( - strip_0x, - add_0x, - ) - -# local imports -#from . import keyapi -from .interface import Keystore -from crypto_dev_signer.error import UnknownAccountError -from crypto_dev_signer.encoding import private_key_to_address - -logg = logging.getLogger(__name__) - - -class DictKeystore(Keystore): - - def __init__(self): - self.keys = {} - - - def get(self, address, password=None): - address_key = strip_0x(address).lower() - if password != None: - logg.debug('password ignored as dictkeystore doesnt do encryption') - try: - return self.keys[address_key] - except KeyError: - raise UnknownAccountError(address_key) - - - def list(self): - return list(self.keys.keys()) - - - def import_key(self, pk, password=None): - address_hex = private_key_to_address(pk) - address_hex_clean = strip_0x(address_hex).lower() - self.keys[address_hex_clean] = pk.secret - logg.debug('added key {}'.format(address_hex)) - return add_0x(address_hex) diff --git a/crypto_dev_signer/keystore/interface.py b/crypto_dev_signer/keystore/interface.py @@ -1,47 +0,0 @@ -# standard imports -import os -import json -import logging - -# local imports -from crypto_dev_signer.keystore import keyfile -from crypto_dev_signer.encoding import private_key_from_bytes - -logg = logging.getLogger(__name__) - -class Keystore: - - def get(self, address, password=None): - raise NotImplementedError - - - def list(self): - raise NotImplementedError - - - def new(self, password=None): - b = os.urandom(32) - return self.import_raw_key(b, password) - - - def import_raw_key(self, b, password=None): - pk = private_key_from_bytes(b) - return self.import_key(pk, password) - - - def import_key(self, pk, password=None): - raise NotImplementedError - - - def import_keystore_data(self, keystore_content, password=''): - if type(keystore_content).__name__ == 'str': - keystore_content = json.loads(keystore_content) - elif type(keystore_content).__name__ == 'bytes': - logg.debug('bytes {}'.format(keystore_content)) - keystore_content = json.loads(keystore_content.decode('utf-8')) - private_key = keyfile.from_dict(keystore_content, password.encode('utf-8')) - return self.import_raw_key(private_key, password) - - def import_keystore_file(self, keystore_file, password=''): - private_key = keyfile.from_file(keystore_file, password) - return self.import_raw_key(private_key) diff --git a/crypto_dev_signer/keystore/keyfile.py b/crypto_dev_signer/keystore/keyfile.py @@ -1,143 +0,0 @@ -# standard imports -import os -import hashlib -import logging -import json -import uuid - -# external imports -import coincurve -from Crypto.Cipher import AES -from Crypto.Util import Counter -import sha3 - -# local imports -from crypto_dev_signer.encoding import private_key_to_address - -logg = logging.getLogger(__name__) - -algo_keywords = [ - 'aes-128-ctr', - ] -hash_keywords = [ - 'scrypt' - ] - -default_kdfparams = { - 'dklen': 32, - 'n': 1 << 18, - 'p': 1, - 'r': 8, - 'salt': os.urandom(32).hex(), - } - - -def to_mac(mac_key, ciphertext_bytes): - h = sha3.keccak_256() - h.update(mac_key) - h.update(ciphertext_bytes) - return h.digest() - - -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']) - 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) - - -class Ciphers: - - aes_128_block_size = 1 << 7 - aes_iv_len = 16 - - @staticmethod - def decrypt_aes_128_ctr(ciphertext, decryption_key, iv): - ctr = Counter.new(Ciphers.aes_128_block_size, initial_value=iv) - cipher = AES.new(decryption_key, AES.MODE_CTR, counter=ctr) - 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) - cipher = AES.new(encryption_key, AES.MODE_CTR, counter=ctr) - ciphertext = cipher.encrypt(plaintext) - return ciphertext - - -def to_dict(private_key_bytes, passphrase=''): - - private_key = coincurve.PrivateKey(secret=private_key_bytes) - - encryption_key = Hashes.from_scrypt(passphrase=passphrase) - - address_hex = private_key_to_address(private_key) - iv_bytes = os.urandom(Ciphers.aes_iv_len) - iv = int.from_bytes(iv_bytes, 'big') - ciphertext_bytes = Ciphers.encrypt_aes_128_ctr(private_key.secret, encryption_key[:16], iv) - - mac = to_mac(encryption_key[16:], ciphertext_bytes) - - crypto_dict = { - 'cipher': 'aes-128-ctr', - 'ciphertext': ciphertext_bytes.hex(), - 'cipherparams': { - 'iv': iv_bytes.hex(), - }, - 'kdf': 'scrypt', - 'kdfparams': default_kdfparams, - 'mac': mac.hex(), - } - - uu = uuid.uuid1() - o = { - 'address': address_hex, - '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)) - - kdf = o['crypto']['kdf'] - if kdf not in hash_keywords: - raise NotImplementedError('kdf "{}" not implemented'.format(kdf)) - - 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) - assert control_mac == calculated_mac - - m = getattr(Ciphers, 'decrypt_{}'.format(cipher.replace('-', '_'))) - pk = m(ciphertext_bytes, decryption_key[:16], iv) - return pk - - -def from_file(filepath, passphrase=''): - - f = open(filepath, 'r') - o = json.load(f) - f.close() - - return from_dict(o, passphrase) diff --git a/crypto_dev_signer/__init__.py b/funga/__init__.py diff --git a/crypto_dev_signer/error.py b/funga/error.py diff --git a/crypto_dev_signer/cli/__init__.py b/funga/eth/__init__.py diff --git a/crypto_dev_signer/eth/__init__.py b/funga/eth/cli/__init__.py diff --git a/crypto_dev_signer/cli/handle.py b/funga/eth/cli/handle.py diff --git a/crypto_dev_signer/cli/http.py b/funga/eth/cli/http.py diff --git a/crypto_dev_signer/cli/jsonrpc.py b/funga/eth/cli/jsonrpc.py diff --git a/crypto_dev_signer/cli/socket.py b/funga/eth/cli/socket.py diff --git a/funga/eth/encoding.py b/funga/eth/encoding.py @@ -0,0 +1,91 @@ +# standard imports +import logging + +# external imports +import coincurve +import sha3 +from hexathon import ( + strip_0x, + uniform, + ) + +logg = logging.getLogger(__name__) + + +def private_key_from_bytes(b): + return coincurve.PrivateKey(secret=b) + + +def public_key_bytes_to_address(pubk_bytes, result_format='hex'): + h = sha3.keccak_256() + logg.debug('public key bytes {}'.format(pubk_bytes.hex())) + h.update(pubk_bytes[1:]) + z = h.digest()[12:] + if result_format == 'hex': + return to_checksum_address(z[:20].hex()) + elif result_format == 'bytes': + return z[:20] + raise ValueError('invalid result format "{}"'.format(result_format)) + + +def public_key_to_address(pubk, result_format='hex'): + pubk_bytes = pubk.format(compressed=False) + return public_key_bytes_to_address(pubk_bytes, result_format='hex') + + +def private_key_to_address(pk, result_format='hex'): + pubk = coincurve.PublicKey.from_secret(pk.secret) + #logg.debug('secret {} '.format(pk.secret.hex())) + return public_key_to_address(pubk, result_format) + + +def is_address(address_hex): + try: + address_hex = strip_0x(address_hex) + except ValueError: + return False + return len(address_hex) == 40 + + +def is_checksum_address(address_hex): + hx = None + try: + hx = to_checksum(address_hex) + except ValueError: + return False + return hx == strip_0x(address_hex) + + +def to_checksum_address(address_hex): + address_hex = strip_0x(address_hex) + address_hex = uniform(address_hex) + if len(address_hex) != 40: + raise ValueError('Invalid address length') + h = sha3.keccak_256() + h.update(address_hex.encode('utf-8')) + z = h.digest() + + #checksum_address_hex = '0x' + checksum_address_hex = '' + + for (i, c) in enumerate(address_hex): + if c in '1234567890': + checksum_address_hex += c + elif c in 'abcdef': + if z[int(i / 2)] & (0x80 >> ((i % 2) * 4)) > 1: + checksum_address_hex += c.upper() + else: + checksum_address_hex += c + + return checksum_address_hex + +to_checksum = to_checksum_address + +ethereum_recid_modifier = 35 + +def chain_id_to_v(chain_id, signature): + v = signature[64] + return (chain_id * 2) + ethereum_recid_modifier + v + +def chainv_to_v(chain_id, v): + return v - ethereum_recid_modifier - (chain_id * 2) diff --git a/funga/eth/helper/__init__.py b/funga/eth/helper/__init__.py @@ -0,0 +1 @@ +from .tx import EthTxExecutor diff --git a/funga/eth/helper/tx.py b/funga/eth/helper/tx.py @@ -0,0 +1,58 @@ +# standard imports +import logging + +# local imports +from crypto_dev_signer.helper import TxExecutor +from crypto_dev_signer.error import NetworkError + +logg = logging.getLogger() +logging.getLogger('web3').setLevel(logging.CRITICAL) +logging.getLogger('urllib3').setLevel(logging.CRITICAL) + + +class EthTxExecutor(TxExecutor): + + def __init__(self, w3, sender, signer, chain_id, verifier=None, block=False): + self.w3 = w3 + nonce = self.w3.eth.getTransactionCount(sender, 'pending') + super(EthTxExecutor, self).__init__(sender, signer, self.translator, self.dispatcher, self.reporter, nonce, chain_id, self.fee_helper, self.fee_price_helper, verifier, block) + + + def fee_helper(self, tx): + estimate = self.w3.eth.estimateGas(tx) + if estimate < 21000: + estimate = 21000 + logg.debug('estimate {} {}'.format(tx, estimate)) + return estimate + + + def fee_price_helper(self): + return self.w3.eth.gasPrice + + + def dispatcher(self, tx): + error_object = None + try: + tx_hash = self.w3.eth.sendRawTransaction(tx) + except ValueError as e: + error_object = e.args[0] + logg.error('node could not intepret rlp {}'.format(tx)) + if error_object != None: + raise NetworkError(error_object) + return tx_hash + + + def reporter(self, tx): + return self.w3.eth.getTransactionReceipt(tx) + + + def translator(self, tx): + if tx.get('feePrice') != None: + tx['gasPrice'] = tx['feePrice'] + del tx['feePrice'] + + if tx.get('feeUnits') != None: + tx['gas'] = tx['feeUnits'] + del tx['feeUnits'] + + return tx diff --git a/crypto_dev_signer/keystore/__init__.py b/funga/eth/keystore/__init__.py diff --git a/funga/eth/keystore/dict.py b/funga/eth/keystore/dict.py @@ -0,0 +1,44 @@ +# standard imports +import logging + +# external imports +from hexathon import ( + strip_0x, + add_0x, + ) + +# local imports +#from . import keyapi +from funga.error import UnknownAccountError +from .interface import EthKeystore +from funga.eth.encoding import private_key_to_address + +logg = logging.getLogger(__name__) + + +class DictKeystore(EthKeystore): + + def __init__(self): + self.keys = {} + + + def get(self, address, password=None): + address_key = strip_0x(address).lower() + if password != None: + logg.debug('password ignored as dictkeystore doesnt do encryption') + try: + return self.keys[address_key] + except KeyError: + raise UnknownAccountError(address_key) + + + def list(self): + return list(self.keys.keys()) + + + def import_key(self, pk, password=None): + address_hex = private_key_to_address(pk) + address_hex_clean = strip_0x(address_hex).lower() + self.keys[address_hex_clean] = pk.secret + logg.debug('added key {}'.format(address_hex)) + return add_0x(address_hex) diff --git a/funga/eth/keystore/interface.py b/funga/eth/keystore/interface.py @@ -0,0 +1,53 @@ +# standard imports +import os +import json +import logging + +# local imports +from funga.keystore import Keystore +from funga.eth.keystore import keyfile +from funga.eth.encoding import private_key_from_bytes + +logg = logging.getLogger(__name__) + + +def native_keygen(self): + return os.urandom(32) + + +class EthKeystore(Keystore): + + def __init__(self, private_key_generator=native_keygen): + super(Keystore, self).__init__( + private_key_generator=private_key_generator, + private_key_parser=private_key_from_bytes, + keystore_parser=keyfile.from_some, + ) + + + def new(self, password=None): + return self.import_raw_key(b, password) + + + def import_raw_key(self, b, password=None): + pk = private_key_from_bytes(b) + return self.import_key(pk, password) + + + def import_key(self, pk, password=None): + raise NotImplementedError + + + def import_keystore_data(self, keystore_content, password=''): + if type(keystore_content).__name__ == 'str': + keystore_content = json.loads(keystore_content) + elif type(keystore_content).__name__ == 'bytes': + logg.debug('bytes {}'.format(keystore_content)) + keystore_content = json.loads(keystore_content.decode('utf-8')) + private_key = keyfile.from_dict(keystore_content, password.encode('utf-8')) + return self.import_raw_key(private_key, password) + + + def import_keystore_file(self, keystore_file, password=''): + private_key = keyfile.from_file(keystore_file, password) + return self.import_raw_key(private_key) diff --git a/funga/eth/keystore/keyfile.py b/funga/eth/keystore/keyfile.py @@ -0,0 +1,158 @@ +# standard imports +import os +import hashlib +import logging +import json +import uuid + +# external imports +import coincurve +from Crypto.Cipher import AES +from Crypto.Util import Counter +import sha3 + +# local imports +from funga.eth.encoding import private_key_to_address + +logg = logging.getLogger(__name__) + +algo_keywords = [ + 'aes-128-ctr', + ] +hash_keywords = [ + 'scrypt' + ] + +default_kdfparams = { + 'dklen': 32, + 'n': 1 << 18, + 'p': 1, + 'r': 8, + 'salt': os.urandom(32).hex(), + } + + +def to_mac(mac_key, ciphertext_bytes): + h = sha3.keccak_256() + h.update(mac_key) + h.update(ciphertext_bytes) + return h.digest() + + +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']) + 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) + + +class Ciphers: + + aes_128_block_size = 1 << 7 + aes_iv_len = 16 + + @staticmethod + def decrypt_aes_128_ctr(ciphertext, decryption_key, iv): + ctr = Counter.new(Ciphers.aes_128_block_size, initial_value=iv) + cipher = AES.new(decryption_key, AES.MODE_CTR, counter=ctr) + 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) + cipher = AES.new(encryption_key, AES.MODE_CTR, counter=ctr) + ciphertext = cipher.encrypt(plaintext) + return ciphertext + + +def to_dict(private_key_bytes, passphrase=''): + + private_key = coincurve.PrivateKey(secret=private_key_bytes) + + encryption_key = Hashes.from_scrypt(passphrase=passphrase) + + address_hex = private_key_to_address(private_key) + iv_bytes = os.urandom(Ciphers.aes_iv_len) + iv = int.from_bytes(iv_bytes, 'big') + ciphertext_bytes = Ciphers.encrypt_aes_128_ctr(private_key.secret, encryption_key[:16], iv) + + mac = to_mac(encryption_key[16:], ciphertext_bytes) + + crypto_dict = { + 'cipher': 'aes-128-ctr', + 'ciphertext': ciphertext_bytes.hex(), + 'cipherparams': { + 'iv': iv_bytes.hex(), + }, + 'kdf': 'scrypt', + 'kdfparams': default_kdfparams, + 'mac': mac.hex(), + } + + uu = uuid.uuid1() + o = { + 'address': address_hex, + '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)) + + kdf = o['crypto']['kdf'] + if kdf not in hash_keywords: + raise NotImplementedError('kdf "{}" not implemented'.format(kdf)) + + 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) + assert control_mac == calculated_mac + + m = getattr(Ciphers, 'decrypt_{}'.format(cipher.replace('-', '_'))) + pk = m(ciphertext_bytes, decryption_key[:16], iv) + return pk + + +def from_file(filepath, passphrase=''): + + f = open(filepath, 'r') + o = json.load(f) + f.close() + + return from_dict(o, passphrase) + + +def from_some(v, passphrase=''): + if isinstance(v, bytes): + v = v.decode('utf-8') + + if isinstance(v, str): + try: + return from_file(v, passphrase) + except Exception: + logg.debug('keyfile parse as file fail') + pass + v = json.loads(v) + + return from_dict(v, passphrase) diff --git a/crypto_dev_signer/keystore/reference.py b/funga/eth/keystore/reference.py diff --git a/crypto_dev_signer/runnable/keyfile.py b/funga/eth/runnable/keyfile.py diff --git a/crypto_dev_signer/runnable/signer.py b/funga/eth/runnable/signer.py diff --git a/funga/eth/signer/__init__.py b/funga/eth/signer/__init__.py @@ -0,0 +1 @@ +from funga.eth.signer.defaultsigner import EIP155Signer diff --git a/funga/eth/signer/defaultsigner.py b/funga/eth/signer/defaultsigner.py @@ -0,0 +1,79 @@ +# standard imports +import logging + +# external imports +import sha3 +import coincurve +from hexathon import int_to_minbytes + +# local imports +from funga.signer import Signer +from funga.eth.encoding import chain_id_to_v + +logg = logging.getLogger(__name__) + + +class EIP155Signer(Signer): + + def __init__(self, keyGetter): + super(EIP155Signer, self).__init__(keyGetter) + + + def sign_transaction(self, tx, password=None): + s = tx.rlp_serialize() + h = sha3.keccak_256() + h.update(s) + message_to_sign = h.digest() + z = self.sign_pure(tx.sender, message_to_sign, password) + + return z + + + def sign_transaction_to_rlp(self, tx, password=None): + chain_id = int.from_bytes(tx.v, byteorder='big') + sig = self.sign_transaction(tx, password) + tx.apply_signature(chain_id, sig) + return tx.rlp_serialize() + + + def sign_transaction_to_wire(self, tx, password=None): + return self.sign_transaction_to_rlp(tx, password=password) + + + def sign_ethereum_message(self, address, message, password=None): + + #k = keys.PrivateKey(self.keyGetter.get(address, password)) + #z = keys.ecdsa_sign(message_hash=g, private_key=k) + if type(message).__name__ == 'str': + logg.debug('signing message in "str" format: {}'.format(message)) + #z = k.sign_msg(bytes.fromhex(message)) + message = bytes.fromhex(message) + elif type(message).__name__ == 'bytes': + logg.debug('signing message in "bytes" format: {}'.format(message.hex())) + #z = k.sign_msg(message) + else: + logg.debug('unhandled format {}'.format(type(message).__name__)) + raise ValueError('message must be type str or bytes, received {}'.format(type(message).__name__)) + + ethereumed_message_header = b'\x19' + 'Ethereum Signed Message:\n{}'.format(len(message)).encode('utf-8') + h = sha3.keccak_256() + h.update(ethereumed_message_header + message) + message_to_sign = h.digest() + + z = self.sign_pure(address, message_to_sign, password) + return z + + + # TODO: generic sign should be moved to non-eth context + def sign_pure(self, address, message, password=None): + pk = coincurve.PrivateKey(secret=self.keyGetter.get(address, password)) + z = pk.sign_recoverable(hasher=None, message=message) + return z + + + def sign_message(self, address, message, password=None, dialect='eth'): + if dialect == None: + return self.sign_pure(address, message, password=password) + elif dialect == 'eth': + return self.sign_ethereum_message(address, message, password=password) + raise ValueError('Unknown message sign dialect "{}"'.format(dialect)) diff --git a/crypto_dev_signer/eth/transaction.py b/funga/eth/transaction.py diff --git a/funga/eth/web3ext/__init__.py b/funga/eth/web3ext/__init__.py @@ -0,0 +1,29 @@ +import logging +import re + +from web3 import Web3 as Web3super +from web3 import WebsocketProvider, HTTPProvider +from .middleware import PlatformMiddleware + +re_websocket = re.compile('^wss?://') +re_http = re.compile('^https?://') + +logg = logging.getLogger(__file__) + + +def create_middleware(ipcpath): + PlatformMiddleware.ipcaddr = ipcpath + return PlatformMiddleware + + +# overrides the original Web3 constructor +#def Web3(blockchain_provider='ws://localhost:8546', ipcpath=None): +def Web3(provider, ipcpath=None): + w3 = Web3super(provider) + + if ipcpath != None: + logg.info('using signer middleware with ipc {}'.format(ipcpath)) + w3.middleware_onion.add(create_middleware(ipcpath)) + + w3.eth.personal = w3.geth.personal + return w3 diff --git a/funga/eth/web3ext/middleware.py b/funga/eth/web3ext/middleware.py @@ -0,0 +1,116 @@ +# standard imports +import logging +import re +import socket +import uuid +import json + +logg = logging.getLogger(__file__) + + +def jsonrpc_request(method, params): + uu = uuid.uuid4() + return { + "jsonrpc": "2.0", + "id": str(uu), + "method": method, + "params": params, + } + +class PlatformMiddleware: + + # id for the request is not available, meaning we cannot easily short-circuit + # hack workaround + id_seq = -1 + re_personal = re.compile('^personal_.*') + ipcaddr = None + + + def __init__(self, make_request, w3): + self.w3 = w3 + self.make_request = make_request + if self.ipcaddr == None: + raise AttributeError('ipcaddr not set') + + + # TODO: understand what format input params come in + # single entry input gives a tuple on params, wtf... + # dict input comes as [{}] and fails if not passed on as an array + @staticmethod + def _translate_params(params): + #if params.__class__.__name__ == 'tuple': + # r = [] + # for p in params: + # r.append(p) + # return r + + if params.__class__.__name__ == 'list' and len(params) > 0: + return params[0] + + return params + + + # TODO: DRY + def __call__(self, method, suspect_params): + + self.id_seq += 1 + logg.debug('in middleware method {} params {} ipcpath {}'.format(method, suspect_params, self.ipcaddr)) + + if self.re_personal.match(method) != None: + params = PlatformMiddleware._translate_params(suspect_params) + # multiple providers is removed in web3.py 5.12.0 + # https://github.com/ethereum/web3.py/issues/1701 + # thus we need a workaround to use the same web3 instance + s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0) + ipc_provider_workaround = s.connect(self.ipcaddr) + + logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params)) + o = jsonrpc_request(method, params[0]) + j = json.dumps(o) + logg.debug('send {}'.format(j)) + s.send(j.encode('utf-8')) + r = s.recv(4096) + s.close() + logg.debug('got recv {}'.format(str(r))) + jr = json.loads(r) + jr['id'] = self.id_seq + #return str(json.dumps(jr)) + return jr + + elif method == 'eth_signTransaction': + params = PlatformMiddleware._translate_params(suspect_params) + s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0) + ipc_provider_workaround = s.connect(self.ipcaddr) + logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params)) + o = jsonrpc_request(method, params[0]) + j = json.dumps(o) + logg.debug('send {}'.format(j)) + s.send(j.encode('utf-8')) + r = s.recv(4096) + s.close() + logg.debug('got recv {}'.format(str(r))) + jr = json.loads(r) + jr['id'] = self.id_seq + #return str(json.dumps(jr)) + return jr + + elif method == 'eth_sign': + params = PlatformMiddleware._translate_params(suspect_params) + s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0) + ipc_provider_workaround = s.connect(self.ipcaddr) + logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params)) + o = jsonrpc_request(method, params) + j = json.dumps(o) + logg.debug('send {}'.format(j)) + s.send(j.encode('utf-8')) + r = s.recv(4096) + s.close() + logg.debug('got recv {}'.format(str(r))) + jr = json.loads(r) + jr['id'] = self.id_seq + return jr + + + + r = self.make_request(method, suspect_params) + return r diff --git a/funga/keystore.py b/funga/keystore.py @@ -0,0 +1,46 @@ +# standard imports +import os +import json +import logging + +logg = logging.getLogger(__name__) + + +class Keystore: + + + def __init__(self, private_key_generator, private_key_parser, keystore_parser): + self.private_key_generator = private_key_generator + self.private_key_parser = private_key_parser + self.keystore_parser = keystore_parser + + + def get(self, address, password=None): + raise NotImplementedError + + + def list(self): + raise NotImplementedError + + + def new(self, password=None): + self.private_key_generator(password=password) + + + def import_raw_key(self, b, password=None): + pk = self.private_key_parser(b) + return self.import_key(pk, password) + + + def import_key(self, pk, password=None): + raise NotImplementedError + + + def import_keystore_data(self, keystore_content, password=''): + private_key = self.keystore_parser(keystore_content, password.encode('utf-8')) + return self.import_raw_key(private_key, password=password) + + + def import_keystore_file(self, keystore_file, password=''): + private_key = self.keystore_parser(keystore_file, password) + return self.import_raw_key(private_key, password=password) diff --git a/funga/signer.py b/funga/signer.py @@ -0,0 +1,8 @@ +class Signer: + + def __init__(self, keyGetter): + self.keyGetter = keyGetter + + + def sign_transaction(self, tx, password=None): + return NotImplementedError diff --git a/setup.py b/setup.py @@ -32,18 +32,18 @@ while True: f.close() setup( - name="crypto-dev-signer", - version="0.4.15a7", + name="funga", + version="0.5.1a1", description="A signer and keystore daemon and library for cryptocurrency software development", author="Louis Holbrook", author_email="dev@holbrook.no", packages=[ - 'crypto_dev_signer.eth.signer', - 'crypto_dev_signer.eth', - 'crypto_dev_signer.cli', - 'crypto_dev_signer.keystore', - 'crypto_dev_signer.runnable', - 'crypto_dev_signer', + 'funga.eth.signer', + 'funga.eth', + 'funga.cli', + 'funga.keystore', + 'funga.runnable', + 'funga', ], install_requires=requirements, extras_require={ @@ -54,9 +54,9 @@ setup( long_description_content_type='text/markdown', entry_points = { 'console_scripts': [ - 'crypto-dev-daemon=crypto_dev_signer.runnable.signer:main', - 'eth-keyfile=crypto_dev_signer.runnable.keyfile:main', + 'crypto-dev-daemon=funga.runnable.signer:main', + 'eth-keyfile=funga.runnable.keyfile:main', ], }, - url='https://gitlab.com/chaintool/crypto-dev-signer', + url='https://gitlab.com/chaintool/funga', ) diff --git a/tests/test_keystore_dict.py b/tests/test_keystore_dict.py @@ -13,9 +13,9 @@ from hexathon import ( ) # local imports -from crypto_dev_signer.keystore.dict import DictKeystore -from crypto_dev_signer.error import UnknownAccountError -from crypto_dev_signer.eth.signer import ReferenceSigner +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() @@ -53,7 +53,7 @@ class TestDict(unittest.TestCase): def test_sign_message(self): - s = ReferenceSigner(self.db) + s = EIP155Signer(self.db) z = s.sign_ethereum_message(strip_0x(self.address_hex), b'foo') logg.debug('zzz {}'.format(str(z)))