Inspection middleware with jsonpickle

libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py

@@ -8,8 +8,9 @@
 import asyncio
 import inspect
 from datetime import datetime
-from typing import Coroutine, Dict, List, Callable, Union
+from typing import Awaitable, Coroutine, Dict, List, Callable, Union
 from copy import copy
+from threading import Lock
 from botbuilder.schema import (
     ActivityTypes,
     Activity,
@@ -56,8 +57,8 @@ class TestAdapter(BotAdapter, UserTokenProvider):
     def __init__(
         self,
         logic: Coroutine = None,
-        conversation: ConversationReference = None,
-        send_trace_activity: bool = False,
+        template: Activity = None,
+        send_trace_activities: bool = False,
     ):  # pylint: disable=unused-argument
         """
         Creates a new TestAdapter instance.
@@ -69,21 +70,42 @@ def __init__(
         self._next_id: int = 0
         self._user_tokens: List[UserToken] = []
         self._magic_codes: List[TokenMagicCode] = []
+        self._conversation_lock = Lock()
         self.activity_buffer: List[Activity] = []
         self.updated_activities: List[Activity] = []
         self.deleted_activities: List[ConversationReference] = []
+        self.send_trace_activities = send_trace_activities
 
-        self.template: Activity = Activity(
+        self.template = template or Activity(
             channel_id="test",
             service_url="https://test.com",
             from_property=ChannelAccount(id="User1", name="user"),
             recipient=ChannelAccount(id="bot", name="Bot"),
             conversation=ConversationAccount(id="Convo1"),
         )
-        if self.template is not None:
-            self.template.service_url = self.template.service_url
-            self.template.conversation = self.template.conversation
-            self.template.channel_id = self.template.channel_id
+
+    async def process_activity(
+        self, activity: Activity, logic: Callable[[TurnContext], Awaitable]
+    ):
+        self._conversation_lock.acquire()
+        try:
+            # ready for next reply
+            if activity.type is None:
+                activity.type = ActivityTypes.message
+
+            activity.channel_id = self.template.channel_id
+            activity.from_property = self.template.from_property
+            activity.recipient = self.template.recipient
+            activity.conversation = self.template.conversation
+            activity.service_url = self.template.service_url
+
+            activity.id = str((self._next_id))
+            self._next_id += 1
+        finally:
+            self._conversation_lock.release()
+
+        activity.timestamp = activity.timestamp or datetime.utcnow()
+        await self.run_pipeline(TurnContext(self, activity), logic)
 
     async def send_activities(self, context, activities: List[Activity]):
         """
@@ -99,12 +121,11 @@ def id_mapper(activity):
             self._next_id += 1
             return ResourceResponse(id=str(self._next_id))
 
-        # TODO This if-else code is temporary until the BotAdapter and Bot/TurnContext are revamped.
-        if isinstance(activities, list):
-            responses = [id_mapper(activity) for activity in activities]
-        else:
-            responses = [id_mapper(activities)]
-        return responses
+        return [
+            id_mapper(activity)
+            for activity in activities
+            if self.send_trace_activities or activity.type != "trace"
+        ]
 
     async def delete_activity(self, context, reference: ConversationReference):
         """

libraries/botbuilder-core/botbuilder/core/bot_state.py

@@ -2,7 +2,8 @@
 # Licensed under the MIT License.
 
 from abc import abstractmethod
-from typing import Callable, Dict
+from copy import deepcopy
+from typing import Callable, Dict, Union
 from botbuilder.core.state_property_accessor import StatePropertyAccessor
 from .turn_context import TurnContext
 from .storage import Storage
@@ -186,17 +187,23 @@ async def delete(self, turn_context: TurnContext) -> None:
         await self._bot_state.delete_property_value(turn_context, self._name)
 
     async def get(
-        self, turn_context: TurnContext, default_value_factory: Callable = None
+        self,
+        turn_context: TurnContext,
+        default_value_or_factory: Union[Callable, object] = None,
     ) -> object:
         await self._bot_state.load(turn_context, False)
         try:
             result = await self._bot_state.get_property_value(turn_context, self._name)
             return result
         except:
             # ask for default value from factory
-            if not default_value_factory:
+            if not default_value_or_factory:
                 return None
-            result = default_value_factory()
+            result = (
+                default_value_or_factory()
+                if callable(default_value_or_factory)
+                else deepcopy(default_value_or_factory)
+            )
             # save default value for any further calls
             await self.set(turn_context, result)
             return result

libraries/botbuilder-core/botbuilder/core/inspection/__init__.py

@@ -0,0 +1,11 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+from .inspection_middleware import InspectionMiddleware
+from .inspection_state import InspectionState
+
+__all__ = ["InspectionMiddleware", "InspectionState"]

libraries/botbuilder-core/botbuilder/core/inspection/inspection_middleware.py

@@ -0,0 +1,197 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from uuid import uuid4
+from typing import Any, List
+
+from jsonpickle import Pickler
+from botbuilder.core import BotState, ConversationState, TurnContext, UserState
+from botbuilder.schema import Activity, ActivityTypes, ConversationReference
+from botframework.connector.auth import MicrosoftAppCredentials
+
+from .inspection_session import InspectionSession
+from .inspection_sessions_by_status import (
+    InspectionSessionsByStatus,
+    DEFAULT_INSPECTION_SESSIONS_BY_STATUS,
+)
+from .inspection_state import InspectionState
+from .interception_middleware import InterceptionMiddleware
+from .trace_activity import from_state, make_command_activity
+
+
+class InspectionMiddleware(InterceptionMiddleware):
+    _COMMAND = "/INSPECT"
+
+    def __init__(  # pylint: disable=super-init-not-called
+        self,
+        inspection_state: InspectionState,
+        user_state: UserState = None,
+        conversation_state: ConversationState = None,
+        credentials: MicrosoftAppCredentials = None,
+    ):
+
+        self.inspection_state = inspection_state
+        self.inspection_state_accessor = inspection_state.create_property(
+            "InspectionSessionByStatus"
+        )
+        self.user_state = user_state
+        self.conversation_state = conversation_state
+        self.credentials = MicrosoftAppCredentials(
+            credentials.microsoft_app_id if credentials else "",
+            credentials.microsoft_app_password if credentials else "",
+        )
+
+    async def process_command(self, context: TurnContext) -> Any:
+        if context.activity.type == ActivityTypes.message and context.activity.text:
+
+            original_text = context.activity.text
+            TurnContext.remove_recipient_mention(context.activity)
+
+            command = context.activity.text.strip().split(" ")
+            if len(command) > 1 and command[0] == InspectionMiddleware._COMMAND:
+
+                if len(command) == 2 and command[1] == "open":
+                    await self._process_open_command(context)
+                    return True
+
+                if len(command) == 3 and command[1] == "attach":
+                    await self.process_attach_command(context, command[2])
+                    return True
+
+            context.activity.text = original_text
+
+        return False
+
+    async def _inbound(self, context: TurnContext, trace_activity: Activity) -> Any:
+        if await self.process_command(context):
+            return False, False
+
+        session = await self._find_session(context)
+        if session:
+            if await self._invoke_send(context, session, trace_activity):
+                return True, True
+        return True, False
+
+    async def _outbound(
+        self, context: TurnContext, trace_activities: List[Activity]
+    ) -> Any:
+        session = await self._find_session(context)
+        if session:
+            for trace_activity in trace_activities:
+                if not await self._invoke_send(context, session, trace_activity):
+                    break
+
+    async def _trace_state(self, context: TurnContext) -> Any:
+        session = await self._find_session(context)
+        if session:
+            if self.user_state:
+                await self.user_state.load(context, False)
+
+            if self.conversation_state:
+                await self.conversation_state.load(context, False)
+
+            bot_state = {}
+
+            if self.user_state:
+                bot_state["user_state"] = InspectionMiddleware._get_serialized_context(
+                    self.user_state, context
+                )
+
+            if self.conversation_state:
+                bot_state[
+                    "conversation_state"
+                ] = InspectionMiddleware._get_serialized_context(
+                    self.conversation_state, context
+                )
+
+            await self._invoke_send(context, session, from_state(bot_state))
+
+    async def _process_open_command(self, context: TurnContext) -> Any:
+        sessions = await self.inspection_state_accessor.get(
+            context, DEFAULT_INSPECTION_SESSIONS_BY_STATUS
+        )
+        session_id = self._open_command(
+            sessions, TurnContext.get_conversation_reference(context.activity)
+        )
+        await context.send_activity(
+            make_command_activity(
+                f"{InspectionMiddleware._COMMAND} attach {session_id}"
+            )
+        )
+        await self.inspection_state.save_changes(context, False)
+
+    async def process_attach_command(
+        self, context: TurnContext, session_id: str
+    ) -> None:
+        sessions = await self.inspection_state_accessor.get(
+            context, DEFAULT_INSPECTION_SESSIONS_BY_STATUS
+        )
+
+        if self._attach_comamnd(context.activity.conversation.id, sessions, session_id):
+            await context.send_activity(
+                "Attached to session, all traffic is being replicated for inspection."
+            )
+        else:
+            await context.send_activity(
+                f"Open session with id {session_id} does not exist."
+            )
+
+        await self.inspection_state.save_changes(context, False)
+
+    def _open_command(
+        self,
+        sessions: InspectionSessionsByStatus,
+        conversation_reference: ConversationReference,
+    ) -> str:
+        session_id = str(uuid4())
+        sessions.opened_sessions[session_id] = conversation_reference
+        return session_id
+
+    def _attach_comamnd(
+        self,
+        conversation_id: str,
+        sessions: InspectionSessionsByStatus,
+        session_id: str,
+    ) -> bool:
+        inspection_session_state = sessions.opened_sessions.get(session_id)
+        if inspection_session_state:
+            sessions.attached_sessions[conversation_id] = inspection_session_state
+            del sessions.opened_sessions[session_id]
+            return True
+
+        return False
+
+    @staticmethod
+    def _get_serialized_context(state: BotState, context: TurnContext):
+        ctx = state.get(context)
+        return Pickler(unpicklable=False).flatten(ctx)
+
+    async def _find_session(self, context: TurnContext) -> Any:
+        sessions = await self.inspection_state_accessor.get(
+            context, DEFAULT_INSPECTION_SESSIONS_BY_STATUS
+        )
+
+        conversation_reference = sessions.attached_sessions.get(
+            context.activity.conversation.id
+        )
+        if conversation_reference:
+            return InspectionSession(conversation_reference, self.credentials)
+
+        return None
+
+    async def _invoke_send(
+        self, context: TurnContext, session: InspectionSession, activity: Activity
+    ) -> bool:
+        if await session.send(activity):
+            return True
+
+        await self._clean_up_session(context)
+        return False
+
+    async def _clean_up_session(self, context: TurnContext) -> None:
+        sessions = await self.inspection_state_accessor.get(
+            context, DEFAULT_INSPECTION_SESSIONS_BY_STATUS
+        )
+
+        del sessions.attached_sessions[context.activity.conversation.id]
+        await self.inspection_state.save_changes(context, False)

libraries/botbuilder-core/botbuilder/core/inspection/inspection_session.py

@@ -0,0 +1,33 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from typing import Any
+
+from botbuilder.core import TurnContext
+from botbuilder.schema import Activity, ConversationReference
+from botframework.connector.aio import ConnectorClient
+from botframework.connector.auth import MicrosoftAppCredentials
+
+
+class InspectionSession:
+    def __init__(
+        self,
+        conversation_reference: ConversationReference,
+        credentials: MicrosoftAppCredentials,
+    ):
+        self._conversation_reference = conversation_reference
+        self._connector_client = ConnectorClient(
+            credentials, base_url=conversation_reference.service_url
+        )
+
+    async def send(self, activity: Activity) -> Any:
+        TurnContext.apply_conversation_reference(activity, self._conversation_reference)
+
+        try:
+            await self._connector_client.conversations.send_to_conversation(
+                activity.conversation.id, activity
+            )
+        except Exception:
+            return False
+
+        return True

libraries/botbuilder-core/botbuilder/core/inspection/inspection_sessions_by_status.py

@@ -0,0 +1,21 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from typing import Dict
+
+from botbuilder.schema import ConversationReference
+
+
+class InspectionSessionsByStatus:
+    def __init__(
+        self,
+        opened_sessions: Dict[str, ConversationReference] = None,
+        attached_sessions: Dict[str, ConversationReference] = None,
+    ):
+        self.opened_sessions: Dict[str, ConversationReference] = opened_sessions or {}
+        self.attached_sessions: Dict[
+            str, ConversationReference
+        ] = attached_sessions or {}
+
+
+DEFAULT_INSPECTION_SESSIONS_BY_STATUS = InspectionSessionsByStatus()