refactor(backend/kernel): PAT-only auth, drop External trampoline

The earlier auth_bridge routed OAuth/MSAL/federation through the
kernel's External token-provider trampoline (a Python callable
the kernel invoked per HTTP request). Removing that for now.

Why: routing OAuth into the kernel inherently requires per-request
token resolution to keep refresh working during a long-running
session. Two viable mechanisms (kernel-native OAuth, or the
External callback); both have costs (duplicate OAuth flows vs
GIL-per-request). Punting the decision until there's actual
demand on use_sea=True.

Today: the bridge accepts PAT (including TokenFederationProvider-
wrapped PAT, which is how `get_python_sql_connector_auth_provider`
always shapes it). Any non-PAT auth_provider raises a clear
NotSupportedError pointing the user at use_sea=False (Thrift).

This shrinks the auth_bridge to ~50 lines and means the kernel-
side External enablement PR is no longer on the connector's
critical path — there's no kernel-side prerequisite for shipping
use_sea=True for PAT users.

Unit tests updated:
- TokenFederationProvider-wrapped PAT still routes to PAT (kept).
- Generic OAuth provider raises NotSupportedError (new).
- ExternalAuthProvider raises NotSupportedError (new).
- Silent non-PAT provider raises NotSupportedError (new) —
  reject the type itself rather than trying to extract a token
  we already know we can't use.

Live e2e against dogfood with use_sea=True (PAT): all checks
still pass (SELECT 1, range(10000), fetchmany pacing, four
metadata calls, session_configuration round-trip, structured
DatabaseError on bad SQL).

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

Author: vikrantpuppala

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

@@ -1,25 +1,19 @@
 """Translate the connector's ``AuthProvider`` into ``databricks_sql_kernel``
 ``Session`` auth kwargs.
 
-The connector already implements every auth flow it supports (PAT,
-OAuth M2M, OAuth U2M, external token providers, federation). The
-kernel must not re-implement them. Decision D9 in the integration
-design: PAT goes through the kernel's PAT path; everything else
-delegates back to the connector via the kernel's ``External``
-trampoline, with a Python callback that returns a fresh bearer
-token.
+This phase ships PAT only. The kernel-side PyO3 binding accepts
+``auth_type='pat'``; OAuth / federation / custom credentials
+providers are reserved but not yet wired in either layer. Non-PAT
+auth raises ``NotSupportedError`` from this bridge so the failure
+surfaces at session-open time with a clear message rather than
+deep inside the kernel.
 
 Token extraction goes through ``AuthProvider.add_headers({})``
 rather than touching auth-provider-specific attributes, so the
-bridge works for every subclass — including custom providers a
-caller may have wired in.
-
-End-to-end limitation: the kernel's
-``build_auth_provider`` currently rejects ``AuthConfig::External``
-("reserved; v0 wires PAT + OAuthM2M + OAuthU2M only"). Until the
-kernel-side follow-up PR lands, non-PAT auth surfaces a clear
-``KernelError(code='InvalidArgument', message='AuthConfig::External
-is reserved...')`` from ``Session.open_session``. PAT works today.
+bridge works uniformly for every PAT shape — including
+``AccessTokenAuthProvider`` wrapped in ``TokenFederationProvider``
+(which ``get_python_sql_connector_auth_provider`` does for every
+provider it builds).
 """
 
 from __future__ import annotations
@@ -29,6 +23,7 @@
 
 from databricks.sql.auth.authenticators import AccessTokenAuthProvider, AuthProvider
 from databricks.sql.auth.token_federation import TokenFederationProvider
+from databricks.sql.exc import NotSupportedError
 
 logger = logging.getLogger(__name__)
 
@@ -64,8 +59,8 @@ def _extract_bearer_token(auth_provider: AuthProvider) -> Optional[str]:
     provider-specific internals.
 
     Returns ``None`` if the provider did not write an Authorization
-    header or wrote a non-Bearer scheme — neither shape is
-    representable in the kernel's auth surface today.
+    header or wrote a non-Bearer scheme — neither is representable
+    in the kernel's PAT auth surface.
     """
     headers: Dict[str, str] = {}
     auth_provider.add_headers(headers)
@@ -80,29 +75,13 @@ def _extract_bearer_token(auth_provider: AuthProvider) -> Optional[str]:
 def kernel_auth_kwargs(auth_provider: AuthProvider) -> Dict[str, Any]:
     """Build the kwargs passed to ``databricks_sql_kernel.Session(...)``.
 
