Ensure SSLSocket is closed after failure verifying cert hostname or fingerprint

3 files changed, 65 insertions, 28 deletions

diff --git a/changelog/2991.bugfix.rst b/changelog/2991.bugfix.rst
new file mode 100644
index 00000000..19e360ba
--- /dev/null
+++ b/changelog/2991.bugfix.rst
@@ -0,0 +1 @@
+avoid a ``ResourceWarning`` if ``assert_fingerprint`` or ``match_hostname`` fail
diff --git a/src/urllib3/connection.py b/src/urllib3/connection.py
index 1f13af0b..9dc04fe0 100644
--- a/src/urllib3/connection.py
+++ b/src/urllib3/connection.py
@@ -784,36 +784,42 @@ def _ssl_wrap_socket_and_match_hostname(
         tls_in_tls=tls_in_tls,
     )
 
-    if assert_fingerprint:
-        _assert_fingerprint(ssl_sock.getpeercert(binary_form=True), assert_fingerprint)
-    elif (
-        context.verify_mode != ssl.CERT_NONE
-        and not context.check_hostname
-        and assert_hostname is not False
-    ):
-        cert: _TYPE_PEER_CERT_RET_DICT = ssl_sock.getpeercert()  # type: ignore[assignment]
-
-        # Need to signal to our match_hostname whether to use 'commonName' or not.
-        # If we're using our own constructed SSLContext we explicitly set 'False'
-        # because PyPy hard-codes 'True' from SSLContext.hostname_checks_common_name.
-        if default_ssl_context:
-            hostname_checks_common_name = False
-        else:
-            hostname_checks_common_name = (
-                getattr(context, "hostname_checks_common_name", False) or False
+    try:
+        if assert_fingerprint:
+            _assert_fingerprint(
+                ssl_sock.getpeercert(binary_form=True), assert_fingerprint
+            )
+        elif (
+            context.verify_mode != ssl.CERT_NONE
+            and not context.check_hostname
+            and assert_hostname is not False
+        ):
+            cert: _TYPE_PEER_CERT_RET_DICT = ssl_sock.getpeercert()  # type: ignore[assignment]
+
+            # Need to signal to our match_hostname whether to use 'commonName' or not.
+            # If we're using our own constructed SSLContext we explicitly set 'False'
+            # because PyPy hard-codes 'True' from SSLContext.hostname_checks_common_name.
+            if default_ssl_context:
+                hostname_checks_common_name = False
+            else:
+                hostname_checks_common_name = (
+                    getattr(context, "hostname_checks_common_name", False) or False
+                )
+
+            _match_hostname(
+                cert,
+                assert_hostname or server_hostname,  # type: ignore[arg-type]
+                hostname_checks_common_name,
             )
 
-        _match_hostname(
-            cert,
-            assert_hostname or server_hostname,  # type: ignore[arg-type]
-            hostname_checks_common_name,
+        return _WrappedAndVerifiedSocket(
+            socket=ssl_sock,
+            is_verified=context.verify_mode == ssl.CERT_REQUIRED
+            or bool(assert_fingerprint),
         )
-
-    return _WrappedAndVerifiedSocket(
-        socket=ssl_sock,
-        is_verified=context.verify_mode == ssl.CERT_REQUIRED
-        or bool(assert_fingerprint),
-    )
+    except BaseException:
+        ssl_sock.close()
+        raise
 
 
 def _match_hostname(
diff --git a/test/test_connection.py b/test/test_connection.py
index 8237311f..9f33a7b3 100644
--- a/test/test_connection.py
+++ b/test/test_connection.py
@@ -17,7 +17,8 @@ from urllib3.connection import (  # type: ignore[attr-defined]
     _url_from_connection,
     _wrap_proxy_error,
 )
-from urllib3.exceptions import HTTPError, ProxyError
+from urllib3.exceptions import HTTPError, ProxyError, SSLError
+from urllib3.util import ssl_
 from urllib3.util.ssl_match_hostname import (
     CertificateError as ImplementationCertificateError,
 )
@@ -235,3 +236,32 @@ class TestConnection:
         # Should error if a request has not been sent
         with pytest.raises(ResponseNotReady):
             conn.getresponse()
+
+    def test_assert_fingerprint_closes_socket(self) -> None:
+        context = mock.create_autospec(ssl_.SSLContext)
+        context.wrap_socket.return_value.getpeercert.return_value = b"fake cert"
+        conn = HTTPSConnection(
+            "google.com",
+            port=443,
+            assert_fingerprint="AA:AA:AA:AA:AA:AAAA:AA:AAAA:AA:AA:AA:AA:AA:AA:AA:AA:AA:AA",
+            ssl_context=context,
+        )
+        with mock.patch.object(conn, "_new_conn"):
+            with pytest.raises(SSLError):
+                conn.connect()
+
+        context.wrap_socket.return_value.close.assert_called_once_with()
+
+    def test_assert_hostname_closes_socket(self) -> None:
+        context = mock.create_autospec(ssl_.SSLContext)
+        context.wrap_socket.return_value.getpeercert.return_value = {
+            "subjectAltName": (("DNS", "google.com"),)
+        }
+        conn = HTTPSConnection(
+            "google.com", port=443, assert_hostname="example.com", ssl_context=context
+        )
+        with mock.patch.object(conn, "_new_conn"):
+            with pytest.raises(ImplementationCertificateError):
+                conn.connect()
+
+        context.wrap_socket.return_value.close.assert_called_once_with()