Use more strict mypy setting and update KeystoreModule type ()

1. [KeystoreModule] change `params: dict` to `params:
Dict[str, Any]`
2. Rename `to_bytes` to `encode_bytes`
This commit is contained in:
Hsiao-Wei Wang 2020-05-19 21:34:16 +08:00 committed by GitHub
parent 4460c7c261
commit 581485f274
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 47 additions and 32 deletions

View File

@ -34,7 +34,7 @@ test: build_test
$(VENV_ACTIVATE) && python -m pytest .
lint: build_test
$(VENV_ACTIVATE) && flake8 --config=flake8.ini ./eth2deposit ./cli ./tests && mypy --config-file mypy.ini -p eth2deposit -p tests -p cli
$(VENV_ACTIVATE) && flake8 --config=flake8.ini ./eth2deposit ./cli ./tests && mypy --config-file mypy.ini -p eth2deposit -p cli
deposit: build
$(VENV_ACTIVATE) && python setup.py install; python ./cli/deposit.py

View File

@ -40,7 +40,7 @@ def generate_mnemonic(language: str, words_path: str) -> str:
return mnemonic
def check_python_version():
def check_python_version() -> None:
'''
Checks that the python version running is sufficient and exits if not.
'''
@ -54,12 +54,12 @@ def check_python_version():
'--num_validators',
prompt='Please choose how many validators you wish to run',
required=True,
type=int,
type=int, # type: ignore
)
@click.option(
'--mnemonic_language',
prompt='Please choose your mnemonic language',
type=click.Choice(languages, case_sensitive=False), # type: ignore
type=click.Choice(languages, case_sensitive=False),
default='english',
)
@click.option(

View File

@ -1,7 +1,7 @@
import os
import time
import json
from typing import List
from typing import Dict, List
from py_ecc.bls import G2ProofOfPossession as bls
from eth2deposit.key_handling.key_derivation.path import mnemonic_and_path_to_key
@ -26,14 +26,14 @@ class ValidatorCredentials:
self.amount = amount
@property
def signing_pk(self):
def signing_pk(self) -> bytes:
return bls.PrivToPub(self.signing_sk)
@property
def withdrawal_pk(self):
def withdrawal_pk(self) -> bytes:
return bls.PrivToPub(self.withdrawal_sk)
def signing_keystore(self, password: str) -> ScryptKeystore:
def signing_keystore(self, password: str) -> Keystore:
secret = self.signing_sk.to_bytes(32, 'big')
return ScryptKeystore.encrypt(secret=secret, password=password, path=self.signing_key_path)
@ -76,8 +76,8 @@ def sign_deposit_data(deposit_data: DepositMessage, sk: int) -> Deposit:
return signed_deposit_data
def export_deposit_data_json(*, credentials: List[ValidatorCredentials], folder: str):
deposit_data: List[dict] = []
def export_deposit_data_json(*, credentials: List[ValidatorCredentials], folder: str) -> str:
deposit_data: List[Dict[bytes, bytes]] = []
for credential in credentials:
deposit_datum = DepositMessage(
pubkey=credential.signing_pk,

View File

@ -4,6 +4,7 @@ from secrets import randbits
from typing import (
List,
Optional,
Sequence,
)
from eth2deposit.utils.crypto import (
@ -12,11 +13,11 @@ from eth2deposit.utils.crypto import (
)
def _get_word_list(language: str, path: str):
def _get_word_list(language: str, path: str) -> Sequence[str]:
return open(os.path.join(path, '%s.txt' % language)).readlines()
def _get_word(*, word_list, index: int) -> str:
def _get_word(*, word_list: Sequence[str], index: int) -> str:
assert index < 2048
return word_list[index][:-1]
@ -30,7 +31,7 @@ def get_seed(*, mnemonic: str, password: str='') -> bytes:
return PBKDF2(password=mnemonic, salt=salt, dklen=64, c=2048, prf='sha512')
def get_languages(path) -> List[str]:
def get_languages(path: str) -> List[str]:
"""
Walk the `path` and list all the languages with word-lists available.
"""

View File

@ -6,6 +6,7 @@ from dataclasses import (
)
import json
from secrets import randbits
from typing import Any, Dict, Union
from uuid import uuid4
from eth2deposit.utils.crypto import (
AES_128_CTR,
@ -18,21 +19,21 @@ from py_ecc.bls import G2ProofOfPossession as bls
hexdigits = set('0123456789abcdef')
def to_bytes(obj):
if isinstance(obj, str):
if all(c in hexdigits for c in obj):
return bytes.fromhex(obj)
def encode_bytes(obj: Union[str, Dict[str, Any]]) -> Union[bytes, str, Dict[str, Any]]:
if isinstance(obj, str) and all(c in hexdigits for c in obj):
return bytes.fromhex(obj)
elif isinstance(obj, dict):
for key, value in obj.items():
obj[key] = to_bytes(value)
obj[key] = encode_bytes(value)
return obj
class BytesDataclass:
def __post_init__(self):
def __post_init__(self) -> None:
for field in fields(self):
if field.type in (dict, bytes):
self.__setattr__(field.name, to_bytes(self.__getattribute__(field.name)))
if field.type in (bytes, Dict[str, Any]):
# Convert hexstring to bytes
self.__setattr__(field.name, encode_bytes(self.__getattribute__(field.name)))
def as_json(self) -> str:
return json.dumps(asdict(self), default=lambda x: x.hex())
@ -41,7 +42,7 @@ class BytesDataclass:
@dataclass
class KeystoreModule(BytesDataclass):
function: str = ''
params: dict = dataclass_field(default_factory=dict)
params: Dict[str, Any] = dataclass_field(default_factory=dict)
message: bytes = bytes()
@ -52,7 +53,7 @@ class KeystoreCrypto(BytesDataclass):
cipher: KeystoreModule = KeystoreModule()
@classmethod
def from_json(cls, json_dict: dict):
def from_json(cls, json_dict: Dict[Any, Any]) -> 'KeystoreCrypto':
kdf = KeystoreModule(**json_dict['kdf'])
checksum = KeystoreModule(**json_dict['checksum'])
cipher = KeystoreModule(**json_dict['cipher'])
@ -67,20 +68,20 @@ class Keystore(BytesDataclass):
uuid: str = str(uuid4()) # Generate a new uuid
version: int = 4
def kdf(self, **kwargs):
def kdf(self, **kwargs: Any) -> bytes:
return scrypt(**kwargs) if 'scrypt' in self.crypto.kdf.function else PBKDF2(**kwargs)
def save(self, file: str):
def save(self, file: str) -> None:
with open(file, 'w') as f:
f.write(self.as_json())
@classmethod
def open(cls, file: str):
def open(cls, file: str) -> 'Keystore':
with open(file, 'r') as f:
return cls.from_json(f.read())
@classmethod
def from_json(cls, path: str):
def from_json(cls, path: str) -> 'Keystore':
with open(path, 'r') as f:
json_dict = json.load(f)
crypto = KeystoreCrypto.from_json(json_dict['crypto'])
@ -93,7 +94,7 @@ class Keystore(BytesDataclass):
@classmethod
def encrypt(cls, *, secret: bytes, password: str, path: str='',
kdf_salt: bytes=randbits(256).to_bytes(32, 'big'),
aes_iv: bytes=randbits(128).to_bytes(16, 'big')):
aes_iv: bytes=randbits(128).to_bytes(16, 'big')) -> 'Keystore':
keystore = cls()
keystore.crypto.kdf.params['salt'] = kdf_salt
decryption_key = keystore.kdf(password=password, **keystore.crypto.kdf.params)

View File

@ -1,3 +1,5 @@
from typing import Any
from Crypto.Hash import (
SHA256 as _sha256,
SHA512 as _sha512,
@ -12,7 +14,7 @@ from Crypto.Cipher import (
)
def SHA256(x):
def SHA256(x: bytes) -> bytes:
return _sha256.new(x).digest()
@ -35,5 +37,5 @@ def HKDF(*, salt: bytes, IKM: bytes, L: int) -> bytes:
return res if isinstance(res, bytes) else res[0] # PyCryptodome can return Tuple[bytes]
def AES_128_CTR(*, key: bytes, iv: bytes):
def AES_128_CTR(*, key: bytes, iv: bytes) -> Any:
return _AES.new(key=key, mode=_AES.MODE_CTR, initial_value=iv, nonce=b'')

View File

@ -3,6 +3,8 @@ from eth_typing import (
BLSPubkey,
BLSSignature,
)
from typing import Any, Dict
from py_ecc.bls import G2ProofOfPossession as bls
from eth2deposit.utils.ssz import (
@ -25,7 +27,7 @@ def verify_deposit_data_json(filefolder: str) -> bool:
return False
def verify_deposit(deposit_data_dict: dict) -> bool:
def verify_deposit(deposit_data_dict: Dict[str, Any]) -> bool:
'''
Checks whether a deposit is valid based on the eth2 rules.
https://github.com/ethereum/eth2.0-specs/blob/dev/specs/phase0/beacon-chain.md#deposits

View File

@ -1,3 +1,12 @@
[mypy]
follow_imports = False
warn_unused_ignores = True
ignore_missing_imports = True
strict_optional = False
check_untyped_defs = True
disallow_incomplete_defs = True
disallow_untyped_defs = True
disallow_any_generics = True
disallow_untyped_calls = True
warn_redundant_casts = True
warn_unused_configs = True
strict_equality = True