Merge "Adapt websocketproxy tests for SimpleHTTPServer fix" into stab…

…le/yoga

nova/tests/unit/console/test_websocketproxy.py

@@ -589,12 +589,12 @@ def test_malformed_cookie(self, validate, check_port):
         self.wh.socket.assert_called_with('node1', 10000, connect=True)
         self.wh.do_proxy.assert_called_with('<socket>')
 
-    def test_reject_open_redirect(self):
+    def test_reject_open_redirect(self, url='//example.com/%2F..'):
         # This will test the behavior when an attempt is made to cause an open
         # redirect. It should be rejected.
         mock_req = mock.MagicMock()
         mock_req.makefile().readline.side_effect = [
-            b'GET //example.com/%2F.. HTTP/1.1\r\n',
+            f'GET {url} HTTP/1.1\r\n'.encode('utf-8'),
             b''
         ]
 
@@ -619,41 +619,32 @@ def test_reject_open_redirect(self):
         result = output.readlines()
 
         # Verify no redirect happens and instead a 400 Bad Request is returned.
-        self.assertIn('400 URI must not start with //', result[0].decode())
+        # NOTE: As of python 3.10.6 there is a fix for this vulnerability,
+        # which will cause a 301 Moved Permanently error to be returned
+        # instead that redirects to a sanitized version of the URL with extra
+        # leading '/' characters removed.
+        # See https://github.com/python/cpython/issues/87389 for details.
+        # We will consider either response to be valid for this test. This will
+        # also help if and when the above fix gets backported to older versions
+        # of python.
+        errmsg = result[0].decode()
+        expected_nova = '400 URI must not start with //'
+        expected_cpython = '301 Moved Permanently'
+
+        self.assertTrue(expected_nova in errmsg or expected_cpython in errmsg)
+
+        # If we detect the cpython fix, verify that the redirect location is
+        # now the same url but with extra leading '/' characters removed.
+        if expected_cpython in errmsg:
+            location = result[3].decode()
+            location = location.removeprefix('Location: ').rstrip('\r\n')
+            self.assertTrue(
+                location.startswith('/example.com/%2F..'),
+                msg='Redirect location is not the expected sanitized URL',
+            )
 
     def test_reject_open_redirect_3_slashes(self):
-        # This will test the behavior when an attempt is made to cause an open
-        # redirect. It should be rejected.
-        mock_req = mock.MagicMock()
-        mock_req.makefile().readline.side_effect = [
-            b'GET ///example.com/%2F.. HTTP/1.1\r\n',
-            b''
-        ]
-
-        # Collect the response data to verify at the end. The
-        # SimpleHTTPRequestHandler writes the response data by calling the
-        # request socket sendall() method.
-        self.data = b''
-
-        def fake_sendall(data):
-            self.data += data
-
-        mock_req.sendall.side_effect = fake_sendall
-
-        client_addr = ('8.8.8.8', 54321)
-        mock_server = mock.MagicMock()
-        # This specifies that the server will be able to handle requests other
-        # than only websockets.
-        mock_server.only_upgrade = False
-
-        # Constructing a handler will process the mock_req request passed in.
-        websocketproxy.NovaProxyRequestHandler(
-            mock_req, client_addr, mock_server)
-
-        # Verify no redirect happens and instead a 400 Bad Request is returned.
-        self.data = self.data.decode()
-        self.assertIn('Error code: 400', self.data)
-        self.assertIn('Message: URI must not start with //', self.data)
+        self.test_reject_open_redirect(url='///example.com/%2F..')
 
     @mock.patch('nova.objects.ConsoleAuthToken.validate')
     def test_no_compute_rpcapi_with_invalid_token(self, mock_validate):