Rename 'test_get_pk2' to 'test_get_eth2_pk' and use rather ETH2 public address

tests/ragger/staking_deposit/key_handling/key_derivation/path.py

@@ -0,0 +1,40 @@
+from typing import List
+
+from bip_utils import Bip39SeedGenerator
+from bip_utils.utils.mnemonic import Mnemonic
+
+from .tree import (
+    derive_master_SK,
+    derive_child_SK,
+)
+
+
+def path_to_nodes(path: str) -> List[int]:
+    """
+    Maps from a path string to a list of indices where each index represents the corresponding level in the path.
+    """
+    path = path.replace(' ', '')
+    if not set(path).issubset(set('m1234567890/')):
+        raise ValueError(f"Invalid path {path}")
+
+    indices = path.split('/')
+
+    if indices[0] != 'm':
+        raise ValueError(f"The first character of path should be `m`. Got {indices[0]}.")
+    indices.pop(0)
+
+    return [int(index) for index in indices]
+
+
+def mnemonic_and_path_to_key(mnemonic: str, path: str) -> int:
+    """
+    Return the SK at position `path`, derived from `mnemonic`. The password is to be
+    compliant with BIP39 mnemonics that use passwords, but is not used by this CLI outside of tests.
+    """
+    #seed = get_seed(mnemonic)
+    seed = Bip39SeedGenerator(Mnemonic.FromString(mnemonic)).Generate()
+
+    sk = derive_master_SK(seed)
+    for node in path_to_nodes(path):
+        sk = derive_child_SK(parent_SK=sk, index=node)
+    return sk

tests/ragger/staking_deposit/key_handling/key_derivation/tree.py

@@ -0,0 +1,85 @@
+from staking_deposit.utils.crypto import (
+    HKDF,
+    SHA256,
+)
+from py_ecc.optimized_bls12_381 import curve_order as bls_curve_order
+from typing import List
+
+
+def _flip_bits_256(input: int) -> int:
+    """
+    Flips 256 bits worth of `input`.
+    """
+    return input ^ (2**256 - 1)
+
+
+def _IKM_to_lamport_SK(*, IKM: bytes, salt: bytes) -> List[bytes]:
+    """
+    Derives the lamport SK for a given `IKM` and `salt`.
+
+    Ref: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-2333.md#ikm_to_lamport_sk
+    """
+    OKM = HKDF(salt=salt, IKM=IKM, L=8160)
+    lamport_SK = [OKM[i: i + 32] for i in range(0, 8160, 32)]
+    return lamport_SK
+
+
+def _parent_SK_to_lamport_PK(*, parent_SK: int, index: int) -> bytes:
+    """
+    Derives the `index`th child's lamport PK from the `parent_SK`.
+
+    Ref: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-2333.md#parent_sk_to_lamport_pk
+    """
+    salt = index.to_bytes(4, byteorder='big')
+    IKM = parent_SK.to_bytes(32, byteorder='big')
+    lamport_0 = _IKM_to_lamport_SK(IKM=IKM, salt=salt)
+    not_IKM = _flip_bits_256(parent_SK).to_bytes(32, byteorder='big')
+    lamport_1 = _IKM_to_lamport_SK(IKM=not_IKM, salt=salt)
+    lamport_SKs = lamport_0 + lamport_1
+    lamport_PKs = [SHA256(sk) for sk in lamport_SKs]
+    compressed_PK = SHA256(b''.join(lamport_PKs))
+    return compressed_PK
+
+
+def _HKDF_mod_r(*, IKM: bytes, key_info: bytes=b'') -> int:
+    """
+    Hashes the IKM using HKDF and returns the answer as an int modulo r, the BLS field order.
+
+    Ref: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-2333.md#hkdf_mod_r
+    """
+    L = 48  # `ceil((3 * ceil(log2(r))) / 16)`, where `r` is the order of the BLS 12-381 curve
+    salt = b'BLS-SIG-KEYGEN-SALT-'
+    SK = 0
+    while SK == 0:
+        salt = SHA256(salt)
+        okm = HKDF(
+            salt=salt,
+            IKM=IKM + b'\x00',  # add postfix `I2OSP(0, 1)`
+            L=L,
+            info=key_info + L.to_bytes(2, 'big'),
+        )
+        SK = int.from_bytes(okm, byteorder='big') % bls_curve_order
+    return SK
+
+
+def derive_child_SK(*, parent_SK: int, index: int) -> int:
+    """
+    Given a parent SK `parent_SK`, return the child SK at the supplied `index`.
+
+    Ref: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-2333.md#derive_child_sk
+    """
+    if index < 0 or index >= 2**32:
+        raise IndexError(f"`index` should be greater than or equal to  0 and less than 2**32. Got index={index}.")
+    lamport_PK = _parent_SK_to_lamport_PK(parent_SK=parent_SK, index=index)
+    return _HKDF_mod_r(IKM=lamport_PK)
+
+
+def derive_master_SK(seed: bytes) -> int:
+    """
+    Given a seed, derive the master SK.
+
+    Ref: https://github.com/ethereum/EIPs/blob/master/EIPS/eip-2333.md#derive_master_sk
+    """
+    if len(seed) < 32:
+        raise ValueError(f"`len(seed)` should be greater than or equal to 32. Got {len(seed)}.")
+    return _HKDF_mod_r(IKM=seed)

