move initial in-memory ECDH-1PU support; enable tests; rename methods…

… and variables for consistency

Signed-off-by: Andrew Whitehead <cywolf@gmail.com>

aries_cloudagent/core/in_memory/__init__.py

@@ -0,0 +1,5 @@
+"""In-memory wallet support."""
+
+from .profile import InMemoryProfile, InMemoryProfileManager, InMemoryProfileSession
+
+__all__ = ["InMemoryProfile", "InMemoryProfileManager", "InMemoryProfileSession"]

aries_cloudagent/core/in_memory/didcomm/derive_1pu.py

@@ -0,0 +1,32 @@
+"""Functions for performing Key Agreement using ECDH-1PU."""
+
+from .derive_ecdh import derive_shared_secret, concat_kdf
+
+
+def derive_1pu(ze, zs, alg, apu, apv, keydatalen):
+    """Generate shared encryption key from two ECDH shared secrets."""
+
+    z = ze + zs
+    key = concat_kdf(z, alg, apu, apv, keydatalen)
+    return key
+
+
+def derive_sender_1pu(epk, sender_sk, recip_pk, alg, apu, apv, keydatalen):
+    """Generate two shared secrets (ze, zs)."""
+
+    ze = derive_shared_secret(epk, recip_pk)
+    zs = derive_shared_secret(sender_sk, recip_pk)
+
+    key = derive_1pu(ze, zs, alg, apu, apv, keydatalen)
+    return key
+
+
+def derive_receiver_1pu(epk, sender_pk, recip_sk, alg, apu, apv, keydatalen):
+    """Generate two shared secrets (ze, zs)."""
+
+    ze = derive_shared_secret(recip_sk, epk)
+    zs = derive_shared_secret(recip_sk, sender_pk)
+
+    key = derive_1pu(ze, zs, alg, apu, apv, keydatalen)
+
+    return key

aries_cloudagent/core/in_memory/didcomm/derive_ecdh.py

@@ -0,0 +1,68 @@
+"""Functions for performing Key Agreement."""
+
+import hashlib
+
+from binascii import unhexlify
+from typing import Union
+
+from ecdsa import ECDH, NIST256p
+
+
+def derive_shared_secret(private_key: bytes, public_key: bytes):
+    """Generate a shared secret from keys in byte format."""
+
+    derive = ECDH(curve=NIST256p)
+    derive.load_private_key_bytes(unhexlify(private_key))
+    derive.load_received_public_key_bytes(unhexlify(public_key))
+
+    secret = derive.generate_sharedsecret_bytes()
+    return secret
+
+
+def derive_shared_secret_from_key(private_key, public_key):
+    """Generate a shared secret from keys in ecdsa.Keys format."""
+
+    derive = ECDH(curve=NIST256p)
+    derive.load_private_key(private_key)
+    derive.load_received_public_key(public_key)
+
+    secret = derive.generate_sharedsecret_bytes()
+    return secret
+
+
+def _to_bytes(s: Union[str, bytes]) -> bytes:
+    if isinstance(s, str):
+        return s.encode("utf-8")
+    return s
+
+
+def concat_kdf(
+    shared_secret: bytes,
+    alg: Union[str, bytes],
+    apu: Union[str, bytes],
+    apv: Union[str, bytes],
+    keydatalen: int,
+):
+    """Generate a shared encryption key from a shared secret."""
+
+    alg = _to_bytes(alg)
+    apu = _to_bytes(apu)
+    apv = _to_bytes(apv)
+
+    # ECDH-1PU requires a "round number 1" to be prefixed onto the shared secret z
+    hasher = hashlib.sha256((1).to_bytes(4, "big"))
+    # Ze + Zs
+    hasher.update(shared_secret)
+    # AlgId
+    hasher.update(len(alg).to_bytes(4, "big"))
+    hasher.update(alg)
+    # PartyUInfo
+    hasher.update(len(apu).to_bytes(4, "big"))
+    hasher.update(apu)
+    # PartyVInfo
+    hasher.update(len(apv).to_bytes(4, "big"))
+    hasher.update(apv)
+    # SuppPubInfo
+    hasher.update((keydatalen * 8).to_bytes(4, "big"))
+
+    return hasher.digest()

