forked from chaintool/funga-eth
Compare commits
No commits in common. "3a94f7e5a2fb3bc76d54460999fdce8f895b5b57" and "39435b67ba7441f86f4703e24a962f6ab1fd607d" have entirely different histories.
3a94f7e5a2
...
39435b67ba
@ -13,20 +13,19 @@ 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,
|
||||||
@ -34,14 +33,8 @@ 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()
|
||||||
@ -54,35 +47,17 @@ 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,
|
return hashlib.scrypt(passphrase.encode('utf-8'), salt=salt,n=n, p=p, r=r, maxmem=1024*1024*1024, dklen=dklen)
|
||||||
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
|
||||||
|
|
||||||
@ -93,6 +68,7 @@ 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)
|
||||||
@ -101,19 +77,11 @@ class Ciphers:
|
|||||||
return ciphertext
|
return ciphertext
|
||||||
|
|
||||||
|
|
||||||
def to_dict(private_key_bytes, kdf :str, passphrase=''):
|
def to_dict(private_key_bytes, passphrase=''):
|
||||||
|
|
||||||
private_key = coincurve.PrivateKey(secret=private_key_bytes)
|
private_key = coincurve.PrivateKey(secret=private_key_bytes)
|
||||||
|
|
||||||
if kdf == 'scrypt':
|
encryption_key = Hashes.from_scrypt(passphrase=passphrase)
|
||||||
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)
|
||||||
@ -127,11 +95,11 @@ def to_dict(private_key_bytes, kdf :str, passphrase=''):
|
|||||||
'ciphertext': ciphertext_bytes.hex(),
|
'ciphertext': ciphertext_bytes.hex(),
|
||||||
'cipherparams': {
|
'cipherparams': {
|
||||||
'iv': iv_bytes.hex(),
|
'iv': iv_bytes.hex(),
|
||||||
},
|
},
|
||||||
'kdf': kdf,
|
'kdf': 'scrypt',
|
||||||
'kdfparams': kdfparams,
|
'kdfparams': default_kdfparams,
|
||||||
'mac': mac.hex(),
|
'mac': mac.hex(),
|
||||||
}
|
}
|
||||||
|
|
||||||
uu = uuid.uuid1()
|
uu = uuid.uuid1()
|
||||||
o = {
|
o = {
|
||||||
@ -139,11 +107,12 @@ def to_dict(private_key_bytes, kdf :str, 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))
|
||||||
@ -152,19 +121,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:
|
||||||
@ -176,6 +145,7 @@ 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)
|
||||||
|
@ -16,10 +16,6 @@ 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,
|
||||||
@ -81,7 +77,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, 'pbkdf2', passphrase)
|
o = to_dict(pk_bytes, passphrase)
|
||||||
r = json.dumps(o)
|
r = json.dumps(o)
|
||||||
|
|
||||||
print(r)
|
print(r)
|
||||||
|
1
tests/testdata/foo2.json
vendored
1
tests/testdata/foo2.json
vendored
@ -1 +0,0 @@
|
|||||||
{"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"}
|
|
Loading…
Reference in New Issue
Block a user