Merge branch 'main' into docs/databases

aries_cloudagent/multitenant/askar_profile_manager.py

@@ -83,3 +83,12 @@ async def get_wallet_profile(
         ).extend(extra_settings)
 
         return AskarProfile(multitenant_wallet.opened, profile_context)
+
+    async def remove_wallet_profile(self, profile: Profile):
+        """Remove the wallet profile instance.
+
+        Args:
+            profile: The wallet profile instance
+
+        """
+        await profile.remove()

aries_cloudagent/multitenant/base.py

@@ -270,8 +270,7 @@ async def remove_wallet(self, wallet_id: str, wallet_key: str = None):
             {"wallet.key": wallet_key},
         )
 
-        del self._instances[wallet_id]
-        await profile.remove()
+        await self.remove_wallet_profile(profile)
 
         # Remove all routing records associated with wallet
         async with self._profile.session() as session:
@@ -282,6 +281,15 @@ async def remove_wallet(self, wallet_id: str, wallet_key: str = None):
 
             await wallet.delete_record(session)
 
+    @abstractmethod
+    async def remove_wallet_profile(self, profile: Profile):
+        """Remove the wallet profile instance.
+
+        Args:
+            profile: The wallet profile instance
+
+        """
+
     async def add_key(
         self, wallet_id: str, recipient_key: str, *, skip_if_exists: bool = False
     ):

aries_cloudagent/multitenant/manager.py

@@ -71,3 +71,14 @@ async def get_wallet_profile(
             self._instances[wallet_id] = profile
 
         return self._instances[wallet_id]
+
+    async def remove_wallet_profile(self, profile: Profile):
+        """Remove the wallet profile instance.
+
+        Args:
+            profile: The wallet profile instance
+
+        """
+        wallet_id = profile.settings.get("wallet.id")
+        del self._instances[wallet_id]
+        await profile.remove()

aries_cloudagent/multitenant/tests/test_askar_profile_manager.py

@@ -170,3 +170,10 @@ def side_effect(context, provision):
                     wallet_config.call_args[0][0].settings.get("wallet.name")
                     == multitenant_sub_wallet_name
                 )
+
+    async def test_remove_wallet_profile(self):
+        test_profile = InMemoryProfile.test_profile()
+
+        with async_mock.patch.object(InMemoryProfile, "remove") as profile_remove:
+            await self.manager.remove_wallet_profile(test_profile)
+            profile_remove.assert_called_once_with()

aries_cloudagent/multitenant/tests/test_base.py

@@ -307,8 +307,8 @@ async def test_remove_wallet_removes_profile_wallet_storage_records(self):
         ) as retrieve_by_id, async_mock.patch.object(
             BaseMultitenantManager, "get_wallet_profile"
         ) as get_wallet_profile, async_mock.patch.object(
-            InMemoryProfile, "remove"
-        ) as remove_profile, async_mock.patch.object(
+            BaseMultitenantManager, "remove_wallet_profile"
+        ) as remove_wallet_profile, async_mock.patch.object(
             WalletRecord, "delete_record"
         ) as wallet_delete_record, async_mock.patch.object(
             InMemoryStorage, "delete_all_records"
@@ -320,17 +320,15 @@ async def test_remove_wallet_removes_profile_wallet_storage_records(self):
             )
             wallet_profile = InMemoryProfile.test_profile()
 
-            self.manager._instances["test"] = wallet_profile
             retrieve_by_id.return_value = wallet_record
             get_wallet_profile.return_value = wallet_profile
 
             await self.manager.remove_wallet("test")
 
-            assert "test" not in self.manager._instances
             get_wallet_profile.assert_called_once_with(
                 self.profile.context, wallet_record, {"wallet.key": "test_key"}
             )
