ai: apply changes for #912 (2 review threads)

Addresses:
  - #3798853448 at src/databricks/sql/auth/auth.py:141
  - #3798853453 at src/databricks/sql/auth/auth.py:128

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>

Author: peco-engineer-bot[bot]

src/databricks/sql/auth/auth.py

@@ -125,9 +125,10 @@ def get_python_sql_connector_auth_provider(hostname: str, http_client, **kwargs)
     oauth_redirect_port = kwargs.get("oauth_redirect_port")
     oauth_scopes = kwargs.get("oauth_scopes")
 
-    scopes = oauth_scopes or PYSQL_OAUTH_SCOPES
     if caller_client_id:
         client_id = caller_client_id
+        # Caller owns the bundle: forward their scopes verbatim.
+        scopes = oauth_scopes or PYSQL_OAUTH_SCOPES
         # Foreign client_id with no explicit port falls through to the base
         # (app-neutral) default, NOT the driver's app-specific range.
         redirect_port_range = (
@@ -137,9 +138,15 @@ def get_python_sql_connector_auth_provider(hostname: str, http_client, **kwargs)
         )
     else:
         client_id = default_client_id
-        redirect_port_range = (
-            [oauth_redirect_port] if oauth_redirect_port else default_redirect_port_range
-        )
+        # The caller relies on the driver's default client_id, whose OAuth app
+        # only has the app-specific default scopes/range registered. A caller-
+        # supplied oauth_scopes must NOT override the driver's app-specific
+        # default scopes here, just as a caller-supplied oauth_redirect_port must
+        # NOT override the default range (doing so risks a redirect_uri_mismatch
+        # or scope error, PECOBLR-4039); both only take effect alongside a
+        # caller-supplied oauth_client_id.
+        scopes = PYSQL_OAUTH_SCOPES
+        redirect_port_range = default_redirect_port_range
 
     cfg = ClientContext(
         hostname=normalize_host_name(hostname),

tests/unit/test_auth.py

@@ -12,6 +12,8 @@
 from databricks.sql.auth.auth import (
     get_python_sql_connector_auth_provider,
     PYSQL_OAUTH_CLIENT_ID,
+    PYSQL_OAUTH_REDIRECT_PORT_RANGE,
+    PYSQL_OAUTH_SCOPES,
 )
 from databricks.sql.auth.oauth import OAuthManager, Token, ClientCredentialsTokenSource
 from databricks.sql.auth.authenticators import (
@@ -251,6 +253,52 @@ def test_get_python_sql_connector_u2m_foreign_client_id_no_port(
         self.assertEqual(provider._client_id, "test-custom-u2m-app")
         self.assertEqual(provider.oauth_manager.port_range, [8030])
 
+    @patch.object(DatabricksOAuthProvider, "_initial_get_token")
+    def test_get_python_sql_connector_u2m_default_client_id_ignores_redirect_port(
+        self, mock__initial_get_token
+    ):
+        # AUTH-013 / PECOBLR-4039: a caller relying on the driver's DEFAULT
+        # client_id must stay pinned to the driver's registered redirect port
+        # range even if oauth_redirect_port is supplied - the driver's OAuth app
+        # only registers 8020-8024, so honoring an arbitrary port would produce a
+        # redirect_uri_mismatch. oauth_redirect_port only applies with a
+        # caller-supplied oauth_client_id.
+        hostname = "foo.cloud.databricks.com"
+        kwargs = {"oauth_redirect_port": 9000}
+        mock_http_client = MagicMock()
+        auth_provider = get_python_sql_connector_auth_provider(
+            hostname, mock_http_client, **kwargs
+        )
+
+        provider = auth_provider.external_provider
+        self.assertEqual(type(provider).__name__, "DatabricksOAuthProvider")
+        self.assertEqual(provider._client_id, PYSQL_OAUTH_CLIENT_ID)
+        self.assertEqual(
+            provider.oauth_manager.port_range, PYSQL_OAUTH_REDIRECT_PORT_RANGE
+        )
+
+    @patch.object(DatabricksOAuthProvider, "_initial_get_token")
+    def test_get_python_sql_connector_u2m_default_client_id_ignores_scopes(
+        self, mock__initial_get_token
+    ):
+        # AUTH-013 / PECOBLR-4039: a caller relying on the driver's DEFAULT
+        # client_id must stay pinned to the driver's app-specific default scopes
+        # even if oauth_scopes is supplied - the driver's OAuth app only has its
+        # own default scopes registered, so honoring arbitrary scopes risks a
+        # scope error. oauth_scopes only applies with a caller-supplied
+        # oauth_client_id.
+        hostname = "foo.cloud.databricks.com"
+        kwargs = {"oauth_scopes": ["all-apis"]}
+        mock_http_client = MagicMock()
+        auth_provider = get_python_sql_connector_auth_provider(
+            hostname, mock_http_client, **kwargs
+        )
+
+        provider = auth_provider.external_provider
+        self.assertEqual(type(provider).__name__, "DatabricksOAuthProvider")
+        self.assertEqual(provider._client_id, PYSQL_OAUTH_CLIENT_ID)
+        self.assertEqual(provider._scopes_as_str, " ".join(PYSQL_OAUTH_SCOPES))
+
 
 class TestClientCredentialsTokenSource:
     @pytest.fixture