Add auth type for Azure storage

.gitignore

@@ -61,6 +61,7 @@ gradle/wrapper/gradle-wrapper-*.sha256
 
 # jenv
 .java-version
+.sdkmanrc
 
 # Log files
 *.log
@@ -78,3 +79,7 @@ hs_err_pid*
 
 # macOS
 *.DS_Store
+
+# python
+polaris-venv
+pyproject.toml

docs/index.html

@@ -1023,7 +1023,7 @@ <h3 id="connecting-with-spark">Connecting with Spark</h3>
 &lt;p&gt;A service connection represents a REST-compatible engine (such as Apache Spark&amp;trade;, Apache Flink&amp;reg;, or Trino) that can read from and write to Polaris
 Catalog. When creating a new service connection, the Polaris administrator grants the service principal that is created with the new service 
 connection either a new or existing principal role. A principal role is a resource in Polaris that you can use to logically group Polaris 
-service principals together and grant privileges on securable objects. For more information, see &lt;a href=&quot;access-control.md#principal-role&quot; title=&quot;Principal role&quot;&gt;Principal role&lt;/a&gt;. Polaris uses a role-based access control (RBAC) model to grant service principals access to resources. For more information, 
+service principals together and grant privileges on securable objects. For more information, see &lt;a href=&quot;access-control.md#principal-role&quot; title=&quot;Principal role&quot;&gt;Principal role&lt;/a&gt;. Polaris uses a role-based access control (RBAC) model to grant service principals access to resources. For more information,
 see &lt;a href=&quot;access-control.md&quot; title=&quot;Access control&quot;&gt;Access control&lt;/a&gt;. For a diagram of this model, see &lt;a href=&quot;access-control.md#rbac-model&quot; title=&quot;RBAC model&quot;&gt;RBAC model&lt;/a&gt;.&lt;/p&gt;
 &lt;p&gt;If the Polaris administrator grants the service principal for the new service connection a new principal role, the service principal
 doesn&amp;#39;t have any privileges granted to it yet. When securing the catalog that the new service connection will connect to, the Polaris
@@ -1154,7 +1154,7 @@ <h3 id="service-connection">Service connection</h3>
 <p>A service connection represents a REST-compatible engine (such as Apache Spark&trade;, Apache Flink&reg;, or Trino) that can read from and write to Polaris
 Catalog. When creating a new service connection, the Polaris administrator grants the service principal that is created with the new service 
 connection either a new or existing principal role. A principal role is a resource in Polaris that you can use to logically group Polaris 
-service principals together and grant privileges on securable objects. For more information, see <a href="access-control.md#principal-role" title="Principal role">Principal role</a>. Polaris uses a role-based access control (RBAC) model to grant service principals access to resources. For more information, 
+service principals together and grant privileges on securable objects. For more information, see <a href="access-control.md#principal-role" title="Principal role">Principal role</a>. Polaris uses a role-based access control (RBAC) model to grant service principals access to resources. For more information,
 see <a href="access-control.md" title="Access control">Access control</a>. For a diagram of this model, see <a href="access-control.md#rbac-model" title="RBAC model">RBAC model</a>.</p>
 <p>If the Polaris administrator grants the service principal for the new service connection a new principal role, the service principal
 doesn&#39;t have any privileges granted to it yet. When securing the catalog that the new service connection will connect to, the Polaris

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -149,6 +149,7 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setConsentUrl(azureConfig.getConsentUrl())
             .setStorageType(AZURE)
             .setAllowedLocations(azureConfig.getAllowedLocations())
+            .setAuthType(azureConfig.getAuthType())
             .build();
       }
       if (configInfo instanceof GcpStorageConfigurationInfo) {
@@ -252,7 +253,9 @@ public Builder setStorageConfigurationInfo(
             AzureStorageConfigInfo azureConfigModel = (AzureStorageConfigInfo) storageConfigModel;
             config =
                 new AzureStorageConfigurationInfo(
-                    new ArrayList<>(allowedLocations), azureConfigModel.getTenantId());
+                    new ArrayList<>(allowedLocations),
+                    azureConfigModel.getTenantId(),
+                    azureConfigModel.getAuthType());
             break;
           case GCS:
             config = new GcpStorageConfigurationInfo(new ArrayList<>(allowedLocations));

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureCredentialsStorageIntegration.java

@@ -92,6 +92,46 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
     String storageDnsName = location.getStorageAccount() + "." + location.getEndpoint();
     String filePath = location.getFilePath();
 
+    AzureStorageConfigInfo.AuthTypeEnum authType = storageConfig.getAuthType();
+
+    if (authType == null) {
+      authType = AzureStorageConfigInfo.AuthTypeEnum.NONE;
+    }
+
+    String sasToken = "";
+
+    switch (authType) {
+      case NONE:
+      case SAS_TOKEN:
+        sasToken =
+            getSasToken(
+                storageConfig,
+                allowListOperation,
+                allowedReadLocations,
+                allowedWriteLocations,
+                location,
+                filePath,
+                storageDnsName,
+                loc);
+        break;
+
+      case APPLICATION_DEFAULT:
+        break;
+    }
+    credentialMap.put(PolarisCredentialProperty.AZURE_SAS_TOKEN, sasToken);
+    credentialMap.put(PolarisCredentialProperty.AZURE_ACCOUNT_HOST, storageDnsName);
+    return credentialMap;
+  }
+
+  private String getSasToken(
+      @NotNull AzureStorageConfigurationInfo storageConfig,
+      boolean allowListOperation,
+      @NotNull Set<String> allowedReadLocations,
+      @NotNull Set<String> allowedWriteLocations,
+      AzureLocation location,
+      String filePath,
+      String storageDnsName,
+      String loc) {
     BlobSasPermission blobSasPermission = new BlobSasPermission();
     // pathSasPermission is for Data lake storage
     PathSasPermission pathSasPermission = new PathSasPermission();
@@ -163,9 +203,7 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
       throw new RuntimeException(
           String.format("Endpoint %s not supported", location.getEndpoint()));
     }
