Custom headers should have validation

changedetectionio/forms.py

@@ -224,27 +224,37 @@ class StringDictKeyValue(StringField):
 
     def _value(self):
         if self.data:
-            output = u''
-            for k in self.data.keys():
-                output += "{}: {}\r\n".format(k, self.data[k])
-
+            output = ''
+            for k, v in self.data.items():
+                output += f"{k}: {v}\r\n"
             return output
         else:
-            return u''
+            return ''
 
-    # incoming
+    # incoming data processing + validation
     def process_formdata(self, valuelist):
+        self.data = {}
+        errors = []
         if valuelist:
-            self.data = {}
-            # Remove empty strings
-            cleaned = list(filter(None, valuelist[0].split("\n")))
-            for s in cleaned:
-                parts = s.strip().split(':', 1)
-                if len(parts) == 2:
-                    self.data.update({parts[0].strip(): parts[1].strip()})
-
-        else:
-            self.data = {}
+            # Remove empty strings (blank lines)
+            cleaned = [line.strip() for line in valuelist[0].split("\n") if line.strip()]
+            for idx, s in enumerate(cleaned, start=1):
+                if ':' not in s:
+                    errors.append(f"Line {idx} is missing a ':' separator.")
+                    continue
+                parts = s.split(':', 1)
+                key = parts[0].strip()
+                value = parts[1].strip()
+
+                if not key:
+                    errors.append(f"Line {idx} has an empty key.")
+                if not value:
+                    errors.append(f"Line {idx} has an empty value.")
+
+                self.data[key] = value
+
+        if errors:
+            raise ValidationError("Invalid input:\n" + "\n".join(errors))
 
 class ValidateContentFetcherIsReady(object):
     """

changedetectionio/tests/test_request.py

@@ -424,3 +424,27 @@ def test_headers_textfile_in_request(client, live_server, measure_memory_usage):
     # unlink headers.txt on start/stop
     res = client.get(url_for("ui.form_delete", uuid="all"), follow_redirects=True)
     assert b'Deleted' in res.data
+
+def test_headers_validation(client, live_server):
+    #live_server_setup(live_server)
+
+    test_url = url_for('test_headers', _external=True)
+    res = client.post(
+        url_for("imports.import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+    res = client.post(
+        url_for("ui.ui_edit.edit_page", uuid="first"),
+        data={
+            "url": test_url,
+            "fetch_backend": 'html_requests',
+            "headers": "User-AGent agent-from-watch\r\nsadfsadfsadfsdaf\r\n:foobar"},
+        follow_redirects=True
+    )
+
+    assert b"Line 1 is missing a ':' separator." in res.data
+    assert b"Line 3 has an empty key." in res.data
+