Merge pull request #2250 from uktrade/LTD-5695-fix-URL-redirection-fr…

…om-remote-source-vulnerability

LTD-5695 fix possibleURL redirection from remote source vulnerability

core/helpers.py

@@ -1,11 +1,13 @@
 from urllib.parse import urlencode
-from django.http import (
-    Http404,
-    StreamingHttpResponse,
-)
+from django.http import StreamingHttpResponse
+from django.core.exceptions import SuspiciousOperation
 from django.utils.http import url_has_allowed_host_and_scheme
 
 
+class UnsafeURLDestination(SuspiciousOperation):
+    pass
+
+
 def dummy_quote(string, safe="", encoding=None, errors=None):
     return string
 
@@ -59,7 +61,7 @@ def check_url(request, url):
     if url_is_safe:
         return url
     else:
-        raise Http404
+        raise UnsafeURLDestination(f"URL destination '{url}' was deemed to be unsafe")
 
 
 def stream_document_response(api_response):

lite_forms/views.py

@@ -20,6 +20,8 @@
     validate_data_unknown,
 )
 from lite_forms.submitters import submit_paged_form
+from core.helpers import check_url
+
 
 ACTION = "_action"
 VALIDATE_ONLY = "validate_only"
@@ -188,9 +190,6 @@ def post(self, request, **kwargs):
         self.init(request, **kwargs)
         submission = self.on_submission(request, **kwargs)  # noqa
 
-        if submission:
-            return redirect(submission)
-
         response, data = submit_paged_form(
             request,
             self.get_forms(),
@@ -340,7 +339,8 @@ def _post(self, request, **kwargs):
         if self.validate_only_until_final_submission:
             return self.generate_summary_list()
 
-        return redirect(request.path)
+        sanitised_url = check_url(request, request.path)
+        return redirect(sanitised_url)
 
     def post(self, request, **kwargs):
         return self._post(request, **kwargs)
@@ -403,7 +403,9 @@ def get_next_form_page(self, form_pk, action, request, post_errors):
         if self.validate_only_until_final_submission:
             return self.generate_summary_list()
 
-        return redirect(request.path)
+        sanitised_url = check_url(request, request.path)
+
+        return redirect(sanitised_url)
 
     def dispatch(self, request, *args, **kwargs):
         if request.method.lower() in self.http_method_names:

unit_tests/core/test_helpers.py

@@ -1,6 +1,5 @@
 import pytest
-from django.http import Http404
-from core.helpers import convert_parameters_to_query_params, check_url
+from core.helpers import convert_parameters_to_query_params, check_url, UnsafeURLDestination
 
 
 @pytest.fixture
@@ -15,13 +14,13 @@ def test_convert_parameters_to_query_params():
     assert convert_parameters_to_query_params(params) == "?org_type=individual&org_type=commercial&page=1"
 
 
-@pytest.mark.parametrize("url", ["/next-url/", "http://testserver/next-url/"])
+@pytest.mark.parametrize("url", ["/next-url/", "http://testserver/next-url/", "\\valid\path"])
 def test_check_url(mock_request, url):
     assert check_url(mock_request, url) == url
 
 
 # Check not valid host and non secure url
-@pytest.mark.parametrize("url", ["http://not-the-same.com/next/", "https://test-server/next/"])
+@pytest.mark.parametrize("url", ["http://not-the-same.com/next/", "https://test-server/next/", "https:/malicious.com"])
 def test_check_url_not_allowed(mock_request, url):
-    with pytest.raises(Http404):
+    with pytest.raises(UnsafeURLDestination):
         check_url(mock_request, url)

unit_tests/exporter/applications/views/test_end_use_details.py

@@ -33,3 +33,28 @@ def test_application_end_use_summary(
         response.context["instruction_text"]
         == "Review your answers below and make any amends you need to. Click 'Save and continue' to save your progress."
     )
+
+
+def test_application_end_use_summary_post_url_has_allowed_host_and_scheme(
+    authorized_client, mock_application_get, application_end_use_summary_url, application_task_list_url
+):
+    response = authorized_client.post(
+        application_end_use_summary_url,
+        data={
+            "_action": "submit",
+        },
+    )
+    assert response.status_code == 302
+
+
+def test_application_end_use_summary_get_next_form_url_has_allowed_host_and_scheme(
+    authorized_client, mock_application_get, application_end_use_summary_url, application_task_list_url
+):
+    response = authorized_client.post(
+        application_end_use_summary_url,
+        data={
+            "_action": "finish",
+            "form_pk": "1",
+        },
+    )
+    assert response.status_code == 302