tests/ragger/staking_deposit/utils/crypto.py

@@ -0,0 +1,55 @@
+from typing import Any
+
+from Crypto.Hash import (
+    SHA256 as _sha256,
+    SHA512 as _sha512,
+)
+from Crypto.Protocol.KDF import (
+    scrypt as _scrypt,
+    HKDF as _HKDF,
+    PBKDF2 as _PBKDF2,
+)
+from Crypto.Cipher import (
+    AES as _AES
+)
+
+
+def SHA256(x: bytes) -> bytes:
+    return _sha256.new(x).digest()
+
+
+def scrypt(*, password: str, salt: str, n: int, r: int, p: int, dklen: int) -> bytes:
+    if n * r * p < 2**20:  # 128 MB memory usage
+        raise ValueError("The Scrypt parameters chosen are not secure.")
+    if n >= 2**(128 * r / 8):
+        raise ValueError("The given `n` should be less than `2**(128 * r / 8)`."
+                         f"\tGot `n={n}`, r={r}, 2**(128 * r / 8)={2**(128 * r / 8)}")
+    res = _scrypt(password=password, salt=salt, key_len=dklen, N=n, r=r, p=p)
+    return res if isinstance(res, bytes) else res[0]  # PyCryptodome can return Tuple[bytes]
+
+
+def PBKDF2(*, password: bytes, salt: bytes, dklen: int, c: int, prf: str) -> bytes:
+    if 'sha' not in prf:
+        raise ValueError(f"String 'sha' is not in `prf`({prf})")
+    if 'sha256' in prf and c < 2**18:
+        '''
+        Verify the number of rounds of SHA256-PBKDF2. SHA512 not checked as use in BIP39
+        does not require, and therefore doesn't use, safe parameters (c=2048).
+
+        Ref: https://github.com/bitcoin/bips/blob/master/bip-0039.mediawiki#from-mnemonic-to-seed
+        '''
+        raise ValueError("The PBKDF2 parameters chosen are not secure.")
+    _hash = _sha256 if 'sha256' in prf else _sha512
+    res = _PBKDF2(password=password, salt=salt, dkLen=dklen, count=c, hmac_hash_module=_hash)  # type: ignore
+    return res if isinstance(res, bytes) else res[0]  # PyCryptodome can return Tuple[bytes]
+
+
+def HKDF(*, salt: bytes, IKM: bytes, L: int, info: bytes=b'') -> bytes:
+    res = _HKDF(master=IKM, key_len=L, salt=salt, hashmod=_sha256, context=info)
+    return res if isinstance(res, bytes) else res[0]  # PyCryptodome can return Tuple[bytes]
+
+
+def AES_128_CTR(*, key: bytes, iv: bytes) -> Any:
+    if len(key) != 16:
+        raise ValueError(f"The key length should be 16. Got {len(key)}.")
+    return _AES.new(key=key, mode=_AES.MODE_CTR, initial_value=iv, nonce=b'')