fix: mediation webhook routing keys

.dockerignore

@@ -9,3 +9,4 @@ test-reports
 .python-version
 docker
 env
+.venv

aries_cloudagent/commands/default_version_upgrade_config.yml

@@ -1,3 +1,5 @@
+v0.11.0:
+  update_routing_keys: true
 v0.8.1:
   resave_records:
     base_record_path:
@@ -13,4 +15,4 @@ v0.7.1:
 v0.7.0:
   update_existing_records: false 
 v0.6.0:
-  update_existing_records: false
+  update_existing_records: false

aries_cloudagent/commands/tests/test_upgrade.py

@@ -1,17 +1,24 @@
 import asyncio
 
-from asynctest import mock as async_mock, TestCase as AsyncTestCase
+from asynctest import TestCase as AsyncTestCase, mock as async_mock
 
-from ...core.in_memory import InMemoryProfile
+from .. import upgrade as test_module
 from ...connections.models.conn_record import ConnRecord
+from ...core.in_memory import InMemoryProfile
+from ...protocols.coordinate_mediation.v1_0.models.mediation_record import (
+    MediationRecord,
+)
+from ...protocols.routing.v1_0.models.route_record import RouteRecord
 from ...storage.base import BaseStorage
 from ...storage.record import StorageRecord
 from ...version import __version__
-
-from .. import upgrade as test_module
 from ..upgrade import UpgradeError
 
 
+TEST_BASE58_VERKEY = "3Dn1SJNPaCXcvvJvSbsFWP2xaCjMom3can8CQNhWrTRx"
+TEST_VERKEY = "did:key:z6Mkgg342Ycpuk263R9d8Aq6MUaxPn1DDeHyGo38EefXmgDL"
+
+
 class TestUpgrade(AsyncTestCase):
     async def setUp(self):
         self.session = InMemoryProfile.test_session()
@@ -619,3 +626,106 @@ async def test_upgrade_explicit_check(self):
             await test_module.upgrade(profile=self.profile)
             assert mock_logger.warning.call_count == 1
             assert mock_logger.info.call_count == 0
+
+    async def test_update_routing_keys(self):
+        """Test update routing keys routine."""
+        routes = [
+            RouteRecord(
+                role=RouteRecord.ROLE_SERVER,
+                recipient_key=TEST_BASE58_VERKEY,
+                connection_id="dummy connection id",
+            ),
+            RouteRecord(
+                role=RouteRecord.ROLE_SERVER,
+                recipient_key=TEST_BASE58_VERKEY,
+                connection_id="dummy connection id",
+            ),
+            RouteRecord(
+                role=RouteRecord.ROLE_SERVER,
+                recipient_key=TEST_VERKEY,
+                connection_id="dummy connection id",
+            ),
+        ]
+        mediations = [
+            MediationRecord(
+                role=MediationRecord.ROLE_CLIENT,
+                state=MediationRecord.STATE_GRANTED,
+                connection_id="dummy connection id",
+                routing_keys=[TEST_BASE58_VERKEY],
+            ),
+            MediationRecord(
+                role=MediationRecord.ROLE_CLIENT,
+                state=MediationRecord.STATE_GRANTED,
+                connection_id="dummy connection id",
+                routing_keys=[TEST_BASE58_VERKEY, TEST_BASE58_VERKEY],
+            ),
+            MediationRecord(
+                role=MediationRecord.ROLE_CLIENT,
+                state=MediationRecord.STATE_GRANTED,
+                connection_id="dummy connection id",
+                routing_keys=[TEST_VERKEY],
+            ),
+            MediationRecord(
+                role=MediationRecord.ROLE_CLIENT,
+                state=MediationRecord.STATE_GRANTED,
+                connection_id="dummy connection id",
+                routing_keys=[TEST_VERKEY, TEST_BASE58_VERKEY],
+            ),
+        ]
+        for route in routes:
+            await route.save(self.session)
+        for mediation in mediations:
+            await mediation.save(self.session)
+
+        await test_module.update_routing_keys(self.profile)
+
+        mediation_records = await MediationRecord.query(self.session)
+        route_records = await RouteRecord.query(self.session)
+        assert len(mediation_records)
+        assert len(route_records)
+
+        for record in mediation_records:
+            for routing_key in record.routing_keys:
+                assert routing_key == TEST_VERKEY
+
+        for record in route_records:
+            assert record.recipient_key == TEST_VERKEY
+
+    async def test_update_routing_keys_called_from_executables(self):
+        version_storage_record = await self.storage.find_record(
+            type_filter="acapy_version", tag_query={}
+        )
+        await self.storage.delete_record(version_storage_record)
+        with async_mock.patch.object(
+            test_module,
+            "wallet_config",
+            async_mock.CoroutineMock(
+                return_value=(
+                    self.profile,
+                    async_mock.CoroutineMock(did="public DID", verkey="verkey"),
+                )
+            ),
+        ), async_mock.patch.object(
+            test_module.yaml,
+            "safe_load",
+            async_mock.MagicMock(
+                return_value={
+                    "v0.11.0": {
+                        "update_routing_keys": True,
+                    },
+                }
+            ),
+        ), async_mock.patch.object(
+            test_module,
+            "update_routing_keys",
+            async_mock.CoroutineMock(),
+        ) as mock_update_routing_keys:
+            test_module.UPGRADE_EXISTING_RECORDS_FUNCTION_MAPPING[
+                "update_routing_keys"
+            ] = mock_update_routing_keys
+            await test_module.upgrade(
+                settings={
+                    "upgrade.from_version": "v0.7.2",
+                }
+            )
+            mock_update_routing_keys.assert_called_once()