-            remove_profile.assert_called_once_with()
+            remove_wallet_profile.assert_called_once_with(wallet_profile)
             assert wallet_delete_record.call_count == 1
             delete_all_records.assert_called_once_with(
                 RouteRecord.RECORD_TYPE, {"wallet_id": "test"}

aries_cloudagent/multitenant/tests/test_manager.py

@@ -173,3 +173,14 @@ def side_effect(context, provision):
             assert profile.settings.get("mediation.invite") == "http://invite.com"
             assert profile.settings.get("mediation.default_id") == "24a96ef5"
             assert profile.settings.get("mediation.clear") == True
+
+    async def test_remove_wallet_profile(self):
+        test_profile = InMemoryProfile.test_profile(
+            settings={"wallet.id": "test"},
+        )
+        self.manager._instances["test"] = test_profile
+
+        with async_mock.patch.object(InMemoryProfile, "remove") as profile_remove:
+            await self.manager.remove_wallet_profile(test_profile)
+            assert "test" not in self.manager._instances
+            profile_remove.assert_called_once_with()

aries_cloudagent/protocols/out_of_band/v1_0/handlers/reuse_handler.py

@@ -27,6 +27,8 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
         profile = context.profile
         mgr = OutOfBandManager(profile)
         try:
-            await mgr.receive_reuse_message(context.message, context.message_receipt)
+            await mgr.receive_reuse_message(
+                context.message, context.message_receipt, context.connection_record
+            )
         except OutOfBandManagerError as e:
             self._logger.exception(f"Error processing Handshake Reuse message, {e}")

aries_cloudagent/protocols/out_of_band/v1_0/handlers/tests/test_reuse_handler.py

@@ -38,10 +38,13 @@ async def test_called(self, mock_oob_mgr, request_context):
         mock_oob_mgr.return_value.receive_reuse_message = async_mock.CoroutineMock()
         request_context.message = HandshakeReuse()
         handler = test_module.HandshakeReuseMessageHandler()
+        request_context.connection_record = ConnRecord()
         responder = MockResponder()
         await handler.handle(request_context, responder)
         mock_oob_mgr.return_value.receive_reuse_message.assert_called_once_with(
-            request_context.message, request_context.message_receipt
+            request_context.message,
+            request_context.message_receipt,
+            request_context.connection_record,
         )
 
     @pytest.mark.asyncio
@@ -52,11 +55,13 @@ async def test_reuse_accepted(self, mock_oob_mgr, request_context):
         mock_oob_mgr.return_value.receive_reuse_message.return_value = reuse_accepted
         request_context.message = HandshakeReuse()
         handler = test_module.HandshakeReuseMessageHandler()
+        request_context.connection_record = ConnRecord()
         responder = MockResponder()
         await handler.handle(request_context, responder)
         mock_oob_mgr.return_value.receive_reuse_message.assert_called_once_with(
             request_context.message,
             request_context.message_receipt,
+            request_context.connection_record,
         )
 
     @pytest.mark.asyncio
@@ -68,6 +73,7 @@ async def test_exception(self, mock_oob_mgr, request_context):
         )
         request_context.message = HandshakeReuse()
         handler = test_module.HandshakeReuseMessageHandler()
+        request_context.connection_record = ConnRecord()
         responder = MockResponder()
         await handler.handle(request_context, responder)
         assert mock_oob_mgr.return_value._logger.exception.called_once_("error")

aries_cloudagent/protocols/out_of_band/v1_0/manager.py

