From 770e8e54deaef99ceeb08645ce75331fda91e243 Mon Sep 17 00:00:00 2001 From: Daniel Bluhm Date: Thu, 19 Sep 2024 15:30:09 -0400 Subject: [PATCH] chore: delete unused keypair storage manager Signed-off-by: Daniel Bluhm --- aries_cloudagent/wallet/key_pair.py | 112 ------------------ .../wallet/tests/test_key_pair.py | 110 ----------------- 2 files changed, 222 deletions(-) delete mode 100644 aries_cloudagent/wallet/key_pair.py delete mode 100644 aries_cloudagent/wallet/tests/test_key_pair.py diff --git a/aries_cloudagent/wallet/key_pair.py b/aries_cloudagent/wallet/key_pair.py deleted file mode 100644 index 963cd256f1..0000000000 --- a/aries_cloudagent/wallet/key_pair.py +++ /dev/null @@ -1,112 +0,0 @@ -"""Key pair storage manager.""" - -import json -from typing import List, Mapping, Optional, Sequence - -from uuid_utils import uuid4 - -from ..storage.base import BaseStorage -from ..storage.record import StorageRecord -from .key_type import KeyType -from .util import bytes_to_b58 - -KEY_PAIR_STORAGE_TYPE = "key_pair" - - -class KeyPairStorageManager: - """Key pair storage manager.""" - - def __init__(self, store: BaseStorage) -> None: - """Create new KeyPairStorageManager instance. - - Args: - store (BaseStorage): The storage to use for the key pairs. - """ - self._store = store - - async def store_key_pair( - self, - public_key: bytes, - secret_key: bytes, - key_type: KeyType, - metadata: Optional[dict] = None, - tags: Optional[dict] = None, - ): - """Store signing key pair in storage. - - Args: - public_key (bytes): The public key - secret_key (bytes): The secret key - key_type (KeyType): The key type - metadata (dict, optional): The metadata - tags (dict, optional): The tags. - """ - metadata = metadata or {} - tags = tags or {} - verkey = bytes_to_b58(public_key) - data = { - "verkey": verkey, - "secret_key": bytes_to_b58(secret_key), - "key_type": key_type.key_type, - "metadata": metadata, - } - record = StorageRecord( - KEY_PAIR_STORAGE_TYPE, - json.dumps(data), - {**tags, "verkey": verkey, "key_type": key_type.key_type}, - uuid4().hex, - ) - - await self._store.add_record(record) - - async def get_key_pair(self, verkey: str) -> dict: - """Retrieve signing key pair from storage by verkey. - - Args: - storage (BaseStorage): The storage to use for querying - verkey (str): The verkey to query for - - Raises: - StorageDuplicateError: If more than one key pair is found for this verkey - StorageNotFoundError: If no key pair is found for this verkey - - Returns: - dict: The key pair data - - """ - - record = await self._store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - data = json.loads(record.value) - - return data - - async def find_key_pairs(self, tag_query: Optional[Mapping] = None) -> List[dict]: - """Find key pairs by tag query.""" - records: Sequence[StorageRecord] = await self._store.find_all_records( - KEY_PAIR_STORAGE_TYPE, tag_query - ) - - return [json.loads(record.value) for record in records] - - async def delete_key_pair(self, verkey: str): - """Remove a previously-stored key pair record. - - Raises: - StorageNotFoundError: If the record is not found - - """ - record = await self._store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - await self._store.delete_record(record) - - async def update_key_pair_metadata(self, verkey: str, metadata: dict): - """Update the metadata of a key pair record by verkey. - - Raises: - StorageNotFoundError: If the record is not found. - - """ - record = await self._store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - data = json.loads(record.value) - data["metadata"] = metadata - - await self._store.update_record(record, json.dumps(data), record.tags) diff --git a/aries_cloudagent/wallet/tests/test_key_pair.py b/aries_cloudagent/wallet/tests/test_key_pair.py deleted file mode 100644 index df5dee848e..0000000000 --- a/aries_cloudagent/wallet/tests/test_key_pair.py +++ /dev/null @@ -1,110 +0,0 @@ -import json -from unittest import IsolatedAsyncioTestCase - -from ...core.in_memory import InMemoryProfile -from ...storage.error import StorageNotFoundError -from ...storage.in_memory import InMemoryStorage -from ..key_pair import KEY_PAIR_STORAGE_TYPE, KeyPairStorageManager -from ..key_type import ED25519 -from ..util import bytes_to_b58 - - -class TestKeyPairStorageManager(IsolatedAsyncioTestCase): - test_public_key = b"somepublickeybytes" - test_secret = b"verysecretkey" - - async def asyncSetUp(self): - self.profile = InMemoryProfile.test_profile() - self.store = InMemoryStorage(self.profile) - self.key_pair_mgr = KeyPairStorageManager(self.store) - - async def test_create_key_pair(self): - await self.key_pair_mgr.store_key_pair( - public_key=self.test_public_key, - secret_key=self.test_secret, - key_type=ED25519, - ) - - verkey = bytes_to_b58(self.test_public_key) - - record = await self.store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - - assert record - - value = json.loads(record.value) - - assert record.tags == {"verkey": verkey, "key_type": ED25519.key_type} - assert value["verkey"] == verkey - assert value["secret_key"] == bytes_to_b58(self.test_secret) - assert value["metadata"] == {} - assert value["key_type"] == ED25519.key_type - - async def test_get_key_pair(self): - await self.key_pair_mgr.store_key_pair( - public_key=self.test_public_key, - secret_key=self.test_secret, - key_type=ED25519, - ) - - verkey = bytes_to_b58(self.test_public_key) - - key_pair = await self.key_pair_mgr.get_key_pair(verkey) - - assert key_pair["verkey"] == verkey - assert key_pair["secret_key"] == bytes_to_b58(self.test_secret) - assert key_pair["metadata"] == {} - assert key_pair["key_type"] == ED25519.key_type - - async def test_get_key_pair_x_not_found(self): - with self.assertRaises(StorageNotFoundError): - await self.key_pair_mgr.get_key_pair("not_existing_verkey") - - async def test_delete_key_pair(self): - await self.key_pair_mgr.store_key_pair( - public_key=self.test_public_key, - secret_key=self.test_secret, - key_type=ED25519, - ) - - verkey = bytes_to_b58(self.test_public_key) - - record = await self.store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - assert record - - await self.key_pair_mgr.delete_key_pair(verkey) - - # should be deleted now - with self.assertRaises(StorageNotFoundError): - await self.key_pair_mgr.delete_key_pair(verkey) - - async def test_delete_key_pair_x_not_found(self): - with self.assertRaises(StorageNotFoundError): - await self.key_pair_mgr.delete_key_pair("non_existing_verkey") - - async def test_update_key_pair_metadata(self): - await self.key_pair_mgr.store_key_pair( - public_key=self.test_public_key, - secret_key=self.test_secret, - key_type=ED25519, - metadata={"some": "data"}, - ) - - verkey = bytes_to_b58(self.test_public_key) - - record = await self.store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - assert record - value = json.loads(record.value) - - assert value["metadata"] == {"some": "data"} - - await self.key_pair_mgr.update_key_pair_metadata(verkey, {"some_other": "data"}) - - record = await self.store.find_record(KEY_PAIR_STORAGE_TYPE, {"verkey": verkey}) - assert record - value = json.loads(record.value) - - assert value["metadata"] == {"some_other": "data"} - - async def test_update_key_pair_metadata_x_not_found(self): - with self.assertRaises(StorageNotFoundError): - await self.key_pair_mgr.update_key_pair_metadata("non_existing_verkey", {})