databricks · cathleeny · Sep 16, 2026 · Sep 16, 2026 · Sep 16, 2026 · Sep 16, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 # Unreleased
 - Transparently auto-recover Thrift connections to Reyden / Real-Time warehouses: when a warehouse rejects the default Thrift protocol (SQLSTATE `KP001`), the session is re-opened on the kernel backend and the warehouse is remembered so later connections skip Thrift. Applies only when no backend was chosen explicitly.
+- Reject an mTLS private key without a client certificate, and identify missing or empty client certificate/key files in connection errors.
 
 # 4.5.0 (2026-09-01)
 - Upgrade Databricks SQL Kernel to 1.0.0.

diff --git a/src/databricks/sql/backend/kernel/client.py b/src/databricks/sql/backend/kernel/client.py
@@ -53,6 +53,7 @@
 if TYPE_CHECKING:
     from databricks.sql.client import Cursor
     from databricks.sql.result_set import ResultSet
+    from databricks.sql.types import SSLOptions
 
     # Type-annotation-only import (deferred by ``from __future__ import
     # annotations``). ``execute_command`` accepts the Thrift-shaped
@@ -1078,7 +1079,7 @@ def max_download_threads(self) -> int:
 }
 
 
-def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
+def _kernel_tls_kwargs(ssl_options: Optional[SSLOptions]) -> Dict[str, Any]:
     """Translate the connector's ``SSLOptions`` into the kernel
     ``Session``'s ``tls_*`` kwargs.
 
@@ -1102,6 +1103,11 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     if ssl_options is None:
         return {}
 
+    # The kernel rejects an in-memory key without a certificate, but a lone key file
+    # used to be dropped here before it could reach that validation. Reject the
+    # incomplete connector configuration directly instead.
+    ssl_options.validate_client_identity()
+
     kwargs: Dict[str, Any] = {}
 
     # Inverted booleans. Emit only the insecure (skip) direction so the
@@ -1111,18 +1117,18 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     # own semantics (``create_ssl_context`` sets ``check_hostname=False``
     # whenever ``tls_verify`` is False). Without this the kernel could
     # still attempt a hostname check the connector considers disabled.
-    if getattr(ssl_options, "tls_verify", True) is False:
+    if ssl_options.tls_verify is False:
         kwargs["tls_skip_verify"] = True
         kwargs["tls_skip_hostname_verify"] = True
-    elif getattr(ssl_options, "tls_verify_hostname", True) is False:
+    elif ssl_options.tls_verify_hostname is False:
         kwargs["tls_skip_hostname_verify"] = True
 
-    ca_file = getattr(ssl_options, "tls_trusted_ca_file", None)
+    ca_file = ssl_options.tls_trusted_ca_file
     if ca_file:
         kwargs["tls_ca_cert"] = _read_pem_bytes(ca_file, "tls_trusted_ca_file")
 
-    cert_file = getattr(ssl_options, "tls_client_cert_file", None)
-    key_file = getattr(ssl_options, "tls_client_cert_key_file", None)
+    cert_file = ssl_options.tls_client_cert_file
+    key_file = ssl_options.tls_client_cert_key_file
     if cert_file:
         # The kernel pairs cert + key for mutual TLS; a cert without a
         # key (or vice versa) is rejected kernel-side. The connector's
@@ -1135,7 +1141,7 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
         # The kernel has no surface for an encrypted client key today.
         # Reject loudly rather than hand the kernel a key it can't
         # decrypt (which would fail with an opaque TLS parse error).
-        if getattr(ssl_options, "tls_client_cert_key_password", None):
+        if ssl_options.tls_client_cert_key_password:
             raise NotSupportedError(
                 "use_kernel=True does not support a password-protected mTLS "
                 "client key (tls_client_cert_key_password). Provide an "

diff --git a/src/databricks/sql/common/unified_http_client.py b/src/databricks/sql/common/unified_http_client.py
@@ -116,16 +116,9 @@ def _setup_pool_managers(self):
                     self.config.ssl_options.tls_trusted_ca_file
                 )
 
-            # Load client certificate if specified
-            if (
-                self.config.ssl_options.tls_client_cert_file
-                and self.config.ssl_options.tls_client_cert_key_file
-            ):
-                ssl_context.load_cert_chain(
-                    self.config.ssl_options.tls_client_cert_file,
-                    self.config.ssl_options.tls_client_cert_key_file,
-                    self.config.ssl_options.tls_client_cert_key_password,
-                )
+            # Load a separate cert/key pair or a combined cert+key PEM. The shared
+            # helper also reports missing/empty paths before opaque stdlib SSL errors.
+            self.config.ssl_options.load_client_cert_chain(ssl_context)
 
         # Create retry policy
         self._retry_policy = DatabricksRetryPolicy(

diff --git a/src/databricks/sql/types.py b/src/databricks/sql/types.py
@@ -21,6 +21,8 @@
 import decimal
 from ssl import SSLContext, CERT_NONE, CERT_REQUIRED, create_default_context
 
+from databricks.sql.exc import ProgrammingError
+
 
 class SSLOptions:
     tls_verify: bool
@@ -46,6 +48,74 @@ def __init__(
         self.tls_client_cert_key_file = tls_client_cert_key_file
         self.tls_client_cert_key_password = tls_client_cert_key_password
 
+    def validate_client_identity(self) -> None:
+        """Validate the shape of the mutual-TLS client identity.
+
+        ``SSLContext.load_cert_chain`` accepts a combined certificate + private-key
+        PEM when ``keyfile`` is omitted, so a certificate without a separate key file
+        is valid. The inverse is never useful: a key without a certificate would be
+        silently ignored by the stdlib and downgrade the connection to one-way TLS.
+        """
+        if self.tls_client_cert_key_file and not self.tls_client_cert_file:
+            raise ProgrammingError(
+                "tls_client_cert_key_file (client private key) requires "
+                "tls_client_cert_file (client certificate) for mutual TLS."
+            )
+
+    @staticmethod
+    def _validate_client_identity_file(
+        path: str, option_name: str, description: str
+    ) -> None:
+        """Reject an unreadable or zero-byte client-identity file clearly.
+
+        Read only one byte: the PEM parser remains responsible for validating
+        non-empty content, while this preflight can identify which of the two input
+        paths failed before ``load_cert_chain`` collapses both into an opaque error.
+        """
+        try:
+            with open(path, "rb") as file:
+                has_content = bool(file.read(1))
+        except OSError as exc:
+            raise ProgrammingError(
+                f"Failed to read {option_name} ({description}) '{path}' for mutual "
+                f"TLS: {exc}"
+            ) from exc
+
+        if not has_content:
+            raise ProgrammingError(
+                f"{option_name} ({description}) '{path}' is empty; expected "
+                "PEM-encoded content for mutual TLS."
+            )
+
+    def load_client_cert_chain(self, ssl_context: SSLContext) -> None:
+        """Load the configured mutual-TLS identity into ``ssl_context``.
+
+        Path validation intentionally precedes PEM parsing. Besides producing useful
+        diagnostics, this ensures a missing/empty key is reported as the failing input
+        even when the readable certificate file contains malformed non-empty bytes.
+        """
+        self.validate_client_identity()
+        if not self.tls_client_cert_file:
+            return
+
+        self._validate_client_identity_file(
+            self.tls_client_cert_file,
+            "tls_client_cert_file",
+            "client certificate",
+        )
+        if self.tls_client_cert_key_file:
+            self._validate_client_identity_file(
+                self.tls_client_cert_key_file,
+                "tls_client_cert_key_file",
+                "client private key",
+            )
+
+        ssl_context.load_cert_chain(
+            certfile=self.tls_client_cert_file,
+            keyfile=self.tls_client_cert_key_file,
+            password=self.tls_client_cert_key_password,
+        )
+
     def create_ssl_context(self) -> SSLContext:
         ssl_context = create_default_context(cafile=self.tls_trusted_ca_file)
 
@@ -59,12 +129,7 @@ def create_ssl_context(self) -> SSLContext:
             ssl_context.check_hostname = True
             ssl_context.verify_mode = CERT_REQUIRED
 
-        if self.tls_client_cert_file:
-            ssl_context.load_cert_chain(
-                certfile=self.tls_client_cert_file,
-                keyfile=self.tls_client_cert_key_file,
-                password=self.tls_client_cert_key_password,
-            )
+        self.load_client_cert_chain(ssl_context)
 
         return ssl_context
 

diff --git a/tests/unit/test_kernel_client.py b/tests/unit/test_kernel_client.py
@@ -2093,6 +2093,60 @@ def test_mtls_cert_only_falls_back_to_cert_for_key(self, tmp_path):
         assert out["tls_client_cert"] == b"COMBINED"
         assert out["tls_client_key"] == b"COMBINED"
 
+    def test_mtls_key_without_cert_is_rejected_before_file_access(self):
+        with pytest.raises(
+            ProgrammingError,
+            match="tls_client_cert_key_file.*requires.*tls_client_cert_file",
+        ):
+            kernel_client._kernel_tls_kwargs(
+                self._ssl_options(
+                    tls_client_cert_key_file="/path/does/not/need/to/exist.pem"
+                )
+            )
+
+    @pytest.mark.parametrize(
+        "failing_input,empty,expected_option",
+        [
+            ("certificate", False, "tls_client_cert_file"),
+            ("private key", False, "tls_client_cert_key_file"),
+            ("certificate", True, "tls_client_cert_file"),
+            ("private key", True, "tls_client_cert_key_file"),
+        ],
+        ids=[
+            "missing-certificate",
+            "missing-private-key",
+            "empty-certificate",
+            "empty-private-key",
+        ],
+    )
+    def test_mtls_unreadable_or_empty_file_names_failing_input(
+        self, tmp_path, failing_input, empty, expected_option
+    ):
+        readable_nonempty = tmp_path / "readable-nonempty.pem"
+        readable_nonempty.write_bytes(b"NOT-NECESSARILY-VALID-PEM")
+        failing_path = tmp_path / ("empty.pem" if empty else "missing.pem")
+        if empty:
+            failing_path.write_bytes(b"")
+
+        cert_file, key_file = (
+            (failing_path, readable_nonempty)
+            if failing_input == "certificate"
+            else (readable_nonempty, failing_path)
+        )
+
+        with pytest.raises(ProgrammingError) as exc_info:
+            kernel_client._kernel_tls_kwargs(
+                self._ssl_options(
+                    tls_client_cert_file=str(cert_file),
+                    tls_client_cert_key_file=str(key_file),
+                )
+            )
+
+        message = str(exc_info.value)
+        assert expected_option in message
+        assert str(failing_path) in message
+        assert ("is empty" in message) is empty
+
     def test_encrypted_client_key_rejected(self, tmp_path):
         cert = tmp_path / "client.crt"
         cert.write_bytes(b"CERT")

diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py
@@ -49,8 +49,9 @@ def test_close_uses_the_correct_session_id(self, mock_client_class):
         assert close_session_call_args.guid == b"\x22"
         assert close_session_call_args.secret == b"\x33"
 
+    @patch("%s.client.UnifiedHttpClient" % PACKAGE_NAME)
     @patch("%s.session.ThriftDatabricksClient" % PACKAGE_NAME)
-    def test_auth_args(self, mock_client_class):
+    def test_auth_args(self, mock_client_class, _mock_http_client):
         # Test that the following auth args work:
         # token = foo,
         # token = None, _tls_client_cert_file = something, _use_cert_as_auth = True
@@ -93,13 +94,15 @@ def test_tls_arg_passthrough(self, mock_client_class, mock_http_client):
             **self.DUMMY_CONNECTION_ARGS,
             _tls_verify_hostname="hostname",
             _tls_trusted_ca_file="trusted ca file",
+            _tls_client_cert_file="trusted client cert",
             _tls_client_cert_key_file="trusted client cert",
             _tls_client_cert_key_password="key password",
         )
 
         kwargs = mock_client_class.call_args[1]
         assert kwargs["_tls_verify_hostname"] == "hostname"
         assert kwargs["_tls_trusted_ca_file"] == "trusted ca file"
+        assert kwargs["_tls_client_cert_file"] == "trusted client cert"
         assert kwargs["_tls_client_cert_key_file"] == "trusted client cert"
         assert kwargs["_tls_client_cert_key_password"] == "key password"
 

diff --git a/tests/unit/test_ssl_options.py b/tests/unit/test_ssl_options.py
@@ -0,0 +1,89 @@
+from unittest.mock import Mock
+
+import pytest
+
+from databricks.sql.exc import ProgrammingError
+from databricks.sql.types import SSLOptions
+
+
+class TestSSLOptionsMutualTls:
+    def test_private_key_without_client_certificate_is_rejected_before_file_access(
+        self,
+    ):
+        options = SSLOptions(
+            tls_client_cert_key_file="/path/does/not/need/to/exist.pem"
+        )
+
+        with pytest.raises(ProgrammingError) as exc_info:
+            options.load_client_cert_chain(Mock())
+
+        message = str(exc_info.value)
+        assert "tls_client_cert_key_file" in message
+        assert "tls_client_cert_file" in message
+        assert "requires" in message
+
+    @pytest.mark.parametrize(
+        "failing_input,empty",
+        [
+            ("certificate", False),
+            ("private key", False),
+            ("certificate", True),
+            ("private key", True),
+        ],
+        ids=[
+            "missing-certificate",
+            "missing-private-key",
+            "empty-certificate",
+            "empty-private-key",
+        ],
+    )
+    def test_unreadable_or_empty_identity_file_names_failing_input(
+        self, tmp_path, failing_input, empty
+    ):
+        # Deliberately not PEM: file readability/emptiness must be checked for both
+        # inputs before SSL parsing begins, so a malformed peer cannot mask the
+        # missing/empty input this case is exercising.
+        readable_nonempty = tmp_path / "readable-nonempty.pem"
+        readable_nonempty.write_bytes(b"not PEM, but readable and non-empty")
+        failing_path = tmp_path / ("empty.pem" if empty else "missing.pem")
+        if empty:
+            failing_path.write_bytes(b"")
+
+        if failing_input == "certificate":
+            cert_file = failing_path
+            key_file = readable_nonempty
+            expected_option = "tls_client_cert_file"
+        else:
+            cert_file = readable_nonempty
+            key_file = failing_path
+            expected_option = "tls_client_cert_key_file"
+
+        ssl_context = Mock()
+        options = SSLOptions(
+            tls_client_cert_file=str(cert_file),
+            tls_client_cert_key_file=str(key_file),
+        )
+
+        with pytest.raises(ProgrammingError) as exc_info:
+            options.load_client_cert_chain(ssl_context)
+
+        message = str(exc_info.value)
+        assert expected_option in message
+        assert str(failing_path) in message
+        assert ("is empty" in message) is empty
+        ssl_context.load_cert_chain.assert_not_called()
+
+    def test_combined_cert_key_file_and_password_are_forwarded(self, tmp_path):
+        combined = tmp_path / "combined.pem"
+        combined.write_bytes(b"non-empty combined PEM placeholder")
+        ssl_context = Mock()
+        password = "encrypted-key-password"
+
+        SSLOptions(
+            tls_client_cert_file=str(combined),
+            tls_client_cert_key_password=password,
+        ).load_client_cert_chain(ssl_context)
+
+        ssl_context.load_cert_chain.assert_called_once_with(
+            certfile=str(combined), keyfile=None, password=password
+        )