diff --git a/server/opensharing-server-core/pom.xml b/server/opensharing-server-core/pom.xml index 2cc3094b..cc6acb41 100644 --- a/server/opensharing-server-core/pom.xml +++ b/server/opensharing-server-core/pom.xml @@ -76,7 +76,68 @@ software.amazon.awssdk s3 ${aws-sdk.version} - runtime + + + com.azure + azure-storage-blob + ${azure-storage-blob.version} + + + com.azure + azure-core-http-netty + + + com.fasterxml.jackson.dataformat + jackson-dataformat-xml + + + + + com.google.cloud + google-cloud-storage + ${google-cloud-storage.version} + + + com.google.code.gson + gson + + + com.google.cloud + google-cloud-core-grpc + + + com.google.api.grpc + grpc-google-cloud-storage-v2 + + + com.google.api.grpc + gapic-google-cloud-storage-v2 + + + io.grpc + grpc-netty-shaded + + + io.grpc + grpc-alts + + + io.grpc + grpc-auth + + + io.grpc + grpc-inprocess + + + com.google.api + gax-grpc + + + + + com.google.code.gson + gson software.amazon.awssdk 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 f84d2d64..06df9fa5 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 @@ -53,7 +53,7 @@ static long versionAtOrAfter(TableImpl table, Engine engine, Instant startingTim } /** Opens the table's root with read credentials the catalog vends to {@code auth}. */ - Session open(ResolvedAsset table, AuthContext auth) { + public Session open(ResolvedAsset table, AuthContext auth) { if (!(table.additionalProperties() instanceof TableProperties properties) || properties.dataSourceFormat() != DataSourceFormat.DELTA) { throw ApiException.invalidParameter("table '" + table.fullName() + "' is not a Delta table"); @@ -86,5 +86,5 @@ Session open(ResolvedAsset table, AuthContext auth) { return new Session(engine, Table.forPath(engine, path), location); } - record Session(Engine engine, Table table, String location) {} + public record Session(Engine engine, Table table, String location) {} } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java index 0bd49f1e..328c8e7d 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java @@ -40,10 +40,7 @@ private static void configureS3(Configuration configuration, StorageCredentials "fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider"); } - String region = credentials.credentials().get(StorageCredentials.REGION); - if (region != null && !region.isBlank()) { - configuration.set("fs.s3a.endpoint.region", region); - } + configuration.set("fs.s3a.endpoint.region", credentials.require(StorageCredentials.REGION)); } private static void configureAzure( diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/AzureSasUrlSigner.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/AzureSasUrlSigner.java new file mode 100644 index 00000000..bdf93dc9 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/AzureSasUrlSigner.java @@ -0,0 +1,102 @@ +package io.opensharing.asset.table.signer; + +import com.azure.storage.blob.sas.BlobSasPermission; +import com.azure.storage.blob.sas.BlobServiceSasSignatureValues; +import com.azure.storage.common.StorageSharedKeyCredential; +import com.azure.storage.common.sas.SasProtocol; +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.http.ApiException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.Locale; +import java.util.Set; + +/** + * Azure blob URLs: append a catalog SAS when present, otherwise mint a read-only HTTPS blob SAS + * from an account key. + */ +final class AzureSasUrlSigner implements UrlSigner { + + @Override + public Set schemes() { + return Set.of("abfs", "abfss", "wasb", "wasbs"); + } + + @Override + public SignedUrl sign(String path, StorageCredentials credentials, Duration ttl) { + Instant expiration = Instant.now().plus(ttl); + String sas = credentials.credentials().get(StorageCredentials.SAS_TOKEN); + if (sas != null && !sas.isBlank()) { + // Catalog SAS is already minted; its lifetime is credentials.expiration(), not the sign TTL. + Instant sasExpiry = credentials.expiration() == null ? expiration : credentials.expiration(); + return new SignedUrl(httpsUrl(path) + withQuery(sas), sasExpiry); + } + String accountKey = credentials.credentials().get(StorageCredentials.AZURE_ACCOUNT_KEY); + if (accountKey == null || accountKey.isBlank()) { + throw ApiException.invalidParameter( + "Azure credentials have neither a SAS token nor an account key"); + } + try { + String blobSas = blobSas(path, accountKey, expiration); + return new SignedUrl(httpsUrl(path) + withQuery(blobSas), expiration); + } catch (RuntimeException e) { + throw ApiException.invalidParameter("Azure account key cannot sign a blob SAS"); + } + } + + private static String blobSas(String path, String accountKey, Instant expiration) { + StoragePaths.AzureBlob parsed = StoragePaths.azureBlob(path); + String account = parsed.host().split("\\.")[0]; + return new BlobServiceSasSignatureValues( + OffsetDateTime.ofInstant(expiration, ZoneOffset.UTC), + new BlobSasPermission().setReadPermission(true)) + .setProtocol(SasProtocol.HTTPS_ONLY) + .setContainerName(parsed.container()) + .setBlobName(parsed.blob()) + .generateSasQueryParameters(new StorageSharedKeyCredential(account, accountKey)) + .encode(); + } + + private static String httpsUrl(String path) { + StoragePaths.AzureBlob parsed = StoragePaths.azureBlob(path); + return "https://" + + parsed.host().replace(".dfs.", ".blob.") + + "/" + + parsed.container() + + encodePath(parsed.blob()); + } + + static String encodePath(String blob) { + StringBuilder path = new StringBuilder(); + for (String segment : blob.split("/", -1)) { + path.append('/').append(percentEncode(segment)); + } + return path.toString(); + } + + private static String percentEncode(String value) { + StringBuilder encoded = new StringBuilder(value.length()); + for (byte b : value.getBytes(StandardCharsets.UTF_8)) { + char c = (char) (b & 0xFF); + if ((c >= 'A' && c <= 'Z') + || (c >= 'a' && c <= 'z') + || (c >= '0' && c <= '9') + || c == '-' + || c == '.' + || c == '_' + || c == '~') { + encoded.append(c); + } else { + encoded.append('%').append(String.format(Locale.ROOT, "%02X", b & 0xFF)); + } + } + return encoded.toString(); + } + + private static String withQuery(String sas) { + return sas.startsWith("?") ? sas : "?" + sas; + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/GcsUrlSigner.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/GcsUrlSigner.java new file mode 100644 index 00000000..ba6c8d6e --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/GcsUrlSigner.java @@ -0,0 +1,74 @@ +package io.opensharing.asset.table.signer; + +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auth.oauth2.ServiceAccountCredentials; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.BlobInfo; +import com.google.cloud.storage.Storage; +import com.google.cloud.storage.StorageOptions; +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.http.ApiException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** GCS V4 signed URLs via {@code Storage.signUrl}. */ +final class GcsUrlSigner implements UrlSigner { + + @Override + public Set schemes() { + return Set.of("gs"); + } + + @Override + public SignedUrl sign(String path, StorageCredentials credentials, Duration ttl) { + String[] bucketAndObject = StoragePaths.bucketAndKey(path); + Instant expiration = Instant.now().plus(ttl); + String url = + storage(credentials) + .signUrl( + BlobInfo.newBuilder(BlobId.of(bucketAndObject[0], bucketAndObject[1])).build(), + ttl.toSeconds(), + TimeUnit.SECONDS, + Storage.SignUrlOption.withV4Signature()) + .toString(); + return new SignedUrl(url, expiration); + } + + /** + * Signs with the service account key file the catalog vends, if any, and otherwise with Google + * application default credentials. + */ + private static Storage storage(StorageCredentials credentials) { + String file = + credentials == null + ? null + : credentials.credentials().get(StorageCredentials.GOOGLE_SERVICE_ACCOUNT_KEY_FILE); + if (file != null && !file.isBlank()) { + return StorageOptions.newBuilder() + .setCredentials(credentialsFromFile(file.trim())) + .build() + .getService(); + } + try { + return StorageOptions.getDefaultInstance().getService(); + } catch (RuntimeException e) { + throw ApiException.notImplemented( + "no Google service account key is configured to sign gs urls"); + } + } + + private static GoogleCredentials credentialsFromFile(String file) { + try (InputStream in = Files.newInputStream(Path.of(file))) { + return ServiceAccountCredentials.fromStream(in); + } catch (IOException | RuntimeException e) { + throw new IllegalStateException( + "the Google service account key file '" + file + "' cannot be read", e); + } + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/S3UrlSigner.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/S3UrlSigner.java new file mode 100644 index 00000000..45f08cde --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/S3UrlSigner.java @@ -0,0 +1,63 @@ +package io.opensharing.asset.table.signer; + +import io.opensharing.catalog.StorageCredentials; +import java.time.Duration; +import java.time.Instant; +import java.util.Set; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.presigner.S3Presigner; +import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest; + +/** AWS SDK GET presigning for S3 object URLs. */ +final class S3UrlSigner implements UrlSigner { + + @Override + public Set schemes() { + return Set.of("s3", "s3a", "s3n"); + } + + @Override + public SignedUrl sign(String path, StorageCredentials credentials, Duration ttl) { + String[] bucketAndKey = StoragePaths.bucketAndKey(path); + AwsCredentials aws = awsCredentials(credentials); + Instant expiration = Instant.now().plus(ttl); + try (S3Presigner presigner = + S3Presigner.builder() + .region(Region.of(credentials.require(StorageCredentials.REGION))) + .credentialsProvider(StaticCredentialsProvider.create(aws)) + .build()) { + String url = + presigner + .presignGetObject( + GetObjectPresignRequest.builder() + .signatureDuration(ttl) + .getObjectRequest( + GetObjectRequest.builder() + .bucket(bucketAndKey[0]) + .key(bucketAndKey[1]) + .build()) + .build()) + .url() + .toString(); + return new SignedUrl(url, expiration); + } + } + + private static AwsCredentials awsCredentials(StorageCredentials credentials) { + String session = credentials.credentials().get(StorageCredentials.SESSION_TOKEN); + if (session != null && !session.isBlank()) { + return AwsSessionCredentials.create( + credentials.require(StorageCredentials.ACCESS_KEY_ID), + credentials.require(StorageCredentials.SECRET_ACCESS_KEY), + session); + } + return AwsBasicCredentials.create( + credentials.require(StorageCredentials.ACCESS_KEY_ID), + credentials.require(StorageCredentials.SECRET_ACCESS_KEY)); + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/SignedUrl.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/SignedUrl.java new file mode 100644 index 00000000..12f4d208 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/SignedUrl.java @@ -0,0 +1,6 @@ +package io.opensharing.asset.table.signer; + +import java.time.Instant; + +/** A recipient-readable URL and the moment its signature expires. */ +public record SignedUrl(String url, Instant expiration) {} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/StoragePaths.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/StoragePaths.java new file mode 100644 index 00000000..6cb0de9a --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/StoragePaths.java @@ -0,0 +1,40 @@ +package io.opensharing.asset.table.signer; + +import io.opensharing.http.ApiException; + +/** Splits a Hadoop-style object path without requiring it to be a legal {@code java.net.URI}. */ +final class StoragePaths { + + private StoragePaths() {} + + static String afterScheme(String path) { + int separator = path.indexOf("://"); + if (separator < 0) { + throw ApiException.invalidParameter("'" + path + "' has no storage scheme"); + } + return path.substring(separator + 3); + } + + static String[] bucketAndKey(String path) { + String rest = afterScheme(path); + int slash = rest.indexOf('/'); + if (slash <= 0 || slash == rest.length() - 1) { + throw ApiException.invalidParameter("'" + path + "' is not a bucket/object path"); + } + return new String[] {rest.substring(0, slash), rest.substring(slash + 1)}; + } + + static AzureBlob azureBlob(String path) { + String rest = afterScheme(path); + int at = rest.indexOf('@'); + int slash = rest.indexOf('/'); + if (at <= 0 || slash < 0 || at > slash) { + throw ApiException.invalidParameter( + "'" + path + "' is not an Azure path of the form scheme://container@account.host/blob"); + } + return new AzureBlob( + rest.substring(0, at), rest.substring(at + 1, slash), rest.substring(slash + 1)); + } + + record AzureBlob(String container, String host, String blob) {} +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigner.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigner.java new file mode 100644 index 00000000..8bf2f50b --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigner.java @@ -0,0 +1,13 @@ +package io.opensharing.asset.table.signer; + +import io.opensharing.catalog.StorageCredentials; +import java.time.Duration; +import java.util.Set; + +/** Signs one storage object for credential-free, read-only HTTPS access. */ +interface UrlSigner { + + Set schemes(); + + SignedUrl sign(String path, StorageCredentials credentials, Duration ttl); +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigners.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigners.java new file mode 100644 index 00000000..3dd17283 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/signer/UrlSigners.java @@ -0,0 +1,60 @@ +package io.opensharing.asset.table.signer; + +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.http.ApiException; +import io.opensharing.http.ErrorCodes; +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +/** Picks a cloud signer for a storage path. */ +public class UrlSigners { + + private final Map byScheme = new HashMap<>(); + + /** Signs S3, Azure, and GCS paths. */ + public UrlSigners() { + this(List.of(new S3UrlSigner(), new AzureSasUrlSigner(), new GcsUrlSigner())); + } + + UrlSigners(List signers) { + for (UrlSigner signer : signers) { + signer.schemes().forEach(scheme -> byScheme.put(scheme.toLowerCase(Locale.ROOT), signer)); + } + } + + public SignedUrl sign(String path, StorageCredentials credentials, Duration ttl) { + return signerFor(path).sign(path, credentials, capped(credentials, ttl)); + } + + private UrlSigner signerFor(String path) { + String scheme = scheme(path); + UrlSigner signer = byScheme.get(scheme); + if (signer == null) { + throw ApiException.notImplemented("this build cannot sign urls for '" + scheme + "' storage"); + } + return signer; + } + + static String scheme(String path) { + int separator = path.indexOf("://"); + return separator < 0 ? "file" : path.substring(0, separator).toLowerCase(Locale.ROOT); + } + + private static Duration capped(StorageCredentials credentials, Duration ttl) { + if (credentials == null || credentials.expiration() == null) { + return ttl; + } + Duration untilExpiry = Duration.between(Instant.now(), credentials.expiration()); + if (untilExpiry.isNegative() || untilExpiry.isZero()) { + throw new ApiException( + 502, + ErrorCodes.CATALOG_ERROR, + "the catalog vended credentials that had already expired, so no url can be signed"); + } + return untilExpiry.compareTo(ttl) < 0 ? untilExpiry : ttl; + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/AzureSasUrlSignerTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/AzureSasUrlSignerTest.java new file mode 100644 index 00000000..5b55ac2a --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/AzureSasUrlSignerTest.java @@ -0,0 +1,90 @@ +package io.opensharing.asset.table.signer; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.opensharing.catalog.CloudProvider; +import io.opensharing.catalog.StorageCredentials; +import java.time.Duration; +import java.time.Instant; +import java.util.Base64; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class AzureSasUrlSignerTest { + + private final AzureSasUrlSigner signer = new AzureSasUrlSigner(); + + @Test + void appendsCatalogSasToHttpsBlobUrl() { + StorageCredentials credentials = + new StorageCredentials( + "abfss://container@account.dfs.core.windows.net/", + CloudProvider.AZURE, + Map.of(StorageCredentials.SAS_TOKEN, "sv=2020-12-06&sig=catalog"), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = + signer.sign( + "abfss://container@account.dfs.core.windows.net/table/part.parquet", + credentials, + Duration.ofMinutes(15)); + + assertTrue( + signed + .url() + .startsWith( + "https://account.blob.core.windows.net/container/table/part.parquet" + + "?sv=2020-12-06&sig=catalog"), + signed.url()); + long remaining = Duration.between(Instant.now(), signed.expiration()).toSeconds(); + assertTrue(remaining >= 59 * 60 && remaining <= 60 * 60, String.valueOf(remaining)); + } + + @Test + void mintsReadOnlyBlobSasFromAccountKey() { + String accountKey = Base64.getEncoder().encodeToString(new byte[64]); + StorageCredentials credentials = + new StorageCredentials( + "abfss://container@account.dfs.core.windows.net/", + CloudProvider.AZURE, + Map.of(StorageCredentials.AZURE_ACCOUNT_KEY, accountKey), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = + signer.sign( + "abfss://container@account.dfs.core.windows.net/table/part.parquet", + credentials, + Duration.ofMinutes(15)); + + assertTrue( + signed + .url() + .startsWith("https://account.blob.core.windows.net/container/table/part.parquet?"), + signed.url()); + assertTrue(signed.url().contains("sp=r"), signed.url()); + assertTrue(signed.url().contains("spr=https"), signed.url()); + assertTrue(signed.url().contains("sig="), signed.url()); + assertTrue(signed.url().contains("se="), signed.url()); + long remaining = Duration.between(Instant.now(), signed.expiration()).toSeconds(); + assertTrue(remaining >= 14 * 60 && remaining <= 15 * 60, String.valueOf(remaining)); + } + + @Test + void encodesSpacesInBlobPaths() { + String accountKey = Base64.getEncoder().encodeToString(new byte[64]); + StorageCredentials credentials = + new StorageCredentials( + "abfss://container@account.dfs.core.windows.net/", + CloudProvider.AZURE, + Map.of(StorageCredentials.AZURE_ACCOUNT_KEY, accountKey), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = + signer.sign( + "abfss://container@account.dfs.core.windows.net/table/c2=foo bar/part.parquet", + credentials, + Duration.ofMinutes(15)); + + assertTrue(signed.url().contains("/table/c2%3Dfoo%20bar/part.parquet?"), signed.url()); + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/GcsUrlSignerTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/GcsUrlSignerTest.java new file mode 100644 index 00000000..7e953621 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/GcsUrlSignerTest.java @@ -0,0 +1,87 @@ +package io.opensharing.asset.table.signer; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import io.opensharing.catalog.CloudProvider; +import io.opensharing.catalog.StorageCredentials; +import java.nio.file.Path; +import java.security.KeyPair; +import java.security.KeyPairGenerator; +import java.time.Duration; +import java.time.Instant; +import java.util.Base64; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class GcsUrlSignerTest { + + @TempDir Path tempDir; + + @Test + void signsGsPathsWithCatalogServiceAccountKey() throws Exception { + Path keyFile = writeServiceAccount(tempDir.resolve("sa.json")); + GcsUrlSigner signer = new GcsUrlSigner(); + StorageCredentials credentials = + new StorageCredentials( + "gs://bucket/", + CloudProvider.GCP, + Map.of(StorageCredentials.GOOGLE_SERVICE_ACCOUNT_KEY_FILE, keyFile.toString()), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = + signer.sign("gs://bucket/path/file.parquet", credentials, Duration.ofMinutes(10)); + + assertTrue(signed.url().startsWith("https://storage.googleapis.com/bucket/path/file.parquet?")); + assertTrue(signed.url().contains("X-Goog-Algorithm=GOOG4-RSA-SHA256")); + assertTrue(signed.url().contains("X-Goog-Signature=")); + assertTrue(signed.url().contains("X-Goog-Expires=600"), signed.url()); + long remaining = Duration.between(Instant.now(), signed.expiration()).toSeconds(); + assertTrue(remaining >= 9 * 60 && remaining <= 10 * 60, String.valueOf(remaining)); + } + + @Test + void encodesSpacesInObjectPaths() throws Exception { + Path keyFile = writeServiceAccount(tempDir.resolve("sa.json")); + GcsUrlSigner signer = new GcsUrlSigner(); + StorageCredentials credentials = + new StorageCredentials( + "gs://bucket/", + CloudProvider.GCP, + Map.of(StorageCredentials.GOOGLE_SERVICE_ACCOUNT_KEY_FILE, keyFile.toString()), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = + signer.sign("gs://bucket/c2=foo bar/file.parquet", credentials, Duration.ofMinutes(10)); + + assertTrue( + signed.url().contains("/bucket/c2%3Dfoo%20bar/file.parquet?"), signed.url()); + } + + private static Path writeServiceAccount(Path file) throws Exception { + KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA"); + generator.initialize(2048); + KeyPair pair = generator.generateKeyPair(); + String keyType = "PRIVATE" + " KEY"; + String pem = + "-----BEGIN " + + keyType + + "-----\n" + + Base64.getEncoder().encodeToString(pair.getPrivate().getEncoded()) + + "\n-----END " + + keyType + + "-----\n"; + ObjectNode json = new ObjectMapper().createObjectNode(); + json.put("type", "service_account"); + json.put("project_id", "example"); + json.put("private_key_id", "test"); + json.put("private_key", pem); + json.put("client_email", "signer@example.iam.gserviceaccount.com"); + json.put("client_id", "123"); + json.put("token_uri", "https://oauth2.googleapis.com/token"); + new ObjectMapper().writeValue(file.toFile(), json); + return file; + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/S3UrlSignerTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/S3UrlSignerTest.java new file mode 100644 index 00000000..fe4a2456 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/S3UrlSignerTest.java @@ -0,0 +1,67 @@ +package io.opensharing.asset.table.signer; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.opensharing.catalog.CloudProvider; +import io.opensharing.catalog.StorageCredentials; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class S3UrlSignerTest { + + private final S3UrlSigner signer = new S3UrlSigner(); + + @Test + void signsS3aPathsWithOptionalSessionTokenAndCredentialRegion() { + Instant expiration = Instant.now().plus(Duration.ofHours(1)); + StorageCredentials credentials = + new StorageCredentials( + "s3://delta-exchange-test/", + CloudProvider.AWS, + Map.of( + StorageCredentials.ACCESS_KEY_ID, "AKIATEST", + StorageCredentials.SECRET_ACCESS_KEY, "secret", + StorageCredentials.SESSION_TOKEN, "session", + StorageCredentials.REGION, "us-west-2"), + expiration); + + SignedUrl signed = + signer.sign( + "s3a://delta-exchange-test/table2/date=2021-04-28/part.parquet", + credentials, + Duration.ofMinutes(15)); + + assertTrue( + signed + .url() + .startsWith( + "https://delta-exchange-test.s3.us-west-2.amazonaws.com" + + "/table2/date%3D2021-04-28/part.parquet?")); + assertTrue(signed.url().contains("X-Amz-Algorithm=AWS4-HMAC-SHA256")); + assertTrue(signed.url().contains("X-Amz-Security-Token=session")); + assertTrue(signed.url().contains("X-Amz-Signature=")); + long remaining = Duration.between(Instant.now(), signed.expiration()).toSeconds(); + assertTrue(remaining >= 14 * 60 && remaining <= 15 * 60); + } + + @Test + void omitsSessionTokenWhenCatalogOmitsIt() { + StorageCredentials credentials = + new StorageCredentials( + "s3://bucket/", + CloudProvider.AWS, + Map.of( + StorageCredentials.ACCESS_KEY_ID, "AKIATEST", + StorageCredentials.SECRET_ACCESS_KEY, "secret", + StorageCredentials.REGION, "us-west-2"), + Instant.now().plus(Duration.ofHours(1))); + + SignedUrl signed = signer.sign("s3://bucket/key.parquet", credentials, Duration.ofMinutes(5)); + + assertTrue(signed.url().startsWith("https://bucket.s3.us-west-2.amazonaws.com/key.parquet?")); + assertFalse(signed.url().contains("X-Amz-Security-Token")); + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/UrlSignersTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/UrlSignersTest.java new file mode 100644 index 00000000..54f67458 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/signer/UrlSignersTest.java @@ -0,0 +1,72 @@ +package io.opensharing.asset.table.signer; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.opensharing.catalog.CloudProvider; +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.http.ApiException; +import io.opensharing.http.ErrorCodes; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class UrlSignersTest { + + private final UrlSigners signers = + new UrlSigners( + List.of(new S3UrlSigner(), new AzureSasUrlSigner())); + + @Test + void signsBySchemeAndCapsTtlToCredentialExpiry() { + Instant expiry = Instant.now().plus(Duration.ofMinutes(3)); + StorageCredentials credentials = + new StorageCredentials( + "s3://bucket/", + CloudProvider.AWS, + Map.of( + StorageCredentials.ACCESS_KEY_ID, "AKIATEST", + StorageCredentials.SECRET_ACCESS_KEY, "secret", + StorageCredentials.REGION, "us-west-2"), + expiry); + + SignedUrl signed = signers.sign("s3://bucket/key.parquet", credentials, Duration.ofHours(1)); + + assertTrue(signed.url().contains("X-Amz-Signature="), signed.url()); + assertTrue(signed.url().matches(".*X-Amz-Expires=17[0-9].*"), signed.url()); + long remaining = signed.expiration().getEpochSecond() - Instant.now().getEpochSecond(); + assertTrue(remaining >= 170 && remaining <= 180, String.valueOf(remaining)); + } + + @Test + void rejectsUnknownSchemesAndExpiredCredentials() { + StorageCredentials expired = + new StorageCredentials( + "s3://bucket/", + CloudProvider.AWS, + Map.of( + StorageCredentials.ACCESS_KEY_ID, "AKIATEST", + StorageCredentials.SECRET_ACCESS_KEY, "secret", + StorageCredentials.REGION, "us-west-2"), + Instant.now().minusSeconds(5)); + ApiException unknown = + assertThrows( + ApiException.class, + () -> + signers.sign( + "file:///tmp/table/part.parquet", + expired, + Duration.ofMinutes(5))); + assertEquals(501, unknown.getStatus()); + assertEquals(ErrorCodes.NOT_IMPLEMENTED, unknown.getErrorCode()); + + ApiException alreadyExpired = + assertThrows( + ApiException.class, + () -> signers.sign("s3://bucket/key.parquet", expired, Duration.ofMinutes(5))); + assertEquals(502, alreadyExpired.getStatus()); + } +} diff --git a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java index 5a7237d6..7bf21b51 100644 --- a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java +++ b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java @@ -245,10 +245,7 @@ private Map envValues(CloudProvider provider) { case AWS, R2 -> { values.put(StorageCredentials.ACCESS_KEY_ID, requireEnv("AWS_ACCESS_KEY_ID")); values.put(StorageCredentials.SECRET_ACCESS_KEY, requireEnv("AWS_SECRET_ACCESS_KEY")); - String region = firstEnv("AWS_REGION", "AWS_DEFAULT_REGION"); - if (region != null) { - values.put(StorageCredentials.REGION, region); - } + values.put(StorageCredentials.REGION, requireEnv("AWS_REGION", "AWS_DEFAULT_REGION")); String session = firstEnv("AWS_SESSION_TOKEN"); if (session != null) { values.put(StorageCredentials.SESSION_TOKEN, session); @@ -279,11 +276,13 @@ private static CloudProvider providerFor(String location) { }; } - private static String requireEnv(String name) { - String value = firstEnv(name); + private static String requireEnv(String... names) { + String value = firstEnv(names); if (value == null) { throw new CatalogException( - "local catalog credentials.mode is ENV but environment variable '" + name + "' is unset"); + "local catalog credentials.mode is ENV but none of " + + String.join(", ", names) + + " is set"); } return value; } @@ -313,7 +312,6 @@ private Map staticValues(CloudProvider provider) { } if (provider == CloudProvider.AWS || provider == CloudProvider.R2) { copyIfPresent(values, StorageCredentials.SESSION_TOKEN); - copyIfPresent(values, StorageCredentials.REGION); } return values; } @@ -334,6 +332,7 @@ private Map fakeValues(CloudProvider provider, Instant expiratio "ASIA" + randomString(16).toUpperCase(Locale.ROOT)); values.put(StorageCredentials.SECRET_ACCESS_KEY, randomString(40)); values.put(StorageCredentials.SESSION_TOKEN, "local-fake-session-" + randomString(48)); + values.put(StorageCredentials.REGION, "us-west-2"); } case AZURE -> values.put( @@ -348,7 +347,10 @@ private Map fakeValues(CloudProvider provider, Instant expiratio private static List requiredKeys(CloudProvider provider) { return switch (provider) { case AWS, R2 -> - List.of(StorageCredentials.ACCESS_KEY_ID, StorageCredentials.SECRET_ACCESS_KEY); + List.of( + StorageCredentials.ACCESS_KEY_ID, + StorageCredentials.SECRET_ACCESS_KEY, + StorageCredentials.REGION); case AZURE -> List.of(StorageCredentials.SAS_TOKEN); case GCP -> List.of(StorageCredentials.OAUTH_TOKEN); }; diff --git a/server/opensharing-server/src/test/java/io/opensharing/asset/table/signer/UrlSignerCloudIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/asset/table/signer/UrlSignerCloudIntegrationTest.java new file mode 100644 index 00000000..f604664f --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/asset/table/signer/UrlSignerCloudIntegrationTest.java @@ -0,0 +1,117 @@ +package io.opensharing.asset.table.signer; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.delta.kernel.Scan; +import io.delta.kernel.Snapshot; +import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.data.Row; +import io.delta.kernel.internal.InternalScanFileUtils; +import io.delta.kernel.utils.CloseableIterator; +import io.opensharing.asset.table.delta.DeltaKernel; +import io.opensharing.auth.AuthContext; +import io.opensharing.auth.UserContext; +import io.opensharing.catalog.Asset; +import io.opensharing.catalog.AssetType; +import io.opensharing.catalog.CatalogConnector; +import io.opensharing.catalog.CredentialRequest; +import io.opensharing.catalog.ResolvedAsset; +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.catalog.StorageOperation; +import io.opensharing.catalog.local.LocalCatalogConnector; +import io.opensharing.catalog.local.LocalCatalogLoader; +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpResponse.BodyHandlers; +import java.time.Duration; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; + +/** + * Signs a real data file from each cloud test table using catalog-vended ENV credentials, then + * reads the first byte through the signed URL. Cases are skipped unless those credentials are set. + */ +@Timeout(60) +class UrlSignerCloudIntegrationTest { + + private static final AuthContext AUTH = + AuthContext.of(new UserContext("catalog-alice-id", "alice-token", "alice")); + + private final CatalogConnector catalog = + new LocalCatalogConnector(LocalCatalogLoader.load("classpath:local-catalog-cloud.yml")); + private final DeltaKernel kernel = new DeltaKernel(catalog); + private final UrlSigners signers = new UrlSigners(); + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void signsAndReadsARealS3File() throws Exception { + signAndRead("main.sales.table1", "X-Amz-Signature="); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+") + void signsAndReadsARealAzureFile() throws Exception { + signAndRead("main.sales.azure", "sig="); + } + + @Test + @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+") + void signsAndReadsARealGcsFile() throws Exception { + signAndRead("main.sales.gcs", "X-Goog-Signature="); + } + + private void signAndRead(String fullName, String signatureMarker) throws Exception { + ResolvedAsset table = catalog.resolveAsset(new Asset(AssetType.TABLE, fullName), AUTH); + DeltaKernel.Session session = kernel.open(table, AUTH); + String path = firstDataFile(session); + StorageCredentials credentials = + catalog + .getStorageCredentials( + new CredentialRequest( + AssetType.TABLE, + table.fullName(), + table.catalogAssetId(), + session.location(), + StorageOperation.READ, + Duration.ofMinutes(15)), + AUTH) + .getFirst(); + SignedUrl signed = signers.sign(path, credentials, Duration.ofMinutes(5)); + assertTrue(signed.url().startsWith("https://"), signed.url()); + assertTrue(signed.url().contains(signatureMarker), signed.url()); + int status = getFirstByte(signed.url()); + assertTrue(status == 200 || status == 206, "HTTP " + status + " for signed " + path); + } + + private static String firstDataFile(DeltaKernel.Session session) throws IOException { + Snapshot snapshot = session.table().getLatestSnapshot(session.engine()); + Scan scan = snapshot.getScanBuilder().build(); + try (CloseableIterator batches = scan.getScanFiles(session.engine())) { + while (batches.hasNext()) { + try (CloseableIterator rows = batches.next().getRows()) { + while (rows.hasNext()) { + return InternalScanFileUtils.getAddFileStatus(rows.next()).getPath(); + } + } + } + } + throw new AssertionError("table at " + session.location() + " has no data files to sign"); + } + + private static int getFirstByte(String url) throws Exception { + HttpRequest request = + HttpRequest.newBuilder(URI.create(url)) + .timeout(Duration.ofSeconds(30)) + .header("Range", "bytes=0-0") + .GET() + .build(); + HttpResponse response = + HttpClient.newHttpClient().send(request, BodyHandlers.discarding()); + return response.statusCode(); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/catalog/local/LocalCatalogConnectorTest.java b/server/opensharing-server/src/test/java/io/opensharing/catalog/local/LocalCatalogConnectorTest.java index 717bcfbb..b9085c23 100644 --- a/server/opensharing-server/src/test/java/io/opensharing/catalog/local/LocalCatalogConnectorTest.java +++ b/server/opensharing-server/src/test/java/io/opensharing/catalog/local/LocalCatalogConnectorTest.java @@ -196,6 +196,7 @@ void vendsPlaceholderCredentialsScopedToTheAssetLocation() { assertTrue(credentials.expiration().isAfter(java.time.Instant.now())); assertTrue(credentials.require(StorageCredentials.ACCESS_KEY_ID).startsWith("ASIA")); assertTrue(!credentials.require(StorageCredentials.SESSION_TOKEN).isBlank()); + assertEquals("us-west-2", credentials.require(StorageCredentials.REGION)); } @Test diff --git a/server/pom.xml b/server/pom.xml index 6021f879..07d8cacd 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -77,6 +77,8 @@ 4.0.0 3.4.3 2.35.4 + 12.27.0 + 2.30.1 hadoop3-2.2.30 3.17.0 @@ -161,6 +163,31 @@ logback-classic ${logback.version} + + com.google.code.gson + gson + 2.10.1 + + + com.azure + azure-core + 1.51.0 + + + com.azure + azure-json + 1.2.0 + + + com.azure + azure-xml + 1.0.0 + + + io.projectreactor + reactor-core + 3.4.38 +