Merge pull request #1739 from domwhewell-sage/enhance_teams_webhook

Changed teams output module to POST an adaptive card to the power automate webhook

bbot/modules/output/slack.py

@@ -16,7 +16,6 @@ class Slack(WebhookOutputModule):
         "event_types": "Types of events to send",
         "min_severity": "Only allow VULNERABILITY events of this severity or higher",
     }
-    good_status_code = 200
     content_key = "text"
 
     def format_message_str(self, event):

bbot/modules/output/teams.py

@@ -10,14 +10,115 @@ class Teams(WebhookOutputModule):
     }
     options = {"webhook_url": "", "event_types": ["VULNERABILITY", "FINDING"], "min_severity": "LOW"}
     options_desc = {
-        "webhook_url": "Discord webhook URL",
+        "webhook_url": "Teams webhook URL",
         "event_types": "Types of events to send",
         "min_severity": "Only allow VULNERABILITY events of this severity or higher",
     }
     _module_threads = 5
-    good_status_code = 200
-    content_key = "text"
+    adaptive_card = {
+        "type": "message",
+        "attachments": [
+            {
+                "contentType": "application/vnd.microsoft.card.adaptive",
+                "contentUrl": None,
+                "content": {
+                    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                    "type": "AdaptiveCard",
+                    "version": "1.2",
+                    "msteams": {"width": "full"},
+                    "body": [],
+                },
+            }
+        ],
+    }
+
+    async def handle_event(self, event):
+        while 1:
+            data = self.format_message(self.adaptive_card.copy(), event)
+
+            response = await self.helpers.request(
+                url=self.webhook_url,
+                method="POST",
+                json=data,
+            )
+            status_code = getattr(response, "status_code", 0)
+            if self.evaluate_response(response):
+                break
+            else:
+                response_data = getattr(response, "text", "")
+                try:
+                    retry_after = response.json().get("retry_after", 1)
+                except Exception:
+                    retry_after = 1
+                self.verbose(
+                    f"Error sending {event}: status code {status_code}, response: {response_data}, retrying in {retry_after} seconds"
+                )
+                await self.helpers.sleep(retry_after)
+
+    def trim_message(self, message):
+        if len(message) > self.message_size_limit:
+            message = message[: self.message_size_limit - 3] + "..."
+        return message
+
+    def format_message_str(self, event):
+        items = []
+        msg = self.trim_message(event.data)
+        items.append({"type": "TextBlock", "text": f"{msg}", "wrap": True})
+        items.append({"type": "FactSet", "facts": [{"title": "Tags:", "value": ", ".join(event.tags)}]})
+        return items
+
+    def format_message_other(self, event):
+        items = [{"type": "FactSet", "facts": []}]
+        for key, value in event.data.items():
+            if key != "severity":
+                msg = self.trim_message(str(value))
+                items[0]["facts"].append({"title": f"{key}:", "value": msg})
+        return items
+
+    def get_severity_color(self, event):
+        color = "Accent"
+        if event.type == "VULNERABILITY":
+            severity = event.data.get("severity", "UNKNOWN")
+            if severity == "CRITICAL":
+                color = "Attention"
+            elif severity == "HIGH":
+                color = "Attention"
+            elif severity == "MEDIUM":
+                color = "Warning"
+            elif severity == "LOW":
+                color = "Good"
+        return color
 
-    def evaluate_response(self, response):
-        text = getattr(response, "text", "")
-        return text == "1"
+    def format_message(self, adaptive_card, event):
+        heading = {"type": "TextBlock", "text": f"{event.type}", "wrap": True, "size": "Large", "style": "heading"}
+        body = adaptive_card["attachments"][0]["content"]["body"]
+        body.append(heading)
+        if event.type in ("VULNERABILITY", "FINDING"):
+            subheading = {
+                "type": "TextBlock",
+                "text": event.data.get("severity", "UNKNOWN"),
+                "spacing": "None",
+                "size": "Large",
+                "wrap": True,
+            }
+            subheading["color"] = self.get_severity_color(event)
+            body.append(subheading)
+        main_text = {
+            "type": "ColumnSet",
+            "separator": True,
+            "spacing": "Medium",
+            "columns": [
+                {
+                    "type": "Column",
+                    "width": "stretch",
+                    "items": [],
+                }
+            ],
+        }
+        if isinstance(event.data, str):
+            items = self.format_message_str(event)
+        else:
+            items = self.format_message_other(event)
+        main_text["columns"][0]["items"] = items
+        body.append(main_text)
+        return adaptive_card

bbot/modules/templates/webhook.py

@@ -9,7 +9,6 @@ class WebhookOutputModule(BaseOutputModule):
     """
 
     accept_dupes = False
-    good_status_code = 204
     message_size_limit = 2000
     content_key = "content"
     vuln_severities = ["UNKNOWN", "LOW", "MEDIUM", "HIGH", "CRITICAL"]
@@ -94,5 +93,4 @@ def format_message(self, event):
         return msg
 
     def evaluate_response(self, response):
-        status_code = getattr(response, "status_code", 0)
-        return status_code == self.good_status_code
+        return getattr(response, "is_success", False)

bbot/test/test_step_2/module_tests/test_module_discord.py

@@ -29,7 +29,7 @@ def custom_response(request: httpx.Request):
             if module_test.request_count == 2:
                 return httpx.Response(status_code=429, json={"retry_after": 0.01})
             else:
-                return httpx.Response(status_code=module_test.module.good_status_code)
+                return httpx.Response(status_code=200)
 
         module_test.httpx_mock.add_callback(custom_response, url=self.webhook_url)
 

bbot/test/test_step_2/module_tests/test_module_teams.py

@@ -16,13 +16,15 @@ def custom_response(request: httpx.Request):
             module_test.request_count += 1
             if module_test.request_count == 2:
                 return httpx.Response(
-                    status_code=200,
-                    text="Webhook message delivery failed with error: Microsoft Teams endpoint returned HTTP error 429 with ContextId tcid=0,server=msgapi-production-eus-azsc2-4-170,cv=deadbeef=2..",
+                    status_code=400,
+                    json={
+                        "error": {
+                            "code": "WorkflowTriggerIsNotEnabled",
+                            "message": "Could not execute workflow 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx' trigger 'manual' with state 'Disabled': trigger is not enabled.",
+                        }
+                    },
                 )
             else:
-                return httpx.Response(
-                    status_code=200,
-                    text="1",
-                )
+                return httpx.Response(status_code=200)
 
         module_test.httpx_mock.add_callback(custom_response, url=self.webhook_url)