aries_cloudagent/crypto/tests/test_1PU.py → aries_cloudagent/core/in_memory/didcomm/tests/test_1pu.py

@@ -1,13 +1,11 @@
-# from derive1PU import *
-from ..derive1PU import *
-from ...wallet.util import *
-
-from ecdsa import ECDH, NIST256p
-import base64
 from binascii import unhexlify
 
+from .....wallet.util import b64_to_bytes
+
+from ..derive_1pu import *
+
 
-def Test_Hex_Example():
+def test_1pu_hex_example():
 
     # Previously randomly generated 3 sets of keys
     aliceSecretKey = "23832cbef38641b8754a35f1f79bbcbc248e09ac93b01c2eaf12474f2ac406b6"
@@ -27,11 +25,11 @@ def Test_Hex_Example():
     apv = "Bob"
     keydatalen = 32  # 32 bytes or 256 bit output key length
 
-    aliceKey = deriveSender1PU(
+    aliceKey = derive_sender_1pu(
         aliceEphemeralSecretKey, aliceSecretKey, bobPublicKey, alg, apu, apv, keydatalen
     )
     print("Alice 1PU key: ", aliceKey.hex())
-    bobKey = deriveReceiver1PU(
+    bobKey = derive_receiver_1pu(
         aliceEphemeralPublicKey, alicePublicKey, bobSecretKey, alg, apu, apv, keydatalen
     )
     print("Bob 1PU key: ", bobKey.hex())
@@ -42,7 +40,7 @@ def Test_Hex_Example():
 
 
 # Example key exchange in https://tools.ietf.org/id/draft-madden-jose-ecdh-1pu-03.html#rfc.appendix.A
-def Test_Appendix_Example():
+def test_1pu_appendix_example():
 
     # Convert the three JWK keys into hex encoded byte format
 
@@ -86,11 +84,11 @@ def Test_Appendix_Example():
     apv = "Bob"
     keydatalen = 32  # 32 bytes or 256 bit output key length
 
-    aliceKey = deriveSender1PU(
+    aliceKey = derive_sender_1pu(
         aliceEphemeralSecretKey, aliceSecretKey, bobPublicKey, alg, apu, apv, keydatalen
     )
     print("Alice 1PU key: ", aliceKey.hex())
-    bobKey = deriveReceiver1PU(
+    bobKey = derive_receiver_1pu(
         aliceEphemeralPublicKey, alicePublicKey, bobSecretKey, alg, apu, apv, keydatalen
     )
     print("Bob 1PU key: ", bobKey.hex())
@@ -109,8 +107,8 @@ def Test_Appendix_Example():
 
 def main():
 
-    Test_Hex_Example()
-    Test_Appendix_Example()
+    test_1pu_hex_example()
+    test_1pu_appendix_example()
 
 
 if __name__ == "__main__":

aries_cloudagent/crypto/tests/test_ECDH.py → aries_cloudagent/core/in_memory/didcomm/tests/test_ecdh.py

@@ -1,9 +1,9 @@
-# from deriveECDH import *
-from ..deriveECDH import *
 from ecdsa import ECDH, NIST256p, SigningKey
 
+from ..derive_ecdh import *
+
 # Generate the same shared secret from imported generated keys
-def Test_DeriveECDHSecret():
+def test_ecdh_derive_shared_secret():
 
     # Import keys for two participating users
     aliceSecretKey = "23832cbef38641b8754a35f1f79bbcbc248e09ac93b01c2eaf12474f2ac406b6"
@@ -13,16 +13,16 @@ def Test_DeriveECDHSecret():
     bobPublicKey = "04e35cde5e3761d075fc87b3b0983a179e1b8e09da242e79965d657cba48f792dfc9b446a098ab0194888cd9d53a21c873c00264275dba925c2db6c458c87ca3d6"
 
     # Each user derives the same shared secret, independantly, using the other's public key which is exchanged
-    aliceSecret = DeriveECDHSecret(aliceSecretKey, bobPublicKey)
+    aliceSecret = derive_shared_secret(aliceSecretKey, bobPublicKey)
     print("Alice secret: ", aliceSecret.hex())
-    bobSecret = DeriveECDHSecret(bobSecretKey, alicePublicKey)
+    bobSecret = derive_shared_secret(bobSecretKey, alicePublicKey)
     print("Bob secret: ", bobSecret.hex())
 
     assert aliceSecret == bobSecret, "Both parties should generate the same secret"
 
 
 # Generate the same shared secret from random keys
-def Test_DeriveECDHSecretRandom():
+def test_ecdh_derive_shared_secret_random():
 
     # Generate random keys for the two participating users
     aliceSecretKey = SigningKey.generate(curve=NIST256p)
@@ -36,26 +36,26 @@ def Test_DeriveECDHSecretRandom():
     bobPublicKey = bob.get_public_key()
 
     # Each user derives the same shared secret, independantly, using the other's public key which is exchanged
-    aliceSecret = DeriveECDHSecretFromKey(aliceSecretKey, bobPublicKey)
+    aliceSecret = derive_shared_secret_from_key(aliceSecretKey, bobPublicKey)
     print("Alice secret: ", aliceSecret.hex())
-    bobSecret = DeriveECDHSecretFromKey(bobSecretKey, alicePublicKey)
+    bobSecret = derive_shared_secret_from_key(bobSecretKey, alicePublicKey)
     print("Bob secret: ", bobSecret.hex())
 
     assert aliceSecret == bobSecret, "Both parties should generate the same secret"
 
 
 # Test the entire key generation flow, DeriveECDHSecret() into ConcatKDF()
-def Test_GenerateKey():
+def test_ecdh_generate_key():
 
     aliceSecretKey = "23832cbef38641b8754a35f1f79bbcbc248e09ac93b01c2eaf12474f2ac406b6"
     alicePublicKey = "04fd4ca9eb7954a03517ac8249e6070aa3112e582f596b10f0d45d757b56d5dc0395a7d207d06503a4d6ad6e2ad3a1fd8cc233c072c0dc0f32213deb712c32cbdf"
 
     bobSecretKey = "2d1b242281944aa58c251ce12db6df8babd703b5c0a1fc0b9a34f5b7b9ad6030"
     bobPublicKey = "04e35cde5e3761d075fc87b3b0983a179e1b8e09da242e79965d657cba48f792dfc9b446a098ab0194888cd9d53a21c873c00264275dba925c2db6c458c87ca3d6"
 
-    aliceSecret = DeriveECDHSecret(aliceSecretKey, bobPublicKey)
+    aliceSecret = derive_shared_secret(aliceSecretKey, bobPublicKey)
     print("Alice secret: ", aliceSecret.hex())
-    bobSecret = DeriveECDHSecret(bobSecretKey, alicePublicKey)
+    bobSecret = derive_shared_secret(bobSecretKey, alicePublicKey)
     print("Bob secret: ", bobSecret.hex())
 
     # Header parameters used in ConcatKDF
@@ -65,19 +65,19 @@ def Test_GenerateKey():
     keydatalen = 32  # 32 bytes or 256 bit output key length
 
     # After each side generates the shared secret, it is used to independantly derive a shared encryption key
-    aliceKey = ConcatKDF(aliceSecret, alg, apu, apv, keydatalen)
+    aliceKey = concat_kdf(aliceSecret, alg, apu, apv, keydatalen)
     print("Alice key: ", aliceKey.hex())
 
-    bobKey = ConcatKDF(bobSecret, alg, apu, apv, keydatalen)
+    bobKey = concat_kdf(bobSecret, alg, apu, apv, keydatalen)
     print("Bob key: ", bobKey.hex())
 
     assert (
         aliceKey == bobKey
     ), "Both parties should generate the same key from the same secret"
 
 
-# Test the entire key generation flow, DeriveECDHSecretFromKey() into ConcatKDF()
-def Test_GenerateKeyRandom():
+# Test the entire key generation flow, derive_shared_secret() into concat_kdf()
+def test_ecdh_generate_key_random():
 
     aliceSecretKey = SigningKey.generate(curve=NIST256p)
     alice = ECDH(curve=NIST256p)
@@ -89,9 +89,9 @@ def Test_GenerateKeyRandom():
     bob.load_private_key(bobSecretKey)
     bobPublicKey = bob.get_public_key()
 
-    aliceSecret = DeriveECDHSecretFromKey(aliceSecretKey, bobPublicKey)
+    aliceSecret = derive_shared_secret_from_key(aliceSecretKey, bobPublicKey)
     print("Alice secret: ", aliceSecret.hex())
-    bobSecret = DeriveECDHSecretFromKey(bobSecretKey, alicePublicKey)
+    bobSecret = derive_shared_secret_from_key(bobSecretKey, alicePublicKey)
     print("Bob secret: ", bobSecret.hex())
 
     # Header parameters used in ConcatKDF
@@ -101,10 +101,10 @@ def Test_GenerateKeyRandom():
     keydatalen = 32  # 32 bytes or 256 bit output key length
 
     # After each side generates the shared secret, it is used to independantly derive a shared encryption key
-    aliceKey = ConcatKDF(aliceSecret, alg, apu, apv, keydatalen)
+    aliceKey = concat_kdf(aliceSecret, alg, apu, apv, keydatalen)
     print("Alice key: ", aliceKey.hex())
 
-    bobKey = ConcatKDF(bobSecret, alg, apu, apv, keydatalen)
+    bobKey = concat_kdf(bobSecret, alg, apu, apv, keydatalen)
     print("Bob key: ", bobKey.hex())
 
     assert (
@@ -114,10 +114,10 @@ def Test_GenerateKeyRandom():
 
 def main():
 
-    Test_DeriveECDHSecret()
-    Test_DeriveECDHSecretRandom()
-    Test_GenerateKey()
-    Test_GenerateKeyRandom()
+    test_ecdh_derive_shared_secret()
+    test_ecdh_derive_shared_secret_random()
+    test_ecdh_generate_key()
+    test_ecdh_generate_key_random()
 
 
 if __name__ == "__main__":

aries_cloudagent/core/in_memory.py → aries_cloudagent/core/in_memory/profile.py

@@ -4,14 +4,14 @@
 from typing import Any, Mapping, Type
 from weakref import ref
 
-from ..config.injection_context import InjectionContext
-from ..config.provider import ClassProvider
-from ..storage.base import BaseStorage
-from ..storage.vc_holder.base import VCHolder
-from ..utils.classloader import DeferLoad
-from ..wallet.base import BaseWallet
-
-from .profile import Profile, ProfileManager, ProfileSession
+from ...config.injection_context import InjectionContext
+from ...config.provider import ClassProvider
+from ...storage.base import BaseStorage
+from ...storage.vc_holder.base import VCHolder
+from ...utils.classloader import DeferLoad
+from ...wallet.base import BaseWallet
+
+from ..profile import Profile, ProfileManager, ProfileSession
 
 STORAGE_CLASS = DeferLoad("aries_cloudagent.storage.in_memory.InMemoryStorage")
 WALLET_CLASS = DeferLoad("aries_cloudagent.wallet.in_memory.InMemoryWallet")

aries_cloudagent/core/tests/test_in_memory.py → aries_cloudagent/core/in_memory/tests/test_profile.py

@@ -1,7 +1,6 @@
 import pytest
-import time
 
-from ..in_memory import InMemoryProfile
+from ..profile import InMemoryProfile
 
 
 @pytest.fixture()