Add handling for non-latin-1 characters in webhook headers

CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## v1.23.0 - Unreleased
+
+### Bug Fixes
+- Add handling for non-latin-1 characters in webhook headers
+
 ## v1.22.0 - 2020-08-06
 
 ### Improvements

hc/api/tests/test_notify_webhook.py

@@ -342,3 +342,20 @@ class NotifyWebhookTestCase(BaseTestCase):
 
         n = Notification.objects.get()
         self.assertEqual(n.error, "Webhook notifications are not enabled.")
+
+    @patch("hc.api.transports.requests.request")
+    def test_webhooks_handle_non_ascii_in_headers(self, mock_request):
+        definition = {
+            "method_down": "GET",
+            "url_down": "http://foo.com",
+            "headers_down": {"X-Foo": "bār"},
+            "body_down": "",
+        }
+
+        self._setup_data(json.dumps(definition))
+        self.check.save()
+
+        self.channel.notify(self.check)
+        args, kwargs = mock_request.call_args
+
+        self.assertEqual(kwargs["headers"]["X-Foo"], "bār")

hc/api/transports.py

@@ -203,7 +203,7 @@ class HttpTransport(Transport):
 
 
 class Webhook(HttpTransport):
-    def prepare(self, template, check, urlencode=False):
+    def prepare(self, template, check, urlencode=False, latin1=False):
         """ Replace variables with actual values. """
 
         def safe(s):
@@ -220,7 +220,12 @@ class Webhook(HttpTransport):
         for i, tag in enumerate(check.tags_list()):
             ctx["$TAG%d" % (i + 1)] = safe(tag)
 
-        return replace(template, ctx)
+        result = replace(template, ctx)
+        if latin1:
+            # Replace non-latin-1 characters with XML character references.
+            result = result.encode("latin-1", "xmlcharrefreplace").decode()
+
+        return result
 
     def is_noop(self, check):
         if check.status == "down" and not self.channel.url_down:
@@ -242,7 +247,8 @@ class Webhook(HttpTransport):
         url = self.prepare(spec["url"], check, urlencode=True)
         headers = {}
         for key, value in spec["headers"].items():
-            headers[key] = self.prepare(value, check)
+            # Header values should contain ASCII and latin-1 only
+            headers[key] = self.prepare(value, check, latin1=True)
 
         body = spec["body"]
         if body:

hc/front/forms.py

@@ -15,6 +15,14 @@ from hc.front.validators import (
 import requests
 
 
+def _is_latin1(s):
+    try:
+        s.encode("latin-1")
+        return True
+    except UnicodeError:
+        return False
+
+
 class HeadersField(forms.Field):
     message = """Use "Header-Name: value" pairs, one per line."""
 
@@ -35,6 +43,11 @@ class HeadersField(forms.Field):
             if not n or not v:
                 raise ValidationError(message=self.message)
 
+            if not _is_latin1(n):
+                raise ValidationError(
+                    message="Header names must not contain special characters"
+                )
+
             headers[n] = v
 
         return headers

hc/front/tests/test_add_webhook.py

@@ -149,6 +149,19 @@ class AddWebhookTestCase(BaseTestCase):
         self.assertContains(r, """invalid-header""")
         self.assertEqual(Channel.objects.count(), 0)
 
+    def test_it_rejects_non_latin1_in_header_name(self):
+        self.client.login(username="[email protected]", password="password")
+        form = {
+            "method_down": "GET",
+            "url_down": "http://example.org",
+            "headers_down": "fō:bar",
+            "method_up": "GET",
+        }
+
+        r = self.client.post(self.url, form)
+        self.assertContains(r, """must not contain special characters""")
+        self.assertEqual(Channel.objects.count(), 0)
+
     def test_it_strips_headers(self):
         form = {
             "method_down": "GET",