fix: prevent web-test errors and validate domain inputs

.github/workflows/django.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       max-parallel: 4
       matrix:
-        python-version: [3.9, 3.11.4]
+        python-version: ["3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v3

testing/forms.py

@@ -1,6 +1,46 @@
+from urllib.parse import urlsplit
+
+import idna
 from django import forms
+from django.core.exceptions import ValidationError
+from django.core.validators import DomainNameValidator
+
+from .models import DMARCRecord, MailDomain, TestReport
+
+
+class WebTestForm(forms.Form):
+    invalid_target = (
+        "Enter a valid domain such as test-domain.lu or www.test-domain.lu, "
+        "or an HTTP(S) website URL without credentials or a port number."
+    )
+    target = forms.CharField(
+        max_length=2048,
+        error_messages={"required": invalid_target, "max_length": invalid_target},
+    )
 
-from .models import DMARCRecord, MailDomain
+    def clean_target(self):
+        value = self.cleaned_data["target"]
+        try:
+            if "\\" in value or any(character.isspace() for character in value):
+                raise ValueError("Invalid whitespace or backslash")
+            parsed = urlsplit(value if "://" in value else f"https://{value}")
+            if (
+                parsed.scheme not in ("http", "https")
+                or parsed.username is not None
+                or parsed.password is not None
+                or parsed.port is not None
+                or not parsed.hostname
+            ):
+                raise ValueError("Expected a website hostname")
+            DomainNameValidator()(parsed.hostname)
+            domain = idna.encode(parsed.hostname.removesuffix("."), uts46=True).decode(
+                "ascii"
+            )
+            if len(domain) > TestReport._meta.get_field("tested_site").max_length:
+                raise ValueError("Domain is too long for a saved report")
+            return domain
+        except (ValidationError, ValueError, UnicodeError):
+            raise forms.ValidationError(self.invalid_target) from None
 
 
 class DMARCRecordForm(forms.ModelForm):

testing/helpers.py

@@ -931,8 +931,12 @@ def analyze_csp(csp, result, header_name):
 
 
 def parse_csp(csp):
-    return dict(
-        directive.split(None, 1) for directive in csp.split(';') if directive.strip())
+    directives = {}
+    for directive in csp.split(';'):
+        parts = directive.split(None, 1)
+        if parts:
+            directives.setdefault(parts[0].lower(), parts[1] if len(parts) > 1 else '')
+    return directives
 
 
 def check_unsafe_directives(directives, result, header_name):
@@ -967,7 +971,9 @@ def check_overly_permissive_directives(directives, result, header_name):
 
 
 def check_csp_syntax(csp, result, header_name):
-    if not re.match(r'^[a-zA-Z0-9\-]+\s+[^;]+(?:;\s*[a-zA-Z0-9\-]+\s+[^;]+)*$', csp):
+    directives = (part.strip() for part in csp.split(';') if part.strip())
+    if any(not re.fullmatch(r'[a-zA-Z0-9-]+(?:[ \t]+[^\r\n;,]*)?', part)
+           for part in directives):
         result['issues'].append(f"{header_name}: CSP syntax appears to be invalid.")
         result['recommendations'].append("Review and correct the CSP syntax.")
 
@@ -1022,7 +1028,7 @@ def check_cookies(domain: str) -> Dict[str, Any]:
             'cookies': cookie_details,
             'message': message
         }
