forked from chaintool/funga-eth
Compare commits
11 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
c402163f63 | ||
|
05435268bf | ||
|
a534f8ca1e | ||
|
d9531c33cb | ||
45c7538528 | |||
|
385578e582 | ||
|
c3033e9572 | ||
|
d50489f0ba | ||
|
e5cd1cad58 | ||
|
30c82b9cd1 | ||
|
903f65936e |
16
CHANGELOG
Normal file
16
CHANGELOG
Normal file
@ -0,0 +1,16 @@
|
||||
* 0.6.2
|
||||
- Enable signing of binary message
|
||||
* 0.6.1
|
||||
- Avoid padding of addresses missing one nibble
|
||||
* 0.6.0
|
||||
- Upgrade confini
|
||||
* 0.5.4
|
||||
- Add message signer cli
|
||||
- Add pbkdf2 support
|
||||
- Add -0 flag for omitting newline on output
|
||||
- Revert RLP to 2.0.1, to not break eth-tester in dependents
|
||||
* 0.5.3
|
||||
- Upgrade RLP to 3.0.0 (eliminates really annoying cytoolz warning on stdout)
|
||||
---
|
||||
|
||||
changelog before 0.5.3 will be written later, sorry
|
@ -41,7 +41,7 @@ def private_key_to_address(pk, result_format='hex'):
|
||||
|
||||
def is_address(address_hex):
|
||||
try:
|
||||
address_hex = strip_0x(address_hex)
|
||||
address_hex = strip_0x(address_hex, pad=False)
|
||||
except ValueError:
|
||||
return False
|
||||
return len(address_hex) == 40
|
||||
@ -57,10 +57,10 @@ def is_checksum_address(address_hex):
|
||||
|
||||
|
||||
def to_checksum_address(address_hex):
|
||||
address_hex = strip_0x(address_hex)
|
||||
address_hex = uniform(address_hex)
|
||||
address_hex = strip_0x(address_hex, pad=False)
|
||||
if len(address_hex) != 40:
|
||||
raise ValueError('Invalid address length')
|
||||
address_hex = uniform(address_hex)
|
||||
h = sha3.keccak_256()
|
||||
h.update(address_hex.encode('utf-8'))
|
||||
z = h.digest()
|
||||
|
@ -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)
|
||||
|
||||
@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)
|
||||
|
@ -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,
|
||||
@ -28,6 +30,7 @@ logg = logging.getLogger()
|
||||
argparser = argparse.ArgumentParser()
|
||||
argparser.add_argument('-d', '--decrypt', dest='d', type=str, help='decrypt file')
|
||||
argparser.add_argument('--private-key', dest='private_key', action='store_true', help='output private key instead of address')
|
||||
argparser.add_argument('-0', dest='nonl', action='store_true', help='no newline at end of output')
|
||||
argparser.add_argument('-z', action='store_true', help='zero-length password')
|
||||
argparser.add_argument('-k', type=str, help='load key from file')
|
||||
argparser.add_argument('-v', action='store_true', help='be verbose')
|
||||
@ -77,10 +80,12 @@ def main():
|
||||
else:
|
||||
pk_bytes = os.urandom(32)
|
||||
pk = coincurve.PrivateKey(secret=pk_bytes)
|
||||
o = to_dict(pk_bytes, passphrase)
|
||||
o = to_dict(pk_bytes, passphrase=passphrase)
|
||||
r = json.dumps(o)
|
||||
|
||||
print(r)
|
||||
if not args.nonl:
|
||||
r += "\n"
|
||||
sys.stdout.write(r)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
62
funga/eth/runnable/msg.py
Normal file
62
funga/eth/runnable/msg.py
Normal file
@ -0,0 +1,62 @@
|
||||
# standard imports
|
||||
import os
|
||||
import logging
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
import getpass
|
||||
|
||||
# external impors
|
||||
import coincurve
|
||||
from hexathon import strip_0x
|
||||
|
||||
# local imports
|
||||
from funga.error import DecryptError
|
||||
from funga.eth.keystore.dict import DictKeystore
|
||||
from funga.eth.signer import EIP155Signer
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
logg = logging.getLogger()
|
||||
|
||||
argparser = argparse.ArgumentParser()
|
||||
argparser.add_argument('-f', type=str, help='Keyfile to use for signing')
|
||||
argparser.add_argument('-z', action='store_true', help='zero-length password')
|
||||
argparser.add_argument('-v', action='store_true', help='be verbose')
|
||||
argparser.add_argument('-0', dest='nonl', action='store_true', help='no newline at end of output')
|
||||
argparser.add_argument('-b', '--binary', dest='binary', action='store_true', help='parse input as binary hex')
|
||||
argparser.add_argument('msg', type=str, help='Message to sign')
|
||||
args = argparser.parse_args()
|
||||
|
||||
if args.v:
|
||||
logg.setLevel(logging.DEBUG)
|
||||
|
||||
|
||||
def main():
|
||||
passphrase = os.environ.get('WALLET_PASSPHRASE', os.environ.get('PASSPHRASE'))
|
||||
if args.z:
|
||||
passphrase = ''
|
||||
if passphrase == None:
|
||||
passphrase = getpass.getpass('decryption phrase: ')
|
||||
|
||||
keystore = DictKeystore()
|
||||
address = keystore.import_keystore_file(args.f, password=passphrase)
|
||||
|
||||
signer = EIP155Signer(keystore)
|
||||
|
||||
msg = None
|
||||
if args.binary:
|
||||
hx = strip_0x(args.msg, pad=True)
|
||||
msg = bytes.fromhex(hx)
|
||||
else:
|
||||
msg = args.msg.encode('utf-8').hex()
|
||||
sig = signer.sign_ethereum_message(address, msg, password=passphrase)
|
||||
|
||||
r = sig.hex()
|
||||
if not args.nonl:
|
||||
r += "\n"
|
||||
sys.stdout.write(r)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
@ -1,9 +1,10 @@
|
||||
cryptography==3.2.1
|
||||
pysha3==1.0.2
|
||||
rlp==2.0.1
|
||||
#rlp==3.0.0
|
||||
json-rpc==1.13.0
|
||||
confini~=0.5.1
|
||||
confini~=0.6.0
|
||||
coincurve==15.0.0
|
||||
hexathon~=0.1.0
|
||||
hexathon~=0.1.6
|
||||
pycryptodome==3.10.1
|
||||
funga==0.5.1
|
||||
funga==0.5.2
|
||||
|
5
setup.py
5
setup.py
@ -33,7 +33,7 @@ f.close()
|
||||
|
||||
setup(
|
||||
name="funga-eth",
|
||||
version="0.5.2",
|
||||
version="0.6.2",
|
||||
description="Ethereum implementation of the funga keystore and signer",
|
||||
author="Louis Holbrook",
|
||||
author_email="dev@holbrook.no",
|
||||
@ -55,8 +55,9 @@ setup(
|
||||
'console_scripts': [
|
||||
'funga-ethd=funga.eth.runnable.signer:main',
|
||||
'eth-keyfile=funga.eth.runnable.keyfile:main',
|
||||
'eth-sign-msg=funga.eth.runnable.msg:main',
|
||||
],
|
||||
},
|
||||
url='https://gitlab.com/chaintool/funga-eth',
|
||||
url='https://git.grassecon.net/chaintool/funga-eth',
|
||||
include_package_data=True,
|
||||
)
|
||||
|
@ -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__':
|
||||
|
59
tests/test_pbkdf2.py
Normal file
59
tests/test_pbkdf2.py
Normal file
@ -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()
|
1
tests/testdata/UTC--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a
vendored
Normal file
1
tests/testdata/UTC--2022-01-24T10-34-04Z--cc47ad90-71a0-7fbe-0224-63326e27263a
vendored
Normal file
@ -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":"{}"}
|
Loading…
Reference in New Issue
Block a user