refactor(kernel): thin azure-oauth — kernel owns Azure resolution

The kernel is the auth core now: for azure-oauth the bridge forwards only
auth_type='azure-oauth' (+ optional client_id/redirect_port passthrough),
and the kernel pins the workspace v2.0 authorize/token endpoints, the
Azure app client id, port 8030, and the user_impersonation scope. Drops
the connector-side endpoint/scope construction (and the
AzureOAuthEndpointCollection / PYSQL_OAUTH_AZURE_* imports) from the
kernel path. Live-verified end-to-end against an Azure workspace.

azure-sp-m2m still routes to oauth-m2m here pending the kernel's
dedicated azure-sp-m2m variant.

Co-authored-by: Isaac
Signed-off-by: eric-wang-1990 <e.wang@databricks.com>

Author: eric-wang-1990

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

@@ -67,15 +67,12 @@
 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.common import get_effective_azure_login_app_id
-from databricks.sql.auth.endpoint import AzureOAuthEndpointCollection
 from databricks.sql.auth.token_federation import TokenFederationProvider
 from databricks.sql.exc import NotSupportedError, ProgrammingError
 
@@ -218,28 +215,21 @@ def kernel_auth_kwargs(
     # creds in azure_* kwargs, not oauth_client_id/secret, so it would
     # otherwise fall through to the final "unsupported" error).
 
-    # azure-oauth (Azure AD U2M): forward the Azure app bundle to oauth-u2m.
-    # The kernel runs the browser flow and discovers endpoints via the
-    # workspace /oidc redirector (which an Azure workspace redirects to Entra).
-    # The AAD delegated scope ({app_id}/user_impersonation [+ offline_access])
-    # is synthesised via AzureOAuthEndpointCollection, which also honors the
-    # DATABRICKS_AZURE_TENANT_ID app-id override. PECOBLR-4120.
+    # azure-oauth (Azure AD U2M): forward the selector; the KERNEL owns Azure
+    # resolution (it is the auth core). The kernel pins the workspace v2.0
+    # authorize/token endpoints (`{host}/oidc/oauth2/v2.0/{authorize,token}` —
+    # NOT the discovered `/oidc/v1/authorize`, which the workspace redirects to
+    # a malformed Entra URL), the Azure app client id, port 8030, and the
+    # `{app_id}/user_impersonation offline_access` scope. So this binding does
+    # NOT construct endpoints/scopes — it just passes `auth_type='azure-oauth'`
+    # plus any optional client_id / redirect_port passthrough. PECOBLR-4120.
     if auth_type == "azure-oauth":
-        redirect_port = opts.get("oauth_redirect_port")
-        caller_scopes = _normalize_scopes(opts.get("oauth_scopes"))
-        mapped_scopes = AzureOAuthEndpointCollection().get_scopes_mapping(
-            caller_scopes if caller_scopes is not None else list(PYSQL_OAUTH_SCOPES)
-        )
-        kwargs = {
-            "auth_type": "oauth-u2m",
-            "client_id": client_id or PYSQL_OAUTH_AZURE_CLIENT_ID,
-            "redirect_ports": (
-                [_coerce_redirect_port(redirect_port)]
-                if client_id and redirect_port is not None
-                else list(PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE)
-            ),
-            "oauth_scopes": mapped_scopes,
-        }
+        kwargs = {"auth_type": "azure-oauth"}
+        if client_id:
+            kwargs["client_id"] = client_id
+            redirect_port = opts.get("oauth_redirect_port")
+            if redirect_port is not None:
+                kwargs["redirect_ports"] = [_coerce_redirect_port(redirect_port)]
         if federation_client_id:
             kwargs["identity_federation_client_id"] = federation_client_id
         return kwargs

tests/unit/test_kernel_auth_bridge.py

@@ -31,11 +31,8 @@
     PYSQL_OAUTH_CLIENT_ID,
     PYSQL_OAUTH_SCOPES,
     PYSQL_OAUTH_REDIRECT_PORT_RANGE,
-    PYSQL_OAUTH_AZURE_CLIENT_ID,
-    PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE,
 )
 from databricks.sql.auth.common import get_effective_azure_login_app_id
-from databricks.sql.auth.endpoint import AzureOAuthEndpointCollection
 from databricks.sql.auth.authenticators import (
     AccessTokenAuthProvider,
     AuthProvider,
@@ -280,30 +277,21 @@ def test_bare_databricks_oauth_forwards_full_python_bundle(self):
             "oauth_scopes": list(PYSQL_OAUTH_SCOPES),
         }
 
-    def test_azure_oauth_routes_to_kernel_u2m(self):
-        # azure-oauth (Azure AD U2M) now routes to the kernel's oauth-u2m with
-        # the Azure app bundle: the Azure client id, its registered port 8030,
-        # and the AAD delegated scope ({app_id}/user_impersonation +
-        # offline_access). The kernel discovers endpoints via the workspace
-        # /oidc redirector (which an Azure workspace redirects to Entra).
-        # PECOBLR-4120.
+    def test_azure_oauth_forwards_selector_kernel_owns_resolution(self):
+        # azure-oauth (Azure AD U2M): the bridge forwards ONLY the selector.
+        # The kernel owns Azure resolution — it pins the workspace v2.0
+        # authorize/token endpoints, the Azure client id, port 8030, and the
+        # {app_id}/user_impersonation scope. So the bridge must NOT construct
+        # client_id / redirect_ports / oauth_scopes here. PECOBLR-4120.
         kwargs = kernel_auth_kwargs(
             _FakeOAuthProvider(),
             {"auth_type": "azure-oauth"},
         )
-        assert kwargs == {
-            "auth_type": "oauth-u2m",
-            "client_id": PYSQL_OAUTH_AZURE_CLIENT_ID,
-            "redirect_ports": list(PYSQL_OAUTH_AZURE_REDIRECT_PORT_RANGE),
-            "oauth_scopes": AzureOAuthEndpointCollection().get_scopes_mapping(
-                list(PYSQL_OAUTH_SCOPES)
-            ),
-        }
-        # Sanity: the mapped scope is the AAD delegated form, not `sql`.
-        assert any(s.endswith("/user_impersonation") for s in kwargs["oauth_scopes"])
-        assert "offline_access" in kwargs["oauth_scopes"]
+        assert kwargs == {"auth_type": "azure-oauth"}
 
     def test_azure_oauth_honors_custom_client_id_and_port(self):
+        # A caller override still passes through (client_id + its coupled port),
+        # but no scopes/endpoints are synthesised by the bridge.
         kwargs = kernel_auth_kwargs(
             _FakeOAuthProvider(),
             {
@@ -312,8 +300,11 @@ def test_azure_oauth_honors_custom_client_id_and_port(self):
                 "oauth_redirect_port": 9100,
             },
         )
-        assert kwargs["client_id"] == "custom-azure-app"
-        assert kwargs["redirect_ports"] == [9100]
+        assert kwargs == {
+            "auth_type": "azure-oauth",
+            "client_id": "custom-azure-app",
+            "redirect_ports": [9100],
+        }
 
     def test_u2m_custom_client_id_port_and_scopes_honored(self):
         # A caller may override the coupled client_id + redirect port and the