SSLObject made available through exceptions

CHANGES/10028.feature.rst

@@ -0,0 +1,3 @@
+Added ``ssl_object`` attribute to ``ClientResponseError`` and its subclasses
+to provide access to SSL certificate information even after connection closure
+-- by :user:`fed239`.

CONTRIBUTORS.txt

@@ -135,6 +135,7 @@ Eugene Nikolaiev
 Eugene Tolmachev
 Evan Kepner
 Evert Lammerts
+Fedor Tyurin
 Felix Yan
 Fernanda Guimarães
 FichteFoll

aiohttp/client.py

@@ -80,6 +80,7 @@
     ClientResponse,
     Fingerprint,
     RequestInfo,
+    _extract_ssl_object,
 )
 from .client_ws import (
     DEFAULT_WS_CLIENT_TIMEOUT,
@@ -749,7 +750,9 @@ async def _connect_and_send_request(
                                 await req._body.close()
                             resp.close()
                             raise TooManyRedirects(
-                                history[0].request_info, tuple(history)
+                                history[0].request_info,
+                                tuple(history),
+                                ssl_object=_extract_ssl_object(resp._connection),
                             )
 
                         # For 301 and 302, mimic IE, now changed in RFC
@@ -1023,6 +1026,7 @@ async def _ws_connect(
                     message="Invalid response status",
                     status=resp.status,
                     headers=resp.headers,
+                    ssl_object=_extract_ssl_object(resp._connection),
                 )
 
             if resp.headers.get(hdrs.UPGRADE, "").lower() != "websocket":
@@ -1032,6 +1036,7 @@ async def _ws_connect(
                     message="Invalid upgrade header",
                     status=resp.status,
                     headers=resp.headers,
+                    ssl_object=_extract_ssl_object(resp._connection),
                 )
 
             if resp.headers.get(hdrs.CONNECTION, "").lower() != "upgrade":
@@ -1041,6 +1046,7 @@ async def _ws_connect(
                     message="Invalid connection header",
                     status=resp.status,
                     headers=resp.headers,
+                    ssl_object=_extract_ssl_object(resp._connection),
                 )
 
             # key calculation
@@ -1053,6 +1059,7 @@ async def _ws_connect(
                     message="Invalid challenge response",
                     status=resp.status,
                     headers=resp.headers,
+                    ssl_object=_extract_ssl_object(resp._connection),
                 )
 
             # websocket protocol
@@ -1082,6 +1089,7 @@ async def _ws_connect(
                             message=exc.args[0],
                             status=resp.status,
                             headers=resp.headers,
+                            ssl_object=_extract_ssl_object(resp._connection),
                         ) from exc
                 else:
                     compress = 0

aiohttp/client_exceptions.py

@@ -69,6 +69,7 @@ class ClientResponseError(ClientError):
     status: HTTP status code.
     message: Error message.
     headers: Response headers.
+    ssl_object: SSL object from the connection, if available.
     """
 
     def __init__(
@@ -79,6 +80,7 @@ def __init__(
         status: Optional[int] = None,
         message: str = "",
         headers: Optional[MultiMapping[str]] = None,
+        ssl_object: Optional[object] = None,
     ) -> None:
         self.request_info = request_info
         if status is not None:
@@ -88,6 +90,7 @@ def __init__(
         self.message = message
         self.headers = headers
         self.history = history
+        self.ssl_object = ssl_object
         self.args = (request_info, history)
 
     def __str__(self) -> str:
@@ -105,6 +108,8 @@ def __repr__(self) -> str:
             args += f", message={self.message!r}"
         if self.headers is not None:
             args += f", headers={self.headers!r}"
+        if self.ssl_object is not None:
+            args += f", ssl_object={self.ssl_object!r}"
         return f"{type(self).__name__}({args})"
 
 

aiohttp/client_reqrep.py

@@ -104,6 +104,33 @@
 _CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
 
 
+def _extract_ssl_object(
+    connection: Optional[Union["Connection", object]],
+) -> Optional[object]:
+    """Extract SSL object from connection or transport if available."""
+    if connection is None:
+        return None
+
+    # Handle both Connection objects and Transport objects
+    if hasattr(connection, "transport"):
+        # This is a Connection object
+        transport = connection.transport
+    elif hasattr(connection, "get_extra_info"):
+        # This is a Transport object
+        transport = connection
+    else:
+        return None
+
+    if transport is None:
+        return None
+
+    try:
+        return transport.get_extra_info("ssl_object")
+    except Exception:
+        # If we can't get the SSL object for any reason, return None
+        return None
+
+
 def _gen_default_accept_encoding() -> str:
     encodings = [
         "gzip",
@@ -476,6 +503,7 @@ async def start(self, connection: "Connection") -> "ClientResponse":
                         status=exc.code,
                         message=exc.message,
                         headers=exc.headers,
+                        ssl_object=_extract_ssl_object(connection),
                     ) from exc
 
                 if message.code < 100 or message.code > 199 or message.code == 101:
@@ -570,6 +598,7 @@ def raise_for_status(self) -> None:
                 status=self.status,
                 message=self.reason,
                 headers=self.headers,
+                ssl_object=_extract_ssl_object(self._connection),
             )
 
     def _release_connection(self) -> None:
@@ -691,6 +720,7 @@ async def json(
                         "unexpected mimetype: %s" % self.content_type
                     ),
                     headers=self.headers,
+                    ssl_object=_extract_ssl_object(self._connection),
                 )
 
         if encoding is None:

aiohttp/connector.py

@@ -50,7 +50,12 @@
     ssl_errors,
 )
 from .client_proto import ResponseHandler
-from .client_reqrep import SSL_ALLOWED_TYPES, ClientRequest, Fingerprint
+from .client_reqrep import (
+    SSL_ALLOWED_TYPES,
+    ClientRequest,
+    Fingerprint,
+    _extract_ssl_object,
+)
 from .helpers import (
     _SENTINEL,
     ceil_timeout,
@@ -1547,6 +1552,7 @@ async def _create_proxy_connection(
                             status=resp.status,
                             message=message,
                             headers=resp.headers,
+                            ssl_object=_extract_ssl_object(transport),
                         )
                 except BaseException:
                     # It shouldn't be closed in `finally` because it's fed to

docs/client_advanced.rst

@@ -711,6 +711,38 @@ DER with e.g::
    to :class:`TCPConnector` as default, the value from
    :meth:`ClientSession.get` and others override default.
 
+Example: Access SSL certificate information from exceptions
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. versionadded:: 3.13
+
+When :class:`ClientResponseError` or its subclasses are raised, you can access
+SSL certificate information even after the connection has been closed using
+the :attr:`~ClientResponseError.ssl_object` attribute::
+
+  try:
+      async with session.get('https://example.com') as resp:
+          resp.raise_for_status()
+  except aiohttp.ClientResponseError as e:
+      if e.ssl_object:
+          # Access peer certificate information
+          peer_cert = e.ssl_object.getpeercert()
+          print(f"Certificate subject: {peer_cert.get('subject')}")
+          print(f"Certificate issuer: {peer_cert.get('issuer')}")
+
+          # Check certificate validity dates
+          not_before = peer_cert.get('notBefore')
+          not_after = peer_cert.get('notAfter')
+          print(f"Valid from {not_before} to {not_after}")
+
+          # Access cipher information
+          cipher = e.ssl_object.cipher()
+          if cipher:
+              print(f"Cipher: {cipher[0]}, Protocol: {cipher[1]}")
+
+This is particularly useful for advanced certificate validation, debugging
+SSL-related issues, or implementing custom certificate verification logic.
+
 .. _aiohttp-client-proxy-support:
 
 Proxy support

docs/client_reference.rst

@@ -2718,6 +2718,27 @@ Response errors
 
       .. deprecated:: 3.1
 
+   .. attribute:: ssl_object
+
+      SSL object from the connection transport, if available
+      (:class:`ssl.SSLSocket` or ``None``).
+
+      This attribute provides access to SSL certificate information even after
+      the connection has been closed. Useful for advanced certificate
+      validation and debugging SSL-related issues.
+
+      Example usage::
+
+          try:
+              async with session.get('https://example.com') as resp:
+                  resp.raise_for_status()
+          except aiohttp.ClientResponseError as e:
+              if e.ssl_object:
+                  peer_cert = e.ssl_object.getpeercert()
+                  print(f"Certificate subject: {peer_cert.get('subject')}")
+
+      .. versionadded:: 3.13
+
 
 .. class:: ContentTypeError
 

tests/test_client_exceptions.py

@@ -1,11 +1,14 @@
 import errno
 import pickle
+import ssl
+from unittest.mock import Mock
 
 import pytest
 from multidict import CIMultiDict, CIMultiDictProxy
 from yarl import URL
 
 from aiohttp import client, client_reqrep
+from aiohttp.client_reqrep import _extract_ssl_object
 
 
 class TestClientResponseError:
@@ -83,6 +86,35 @@ def test_str(self) -> None:
         )
         assert str(err) == ("400, message='Something wrong', url='http://example.com'")
 
+    def test_ssl_object_none_by_default(self) -> None:
+        err = client.ClientResponseError(request_info=self.request_info, history=())
+        assert err.ssl_object is None
+
+    def test_ssl_object_stored(self) -> None:
+        mock_ssl_object = Mock(spec=ssl.SSLObject)
+        err = client.ClientResponseError(
+            request_info=self.request_info, history=(), ssl_object=mock_ssl_object
+        )
+        assert err.ssl_object is mock_ssl_object
+
+    def test_repr_with_ssl_object(self) -> None:
+        mock_ssl_object = Mock(spec=ssl.SSLObject)
+        mock_ssl_object.__repr__ = Mock(return_value="<MockSSLObject>")
+        err = client.ClientResponseError(
+            request_info=self.request_info,
+            history=(),
+            status=400,
+            message="Something wrong",
+            headers=CIMultiDict(),
+            ssl_object=mock_ssl_object,
+        )
+        expected_repr = (
+            "ClientResponseError(%r, (), status=400, "
+            "message='Something wrong', headers=<CIMultiDict()>, ssl_object=<MockSSLObject>)"
+            % (self.request_info,)
+        )
+        assert repr(err) == expected_repr
+
 
 class TestClientConnectorError:
     connection_key = client_reqrep.ConnectionKey(
@@ -305,3 +337,37 @@ def test_none_description(self) -> None:
     def test_str_with_description(self) -> None:
         err = client.InvalidURL(url=":wrong:url:", description=":description:")
         assert str(err) == ":wrong:url: - :description:"
+
+
+class TestExtractSSLObject:
+    def test_extract_ssl_object_none_connection(self) -> None:
+        result = _extract_ssl_object(None)
+        assert result is None
+
+    def test_extract_ssl_object_no_transport(self) -> None:
+        mock_connection = Mock()
+        mock_connection.transport = None
+        result = _extract_ssl_object(mock_connection)
+        assert result is None
+
+    def test_extract_ssl_object_success(self) -> None:
+        mock_ssl_object = Mock(spec=ssl.SSLObject)
+        mock_transport = Mock()
+        mock_transport.get_extra_info.return_value = mock_ssl_object
+        mock_connection = Mock()
+        mock_connection.transport = mock_transport
+
+        result = _extract_ssl_object(mock_connection)
+
+        assert result is mock_ssl_object
+        mock_transport.get_extra_info.assert_called_once_with("ssl_object")
+
+    def test_extract_ssl_object_exception_handling(self) -> None:
+        mock_transport = Mock()
+        mock_transport.get_extra_info.side_effect = Exception("Transport error")
+        mock_connection = Mock()
+        mock_connection.transport = mock_transport
+
+        result = _extract_ssl_object(mock_connection)
+
+        assert result is None