Support alternate, custom CA certificate bundles

changelog.d/20231115_101055_kurtmckee_custom_ca_bundles_sc_28404.rst

@@ -0,0 +1,11 @@
+Added
+~~~~~
+
+- Support custom CA certificate bundles. (:pr:`NUMBER`)
+
+  Previously, SSL/TLS verification allowed only a boolean ``True`` or ``False`` value.
+  It is now possible to specify a CA certificate bundle file
+  using the existing ``verify_ssl`` parameter
+  or ``GLOBUS_SDK_VERIFY_SSL`` environment variable.
+
+  This may be useful for interacting with Globus through certain proxy firewalls.

docs/config.rst

@@ -23,9 +23,11 @@ created with ``verify_ssl=True``, the resulting client will have SSL
 verification turned on.
 
 ``GLOBUS_SDK_VERIFY_SSL``
-    Used to disable SSL verification, typically to handle SSL-intercepting
-    firewalls. By default, all connections to servers are verified. Set
-    ``GLOBUS_SDK_VERIFY_SSL="false"`` to disable verification.
+    Used to configure SSL/TLS verification, typically to handle SSL/TLS-intercepting firewalls.
+    By default, all connections to servers are verified.
+    Set ``GLOBUS_SDK_VERIFY_SSL="false"`` to disable verification,
+    or set ``GLOBUS_SDK_VERIFY_SSL="/path/to/ca-bundle.cert"``
+    to use an alternate certificate authority bundle file.
 
 ``GLOBUS_SDK_HTTP_TIMEOUT``
     Adjust the timeout when HTTP requests are made. By default, requests have a

src/globus_sdk/config/env_vars.py

@@ -8,6 +8,7 @@
 
 import logging
 import os
+import pathlib
 import typing as t
 
 log = logging.getLogger(__name__)
@@ -19,16 +20,6 @@
 SSL_VERIFY_VAR = "GLOBUS_SDK_VERIFY_SSL"
 
 