-    except RequestException as e:
+    except requests.RequestException as e:
         return {
             'status': False,
             'cookies': [],
@@ -1322,7 +1328,18 @@ def check_hsts(domain: str) -> Dict[str, Union[bool, str, Dict[str, Union[str, b
         response = requests.get(url, timeout=10)
         response.raise_for_status()
         hsts_header = response.headers.get('strict-transport-security')
-        parsed_hsts = parse_hsts_header(hsts_header) if hsts_header else {}
+        try:
+            parsed_hsts = parse_hsts_header(hsts_header) if hsts_header else {}
+        except ValueError as e:
+            return {
+                'status': False,
+                'data': f'Invalid HSTS header: {e}',
+                'parsed': {},
+                'http_status': response.status_code,
+                'preload_ready': False,
+                'strength': 'Invalid',
+                'recommendations': ['Provide one non-negative integer max-age value.']
+            }
 
         preload_ready = parsed_hsts.get('preload', False)
         strength, recommendations = evaluate_hsts_strength(parsed_hsts)
@@ -1380,13 +1397,21 @@ def parse_hsts_header(header: str) -> Dict[str, Union[str, bool, int]]:
     components = header.split(';')
     parsed = {}
     for component in components:
-        component = component.strip().lower()
-        if component.startswith('max-age='):
-            parsed['max-age'] = int(component.split('=')[1])
-        elif component == 'includesubdomains':
+        name, separator, value = component.partition('=')
+        name = name.strip().lower()
+        if name == 'max-age':
+            value = value.strip()
+            if len(value) >= 2 and value.startswith('"') and value.endswith('"'):
+                value = value[1:-1]
+            if not separator or not re.fullmatch(r'[0-9]+', value) or 'max-age' in parsed:
+                raise ValueError('max-age must appear once and contain a non-negative integer.')
+            parsed['max-age'] = int(value)
+        elif name == 'includesubdomains' and not separator:
             parsed['includeSubDomains'] = True
-        elif component == 'preload':
+        elif name == 'preload' and not separator:
             parsed['preload'] = True
+    if 'max-age' not in parsed:
+        raise ValueError('The required max-age directive is missing.')
     return parsed
 
 

testing/templates/check_webapp.html

@@ -31,9 +31,10 @@ <h1 class="font-size-48 font-weight-800 lh-base">Assesses web security, includin
                 <span class="text-danger">{{ error }}</span>
                 {% endif %}
                 <div class="mb-3">
-                    <label class="mb-1">Enter your domain name:</label>
+                    <label class="mb-1" for="target">Enter your domain name or website URL:</label>
                     <input class="form-control" type="text" placeholder="domain.com" id="target" name="target"
-                           value="{{ domain }}" required>
+                           value="{% firstof target domain %}" maxlength="2048" required>
+                    <small class="form-text">For example, test-domain.lu or https://www.test-domain.lu/. The tests check the hostname's homepage.</small>
                 </div>
                 <input type="submit" value="Test"
                        class="text-white button-text-background primary-primary-blue_background">

testing/test_web_checks.py

@@ -0,0 +1,216 @@
+from unittest.mock import patch
+
+import requests
+from django.test import SimpleTestCase, TestCase
+
+from testing import helpers
+from testing.models import TestReport
+
+
+def web_response(headers=None, status=200):
+    response = requests.Response()
+    response.status_code = status
+    response.url = "https://test-domain.lu/"
+    response._content = b"<html><body>Test website</body></html>"
+    response.headers.update(headers or {})
+    return response
+
+
+class SecurityHeaderTests(SimpleTestCase):
+    def test_csp_accepts_valueless_directives_and_empty_segments(self):
+        self.assertEqual(
+            helpers.parse_csp(
+                " ; default-src 'self'; upgrade-insecure-requests; sandbox;; "
+            ),
+            {"default-src": "'self'", "upgrade-insecure-requests": "", "sandbox": ""},
+        )
+
+    def test_csp_keeps_first_case_insensitive_directive(self):
+        self.assertEqual(
+            helpers.parse_csp("DEFAULT-SRC 'self'; default-src *"),
+            {"default-src": "'self'"},
+        )
+
+    def test_csp_syntax_accepts_flags_and_trailing_semicolons(self):
+        result = {"issues": [], "recommendations": []}
+        helpers.check_csp_syntax(
+            "default-src 'self'; upgrade-insecure-requests; sandbox;", result, "CSP"
+        )
+        self.assertEqual(result["issues"], [])
+
+    def test_csp_syntax_reports_invalid_directive_names(self):
+        result = {"issues": [], "recommendations": []}
+        helpers.check_csp_syntax("default_src 'self'", result, "CSP")
+        self.assertTrue(result["issues"])
+
+    def test_cookie_http_error_returns_a_failed_check(self):
+        with patch(
+            "testing.helpers.requests.get", return_value=web_response(status=403)
+        ):
+            result = helpers.check_cookies("test-domain.lu")
+        self.assertFalse(result["status"])
+        self.assertEqual(result["cookies"], [])
+        self.assertIn("403", result["message"])
+
+    def test_cookie_network_errors_return_a_failed_check(self):
+        for error in (
+            requests.ConnectionError("DNS failure"),
+            requests.Timeout("Timed out"),
+        ):
+            with self.subTest(error=type(error).__name__):
+                with patch("testing.helpers.requests.get", side_effect=error):
+                    result = helpers.check_cookies("test-domain.lu")
+                self.assertFalse(result["status"])
+                self.assertEqual(result["cookies"], [])
+                self.assertIn(str(error), result["message"])
+
+    def test_malformed_hsts_returns_a_failed_check(self):
+        for header in (
+            "max-age=invalid",
+            "max-age=",
+            "max-age=-1",
+            "includeSubDomains",
+            "max-age=1; max-age=2",
+            'max-age="31536000',
+        ):
+            with self.subTest(header=header):
+                response = web_response({"Strict-Transport-Security": header})
+                with patch("testing.helpers.requests.get", return_value=response):
+                    result = helpers.check_hsts("test-domain.lu")
+                self.assertFalse(result["status"])
+                self.assertEqual(result["http_status"], 200)
+                self.assertIn("Invalid HSTS", result["data"])
+
+    def test_hsts_accepts_quoted_max_age(self):
+        self.assertEqual(
+            helpers.parse_hsts_header('max-age="31536000"; includeSubDomains; preload'),
+            {"max-age": 31536000, "includeSubDomains": True, "preload": True},
+        )
+
+
+class WebTestInputTests(TestCase):
+    endpoint = "/infra-testing/web-test/"
+
+    def test_get_displays_the_form_without_making_requests(self):
+        with patch(
+            "requests.sessions.Session.request",
+            side_effect=AssertionError("Unexpected outbound request"),
+        ):
+            response = self.client.get(self.endpoint)
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, 'name="target"')
+
+    def test_valid_targets_are_normalized_before_checks_and_saved(self):
+        examples = {
+            "test-domain.lu": "test-domain.lu",
+            "www.test-domain.lu": "www.test-domain.lu",
+            "  WWW.Test-Domain.LU  ": "www.test-domain.lu",
+            "https://www.test-domain.lu/path?query=1#section": "www.test-domain.lu",
+            "http://test-domain.lu/": "test-domain.lu",
+            "test-domain.lu/": "test-domain.lu",
+            "test-domain.lu.": "test-domain.lu",
+            "münich.lu": "xn--mnich-kva.lu",
+        }
+        for target, expected in examples.items():
+            with self.subTest(target=target):
+                with patch(
+                    "requests.sessions.Session.request", return_value=web_response()
+                ) as network:
+                    response = self.client.post(self.endpoint, {"target": target})
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(response.context["domain"], expected)
+                self.assertEqual(
+                    network.call_args_list[0].kwargs["url"], f"https://{expected}"
+                )
+                self.assertTrue(
+                    TestReport.objects.filter(
+                        tested_site=expected, test_ran="web-test"
+                    ).exists()
+                )
+
+    def test_invalid_targets_do_not_make_requests_or_save_reports(self):
+        examples = (
+            "",
+            " ",
+            "www.",
+            "www",
+            "https://",
+            "test domain.lu",
+            "test_domain.lu",
+            "-test.lu",
+            "test-.lu",
+            "test..lu",
+            "(test-domain.lu)",
+            "test.lu\\path",
+            "https://user:password@test-domain.lu/",
+            "ftp://test-domain.lu/",
+            "test-domain.lu:8443",
+            "test-domain.lu:invalid",
+            "https://[invalid/",
+            "test\ndomain.lu",
+            "localhost",
+            "127.0.0.1",
+            "a" * 64 + ".lu",
+        )
+        for target in examples:
+            with self.subTest(target=target):
+                with self.assertLogs("django.request", level="WARNING"), patch(
+                    "requests.sessions.Session.request",
+                    side_effect=AssertionError("Unexpected outbound request"),
+                ):
+                    response = self.client.post(self.endpoint, {"target": target})
+                self.assertEqual(response.status_code, 400)
+                self.assertTrue(response.context["error"])
+                self.assertContains(response, "Enter a valid", status_code=400)
+                self.assertNotContains(response, "Overview of", status_code=400)
+        self.assertEqual(TestReport.objects.count(), 0)
+
+    def test_missing_target_is_a_validation_error(self):
+        with self.assertLogs("django.request", level="WARNING"), patch(
+            "requests.sessions.Session.request",
+            side_effect=AssertionError("Unexpected outbound request"),
+        ):
+            response = self.client.post(self.endpoint, {})
+        self.assertEqual(response.status_code, 400)
+        self.assertTrue(response.context["error"])
+
+    def test_invalid_input_is_preserved_and_escaped(self):
+        with self.assertLogs("django.request", level="WARNING"), patch(
+            "requests.sessions.Session.request",
+            side_effect=AssertionError("Unexpected outbound request"),
+        ):
+            response = self.client.post(
+                self.endpoint, {"target": '<script>alert("x")</script>'}
+            )
+        self.assertContains(response, "&lt;script&gt;", status_code=400)
+        self.assertNotContains(response, '<script>alert("x")</script>', status_code=400)
+
+    def test_unresolvable_target_returns_results_instead_of_500(self):
+        with patch(
+            "requests.sessions.Session.request",
+            side_effect=requests.ConnectionError("DNS failure"),
+        ):
+            response = self.client.post(self.endpoint, {"target": "test-domain.lu"})
+        self.assertEqual(response.status_code, 200)
+        self.assertFalse(response.context["cookies_result"]["status"])
+        self.assertTrue(
+            TestReport.objects.filter(tested_site="test-domain.lu").exists()
+        )
+
+    def test_valid_csp_and_forbidden_target_do_not_break_results(self):
+        for result in (
+            web_response(
+                {
+                    "Content-Security-Policy": "default-src 'self'; upgrade-insecure-requests;"
+                }
+            ),
+            web_response(status=403),
+            web_response({"Strict-Transport-Security": "max-age=invalid"}),
+        ):
+            with self.subTest(headers=dict(result.headers), status=result.status_code):
+                with patch("requests.sessions.Session.request", return_value=result):
+                    response = self.client.post(
+                        self.endpoint, {"target": "test-domain.lu"}
+                    )
+                self.assertEqual(response.status_code, 200)
+                self.assertContains(response, "Overview of test-domain.lu")

testing/views.py

@@ -29,7 +29,7 @@
 
 from testing_platform import settings
 
-from .forms import DMARCRecordForm, SPFRecordForm
+from .forms import DMARCRecordForm, SPFRecordForm, WebTestForm
 from .helpers import (
     check_soa_record,
     file_check,
@@ -131,7 +131,13 @@ def test_landing(request):
 @csrf_exempt
 def check_website_security(request):
     if request.method == 'POST':
-        domain = request.POST.get('target')
+        form = WebTestForm(request.POST)
+        if not form.is_valid():
+            return render(request, 'check_webapp.html', {
+                'error': form.errors['target'][0],
+                'target': request.POST.get('target', ''),
+            }, status=400)
+        domain = form.cleaned_data['target']
 
         csp_result = check_csp(domain)
         cookies_result = check_cookies(domain)