Enable manually triggering keylist updates during connection

aries_cloudagent/config/wallet.py

@@ -7,13 +7,12 @@
 from ..core.profile import Profile, ProfileManager, ProfileSession
 from ..storage.base import BaseStorage
 from ..storage.error import StorageNotFoundError
-from ..version import __version__, RECORD_TYPE_ACAPY_VERSION
+from ..version import RECORD_TYPE_ACAPY_VERSION, __version__
 from ..wallet.base import BaseWallet
-from ..wallet.did_info import DIDInfo
 from ..wallet.crypto import seed_to_did
-from ..wallet.key_type import KeyType
+from ..wallet.did_info import DIDInfo
 from ..wallet.did_method import DIDMethod
-
+from ..wallet.key_type import KeyType
 from .base import ConfigError
 from .injection_context import InjectionContext
 

aries_cloudagent/connections/base_manager.py

@@ -24,6 +24,9 @@
 from ..protocols.coordinate_mediation.v1_0.models.mediation_record import (
     MediationRecord,
 )
+from ..protocols.coordinate_mediation.v1_0.route_manager import (
+    RouteManager,
+)
 from ..resolver.base import ResolverError
 from ..resolver.did_resolver import DIDResolver
 from ..storage.base import BaseStorage
@@ -56,6 +59,7 @@ def __init__(self, profile: Profile):
         """
         self._logger = logging.getLogger(__name__)
         self._profile = profile
+        self._route_manager = profile.inject(RouteManager)
 
     async def create_did_document(
         self,

aries_cloudagent/core/conductor.py

@@ -11,19 +11,20 @@
 import hashlib
 import json
 import logging
+
 from qrcode import QRCode
 
 from ..admin.base_server import BaseAdminServer
 from ..admin.server import AdminResponder, AdminServer
 from ..config.default_context import ContextBuilder
 from ..config.injection_context import InjectionContext
-from ..config.provider import ClassProvider
 from ..config.ledger import (
     get_genesis_transactions,
     ledger_config,
     load_multiple_genesis_transactions_from_config,
 )
 from ..config.logging import LoggingConfigurator
+from ..config.provider import ClassProvider
 from ..config.wallet import wallet_config
 from ..core.profile import Profile
 from ..indy.verifier import IndyVerifier
@@ -33,8 +34,8 @@
     BaseMultipleLedgerManager,
     MultipleLedgerManagerError,
 )
-from ..ledger.multiple_ledger.manager_provider import MultiIndyLedgerManagerProvider
 from ..ledger.multiple_ledger.ledger_requests_executor import IndyLedgerRequestsExecutor
+from ..ledger.multiple_ledger.manager_provider import MultiIndyLedgerManagerProvider
 from ..messaging.responder import BaseResponder
 from ..multitenant.base import BaseMultitenantManager
 from ..multitenant.manager_provider import MultitenantManagerProvider
@@ -45,8 +46,12 @@
 from ..protocols.connections.v1_0.messages.connection_invitation import (
     ConnectionInvitation,
 )
-from ..protocols.coordinate_mediation.v1_0.manager import MediationManager
 from ..protocols.coordinate_mediation.mediation_invite_store import MediationInviteStore
+from ..protocols.coordinate_mediation.v1_0.manager import MediationManager
+from ..protocols.coordinate_mediation.v1_0.route_manager import RouteManager
+from ..protocols.coordinate_mediation.v1_0.route_manager_provider import (
+    RouteManagerProvider,
+)
 from ..protocols.out_of_band.v1_0.manager import OutOfBandManager
 from ..protocols.out_of_band.v1_0.messages.invitation import HSProto, InvitationMessage
 from ..storage.base import BaseStorage
@@ -61,12 +66,11 @@
 from ..utils.stats import Collector
 from ..utils.task_queue import CompletedTask, TaskQueue
 from ..vc.ld_proofs.document_loader import DocumentLoader
-from ..version import __version__, RECORD_TYPE_ACAPY_VERSION
+from ..version import RECORD_TYPE_ACAPY_VERSION, __version__
 from ..wallet.did_info import DIDInfo
-from .oob_processor import OobMessageProcessor
-
 from .dispatcher import Dispatcher
-from .util import STARTUP_EVENT_TOPIC, SHUTDOWN_EVENT_TOPIC
+from .oob_processor import OobMessageProcessor
+from .util import SHUTDOWN_EVENT_TOPIC, STARTUP_EVENT_TOPIC
 
 LOGGER = logging.getLogger(__name__)
 
@@ -204,6 +208,11 @@ async def setup(self):
                 BaseMultitenantManager, MultitenantManagerProvider(self.root_profile)
             )
 
+        # Bind route manager provider
+        context.injector.bind_provider(
+            RouteManager, RouteManagerProvider(self.root_profile)
+        )
+
         # Bind oob message processor to be able to receive and process un-encrypted
         # messages
         context.injector.bind_instance(

aries_cloudagent/core/tests/test_dispatcher.py

@@ -18,6 +18,7 @@
     V20CredProblemReport,
 )
 from ...protocols.problem_report.v1_0.message import ProblemReport
+from ...protocols.coordinate_mediation.v1_0.route_manager import RouteManager
 from ...transport.inbound.message import InboundMessage
 from ...transport.inbound.receipt import MessageReceipt
 from ...transport.outbound.message import OutboundMessage
@@ -31,6 +32,7 @@ def make_profile() -> Profile:
     profile.context.injector.bind_instance(ProtocolRegistry, ProtocolRegistry())
     profile.context.injector.bind_instance(Collector, Collector())
     profile.context.injector.bind_instance(EventBus, EventBus())
+    profile.context.injector.bind_instance(RouteManager, async_mock.MagicMock())
     return profile
 
 

aries_cloudagent/multitenant/base.py

@@ -1,31 +1,26 @@
 """Manager for multitenancy."""
 
+from abc import ABC, abstractmethod
 from datetime import datetime
 import logging
-from abc import abstractmethod, ABC
+from typing import Iterable, List, Optional, cast
 
 import jwt
-from typing import Iterable, List, Optional, cast
 
-from ..core.profile import (
-    Profile,
-    ProfileSession,
-)
-from ..messaging.responder import BaseResponder
 from ..config.injection_context import InjectionContext
-from ..wallet.models.wallet_record import WalletRecord
-from ..wallet.base import BaseWallet
 from ..core.error import BaseError
-from ..protocols.routing.v1_0.manager import RouteNotFoundError, RoutingManager
-from ..protocols.routing.v1_0.models.route_record import RouteRecord
-from ..transport.wire_format import BaseWireFormat
-from ..storage.base import BaseStorage
-from ..storage.error import StorageNotFoundError
+from ..core.profile import Profile, ProfileSession
 from ..protocols.coordinate_mediation.v1_0.manager import (
     MediationManager,
     MediationRecord,
 )
-
+from ..protocols.coordinate_mediation.v1_0.route_manager import RouteManager
+from ..protocols.routing.v1_0.manager import RouteNotFoundError, RoutingManager
+from ..protocols.routing.v1_0.models.route_record import RouteRecord
+from ..storage.base import BaseStorage
+from ..transport.wire_format import BaseWireFormat
+from ..wallet.base import BaseWallet
+from ..wallet.models.wallet_record import WalletRecord
 from .error import WalletKeyMissingError
 
 LOGGER = logging.getLogger(__name__)
@@ -204,8 +199,8 @@ async def create_wallet(
                 public_did_info = await wallet.get_public_did()
 
             if public_did_info:
-                await self.add_key(
-                    wallet_record.wallet_id, public_did_info.verkey, skip_if_exists=True
+                await profile.inject(RouteManager).route_public_did(
+                    profile, public_did_info.verkey
                 )
         except Exception:
             await wallet_record.delete_record(session)
@@ -285,49 +280,6 @@ async def remove_wallet_profile(self, profile: Profile):
 
         """
 