-    Two routing decisions:
-
-    1. ``AccessTokenAuthProvider`` → ``auth_type='pat'`` with the
-       static token. Kernel uses it verbatim for every request.
-    2. Anything else → ``auth_type='external'`` with a callback that
-       calls ``auth_provider.add_headers({})`` and returns the
-       fresh bearer token. The connector keeps owning the OAuth /
-       MSAL / federation flow; the kernel asks for a token whenever
-       it needs one.
-
-    The PAT special-case exists because it's the only path the
-    kernel actually serves end-to-end today. Once the kernel-side
-    External enablement lands, PAT could collapse into the
-    External path too (one callback that returns the static token);
-    but keeping the explicit ``pat`` route means the kernel does
-    not pay the GIL-reacquire cost on every HTTP request for PAT
-    users.
+    PAT (including ``TokenFederationProvider``-wrapped PAT) routes
+    through the kernel's PAT path. Anything else raises
+    ``NotSupportedError`` — the kernel binding doesn't accept OAuth
+    today, and routing OAuth through PAT would silently break
+    token refresh during long-running sessions.
     """
     if _is_pat(auth_provider):
-        # PAT case: pull the static token out and feed the kernel's
-        # PAT path. We go through ``add_headers`` regardless of
-        # whether the provider was wrapped in TokenFederation or
-        # not — both shapes write the same Authorization header.
         token = _extract_bearer_token(auth_provider)
         if not token:
             raise ValueError(
@@ -111,21 +90,8 @@ def kernel_auth_kwargs(auth_provider: AuthProvider) -> Dict[str, Any]:
             )
         return {"auth_type": "pat", "access_token": token}
 
-    # Every other provider: trampoline a callback. The callback is
-    # invoked once per HTTP request that needs auth (the kernel does
-    # not cache the returned token), so the auth_provider's own
-    # caching is what keeps this fast.
-    def token_callback() -> str:
-        token = _extract_bearer_token(auth_provider)
-        if not token:
-            raise RuntimeError(
-                f"{type(auth_provider).__name__}.add_headers did not produce "
-                "a Bearer Authorization header; cannot supply a token to the kernel"
-            )
-        return token
-
-    logger.debug(
-        "Routing %s through kernel External trampoline",
-        type(auth_provider).__name__,
+    raise NotSupportedError(
+        f"The kernel backend (use_sea=True) currently only supports PAT auth, "
+        f"but got {type(auth_provider).__name__}. Use use_sea=False (Thrift) "
+        "for OAuth / federation / custom credential providers."
     )
-    return {"auth_type": "external", "token_callback": token_callback}

tests/unit/test_kernel_auth_bridge.py

@@ -1,9 +1,12 @@
 """Unit tests for the kernel backend's auth bridge.
 
-The bridge translates the connector's ``AuthProvider`` hierarchy
-into ``databricks_sql_kernel.Session`` auth kwargs. PAT goes through
-the kernel's PAT path; everything else trampolines through the
-``External`` path with a Python callback.
+Phase 1 ships PAT only. Tests verify:
+  - PAT routes through ``auth_type='pat'``.
+  - ``TokenFederationProvider``-wrapped PAT also routes through
+    PAT (every provider built by ``get_python_sql_connector_auth_provider``
+    is federation-wrapped, so the naive isinstance check has to
+    look through the wrapper).
+  - Anything else raises ``NotSupportedError`` with a clear message.
 """
 
 from __future__ import annotations
@@ -12,38 +15,36 @@
 
 import pytest
 
-from databricks.sql.auth.authenticators import AccessTokenAuthProvider, AuthProvider
+from databricks.sql.auth.authenticators import (
+    AccessTokenAuthProvider,
+    AuthProvider,
+    DatabricksOAuthProvider,
+    ExternalAuthProvider,
+)
 from databricks.sql.backend.kernel.auth_bridge import (
     _extract_bearer_token,
     kernel_auth_kwargs,
 )
+from databricks.sql.exc import NotSupportedError
 
 
 class _FakeOAuthProvider(AuthProvider):
-    """Stand-in for OAuth/MSAL/federation providers — anything that
-    isn't ``AccessTokenAuthProvider``. Returns a counter-stamped
-    token so tests can prove the callback is invoked each call."""
-
-    def __init__(self):
-        self.calls = 0
+    """Stand-in for any non-PAT provider. The bridge should reject
+    these with NotSupportedError."""
 
     def add_headers(self, request_headers):
-        self.calls += 1
-        request_headers["Authorization"] = f"Bearer token-{self.calls}"
+        request_headers["Authorization"] = "Bearer oauth-token-xyz"
 
 
 class _MalformedProvider(AuthProvider):
-    """Provider that returns a non-Bearer Authorization header
-    (e.g. Basic auth). The bridge should reject this rather than
-    silently sending the wrong shape to the kernel."""
+    """Provider that returns a non-Bearer Authorization header."""
 
     def add_headers(self, request_headers):
         request_headers["Authorization"] = "Basic dXNlcjpwYXNz"
 
 
 class _SilentProvider(AuthProvider):
-    """Provider that writes nothing — represents misconfigured
-    auth or a placeholder. The bridge must surface this clearly."""
+    """Provider that writes nothing — misconfigured auth."""
 
     def add_headers(self, request_headers):
         pass
@@ -74,43 +75,53 @@ def test_federation_wrapped_pat_routes_to_kernel_pat(self):
         underlying ``AccessTokenAuthProvider``."""
         from databricks.sql.auth.token_federation import TokenFederationProvider
 
