Doyensec fixes (#520)

* Replace requests with advocate to prevent blind SSRF

* Update python-multipart to fix GHSA-2jv5-9r88-3w3p

* Fix stored XSS in slow redirect token

canarytokens/channel_output_webhook.py

@@ -3,6 +3,7 @@
 """
 from typing import Dict
 
+import advocate
 import requests
 from pydantic import HttpUrl
 from twisted.logger import Logger
@@ -63,16 +64,24 @@ def generic_webhook_send(
     ) -> bool:
         # Design: wrap in a retry?
         try:
-            response = requests.post(
-                url=str(alert_webhook_url), json=payload, timeout=(2, 2)
+            validator = advocate.AddrValidator(port_whitelist=set(range(0, 65535)))
+            response = advocate.post(
+                url=str(alert_webhook_url),
+                json=payload,
+                timeout=(2, 2),
+                validator=validator,
             )
             response.raise_for_status()
             log.info(f"Successfully sent to {alert_webhook_url}")
             return True
-        except requests.exceptions.HTTPError:
+        except advocate.HTTPError:
             log.debug(
                 f"Failed sending request to webhook {alert_webhook_url}.",
             )
+        except advocate.exceptions.UnacceptableAddressException:
+            log.debug(
+                f"Disallowed requests to {alert_webhook_url}.",
+            )
         except (requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
             log.debug(
                 f"Failed connecting to webhook {alert_webhook_url}.",

canarytokens/queries.py

@@ -9,6 +9,7 @@
 from ipaddress import IPv4Address
 from typing import Dict, List, Literal, Optional, Tuple, Union
 
+import advocate
 import requests
 from pydantic import EmailStr, HttpUrl, ValidationError, parse_obj_as
 from twisted.logger import Logger
@@ -908,11 +909,12 @@ def validate_webhook(url, token_type: models.TokenTypes):
             },
             time=datetime.datetime.now(),
         )
-    response = requests.post(
+    response = advocate.post(
         url,
         payload.json(),
         headers={"content-type": "application/json"},
         timeout=10,
+        validator=advocate.AddrValidator(port_whitelist=set(range(0, 65535))),
     )
     # TODO: this accepts 3xx which is probably too lenient. We probably want any 2xx code.
     response.raise_for_status()

canarytokens/tokens.py

@@ -523,7 +523,9 @@ def _get_response_for_fast_redirect(
     ):
         redirect_url = canarydrop.redirect_url
         if redirect_url:
-            if ":" not in redirect_url:
+            if not redirect_url.startswith("http://") and not redirect_url.startswith(
+                "https://"
+            ):
                 redirect_url = "http://" + redirect_url
             return redirectTo(redirect_url.encode(), request)
         return GIF
@@ -544,7 +546,9 @@ def _get_response_for_slow_redirect(
         canarydrop: canarydrop.Canarydrop, request: Request
     ) -> bytes:
         redirect_url = canarydrop.redirect_url
-        if redirect_url and ":" not in redirect_url:
+        if not redirect_url.startswith("http://") and not redirect_url.startswith(
+            "https://"
+        ):
             redirect_url = "http://" + redirect_url
         template = get_template_env().get_template("browser_scanner.html")
         return template.render(

pyproject.toml

@@ -13,7 +13,7 @@ redis = "^4.2.0"
 httpx = "^0.23.0"
 Jinja2 = "^3.1.2"
 requests = "^2.27.1"
-python-multipart = "^0.0.5"
+python-multipart = "^0.0.9"
 pyOpenSSL = "^23.0.0"
 service-identity = "^21.1.0"
 PyYAML = "^6.0"
@@ -32,6 +32,7 @@ faker = "^17.6.0"
 azure-core = "^1.23.1"
 azure-identity = "^1.10.0"
 cssutils = "^1.0.2"
+advocate = {git = "https://github.com/JordanMilne/Advocate.git"}
 
 [tool.poetry.extras]
 web = ["fastapi", "uvicorn", "sentry-sdk"]
@@ -63,6 +64,7 @@ pytest-memray = {version = "^1.0.0", platform = "linux"}
 python-docx = "^0.8.11"
 boto3 = "^1.23.10"
 boto3-stubs = {extras = ["essential"], version = "^1.24.17"}
+botocore = "^1.34.141"
 modernize = "^0.8.0"
 deepdiff = "^5.8.1"
 pypdf = "^4.2.0"

templates/browser_scanner.html

@@ -2,7 +2,7 @@
     {% if redirect_url %}
     <head>
         <link rel="shortcut icon" href="/resources/favicon.ico">
-        <noscript><meta http-equiv="refresh" content="0; {{redirect_url}}"></noscript>
+        <noscript><meta http-equiv="refresh" content="0; {{redirect_url|e}}"></noscript>
     </head>
     {% endif %}    
 <body>
@@ -498,7 +498,7 @@
       detectJava();
 
       {% if redirect_url %}
-      window.location = '{{redirect_url}}';
+      window.location = {{redirect_url|tojson}};
       {% endif %}
 
       return;

tests/units/test_channel_output_webhook.py

@@ -1,4 +1,6 @@
 # pytest caplog
+import pytest
+
 from twisted.logger import capturedLogs
 
 from canarytokens.canarydrop import Canarydrop
@@ -338,3 +340,58 @@ def test_canaryalert_ms_teams_webhook(
         host=input_channel.hostname,
     )
     assert isinstance(canaryalert_webhook_payload, TokenAlertDetailsMsTeams)
+
+
+@pytest.mark.parametrize(
+    "bad_url",
+    [
+        "http://127.0.0.1",
+        "http://169.254.196.254",
+        "https://localhost",
+        "https://dev-docker.canary.tools",
+    ],
+)
+def test_ssrf_protection(
+    setup_db,
+    frontend_settings: FrontendSettings,
+    settings: SwitchboardSettings,
+    bad_url: str,
+):
+    switchboard = Switchboard()
+    switchboard.switchboard_settings = settings
+    webhook_channel = WebhookOutputChannel(
+        switchboard=switchboard,
+        switchboard_scheme=settings.SWITCHBOARD_SCHEME,
+        frontend_domain="test.com",
+    )
+    cd = Canarydrop(
+        type=TokenTypes.DNS,
+        generate=True,
+        alert_email_enabled=False,
+        alert_email_recipient="[email protected]",
+        alert_webhook_enabled=False,
+        alert_webhook_url=bad_url,
+        canarytoken=Canarytoken(),
+        memo="memo",
+        browser_scanner_enabled=False,
+    )
+
+    token_hit = Canarytoken.create_token_hit(
+        token_type=TokenTypes.DNS,
+        input_channel="not_valid",
+        src_ip="127.0.0.1",
+        hit_info={"some": "data"},
+    )
+    cd.add_canarydrop_hit(token_hit=token_hit)
+    with capturedLogs() as captured:
+        webhook_channel.send_alert(
+            canarydrop=cd,
+            token_hit=token_hit,
+            input_channel=ChannelDNS(
+                switchboard=switchboard,
+                frontend_settings=frontend_settings,
+                switchboard_hostname="test.com",
+                switchboard_scheme=settings.SWITCHBOARD_SCHEME,
+            ),
+        )
+    assert any(["Disallowed requests to" in log["log_format"] for log in captured])