@@ -985,9 +985,7 @@ async def create_handshake_reuse_message(
 
         """
         try:
-            # ID of Out-of-Band invitation to use as a pthid
-            # pthid = invi_msg._id
-            pthid = conn_record.invitation_msg_id
+            pthid = invi_msg._id
             reuse_msg = HandshakeReuse()
             thid = reuse_msg._id
             reuse_msg.assign_thread_id(thid=thid, pthid=pthid)
@@ -1012,10 +1010,25 @@ async def create_handshake_reuse_message(
                 f"Error on creating and sending a handshake reuse message: {err}"
             )
 
+    async def delete_stale_connection_by_invitation(self, invi_msg_id: str):
+        """Delete unused connections, using existing an active connection instead."""
+        post_filter = {}
+        post_filter["invitation_msg_id"] = invi_msg_id
+        post_filter["invitation_mode"] = "once"
+        post_filter["state"] = "invitation"
+        async with self.profile.session() as session:
+            conn_records = await ConnRecord.query(
+                session,
+                post_filter_positive=post_filter,
+            )
+            for conn_rec in conn_records:
+                await conn_rec.delete_record(session)
+
     async def receive_reuse_message(
         self,
         reuse_msg: HandshakeReuse,
         receipt: MessageReceipt,
+        conn_rec: ConnRecord,
     ) -> None:
         """
         Receive and process a HandshakeReuse message under RFC 0434.
@@ -1034,67 +1047,23 @@ async def receive_reuse_message(
             or the connection does not exists
 
         """
-        try:
-            invi_msg_id = reuse_msg._thread.pthid
-            reuse_msg_id = reuse_msg._thread.thid
-            tag_filter = {}
-            post_filter = {}
-            # post_filter["state"] = "active"
-            # tag_filter["their_did"] = receipt.sender_did
-            post_filter["invitation_msg_id"] = invi_msg_id
-            conn_record = await self.find_existing_connection(
-                tag_filter=tag_filter, post_filter=post_filter
-            )
-            responder = self.profile.inject_or(BaseResponder)
-            if conn_record is not None:
-                # For ConnRecords created using did-exchange
-                reuse_accept_msg = HandshakeReuseAccept()
-                reuse_accept_msg.assign_thread_id(thid=reuse_msg_id, pthid=invi_msg_id)
-                connection_targets = await self.fetch_connection_targets(
-                    connection=conn_record
-                )
-                if responder:
-                    await responder.send(
-                        message=reuse_accept_msg,
-                        target_list=connection_targets,
-                    )
-                # This is not required as now we attaching the invitation_msg_id
-                # using original invitation [from existing connection]
-                #
-                # Delete the ConnRecord created; re-use existing connection
-                # invi_id_post_filter = {}
-                # invi_id_post_filter["invitation_msg_id"] = invi_msg_id
-                # conn_rec_to_delete = await self.find_existing_connection(
-                #     tag_filter={},
-                #     post_filter=invi_id_post_filter,
-                # )
-                # if conn_rec_to_delete is not None:
-                #     if conn_record.connection_id != conn_rec_to_delete.connection_id:
-                #         await conn_rec_to_delete.delete_record(session=self._session)
-            else:
-                conn_record = await self.find_existing_connection(
-                    tag_filter={"their_did": receipt.sender_did}, post_filter={}
-                )
-                # Problem Report is redundant in this case as with no active
-                # connection, it cannot reach the invitee any way
-                if conn_record is not None:
-                    # For ConnRecords created using RFC 0160 connections
-                    reuse_accept_msg = HandshakeReuseAccept()
-                    reuse_accept_msg.assign_thread_id(
-                        thid=reuse_msg_id, pthid=invi_msg_id
-                    )
-                    connection_targets = await self.fetch_connection_targets(
-                        connection=conn_record
-                    )
-                    if responder:
-                        await responder.send(
-                            message=reuse_accept_msg,
-                            target_list=connection_targets,
-                        )
-        except StorageNotFoundError:
-            raise OutOfBandManagerError(
-                (f"No existing ConnRecord found for OOB Invitee, {receipt.sender_did}"),
+        invi_msg_id = reuse_msg._thread.pthid
+        reuse_msg_id = reuse_msg._thread.thid
+        responder = self.profile.inject_or(BaseResponder)
+        reuse_accept_msg = HandshakeReuseAccept()
+        reuse_accept_msg.assign_thread_id(thid=reuse_msg_id, pthid=invi_msg_id)
+        connection_targets = await self.fetch_connection_targets(connection=conn_rec)
+        if responder:
+            await responder.send(
+                message=reuse_accept_msg,
+                target_list=connection_targets,
             )
+        # Update ConnRecord's invi_msg_id
+        async with self._profile.session() as session:
+            conn_rec.invitation_msg_id = invi_msg_id
+            await conn_rec.save(session, reason="Assigning new invitation_msg_id")
+        # Delete the ConnRecord created; re-use existing connection
+        await self.delete_stale_connection_by_invitation(invi_msg_id)
 
     async def receive_reuse_accepted_message(
         self,
@@ -1130,6 +1099,10 @@ async def receive_reuse_accepted_message(
                 await conn_record.metadata_set(
                     session=session, key="reuse_msg_state", value="accepted"
                 )
+                conn_record.invitation_msg_id = invi_msg_id
+                await conn_record.save(
+                    session, reason="Assigning new invitation_msg_id"
+                )
         except Exception as e:
             raise OutOfBandManagerError(
                 (