Add Slack webhook exposed key handling

canarytokens/channel.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import datetime
-from typing import Any, Coroutine, List, Optional, Union
+from typing import Any, Coroutine, Optional, Union
 
 import twisted.internet.reactor
 from twisted.internet import threads
@@ -24,8 +24,6 @@
     GoogleChatHeader,
     GoogleChatSection,
     Memo,
-    SlackAttachment,
-    SlackField,
     DiscordDetails,
     DiscordEmbeds,
     DiscordAuthorField,
@@ -34,35 +32,13 @@
     MsTeamsPotentialAction,
     TokenAlertDetails,
     TokenAlertDetailsGoogleChat,
-    TokenAlertDetailsSlack,
     TokenAlertDetailsDiscord,
     TokenAlertDetailsMsTeams,
 )
 
 log = Logger()
 
 
-def format_as_slack_canaryalert(details: TokenAlertDetails) -> TokenAlertDetailsSlack:
-    """
-    Transforms `TokenAlertDetails` to `TokenAlertDetailsSlack`.
-    """
-    fields: List[SlackField] = [
-        SlackField(title="Channel", value=details.channel),
-        SlackField(title="Memo", value=details.memo),
-        SlackField(
-            title="time",
-            value=details.time.strftime("%Y-%m-%d %H:%M:%S (UTC)"),
-        ),
-        SlackField(title="Manage", value=details.manage_url),
-    ]
-
-    attchments = [SlackAttachment(title_link=details.manage_url, fields=fields)]
-    return TokenAlertDetailsSlack(
-        # channel="#general",
-        attachments=attchments,
-    )
-
-
 def format_as_googlechat_canaryalert(
     details: TokenAlertDetails,
 ) -> TokenAlertDetailsGoogleChat:

canarytokens/models.py

@@ -49,7 +49,12 @@
     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,
+    prettify_snake_case,
+    dict_to_csv,
+    get_src_ip_continent,
+)
 
 CANARYTOKEN_RE = re.compile(
     ".*([" + "".join(CANARYTOKEN_ALPHABET) + "]{" + str(CANARYTOKEN_LENGTH) + "}).*",
@@ -401,10 +406,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.
@@ -2138,25 +2139,6 @@ class Config:
         }
 
 
-class SlackField(BaseModel):
-    title: str
-    value: str
-    short: bool = True
-
-
-class SlackAttachment(BaseModel):
-    title: str = "Canarytoken Triggered"
-    title_link: HttpUrl
-    mrkdwn_in: List[str] = ["title"]
-    fallback: str = ""
-    fields: List[SlackField]
-
-    def __init__(__pydantic_self__, **data: Any) -> None:
-        # HACK: We can do better here.
-        data["fallback"] = f"Canarytoken Triggered: {data['title_link']}"
-        super().__init__(**data)
-
-
 class GoogleChatDecoratedText(BaseModel):
     topLabel: str = ""
     text: str = ""
@@ -2301,15 +2283,6 @@ def json_safe_dict(self) -> Dict[str, str]:
         return json_safe_dict(self)
 
 
-class TokenAlertDetailsSlack(BaseModel):
-    """Details that are sent to slack webhooks."""
-
-    attachments: List[SlackAttachment]
-
-    def json_safe_dict(self) -> Dict[str, str]:
-        return json_safe_dict(self)
-
-
 class MsTeamsDetailsSection(BaseModel):
     canarytoken: Canarytoken
     token_reminder: Memo

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,29 +859,8 @@ 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(
-            attachments=[
-                models.SlackAttachment(
-                    title_link=HttpUrl("https://test.com/check", scheme="https"),
-                    fields=[
-                        models.SlackField(
-                            title="test",
-                            value="Working",
-                        )
-                    ],
-                )
-            ]
-        )
-    elif webhook_type == WebhookType.GOOGLE_CHAT:
+    if webhook_type == WebhookType.GOOGLE_CHAT:
         # construct google chat alert card
         card = models.GoogleChatCard(
             header=models.GoogleChatHeader(

canarytokens/utils.py

@@ -1,8 +1,14 @@
+import json
 import subprocess
 from pathlib import Path
-from typing import Any, Literal, Union
+from typing import Any, Dict, Literal, Tuple, Union
 
 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: