From 996e3779c9f6db8e673991b5a78029b16485aad4 Mon Sep 17 00:00:00 2001 From: Philippe Parage <69145356+pparage@users.noreply.github.com> Date: Mon, 14 Sep 2026 13:11:39 +0200 Subject: [PATCH] fix: handle web-test errors and normalize domain inputs --- .github/workflows/django.yml | 2 +- testing/forms.py | 42 +++++- testing/helpers.py | 45 ++++-- testing/templates/check_webapp.html | 5 +- testing/test_web_checks.py | 216 ++++++++++++++++++++++++++++ testing/views.py | 10 +- 6 files changed, 304 insertions(+), 16 deletions(-) create mode 100644 testing/test_web_checks.py diff --git a/.github/workflows/django.yml b/.github/workflows/django.yml index 7d4ee1ef..ddfe25f2 100644 --- a/.github/workflows/django.yml +++ b/.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 diff --git a/testing/forms.py b/testing/forms.py index b373f400..13728f04 100644 --- a/testing/forms.py +++ b/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): diff --git a/testing/helpers.py b/testing/helpers.py index 19d11a91..afcf982e 100644 --- a/testing/helpers.py +++ b/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 diff --git a/testing/templates/check_webapp.html b/testing/templates/check_webapp.html index 0634ff79..699b85c0 100644 --- a/testing/templates/check_webapp.html +++ b/testing/templates/check_webapp.html @@ -31,9 +31,10 @@

Assesses web security, includin {{ error }} {% endif %}
- + + value="{% firstof target domain %}" maxlength="2048" required> + For example, test-domain.lu or https://www.test-domain.lu/. The tests check the hostname's homepage.
diff --git a/testing/test_web_checks.py b/testing/test_web_checks.py new file mode 100644 index 00000000..9a718559 --- /dev/null +++ b/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"Test website" + 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": ''} + ) + self.assertContains(response, "<script>", status_code=400) + self.assertNotContains(response, '', 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") diff --git a/testing/views.py b/testing/views.py index dee750f5..16cab5c2 100644 --- a/testing/views.py +++ b/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)