fix(kernel): address review — re-pin KERNEL_REV, secret hygiene, auth ambiguity, TLS superset

Addresses @gopalldb's review on #819.

P0:
- Re-pin KERNEL_REV to the squash-merge SHA of the now-merged kernel
  PR (ec2288742cbac0cd9fab50da353e8405972eefe9 on kernel main),
  replacing the orphaned branch-HEAD SHA.

P1:
- Scrub oauth_client_secret from the long-lived self._auth_options in
  open_session's finally (even on the failure path). It outlives the
  method, so a retained secret was exposed to vars()/pickle/debugger
  far longer than needed; the kernel now owns it.
- tls_verify=False now also emits tls_skip_hostname_verify=True —
  no-verify subsumes hostname verification, matching SSLOptions'
  create_ssl_context (check_hostname=False when tls_verify is False).
- Reject ambiguous auth combos before resolving, instead of silently
  picking M2M: (a) credentials_provider + oauth_client_secret, and
  (b) a U2M auth_type (databricks-oauth/azure-oauth) + oauth_client_secret.
  Both raise NotSupportedError naming the conflict, so the failure is at
  session-open rather than a confusing first-call 401 against the wrong
  principal.
- Secret-leak tests: oauth_client_secret is forwarded to the kernel but
  scrubbed from _auth_options, and absent from repr(conn) / vars(conn);
  scrub runs even when open_session raises.

P2:
- _read_pem_bytes rejects an empty/whitespace CA/cert file with a clear
  ProgrammingError instead of passing empty PEM to the kernel.
- _normalize_scopes raises ProgrammingError on a non-str/list/tuple
  oauth_scopes rather than silently dropping to default scopes.
- Added U2M oauth_scopes forwarding test.

150 kernel unit tests pass; black + mypy clean.

Note: KERNEL_REV now points at the merged kernel main SHA; no further
re-pin needed before merge.

Co-authored-by: Isaac
Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>

Author: vikrantpuppala

KERNEL_REV

@@ -1 +1 @@
-774569bc6d816f662c0f4f8a05859d1f4f26e9d3
+ec2288742cbac0cd9fab50da353e8405972eefe9

src/databricks/sql/backend/kernel/auth_bridge.py

