Delta Sharing protocol: generate temporary table credentials

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java

@@ -13,8 +13,12 @@
 import io.opensharing.catalog.AssetPage;
 import io.opensharing.catalog.AssetType;
 import io.opensharing.catalog.CatalogConnector;
+import io.opensharing.catalog.CredentialRequest;
 import io.opensharing.catalog.ResolvedAsset;
 import io.opensharing.catalog.TableProperties;
+import io.opensharing.catalog.StorageCredentials;
+import io.opensharing.catalog.StorageOperation;
+import io.opensharing.exception.CatalogException;
 import io.opensharing.http.ApiException;
 import io.opensharing.http.ListResponse;
 import io.opensharing.http.Pagination;
@@ -124,6 +128,43 @@ public DeltaTableMetadataReader.Result metadata(
         resolve(granted, schema, table), version, asOf, owner(granted), capabilities);
   }
 
+  /**
+   * Read credentials for one table in the share that supports directory access, scoped to its root
+   * or to the auxiliary location {@code request} names. The catalog vends them to the share owner.
+   */
+  public TemporaryTableCredentialsResponse temporaryCredentials(
+      RecipientPrincipal recipient,
+      String share,
+      String schema,
+      String table,
+      TemporaryTableCredentialsRequest request) {
+    ShareEntity granted = shares.requireGranted(recipient, share);
+    ResolvedAsset resolved = resolve(granted, schema, table);
+    List<String> accessModes = TableAccessModes.forTable(resolved);
+    if (accessModes == null || !accessModes.contains("dir")) {
+      throw ApiException.invalidParameter(
+          "table '" + schema + "." + table + "' does not support directory access");
+    }
+    String location = credentialLocation(resolved, request == null ? null : request.location());
+    CredentialRequest credentials =
+        new CredentialRequest(
+            AssetType.TABLE,
+            resolved.fullName(),
+            resolved.catalogAssetId(),
+            location,
+            StorageOperation.READ,
+            null);
+    StorageCredentials minted =
+        catalog.getStorageCredentials(credentials, owner(granted)).stream()
+            .filter(candidate -> candidate.covers(location))
+            .findFirst()
+            .orElseThrow(
+                () ->
+                    new CatalogException(
+                        "catalog returned no credentials for location '" + location + "'"));
+    return new TemporaryTableCredentialsResponse(TemporaryCredentials.from(minted));
+  }
+
   private List<SharedDataObjectEntity> inSchema(ShareEntity share, String schema) {
     List<SharedDataObjectEntity> included =
         objects.listInSchema(share, ObjectNames.normalize(schema));
@@ -155,6 +196,24 @@ private ResolvedAsset resolve(ShareEntity share, String schema, String table) {
     return find(share, schema, table).resolved().get();
   }
 
+  private static String credentialLocation(ResolvedAsset table, String requested) {
+    AssetLocation location = table.location();
+    String root = location == null ? null : location.storageLocation();
+    if (requested == null || requested.isBlank()) {
+      if (root == null || root.isBlank()) {
+        throw ApiException.invalidParameter(
+            "table '" + table.fullName() + "' has no storage location");
+      }
+      return root;
+    }
+    if (requested.equals(root)
+        || (location != null && location.auxiliaryLocations().contains(requested))) {
+      return requested;
+    }
+    throw ApiException.invalidParameter(
+        "location '" + requested + "' is not part of table '" + table.fullName() + "'");
+  }
+
   private static Instant parseTimestamp(String name, String value) {
     if (value == null || value.isBlank()) {
       return null;

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryCredentials.java

@@ -0,0 +1,66 @@
+package io.opensharing.asset.table;
+
+import io.opensharing.catalog.StorageCredentials;
+
+/**
+ * Directory-scoped cloud credentials for tables whose {@code accessModes} include {@code dir}. Only
+ * the field for the storage's cloud is set:
+ *
+ * <pre>{@code
+ * {"location": "<storage location>",
+ *  "awsTempCredentials": {"accessKeyId": "<key id>", "secretAccessKey": "<secret>",
+ *                         "sessionToken": "<session token>"},
+ *  "azureUserDelegationSas": {"sasToken": "<SAS token>"},
+ *  "gcpOauthToken": {"oauthToken": "<OAuth token>"},
+ *  "r2Credentials": {"accessKeyId": "<key id>", "secretAccessKey": "<secret>",
+ *                    "sessionToken": "<session token>"},
+ *  "expirationTime": <epoch milliseconds>}
+ * }</pre>
+ */
+public record TemporaryCredentials(
+    String location,
+    KeyCredentials awsTempCredentials,
+    AzureUserDelegationSas azureUserDelegationSas,
+    GcpOauthToken gcpOauthToken,
+    KeyCredentials r2Credentials,
+    Long expirationTime) {
+
+  static TemporaryCredentials from(StorageCredentials minted) {
+    String location = minted.prefix();
+    Long expiration = minted.expiration() == null ? null : minted.expiration().toEpochMilli();
+    return switch (minted.provider()) {
+      case AWS -> new TemporaryCredentials(location, keys(minted), null, null, null, expiration);
+      case AZURE ->
+          new TemporaryCredentials(
+              location,
+              null,
+              new AzureUserDelegationSas(minted.require(StorageCredentials.SAS_TOKEN)),
+              null,
+              null,
+              expiration);
+      case GCP ->
+          new TemporaryCredentials(
+              location,
+              null,
+              null,
+              new GcpOauthToken(minted.require(StorageCredentials.OAUTH_TOKEN)),
+              null,
+              expiration);
+      case R2 -> new TemporaryCredentials(location, null, null, null, keys(minted), expiration);
+    };
+  }
+
+  private static KeyCredentials keys(StorageCredentials minted) {
+    String session = minted.credentials().get(StorageCredentials.SESSION_TOKEN);
+    return new KeyCredentials(
+        minted.require(StorageCredentials.ACCESS_KEY_ID),
+        minted.require(StorageCredentials.SECRET_ACCESS_KEY),
+        session == null || session.isBlank() ? null : session);
+  }
+
+  public record KeyCredentials(String accessKeyId, String secretAccessKey, String sessionToken) {}
+
+  public record AzureUserDelegationSas(String sasToken) {}
+
+  public record GcpOauthToken(String oauthToken) {}
+}

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsRequest.java

@@ -0,0 +1,8 @@
+package io.opensharing.asset.table;
+
+/**
+ * Optional body for Generate Temporary Table Credential.
+ *
+ * @param location the table root or one of its auxiliary locations; the table root if null
+ */
+public record TemporaryTableCredentialsRequest(String location) {}

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsResponse.java

@@ -0,0 +1,10 @@
+package io.opensharing.asset.table;
+
+/**
+ * Generate Temporary Table Credential response:
+ *
+ * <pre>{@code
+ * {"credentials": {"location": "<storage location>", ..., "expirationTime": <epoch milliseconds>}}
+ * }</pre>
+ */
+public record TemporaryTableCredentialsResponse(TemporaryCredentials credentials) {}

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java

@@ -74,7 +74,7 @@ Session open(ResolvedAsset table, AuthContext auth) {
             auth);
     StorageCredentials rootCredentials =
         credentials.stream()
-            .filter(candidate -> covers(candidate.prefix(), location))
+            .filter(candidate -> candidate.covers(location))
             .findFirst()
             .orElseThrow(
                 () ->
@@ -86,13 +86,5 @@ Session open(ResolvedAsset table, AuthContext auth) {
     return new Session(engine, Table.forPath(engine, path), location);
   }
 
-  private static boolean covers(String prefix, String location) {
-    if (prefix == null || prefix.isBlank()) {
-      return false;
-    }
-    String normalized = prefix.endsWith("/") ? prefix : prefix + "/";
-    return location.equals(prefix) || location.startsWith(normalized);
-  }
-
   record Session(Engine engine, Table table, String location) {}
 }

server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java

@@ -43,6 +43,15 @@ public record StorageCredentials(
     credentials = credentials == null ? Map.of() : Map.copyOf(credentials);
   }
 
+  /** Whether these credentials' prefix is {@code location} or a directory above it. */
+  public boolean covers(String location) {
+    if (prefix == null || prefix.isBlank()) {
+      return false;
+    }
+    String directory = prefix.endsWith("/") ? prefix : prefix + "/";
+    return location.equals(prefix) || location.startsWith(directory);
+  }
+
   public String require(String key) {
     String value = credentials.get(key);
     if (value == null || value.isBlank()) {

server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java

@@ -1,5 +1,7 @@
 package io.opensharing.protocol;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
 import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.HttpStatus;
@@ -11,11 +13,15 @@
 import com.linecorp.armeria.server.annotation.Get;
 import com.linecorp.armeria.server.annotation.Header;
 import com.linecorp.armeria.server.annotation.Param;
+import com.linecorp.armeria.server.annotation.Post;
 import com.linecorp.armeria.server.annotation.Produces;
 import com.linecorp.armeria.server.annotation.ProducesJson;
 import io.opensharing.asset.table.DeltaSharingCapabilities;
 import io.opensharing.asset.table.ProtocolTables;
+import io.opensharing.asset.table.TemporaryTableCredentialsRequest;
+import io.opensharing.asset.table.TemporaryTableCredentialsResponse;
 import io.opensharing.asset.table.delta.DeltaTableMetadataReader;
+import io.opensharing.http.ApiJson;
 
 /**
  * The Delta Sharing protocol endpoints for one shared table, which recipients call with their
@@ -83,4 +89,25 @@ public HttpResponse getTableMetadata(
             .build();
     return HttpResponse.of(headers, HttpData.ofUtf8(result.ndjson()));
   }
+
+  /**
+   * {@code POST /shares/{share}/schemas/{schema}/tables/{table}/temporary-table-credentials}: read
+   * credentials for a table that supports directory access. The request body is optional.
+   */
+  @Post("/shares/{share}/schemas/{schema}/tables/{table}/temporary-table-credentials")
+  public TemporaryTableCredentialsResponse generateTemporaryTableCredentials(
+      ServiceRequestContext ctx,
+      @Param String share,
+      @Param String schema,
+      @Param String table,
+      AggregatedHttpRequest request)
+      throws JsonProcessingException {
+    String body = request.contentUtf8();
+    TemporaryTableCredentialsRequest options =
+        body.isBlank()
+            ? null
+            : ApiJson.MAPPER.readValue(body, TemporaryTableCredentialsRequest.class);
+    return tables.temporaryCredentials(
+        RecipientTokenDecorator.recipient(ctx), share, schema, table, options);
+  }
 }

server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TemporaryCredentialsTest.java

@@ -0,0 +1,93 @@
+package io.opensharing.asset.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import io.opensharing.catalog.CloudProvider;
+import io.opensharing.catalog.StorageCredentials;
+import java.time.Instant;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class TemporaryCredentialsTest {
+
+  private static final Instant EXPIRATION = Instant.parse("2026-01-01T00:00:00Z");
+
+  @Test
+  void mapsAwsSessionCredentials() {
+    TemporaryCredentials credentials =
+        TemporaryCredentials.from(
+            minted(
+                "s3://bucket/table/",
+                CloudProvider.AWS,
+                Map.of(
+                    StorageCredentials.ACCESS_KEY_ID, "ASIATEST",
+                    StorageCredentials.SECRET_ACCESS_KEY, "secret",
+                    StorageCredentials.SESSION_TOKEN, "session")));
+    assertEquals("s3://bucket/table/", credentials.location());
+    assertEquals("ASIATEST", credentials.awsTempCredentials().accessKeyId());
+    assertEquals("secret", credentials.awsTempCredentials().secretAccessKey());
+    assertEquals("session", credentials.awsTempCredentials().sessionToken());
+    assertEquals(EXPIRATION.toEpochMilli(), credentials.expirationTime());
+    assertNull(credentials.azureUserDelegationSas());
+    assertNull(credentials.gcpOauthToken());
+    assertNull(credentials.r2Credentials());
+  }
+
+  @Test
+  void mapsAzureUserDelegationSas() {
+    TemporaryCredentials credentials =
+        TemporaryCredentials.from(
+            minted(
+                "abfss://c@a.dfs.core.windows.net/t",
+                CloudProvider.AZURE,
+                Map.of(StorageCredentials.SAS_TOKEN, "sv=sig")));
+    assertEquals("abfss://c@a.dfs.core.windows.net/t", credentials.location());
+    assertEquals("sv=sig", credentials.azureUserDelegationSas().sasToken());
+    assertEquals(EXPIRATION.toEpochMilli(), credentials.expirationTime());
+    assertNull(credentials.awsTempCredentials());
+    assertNull(credentials.gcpOauthToken());
+    assertNull(credentials.r2Credentials());
+  }
+
+  @Test
+  void mapsGcpOauthToken() {
+    TemporaryCredentials credentials =
+        TemporaryCredentials.from(
+            minted(
+                "gs://bucket/table",
+                CloudProvider.GCP,
+                Map.of(StorageCredentials.OAUTH_TOKEN, "ya29.token")));
+    assertEquals("gs://bucket/table", credentials.location());
+    assertEquals("ya29.token", credentials.gcpOauthToken().oauthToken());
+    assertEquals(EXPIRATION.toEpochMilli(), credentials.expirationTime());
+    assertNull(credentials.awsTempCredentials());
+    assertNull(credentials.azureUserDelegationSas());
+    assertNull(credentials.r2Credentials());
+  }
+
+  @Test
+  void mapsR2Credentials() {
+    TemporaryCredentials credentials =
+        TemporaryCredentials.from(
+            minted(
+                "s3://r2/table",
+                CloudProvider.R2,
+                Map.of(
+                    StorageCredentials.ACCESS_KEY_ID, "r2key",
+                    StorageCredentials.SECRET_ACCESS_KEY, "r2secret")));
+    assertEquals("s3://r2/table", credentials.location());
+    assertEquals("r2key", credentials.r2Credentials().accessKeyId());
+    assertEquals("r2secret", credentials.r2Credentials().secretAccessKey());
+    assertNull(credentials.r2Credentials().sessionToken());
+    assertEquals(EXPIRATION.toEpochMilli(), credentials.expirationTime());
+    assertNull(credentials.awsTempCredentials());
+    assertNull(credentials.azureUserDelegationSas());
+    assertNull(credentials.gcpOauthToken());
+  }
+
+  private static StorageCredentials minted(
+      String location, CloudProvider provider, Map<String, String> values) {
+    return new StorageCredentials(location, provider, values, EXPIRATION);
+  }
+}