Delta Sharing protocol: sign recipient file URLs for cloud storage

server/opensharing-server-core/pom.xml

@@ -76,7 +76,68 @@
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>s3</artifactId>
       <version>${aws-sdk.version}</version>
-      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.azure</groupId>
+      <artifactId>azure-storage-blob</artifactId>
+      <version>${azure-storage-blob.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.azure</groupId>
+          <artifactId>azure-core-http-netty</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.dataformat</groupId>
+          <artifactId>jackson-dataformat-xml</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-storage</artifactId>
+      <version>${google-cloud-storage.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.gson</groupId>
+          <artifactId>gson</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.google.cloud</groupId>
+          <artifactId>google-cloud-core-grpc</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.google.api.grpc</groupId>
+          <artifactId>grpc-google-cloud-storage-v2</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.google.api.grpc</groupId>
+          <artifactId>gapic-google-cloud-storage-v2</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>io.grpc</groupId>
+          <artifactId>grpc-netty-shaded</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>io.grpc</groupId>
+          <artifactId>grpc-alts</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>io.grpc</groupId>
+          <artifactId>grpc-auth</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>io.grpc</groupId>
+          <artifactId>grpc-inprocess</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.google.api</groupId>
+          <artifactId>gax-grpc</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
     </dependency>
     <dependency>
       <groupId>software.amazon.awssdk</groupId>

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) {}
 }

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(

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<String> 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;
+  }
+}

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<String> 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);
+    }
+  }
+}

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<String> 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));
+  }
+}

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) {}

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) {}
+}