Reject timestampAsString override on kernel sessions

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Release History
 
 # Unreleased
+- Kernel backend (`use_kernel=True`): Reject attempts to set `spark.thriftserver.arrowBasedRowSet.timestampAsString` to a non-false value during session creation, matching the Thrift backend's validation.
 - Kernel backend (`use_kernel=True`): OAuth **M2M with a JWT private-key client assertion** (RFC 7523) is now supported. Pass `oauth_client_id` + `oauth_jwt_key_file` + `oauth_jwt_kid` (with optional `oauth_jwt_passphrase` for an encrypted PKCS#8 key, `oauth_jwt_algorithm` defaulting to `RS256`, `oauth_scopes`, and `token_url` for the IdP token endpoint) and the connector routes them to the kernel's `auth_type="oauth-m2m-jwt"`, which signs a short-lived assertion with the private key instead of sending a client secret. The kernel owns the token lifecycle. A private-key file is treated as unambiguous JWT M2M intent and is mutually exclusive with `oauth_client_secret` / `credentials_provider` (both raise `NotSupportedError`). Verified end-to-end against an Azure Databricks workspace with the service principal's public certificate registered on its Entra ID app registration. Requires `databricks-sql-kernel >= 0.2.0` with JWT support.
 - Kernel backend (`use_kernel=True`): OAuth U2M with `auth_type="databricks-oauth"` now forwards the connector's `databricks-sql-python` OAuth-app bundle (`client_id` + `sql offline_access` scopes + redirect port) into the kernel, so a bare U2M connection authenticates as `databricks-sql-python` — parity with the Thrift path — instead of inheriting the kernel's own `databricks-sql-connector` default. A caller-supplied `oauth_client_id` (with its coupled `oauth_redirect_port`) is honored, as is a caller-supplied `oauth_scopes`; absent one, the connector default (`sql offline_access`) is forwarded. Note: the kernel binds a single U2M redirect port, so unlike the Thrift path (which tries the full `8020..8024` range) the kernel path uses only one port and does not fall back to the next port if it is already bound — pass `oauth_redirect_port` (with `oauth_client_id`) to pick a free one on a port collision (PECOBLR-4040)
 - Kernel backend (`use_kernel=True`): **Azure Entra (Azure AD) service-principal M2M is now supported.** `auth_type="azure-sp-m2m"` forwards `azure_client_id` / `azure_client_secret`; the kernel is the Azure-aware auth core — it builds the Entra v2.0 token endpoint and the `{app_id}/.default` scope, and **auto-discovers the tenant** from the workspace's `/aad/auth` redirect when `azure_tenant_id` is omitted (matching Thrift). The `Authorization` bearer is the Databricks-audience data token, which alone authenticates a workspace-member SP. Set `azure_workspace_resource_id` and the kernel also sends the Azure SP management token (`X-Databricks-Azure-SP-Management-Token`) + `X-Databricks-Azure-Workspace-Resource-Id` header (matching the JDBC driver), so a service principal with an Azure RBAC role but no workspace membership can authenticate; omit it and no ARM management-scope token is fetched. Azure AD **U2M** (`auth_type="azure-oauth"`) now routes to the kernel's OAuth U2M flow, identically to `auth_type="databricks-oauth"`: the kernel runs the in-house workspace-federated browser flow, which Azure workspaces support (the workspace federates login to Entra). It forwards the connector's `databricks-sql-python` OAuth app, not the Thrift Azure app (`96eecda7` / port 8030), which is registered for Thrift's direct-Entra flow the kernel does not perform (PECOBLR-4141; PECOBLR-4120)

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

@@ -43,6 +43,7 @@
     SessionId,
 )
 from databricks.sql.exc import (
+    Error,
     InterfaceError,
     NotSupportedError,
     ProgrammingError,
@@ -63,6 +64,8 @@
 # per-request skip-and-warn.
 _KERNEL_MANAGED_HEADERS = frozenset({"authorization", "x-databricks-org-id"})
 
+TIMESTAMP_AS_STRING_CONFIG = "spark.thriftserver.arrowBasedRowSet.timestampAsString"
+
 # Leading verbs of SQL volume/staging statements. Detected by the
 # leading token (case-insensitive) so the kernel backend can fail loud
 # on staging ops it can't service — see ``execute_command``.
@@ -99,6 +102,21 @@ def _strip_leading_sql_comments(sql: str) -> str:
     return sql[i:]
 
 
+def _check_session_configuration(session_configuration: Dict[str, str]) -> None:
+    # This client expects timestampAsString to be false, so do not allow overrides.
+    if (
+        session_configuration.get(TIMESTAMP_AS_STRING_CONFIG, "false").lower()
+        != "false"
+    ):
+        raise Error(
+            "Invalid session configuration: {} cannot be changed "
+            "while using the Databricks SQL connector, it must be false not {}".format(
+                TIMESTAMP_AS_STRING_CONFIG,
+                session_configuration[TIMESTAMP_AS_STRING_CONFIG],
+            )
+        )
+
+
 def _is_not_found(exc: BaseException) -> bool:
     """True iff ``exc`` is a kernel ``NotFound`` error (HTTP 404 /
     ``STATEMENT_NOT_FOUND``).
@@ -310,16 +328,29 @@ def open_session(
         session_conf: Optional[Dict[str, str]] = None
         if session_configuration:
             session_conf = {k: str(v) for k, v in session_configuration.items()}
-        # The kwarg builds run INSIDE the try so the ``finally`` scrub
-        # below always fires — including when ``kernel_auth_kwargs``
+        # The kwarg builds — and the session-conf validation — run INSIDE
+        # the try so the ``finally`` scrub below always fires, including
+        # when the validation rejects or when ``kernel_auth_kwargs``
         # itself raises mid-build (e.g. an OAuth token-exchange failure
-        # while the M2M secret is in hand). Pre-declared empty so the
-        # ``finally`` can reference them unconditionally even on an early
-        # raise. Building here (not in ``__init__``) keeps the bearer
-        # token's in-process lifetime as short as possible.
+        # while the M2M secret is in hand). ``self._auth_options`` already
+        # holds the secret at this point (set in ``__init__``), so an
+        # early raise before the try would leave it un-scrubbed. Pre-declared
+        # empty so the ``finally`` can reference them unconditionally even
+        # on an early raise. Building here (not in ``__init__``) keeps the
+        # bearer token's in-process lifetime as short as possible.
         auth_kwargs: Dict[str, Any] = {}
         tls_kwargs: Dict[str, Any] = {}
         try:
+            if session_conf is not None:
+                # ``_check_session_configuration`` only *rejects* a non-``false``
+                # ``TIMESTAMP_AS_STRING_CONFIG`` override; unlike the Thrift backend
+                # (``thrift_backend.py``), it deliberately does NOT inject
+                # ``TIMESTAMP_AS_STRING_CONFIG = "false"`` into ``session_conf``.
+                # The kernel owns type handling natively (Arrow / complex-types),
+                # and the Thrift-specific conf may be unsupported on the SEA session
+                # path, so pinning it here would be redundant at best and rejected
+                # at worst. The divergence is intentional, not accidental.
+                _check_session_configuration(session_conf)
             auth_kwargs = kernel_auth_kwargs(
                 self._auth_provider,
                 self._auth_options,

tests/unit/test_kernel_client.py

@@ -89,6 +89,7 @@ def __init__(
 from databricks.sql.backend.types import CommandId, CommandState
 from databricks.sql.exc import (
     DatabaseError,
+    Error,
     InterfaceError,
     NotSupportedError,
     OperationalError,
@@ -306,6 +307,75 @@ def test_open_session_rejects_double_open(monkeypatch):
         c.open_session(session_configuration=None, catalog=None, schema=None)
 
 
+def test_open_session_rejects_timestamp_as_string_true(monkeypatch):
+    session_ctor = MagicMock()
+    monkeypatch.setattr(kernel_client._kernel, "Session", session_ctor)
+
+    c = _make_client()
+
+    with pytest.raises(Error, match="timestampAsString cannot be changed"):
+        c.open_session(
+            session_configuration={
+                "spark.thriftserver.arrowBasedRowSet.timestampAsString": True
+            },
+            catalog=None,
+            schema=None,
+        )
+
+    session_ctor.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["TRUE", "True", "true"],  # ``.lower()`` normalizes case before comparison
+)
+def test_open_session_rejects_timestamp_as_string_true_case_insensitive(
+    monkeypatch, value
+):
+    """A truthy ``TIMESTAMP_AS_STRING_CONFIG`` string is rejected
+    regardless of case — the guard lower-cases before comparing."""
+    session_ctor = MagicMock()
+    monkeypatch.setattr(kernel_client._kernel, "Session", session_ctor)
+
+    c = _make_client()
+
+    with pytest.raises(Error, match="timestampAsString cannot be changed"):
+        c.open_session(
+            session_configuration={
+                "spark.thriftserver.arrowBasedRowSet.timestampAsString": value
+            },
+            catalog=None,
+            schema=None,
+        )
+
+    session_ctor.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["false", "False", "FALSE"],  # a valid pin must NOT be rejected, any case
+)
+def test_open_session_accepts_timestamp_as_string_false(monkeypatch, value):
+    """A legitimate ``TIMESTAMP_AS_STRING_CONFIG = "false"`` pin (any
+    case) still opens the session — the guard rejects only non-false
+    overrides, so a valid config is not wrongly refused."""
+    fake_session = MagicMock()
+    fake_session.return_value.session_id = "sess-id"
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+
+    c = _make_client()
+
+    c.open_session(
+        session_configuration={
+            "spark.thriftserver.arrowBasedRowSet.timestampAsString": value
+        },
+        catalog=None,
+        schema=None,
+    )
+
+    fake_session.assert_called_once()
+
+
 @pytest.mark.parametrize(
     "kwargs, expected_flag",
     [