SNOW-2176203: Support intermediates in trust stores (#2520)

Our current python connector does not support a configuration with intermediate certificates in a trust store as roots of trust. This is allowed by RFCs; the root does not need to be self signed, and this is the behavior we have in our Go client.

These changes enable partial chain validation to normalize behavior across clients.

Author: sfc-gh-snoonan

DESCRIPTION.md

@@ -11,6 +11,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Added the `workload_identity_impersonation_path` parameter to support service account impersonation for Workload Identity Federation on GCP and AWS workloads only
   - Fixed `get_results_from_sfqid` when using `DictCursor` and executing multiple statements at once
   - Added the `oauth_credentials_in_body` parameter supporting an option to send the oauth client credentials in the request body
+  - Added support for intermediate certificates as roots when they are stored in the trust store
 
 - v3.17.3(September 02,2025)
   - Enhanced configuration file permission warning messages.

src/snowflake/connector/connection_diagnostic.py

@@ -240,6 +240,10 @@ def __test_socket_get_cert(
 
                 context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                 context.load_verify_locations(certifi.where())
+                # Best-effort: enable partial-chain when supported
+                _partial_flag = getattr(ssl, "VERIFY_X509_PARTIAL_CHAIN", 0)
+                if _partial_flag and hasattr(context, "verify_flags"):
+                    context.verify_flags |= _partial_flag
                 sock = context.wrap_socket(conn, server_hostname=host)
                 certificate = ssl.DER_cert_to_PEM_cert(sock.getpeercert(True))
                 http_request = f"""GET / {host}:{port} HTTP/1.1\r\n

src/snowflake/connector/ssl_wrap_socket.py

@@ -8,11 +8,13 @@
 #
 # and added OCSP validator on the top.
 import logging
+import os
+import ssl
 import time
 import weakref
 from contextvars import ContextVar
 from functools import wraps
-from inspect import getfullargspec as get_args
+from inspect import signature as _sig
 from socket import socket
 from typing import Any
 
@@ -38,9 +40,57 @@
 log = logging.getLogger(__name__)
 
 
+# Helper utilities (private)
+def _resolve_cafile(kwargs: dict[str, Any]) -> str | None:
+    """Resolve CA bundle path from kwargs or standard environment variables.
+
+    Precedence:
+      1) kwargs['ca_certs'] if provided by caller
+      2) REQUESTS_CA_BUNDLE
+      3) SSL_CERT_FILE
+    """
+    caf = kwargs.get("ca_certs")
+    if caf:
+        return caf
+    return os.environ.get("REQUESTS_CA_BUNDLE") or os.environ.get("SSL_CERT_FILE")
+
+
+def _ensure_partial_chain_on_context(ctx: PyOpenSSLContext, cafile: str | None) -> None:
+    """Load CA bundle (when provided) and enable OpenSSL partial-chain support on ctx."""
+    if cafile:
+        try:
+            ctx.load_verify_locations(cafile=cafile, capath=None)
+        except (ssl.SSLError, OSError, ValueError):
+            # Leave context unchanged; handshake/validation surfaces failures
+            pass
+    try:
+        store = ctx._ctx.get_cert_store()
+        from OpenSSL import crypto as _crypto
+
+        if hasattr(_crypto, "X509StoreFlags") and hasattr(
+            _crypto.X509StoreFlags, "PARTIAL_CHAIN"
+        ):
+            store.set_flags(_crypto.X509StoreFlags.PARTIAL_CHAIN)
+    except (AttributeError, ImportError, OpenSSL.SSL.Error, OSError, ValueError):
+        # Best-effort; if not available, default chain building applies
+        pass
+
+
+def _build_context_with_partial_chain(cafile: str | None) -> PyOpenSSLContext:
+    """Create PyOpenSSL context configured for CERT_REQUIRED and partial-chain trust."""
+    ctx = PyOpenSSLContext(ssl_.PROTOCOL_TLS_CLIENT)
+    try:
+        ctx.verify_mode = ssl.CERT_REQUIRED
+    except Exception:
+        pass
+    _ensure_partial_chain_on_context(ctx, cafile)
+    return ctx
+
+
 # Store a *weak* reference so that the context variable doesn’t prolong the
 # lifetime of the SessionManager. Once all owning connections are GC-ed the
-# weakref goes dead and OCSP will fall back to its local manager (but most likely won't be used ever again anyway).
+# weakref goes dead and OCSP will fall back to its local manager (but most
+# likely won't be used ever again anyway).
 _CURRENT_SESSION_MANAGER: ContextVar[weakref.ref[SessionManager] | None] = ContextVar(
     "_CURRENT_SESSION_MANAGER",
     default=None,
@@ -71,7 +121,10 @@ def set_current_session_manager(sm: SessionManager | None) -> Any:
     Called from SnowflakeConnection so that OCSP downloads
     use the same proxy / header configuration as the initiating connection.
 
-    Alternative approach would be moving method inject_into_urllib3() inside connection initialization, but in case this delay (from module import time to connection initialization time) would cause some code to break we stayed with this approach, having in mind soon OCSP deprecation.
+    Alternative approach would be moving method inject_into_urllib3() inside
+    connection initialization, but in case this delay (from module import time
+    to connection initialization time) would cause some code to break we stayed
+    with this approach, having in mind soon OCSP deprecation.
     """
     return _CURRENT_SESSION_MANAGER.set(weakref.ref(sm) if sm is not None else None)
 
@@ -93,37 +146,29 @@ def inject_into_urllib3() -> None:
 
 @wraps(ssl_.ssl_wrap_socket)
 def ssl_wrap_socket_with_ocsp(*args: Any, **kwargs: Any) -> WrappedSocket:
