Compare commits

...

3 Commits

Author SHA1 Message Date
3a94f7e5a2 Merge pull request 'Ida/pbkdf2' (#1) from Ida/pbkdf2 into master
Reviewed-on: idaapayo/funga-eth#1
2022-01-19 13:52:48 +00:00
1f5d057a9a second pbkdf2 implementation 2022-01-19 14:37:22 +03:00
01598a8c59 pkdf2 implementation 2022-01-19 13:49:29 +03:00
3 changed files with 64 additions and 29 deletions

View File

@ -13,19 +13,20 @@ import sha3
# local imports # local imports
from funga.error import ( from funga.error import (
DecryptError, DecryptError,
KeyfileError, KeyfileError,
) )
from funga.eth.encoding import private_key_to_address from funga.eth.encoding import private_key_to_address
logg = logging.getLogger(__name__) logg = logging.getLogger(__name__)
algo_keywords = [ algo_keywords = [
'aes-128-ctr', 'aes-128-ctr',
] ]
hash_keywords = [ hash_keywords = [
'scrypt' 'scrypt',
] 'pbkdf2'
]
default_kdfparams = { default_kdfparams = {
'dklen': 32, 'dklen': 32,
@ -33,8 +34,14 @@ default_kdfparams = {
'p': 1, 'p': 1,
'r': 8, 'r': 8,
'salt': os.urandom(32).hex(), 'salt': os.urandom(32).hex(),
} }
pbkdf2_kdfparams = {
'c': 100000,
'dklen': 32,
'prf': 'sha256',
'salt': os.urandom(32).hex(),
}
def to_mac(mac_key, ciphertext_bytes): def to_mac(mac_key, ciphertext_bytes):
h = sha3.keccak_256() h = sha3.keccak_256()
@ -47,17 +54,35 @@ class Hashes:
@staticmethod @staticmethod
def from_scrypt(kdfparams=default_kdfparams, passphrase=''): def from_scrypt(kdfparams=default_kdfparams, passphrase=''):
dklen = int(kdfparams['dklen']) dklen = int(kdfparams['dklen'])
n = int(kdfparams['n']) n = int(kdfparams['n'])
p = int(kdfparams['p']) p = int(kdfparams['p'])
r = int(kdfparams['r']) r = int(kdfparams['r'])
salt = bytes.fromhex(kdfparams['salt']) 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)
@staticmethod
def from_pbkdf2(kdfparams=pbkdf2_kdfparams, passphrase=''):
hashname = kdfparams['prf']
pwd = passphrase.encode('utf-8')
salt = bytes.fromhex(kdfparams['salt'])
itr = int(kdfparams['c'])
dklen = int(kdfparams['dklen'])
derived_key = hashlib.pbkdf2_hmac(
hash_name=hashname,
password=pwd,
salt=salt,
iterations=itr,
dklen=dklen
)
return derived_key
class Ciphers: class Ciphers:
aes_128_block_size = 1 << 7 aes_128_block_size = 1 << 7
aes_iv_len = 16 aes_iv_len = 16
@ -68,7 +93,6 @@ class Ciphers:
plaintext = cipher.decrypt(ciphertext) plaintext = cipher.decrypt(ciphertext)
return plaintext return plaintext
@staticmethod @staticmethod
def encrypt_aes_128_ctr(plaintext, encryption_key, iv): def encrypt_aes_128_ctr(plaintext, encryption_key, iv):
ctr = Counter.new(Ciphers.aes_128_block_size, initial_value=iv) ctr = Counter.new(Ciphers.aes_128_block_size, initial_value=iv)
@ -77,11 +101,19 @@ class Ciphers:
return ciphertext return ciphertext
def to_dict(private_key_bytes, passphrase=''): def to_dict(private_key_bytes, kdf :str, passphrase=''):
private_key = coincurve.PrivateKey(secret=private_key_bytes) 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_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) address_hex = private_key_to_address(private_key)
iv_bytes = os.urandom(Ciphers.aes_iv_len) iv_bytes = os.urandom(Ciphers.aes_iv_len)
@ -95,11 +127,11 @@ def to_dict(private_key_bytes, passphrase=''):
'ciphertext': ciphertext_bytes.hex(), 'ciphertext': ciphertext_bytes.hex(),
'cipherparams': { 'cipherparams': {
'iv': iv_bytes.hex(), 'iv': iv_bytes.hex(),
}, },
'kdf': 'scrypt', 'kdf': kdf,
'kdfparams': default_kdfparams, 'kdfparams': kdfparams,
'mac': mac.hex(), 'mac': mac.hex(),
} }
uu = uuid.uuid1() uu = uuid.uuid1()
o = { o = {
@ -107,12 +139,11 @@ def to_dict(private_key_bytes, passphrase=''):
'version': 3, 'version': 3,
'crypto': crypto_dict, 'crypto': crypto_dict,
'id': str(uu), 'id': str(uu),
} }
return o return o
def from_dict(o, passphrase=''): def from_dict(o, passphrase=''):
cipher = o['crypto']['cipher'] cipher = o['crypto']['cipher']
if cipher not in algo_keywords: if cipher not in algo_keywords:
raise NotImplementedError('cipher "{}" not implemented'.format(cipher)) raise NotImplementedError('cipher "{}" not implemented'.format(cipher))
@ -121,19 +152,19 @@ def from_dict(o, passphrase=''):
if kdf not in hash_keywords: if kdf not in hash_keywords:
raise NotImplementedError('kdf "{}" not implemented'.format(kdf)) 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) decryption_key = m(o['crypto']['kdfparams'], passphrase)
control_mac = bytes.fromhex(o['crypto']['mac']) control_mac = bytes.fromhex(o['crypto']['mac'])
iv_bytes = bytes.fromhex(o['crypto']['cipherparams']['iv']) iv_bytes = bytes.fromhex(o['crypto']['cipherparams']['iv'])
iv = int.from_bytes(iv_bytes, "big") iv = int.from_bytes(iv_bytes, "big")
ciphertext_bytes = bytes.fromhex(o['crypto']['ciphertext']) ciphertext_bytes = bytes.fromhex(o['crypto']['ciphertext'])
# check mac # check mac
calculated_mac = to_mac(decryption_key[16:], ciphertext_bytes) calculated_mac = to_mac(decryption_key[16:], ciphertext_bytes)
if control_mac != calculated_mac: if control_mac != calculated_mac:
raise DecryptError('mac mismatch when decrypting passphrase') raise DecryptError('mac mismatch when decrypting passphrase')
m = getattr(Ciphers, 'decrypt_{}'.format(cipher.replace('-', '_'))) m = getattr(Ciphers, 'decrypt_{}'.format(cipher.replace('-', '_')))
try: try:
@ -145,7 +176,6 @@ def from_dict(o, passphrase=''):
def from_file(filepath, passphrase=''): def from_file(filepath, passphrase=''):
f = open(filepath, 'r') f = open(filepath, 'r')
try: try:
o = json.load(f) o = json.load(f)

