DIDComm V2 Interopathon

acapy_agent/config/default_context.py

@@ -8,6 +8,7 @@
 from ..core.plugin_registry import PluginRegistry
 from ..core.profile import ProfileManager, ProfileManagerProvider
 from ..core.protocol_registry import ProtocolRegistry
+from ..didcomm_v2.protocol_registry import V2ProtocolRegistry
 from ..protocols.actionmenu.v1_0.base_service import BaseMenuService
 from ..protocols.actionmenu.v1_0.driver_service import DriverMenuService
 from ..protocols.introduction.v0_1.base_service import BaseIntroductionService
@@ -45,6 +46,7 @@ async def build_context(self) -> InjectionContext:
 
         # Global protocol registry
         context.injector.bind_instance(ProtocolRegistry, ProtocolRegistry())
+        context.injector.bind_instance(V2ProtocolRegistry, V2ProtocolRegistry())
 
         # Global goal code registry
         context.injector.bind_instance(GoalCodeRegistry, GoalCodeRegistry())
@@ -129,6 +131,7 @@ async def load_plugins(self, context: InjectionContext):
         # Register standard protocol plugins
         if not self.settings.get("transport.disabled"):
             plugin_registry.register_package("acapy_agent.protocols")
+            plugin_registry.register_package("acapy_agent.protocols_v2")
 
         # Currently providing admin routes only
         plugin_registry.register_plugin("acapy_agent.holder")

acapy_agent/core/dispatcher.py

@@ -12,9 +12,11 @@
 from typing import Callable, Coroutine, Optional, Union
 
 from aiohttp.web import HTTPException
+from didcomm_messaging import DIDCommMessaging, RoutingService
 
 from ..connections.base_manager import BaseConnectionManager
 from ..connections.models.conn_record import ConnRecord
+from ..connections.models.connection_target import ConnectionTarget
 from ..core.profile import Profile
 from ..messaging.agent_message import AgentMessage
 from ..messaging.base_message import BaseMessage, DIDCommVersion
@@ -34,6 +36,7 @@
 from ..utils.tracing import get_timer, trace_event
 from .error import ProtocolMinorVersionNotSupported
 from .protocol_registry import ProtocolRegistry
+from ..didcomm_v2.protocol_registry import V2ProtocolRegistry
 
 
 class ProblemReportParseError(MessageParseError):
@@ -137,31 +140,176 @@ async def handle_v2_message(
     ):
         """Handle a DIDComm V2 message."""
 
+        error_result = None
+        message = None
+
+        try:
+            message = await self.make_v2_message(profile, inbound_message.payload)
+        except ProblemReportParseError:
+            pass  # avoid problem report recursion
+        except MessageParseError as e:
+            self.logger.error(f"Message parsing failed: {str(e)}, sending problem report")
+            error_result = ProblemReport(
+                description={
+                    "en": str(e),
+                    "code": "message-parse-failure",
+                }
+            )
+            if inbound_message.receipt.thread_id:
+                error_result.assign_thread_id(inbound_message.receipt.thread_id)
+
+        session = await profile.session()
+        ctx = session
+        messaging = ctx.inject(DIDCommMessaging)
+        routing_service = ctx.inject(RoutingService)
+        frm = inbound_message.payload.get("from")
+        services = await routing_service._resolve_services(messaging.resolver, frm)
+        chain = [
+            {
+                "did": frm,
+                "service": services,
+            }
+        ]
+
+        # Loop through service DIDs until we run out of DIDs to forward to
+        to_did = services[0].service_endpoint.uri
+        found_forwardable_service = await routing_service.is_forwardable_service(
+            messaging.resolver, services[0]
+        )
+        while found_forwardable_service:
+            services = await routing_service._resolve_services(messaging.resolver, to_did)
+            if services:
+                chain.append(
+                    {
+                        "did": to_did,
+                        "service": services,
+                    }
+                )
+                to_did = services[0].service_endpoint.uri
+            found_forwardable_service = (
+                await routing_service.is_forwardable_service(messaging.resolver, services[0])
+                if services
+                else False
+            )
+        reply_destination = [
+            ConnectionTarget(
+                did=inbound_message.receipt.sender_verkey,
+                endpoint=service.service_endpoint.uri,
+                recipient_keys=[inbound_message.receipt.sender_verkey],
+                sender_key=inbound_message.receipt.recipient_verkey,
+            )
+            for service in chain[-1]["service"]
+        ]
+
         # send a DCV2 Problem Report here for testing, and to punt procotol handling down
         # the road a bit
         context = RequestContext(profile)
+        context.message = message
         context.message_receipt = inbound_message.receipt
         responder = DispatcherResponder(
             context,
             inbound_message,
             send_outbound,
             reply_session_id=inbound_message.session_id,
             reply_to_verkey=inbound_message.receipt.sender_verkey,
+            target_list=reply_destination
         )
 
         context.injector.bind_instance(BaseResponder, responder)
-        error_result = V2AgentMessage(
-            message={
-                "type": "https://didcomm.org/report-problem/2.0/problem-report",
-                "body": {
-                    "comment": "No Handlers Found",
-                    "code": "e.p.msg.not-found",
-                },
-            }
-        )
-        if inbound_message.receipt.thread_id:
-            error_result.message["pthid"] = inbound_message.receipt.thread_id
-        await responder.send_reply(error_result)
+        if not message:
+            error_result = V2AgentMessage(
+                message={
+                    "type": "https://didcomm.org/report-problem/2.0/problem-report",
+                    "body": {
+                        "comment": "No Handlers Found",
+                        "code": "e.p.msg.not-found",
+                    },
+                }
+            )
+            if inbound_message.receipt.thread_id:
+                error_result.message["pthid"] = inbound_message.receipt.thread_id
+
+        # # When processing oob attach message we supply the connection id
+        # # associated with the inbound message
+        # if inbound_message.connection_id:
+        #     async with self.profile.session() as session:
+        #         connection = await ConnRecord.retrieve_by_id(
+        #             session, inbound_message.connection_id
+        #         )
+        # else:
+        #     connection_mgr = BaseConnectionManager(profile)
+        #     connection = await connection_mgr.find_inbound_connection(
+        #         inbound_message.receipt
+        #     )
+        #     del connection_mgr
+
+        # if connection:
+        #     inbound_message.connection_id = connection.connection_id
+
+        # context.connection_ready = connection and connection.is_ready
+        # context.connection_record = connection
+        # responder.connection_id = connection and connection.connection_id
+
+        if error_result:
+            await responder.send_reply(error_result)
+        elif context.message:
+            context.injector.bind_instance(BaseResponder, responder)
+
+            handler = context.message
+            if self.collector:
+                handler = self.collector.wrap_coro(handler, [handler.__qualname__])
+            await handler()(context, responder, payload=inbound_message.payload)
+
+    async def make_v2_message(self, profile: Profile, parsed_msg: dict) -> BaseMessage:
+        """Deserialize a message dict into the appropriate message instance.
+
+        Given a dict describing a message, this method
+        returns an instance of the related message class.
+
+        Args:
+            parsed_msg: The parsed message
+            profile: Profile
+
+        Returns:
+            An instance of the corresponding message class for this message
+
+        Raises:
+            MessageParseError: If the message doesn't specify @type
+            MessageParseError: If there is no message class registered to handle
+            the given type
+
+        """
+        if not isinstance(parsed_msg, dict):
+            raise MessageParseError("Expected a JSON object")
+        message_type = parsed_msg.get("type")
+
+        if not message_type:
+            raise MessageParseError("Message does not contain 'type' parameter")
+
+        registry: V2ProtocolRegistry = self.profile.inject(V2ProtocolRegistry)
+        try:
+            #message_cls = registry.resolve_message_class(message_type)
+            #if isinstance(message_cls, DeferLoad):
+            #    message_cls = message_cls.resolved
+            message_cls = registry.protocols_matching_query(message_type)
+        except ProtocolMinorVersionNotSupported as e:
+            raise MessageParseError(f"Problem parsing message type. {e}")
+
+        if not message_cls:
+            raise MessageParseError(f"Unrecognized message type {message_type}")
+
+        try:
+            #instance = message_cls[0] #message_cls.deserialize(parsed_msg)
+            instance = registry.handlers[message_cls[0]]
+            if isinstance(instance, DeferLoad):
+                instance = instance.resolved
+        except BaseModelError as e:
+            if "/problem-report" in message_type:
+                raise ProblemReportParseError("Error parsing problem report message")
+            raise MessageParseError(f"Error deserializing message: {e}") from e
+
+        return instance
+
 
     async def handle_v1_message(
         self,

acapy_agent/core/plugin_registry.py

@@ -11,6 +11,7 @@
 from .error import ProtocolDefinitionValidationError
 from .goal_code_registry import GoalCodeRegistry
 from .protocol_registry import ProtocolRegistry
+from ..didcomm_v2.protocol_registry import V2ProtocolRegistry
 
 LOGGER = logging.getLogger(__name__)
 
@@ -218,8 +219,14 @@ async def load_protocol_version(
         version_definition: Optional[dict] = None,
     ):
         """Load a particular protocol version."""
+        v2_protocol_registry = context.inject(V2ProtocolRegistry)
         protocol_registry = context.inject(ProtocolRegistry)
         goal_code_registry = context.inject(GoalCodeRegistry)
+        if hasattr(mod, "HANDLERS"):
+            for message_type, handler in mod.HANDLERS:
+                v2_protocol_registry.register_handler(
+                    message_type, handler
+                )
         if hasattr(mod, "MESSAGE_TYPES"):
             protocol_registry.register_message_types(
                 mod.MESSAGE_TYPES, version_definition=version_definition

acapy_agent/didcomm_v2/protocol_registry.py

@@ -0,0 +1,43 @@
+"""Registry for DIDComm V2 Protocols."""
+
+from ..utils.classloader import DeferLoad
+from typing import Coroutine, Dict, Sequence, Union
+
+
+class V2ProtocolRegistry:
+    """DIDComm V2 Protocols."""
+
+    def __init__(self):
+        """Initialize a V2ProtocolRegistry instance."""
+        self._type_to_message_handler: Dict[str, Coroutine] = {}
+
+    @property
+    def handlers(self) -> Dict[str, Coroutine]:
+        """Accessor for a list of all message protocols."""
+        return self._type_to_message_handler
+
+    @property
+    def protocols(self) -> Sequence[str]:
+        """Accessor for a list of all message protocols."""
+        return [str(key) for key in self._type_to_message_handler.keys()]
+
+    def protocols_matching_query(self, query: str) -> Sequence[str]:
+        """Return a list of message protocols matching a query string."""
+        all_types = self.protocols
+        result = None
+
+        if query == "*" or query is None:
+            result = all_types
+        elif query:
+            if query.endswith("*"):
+                match = query[:-1]
+                result = tuple(k for k in all_types if k.startswith(match))
+            elif query in all_types:
+                result = (query,)
+        return result or ()
+
+    def register_handler(self, message_type: str, handler: Union[Coroutine, str]):
+        """Register a new message type to handler association."""
+        if isinstance(handler, str):
+                handler = DeferLoad(handler)
+        self._type_to_message_handler[message_type] = handler

acapy_agent/messaging/responder.py

@@ -45,11 +45,14 @@ def __init__(
         connection_id: Optional[str] = None,
         reply_session_id: Optional[str] = None,
         reply_to_verkey: Optional[str] = None,
+        target: Optional[ConnectionTarget] = None,
+        target_list: Sequence[ConnectionTarget] = None,
     ):
         """Initialize a base responder."""
         self.connection_id = connection_id
         self.reply_session_id = reply_session_id
         self.reply_to_verkey = reply_to_verkey
+        self.target_list = target_list
 
     async def create_outbound(
         self,
@@ -133,7 +136,7 @@ async def send_reply(
             reply_session_id=self.reply_session_id,
             reply_to_verkey=self.reply_to_verkey,
             target=target,
-            target_list=target_list,
+            target_list=target_list or self.target_list,
         )
         if isinstance(message, BaseMessage):
             msg_type = message._message_type

acapy_agent/protocols_v2/basicmessage/definition.py

@@ -0,0 +1,10 @@
+"""Version definitions for this protocol."""
+
+versions = [
+    {
+        "major_version": 1,
+        "minimum_minor_version": 0,
+        "current_minor_version": 0,
+        "path": "v1_0",
+    }
+]

acapy_agent/protocols_v2/basicmessage/v1_0/message_types.py

@@ -0,0 +1,46 @@
+"""Message type identifiers for Trust Pings."""
+
+#from ...didcomm_prefix import DIDCommPrefix
+import logging
+from ....messaging.v2_agent_message import V2AgentMessage
+from ....connections.models.connection_target import ConnectionTarget
+from didcomm_messaging import DIDCommMessaging, RoutingService
+
+SPEC_URI = (
+    "https://didcomm.org/basicmessage/2.0/message"
+)
+
+# Message types
+BASIC_MESSAGE = "https://didcomm.org/basicmessage/2.0/message"
+
+PROTOCOL_PACKAGE = "acapy_agent.protocols_v2.basicmessage.v1_0"
+
+class basic_message:
+    async def __call__(self, *args, **kwargs):
+        await self.handle(*args, **kwargs)
+    @staticmethod
+    async def handle(context, responder, payload):
+        logger = logging.getLogger(__name__)
+        their_did = context.message_receipt.sender_verkey.split('#')[0]
+        our_did = context.message_receipt.recipient_verkey.split('#')[0]
+        error_result = V2AgentMessage(
+            message={
+                "type": BASIC_MESSAGE,
+                "body": {
+                    "content": "Hello from acapy",
+                },
+                "to": [their_did],
+                "from": our_did,
+                "lang": "en",
+            }
+        )
+        await responder.send_reply(error_result)
+
+
+HANDLERS = {
+    BASIC_MESSAGE: f"{PROTOCOL_PACKAGE}.message_types.basic_message",
+}.items()
+
+MESSAGE_TYPES = {
+    BASIC_MESSAGE: f"{PROTOCOL_PACKAGE}.message_types.basic_message",
+}