-def _str2bool(val: str) -> bool:
-    val = val.lower()
-    if val in ("y", "yes", "t", "true", "on", "1"):
-        return True
-    elif val in ("n", "no", "f", "false", "off", "0"):
-        return False
-    else:
-        raise ValueError(f"invalid truth value: {val}")
-
-
 @t.overload
 def _load_var(
     varname: str,
@@ -70,12 +61,28 @@ def _load_var(
     return value
 
 
-def _bool_cast(value: t.Any, default: t.Any) -> bool:  # pylint: disable=unused-argument
+def _ssl_verify_cast(
+    value: t.Any, default: t.Any  # pylint: disable=unused-argument
+) -> bool | str:
     if isinstance(value, bool):
         return value
-    elif not isinstance(value, str):
-        raise ValueError(f"cannot cast value {value} of type {type(value)} to bool")
-    return _str2bool(value)
+    if not isinstance(value, (str, pathlib.Path)):
+        msg = f"Value {value} of type {type(value)} cannot be used for SSL verification"
+        raise ValueError(msg)
+    if isinstance(value, str):
+        if value.lower() in {"y", "yes", "t", "true", "on", "1"}:
+            return True
+        if value.lower() in {"n", "no", "f", "false", "off", "0"}:
+            return False
+        if os.path.isfile(value):
+            return value
+    if isinstance(value, pathlib.Path):
+        if value.is_file():
+            return str(value.absolute())
+    raise ValueError(
+        f"SSL verification {value} is not a valid boolean value "
+        "nor a path to a file that exists"
+    )
 
 
 def _optfloat_cast(value: t.Any, default: t.Any) -> float | None:
@@ -93,8 +100,10 @@ def get_environment_name(inputenv: str | None = None) -> str:
     return _load_var(ENVNAME_VAR, "production", explicit_value=inputenv)
 
 
-def get_ssl_verify(value: bool | None = None) -> bool:
-    return _load_var(SSL_VERIFY_VAR, True, explicit_value=value, convert=_bool_cast)
+def get_ssl_verify(value: bool | str | pathlib.Path | None = None) -> bool | str:
+    return _load_var(
+        SSL_VERIFY_VAR, default=True, explicit_value=value, convert=_ssl_verify_cast
+    )
 
 
 def get_http_timeout(value: float | None = None) -> float | None:

src/globus_sdk/transport/requests.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import logging
+import pathlib
 import random
 import time
 import typing as t
@@ -110,7 +111,7 @@ class RequestsTransport:
 
     def __init__(
         self,
-        verify_ssl: bool | None = None,
+        verify_ssl: bool | str | None = None,
         http_timeout: float | None = None,
         retry_backoff: t.Callable[[RetryContext], float] = _exponential_backoff,
         retry_checks: list[RetryCheck] | None = None,
@@ -148,7 +149,7 @@ def _headers(self) -> dict[str, str]:
     def tune(
         self,
         *,
-        verify_ssl: bool | None = None,
+        verify_ssl: bool | str | pathlib.Path | None = None,
         http_timeout: float | None = None,
         retry_backoff: t.Callable[[RetryContext], float] | None = None,
         max_sleep: float | int | None = None,
@@ -162,8 +163,9 @@ def tune(
         In particular, this can be used to temporarily adjust request-sending minutiae
         like the ``http_timeout`` used.
 
-        :param verify_ssl: Explicitly enable or disable SSL verification
-        :type verify_ssl: bool, optional
+        :param verify_ssl: Explicitly enable or disable SSL verification,
+            or configure the path to a CA certificate bundle to use for SSL verification
+        :type verify_ssl: bool or str, optional
         :param http_timeout: Explicitly set an HTTP timeout value in seconds
         :type http_timeout: float, optional
         :param retry_backoff: A function which determines how long to sleep between
@@ -201,7 +203,10 @@ def tune(
             self.max_retries,
         )
         if verify_ssl is not None:
-            self.verify_ssl = verify_ssl
+            if isinstance(verify_ssl, bool):
+                self.verify_ssl = verify_ssl
+            else:
+                self.verify_ssl = str(verify_ssl)
         if http_timeout is not None:
             self.http_timeout = http_timeout
         if retry_backoff is not None:

tests/unit/CA-Bundle.cert

@@ -0,0 +1,2 @@
+This file exists to ensure that custom CA bundle files can be used.
+You can find references to this file by name in the test suite.

tests/unit/test_config.py

@@ -1,4 +1,5 @@
 import os
+import pathlib
 from unittest import mock
 
 import pytest
@@ -36,25 +37,48 @@ def test_get_service_url():
         globus_sdk.config.get_service_url("auth", environment="nonexistent")
 
 
+ca_bundle_file = pathlib.Path(__file__).parent.absolute() / "CA-Bundle.cert"
+ca_bundle_directory = pathlib.Path(__file__).parent.absolute()
+ca_bundle_non_existent = pathlib.Path(__file__).parent.absolute() / "bogus.bogus"
+
+
 @pytest.mark.parametrize(
     "value, expected_result",
-    [(x, True) for x in [str(True), "1", "YES", "true", "t", "True", "ON"]]
-    + [(x, False) for x in [str(False), "0", "NO", "false", "f", "False", "OFF"]]
-    + [(x, ValueError) for x in ["invalid", "1.0", "0.0"]],  # type: ignore
+    [(x, True) for x in ["1", "YES", "true", "t", "True", "ON"]]
+    + [(x, False) for x in ["0", "NO", "false", "f", "False", "OFF"]]
+    + [(str(ca_bundle_file), str(ca_bundle_file))]
+    + [
+        ("invalid", ValueError),
+        ("1.0", ValueError),
+        ("0.0", ValueError),
+        (str(ca_bundle_directory), ValueError),
+        (str(ca_bundle_non_existent), ValueError),
+    ],
 )
 def test_get_ssl_verify(value, expected_result, monkeypatch):
     """
     Confirms bool cast returns correct bools from sets of string values
     """
     monkeypatch.setenv("GLOBUS_SDK_VERIFY_SSL", value)
-    if isinstance(expected_result, bool):
+    if expected_result is not ValueError:
         assert globus_sdk.config.get_ssl_verify() == expected_result
     else:
         with pytest.raises(expected_result):
             globus_sdk.config.get_ssl_verify()
 
 
-@pytest.mark.parametrize("value", ["invalid", 1.0, object()])
+@pytest.mark.parametrize(
+    "value",
+    [
+        "invalid",
+        1.0,
+        object(),
+        ca_bundle_directory,
+        str(ca_bundle_directory),
+        ca_bundle_non_existent,
+        str(ca_bundle_non_existent),
+    ],
+)
 def test_get_ssl_verify_rejects_bad_explicit_value(value, monkeypatch):
     monkeypatch.delenv("GLOBUS_SDK_VERIFY_SSL", raising=False)
     with pytest.raises(ValueError):
@@ -66,9 +90,16 @@ def test_get_ssl_verify_with_explicit_value():
         os.environ["GLOBUS_SDK_VERIFY_SSL"] = "false"
         assert globus_sdk.config.get_ssl_verify(True) is True
         assert globus_sdk.config.get_ssl_verify(False) is False
+        assert globus_sdk.config.get_ssl_verify(ca_bundle_file) == str(ca_bundle_file)
+        assert globus_sdk.config.get_ssl_verify(str(ca_bundle_file)) == str(
+            ca_bundle_file
+        )
         os.environ["GLOBUS_SDK_VERIFY_SSL"] = "on"
         assert globus_sdk.config.get_ssl_verify(True) is True
         assert globus_sdk.config.get_ssl_verify(False) is False
+        assert globus_sdk.config.get_ssl_verify(str(ca_bundle_file)) == str(
+            ca_bundle_file
+        )
 
 
 @pytest.mark.parametrize(

tests/unit/transport/test_transport.py

@@ -1,3 +1,5 @@
+import pathlib
+
 import pytest
 
 from globus_sdk.transport import RequestsTransport, RetryContext
@@ -10,11 +12,22 @@ def _linear_backoff(ctx: RetryContext) -> float:
     return 0.5 * (2**ctx.attempt)
 
 
+ca_bundle_file = pathlib.Path(__file__).parent.parent.absolute() / "CA-Bundle.cert"
+ca_bundle_directory = ca_bundle_file.parent
+ca_bundle_non_existent = ca_bundle_directory / "bogus.bogus"
+
+
 @pytest.mark.parametrize(
     "param_name, init_value, tune_value",
     [
         ("verify_ssl", True, True),
         ("verify_ssl", True, False),
+        ("verify_ssl", True, ca_bundle_file),
+        ("verify_ssl", True, str(ca_bundle_file)),
+        ("verify_ssl", True, ca_bundle_directory),
+        ("verify_ssl", True, str(ca_bundle_directory)),
+        ("verify_ssl", True, ca_bundle_non_existent),
+        ("verify_ssl", True, str(ca_bundle_non_existent)),
         ("http_timeout", 60, 120),
         ("retry_backoff", _exponential_backoff, _linear_backoff),
         ("max_sleep", 10, 10),
@@ -24,13 +37,16 @@ def _linear_backoff(ctx: RetryContext) -> float:
     ],
 )
 def test_transport_tuning(param_name, init_value, tune_value):
+    expected_value = (
+        str(tune_value) if isinstance(tune_value, pathlib.Path) else tune_value
+    )
     init_kwargs = {param_name: init_value}
     transport = RequestsTransport(**init_kwargs)
 
     assert getattr(transport, param_name) == init_value
 
     tune_kwargs = {param_name: tune_value}
     with transport.tune(**tune_kwargs):
-        assert getattr(transport, param_name) == tune_value
+        assert getattr(transport, param_name) == expected_value
 
     assert getattr(transport, param_name) == init_value