refactor code for webhook listener

Author: wip-abramson

libs/aries-basic-controller/aries_basic_controller/aries_controller.py

@@ -23,6 +23,8 @@
 from .controllers.action_menu import ActionMenuController
 from .controllers.revocation import RevocationController
 
+from .aries_webhook_listener import AriesWebhookListener
+
 import logging
 
 logger = logging.getLogger("aries_controller")
@@ -48,7 +50,7 @@ class AriesAgentController:
         Specify whether to create connecitons (default is True)
     messaging : bool
         Initialise the messaging interface (default is True)
-    multitenant : bool
+    is_multitenant : bool
         Initialise the multitenant interface (default is False)
     mediation : bool
         Initialise the mediation interface (default is False)
@@ -62,24 +64,26 @@ class AriesAgentController:
         The API key (default is None)
     tenant_jwt: str
         The tenant JW token (default is None)
+    wallet_id : str
+        The tenant wallet identifier
     """
 
     ## TODO rethink how to initialise. Too many args?
     ## is it important to let users config connections/issuer etc
-    webhook_host: str
-    webhook_port: int
     admin_url: str
+    webhook_host: str = None
+    webhook_port: int = None
     webhook_base: str = ""
     connections: bool = True
     messaging: bool = True
-    multitenant: bool = False
+    is_multitenant: bool = False
     mediation: bool = False
     issuer: bool = True
     action_menu: bool = True
     revocations: bool = True
     api_key: str = None
     tenant_jwt: str = None
-    wallet_id: str = "base"
+    wallet_id: str = None
 
 
     def __post_init__(self):
@@ -101,7 +105,9 @@ def __post_init__(self):
 
 
         self.client_session: ClientSession = ClientSession(headers=self.headers)
-    
+
+        self.webhook_listener: AriesWebhookListener = AriesWebhookListener(webhook_host=self.webhook_host, webhook_port=self.webhook_port, webhook_base=self.webhook_base, is_multitenant=self.is_multitenant)
+
         # Instantiate controllers based on the provided attributes
         if self.connections:
             self.connections = ConnectionsController(self.admin_url, self.client_session)
@@ -115,7 +121,7 @@ def __post_init__(self):
         self.server = ServerController(self.admin_url, self.client_session)
         self.oob = OOBController(self.admin_url, self.client_session)
 
-        if self.multitenant:
+        if self.is_multitenant:
             self.multitenant = MultitenancyController(self.admin_url, self.client_session)
 
         if self.mediation:
@@ -137,17 +143,29 @@ def __post_init__(self):
                 self.client_session
             )
 
+    def update_wallet_id(self, wallet_id: str):
+        """This wallet_id is used to register for webhooks specific to this sub_wallet
+
+        Args:
+        ----
+        wallet_id : str
+            The tenant wallet identifier
+        """
+        self.wallet_id = wallet_id
 
 
-    def update_tenant_jwt(self, tenant_jwt: str): 
+    def update_tenant_jwt(self, tenant_jwt: str, wallet_id: str):
         """Update the tenant JW token attribute and the header
         
         Args:
         ----
         tenant_jwt : str 
             The tenant's JW token
+        wallet_id : str
+            The tenant wallet identifier
         """
         self.tenant_jwt = tenant_jwt
+        self.update_wallet_id(wallet_id)
         self.headers.update({'Authorization': 'Bearer ' + tenant_jwt, 'content-type': "application/json"})
         self.client_session.headers.update(self.headers)
 
@@ -210,8 +228,6 @@ def register_listeners(self, listeners, defaults=True):
             print(f"Register webhooks listeners failed! {exc!r} occurred.")
             logger.warn(f"Register webhooks listeners failed! {exc!r} occurred.")
 
-
-
     def add_listener(self, listener):
         """Subscribe to a listeners for a topic
         
@@ -267,77 +283,15 @@ def remove_all_listeners(self, topic: str = None):
             pub.unsubAll(topicName=topic)
         except Exception as exc:
             print(f"Removing all webhooks listeners failed! {exc!r} occurred.")
-            logger.warn(f"Removing all webhooks listeners failed! {exc!r} occurred.")
-            
+            logger.warning(f"Removing all webhooks listeners failed! {exc!r} occurred.")
 
 
     async def listen_webhooks(self):
