From 60424dd2bd9c3b7303ebe71fd6fc86c21ee1b032 Mon Sep 17 00:00:00 2001 From: Sun Cao Date: Tue, 13 Oct 2026 18:36:21 -0700 Subject: [PATCH] Vend protocol temporary table credentials from the catalog, on Armeria. --- .../asset/table/ProtocolTables.java | 59 ++++++++++++ .../asset/table/TemporaryCredentials.java | 66 +++++++++++++ .../TemporaryTableCredentialsRequest.java | 8 ++ .../TemporaryTableCredentialsResponse.java | 10 ++ .../asset/table/delta/DeltaKernel.java | 10 +- .../catalog/StorageCredentials.java | 9 ++ .../protocol/DeltaSharingProtocolService.java | 27 ++++++ .../asset/table/TemporaryCredentialsTest.java | 93 +++++++++++++++++++ .../ProtocolTableCredentialsTest.java | 82 ++++++++++++++++ 9 files changed, 355 insertions(+), 9 deletions(-) create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryCredentials.java create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsRequest.java create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsResponse.java create mode 100644 server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TemporaryCredentialsTest.java create mode 100644 server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableCredentialsTest.java diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java index 171e49340..7f736c3e3 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java +++ b/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 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 inSchema(ShareEntity share, String schema) { List 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; diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryCredentials.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryCredentials.java new file mode 100644 index 000000000..975a191f2 --- /dev/null +++ b/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: + * + * 
{@code
+ * {"location": "",
+ *  "awsTempCredentials": {"accessKeyId": "", "secretAccessKey": "",
+ *                         "sessionToken": ""},
+ *  "azureUserDelegationSas": {"sasToken": ""},
+ *  "gcpOauthToken": {"oauthToken": ""},
+ *  "r2Credentials": {"accessKeyId": "", "secretAccessKey": "",
+ *                    "sessionToken": ""},
+ *  "expirationTime": }
+ * }
+ */ +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) {} +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsRequest.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsRequest.java new file mode 100644 index 000000000..13c7ce6e5 --- /dev/null +++ b/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) {} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsResponse.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TemporaryTableCredentialsResponse.java new file mode 100644 index 000000000..08f185816 --- /dev/null +++ b/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: + * + * 
{@code
+ * {"credentials": {"location": "", ..., "expirationTime": }}
+ * }
+ */ +public record TemporaryTableCredentialsResponse(TemporaryCredentials credentials) {} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java index 123525dd9..f84d2d642 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java +++ b/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) {} } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java b/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java index 5623a5844..3758a78da 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java +++ b/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()) { diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java index d3492ea0a..c739194cc 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java +++ b/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); + } } diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TemporaryCredentialsTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TemporaryCredentialsTest.java new file mode 100644 index 000000000..0bca5d3a0 --- /dev/null +++ b/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 values) { + return new StorageCredentials(location, provider, values, EXPIRATION); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableCredentialsTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableCredentialsTest.java new file mode 100644 index 000000000..9cf97dd34 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableCredentialsTest.java @@ -0,0 +1,82 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.JsonNode; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import org.junit.jupiter.api.Test; + +class ProtocolTableCredentialsTest extends ProtocolApiSupport { + + @Test + void vendsTemporaryCredentialsForAGrantedTable() { + createShare("table-creds"); + addObject("table-creds", "TABLE", "main.sales.orders", "sales.orders"); + addObject("table-creds", "SCHEMA", "main.hr", "hr"); + createShare("hidden-creds"); + addObject("hidden-creds", "TABLE", "main.sales.customers", "sales.customers"); + String bearer = createAndActivateRecipient("table-creds-partner"); + grant("table-creds", "table-creds-partner"); + + // No body vends the local catalog's placeholder AWS credentials for the table root. + Response root = server.post(credentials("TABLE-CREDS", "SALES", "ORDERS"), bearer, null); + assertEquals(200, root.status()); + JsonNode credentials = root.body().path("credentials"); + assertEquals("s3://test/main.sales.orders/", credentials.path("location").asText()); + JsonNode aws = credentials.path("awsTempCredentials"); + assertTrue(aws.path("accessKeyId").asText().startsWith("ASIA")); + assertTrue(aws.has("secretAccessKey")); + assertTrue(aws.has("sessionToken")); + assertTrue(credentials.path("azureUserDelegationSas").isMissingNode()); + assertTrue(credentials.path("expirationTime").asLong() > 0); + + // An explicit location may repeat the table root. + Response explicit = + server.post( + credentials("table-creds", "sales", "orders"), + bearer, + "{\"location\":\"s3://test/main.sales.orders/\"}"); + assertEquals(200, explicit.status()); + assertEquals( + "s3://test/main.sales.orders/", + explicit.body().path("credentials").path("location").asText()); + + // A SCHEMA grant can mint credentials for a catalog child. + Response child = server.post(credentials("table-creds", "hr", "employees"), bearer, "{}"); + assertEquals(200, child.status()); + assertEquals( + "s3://test/main.hr.employees/", child.body().path("credentials").path("location").asText()); + + // A body that is not JSON is malformed. + Response malformed = server.post(credentials("table-creds", "sales", "orders"), bearer, "{"); + assertEquals(400, malformed.status()); + assertEquals(ErrorCodes.MALFORMED_REQUEST, malformed.errorCode()); + + // A location outside the table root and auxiliaries is INVALID_PARAMETER_VALUE. + Response outside = + server.post( + credentials("table-creds", "sales", "orders"), + bearer, + "{\"location\":\"s3://other/prefix/\"}"); + assertEquals(400, outside.status()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, outside.errorCode()); + + // A table in an ungranted share is 404, not 403. + Response hidden = server.post(credentials("hidden-creds", "sales", "customers"), bearer, null); + assertEquals(404, hidden.status()); + assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode()); + } + + private static String credentials(String share, String schema, String table) { + return PROTOCOL + + "/shares/" + + share + + "/schemas/" + + schema + + "/tables/" + + table + + "/temporary-table-credentials"; + } +}