Add Fake IdP SAML App tokens: Backend and tests (#622)

* Add SAML IdP App token backend and tests

* Update notification template and rename generated file

* Add token alert icon for email

* Improve webdav tests; add missing env vars to CI for tests

.github/workflows/test.yml

@@ -110,6 +110,10 @@ jobs:
           export CANARY_MAILGUN_API_KEY=${{ secrets.TESTING_MAILGUN_API_KEY }}
           export CANARY_SENTRY_ENVIRONMENT=ci
           export CANARY_WEB_IMAGE_UPLOAD_PATH=../uploads
+          export CANARY_WEBDAV_SERVER=test
+          export CANARY_CLOUDFLARE_ACCOUNT_ID=test
+          export CANARY_CLOUDFLARE_API_TOKEN=test
+          export CANARY_CLOUDFLARE_NAMESPACE=test
           cd tests
           poetry run coverage run  --source=../canarytokens --omit="integration/test_custom_binary.py,integration/test_sql_server_token.py" -m pytest units --runv3 -v
 

canarytokens/canarydrop.py

@@ -36,6 +36,7 @@
     AnyTokenExposedHit,
     BrowserScannerSettingsRequest,
     EmailSettingsRequest,
+    IdPAppType,
     PWAType,
     TokenTypes,
     User,
@@ -170,6 +171,9 @@ class Canarydrop(BaseModel):
 
     key_exposed_details: Optional[AnyTokenExposedHit] = None
 
+    idp_app_entity_id: Optional[str]
+    idp_app_type: Optional[IdPAppType]
+
     @root_validator(pre=True)
     def _validate_triggered_details(cls, values):
         """
@@ -308,6 +312,7 @@ def generate_random_url(
         self,
         canary_domains: list[str],
         page: Optional[str] = None,
+        use_path_elements: bool = True,
         skip_cache: bool = False,
     ) -> str:
         """
@@ -320,26 +325,28 @@ def generate_random_url(
             return self.generated_url
         (path_elements, pages) = self.get_url_components()
 
-        generated_url = random.choice(canary_domains) + "/"
         path = []
-        for count in range(0, random.randint(1, 3)):
-            if len(path_elements) == 0:
-                break
-
-            elem = path_elements[random.randint(0, len(path_elements) - 1)]
-            path.append(elem)
-            path_elements.remove(elem)
+        if use_path_elements:
+            path = random.sample(
+                path_elements, random.randint(1, min(3, len(path_elements)))
+            )
         path.append(self.canarytoken.value())
+        path.append(random.choice(pages) if page is None else page)
 
-        path.append(pages[random.randint(0, len(pages) - 1)] if page is None else page)
+        generated_url = random.choice(canary_domains) + "/"
         generated_url += "/".join(path)
         # cache
         if not skip_cache:
             self.generated_url = generated_url
         return generated_url
 
-    def get_url(self, canary_domains: list[str], page: Optional[str] = None):
-        return self.generate_random_url(canary_domains, page)
+    def get_url(
+        self,
+        canary_domains: list[str],
+        page: Optional[str] = None,
+        use_path_elements: Optional[bool] = True,
+    ):
+        return self.generate_random_url(canary_domains, page, use_path_elements)
 
     def generate_random_hostname(self, with_random=False, nxdomain=False):
         """

canarytokens/channel_http.py

@@ -16,6 +16,7 @@
 from canarytokens.exceptions import NoCanarytokenFound, NoCanarydropFound
 from canarytokens.models import AnyTokenHit, AWSKeyTokenHit, TokenTypes
 from canarytokens.queries import get_canarydrop
+from canarytokens.saml import SAML_POST_ARG
 from canarytokens.settings import FrontendSettings, SwitchboardSettings
 from canarytokens.switchboard import Switchboard
 from canarytokens.tokens import Canarytoken, GIF, get_template_env
@@ -179,7 +180,6 @@ def render_POST(self, request: Request):  # noqa: C901
         #    -getting an aws trigger (key == aws_s3)
         # otherwise, slack api token data perhaps
         # store the info and don't re-render
-
         if canarydrop.type == TokenTypes.AWS_KEYS:
             token_hit = Canarytoken._parse_aws_key_trigger(request)
             if isinstance(token_hit, AWSKeyTokenHit):
@@ -237,8 +237,34 @@ def render_POST(self, request: Request):  # noqa: C901
                 )
                 # TODO: These returns are not really needed
                 return b"failed"
-        else:
-            return self.render_GET(request)
+        elif canarydrop.type == TokenTypes.IDP_APP:
+            if SAML_POST_ARG in request.args:
+                return self.render_GET(request)
+            key = request.args.get(b"key", [None])[0]
+            if (key := coerce_to_float(key)) and token:
+                additional_info = {
+                    k.decode(): v
+                    for k, v in request.args.items()
+                    if k.decode() not in ["key", "canarytoken", "name"]
+                }
+                canarydrop.add_additional_info_to_hit(
+                    hit_time=key,
+                    additional_info={
+                        request.args[b"name"][0].decode(): additional_info
+                    },
+                )
+                self.dispatch(
+                    canarydrop=canarydrop,
+                    token_hit=canarydrop.triggered_details.hits[-1],
+                )
+                return b"success"
+            else:
+                log.info(
+                    f"Either {key=} or {token=} were falsy. Dropping this request."
+                )
+                # TODO: These returns are not really needed
+                return b"failed"
+        return self.render_GET(request)
 
 
 class ChannelHTTP:

canarytokens/models.py

@@ -304,6 +304,7 @@ class TokenTypes(str, enum.Enum):
     WINDOWS_FAKE_FS = "windows_fake_fs"
     CC = "cc"
     PWA = "pwa"
+    IDP_APP = "idp_app"
     SLACK_API = "slack_api"
     LEGACY = "legacy"
 
@@ -352,6 +353,7 @@ def __str__(self) -> str:
     TokenTypes.PWA: "Fake app",
     TokenTypes.SLACK_API: "Slack API",
     TokenTypes.LEGACY: "Legacy",
+    TokenTypes.IDP_APP: "SAML2 IdP App",
 }
 
 GeneralHistoryTokenType = Literal[
@@ -826,6 +828,91 @@ class CreditCardV2TokenRequest(TokenRequest):
     cf_turnstile_response: Optional[str]
 
 
+class IdPAppType(enum.Enum):
+    AWS = "aws"
+    AZURE = "azure"
+    BITWARDEN = "bitwarden"
+    DROPBOX = "dropbox"
+    DUO = "duo"
+    ELASTICSEARCH = "elasticsearch"
+    FRESHBOOKS = "freshbooks"
+    GCLOUD = "gcloud"
+    GDRIVE = "gdrive"
+    GITHUB = "github"
+    GITLAB = "gitlab"
+    GMAIL = "gmail"
+    INTUNE = "intune"
+    JAMF = "jamf"
+    JIRA = "jira"
+    KIBANA = "kibana"
+    LASTPASS = "lastpass"
+    MS365 = "ms365"
+    MSTEAMS = "msteams"
+    ONEDRIVE = "onedrive"
+    ONEPASSWORD = "onepassword"
+    OUTLOOK = "outlook"
+    PAGERDUTY = "pagerduty"
+    SAGE = "sage"
+    SALESFORCE = "salesforce"
+    SAP = "sap"
+    SLACK = "slack"
+    VIRTRU = "virtru"
+    ZENDESK = "zendesk"
+    ZOHO = "zoho"
+    ZOOM = "zoom"
+
+
+IDP_APP_TITLES = {
+    IdPAppType.AWS: "AWS",
+    IdPAppType.AZURE: "Azure",
+    IdPAppType.BITWARDEN: "Bitwarden",
+    IdPAppType.DROPBOX: "Dropbox",
+    IdPAppType.DUO: "Duo",
+    IdPAppType.ELASTICSEARCH: "Elasticsearch",
+    IdPAppType.FRESHBOOKS: "Freshbooks",
+    IdPAppType.GCLOUD: "Google Cloud",
+    IdPAppType.GDRIVE: "Google Drive",
+    IdPAppType.GITHUB: "GitHub",
+    IdPAppType.GITLAB: "GitLab",
+    IdPAppType.GMAIL: "Gmail",
+    IdPAppType.INTUNE: "Intune",
+    IdPAppType.JAMF: "JAMF",
+    IdPAppType.JIRA: "Jira",
+    IdPAppType.KIBANA: "Kibana",
+    IdPAppType.LASTPASS: "LastPass",
+    IdPAppType.MS365: "Microsoft 365",
+    IdPAppType.MSTEAMS: "MS Teams",
+    IdPAppType.ONEDRIVE: "OneDrive",
+    IdPAppType.ONEPASSWORD: "1Password",
+    IdPAppType.OUTLOOK: "Outlook",
+    IdPAppType.PAGERDUTY: "PagerDuty",
+    IdPAppType.SAGE: "Sage",
+    IdPAppType.SALESFORCE: "Salesforce",
+    IdPAppType.SAP: "SAP",
+    IdPAppType.SLACK: "Slack",
+    IdPAppType.VIRTRU: "Virtru",
+    IdPAppType.ZENDESK: "Zendesk",
+    IdPAppType.ZOHO: "Zoho",
+    IdPAppType.ZOOM: "Zoom",
+}
+
+
+class IdPAppTokenRequest(TokenRequest):
+    token_type: Literal[TokenTypes.IDP_APP] = TokenTypes.IDP_APP
+    app_type: IdPAppType
+    redirect_url: Optional[str] = None
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "token_type": TokenTypes.IDP_APP,
+                "memo": "Reminder note when this token is triggered",
+                "email": "[email protected]",
+                "redirect_url": "https://youtube.com",
+            },
+        }
+
+
 AnyTokenRequest = Annotated[
     Union[
         CCTokenRequest,
@@ -856,6 +943,7 @@ class CreditCardV2TokenRequest(TokenRequest):
         SQLServerTokenRequest,
         KubeconfigTokenRequest,
         CreditCardV2TokenRequest,
+        IdPAppTokenRequest,
     ],
     Field(discriminator="token_type"),
 ]
@@ -1160,6 +1248,12 @@ class CreditCardV2TokenResponse(TokenResponse):
     expiry_year: int
 
 
+class IdPAppTokenResponse(TokenResponse):
+    token_type: Literal[TokenTypes.IDP_APP] = TokenTypes.IDP_APP
+    entity_id: str
+    app_type: IdPAppType
+
+
 AnyTokenResponse = Annotated[
     Union[
         CCTokenResponse,
@@ -1200,6 +1294,7 @@ class CreditCardV2TokenResponse(TokenResponse):
         MsExcelDocumentTokenResponse,
         KubeconfigTokenResponse,
         CreditCardV2TokenResponse,
+        IdPAppTokenResponse,
     ],
     Field(discriminator="token_type"),
 ]
@@ -1853,6 +1948,11 @@ class LegacyTokenHit(TokenHit):
     mail: Optional[SMTPMailField]
 
 
+class IdPAppTokenHit(TokenHit):
+    token_type: Literal[TokenTypes.IDP_APP] = TokenTypes.IDP_APP
+    additional_info: AdditionalInfo = AdditionalInfo()
+
+
 AnyTokenHit = Annotated[
     Union[
         CCTokenHit,
@@ -1886,6 +1986,7 @@ class LegacyTokenHit(TokenHit):
         KubeconfigTokenHit,
         LegacyTokenHit,
         CreditCardV2TokenHit,
+        IdPAppTokenHit,
     ],
     Field(discriminator="token_type"),
 ]
@@ -2110,6 +2211,11 @@ class LegacyTokenHistory(TokenHistory[LegacyTokenHit]):
     hits: List[LegacyTokenHit] = []
 
 
+class IdPAppTokenHistory(TokenHistory[IdPAppTokenHit]):
+    token_type: Literal[TokenTypes.IDP_APP] = TokenTypes.IDP_APP
+    hits: List[IdPAppTokenHit] = []
+
+
 # AnyTokenHistory is used to type annotate functions that
 # handle any token history. It makes use of an annotated type
 # that discriminates on `token_type` so pydantic can parse
@@ -2147,6 +2253,7 @@ class LegacyTokenHistory(TokenHistory[LegacyTokenHit]):
         KubeconfigTokenHistory,
         LegacyTokenHistory,
         CreditCardV2TokenHistory,
+        IdPAppTokenHistory,
     ],
     Field(discriminator="token_type"),
 ]

canarytokens/queries.py

@@ -374,6 +374,7 @@ def add_additional_info_to_hit(canarytoken, hit_time, additional_info):
             models.SlowRedirectTokenHit,
             models.CustomImageTokenHit,
             models.WebBugTokenHit,
+            models.IdPAppTokenHit,
         ),
     ):
         info = enriched_hit.additional_info.dict(exclude_unset=True, exclude_none=None)

canarytokens/saml.py

@@ -0,0 +1,22 @@
+import re
+from base64 import b64decode
+from typing import Optional
+from twisted.web.http import Request
+
+SAML_POST_ARG = b"SAMLResponse"
+
+
+def prepare_request(request: Request) -> Optional[bytes]:
+    return request.args.pop(SAML_POST_ARG, [None])[0]
+
+
+def extract_identity(saml_request: bytes) -> Optional[str]:
+    data = b64decode(saml_request).decode()
+
+    SEARCH_PATTERN = re.compile(r"<(saml2:)?NameID[^>]*>(.+)</(saml2:)?NameID>")
+    result = SEARCH_PATTERN.search(data)
+    if result is None:
+        return None
+
+    email: str = result.groups()[1]
+    return email

canarytokens/tokens.py

@@ -14,6 +14,7 @@
 from twisted.web.http import Request
 from twisted.web.util import redirectTo
 
+from canarytokens.saml import extract_identity, prepare_request
 from canarytokens.settings import SwitchboardSettings
 
 from canarytokens import canarydrop, msreg, queries
@@ -655,6 +656,24 @@ def _get_response_for_fast_redirect(
             return redirectTo(redirect_url.encode(), request)
         return GIF
 
+    @staticmethod
+    def _get_info_for_idp_app(request: Request):
+        src_data = {}
+        saml_request = prepare_request(request)
+        if saml_request:
+            identity = extract_identity(saml_request)
+            src_data["identity"] = identity
+        http_general_info = Canarytoken._grab_http_general_info(request=request)
+        return http_general_info, {"src_data": src_data}
+
+    @staticmethod
+    def _get_response_for_idp_app(canarydrop: canarydrop.Canarydrop, request: Request):
+        if not canarydrop.redirect_url:
+            return Canarytoken._get_response_for_web(canarydrop, request)
+        if not canarydrop.browser_scanner_enabled:
+            return Canarytoken._get_response_for_fast_redirect(canarydrop, request)
+        return Canarytoken._get_response_for_slow_redirect(canarydrop, request)
+
     @staticmethod
     def _get_info_for_slow_redirect(request):
         http_general_info = Canarytoken._grab_http_general_info(request=request)
@@ -700,7 +719,7 @@ def _get_response_for_web(
                 # set-up response template
                 browser_scanner_template_params = {
                     "key": latest_hit_time,
-                    "canarytoken": canarydrop.canarytoken.value,
+                    "canarytoken": canarydrop.canarytoken.value(),
                     "redirect_url": "",
                 }
                 template = get_template_env().get_template("browser_scanner.html")
@@ -775,7 +794,7 @@ def _get_response_for_web_image(
                 # set-up response template
                 browser_scanner_template_params = {
                     "key": latest_hit_time,
-                    "canarytoken": canarydrop.canarytoken.value,
+                    "canarytoken": canarydrop.canarytoken.value(),
                     "redirect_url": "",
                 }
                 template = get_template_env().get_template("browser_scanner.html")
@@ -848,7 +867,7 @@ def _get_response_for_legacy(canarydrop: canarydrop.Canarydrop, request: Request
                 # set-up response template
                 browser_scanner_template_params = {
                     "key": latest_hit_time,
-                    "canarytoken": canarydrop.canarytoken.value,
+                    "canarytoken": canarydrop.canarytoken.value(),
                     "redirect_url": "",
                 }
                 template = get_template_env().get_template("browser_scanner.html")