-    async def add_key(
-        self, wallet_id: str, recipient_key: str, *, skip_if_exists: bool = False
-    ):
-        """
-        Add a wallet key to map incoming messages to specific subwallets.
-
-        Args:
-            wallet_id: The wallet id the key corresponds to
-            recipient_key: The recipient key belonging to the wallet
-            skip_if_exists: Whether to skip the action if the key is already registered
-                            for relaying / mediation
-        """
-
-        LOGGER.info(
-            f"Add route record for recipient {recipient_key} to wallet {wallet_id}"
-        )
-        routing_mgr = RoutingManager(self._profile)
-        mediation_mgr = MediationManager(self._profile)
-        mediation_record = await mediation_mgr.get_default_mediator()
-
-        if skip_if_exists:
-            try:
-                async with self._profile.session() as session:
-                    await RouteRecord.retrieve_by_recipient_key(session, recipient_key)
-
-                # If no error is thrown, it means there is already a record
-                return
-            except (StorageNotFoundError):
-                pass
-
-        await routing_mgr.create_route_record(
-            recipient_key=recipient_key, internal_wallet_id=wallet_id
-        )
-
-        # External mediation
-        if mediation_record:
-            keylist_updates = await mediation_mgr.add_key(recipient_key)
-
-            responder = self._profile.inject(BaseResponder)
-            await responder.send(
-                keylist_updates, connection_id=mediation_record.connection_id
-            )
-
     async def create_auth_token(
         self, wallet_record: WalletRecord, wallet_key: str = None
     ) -> str:

