Add support for MS Teams webhooks (#339)

canarytokens/channel.py

@@ -5,13 +5,15 @@
 
 import datetime
 from typing import Any, Coroutine, List, Optional, Union
+import re
 
 import twisted.internet.reactor
 from twisted.internet import threads
 from twisted.logger import Logger
 
 from canarytokens import switchboard as sb
 from canarytokens.canarydrop import Canarydrop
+from canarytokens import constants
 
 # from canarytokens.exceptions import DuplicateChannel
 from canarytokens.models import (
@@ -27,11 +29,15 @@
     DiscordDetails,
     DiscordEmbeds,
     DiscordAuthorField,
+    MsTeamsTitleSection,
+    MsTeamsDetailsSection,
+    MsTeamsPotentialAction,
     TokenAlertDetailGeneric,
     TokenAlertDetails,
     TokenAlertDetailsGoogleChat,
     TokenAlertDetailsSlack,
     TokenAlertDetailsDiscord,
+    TokenAlertDetailsMsTeams,
 )
 
 log = Logger()
@@ -122,6 +128,28 @@ def format_as_discord_canaryalert(
     return TokenAlertDetailsDiscord(embeds=[embeds])
 
 
+def format_as_ms_teams_canaryalert(
+    details: TokenAlertDetails,
+) -> TokenAlertDetailsMsTeams:
+    sections = [
+        MsTeamsTitleSection(activityTitle="<b>Canarytoken triggered</b>"),
+        MsTeamsDetailsSection(
+            canarytoken=details.token,
+            token_reminder=details.memo,
+            src_data=details.src_data if details.src_data else None,
+            additional_data=details.additional_data,
+        ),
+    ]
+
+    return TokenAlertDetailsMsTeams(
+        summary="Canarytoken triggered",
+        sections=sections,
+        potentialAction=[
+            MsTeamsPotentialAction(name="Manage", target=[details.manage_url])
+        ],
+    )
+
+
 class Channel(object):
     CHANNEL = "Base"
 
@@ -207,26 +235,33 @@ def format_webhook_canaryalert(
         TokenAlertDetailsSlack, TokenAlertDetailGeneric, TokenAlertDetailsGoogleChat
     ]:
         # TODO: Need to add `host` and `protocol` that can be used to manage the token.
-        slack_hook_base_url = "https://hooks.slack.com"
-        googlechat_hook_base_url = "https://chat.googleapis.com"
-        discord_hook_base_url = "https://discord.com/api/webhooks"
         details = cls.gather_alert_details(
             canarydrop,
             protocol=protocol,
             host=host,
         )
         if canarydrop.alert_webhook_url and (
-            str(canarydrop.alert_webhook_url).startswith(slack_hook_base_url)
+            str(canarydrop.alert_webhook_url).startswith(
+                constants.WEBHOOK_BASE_URL_SLACK
+            )
         ):
             return format_as_slack_canaryalert(details=details)
         elif canarydrop.alert_webhook_url and (
-            str(canarydrop.alert_webhook_url).startswith(googlechat_hook_base_url)
+            str(canarydrop.alert_webhook_url).startswith(
+                constants.WEBHOOK_BASE_URL_GOOGLE_CHAT
+            )
         ):
             return format_as_googlechat_canaryalert(details=details)
         elif canarydrop.alert_webhook_url and (
-            str(canarydrop.alert_webhook_url).startswith(discord_hook_base_url)
+            str(canarydrop.alert_webhook_url).startswith(
+                constants.WEBHOOK_BASE_URL_DISCORD
+            )
         ):
             return format_as_discord_canaryalert(details=details)
+        elif re.match(
+            constants.WEBHOOK_BASE_URL_REGEX_MS_TEAMS, str(canarydrop.alert_webhook_url)
+        ):
+            return format_as_ms_teams_canaryalert(details=details)
         else:
             return TokenAlertDetailGeneric(**details.dict())
 

canarytokens/constants.py

@@ -24,6 +24,10 @@
 # fmt: on
 CANARYTOKEN_LENGTH = 25  # equivalent to 128-bit id
 
+CANARY_IMAGE_URL = (
+    "https://s3-eu-west-1.amazonaws.com/email-images.canary.tools/canary-logo-round.png"
+)
+
 CANARY_PDF_TEMPLATE_OFFSET: int = 793
 
 
@@ -32,3 +36,7 @@
 ]
 
 MAX_WEBHOOK_URL_LENGTH = 1024
+WEBHOOK_BASE_URL_SLACK = "https://hooks.slack.com"
+WEBHOOK_BASE_URL_GOOGLE_CHAT = "https://chat.googleapis.com"
+WEBHOOK_BASE_URL_DISCORD = "https://discord.com/api/webhooks"
+WEBHOOK_BASE_URL_REGEX_MS_TEAMS = r"^https://[\w.]+\.webhook\.office\.com/webhookb2/.*"

canarytokens/models.py

@@ -46,8 +46,10 @@
 from canarytokens.constants import (
     CANARYTOKEN_ALPHABET,
     CANARYTOKEN_LENGTH,
+    CANARY_IMAGE_URL,
     MEMO_MAX_CHARACTERS,
 )
+from canarytokens.utils import prettify_snake_case, dict_to_csv
 
 CANARYTOKEN_RE = re.compile(
     ".*([" + "".join(CANARYTOKEN_ALPHABET) + "]{" + str(CANARYTOKEN_LENGTH) + "}).*",
@@ -2047,6 +2049,70 @@ def json_safe_dict(self) -> Dict[str, str]:
         return json_safe_dict(self)
 
 
+class MsTeamsDetailsSection(BaseModel):
+    canarytoken: Canarytoken
+    token_reminder: Memo
+    src_data: Optional[dict[str, Any]] = None
+    additional_data: Optional[dict[str, Any]] = None
+
+    def dict(self, *args, **kwargs):
+        data = json_safe_dict(self)
+        data["Canarytoken"] = data.pop("canarytoken", "")
+        data["Token Reminder"] = data.pop("token_reminder", "")
+        if "src_data" in data:
+            data["Source Data"] = data.pop("src_data", "")
+
+        if data["additional_data"]:
+            add_data = data.pop("additional_data", {})
+            data.update(add_data)
+
+        facts = []
+        for k, v in data.items():
+            if not v:
+                continue
+
+            if isinstance(v, dict):
+                v = dict_to_csv(v)
+            else:
+                v = str(v)
+
+            facts.append({"name": prettify_snake_case(k), "value": v})
+
+        return {"facts": facts}
+
+
+class MsTeamsTitleSection(BaseModel):
+    activityTitle: str
+    activityImage = CANARY_IMAGE_URL
+
+
+class MsTeamsPotentialAction(BaseModel):
+    name: str
+    target: List[AnyHttpUrl]
+    type: str = "ViewAction"
+    context: str = "http://schema.org"
+
+    def dict(self, *args, **kwargs):
+        d = super().dict(*args, **kwargs)
+
+        d["@type"] = d.pop("type")
+        d["@context"] = d.pop("context")
+
+        return d
+
+
+class TokenAlertDetailsMsTeams(BaseModel):
+    """Details that are sent to MS Teams webhooks."""
+
+    summary: str
+    themeColor = "ff0000"
+    sections: Optional[List[Union[MsTeamsTitleSection, MsTeamsDetailsSection]]] = None
+    potentialAction: Optional[List[MsTeamsPotentialAction]] = None
+
+    def json_safe_dict(self) -> Dict[str, str]:
+        return json_safe_dict(self)
+
+
 class TokenAlertDetailsDiscord(BaseModel):
     """Details that are sent to Discord webhooks"""
 

canarytokens/queries.py

@@ -809,16 +809,14 @@ def validate_webhook(url, token_type: models.TokenTypes):
     if len(url) > constants.MAX_WEBHOOK_URL_LENGTH:
         raise WebhookTooLongError()
 
-    slack = "https://hooks.slack.com"
-    googlechat_hook_base_url = "https://chat.googleapis.com"
-    discord = "https://discord.com/api/webhooks"
     payload: Union[
         models.TokenAlertDetails,
         models.TokenAlertDetailsSlack,
         models.TokenAlertDetailsGoogleChat,
         models.TokenAlertDetailsDiscord,
+        models.TokenAlertDetailsMsTeams,
     ]
-    if url.startswith(slack):
+    if url.startswith(constants.WEBHOOK_BASE_URL_SLACK):
         payload = models.TokenAlertDetailsSlack(
             attachments=[
                 models.SlackAttachment(
@@ -832,7 +830,7 @@ def validate_webhook(url, token_type: models.TokenTypes):
                 )
             ]
         )
-    elif url.startswith(googlechat_hook_base_url):
+    elif url.startswith(constants.WEBHOOK_BASE_URL_GOOGLE_CHAT):
         # construct google chat alert card
         card = models.GoogleChatCard(
             header=models.GoogleChatHeader(
@@ -847,7 +845,7 @@ def validate_webhook(url, token_type: models.TokenTypes):
         payload = models.TokenAlertDetailsGoogleChat(
             cardsV2=[models.GoogleChatCardV2(cardId="unique-card-id", card=card)]
         )
-    elif url.startswith(discord):
+    elif url.startswith(constants.WEBHOOK_BASE_URL_DISCORD):
         # construct discord alert card
         embeds = models.DiscordEmbeds(
             author=models.DiscordAuthorField(
@@ -858,7 +856,13 @@ def validate_webhook(url, token_type: models.TokenTypes):
             timestamp=datetime.datetime.now(),
         )
         payload = models.TokenAlertDetailsDiscord(embeds=[embeds])
-
+    elif re.match(constants.WEBHOOK_BASE_URL_REGEX_MS_TEAMS, url):
+        section = models.MsTeamsTitleSection(
+            activityTitle="<b>Validating new Canarytokens webhook</b>"
+        )
+        payload = models.TokenAlertDetailsMsTeams(
+            summary="Validating new Canarytokens webhook", sections=[section]
+        )
     else:
         payload = models.TokenAlertDetails(
             manage_url=HttpUrl(

canarytokens/utils.py

@@ -3,6 +3,18 @@
 from typing import Any, Literal, Union
 
 
+def dict_to_csv(d: dict) -> str:
+    """Convert dict to CSV"""
+    return ", ".join(f"{k}: {v}" for k, v in d.items())
+
+
+def prettify_snake_case(s: str):
+    """Capitalize first letter and convert underscores to spaces"""
+    s = s.replace("_", " ")
+    s = s[0].upper() + s[1:]
+    return s
+
+
 def coerce_to_float(value: Any) -> Union[Literal[False], float]:
     """
     Tries to convert `value` to a float and returns