-        """Create a server to listen to webhooks"""
-        try:
-            app = web.Application()
-            app.add_routes([web.post(self.webhook_base + "/{wallet}/topic/{topic}/", self._receive_webhook)])
-            runner = web.AppRunner(app)
-            await runner.setup()
-            self.webhook_site = web.TCPSite(runner, self.webhook_host, self.webhook_port)
-            await self.webhook_site.start()
-        except Exception as exc:
-            print(f"Listening webhooks failed! {exc!r} occurred.")
-            logger.warn(f"Listening webhooks failed! {exc!r} occurred.")
-
-
-
-    async def _receive_webhook(self, request: ClientRequest):
-        """Helper to receive webhooks by requesting it
-        
-        Args:
-        ----
-        request : ClientRequest
-            The client request to which the corresponding webhooks shall be received
-            
-        Returns:
-        -------
-        Response:
-            A response with status 200
-        """
-        topic = request.match_info["topic"]
-        wallet = request.match_info["wallet"]
-        print("wallet", wallet)
-        try:
-            payload = await request.json()
-            await self._handle_webhook(wallet, topic, payload)
-            return web.Response(status=200)
-        except Exception as exc:
-            logger.warn(f"Receiving webhooks failed! {exc!r} occurred.")
-        
-
-
-    async def _handle_webhook(self, wallet, topic, payload):
-        """Helper handling a webhook
-        
-        Args:
-        ----
-        topic : str
-            The topic to handle webhooks for
-        payload : dict
-            A JSON-like dictionary representation of the payload
-        """
-        try:
-            pub_topic_path = f"{wallet}.{topic}"
-            print(f"Handle Webhook - {pub_topic_path}", payload)
-            logging.debug(f"Handle Webhook - {pub_topic_path}", payload)
-            pub.sendMessage(pub_topic_path, payload=payload)
-            # return web.Response(status=200)
-        except Exception as exc:
-            logger.warn(f"Handling webhooks failed! {exc!r} occurred when trying to handle this topic: {topic}")
-            
+        # self.webhook_listener: AriesWebhookListener = AriesWebhookListener(webhook_host=webhook_host, webhook_port=webhook_port, webhook_base=webhook_base, is_multitenant=is_multitenant)
 
+        if self.webhook_listener:
+            await self.webhook_listener.listen_webhooks()
 
     async def terminate(self):
-        """Terminate the controller client session and webhook listeners"""
-        try:
-            await self.client_session.close()
-            if self.webhook_site:
-                await self.webhook_site.stop()
-        except Exception as exc:
-            print(f"Terminating webhooks listener failed! {exc!r} occurred.")
-            logger.warn(f"Terminating webhooks listener failed! {exc!r} occurred.")
+        await self.client_session.close()
+        await self.webhook_listener.terminate()

libs/aries-basic-controller/aries_basic_controller/aries_webhook_listener.py

@@ -0,0 +1,109 @@
+from aiohttp import (
+    web,
+    ClientSession,
+    ClientRequest,
+)
+from dataclasses import dataclass
+from pubsub import pub
+import logging
+
+logger = logging.getLogger("aries_webhook_listener")
+
+@dataclass
+class AriesWebhookListener:
+    """The Aries Webhook Listener class
+
+    This class allows you to interact with Aries by exposing the aca-py API.
+
+    Attributes
+    ----------
+    webhook_host : str
+        The url of the webhook host
+    webhook_port : int
+        The exposed port for webhooks on the host
+    admin_url : str
+        The URL for the Admin API
+    webhook_base : str
+        The base url for webhooks (default is "")
+    wallet_id : str
+        The tenant wallet identifier
+    is_multitenant : bool
+        Initialise the multitenant interface (default is False)
+    """
+
+
+    webhook_host: str
+    webhook_port: int
+    webhook_base: str = ""
+    is_multitenant: bool = False
+
+    async def listen_webhooks(self):
+        """Create a server to listen to webhooks"""
+        try:
+            app = web.Application()
+            app.add_routes([web.post(self.webhook_base + "/topic/{topic}/", self._receive_webhook)])
+            if self.is_multitenant:
+                app.add_routes([web.post(self.webhook_base + "/{wallet_id}/topic/{topic}/", self._receive_webhook)])
+            runner = web.AppRunner(app)
+            await runner.setup()
+            self.webhook_site = web.TCPSite(runner, self.webhook_host, self.webhook_port)
+            await self.webhook_site.start()
+        except Exception as exc:
+            print(f"Listening webhooks failed! {exc!r} occurred.")
+            logger.warn(f"Listening webhooks failed! {exc!r} occurred.")
+
+    async def _receive_webhook(self, request: ClientRequest):
+        """Helper to receive webhooks by requesting it
+
+        Args:
+        ----
+        request : ClientRequest
+            The client request to which the corresponding webhooks shall be received
+
+        Returns:
+        -------
+        Response:
+            A response with status 200
+        """
+        topic = request.match_info["topic"]
+        wallet_id = None
+        if self.is_multitenant:
+            wallet_id = request.match_info["wallet_id"]
+        print("wallet", wallet_id)
+        try:
+            payload = await request.json()
+            await self._handle_webhook(wallet_id, topic, payload)
+            return web.Response(status=200)
+        except Exception as exc:
+            logger.warning(f"Receiving webhooks failed! {exc!r} occurred.")
+
+    async def _handle_webhook(self, wallet_id, topic, payload):
+        """Helper handling a webhook
+
+        Args:
+        ----
+        wallet_id: str
+            The identifier for the wallet the webhook is for
+        topic : str
+            The topic to handle webhooks for
+        payload : dict
+            A JSON-like dictionary representation of the payload
+        """
+        try:
+            pub_topic_path = topic
+            if wallet_id:
+                pub_topic_path = f"{wallet_id}.{topic}"
+            logging.debug(f"Handle Webhook - {pub_topic_path}", payload)
+            pub.sendMessage(pub_topic_path, payload=payload)
+            # return web.Response(status=200)
+        except Exception as exc:
+            logger.warn(f"Handling webhooks failed! {exc!r} occurred when trying to handle this topic: {topic}")
+
+    async def terminate(self):
+        """Terminate the controller client session and webhook listeners"""
+        try:
+            if self.webhook_site:
+                await self.webhook_site.stop()
+        except Exception as exc:
+            print(f"Terminating webhooks listener failed! {exc!r} occurred.")
+            logger.warn(f"Terminating webhooks listener failed! {exc!r} occurred.")

