Add external scopes to U2M workflow (#1035)

## Description
- We had issues with SCOPES which we fixed in SDK
[[Link](databricks/databricks-sdk-java@11237a2)],
it works for M2M use-cases. But, the current ask is to now resolve this
for U2M too.


## Testing
- Added unit tests

## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

NO_CHANGELOG=true

Author: samikshya-db

src/main/java/com/databricks/jdbc/api/impl/DatabricksConnectionContext.java

@@ -1,6 +1,7 @@
 package com.databricks.jdbc.api.impl;
 
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.*;
+import static com.databricks.jdbc.common.DatabricksJdbcUrlParams.AUTH_SCOPE;
 import static com.databricks.jdbc.common.DatabricksJdbcUrlParams.DEFAULT_STRING_COLUMN_LENGTH;
 import static com.databricks.jdbc.common.EnvironmentVariables.DEFAULT_ROW_LIMIT_PER_BLOCK;
 import static com.databricks.jdbc.common.util.UserAgentManager.USER_AGENT_SEA_CLIENT;
@@ -307,6 +308,9 @@ public String getNullableClientId() {
 
   @Override
   public List<String> getOAuthScopesForU2M() throws DatabricksParsingException {
+    if (getParameter(AUTH_SCOPE) != null) {
+      return Collections.singletonList(getAuthScope());
+    }
     if (getCloud() == Cloud.AWS || getCloud() == Cloud.GCP) {
       return Arrays.asList(
           DatabricksJdbcConstants.SQL_SCOPE, DatabricksJdbcConstants.OFFLINE_ACCESS_SCOPE);

src/main/java/com/databricks/jdbc/api/internal/IDatabricksConnectionContext.java

@@ -53,6 +53,17 @@ public interface IDatabricksConnectionContext {
 
   String getClientSecret();
 
+  /**
+   * Returns the OAuth scopes to request for the user-to-machine (U2M) authorization flow.
+   *
+   * <p>If an explicit auth scope is provided via connection parameters, this returns a singleton
+   * list containing that scope. On AWS and GCP, this returns the SQL scope and offline access
+   * scope. On Azure, this returns {@code null} because the default scope is set by the Databricks
+   * SDK.
+   *
+   * @return a list of OAuth scopes to request, or {@code null} on Azure to use the SDK default
+   * @throws DatabricksParsingException if connection parameters cannot be parsed
+   */
   List<String> getOAuthScopesForU2M() throws DatabricksParsingException;
 
   AuthMech getAuthMech();

src/main/java/com/databricks/jdbc/common/DatabricksJdbcUrlParams.java

@@ -47,7 +47,7 @@ public enum DatabricksJdbcUrlParams {
   USE_JWT_ASSERTION("UseJWTAssertion", "Use JWT assertion", "0"),
   OIDC_DISCOVERY_MODE("EnableOIDCDiscovery", "OIDC discovery mode", "1"),
   DISCOVERY_MODE("OAuthDiscoveryMode", "OAuth discovery mode", "1"), // Same as OIDC_DISCOVERY_MODE
-  AUTH_SCOPE("Auth_Scope", "Authentication scope", "all-apis"),
+  AUTH_SCOPE("Auth_Scope", "Authentication scope"),
   OIDC_DISCOVERY_ENDPOINT("OIDCDiscoveryEndpoint", "OIDC Discovery Endpoint"),
   DISCOVERY_URL("OAuthDiscoveryURL", "OAuth discovery URL"), // Same as OIDC_DISCOVERY_ENDPOINT
   IDENTITY_FEDERATION_CLIENT_ID(

src/test/java/com/databricks/jdbc/TestConstants.java

@@ -38,6 +38,7 @@ public class TestConstants {
   public static final StatementId TEST_STATEMENT_ID = new StatementId("statement_id");
   public static final String UC_VOLUME_CATALOG = "uc_volume_test_catalog";
   public static final String UC_VOLUME_SCHEMA = "uc_volume_test_schema";
+  public static final String TEST_SCOPE_STRING = "URI:RS:000001:000999:DatabricksOAuth:TEST";
 
   public static final String DEFAULT_BENCHFOOD_HOST = "e2-benchfood.cloud.databricks.com";
   public static final String DEFAULT_BENCHFOOD_HTTP_PATH = "/sql/1.0/warehouses/6e681b20741e4674";
@@ -202,7 +203,7 @@ public class TestConstants {
   public static final String VALID_URL_3 =
       "jdbc:databricks://sample-host.cloud.databricks.com:9999/default;transportMode=http;"
           + "ssl=0;AuthMech=3;httpPath=/sql/1.0/warehouses/9999999999999999;EnableQueryResultLZ4Compression=0;"
-          + "UseThriftClient=1;LogLevel=1234";
+          + "UseThriftClient=1;LogLevel=1234;Auth_scope=URI:RS:000001:000999:DatabricksOAuth:TEST";
 
   public static final String VALID_URL_4 =
       "jdbc:databricks://sample-host.cloud.databricks.com:9999/default;AuthMech=3;"

src/test/java/com/databricks/jdbc/api/impl/DatabricksConnectionContextTest.java

@@ -1,5 +1,6 @@
 package com.databricks.jdbc.api.impl;
 
+import static com.databricks.jdbc.TestConstants.TEST_SCOPE_STRING;
 import static com.databricks.jdbc.api.impl.DatabricksConnectionContext.buildPropertiesMap;
 import static com.databricks.jdbc.api.impl.DatabricksConnectionContext.getLogLevel;
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.GCP_GOOGLE_CREDENTIALS_AUTH_TYPE;
@@ -16,6 +17,7 @@
 import com.databricks.jdbc.exception.DatabricksVendorCode;
 import com.databricks.sdk.core.ProxyConfig;
 import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 import org.junit.jupiter.api.BeforeAll;
@@ -109,9 +111,10 @@ public void testParseValid() throws DatabricksSQLException {
     assertEquals(AuthFlow.TOKEN_PASSTHROUGH, connectionContext.getAuthFlow());
     assertEquals(AuthMech.PAT, connectionContext.getAuthMech());
     assertEquals(CompressionCodec.NONE, connectionContext.getCompressionCodec());
-    assertEquals(8, connectionContext.parameters.size());
+    assertEquals(9, connectionContext.parameters.size());
     assertEquals(LogLevel.OFF, connectionContext.getLogLevel());
-    assertEquals(connectionContext.getOAuthScopesForU2M(), expected_scopes);
+    assertEquals(
+        connectionContext.getOAuthScopesForU2M(), Collections.singletonList(TEST_SCOPE_STRING));
     assertFalse(connectionContext.isAllPurposeCluster());
     assertEquals(DatabricksClientType.THRIFT, connectionContext.getClientType());
 

src/test/java/com/databricks/jdbc/dbclient/impl/common/ClientConfiguratorTest.java

@@ -1,6 +1,7 @@
 package com.databricks.jdbc.dbclient.impl.common;
 
 import static com.databricks.jdbc.TestConstants.TEST_DISCOVERY_URL;
+import static com.databricks.jdbc.TestConstants.TEST_SCOPE_STRING;
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.GCP_GOOGLE_CREDENTIALS_AUTH_TYPE;
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.M2M_AZURE_CLIENT_SECRET_AUTH_TYPE;
 import static org.junit.jupiter.api.Assertions.*;
@@ -26,6 +27,7 @@
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 import org.junit.jupiter.api.Test;
@@ -207,9 +209,10 @@ void getWorkspaceClient_OAuthWithBrowserBasedAuthentication_AuthenticatesCorrect
     when(mockContext.getHostForOAuth()).thenReturn("https://oauth-browser.databricks.com");
     when(mockContext.getClientId()).thenReturn("browser-client-id");
     when(mockContext.getClientSecret()).thenReturn("browser-client-secret");
-    when(mockContext.getOAuthScopesForU2M()).thenReturn(List.of("scope1", "scope2"));
     when(mockContext.isOAuthDiscoveryModeEnabled()).thenReturn(true);
     when(mockContext.getOAuthDiscoveryURL()).thenReturn(TEST_DISCOVERY_URL);
+    when(mockContext.getOAuthScopesForU2M())
+        .thenReturn(Collections.singletonList(TEST_SCOPE_STRING));
     when(mockContext.getHttpConnectionPoolSize()).thenReturn(100);
     when(mockContext.getOAuth2RedirectUrlPorts()).thenReturn(List.of(8020));
     when(mockContext.getHttpMaxConnectionsPerRoute()).thenReturn(100);
@@ -222,7 +225,7 @@ void getWorkspaceClient_OAuthWithBrowserBasedAuthentication_AuthenticatesCorrect
     assertEquals("https://oauth-browser.databricks.com", config.getHost());
     assertEquals("browser-client-id", config.getClientId());
     assertEquals("browser-client-secret", config.getClientSecret());
-    assertEquals(List.of("scope1", "scope2"), config.getScopes());
+    assertEquals(List.of(TEST_SCOPE_STRING), config.getScopes());
     assertEquals("http://localhost:8020", config.getOAuthRedirectUrl());
     assertEquals(DatabricksJdbcConstants.U2M_AUTH_TYPE, config.getAuthType());
   }