Add experimental support for pinning chain certificates

sethmlarson · web-flow · commit ff37321b515c · 2021-10-01T09:36:16.000-05:00
`SSLObject.get_verified_chain()` and `Certificate.public_bytes()` are private APIs in CPython 3.10. They're not documented anywhere yet but seem to work and we need them for Security on by Default. See: python/cpython#25467
diff --git a/elastic_transport/_models.py b/elastic_transport/_models.py
@@ -297,6 +297,11 @@ class NodeConfig:
     #: SHA-256 fingerprint of the node's certificate. If this value is
     #: given then root-of-trust verification isn't done and only the
     #: node's certificate fingerprint is verified.
+    #:
+    #: On CPython 3.10+ this also verifies if any certificate in the
+    #: chain including the Root CA matches this fingerprint. However
+    #: because this requires using private APIs support for this is
+    #: **experimental**.
     ssl_assert_fingerprint: Optional[str] = None
     #: Minimum TLS version to use to connect to the node.
     ssl_version: Optional[int] = None
diff --git a/elastic_transport/_node/_http_aiohttp.py b/elastic_transport/_node/_http_aiohttp.py
@@ -43,6 +43,8 @@
 
 
 class AiohttpHttpNode(BaseAsyncNode):
+    """Default asynchronous node class using the ``aiohttp`` library via HTTP"""
+
     _ELASTIC_CLIENT_META = ("ai", _AIOHTTP_META_VERSION)
 
     def __init__(self, config: NodeConfig):
@@ -178,10 +180,10 @@ async def perform_request(
             ):
                 raise ConnectionTimeout(
                     "Connection timed out during request", errors=(e,)
-                )
+                ) from None
             elif isinstance(e, (ssl.SSLError, aiohttp_exceptions.ClientSSLError)):
