fix(kernel): forward full OAuth U2M app bundle into kernel (PECOBLR-4040)

On the use_kernel path, OAuth U2M forwarded only whatever the caller
explicitly set, sending a bare oauth-u2m otherwise. Since PECOBLR-4039
changed the kernel core default U2M app to databricks-sql-connector /
sql offline_access / port 8030, a bare U2M connection authenticated as
the wrong identity.

The connector is an OVERRIDE of the kernel default, so it now forwards
its full coupled bundle (client_id + oauth_scopes + redirect_port). Each
field falls back to the connector's registered databricks-sql-python
(or azure) default from the existing PYSQL_OAUTH_* constants, giving
parity with the Thrift path. Explicit caller overrides and
identity_federation_client_id forwarding are preserved.

Co-authored-by: Isaac

Author: eric-wang-1990

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Release History
 
+# Unreleased
+- Kernel backend (`use_kernel=True`): OAuth U2M now forwards the connector's full OAuth-app bundle (`client_id` + scopes + redirect port) into the kernel, so a bare U2M connection authenticates as `databricks-sql-python` with `sql offline_access` — parity with the Thrift path — instead of inheriting the kernel's own `databricks-sql-connector` default. Caller-supplied `oauth_client_id` / `oauth_scopes` / `oauth_redirect_port` are still honored (PECOBLR-4040)
+
 # 4.4.0 (2026-07-22)
 - Raised the minimum supported Python version to 3.10, dropping the end-of-life 3.8/3.9, to update the lockfile and clear CVE-flagged dependencies in the repo (databricks/databricks-sql-python#798)
 - Fix: `REMOVE` staging operations no longer require `staging_allowed_local_path` to be set, since removing a remote file does not touch the local filesystem (databricks/databricks-sql-python#726)

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

@@ -48,6 +48,13 @@
 import re
 from typing import Any, Dict, Optional
 
+from databricks.sql.auth.auth import (
+    PYSQL_OAUTH_AZURE_CLIENT_ID,
+    PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE,
+    PYSQL_OAUTH_CLIENT_ID,
+    PYSQL_OAUTH_REDIRECT_PORT_RANGE,
+    PYSQL_OAUTH_SCOPES,
+)
 from databricks.sql.auth.authenticators import AccessTokenAuthProvider, AuthProvider
 from databricks.sql.auth.token_federation import TokenFederationProvider
 from databricks.sql.exc import NotSupportedError, ProgrammingError
@@ -148,8 +155,14 @@ def kernel_auth_kwargs(
     2. **PAT** — the built provider is (or wraps) an
        ``AccessTokenAuthProvider`` → extract the bearer token.
     3. **OAuth U2M** — ``auth_type`` is ``databricks-oauth`` /
-       ``azure-oauth`` → forward optional ``oauth_client_id`` /
-       ``oauth_redirect_port`` to the kernel's ``oauth-u2m``.
+       ``azure-oauth`` → forward the connector's *full* coupled OAuth-app
+       bundle (``client_id`` + ``oauth_scopes`` + ``redirect_port``) to
+       the kernel's ``oauth-u2m``. Each field falls back to the
+       connector's registered ``databricks-sql-python`` (or azure)
+       default when the caller doesn't override it, so a bare U2M
+       connection authenticates as ``databricks-sql-python`` — parity
+       with the Thrift path — rather than the kernel's own
+       ``databricks-sql-connector`` default (PECOBLR-4039/4040).
     4. **Custom credentials_provider** → ``NotSupportedError`` (opaque
        token source; no raw creds for the kernel to own).
     5. Anything else → ``NotSupportedError``.
@@ -214,16 +227,51 @@ def kernel_auth_kwargs(
         return kwargs
 
     # 3. OAuth U2M — browser authorization-code flow; the kernel runs it.
+    #
+    #    The kernel's core default U2M app is databricks-sql-connector /
+    #    sql offline_access / port 8030 (PECOBLR-4039). The Python
+    #    connector is an OVERRIDE of that default: on this path we forward
+    #    its OWN full bundle rather than letting the kernel fall back to
+    #    the connector default. Forwarding a bare oauth-u2m would
+    #    authenticate as databricks-sql-connector, breaking parity with
+    #    the Thrift path (which authenticates as databricks-sql-python).
+    #
+    #    client_id + scopes + redirect_port are coupled per OAuth app —
+    #    each app registers its own redirect URI — so all three are
+    #    resolved together: an explicit caller value wins; otherwise the
+    #    connector's registered databricks-sql-python (or azure) bundle is
+    #    used, mirroring the defaults get_python_sql_connector_auth_provider
+    #    applies on the Thrift path.
+    #
+    #    Only the redirect PORT is routable into the kernel: it derives
+    #    http://localhost:{port}, with scheme/host/path fixed. The
+    #    connector registers a port *range* for its app but the kernel
+    #    accepts a single port, so we forward the first (canonical)
+    #    registered port.
     if auth_type in ("databricks-oauth", "azure-oauth"):
-        kwargs = {"auth_type": "oauth-u2m"}
-        if client_id:
-            kwargs["client_id"] = client_id
+        is_azure = auth_type == "azure-oauth"
+        default_client_id = (
+            PYSQL_OAUTH_AZURE_CLIENT_ID if is_azure else PYSQL_OAUTH_CLIENT_ID
+        )
+        default_port_range = (
+            PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE
+            if is_azure
+            else PYSQL_OAUTH_REDIRECT_PORT_RANGE
+        )
         redirect_port = opts.get("oauth_redirect_port")
-        if redirect_port is not None:
-            kwargs["redirect_port"] = int(redirect_port)
         scopes = _normalize_scopes(opts.get("oauth_scopes"))
-        if scopes is not None:
-            kwargs["oauth_scopes"] = scopes
+        kwargs = {
+            "auth_type": "oauth-u2m",
+            "client_id": client_id or default_client_id,
+            "redirect_port": (
+                int(redirect_port)
+                if redirect_port is not None
+                else default_port_range[0]
+            ),
+            "oauth_scopes": (
+                scopes if scopes is not None else list(PYSQL_OAUTH_SCOPES)
+            ),
+        }
         if federation_client_id:
             kwargs["identity_federation_client_id"] = federation_client_id
         return kwargs

tests/unit/test_kernel_auth_bridge.py

@@ -26,6 +26,13 @@
 # require the kernel wheel). So this test can run on the
 # default-deps CI matrix without any extras. No importorskip needed.
 
+from databricks.sql.auth.auth import (
+    PYSQL_OAUTH_CLIENT_ID,
+    PYSQL_OAUTH_AZURE_CLIENT_ID,
+    PYSQL_OAUTH_SCOPES,
+    PYSQL_OAUTH_REDIRECT_PORT_RANGE,
+    PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE,
+)
 from databricks.sql.auth.authenticators import (
     AccessTokenAuthProvider,
     AuthProvider,
@@ -240,31 +247,93 @@ def test_client_id_without_secret_does_not_trigger_m2m(self):
 
 
 class TestKernelOAuthU2M:
-    @pytest.mark.parametrize("auth_type", ["databricks-oauth", "azure-oauth"])
-    def test_u2m_routes_to_kernel_u2m(self, auth_type):
+    """The kernel core default U2M app is ``databricks-sql-connector`` /
+    ``sql offline_access`` / port 8030 (see PECOBLR-4039). The Python
+    connector is an OVERRIDE: on the kernel path it must forward its OWN
+    full bundle — ``client_id`` + ``oauth_scopes`` + ``redirect_port`` —
+    because the three are coupled per OAuth app. Forwarding a partial
+    bundle would let the kernel fill the rest from the connector default,
+    authenticating as the wrong principal / against an unregistered
+    redirect URI. So bare U2M must forward the complete
+    ``databricks-sql-python`` bundle for parity with the Thrift path."""
+
+    def test_bare_databricks_oauth_forwards_full_python_bundle(self):
+        # No overrides → forward the databricks-sql-python bundle in full
+        # so the kernel does NOT fall back to its databricks-sql-connector
+        # default. This is the parity-with-Thrift acceptance criterion.
+        kwargs = kernel_auth_kwargs(
+            _FakeOAuthProvider(),
+            {"auth_type": "databricks-oauth"},
+        )
+        assert kwargs == {
+            "auth_type": "oauth-u2m",
+            "client_id": PYSQL_OAUTH_CLIENT_ID,
+            "redirect_port": PYSQL_OAUTH_REDIRECT_PORT_RANGE[0],
+            "oauth_scopes": list(PYSQL_OAUTH_SCOPES),
+        }
+
+    def test_bare_azure_oauth_forwards_full_azure_bundle(self):
+        kwargs = kernel_auth_kwargs(
+            _FakeOAuthProvider(),
+            {"auth_type": "azure-oauth"},
+        )
+        assert kwargs == {
+            "auth_type": "oauth-u2m",
+            "client_id": PYSQL_OAUTH_AZURE_CLIENT_ID,
+            "redirect_port": PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE[0],
+            "oauth_scopes": list(PYSQL_OAUTH_SCOPES),
+        }
+
+    def test_u2m_custom_client_id_scopes_and_port_honored(self):
+        # A caller overriding the app supplies the full coupled bundle;
+        # every field is forwarded verbatim.
         kwargs = kernel_auth_kwargs(
             _FakeOAuthProvider(),
-            {"auth_type": auth_type},
+            {
+                "auth_type": "databricks-oauth",
+                "oauth_client_id": "custom-client",
+                "oauth_scopes": ["custom-scope", "offline_access"],
+                "oauth_redirect_port": 9999,
+            },
         )
-        assert kwargs == {"auth_type": "oauth-u2m"}
+        assert kwargs == {
+            "auth_type": "oauth-u2m",
+            "client_id": "custom-client",
+            "redirect_port": 9999,
+            "oauth_scopes": ["custom-scope", "offline_access"],
+        }
 
-    def test_u2m_forwards_client_id_and_redirect_port(self):
+    def test_u2m_custom_client_id_only_falls_back_to_connector_defaults(self):
+        # A custom client_id without explicit scopes/port fills the
+        # remaining two from the connector defaults — matching the Thrift
+        # path, where a custom client_id still uses PYSQL_OAUTH_SCOPES and
+        # the default redirect-port range.
         kwargs = kernel_auth_kwargs(
             _FakeOAuthProvider(),
             {
                 "auth_type": "databricks-oauth",
                 "oauth_client_id": "custom-client",
-                "oauth_redirect_port": 8030,
             },
         )
         assert kwargs == {
             "auth_type": "oauth-u2m",
             "client_id": "custom-client",
-            "redirect_port": 8030,
+            "redirect_port": PYSQL_OAUTH_REDIRECT_PORT_RANGE[0],
+            "oauth_scopes": list(PYSQL_OAUTH_SCOPES),
         }
 
+    def test_u2m_redirect_port_coerced_to_int(self):
+        # oauth_redirect_port may arrive as a string (e.g. from a DSN);
+        # the kernel binding wants an int.
+        kwargs = kernel_auth_kwargs(
+            _FakeOAuthProvider(),
+            {"auth_type": "databricks-oauth", "oauth_redirect_port": "8021"},
+        )
+        assert kwargs["redirect_port"] == 8021
+        assert isinstance(kwargs["redirect_port"], int)
+
     @pytest.mark.parametrize("auth_type", ["databricks-oauth", "azure-oauth"])
-    def test_u2m_forwards_scopes(self, auth_type):
+    def test_u2m_forwards_custom_scopes(self, auth_type):
         kwargs = kernel_auth_kwargs(
             _FakeOAuthProvider(),
             {"auth_type": auth_type, "oauth_scopes": ["all-apis", "offline_access"]},