View File

@ -16,6 +16,10 @@ from funga.eth.keystore.keyfile import (
from_file, from_file,
to_dict, to_dict,
) )
# from testkeyfile import (
# from_file,
# to_dict
# )
from funga.eth.encoding import ( from funga.eth.encoding import (
private_key_to_address, private_key_to_address,
private_key_from_bytes, private_key_from_bytes,
@ -77,7 +81,7 @@ def main():
else: else:
pk_bytes = os.urandom(32) pk_bytes = os.urandom(32)
pk = coincurve.PrivateKey(secret=pk_bytes) pk = coincurve.PrivateKey(secret=pk_bytes)
o = to_dict(pk_bytes, passphrase) o = to_dict(pk_bytes, 'pbkdf2', passphrase)
r = json.dumps(o) r = json.dumps(o)
print(r) print(r)

1
tests/testdata/foo2.json vendored Normal file
View File

@ -0,0 +1 @@
{"address": "a1De08A738F0bD3B261dC41d7E2102599bf648CA", "version": 3, "crypto": {"cipher": "aes-128-ctr", "ciphertext": "2240149943557906e6ee9bcb864d6148dce0c0d8245d1da8d466dcb39edcde6a", "cipherparams": {"iv": "a0c366fdfe86f21a4168b4f12059e7fa"}, "kdf": "pbkdf2", "kdfparams": {"c": 100000, "dklen": 32, "prf": "sha256", "salt": "73cb66c8cf0e60e36f7ec835a057638a17ed315cebb3b691574c32cfed827b6f"}, "mac": "6760ad0841da7976a41f134e021be450baa453efdec8fe5ba7790cfa79d0c14b"}, "id": "22e597ea-777e-11ec-be42-a8a7956d3851"}