-                raise TlsError(str(e), errors=(e,))
-            raise ConnectionError(str(e), errors=(e,))
+                raise TlsError(str(e), errors=(e,)) from None
+            raise ConnectionError(str(e), errors=(e,)) from None
 
         return (
             ApiResponseMeta(
@@ -221,7 +223,7 @@ def _create_aiohttp_session(self) -> None:
 
 
 @functools.lru_cache(maxsize=64, typed=True)
-def aiohttp_fingerprint(ssl_assert_fingerprint: str) -> aiohttp.Fingerprint:
+def aiohttp_fingerprint(ssl_assert_fingerprint: str) -> "aiohttp.Fingerprint":
     """Changes 'ssl_assert_fingerprint' into a configured 'aiohttp.Fingerprint' instance.
     Uses a cache to prevent creating tons of objects needlessly.
     """
diff --git a/elastic_transport/_node/_http_requests.py b/elastic_transport/_node/_http_requests.py
@@ -35,29 +35,39 @@
 
     _REQUESTS_AVAILABLE = True
     _REQUESTS_META_VERSION = client_meta_version(requests.__version__)
+
+    # Use our custom HTTPSConnectionPool for chain cert fingerprint support.
+    try:
+        from ._urllib3_chain_certs import HTTPSConnectionPool
+    except (ImportError, AttributeError):
+        HTTPSConnectionPool = urllib3.HTTPSConnectionPool
+
+    class _ElasticHTTPAdapter(HTTPAdapter):
+        def __init__(self, node_config: NodeConfig, **kwargs):
+            self._node_config = node_config
+            super().__init__(**kwargs)
+
+        def init_poolmanager(
+            self, connections, maxsize, block=False, **pool_kwargs
+        ) -> None:
+            if self._node_config.ssl_context:
+                pool_kwargs.setdefault("ssl_context", self._node_config.ssl_context)
+            if self._node_config.ssl_assert_fingerprint:
+                pool_kwargs.setdefault(
+                    "assert_fingerprint", self._node_config.ssl_assert_fingerprint
+                )
+
+            super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
+            self.poolmanager.pool_classes_by_scheme["https"] = HTTPSConnectionPool
+
+
 except ImportError:  # pragma: nocover
     _REQUESTS_AVAILABLE = False
     _REQUESTS_META_VERSION = ""
 
 
 class RequestsHttpNode(BaseNode):
-    """
-    Connection using the `requests` library communicating via HTTP.
-
-    :arg use_ssl: use ssl for the connection if `True`
-    :arg verify_certs: whether to verify SSL certificates
-    :arg ssl_show_warn: show warning when verify certs is disabled
-    :arg ca_certs: optional path to CA bundle. By default standard requests'
-        bundle will be used.
-    :arg client_cert: path to the file containing the private key and the
-        certificate, or cert only if using client_key
-    :arg client_key: path to the file containing the private key if using
-        separate cert and key files (client_cert will contain only the cert)
-    :arg headers: any custom http headers to be add to requests
-    :arg http_compress: Use gzip compression
-    :arg opaque_id: Send this value in the 'X-Opaque-Id' HTTP header
-        For tracing all requests made by this transport.
-    """
+    """Synchronous node using the ``requests`` library communicating via HTTP"""
 
     _ELASTIC_CLIENT_META = ("rq", _REQUESTS_META_VERSION)
 
@@ -109,8 +119,10 @@ def __init__(self, config: NodeConfig):
             pool_maxsize=config.connections_per_node,
             pool_block=True,
         )
-        for prefix in ("http://", "https://"):
-            self.session.mount(prefix=prefix, adapter=adapter)
+        # Preload the HTTPConnectionPool so initialization issues
+        # are raised here instead of in perform_request()
+        adapter.get_connection(self.base_url)
+        self.session.mount(prefix=f"{self.scheme}://", adapter=adapter)
 
     def perform_request(
         self,
@@ -161,10 +173,10 @@ def perform_request(
             if isinstance(e, requests.Timeout):
                 raise ConnectionTimeout(
                     "Connection timed out during request", errors=(e,)
-                )
+                ) from None
             elif isinstance(e, (ssl.SSLError, requests.exceptions.SSLError)):
-                raise TlsError(str(e), errors=(e,))
-            raise ConnectionError(str(e), errors=(e,))
+                raise TlsError(str(e), errors=(e,)) from None
+            raise ConnectionError(str(e), errors=(e,)) from None
 
         response = ApiResponseMeta(
             node=self.config,
@@ -180,22 +192,3 @@ def close(self) -> None:
         Explicitly closes connections
         """
         self.session.close()
-
-
-class _ElasticHTTPAdapter(HTTPAdapter):
-    def __init__(self, node_config: NodeConfig, **kwargs):
-        self._node_config = node_config
-        super().__init__(**kwargs)
-
-    def init_poolmanager(
-        self, connections, maxsize, block=False, **pool_kwargs
-    ) -> urllib3.PoolManager:
-        if self._node_config.ssl_context:
-            pool_kwargs.setdefault("ssl_context", self._node_config.ssl_context)
-        if self._node_config.ssl_assert_fingerprint:
-            pool_kwargs.setdefault(
-                "ssl_assert_fingerprint", self._node_config.ssl_assert_fingerprint
-            )
-        return super().init_poolmanager(
-            connections, maxsize, block=block, **pool_kwargs
-        )
diff --git a/elastic_transport/_node/_http_urllib3.py b/elastic_transport/_node/_http_urllib3.py
@@ -31,9 +31,14 @@
 from ..client_utils import DEFAULT, client_meta_version
 from ._base import DEFAULT_CA_CERTS, RERAISE_EXCEPTIONS, BaseNode
 
+try:
+    from ._urllib3_chain_certs import HTTPSConnectionPool
+except (ImportError, AttributeError):
+    HTTPSConnectionPool = urllib3.HTTPSConnectionPool
+
 
 class Urllib3HttpNode(BaseNode):
-    """Default synchronous node class using the `urllib3` library via HTTP."""
+    """Default synchronous node class using the ``urllib3`` library via HTTP"""
 
     _ELASTIC_CLIENT_META = ("ur", client_meta_version(urllib3.__version__))
 
@@ -45,7 +50,7 @@ def __init__(self, config: NodeConfig):
 
         # if ssl_context provided use SSL by default
         if config.scheme == "https" and config.ssl_context:
-            pool_class = urllib3.HTTPSConnectionPool
+            pool_class = HTTPSConnectionPool
             kw.update(
                 {
                     "assert_fingerprint": config.ssl_assert_fingerprint,
@@ -54,7 +59,7 @@ def __init__(self, config: NodeConfig):
             )
 
         elif config.scheme == "https":
-            pool_class = urllib3.HTTPSConnectionPool
+            pool_class = HTTPSConnectionPool
             kw.update(
                 {
                     "ssl_version": config.ssl_version,
@@ -147,10 +152,10 @@ def perform_request(
             if isinstance(e, (ConnectTimeoutError, ReadTimeoutError)):
                 raise ConnectionTimeout(
                     "Connection timed out during request", errors=(e,)
-                )
+                ) from None
             elif isinstance(e, (ssl.SSLError, urllib3.exceptions.SSLError)):
-                raise TlsError(str(e), errors=(e,))
-            raise ConnectionError(str(e), errors=(e,))
+                raise TlsError(str(e), errors=(e,)) from None
+            raise ConnectionError(str(e), errors=(e,)) from None
 
         response = ApiResponseMeta(
             node=self.config,
diff --git a/elastic_transport/_node/_urllib3_chain_certs.py b/elastic_transport/_node/_urllib3_chain_certs.py
@@ -0,0 +1,121 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+import hashlib
+import sys
+from binascii import hexlify, unhexlify
+from hmac import compare_digest
+from typing import Any, List, Optional
+
+import _ssl
+import urllib3
+
+from ._base import RERAISE_EXCEPTIONS
+
+if sys.version_info < (3, 10) or sys.implementation.name != "cpython":
+    raise ImportError("Only supported on CPython 3.10+")
+
+_ENCODING_DER = _ssl.ENCODING_DER
+_HASHES_BY_LENGTH = {32: hashlib.md5, 40: hashlib.sha1, 64: hashlib.sha256}
+
+__all__ = ["HTTPSConnectionPool"]
+
+
+class HTTPSConnectionPool(urllib3.HTTPSConnectionPool):
+    """HTTPSConnectionPool implementation which supports ``assert_fingerprint``
+    on certificates within the chain instead of only the leaf cert using private
+    APIs in CPython 3.10+
+    """
+
+    def __init__(
+        self, *args: Any, assert_fingerprint: Optional[str] = None, **kwargs: Any
+    ) -> None:
+        self._elastic_assert_fingerprint = (
+            assert_fingerprint.replace(":", "").lower() if assert_fingerprint else None
+        )
+
+        # Complain about fingerprint length earlier than urllib3 does.
+        if (
+            self._elastic_assert_fingerprint
+            and len(self._elastic_assert_fingerprint) not in _HASHES_BY_LENGTH
+        ):
+            valid_lengths = "', '".join(map(str, sorted(_HASHES_BY_LENGTH.keys())))
+            raise ValueError(
+                f"Fingerprint of invalid length '{len(self._elastic_assert_fingerprint)}'"
+                f", should be one of '{valid_lengths}'"
+            )
+
+        if assert_fingerprint:
+            # Falsey but not None. This is a hack to skip fingerprinting by urllib3
+            # but still set 'is_verified=True' within HTTPSConnectionPool._validate_conn()
+            kwargs["assert_fingerprint"] = ""
+
+        super().__init__(*args, **kwargs)
+
+    def _validate_conn(self, conn):
+        """
+        Called right before a request is made, after the socket is created.
+        """
+        super(HTTPSConnectionPool, self)._validate_conn(conn)
+
+        if self._elastic_assert_fingerprint:
+            hash_func = _HASHES_BY_LENGTH[len(self._elastic_assert_fingerprint)]
+            assert_fingerprint = unhexlify(
+                self._elastic_assert_fingerprint.lower()
+                .replace(":", "")
+                .encode("ascii")
+            )
+
+            fingerprints: List[bytes]
+            try:
+                # 'get_verified_chain()' and 'Certificate.public_bytes()' are private APIs
+                # in CPython 3.10. They're not documented anywhere yet but seem to work
+                # and we need them for Security on by Default so... onwards we go!
+                # See: https://github.com/python/cpython/pull/25467
+                fingerprints = [
+                    hash_func(cert.public_bytes(_ENCODING_DER)).digest()
+                    for cert in conn.sock._sslobj.get_verified_chain()
+                ]
+            except RERAISE_EXCEPTIONS:  # pragma: nocover
+                raise
+            # Because these are private APIs we are super careful here
+            # so that if anything "goes wrong" we fallback on the old behavior.
+            except Exception:  # pragma: nocover
+                fingerprints = []
+
+            # Only add the peercert in front of the chain if it's not there for some reason.
+            # This is to make sure old behavior of 'ssl_assert_fingerprint' still works.
+            peercert_fingerprint = hash_func(conn.sock.getpeercert(True)).digest()
+            if peercert_fingerprint not in fingerprints:  # pragma: nocover
+                fingerprints.insert(0, peercert_fingerprint)
+
+            # If any match then that's a success! We always run them
+            # all through though because of constant time concerns.
+            success = False
+            for fingerprint in fingerprints:
+                success |= compare_digest(fingerprint, assert_fingerprint)
+
+            # Give users all the fingerprints we checked against in
+            # order of peer -> root CA.
+            if not success:
+                raise urllib3.exceptions.SSLError(
+                    'Fingerprints did not match. Expected "{0}", got "{1}".'.format(
+                        self._elastic_assert_fingerprint,
+                        '", "'.join([x.decode() for x in map(hexlify, fingerprints)]),
+                    )
+                )
+            conn.is_verified = success
diff --git a/tests/node/test_http_aiohttp.py b/tests/node/test_http_aiohttp.py
@@ -264,7 +264,7 @@ async def test_ssl_assert_fingerprint(httpbin_cert_fingerprint):
         resp, _ = await node.perform_request("GET", "/")
 
     assert resp.status == 200
-    assert w == []
+    assert [str(x.message) for x in w if x.category != DeprecationWarning] == []
 
 
 async def test_default_headers():
diff --git a/tests/node/test_urllib3_chain_certs.py b/tests/node/test_urllib3_chain_certs.py
