Merge branch 'main' into add_proxy_response_headers

Lib/http/client.py

@@ -945,15 +945,18 @@ def _tunnel(self):
         try:
             (version, code, message) = response._read_status()
 
-        self._proxy_response_headers = parse_headers(response.fp)
+            self._proxy_response_headers = parse_headers(response.fp)
 
-        if self.debuglevel > 0:
-            for hdr, val in self._proxy_response_headers.items():
-                print("header:", hdr + ":", val)
+            if self.debuglevel > 0:
+                for hdr, val in self._proxy_response_headers.items():
+                    print("header:", hdr + ":", val)
 
-        if code != http.HTTPStatus.OK:
-            self.close()
-            raise OSError(f"Tunnel connection failed: {code} {message.strip()}")
+            if code != http.HTTPStatus.OK:
+                self.close()
+                raise OSError(f"Tunnel connection failed: {code} {message.strip()}")
+
+        finally:
+            response.close()
 
     def connect(self):
         """Connect to the host and port specified in __init__."""

Lib/test/test_httplib.py

@@ -2404,6 +2404,29 @@ def test_proxy_response_headers(self):
         headers = self.conn._proxy_response_headers
         self.assertIn(expected_header, headers.items())
 
+    def test_tunnel_leak(self):
+        sock = None
+
+        def _create_connection(address, timeout=None, source_address=None):
+            nonlocal sock
+            sock = FakeSocket(
+                'HTTP/1.1 404 NOT FOUND\r\n\r\n',
+                host=address[0],
+                port=address[1],
+            )
+            return sock
+
+        self.conn._create_connection = _create_connection
+        self.conn.set_tunnel('destination.com')
+        exc = None
+        try:
+            self.conn.request('HEAD', '/', '')
+        except OSError as e:
+            # keeping a reference to exc keeps response alive in the traceback
+            exc = e
+        self.assertIsNotNone(exc)
+        self.assertTrue(sock.file_closed)
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)