fix: self-issuance/self-connection OOB and DID exchange failures

acapy_agent/core/dispatcher.py

@@ -221,9 +221,11 @@ async def handle_v1_message(
         context.injector.bind_instance(BaseResponder, responder)
 
         # When processing oob attach message we supply the connection id
-        # associated with the inbound message
+        # associated with the inbound message. The connection must be looked up
+        # in the profile the message belongs to (e.g. a multitenant subwallet),
+        # not the dispatcher's root profile.
         if inbound_message.connection_id:
-            async with self.profile.session() as session:
+            async with profile.session() as session:
                 connection = await ConnRecord.retrieve_by_id(
                     session, inbound_message.connection_id
                 )

acapy_agent/core/oob_processor.py

@@ -10,10 +10,13 @@
 from ..messaging.decorators.service_decorator import ServiceDecorator
 from ..messaging.request_context import RequestContext
 from ..protocols.didcomm_prefix import DIDCommPrefix
+from ..protocols.didexchange.v1_0.message_types import ARIES_PROTOCOL as DIDEX_1_1
+from ..protocols.didexchange.v1_0.message_types import DIDEX_1_0
 from ..protocols.issue_credential.v2_0.message_types import CRED_20_OFFER
+from ..protocols.out_of_band.v1_0.message_types import MESSAGE_REUSE
 from ..protocols.out_of_band.v1_0.models.oob_record import OobRecord
 from ..protocols.present_proof.v2_0.message_types import PRES_20_REQUEST
-from ..storage.error import StorageNotFoundError
+from ..storage.error import StorageDuplicateError, StorageNotFoundError
 from ..transport.inbound.message import InboundMessage
 from ..transport.outbound.message import OutboundMessage
 from ..transport.wire_format import JsonWireFormat
@@ -22,6 +25,30 @@
 
 LOGGER = logging.getLogger(__name__)
 
+# Message types that, per the OOB/DID-exchange protocols, are always sent by the
+# *receiver* of an invitation back to the *sender* (e.g. a connection request in
+# response to a handshake, or a request to reuse an existing connection). These are
+# processed by the party holding the role=sender OobRecord for the invitation.
+_SENDER_ROLE_MESSAGE_TYPES = {
+    f"{DIDEX_1_0}/request",
+    f"{DIDEX_1_1}/request",
+    MESSAGE_REUSE,
+}
+
+# Message types that are always sent by the *sender* of an invitation to its
+# *receiver* (the attachable messages). An inbound message of one of these types can
+# therefore never belong to our own role=sender OobRecord: it can only match our
+# role=receiver record. This matters for self-connections (one wallet playing both
+# roles for one invitation): once the receiver-role record has been consumed, an
+# inbound attached offer/present-request must not consume the sender-role record,
+# which is still needed to process the receiver's reply (e.g. request-credential).
+_RECEIVER_ROLE_MESSAGE_TYPES = {
+    CRED_20_OFFER,
+    PRES_20_REQUEST,
+    "issue-credential/1.0/offer-credential",
+    "present-proof/1.0/request-presentation",
+}
+
 
 class OobMessageProcessorError(BaseError):
     """Base error for OobMessageProcessor."""
@@ -137,6 +164,105 @@ async def find_oob_record_for_inbound_message(
                 except StorageNotFoundError:
                     # Fine if record is not found
                     pass
+                except StorageDuplicateError:
+                    # Multiple oob records share this invi_msg_id. This happens when
+                    # an agent creates an OOB invitation and then receives that same
+                    # invitation back into the same wallet (e.g. self-issuance): one
+                    # OobRecord has role=sender (from create-invitation) and another
+                    # has role=receiver (from receive-invitation).
+                    #
+                    # Disambiguate primarily using the recipient verkey the inbound
+                    # message was actually delivered to: each side of a connectionless
+                    # exchange uses its own key (OobRecord.our_recipient_key), so the
+                    # recipient_verkey on the envelope tells us unambiguously which of
+                    # the two records this message belongs to, regardless of protocol
+                    # or message type (credential offers/requests, presentation
+                    # requests/presentations/acks, did-exchange requests, handshake
+                    # reuse, ...).
+                    if context.message_receipt.recipient_verkey:
+                        try:
+                            oob_record = await OobRecord.retrieve_by_tag_filter(
+                                session,
+                                {
+                                    "invi_msg_id": (
+                                        context.message_receipt.parent_thread_id
+                                    ),
+                                    "our_recipient_key": (
+                                        context.message_receipt.recipient_verkey
+                                    ),
+                                },
+                            )
+                            LOGGER.debug(
+                                "Multiple OOB records found for invi_msg_id %s; "
+                                "disambiguated using recipient verkey %s to oob "
+                                "record %s",
+                                context.message_receipt.parent_thread_id,
+                                context.message_receipt.recipient_verkey,
+                                oob_record.oob_id,
+                            )
+                        except (StorageNotFoundError, StorageDuplicateError):
+                            # Fine if record is not found (or, unexpectedly, still
+                            # ambiguous); fall back to the message-type heuristic
+                            oob_record = None
+
+                    # Fall back to disambiguating by message type. This is needed
+                    # when the message was delivered without going through the wire
+                    # format (e.g. the initial oob attachment message, which is
+                    # handled locally and therefore has no recipient_verkey). Some
+                    # message types are always sent by the receiver of an invitation
+                    # back to the sender, so we must be looking for our role=sender
+                    # record to process them; all other message types (e.g. attached
+                    # protocol messages like credential offers) are sent by the
+                    # sender to the receiver, so we must be looking for our
+                    # role=receiver record.
+                    if not oob_record:
+                        expected_role = (
+                            OobRecord.ROLE_SENDER
+                            if DIDCommPrefix.unqualify(message_type)
+                            in _SENDER_ROLE_MESSAGE_TYPES
+                            else OobRecord.ROLE_RECEIVER
+                        )
+                        LOGGER.debug(
+                            "Multiple OOB records found for invi_msg_id %s; "
+                            "disambiguating using role %s for message type %s",
+                            context.message_receipt.parent_thread_id,
+                            expected_role,
+                            message_type,
+                        )
+                        try:
+                            oob_record = await OobRecord.retrieve_by_tag_filter(
+                                session,
+                                {
+                                    "invi_msg_id": (
+                                        context.message_receipt.parent_thread_id
+                                    )
+                                },
+                                {"role": expected_role},
+                            )
+                        except StorageNotFoundError:
+                            # Fine if record is not found
+                            pass
+
+                # An attached message (offer, presentation request, ...) always
+                # flows sender -> receiver, so it can never belong to our own
+                # role=sender OobRecord. This happens in a self-connection when
+                # the receiver-role record has already been consumed (deleted
+                # after the handshake): the attached offer must not consume the
+                # sender-role record here, or the later reply from the receiver
+                # (e.g. request-credential) finds no OobRecord at all.
+                if (
+                    oob_record
+                    and oob_record.role == OobRecord.ROLE_SENDER
+                    and DIDCommPrefix.unqualify(message_type)
+                    in _RECEIVER_ROLE_MESSAGE_TYPES
+                ):
+                    LOGGER.debug(
+                        "Ignoring role=sender OOB record %s for inbound "
+                        "receiver-bound message type %s",
+                        oob_record.oob_id,
+                        message_type,
+                    )
+                    oob_record = None
             # Otherwise try to find it using the attach thread id. This is only needed
             # for connectionless exchanges where every handler needs the context of the
             # oob record for verification. We could attach the oob_record to all messages,
@@ -208,17 +334,46 @@ async def find_oob_record_for_inbound_message(
                 oob_record.invitation.requests_attach
                 and oob_record.state == OobRecord.STATE_AWAIT_RESPONSE
             ):
-                LOGGER.debug(
-                    f"Removing stale connection {oob_record.connection_id} due "
-                    "to connection reuse"
-                )
-                # Remove stale connection due to connection reuse
+                # Remove stale connection due to connection reuse. But first check
+                # whether the "old" connection is actually the mirror-image
+                # ConnRecord of a self-connection: when an agent's own wallet plays
+                # both the inviter and invitee roles for one invitation (e.g.
+                # self-issuance), each role gets its own ConnRecord, and both DIDs
+                # are reciprocal (old.my_did == new.their_did and vice versa). That
+                # "old" record is not stale -- it is still awaiting its own
+                # DIDXComplete -- so deleting it here causes accept_complete to
+                # fail with "No corresponding connection request found".
+                old_conn_record = None
                 if oob_record.connection_id:
                     async with context.profile.session() as session:
-                        old_conn_record = await ConnRecord.retrieve_by_id(
-                            session, oob_record.connection_id
-                        )
+                        try:
+                            old_conn_record = await ConnRecord.retrieve_by_id(
+                                session, oob_record.connection_id
+                            )
+                        except StorageNotFoundError:
+                            old_conn_record = None
+
+                is_self_connection_mirror = bool(
+                    old_conn_record
+                    and old_conn_record.my_did
+                    and old_conn_record.their_did
+                    and old_conn_record.my_did == context.connection_record.their_did
+                    and old_conn_record.their_did == context.connection_record.my_did
+                )
+
+                if old_conn_record and not is_self_connection_mirror:
+                    LOGGER.debug(
+                        f"Removing stale connection {oob_record.connection_id} due "
+                        "to connection reuse"
+                    )
+                    async with context.profile.session() as session:
                         await old_conn_record.delete_record(session)
+                elif is_self_connection_mirror:
+                    LOGGER.debug(
+                        f"Not removing connection {oob_record.connection_id}: it is "
+                        "the other role's own ConnRecord for this self-connection "
+                        "and is still awaiting its own handshake completion"
+                    )
 
                 oob_record.connection_id = context.connection_record.connection_id
 

acapy_agent/core/tests/test_dispatcher.py

@@ -128,6 +128,57 @@ async def test_dispatch(self):
                 handler_mock.call_args[0][2], test_module.DispatcherResponder
             )
 
+    async def test_dispatch_oob_attach_connection_from_message_profile(self):
+        """Regression test: connection lookup must use the inbound message's profile.
+
+        For OOB attached messages the inbound message carries a connection_id.
+        In a multitenant agent the dispatcher is constructed with the root
+        profile, while messages are dispatched with the recipient subwallet's
+        profile: the ConnRecord only exists in the latter. Looking it up in the
+        dispatcher's own (root) profile raised StorageNotFoundError and the
+        attached message was never handled.
+        """
+        from ...connections.models.conn_record import ConnRecord
+
+        root_profile = self.profile
+        registry = root_profile.inject(ProtocolRegistry)
+        registry.register_message_types(
+            {
+                pfx.qualify(StubAgentMessage.Meta.message_type): StubAgentMessage
+                for pfx in DIDCommPrefix
+            }
+        )
+
+        # Separate profile (with its own storage) standing in for a subwallet
+        tenant_profile = await create_test_profile()
+
+        conn_rec = ConnRecord(
+            state=ConnRecord.State.COMPLETED.rfc23,
+            their_role=ConnRecord.Role.REQUESTER.rfc23,
+        )
+        async with tenant_profile.session() as session:
+            await conn_rec.save(session)
+
+        dispatcher = test_module.Dispatcher(root_profile)
+        await dispatcher.setup()
+        rcv = Receiver()
+        message = {
+            "@type": DIDCommPrefix.qualify_current(StubAgentMessage.Meta.message_type)
+        }
+        inbound = make_inbound(message)
+        inbound.connection_id = conn_rec.connection_id
+
+        with mock.patch.object(
+            StubAgentMessageHandler, "handle", autospec=True
+        ) as handler_mock:
+            await dispatcher.queue_message(tenant_profile, inbound, rcv.send)
+            await dispatcher.task_queue
+
+            handler_mock.assert_awaited_once()
+            context = handler_mock.call_args[0][1]
+            assert context.connection_record is not None
+            assert context.connection_record.connection_id == conn_rec.connection_id
+
     async def test_dispatch_versioned_message(self):
         profile = self.profile
         registry = profile.inject(ProtocolRegistry)