Add MS Teams webhook exposed key handling

canarytokens/channel.py

@@ -29,14 +29,10 @@
     DiscordDetails,
     DiscordEmbeds,
     DiscordAuthorField,
-    MsTeamsTitleSection,
-    MsTeamsDetailsSection,
-    MsTeamsPotentialAction,
     TokenAlertDetails,
     TokenAlertDetailsGoogleChat,
     TokenAlertDetailsSlack,
     TokenAlertDetailsDiscord,
-    TokenAlertDetailsMsTeams,
 )
 
 log = Logger()
@@ -127,28 +123,6 @@ 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"
 

canarytokens/models.py

@@ -46,10 +46,9 @@
 from canarytokens.constants import (
     CANARYTOKEN_ALPHABET,
     CANARYTOKEN_LENGTH,
-    CANARY_IMAGE_URL,
     MEMO_MAX_CHARACTERS,
 )
-from canarytokens.utils import prettify_snake_case, dict_to_csv, get_src_ip_continent
+from canarytokens.utils import json_safe_dict, get_src_ip_continent
 
 CANARYTOKEN_RE = re.compile(
     ".*([" + "".join(CANARYTOKEN_ALPHABET) + "]{" + str(CANARYTOKEN_LENGTH) + "}).*",
@@ -401,10 +400,6 @@ def __str__(self) -> str:
 ]
 
 
-def json_safe_dict(m: BaseModel, exclude: Tuple = ()) -> Dict[str, str]:
-    return json.loads(m.json(exclude_none=True, exclude=set(exclude)))
-
-
 class TokenRequest(BaseModel):
     """
     TokenRequest holds fields needed to create a Canarytoken.
@@ -2310,70 +2305,6 @@ 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

@@ -7,7 +7,7 @@
 import re
 import secrets
 from ipaddress import IPv4Address
-from typing import Dict, List, Literal, Optional, Tuple, Union
+from typing import Dict, List, Literal, Optional, Tuple
 
 import advocate
 import requests
@@ -859,13 +859,6 @@ def validate_webhook(url, token_type: models.TokenTypes):
     if len(url) > constants.MAX_WEBHOOK_URL_LENGTH:
         raise WebhookTooLongError()
 
-    payload: Union[
-        models.TokenAlertDetails,
-        models.TokenAlertDetailsSlack,
-        models.TokenAlertDetailsGoogleChat,
-        models.TokenAlertDetailsDiscord,
-        models.TokenAlertDetailsMsTeams,
-    ]
     webhook_type = get_webhook_type(url)
     if webhook_type == WebhookType.SLACK:
         payload = models.TokenAlertDetailsSlack(
@@ -907,13 +900,6 @@ def validate_webhook(url, token_type: models.TokenTypes):
             timestamp=datetime.datetime.now(),
         )
         payload = models.TokenAlertDetailsDiscord(embeds=[embeds])
-    elif webhook_type == WebhookType.MS_TEAMS:
-        section = models.MsTeamsTitleSection(
-            activityTitle="<b>Validating new Canarytokens webhook</b>"
-        )
-        payload = models.TokenAlertDetailsMsTeams(
-            summary="Validating new Canarytokens webhook", sections=[section]
-        )
     else:
         payload = generate_webhook_test_payload(webhook_type, token_type)
 

canarytokens/utils.py

@@ -1,8 +1,14 @@
 import subprocess
 from pathlib import Path
-from typing import Any, Literal, Union
+from typing import Any, Literal, Union, Tuple, Dict
+import json
 
 import pycountry_convert
+from pydantic import BaseModel
+
+
+def json_safe_dict(m: BaseModel, exclude: Tuple = ()) -> Dict[str, str]:
+    return json.loads(m.json(exclude_none=True, exclude=set(exclude)))
 
 
 def dict_to_csv(d: dict) -> str: