Add set_max_tls_version to PrometheusAPIClient

Complement set_min_tls_version so callers can pin both TLS protocol
bounds for HTTPS connections to Prometheus. Accepted values remain
"1.2" and "1.3". The max is applied through the same HTTPAdapter SSL
context path as the minimum.

Change-Id: Ic720bba77bbd99b17309a4f2852ac3cab4d276ec
Signed-off-by: Daniel Pawlik <dpawlik@redhat.com>

observabilityclient/prometheus_client.py

@@ -31,10 +31,11 @@ _TLS_VERSIONS = {
 
 
 class _PrometheusTLSAdapter(HTTPAdapter):
-    """HTTP adapter pinning a minimum TLS version on new connections."""
+    """HTTP adapter pinning TLS version bounds on new connections."""
 
-    def __init__(self, min_version, verify, **kwargs):
+    def __init__(self, min_version, max_version, verify, **kwargs):
         self._min_version = min_version
+        self._max_version = max_version
         # requests/urllib3 set verify_mode and load the CA bundle per request,
         # but CERT_NONE on a context with check_hostname enabled raises, so the
         # context has to be built knowing whether we verify.
@@ -44,10 +45,19 @@ class _PrometheusTLSAdapter(HTTPAdapter):
     def init_poolmanager(self, *args, **kwargs):
         kwargs['ssl_context'] = create_urllib3_context(
             ssl_minimum_version=self._min_version,
+            ssl_maximum_version=self._max_version,
             cert_reqs=self._cert_reqs)
         return super().init_poolmanager(*args, **kwargs)
 
 
+def _resolve_tls_version(version):
+    if version not in _TLS_VERSIONS:
+        raise ValueError(
+            f"Unknown TLS version: {version}. Supported values are "
+            f"{', '.join(sorted(_TLS_VERSIONS))}")
+    return _TLS_VERSIONS[version]
+
+
 class PrometheusAPIClientError(Exception):
     def __init__(self, response):
         self.resp = response
@@ -92,6 +102,7 @@ class PrometheusAPIClient:
         else:
             self._session = session
         self._min_tls_version = None
+        self._max_tls_version = None
 
     def set_ca_cert(self, ca_cert):
         self._session.verify = ca_cert
@@ -103,18 +114,24 @@ class PrometheusAPIClient:
         :param version: TLS version, ``1.2`` or ``1.3``
         :type version: str
         """
-        if version not in _TLS_VERSIONS:
-            raise ValueError(
-                f"Unknown TLS version: {version}. Supported values are "
-                f"{', '.join(sorted(_TLS_VERSIONS))}")
-        self._min_tls_version = _TLS_VERSIONS[version]
+        self._min_tls_version = _resolve_tls_version(version)
+        self._mount_tls_adapter()
+
+    def set_max_tls_version(self, version):
+        """Set the maximum TLS version for HTTPS connections to Prometheus.
+
+        :param version: TLS version, ``1.2`` or ``1.3``
+        :type version: str
+        """
+        self._max_tls_version = _resolve_tls_version(version)
         self._mount_tls_adapter()
 
     def _mount_tls_adapter(self):
-        if self._min_tls_version is None:
+        if self._min_tls_version is None and self._max_tls_version is None:
             return
         self._session.mount('https://', _PrometheusTLSAdapter(
-            self._min_tls_version, self._session.verify))
+            self._min_tls_version, self._max_tls_version,
+            self._session.verify))
 
     def set_client_cert(self, client_cert, client_key):
         self._session.cert = (client_cert, client_key)

observabilityclient/tests/unit/test_prometheus_client.py

@@ -112,6 +112,35 @@ class PrometheusAPIClientTest(PrometheusAPIClientTestBase):
         self.assertRaises(ValueError, c.set_min_tls_version, "TLSv1_2")
         self.assertRaises(ValueError, c.set_min_tls_version, "1.1")
 
+    def test_set_max_tls_version(self):
+        c = client.PrometheusAPIClient("localhost:9090")
+        c.set_max_tls_version("1.2")
+        ctx = self._ssl_context(c)
+        self.assertEqual(ssl.TLSVersion.TLSv1_2, ctx.maximum_version)
+        self.assertEqual(ssl.CERT_NONE, ctx.verify_mode)
+
+    def test_set_max_tls_version_with_ca_cert(self):
+        c = client.PrometheusAPIClient("localhost:9090")
+        c.set_ca_cert(True)
+        c.set_max_tls_version("1.3")
+        ctx = self._ssl_context(c)
+        self.assertEqual(ssl.TLSVersion.TLSv1_3, ctx.maximum_version)
+        self.assertEqual(ssl.CERT_REQUIRED, ctx.verify_mode)
+        self.assertTrue(ctx.check_hostname)
+
+    def test_set_min_and_max_tls_version(self):
+        c = client.PrometheusAPIClient("localhost:9090")
+        c.set_min_tls_version("1.2")
+        c.set_max_tls_version("1.3")
+        ctx = self._ssl_context(c)
+        self.assertEqual(ssl.TLSVersion.TLSv1_2, ctx.minimum_version)
+        self.assertEqual(ssl.TLSVersion.TLSv1_3, ctx.maximum_version)
+
+    def test_set_max_tls_version_invalid(self):
+        c = client.PrometheusAPIClient("localhost:9090")
+        self.assertRaises(ValueError, c.set_max_tls_version, "TLSv1_2")
+        self.assertRaises(ValueError, c.set_max_tls_version, "1.1")
+
     def test_init_scheme_default(self):
         c = client.PrometheusAPIClient("localhost:9090")
         self.assertIsNone(c._scheme)

releasenotes/notes/prometheus-tls-versions.yaml

@@ -1,7 +1,8 @@
 ---
 features:
   - |
-    ``PrometheusAPIClient`` now supports ``set_min_tls_version`` to configure
-    the minimum TLS protocol version used for HTTPS connections to Prometheus.
-    Accepted values are ``1.2`` and ``1.3``, matching the spelling used by
-    ``keystoneauth1``'s ``tls_min_version``.
+    ``PrometheusAPIClient`` now supports ``set_min_tls_version`` and
+    ``set_max_tls_version`` to configure the TLS protocol version bounds used
+    for HTTPS connections to Prometheus. Accepted values are ``1.2`` and
+    ``1.3``, matching the spelling used by ``keystoneauth1``'s
+    ``tls_min_version``.