aries_cloudagent/multitenant/route_manager.py

@@ -0,0 +1,105 @@
+"""Multitenancy route manager."""
+
+
+import logging
+from typing import List, Optional, Tuple
+
+from ..core.profile import Profile
+from ..messaging.responder import BaseResponder
+from ..protocols.coordinate_mediation.v1_0.manager import MediationManager
+from ..protocols.coordinate_mediation.v1_0.models.mediation_record import (
+    MediationRecord,
+)
+from ..protocols.coordinate_mediation.v1_0.route_manager import RouteManager
+from ..protocols.routing.v1_0.manager import RoutingManager
+from ..protocols.routing.v1_0.models.route_record import RouteRecord
+from ..storage.error import StorageNotFoundError
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class MultitenantRouteManager(RouteManager):
+    """Multitenancy route manager."""
+
+    def __init__(
+        self,
+        root_profile: Profile,
+    ):
+        """Initialize multitenant route manager."""
+        self.root_profile = root_profile
+
+    async def get_base_wallet_mediator(self) -> Optional[MediationRecord]:
+        """Get base wallet's default mediator."""
+        return await MediationManager(self.root_profile).get_default_mediator()
+
+    async def _route_for_key(
+        self,
+        profile: Profile,
+        recipient_key: str,
+        mediation_record: Optional[MediationRecord] = None,
+        *,
+        skip_if_exists: bool = False,
+        replace_key: Optional[str] = None,
+    ):
+        wallet_id = profile.settings["wallet.id"]
+        LOGGER.info(
+            f"Add route record for recipient {recipient_key} to wallet {wallet_id}"
+        )
+        routing_mgr = RoutingManager(self.root_profile)
+        mediation_mgr = MediationManager(self.root_profile)
+        # If base wallet had mediator, only notify that mediator.
+        # Else, if subwallet has mediator, notify that mediator.
+        base_mediation_record = await self.get_base_wallet_mediator()
+        mediation_record = base_mediation_record or mediation_record
+
+        if skip_if_exists:
+            try:
+                async with self.root_profile.session() as session:
+                    await RouteRecord.retrieve_by_recipient_key(session, recipient_key)
+
+                # If no error is thrown, it means there is already a record
+                return None
+            except (StorageNotFoundError):
+                pass
+
+        await routing_mgr.create_route_record(
+            recipient_key=recipient_key, internal_wallet_id=wallet_id
+        )
+
+        # External mediation
+        keylist_updates = None
+        if mediation_record:
+            keylist_updates = await mediation_mgr.add_key(recipient_key)
+            if replace_key:
+                keylist_updates = await mediation_mgr.remove_key(
+                    replace_key, keylist_updates
+                )
+
+            responder = self.root_profile.inject(BaseResponder)
+            await responder.send(
+                keylist_updates, connection_id=mediation_record.connection_id
+            )
+
+        return keylist_updates
+
+    async def routing_info(
+        self,
+        profile: Profile,
+        my_endpoint: str,
+        mediation_record: Optional[MediationRecord] = None,
+    ) -> Tuple[List[str], str]:
+        """Return routing info."""
+        routing_keys = []
+
+        base_mediation_record = await self.get_base_wallet_mediator()
+
+        if base_mediation_record:
+            routing_keys = base_mediation_record.routing_keys
+            my_endpoint = base_mediation_record.endpoint
+
+        if mediation_record:
+            routing_keys = [*routing_keys, *mediation_record.routing_keys]
+            my_endpoint = mediation_record.endpoint
+
+        return routing_keys, my_endpoint