scripts/startup.sh

@@ -1,4 +1,4 @@
-if [[ -n "${MULTITENANT_TUTORIAL}" ]]; then
+If [ “$MULTITENANT_TUTORIAL” == “MULTITENANT_TUTORIAL” ]; then
     echo $MULTITENANT_TUTORIAL
     python3 ./scripts/parse_yml_env_variables.py -c ./configuration/aries-args-multitenant.yaml;aca-py start --arg-file /tmp/agent_conf.yml
 else

tutorials/4. Multitenancy/docker-compose.yml

@@ -99,7 +99,7 @@ services:
       - AGENT_NAME=${MULTITENANT_AGENT_NAME}
       - ADMIN_PORT=${MULTITENANT_ADMIN_PORT}
       - HTTP_PORT=${MULTITENANT_HTTP_PORT}
-      - ACAPY_WEBHOOK_URL=${MULTITENANT_WEBHOOK_URL}/base
+      - ACAPY_WEBHOOK_URL=${MULTITENANT_WEBHOOK_URL}
       - AGENT_ENDPOINT=${MULTITENANT_AGENT_ENDPOINT}
       - WALLET_SEED=${MULTITENANT_WALLET_SEED}
       - WALLET_NAME=${MULTITENANT_WALLET_NAME}

tutorials/4. Multitenancy/notebooks/mediator/Configure Mediator.ipynb

