- Fix open redirect issue in redirect handling

Author: dataflake

CHANGES.rst

@@ -4,6 +4,8 @@ Change Log
 2.6.0 (unreleased)
 ------------------
 
+- Fix open redirect issue in `Cookie Auth Helper` redirect handling
+
 - Add support for Python 3.9.
 
 

src/Products/PluggableAuthService/plugins/CookieAuthHelper.py

@@ -28,6 +28,8 @@
 import six
 from six.moves.urllib.parse import quote
 from six.moves.urllib.parse import unquote
+from six.moves.urllib.parse import urlparse
+from six.moves.urllib.parse import urlunparse
 
 from AccessControl.class_init import InitializeClass
 from AccessControl.Permissions import view
@@ -225,6 +227,12 @@ def unauthorized(self):
                     # in an endless redirect loop.
                     return 0
 
+            # Sanitize the return URL ``came_from`` and only allow local URLs
+            # to prevent an open exploitable redirect issue
+            if came_from:
+                parsed = urlparse(came_from)
+                came_from = urlunparse(('', '') + parsed[2:])
+
             if '?' in url:
                 sep = '&'
             else:

src/Products/PluggableAuthService/plugins/tests/test_CookieAuthHelper.py

@@ -128,7 +128,8 @@ def test_extractCredentials_with_deleted_cookie(self):
     def test_challenge(self):
         rc, root, folder, object = self._makeTree()
         response = FauxCookieResponse()
-        testURL = 'http://test'
+        testPath = '/some/path'
+        testURL = 'http://test' + testPath
         request = FauxRequest(RESPONSE=response, URL=testURL,
                               ACTUAL_URL=testURL)
         root.REQUEST = request
@@ -138,7 +139,9 @@ def test_challenge(self):
         helper.challenge(request, response)
         self.assertEqual(response.status, 302)
         self.assertEqual(len(response.headers), 3)
-        self.assertTrue(response.headers['Location'].endswith(quote(testURL)))
+        loc = response.headers['Location']
+        self.assertTrue(loc.endswith(quote(testPath)))
+        self.assertNotIn(testURL, loc)
         self.assertEqual(response.headers['Cache-Control'], 'no-cache')
         self.assertEqual(response.headers['Expires'],
                          'Sat, 01 Jan 2000 00:00:00 GMT')
@@ -159,8 +162,9 @@ def test_challenge_with_vhm(self):
         self.assertEqual(response.status, 302)
         self.assertEqual(len(response.headers), 3)
         loc = response.headers['Location']
-        self.assertTrue(loc.endswith(quote(actualURL)))
+        self.assertTrue(loc.endswith(quote('/xxx')))
         self.assertFalse(loc.endswith(quote(vhm)))
+        self.assertNotIn(actualURL, loc)
         self.assertEqual(response.headers['Cache-Control'], 'no-cache')
         self.assertEqual(response.headers['Expires'],
                          'Sat, 01 Jan 2000 00:00:00 GMT')