-    # Extract host_name
-    hostname_index = get_args(ssl_.ssl_wrap_socket).args.index("server_hostname")
-    server_hostname = (
-        args[hostname_index]
-        if len(args) > hostname_index
-        else kwargs.get("server_hostname", None)
-    )
-    # Remove context if present
-    ssl_context_index = get_args(ssl_.ssl_wrap_socket).args.index("ssl_context")
-    context_in_args = len(args) > ssl_context_index
-    ssl_context = (
-        args[hostname_index] if context_in_args else kwargs.get("ssl_context", None)
-    )
-    if not isinstance(ssl_context, PyOpenSSLContext):
-        # Create new default context
-        if context_in_args:
-            new_args = list(args)
-            new_args[ssl_context_index] = None
-            args = tuple(new_args)
-        else:
-            del kwargs["ssl_context"]
-    # Fix ca certs location
-    ca_certs_index = get_args(ssl_.ssl_wrap_socket).args.index("ca_certs")
-    ca_certs_in_args = len(args) > ca_certs_index
-    if not ca_certs_in_args and not kwargs.get("ca_certs"):
-        kwargs["ca_certs"] = certifi.where()
-
-    ret = ssl_.ssl_wrap_socket(*args, **kwargs)
+    # Bind passed args/kwargs to the underlying signature to support both positional and keyword calls
+    bound = _sig(ssl_.ssl_wrap_socket).bind_partial(*args, **kwargs)
+    params = bound.arguments
+
+    server_hostname = params.get("server_hostname")
+
+    # Ensure CA bundle default if not provided
+    if not params.get("ca_certs"):
+        params["ca_certs"] = certifi.where()
+
+    # Ensure PyOpenSSL context with partial-chain is used if none or wrong type provided
+    provided_ctx = params.get("ssl_context")
+    if not isinstance(provided_ctx, PyOpenSSLContext):
+        cafile_for_ctx = _resolve_cafile(params)
+        params["ssl_context"] = _build_context_with_partial_chain(cafile_for_ctx)
+    else:
+        # If a PyOpenSSLContext is provided, ensure it trusts the provided CA and partial-chain is enabled
+        _ensure_partial_chain_on_context(provided_ctx, _resolve_cafile(params))
+
+    ret = ssl_.ssl_wrap_socket(**params)
 
     log.debug(
-        "OCSP Mode: %s, " "OCSP response cache file name: %s",
+        "OCSP Mode: %s, OCSP response cache file name: %s",
         FEATURE_OCSP_MODE.name,
         FEATURE_OCSP_RESPONSE_CACHE_FILE_NAME,
     )
@@ -137,10 +182,7 @@ def ssl_wrap_socket_with_ocsp(*args: Any, **kwargs: Any) -> WrappedSocket:
         ).validate(server_hostname, ret.connection)
         if not v:
             raise OperationalError(
-                msg=(
-                    "The certificate is revoked or "
-                    "could not be validated: hostname={}".format(server_hostname)
-                ),
+                msg=f"The certificate is revoked or could not be validated: hostname={server_hostname}",
                 errno=ER_OCSP_RESPONSE_CERT_STATUS_REVOKED,
             )
     else:

test/unit/test_ssl_partial_chain.py

@@ -0,0 +1,58 @@
+"""Unit tests for SSL wrapper partial-chain context injection."""
+
+import types
+
+import pytest
+
+import snowflake.connector.ssl_wrap_socket as ssw  # pylint: disable=import-error
+from snowflake.connector.constants import OCSPMode  # pylint: disable=import-error
+from snowflake.connector.vendored.urllib3.contrib.pyopenssl import (  # pylint: disable=import-error
+    PyOpenSSLContext,
+)
+
+
+@pytest.fixture(autouse=True)
+def disable_ocsp_checks():
+    """Disable OCSP checks for offline unit testing."""
+    # Ensure wrapper doesn't perform OCSP to keep this unit test offline
+    orig = ssw.FEATURE_OCSP_MODE
+    ssw.FEATURE_OCSP_MODE = OCSPMode.DISABLE_OCSP_CHECKS
+    try:
+        yield
+    finally:
+        ssw.FEATURE_OCSP_MODE = orig
+
+
+def test_wrapper_injects_pyopenssl_context(monkeypatch):
+    """Wrapper should inject a PyOpenSSLContext when none is given."""
+    captured = {}
+
+    def fake_ssl_wrap_socket(  # pylint: disable=unused-argument,too-many-arguments,too-many-positional-arguments
+        sock, ssl_context=None, **kwargs
+    ):
+        # Assert that our wrapper provided a PyOpenSSLContext
+        captured["ctx_is_pyopenssl"] = isinstance(ssl_context, PyOpenSSLContext)
+        # Return a minimal object with a 'connection' attribute expected by wrapper
+        return types.SimpleNamespace(connection=None)
+
+    # Patch underlying urllib3 ssl_wrap_socket used by our wrapper
+    monkeypatch.setattr(ssw.ssl_, "ssl_wrap_socket", fake_ssl_wrap_socket)
+
+    # Call our wrapper without providing ssl_context; it should inject one
+    ssw.ssl_wrap_socket_with_ocsp(
+        sock=None,
+        keyfile=None,
+        certfile=None,
+        cert_reqs=None,
+        ca_certs=None,
+        server_hostname="localhost",
+        ssl_version=None,
+        ciphers=None,
+        ssl_context=None,
+        ca_cert_dir=None,
+        key_password=None,
+        ca_cert_data=None,
+        tls_in_tls=False,
+    )
+
+    assert captured.get("ctx_is_pyopenssl") is True