-    credentialMap.put(PolarisCredentialProperty.AZURE_SAS_TOKEN, sasToken);
-    credentialMap.put(PolarisCredentialProperty.AZURE_ACCOUNT_HOST, storageDnsName);
-    return credentialMap;
+    return sasToken;
   }
 
   private String getBlobUserDelegationSas(

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureStorageConfigurationInfo.java

@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
+import io.polaris.core.admin.model.AzureStorageConfigInfo;
+import io.polaris.core.storage.PolarisStorageConfigurationInfo;
 import java.util.List;
 import java.util.Objects;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -45,13 +47,18 @@ public class AzureStorageConfigurationInfo extends PolarisStorageConfigurationIn
   @JsonProperty(value = "consentUrl", required = false)
   private @Nullable String consentUrl = null;
 
+  private final @Nullable AzureStorageConfigInfo.AuthTypeEnum authType;
+
   @JsonCreator
   public AzureStorageConfigurationInfo(
       @JsonProperty(value = "allowedLocations", required = true) @NotNull
           List<String> allowedLocations,
-      @JsonProperty(value = "tenantId", required = true) @NotNull String tenantId) {
+      @JsonProperty(value = "tenantId", required = true) @NotNull String tenantId,
+      @JsonProperty(value = "authType", required = false) @Nullable
+          AzureStorageConfigInfo.AuthTypeEnum authType) {
     super(StorageType.AZURE, allowedLocations);
     this.tenantId = tenantId;
+    this.authType = authType;
     validateMaxAllowedLocations(MAX_ALLOWED_LOCATIONS);
   }
 
@@ -80,6 +87,10 @@ public void setConsentUrl(String consentUrl) {
     this.consentUrl = consentUrl;
   }
 
+  public AzureStorageConfigInfo.AuthTypeEnum getAuthType() {
+    return authType;
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
@@ -88,6 +99,7 @@ public String toString() {
         .add("allowedLocation", getAllowedLocations())
         .add("multiTenantAppName", multiTenantAppName)
         .add("consentUrl", consentUrl)
+        .add("authType", authType)
         .toString();
   }
 

polaris-core/src/test/java/org/apache/polaris/service/storage/azure/AzureCredentialStorageIntegrationTest.java

@@ -341,7 +341,7 @@ private Map<PolarisCredentialProperty, String> subscopedCredsForOperations(
     allowedLoc.addAll(allowedReadLoc);
     allowedLoc.addAll(allowedWriteLoc);
     AzureStorageConfigurationInfo azureConfig =
-        new AzureStorageConfigurationInfo(allowedLoc, tenantId);
+        new AzureStorageConfigurationInfo(allowedLoc, tenantId, null);
     AzureCredentialsStorageIntegration azureCredsIntegration =
         new AzureCredentialsStorageIntegration();
     EnumMap<PolarisCredentialProperty, String> credsMap =

spec/index.yaml

@@ -3911,6 +3911,13 @@ components:
         consentUrl:
           type: string
           description: URL to the Azure permissions request page
+        authType:
+          type: string
+          enum:
+            - NONE
+            - SAS_TOKEN
+            - APPLICATION_DEFAULT
+          description: the type of authentication to use
       required:
         - tenantId
     Polaris_Management_Service_GcpStorageConfigInfo:

spec/polaris-management-service.yml

@@ -917,6 +917,13 @@ components:
         consentUrl:
           type: string
           description: URL to the Azure permissions request page
+        authType:
+          type: string
+          enum:
+            - NONE
+            - SAS_TOKEN
+            - APPLICATION_DEFAULT
+          description: the type of authentication to use
       required:
         - tenantId