aries_cloudagent/commands/upgrade.py

@@ -1,6 +1,7 @@
 """Upgrade command for handling breaking changes when updating ACA-PY versions."""
 
 import asyncio
+import json
 import logging
 import os
 import yaml
@@ -19,14 +20,22 @@
     Tuple,
 )
 
+from ..protocols.coordinate_mediation.v1_0.models.mediation_record import (
+    MediationRecord,
+)
+from ..protocols.coordinate_mediation.v1_0.normalization import (
+    normalize_from_public_key,
+)
+from ..protocols.routing.v1_0.models.route_record import RouteRecord
+
 from ..core.profile import Profile
 from ..config import argparse as arg
 from ..config.default_context import DefaultContextBuilder
 from ..config.base import BaseError, BaseSettings
 from ..config.util import common_config
 from ..config.wallet import wallet_config
 from ..messaging.models.base_record import BaseRecord
-from ..storage.base import BaseStorage
+from ..storage.base import BaseStorage, BaseStorageSearch
 from ..storage.error import StorageNotFoundError
 from ..storage.record import StorageRecord
 from ..utils.classloader import ClassLoader, ClassNotFoundError
@@ -341,6 +350,7 @@ async def upgrade(
                         f"Only BaseRecord can be resaved, found: {str(rec_type)}"
                     )
                 async with root_profile.session() as session:
+                    # TODO This should use BaseStorageSearch so we don't risk OOM errors
                     all_records = await rec_type.query(session)
                     for record in all_records:
                         await record.save(
@@ -394,6 +404,61 @@ async def update_existing_records(profile: Profile):
     pass
 
 
+async def update_routing_keys(profile: Profile):
+    """Update routing keys previously stored in wallet.
+
+    This update step will transform the stored routing keys into did:key values
+    from raw base58 encoded values.
+
+    Steps:
+        for each mediation record stored in the wallet:
+            ensure the stored routing_keys list is formated as did:keys
+            save the record if modified
+        for each routing record stored in the wallet:
+            ensure the stored recipient keys are formated as did:keys
+            save the record if modified
+    """
+    async with profile.transaction() as txn:
+        searcher = txn.inject(BaseStorageSearch)
+        search = searcher.search_records(
+            MediationRecord.RECORD_TYPE,
+        )
+        async for record in search:
+            try:
+                value = json.loads(record.value)
+                record = MediationRecord.from_storage(record.id, value)
+                original = record.routing_keys
+                record.routing_keys = [
+                    normalize_from_public_key(key) for key in record.routing_keys
+                ]
+                if original != record.routing_keys:
+                    await record.save(
+                        txn, reason="Normalize routing keys to did:key", event=False
+                    )
+            except Exception:
+                LOGGER.exception("Error normalizing routing keys in mediation record")
+        await txn.commit()
+
+    async with profile.transaction() as txn:
+        searcher = txn.inject(BaseStorageSearch)
+        search = searcher.search_records(
+            RouteRecord.RECORD_TYPE,
+        )
+        async for record in search:
+            try:
+                value = json.loads(record.value)
+                record = RouteRecord.from_storage(record.id, value)
+                original = record.recipient_key
+                record.recipient_key = normalize_from_public_key(record.recipient_key)
+                if original != record.recipient_key:
+                    await record.save(
+                        txn, reason="Normalize recipient key to did:key", event=False
+                    )
+            except Exception:
+                LOGGER.exception("Error normalizing recipient key in route record")
+        await txn.commit()
+
+
 def execute(argv: Sequence[str] = None):
     """Entrypoint."""
     parser = arg.create_argument_parser(prog=PROG)
@@ -413,7 +478,8 @@ def main():
 
 
 UPGRADE_EXISTING_RECORDS_FUNCTION_MAPPING = {
-    "update_existing_records": update_existing_records
+    "update_existing_records": update_existing_records,
+    "update_routing_keys": update_routing_keys,
 }
 
 main()

aries_cloudagent/core/in_memory/profile.py

@@ -6,7 +6,7 @@
 
 from ...config.injection_context import InjectionContext
 from ...config.provider import ClassProvider
-from ...storage.base import BaseStorage
+from ...storage.base import BaseStorage, BaseStorageSearch
 from ...storage.vc_holder.base import VCHolder
 from ...utils.classloader import DeferLoad
 from ...wallet.base import BaseWallet
@@ -114,7 +114,9 @@ async def _setup(self):
 
     def _init_context(self):
         """Initialize the session context."""
-        self._context.injector.bind_instance(BaseStorage, STORAGE_CLASS(self.profile))
+        storage = STORAGE_CLASS(self.profile)
+        self._context.injector.bind_instance(BaseStorageSearch, storage)
+        self._context.injector.bind_instance(BaseStorage, storage)
         self._context.injector.bind_instance(BaseWallet, WALLET_CLASS(self.profile))
 
     @property

aries_cloudagent/messaging/valid.py

@@ -200,6 +200,26 @@ def __init__(self):
         )
 
 
+class NonSDList(Regexp):
+    """Validate NonSD List."""
+
+    EXAMPLE = [
+        "name",
+        "address",
+        "address.street_address",
+        "nationalities[1:3]",
+    ]
+    PATTERN = r"[a-z0-9:\[\]_\.@?\(\)]"
+
+    def __init__(self):
+        """Initialize the instance."""
+
+        super().__init__(
+            NonSDList.PATTERN,
+            error="Value {input} is not a valid NonSDList",
+        )
+
+
 class JSONWebToken(Regexp):
     """Validate JSON Web Token."""
 
@@ -208,7 +228,7 @@ class JSONWebToken(Regexp):
         "eyJhIjogIjAifQ."
         "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
     )
-    PATTERN = r"^[-_a-zA-Z0-9]*\.[-_a-zA-Z0-9]*\.[-_a-zA-Z0-9]*$"
+    PATTERN = r"^[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]*\.[a-zA-Z0-9_-]+$"
 
     def __init__(self):
         """Initialize the instance."""
@@ -219,6 +239,28 @@ def __init__(self):
         )
 
 
+class SDJSONWebToken(Regexp):
+    """Validate SD-JSON Web Token."""
+
+    EXAMPLE = (
+        "eyJhbGciOiJFZERTQSJ9."
+        "eyJhIjogIjAifQ."
+        "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
+        "~WyJEM3BUSFdCYWNRcFdpREc2TWZKLUZnIiwgIkRFIl0"
+        "~WyJPMTFySVRjRTdHcXExYW9oRkd0aDh3IiwgIlNBIl0"
+        "~WyJkVmEzX1JlTGNsWTU0R1FHZm5oWlRnIiwgInVwZGF0ZWRfYXQiLCAxNTcwMDAwMDAwXQ"
+    )
+    PATTERN = r"^[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]*\.[a-zA-Z0-9_-]+(?:~[a-zA-Z0-9._-]+)*~?$"
+
+    def __init__(self):
+        """Initialize the instance."""
+
+        super().__init__(
+            SDJSONWebToken.PATTERN,
+            error="Value {input} is not a valid SD-JSON Web token",
+        )
+
+
 class DIDKey(Regexp):
     """Validate value against DID key specification."""
 
@@ -800,9 +842,15 @@ def __init__(
 JWS_HEADER_KID_VALIDATE = JWSHeaderKid()
 JWS_HEADER_KID_EXAMPLE = JWSHeaderKid.EXAMPLE
 
+NON_SD_LIST_VALIDATE = NonSDList()
+NON_SD_LIST_EXAMPLE = NonSDList().EXAMPLE
+
 JWT_VALIDATE = JSONWebToken()
 JWT_EXAMPLE = JSONWebToken.EXAMPLE
 
+SD_JWT_VALIDATE = SDJSONWebToken()
+SD_JWT_EXAMPLE = SDJSONWebToken.EXAMPLE
+
 DID_KEY_VALIDATE = DIDKey()
 DID_KEY_EXAMPLE = DIDKey.EXAMPLE
 

aries_cloudagent/protocols/coordinate_mediation/v1_0/handlers/tests/test_mediation_grant_handler.py

@@ -18,7 +18,7 @@
 from .. import mediation_grant_handler as test_module
 
 TEST_CONN_ID = "conn-id"
-TEST_RECORD_VERKEY = "3Dn1SJNPaCXcvvJvSbsFWP2xaCjMom3can8CQNhWrTRx"
+TEST_BASE58_VERKEY = "3Dn1SJNPaCXcvvJvSbsFWP2xaCjMom3can8CQNhWrTRx"
 TEST_VERKEY = "did:key:z6Mkgg342Ycpuk263R9d8Aq6MUaxPn1DDeHyGo38EefXmgDL"
 TEST_ENDPOINT = "https://example.com"
 
@@ -59,7 +59,21 @@ async def test_handler(self):
         assert record
         assert record.state == MediationRecord.STATE_GRANTED
         assert record.endpoint == TEST_ENDPOINT
-        assert record.routing_keys == [TEST_RECORD_VERKEY]
+        assert record.routing_keys == [TEST_VERKEY]
+
+    async def test_handler_with_base58_received(self):
+        assert isinstance(self.context.message, MediationGrant)
+        self.context.message.routing_keys = [TEST_BASE58_VERKEY]
+        handler, responder = MediationGrantHandler(), MockResponder()
+        await MediationRecord(connection_id=TEST_CONN_ID).save(self.session)
+        await handler.handle(self.context, responder)
+        record = await MediationRecord.retrieve_by_connection_id(
+            self.session, TEST_CONN_ID
+        )
+        assert record
+        assert record.state == MediationRecord.STATE_GRANTED
+        assert record.endpoint == TEST_ENDPOINT
+        assert record.routing_keys == [TEST_VERKEY]
 
     async def test_handler_connection_has_set_to_default_meta(self):
         handler, responder = MediationGrantHandler(), MockResponder()