@@ -127,6 +127,13 @@ def kernel_auth_kwargs(
 
     Resolution order:
 
+    0. **Ambiguity guards** — reject conflicting auth signals *before*
+       resolving, so an ambiguous request fails loudly at session-open
+       rather than silently picking one flow (and failing later as a
+       confusing 401 against the wrong principal):
+       - a custom ``credentials_provider`` *and* M2M kwargs together;
+       - a U2M ``auth_type`` (``databricks-oauth`` / ``azure-oauth``)
+         *and* ``oauth_client_secret`` together.
     1. **OAuth M2M** — ``oauth_client_id`` + ``oauth_client_secret``
        both present → forward raw creds to the kernel's ``oauth-m2m``.
     2. **PAT** — the built provider is (or wraps) an
@@ -140,14 +147,37 @@ def kernel_auth_kwargs(
 
     M2M is checked before PAT so that a workload passing both an
     access token *and* M2M creds resolves to the (refreshing) M2M path
-    rather than a static token.
+    rather than a static token. (Token + M2M is not treated as
+    ambiguous: a PAT is often present as ambient config the caller
+    didn't intend as the primary credential, whereas an explicit
+    ``oauth_client_secret`` is unambiguous M2M intent.)
     """
     opts = auth_options or {}
 
-    # 1. OAuth M2M — raw client-credentials pair forwarded to the kernel.
     client_id = opts.get("oauth_client_id")
     client_secret = opts.get("oauth_client_secret")
-    if client_id and client_secret:
+    auth_type = opts.get("auth_type")
+    has_m2m = bool(client_id and client_secret)
+
+    # 0. Ambiguity guards — fail before any flow is chosen.
+    if client_secret and opts.get("credentials_provider") is not None:
+        raise NotSupportedError(
+            "Ambiguous auth on use_kernel=True: both a custom "
+            "credentials_provider and oauth_client_secret were provided. "
+            "Pass exactly one — oauth_client_id + oauth_client_secret for "
+            "kernel-managed M2M, or use the Thrift backend (default) for "
+            "credentials_provider."
+        )
+    if client_secret and auth_type in ("databricks-oauth", "azure-oauth"):
+        raise NotSupportedError(
+            f"Ambiguous auth on use_kernel=True: auth_type={auth_type!r} selects "
+            "the U2M browser flow, but oauth_client_secret was also provided "
+            "(machine-to-machine). Drop oauth_client_secret for U2M, or drop "
+            "auth_type for M2M."
+        )
+
+    # 1. OAuth M2M — raw client-credentials pair forwarded to the kernel.
+    if has_m2m:
         kwargs: Dict[str, Any] = {
             "auth_type": "oauth-m2m",
             "client_id": client_id,
@@ -169,7 +199,6 @@ def kernel_auth_kwargs(
         return {"auth_type": "pat", "access_token": token}
 
     # 3. OAuth U2M — browser authorization-code flow; the kernel runs it.
-    auth_type = opts.get("auth_type")
     if auth_type in ("databricks-oauth", "azure-oauth"):
         kwargs = {"auth_type": "oauth-u2m"}
         if client_id:
@@ -220,4 +249,10 @@ def _normalize_scopes(scopes: Any) -> Optional[list]:
     if isinstance(scopes, (list, tuple)):
         parts = [str(s) for s in scopes if s]
         return parts or None
-    return None
+    # Anything else (int, dict, bool, …) is a caller error. Fail loudly
+    # rather than silently dropping the scopes to None and surprising
+    # the user with default scopes.
+    raise ProgrammingError(
+        f"oauth_scopes must be a list/tuple of strings or a space-delimited "
+        f"string, got {type(scopes).__name__}."
+    )

src/databricks/sql/backend/kernel/client.py

@@ -211,6 +211,14 @@ def open_session(
             auth_kwargs.pop("access_token", None)
             auth_kwargs.pop("client_secret", None)
             tls_kwargs.pop("tls_client_key", None)
+            # Also scrub the long-lived copy. ``self._auth_options``
+            # outlives this method (it's set in ``__init__`` and the
+            # connector object can live for the whole connection), so a
+            # retained ``oauth_client_secret`` would be exposed to
+            # ``vars(conn)`` / pickle / a debugger dump for far longer
+            # than the credential needs to exist. The kernel now owns
+            # the secret, so drop ours.
+            self._auth_options.pop("oauth_client_secret", None)
 
         # Use the kernel's real server-issued session id, not a
         # synthetic UUID. Matches what the native SEA backend does.
@@ -665,7 +673,8 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     Mappings (note the inverted booleans — the connector expresses
     *verify*, the kernel expresses *skip*):
 
-    - ``tls_verify=False``          → ``tls_skip_verify=True``
+    - ``tls_verify=False``          → ``tls_skip_verify=True`` **and**
+      ``tls_skip_hostname_verify=True``
     - ``tls_verify_hostname=False`` → ``tls_skip_hostname_verify=True``
     - ``tls_trusted_ca_file``       → ``tls_ca_cert`` (PEM bytes)
     - ``tls_client_cert_file`` (+ key file) → ``tls_client_cert`` /
@@ -680,10 +689,16 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     kwargs: Dict[str, Any] = {}
 
     # Inverted booleans. Emit only the insecure (skip) direction so the
-    # default secure path stays implicit.
+    # default secure path stays implicit. ``tls_verify=False`` disables
+    # all chain validation, which subsumes hostname verification — so we
+    # also emit ``tls_skip_hostname_verify`` to match ``SSLOptions``'s
+    # 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:
         kwargs["tls_skip_verify"] = True
-    if getattr(ssl_options, "tls_verify_hostname", True) is False:
+        kwargs["tls_skip_hostname_verify"] = True
+    elif getattr(ssl_options, "tls_verify_hostname", True) is False:
         kwargs["tls_skip_hostname_verify"] = True
 
     ca_file = getattr(ssl_options, "tls_trusted_ca_file", None)
@@ -716,14 +731,23 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
 
 def _read_pem_bytes(path: str, label: str) -> bytes:
     """Read a PEM file into bytes, mapping IO errors to a clear
-    ``ProgrammingError`` that names the offending TLS option."""
+    ``ProgrammingError`` that names the offending TLS option. An empty
+    file is rejected here too — otherwise it reaches the kernel as
+    empty PEM and surfaces as a cryptic ``no certificates found`` /
+    parse error far from the misconfigured path."""
     try:
         with open(path, "rb") as f:
-            return f.read()
+            data = f.read()
     except OSError as exc:
         raise ProgrammingError(
             f"Failed to read {label} '{path}' for the kernel TLS config: {exc}"
         ) from exc
+    if not data.strip():
+        raise ProgrammingError(
+            f"{label} '{path}' is empty; expected PEM-encoded content for the "
+            "kernel TLS config."
+        )
+    return data
 
 
 def _drain_kernel_handle(handle: Any) -> Any:

tests/unit/test_kernel_auth_bridge.py

@@ -262,3 +262,60 @@ def test_u2m_forwards_client_id_and_redirect_port(self):
             "client_id": "custom-client",
             "redirect_port": 8030,
         }
+
+    @pytest.mark.parametrize("auth_type", ["databricks-oauth", "azure-oauth"])
+    def test_u2m_forwards_scopes(self, auth_type):
+        kwargs = kernel_auth_kwargs(
+            _FakeOAuthProvider(),
+            {"auth_type": auth_type, "oauth_scopes": ["all-apis", "offline_access"]},
+        )
+        assert kwargs["oauth_scopes"] == ["all-apis", "offline_access"]
+
+
+class TestKernelAuthAmbiguity:
+    """Conflicting auth signals must fail loudly at session-open rather
+    than silently resolving to one flow (which would surface later as a
+    confusing 401 against the wrong principal)."""
+
+    def test_credentials_provider_plus_m2m_is_rejected(self):
+        def _creds_provider():
+            return lambda: {"Authorization": "Bearer x"}
+
+        with pytest.raises(NotSupportedError, match="Ambiguous auth"):
+            kernel_auth_kwargs(
+                _FakeOAuthProvider(),
+                {
+                    "oauth_client_id": "id",
+                    "oauth_client_secret": "sec",
+                    "credentials_provider": _creds_provider,
+                },
+            )
+
+    @pytest.mark.parametrize("auth_type", ["databricks-oauth", "azure-oauth"])
+    def test_u2m_auth_type_plus_client_secret_is_rejected(self, auth_type):
+        # User asked for U2M (browser) but also passed a secret (M2M).
+        # Don't silently route M2M against the wrong principal.
+        with pytest.raises(NotSupportedError, match="Ambiguous auth"):
+            kernel_auth_kwargs(
+                _FakeOAuthProvider(),
+                {
+                    "auth_type": auth_type,
+                    "oauth_client_id": "id",
+                    "oauth_client_secret": "sec",
+                },
+            )
+
+
+class TestKernelScopesNormalization:
+    def test_unknown_scope_type_raises(self):
+        # A non-str/list/tuple oauth_scopes is a caller error; fail loudly
+        # rather than silently dropping to default scopes.
+        with pytest.raises(ProgrammingError, match="oauth_scopes must be"):
+            kernel_auth_kwargs(
+                _FakeOAuthProvider(),
+                {
+                    "oauth_client_id": "id",
+                    "oauth_client_secret": "sec",
+                    "oauth_scopes": 123,
+                },
+            )

tests/unit/test_kernel_client.py

@@ -803,11 +803,20 @@ def test_default_ssl_options_emit_no_tls_kwargs(self):
     def test_none_ssl_options_emit_no_tls_kwargs(self):
         assert kernel_client._kernel_tls_kwargs(None) == {}
 
-    def test_verify_off_maps_to_skip_verify(self):
+    def test_verify_off_maps_to_skip_verify_and_hostname(self):
+        # tls_verify=False disables all chain validation, which subsumes
+        # hostname verification — so both skip flags are emitted, matching
+        # SSLOptions.create_ssl_context (check_hostname=False when
+        # tls_verify is False).
         out = kernel_client._kernel_tls_kwargs(self._ssl_options(tls_verify=False))
-        assert out == {"tls_skip_verify": True}
+        assert out == {
+            "tls_skip_verify": True,
+            "tls_skip_hostname_verify": True,
+        }
 
     def test_hostname_verify_off_maps_to_skip_hostname(self):
+        # Only hostname verification off (chain still validated) → just
+        # the hostname skip, no tls_skip_verify.
         out = kernel_client._kernel_tls_kwargs(
             self._ssl_options(tls_verify_hostname=False)
         )
@@ -862,3 +871,88 @@ def test_missing_ca_file_raises_programming_error(self):
             kernel_client._kernel_tls_kwargs(
                 self._ssl_options(tls_trusted_ca_file="/no/such/ca.pem")
             )
+
+
+# ---------------------------------------------------------------------------
+# Secret hygiene: oauth_client_secret must not be retained or leak.
+# ---------------------------------------------------------------------------
+
+
+class TestKernelSecretHygiene:
+    """The new ``oauth_client_secret`` M2M kwarg is credential material.
+    Verify it's scrubbed after ``open_session`` and never exposed via
+    the connector's ``repr`` / ``vars`` or through a mapped exception."""
+
+    _SECRET = "super-secret-m2m-value"
+
+    def _client_with_m2m(self, monkeypatch, captured):
+        def fake_session(**kw):
+            captured.update(kw)
+            sess = MagicMock()
+            sess.session_id = "sess-id"
+            return sess
+
+        monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+        return kernel_client.KernelDatabricksClient(
+            server_hostname="example.cloud.databricks.com",
+            http_path="/sql/1.0/warehouses/abc",
+            auth_provider=AccessTokenAuthProvider("dapi-test"),
+            ssl_options=None,
+            auth_options={
+                "oauth_client_id": "sp-uuid",
+                "oauth_client_secret": self._SECRET,
+            },
+        )
+
+    def test_secret_forwarded_then_scrubbed_from_auth_options(self, monkeypatch):
+        captured = {}
+        c = self._client_with_m2m(monkeypatch, captured)
+        c.open_session(session_configuration=None, catalog=None, schema=None)
+
+        # Forwarded to the kernel Session as client_secret...
+        assert captured.get("client_secret") == self._SECRET
+        # ...but not retained on the long-lived connector.
+        assert "oauth_client_secret" not in c._auth_options
+
+    def test_secret_absent_from_repr_and_vars_after_open(self, monkeypatch):
+        captured = {}
+        c = self._client_with_m2m(monkeypatch, captured)
+        c.open_session(session_configuration=None, catalog=None, schema=None)
+
+        assert self._SECRET not in repr(c)
+        assert self._SECRET not in str(vars(c))
+
+    def test_secret_not_in_mapped_open_session_exception(self, monkeypatch):
+        # If the kernel Session constructor raises with the secret in its
+        # message/args, the mapped PEP-249 exception must still be raised
+        # (we don't assert the kernel scrubs its own error — that's the
+        # kernel's job — but we confirm the connector's scrub still runs
+        # via the finally block even on the failure path).
+        def boom_session(**kw):
+            raise RuntimeError("kernel open failed")
+
+        monkeypatch.setattr(kernel_client._kernel, "Session", boom_session)
+        c = kernel_client.KernelDatabricksClient(
+            server_hostname="example.cloud.databricks.com",
+            http_path="/sql/1.0/warehouses/abc",
+            auth_provider=AccessTokenAuthProvider("dapi-test"),
+            ssl_options=None,
+            auth_options={
+                "oauth_client_id": "sp-uuid",
+                "oauth_client_secret": self._SECRET,
+            },
+        )
+        with pytest.raises(Exception):
+            c.open_session(session_configuration=None, catalog=None, schema=None)
+        # Scrubbed even though open_session raised (finally block).
+        assert "oauth_client_secret" not in c._auth_options
+
+
+class TestKernelTlsEmptyCaFile:
+    def test_empty_ca_file_raises_programming_error(self, tmp_path):
+        from databricks.sql.types import SSLOptions
+
+        ca = tmp_path / "empty.pem"
+        ca.write_bytes(b"   \n")
+        with pytest.raises(ProgrammingError, match="is empty"):
+            kernel_client._kernel_tls_kwargs(SSLOptions(tls_trusted_ca_file=str(ca)))