Add token exposed processing to tokens server (#601)

canarytokens/canarydrop.py

@@ -19,7 +19,7 @@
 from urllib.parse import quote
 from typing import Any, Literal, Optional, Union
 
-from pydantic import BaseModel, Field, parse_obj_as, root_validator
+from pydantic import AnyHttpUrl, BaseModel, Field, parse_obj_as, root_validator
 
 from canarytokens import queries, tokens
 from canarytokens.constants import (
@@ -32,6 +32,7 @@
     AnySettingsRequest,
     AnyTokenHistory,
     AnyTokenHit,
+    AnyTokenExposedHit,
     BrowserScannerSettingsRequest,
     EmailSettingsRequest,
     PWAType,
@@ -159,12 +160,13 @@ class Canarydrop(BaseModel):
     cc_v2_expiry_year: Optional[int]
     cc_v2_name_on_card: Literal["Canarytokens.org"] = "Canarytokens.org"
 
+    key_exposed_details: Optional[AnyTokenExposedHit] = None
+
     @root_validator(pre=True)
     def _validate_triggered_details(cls, values):
         """
         Ensure canarydrop `type` and `triggered_details` `token_type` match.
         """
-
         if values.get("triggered_details", None) is None:
             values["triggered_details"] = parse_obj_as(
                 AnyTokenHistory, {"token_type": values["type"], "hits": []}
@@ -181,8 +183,33 @@ def _validate_triggered_details(cls, values):
             {getattr(values["triggered_details"], "token_type")} != {values["type"]}
             """
             )
+
+        if "key_exposed_details" in values:
+            values["key_exposed_details"] = parse_obj_as(
+                AnyTokenExposedHit, values["key_exposed_details"]
+            )
+            token_type, expected_token_type = (
+                values["key_exposed_details"].token_type,
+                values["type"],
+            )
+            if token_type != expected_token_type:
+                raise ValueError(
+                    f"key_exposed_details token_type must match drop type. Got {token_type} instead of {expected_token_type}."
+                )
+
         return values
 
+    def build_manage_url(self, protocol: str, host: str) -> AnyHttpUrl:
+        return parse_obj_as(
+            AnyHttpUrl,
+            "{protocol}://{host}/manage?token={token}&auth={auth}".format(
+                protocol=protocol,
+                host=host,
+                token=self.canarytoken.value(),
+                auth=self.auth,
+            ),
+        )
+
     class Config:
         arbitrary_types_allowed = True
         allow_population_by_field_name = True
@@ -224,6 +251,16 @@ def add_canarydrop_hit(self, *, token_hit: AnyTokenHit):
             canarytoken=self.canarytoken,
         )
 
+    def add_key_exposed_hit(self, token_exposed_hit: AnyTokenExposedHit):
+        if self.key_exposed_details is not None:
+            # Only store the first event since that's the most important - gives the best indication of how
+            # long the key has been exposed
+            return
+
+        queries.save_canarydrop(self)
+        self.key_exposed_details = token_exposed_hit
+        queries.add_key_exposed_hit(token_exposed_hit, self.canarytoken)
+
     def apply_settings_change(self, setting_request: AnySettingsRequest) -> bool:
         """
         Modifies a the drop applying (enable/disable) on any one
@@ -419,6 +456,7 @@ def serialize(
                     "canarytoken",
                     "switchboard_settings",
                     "triggered_details",  # V2 compatible.
+                    "key_exposed_details",
                 }
             ),
         )  # TODO: check https://github.com/samuelcolvin/pydantic/issues/1409 and swap out when possible

canarytokens/channel.py

@@ -1,23 +1,23 @@
 """"
 Base class for all canarydrop channels.
 """
+
 from __future__ import annotations
 
 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 (
     AnyTokenHit,
+    AnyTokenExposedHit,
     GoogleChatAlertDetailsSectionData,
     GoogleChatCard,
     GoogleChatCardV2,
@@ -32,7 +32,6 @@
     MsTeamsTitleSection,
     MsTeamsDetailsSection,
     MsTeamsPotentialAction,
-    TokenAlertDetailGeneric,
     TokenAlertDetails,
     TokenAlertDetailsGoogleChat,
     TokenAlertDetailsSlack,
@@ -216,57 +215,17 @@ def gather_alert_details(
             memo=Memo(canarydrop.memo),
             token=canarydrop.canarytoken.value(),
             # TODO: this manage url should come from the frontend / settings object.
-            manage_url="{protocol}://{host}/manage?token={token}&auth={auth}".format(
-                protocol=protocol,
-                host=host,
-                token=canarydrop.canarytoken.value(),
-                auth=canarydrop.auth,
-            ),
+            manage_url=canarydrop.build_manage_url(protocol, host),
             additional_data=additional_data,  # TODO: additional details need to be re-worked.
             public_domain=host,
         )
 
-    @classmethod
-    def format_webhook_canaryalert(
-        cls,
+    def dispatch(
+        self,
+        *,
         canarydrop: Canarydrop,
-        protocol: str,
-        host: str,  # DESIGN: Shift this to settings. Do we need to have this logic here?
-    ) -> Union[
-        TokenAlertDetailsSlack, TokenAlertDetailGeneric, TokenAlertDetailsGoogleChat
-    ]:
-        # TODO: Need to add `host` and `protocol` that can be used to manage the token.
-        details = cls.gather_alert_details(
-            canarydrop,
-            protocol=protocol,
-            host=host,
-        )
-        if canarydrop.alert_webhook_url and (
-            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(
-                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(
-                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())
-
-    def dispatch(self, *, canarydrop: Canarydrop, token_hit: AnyTokenHit) -> None:
+        token_hit: Union[AnyTokenHit, AnyTokenExposedHit],
+    ) -> None:
         """
         Spins off a `switchboard.dispatch` which notifies on all necessary channels.
         """
@@ -313,7 +272,7 @@ def send_alert(
         self,
         input_channel: InputChannel,
         canarydrop: Canarydrop,
-        token_hit: AnyTokenHit,
+        token_hit: Union[AnyTokenHit, AnyTokenExposedHit],
     ) -> None:
         self.do_send_alert(
             input_channel=input_channel,
@@ -325,7 +284,7 @@ def do_send_alert(
         self,
         input_channel: InputChannel,
         canarydrop: Canarydrop,
-        token_hit: AnyTokenHit,
+        token_hit: Union[AnyTokenHit, AnyTokenExposedHit],
     ) -> Coroutine[Any, Any, None]:
         #  Design: Make this a typing.protocol and drop this.
         raise NotImplementedError("Generic Output channel cannot `do_send_alert`")

canarytokens/channel_http.py

@@ -14,7 +14,7 @@
 from canarytokens.channel import InputChannel
 from canarytokens.constants import INPUT_CHANNEL_HTTP
 from canarytokens.exceptions import NoCanarytokenFound, NoCanarydropFound
-from canarytokens.models import AnyTokenHit, TokenTypes
+from canarytokens.models import AnyTokenHit, AWSKeyTokenHit, TokenTypes
 from canarytokens.queries import get_canarydrop
 from canarytokens.settings import FrontendSettings, SwitchboardSettings
 from canarytokens.switchboard import Switchboard
@@ -162,7 +162,7 @@ def render_OPTIONS(self, request: Request):
         request.responseHeaders.removeHeader("Content-Type")
         return b""
 
-    def render_POST(self, request: Request):
+    def render_POST(self, request: Request):  # noqa: C901
         try:
             token = Canarytoken(value=request.path)
         except NoCanarytokenFound:
@@ -182,7 +182,10 @@ def render_POST(self, request: Request):
 
         if canarydrop.type == TokenTypes.AWS_KEYS:
             token_hit = Canarytoken._parse_aws_key_trigger(request)
-            canarydrop.add_canarydrop_hit(token_hit=token_hit)
+            if isinstance(token_hit, AWSKeyTokenHit):
+                canarydrop.add_canarydrop_hit(token_hit=token_hit)
+            else:
+                canarydrop.add_key_exposed_hit(token_hit)
             self.dispatch(canarydrop=canarydrop, token_hit=token_hit)
             return b"success"
         elif canarydrop.type == TokenTypes.AZURE_ID:

canarytokens/channel_output_email.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from textwrap import dedent
 import textwrap
-from typing import Optional
+from typing import Optional, Union
 import enum
 import uuid
 
@@ -26,6 +26,7 @@
 from canarytokens.constants import OUTPUT_CHANNEL_EMAIL, MAILGUN_IGNORE_ERRORS
 from canarytokens.models import (
     AnyTokenHit,
+    AnyTokenExposedHit,
     readable_token_type_names,
     TokenAlertDetails,
     token_types_with_article_an,
@@ -418,7 +419,7 @@ def do_send_alert(
         self,
         input_channel: InputChannel,
         canarydrop: Canarydrop,
-        token_hit: AnyTokenHit,
+        token_hit: Union[AnyTokenHit, AnyTokenExposedHit],
     ):
         alert_details = input_channel.gather_alert_details(
             canarydrop=canarydrop,

canarytokens/channel_output_webhook.py

@@ -1,7 +1,8 @@
 """
 Output channel that sends to webhooks.
 """
-from typing import Dict
+
+from typing import Dict, Union
 
 import advocate
 import requests
@@ -11,7 +12,14 @@
 from canarytokens import canarydrop
 from canarytokens.channel import InputChannel, OutputChannel
 from canarytokens.constants import OUTPUT_CHANNEL_WEBHOOK
-from canarytokens.models import AnyTokenHit
+from canarytokens.models import (
+    AnyTokenHit,
+    AnyTokenExposedHit,
+    Memo,
+    TokenExposedDetails,
+    TokenExposedHit,
+)
+from canarytokens.webhook_formatting import format_details_for_webhook, get_webhook_type
 
 log = Logger()
 
@@ -23,7 +31,7 @@ def do_send_alert(
         self,
         input_channel: InputChannel,
         canarydrop: canarydrop.Canarydrop,
-        token_hit: AnyTokenHit,
+        token_hit: Union[AnyTokenHit, AnyTokenExposedHit],
     ) -> None:
         # TODO we should format using the hit directly,
         #      we use the drop to get the latest when we already have it
@@ -33,11 +41,28 @@ def do_send_alert(
                 f"alert_webhook_url must start with http[s]://; url found for drop {canarydrop.canarytoken.value()}: {url}"
             )
 
-        payload = input_channel.format_webhook_canaryalert(
-            canarydrop=canarydrop,
-            host=self.hostname,
-            protocol=self.switchboard_scheme,
-        )
+        if isinstance(token_hit, TokenExposedHit):
+            details = TokenExposedDetails(
+                token_type=token_hit.token_type,
+                token=canarydrop.canarytoken.value(),
+                key_id=canarydrop.aws_access_key_id,
+                memo=Memo(canarydrop.memo),
+                public_location=token_hit.public_location,
+                exposed_time=token_hit.time_of_hit,
+                manage_url=canarydrop.build_manage_url(
+                    self.switchboard_scheme, self.hostname
+                ),
+                public_domain=self.hostname,
+            )
+        else:
+            details = input_channel.gather_alert_details(
+                canarydrop=canarydrop,
+                protocol=self.switchboard_scheme,
+                host=self.hostname,
+            )
+
+        webhook_type = get_webhook_type(url)
+        payload = format_details_for_webhook(webhook_type, details)
 
         success = self.generic_webhook_send(
             payload=payload.json_safe_dict(),