@@ -109,14 +109,14 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "base.connections\n"
+      "Subscribing too: connections\n"
      ]
     }
    ],
    "source": [
     "\n",
     "loop = asyncio.get_event_loop()\n",
-    "loop.create_task(agent_controller.listen_webhooks())\n",
+    "loop.create_task(agent_controller.webhook_listener.listen_webhooks())\n",
     "\n",
     "def connection_handler(payload):\n",
     "    print(\"Connection Handler Called\")\n",
@@ -154,9 +154,29 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "Connection ID aa9a800e-9e65-4070-98d8-4f56971d6292\n",
+      "Connection ID a347eb87-8018-4d3e-a051-a945fcf52ff1\n",
       "Invitation\n",
-      "{'@type': 'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/connections/1.0/invitation', '@id': '61520ae1-80f8-473c-ae56-31472c252ceb', 'label': 'MEDIATOR', 'recipientKeys': ['EZ1hdmdntvBTMDW7SHFmQrL28hSV2U3Ctj83ERCzL34C'], 'serviceEndpoint': 'https://7976bda13d4b.ngrok.io'}\n"
+      "{'@type': 'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/connections/1.0/invitation', '@id': 'e08a7e80-d96b-43ba-9ea8-368e5e61eeb6', 'serviceEndpoint': 'https://7eceb3b5d5a4.ngrok.io', 'label': 'MEDIATOR', 'recipientKeys': ['69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU']}\n",
+      "wallet None\n",
+      "Handle Webhook - connections {'connection_id': 'a347eb87-8018-4d3e-a051-a945fcf52ff1', 'accept': 'auto', 'invitation_key': '69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU', 'created_at': '2021-03-31 17:42:19.462303Z', 'invitation_mode': 'multi', 'their_role': 'invitee', 'routing_state': 'none', 'state': 'invitation', 'updated_at': '2021-03-31 17:42:19.462303Z', 'rfc23_state': 'invitation-sent'}\n",
+      "Connection Handler Called\n",
+      "Connection a347eb87-8018-4d3e-a051-a945fcf52ff1 in State invitation\n",
+      "wallet None\n",
+      "Handle Webhook - connections {'connection_id': '49431565-9d12-42e9-8341-8699834fc785', 'accept': 'auto', 'invitation_key': '69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU', 'created_at': '2021-03-31 17:43:17.833726Z', 'invitation_mode': 'once', 'their_role': 'invitee', 'routing_state': 'none', 'state': 'invitation', 'updated_at': '2021-03-31 17:43:17.833726Z', 'my_did': 'ArH29nQAn4GYa3RCTLJehD', 'rfc23_state': 'invitation-sent'}\n",
+      "Connection Handler Called\n",
+      "Connection 49431565-9d12-42e9-8341-8699834fc785 in State invitation\n",
+      "wallet None\n",
+      "Handle Webhook - connections {'connection_id': '49431565-9d12-42e9-8341-8699834fc785', 'accept': 'auto', 'invitation_key': '69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU', 'created_at': '2021-03-31 17:43:17.833726Z', 'their_label': 'Alice', 'invitation_mode': 'once', 'their_role': 'invitee', 'routing_state': 'none', 'state': 'request', 'updated_at': '2021-03-31 17:43:17.851433Z', 'their_did': '9RUfRQwTJ68tZeuE38HXRj', 'my_did': 'ArH29nQAn4GYa3RCTLJehD', 'rfc23_state': 'request-received'}\n",
+      "Connection Handler Called\n",
+      "Connection 49431565-9d12-42e9-8341-8699834fc785 in State request\n",
+      "wallet None\n",
+      "Handle Webhook - connections {'connection_id': '49431565-9d12-42e9-8341-8699834fc785', 'accept': 'auto', 'invitation_key': '69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU', 'created_at': '2021-03-31 17:43:17.833726Z', 'their_label': 'Alice', 'invitation_mode': 'once', 'their_role': 'invitee', 'routing_state': 'none', 'state': 'response', 'updated_at': '2021-03-31 17:43:17.866994Z', 'their_did': '9RUfRQwTJ68tZeuE38HXRj', 'my_did': 'ArH29nQAn4GYa3RCTLJehD', 'rfc23_state': 'response-sent'}\n",
+      "Connection Handler Called\n",
+      "Connection 49431565-9d12-42e9-8341-8699834fc785 in State response\n",
+      "wallet None\n",
+      "Handle Webhook - connections {'connection_id': '49431565-9d12-42e9-8341-8699834fc785', 'accept': 'auto', 'invitation_key': '69hmbswFgse281vL3Z7eAzv8mc6m5RwCYuTgZxhGk4GU', 'created_at': '2021-03-31 17:43:17.833726Z', 'their_label': 'Alice', 'invitation_mode': 'once', 'their_role': 'invitee', 'routing_state': 'none', 'state': 'active', 'updated_at': '2021-03-31 17:43:18.488996Z', 'their_did': '9RUfRQwTJ68tZeuE38HXRj', 'my_did': 'ArH29nQAn4GYa3RCTLJehD', 'rfc23_state': 'completed'}\n",
+      "Connection Handler Called\n",
+      "Connection 49431565-9d12-42e9-8341-8699834fc785 in State active\n"
      ]
     }
    ],
@@ -202,7 +222,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 7,
+   "execution_count": 6,
    "metadata": {
     "pycharm": {
      "name": "#%%\n"
@@ -214,8 +234,8 @@
      "output_type": "stream",
      "text": [
       "Mediation Record\n",
-      "connection_id ab02b489-033d-4378-8202-ad31751d64f0\n",
-      "State granted\n"
+      "connection_id 49431565-9d12-42e9-8341-8699834fc785\n",
+      "State request\n"
      ]
     }
    ],