-        # TokenFederationProvider needs an http_client; a MagicMock
-        # is sufficient because we don't trigger any token exchange
-        # in the test (the cached-token path is never hit).
         base = AccessTokenAuthProvider("dapi-abc")
+        # TokenFederationProvider's __init__ requires an http_client
+        # to construct cleanly; for this unit test we only exercise
+        # the add_headers passthrough + the external_provider
+        # attribute. Bypass __init__ with __new__ and stash just
+        # the fields the bridge touches.
         federated = TokenFederationProvider.__new__(TokenFederationProvider)
         federated.external_provider = base
-        # The bridge only touches `add_headers` (delegated to the
-        # base) and `external_provider`. Other attrs would be set
-        # by __init__ but aren't exercised here.
         federated.add_headers = base.add_headers
         kwargs = kernel_auth_kwargs(federated)
         assert kwargs == {"auth_type": "pat", "access_token": "dapi-abc"}
 
-    def test_pat_with_silent_provider_raises(self):
+    def test_pat_with_silent_provider_raises_value_error(self):
         """An AccessTokenAuthProvider that produces no Authorization
         header is misconfigured; surface that at bridge-build time,
         not on the first kernel HTTP request."""
         broken = AccessTokenAuthProvider("dapi-x")
-        # Force the broken state by monkey-patching add_headers.
         broken.add_headers = lambda h: None  # type: ignore[method-assign]
         with pytest.raises(ValueError, match="Bearer"):
             kernel_auth_kwargs(broken)
 
-    def test_oauth_routes_to_external_trampoline(self):
-        provider = _FakeOAuthProvider()
-        kwargs = kernel_auth_kwargs(provider)
-        assert kwargs["auth_type"] == "external"
-        callback = kwargs["token_callback"]
-        assert callable(callback)
-        # First call -> token-1, second call -> token-2. Proves the
-        # callback delegates to the live auth_provider each time
-        # rather than caching.
-        assert callback() == "token-1"
-        assert callback() == "token-2"
-        assert provider.calls == 2
-
-    def test_external_callback_raises_on_missing_header(self):
-        kwargs = kernel_auth_kwargs(_SilentProvider())
-        with pytest.raises(RuntimeError, match="Bearer"):
-            kwargs["token_callback"]()
+    def test_generic_oauth_provider_raises_not_supported(self):
+        with pytest.raises(NotSupportedError, match="only supports PAT"):
+            kernel_auth_kwargs(_FakeOAuthProvider())
+
+    def test_external_credentials_provider_raises_not_supported(self):
+        """``ExternalAuthProvider`` wraps user-supplied
+        credentials_provider — kernel doesn't accept these today,
+        and the bridge surfaces that explicitly."""
+        # ExternalAuthProvider's __init__ calls the credentials
+        # provider; supply a noop one.
+        from databricks.sql.auth.authenticators import CredentialsProvider
+
+        class _NoopCreds(CredentialsProvider):
+            def auth_type(self):
+                return "noop"
+
+            def __call__(self, *args, **kwargs):
+                return lambda: {"Authorization": "Bearer noop"}
+
+        ext = ExternalAuthProvider(_NoopCreds())
+        with pytest.raises(NotSupportedError, match="only supports PAT"):
+            kernel_auth_kwargs(ext)
+
+    def test_silent_non_pat_provider_also_raises_not_supported(self):
+        """Even if a non-PAT provider produces no header, the bridge
+        rejects the type itself — we don't try to extract a token
+        from something we already know is unsupported."""
+        with pytest.raises(NotSupportedError):
+            kernel_auth_kwargs(_SilentProvider())