HADOOP-17922. move to fs.s3a.encryption.algorithm - JCEKS integration (#3466)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeysPublic.java

@@ -969,6 +969,8 @@ public class CommonConfigurationKeysPublic {
           "ssl.keystore.pass$",
           "fs.s3.*[Ss]ecret.?[Kk]ey",
           "fs.s3a.*.server-side-encryption.key",
+          "fs.s3a.encryption.algorithm",
+          "fs.s3a.encryption.key",
           "fs.azure\\.account.key.*",
           "credential$",
           "oauth.*secret",

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -667,6 +667,8 @@
       ssl.keystore.pass$
       fs.s3a.server-side-encryption.key
       fs.s3a.*.server-side-encryption.key
+      fs.s3a.encryption.algorithm
+      fs.s3a.encryption.key
       fs.s3a.secret.key
       fs.s3a.*.secret.key
       fs.s3a.session.key
@@ -1569,20 +1571,22 @@
 </property>
 
 <property>
-  <name>fs.s3a.server-side-encryption-algorithm</name>
-  <description>Specify a server-side encryption algorithm for s3a: file system.
-    Unset by default.  It supports the following values: 'AES256' (for SSE-S3),
-    'SSE-KMS' and 'SSE-C'.
+  <name>fs.s3a.encryption.algorithm</name>
+  <description>Specify a server-side encryption or client-side
+    encryption algorithm for s3a: file system. Unset by default. It supports the
+    following values: 'AES256' (for SSE-S3), 'SSE-KMS', 'SSE-C', and 'CSE-KMS'
   </description>
 </property>
 
 <property>
-  <name>fs.s3a.server-side-encryption.key</name>
-  <description>Specific encryption key to use if fs.s3a.server-side-encryption-algorithm
-    has been set to 'SSE-KMS' or 'SSE-C'. In the case of SSE-C, the value of this property
-    should be the Base64 encoded key. If you are using SSE-KMS and leave this property empty,
-    you'll be using your default's S3 KMS key, otherwise you should set this property to
-    the specific KMS key id.
+  <name>fs.s3a.encryption.key</name>
+  <description>Specific encryption key to use if fs.s3a.encryption.algorithm
+    has been set to 'SSE-KMS', 'SSE-C' or 'CSE-KMS'. In the case of SSE-C
+    , the value of this property should be the Base64 encoded key. If you are
+    using SSE-KMS and leave this property empty, you'll be using your default's
+    S3 KMS key, otherwise you should set this property to the specific KMS key
+    id. In case of 'CSE-KMS' this value needs to be the AWS-KMS Key ID
+    generated from AWS console.
   </description>
 </property>
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMultipartUploaderTest.java

@@ -89,6 +89,9 @@ public void setup() throws Exception {
 
     final FileSystem fs = getFileSystem();
     Path testPath = getContract().getTestPath();
+    Assume.assumeTrue("Multipart uploader is not supported",
+        fs.hasPathCapability(testPath,
+            CommonPathCapabilities.FS_MULTIPART_UPLOADER));
     uploader0 = fs.createMultipartUploader(testPath).build();
     uploader1 = fs.createMultipartUploader(testPath).build();
   }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -420,7 +420,13 @@ private Constants() {
       "fs.s3a.multipart.purge.age";
   public static final long DEFAULT_PURGE_EXISTING_MULTIPART_AGE = 86400;
 
-  // s3 server-side encryption, see S3AEncryptionMethods for valid options
+  /**
+   * s3 server-side encryption, see
+   * {@link S3AEncryptionMethods} for valid options.
+   *
+   * {@value}
+   */
+  @Deprecated
   public static final String SERVER_SIDE_ENCRYPTION_ALGORITHM =
       "fs.s3a.server-side-encryption-algorithm";
 
@@ -444,9 +450,32 @@ private Constants() {
    * May be set within a JCEKS file.
    * Value: "{@value}".
    */
+  @Deprecated
   public static final String SERVER_SIDE_ENCRYPTION_KEY =
       "fs.s3a.server-side-encryption.key";
 
+  /**
+   * Set S3-server side encryption(SSE) or S3-Client side encryption(CSE)
+   * algorithm. Check {@link S3AEncryptionMethods} for valid options.
+   * <br>
+   * value: {@value}
+   */
+  public static final String S3_ENCRYPTION_ALGORITHM =
+      "fs.s3a.encryption.algorithm";
+
+  /**
+   * Set S3-SSE or S3-CSE encryption Key if required.
+   * <br>
+   * <i>Note:</i>
+   *   <ul>
+   *     <li>In case of S3-CSE this value needs to be set for CSE to work.</li>
+   *     <li>In case of S3-SSE follow {@link #SERVER_SIDE_ENCRYPTION_KEY}</li>
+   *   </ul>
+   * value:{@value}
+   */
+  public static final String S3_ENCRYPTION_KEY =
+      "fs.s3a.encryption.key";
+
   /**
    * List of custom Signers. The signer class will be loaded, and the signer
    * name will be associated with this signer class in the S3 SDK.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java

@@ -25,13 +25,23 @@
 import com.amazonaws.SdkClientException;
 import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.handlers.RequestHandler2;
+import com.amazonaws.regions.RegionUtils;
 import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3Builder;
 import com.amazonaws.services.s3.AmazonS3Client;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.AmazonS3EncryptionClientV2Builder;
+import com.amazonaws.services.s3.AmazonS3EncryptionV2;
 import com.amazonaws.services.s3.S3ClientOptions;
 import com.amazonaws.services.s3.internal.ServiceUtils;
+import com.amazonaws.services.s3.model.CryptoConfigurationV2;
+import com.amazonaws.services.s3.model.CryptoMode;
+import com.amazonaws.services.s3.model.CryptoRangeGetMode;
+import com.amazonaws.services.s3.model.EncryptionMaterialsProvider;
+import com.amazonaws.services.s3.model.KMSEncryptionMaterialsProvider;
 import com.amazonaws.util.AwsHostNameUtils;
 import com.amazonaws.util.RuntimeHttpUtils;
+import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,6 +58,9 @@
 import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_CENTRAL_REGION;
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING;
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING_DEFAULT;
+import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_KEY;
+import static org.apache.hadoop.fs.s3a.S3AUtils.getEncryptionAlgorithm;
+import static org.apache.hadoop.fs.s3a.S3AUtils.getS3EncryptionKey;
 import static org.apache.hadoop.fs.s3a.S3AUtils.translateException;
 
 /**
@@ -81,6 +94,12 @@ public class DefaultS3ClientFactory extends Configured
       "S3A filesystem client is using"
           + " the SDK region resolution chain.";
 
+  /** Exactly once log to inform about ignoring the AWS-SDK Warnings for CSE. */
+  private static final LogExactlyOnce IGNORE_CSE_WARN = new LogExactlyOnce(LOG);
+
+  /** Bucket name. */
+  private String bucket;
+
   /**
    * Create the client by preparing the AwsConf configuration
    * and then invoking {@code buildAmazonS3Client()}.
@@ -90,9 +109,10 @@ public AmazonS3 createS3Client(
       final URI uri,
       final S3ClientCreationParameters parameters) throws IOException {
     Configuration conf = getConf();
+    bucket = uri.getHost();
     final ClientConfiguration awsConf = S3AUtils
         .createAwsConf(conf,
-            uri.getHost(),
+            bucket,
             Constants.AWS_SERVICE_IDENTIFIER_S3);
     // add any headers
     parameters.getHeaders().forEach((h, v) ->
@@ -111,16 +131,82 @@ public AmazonS3 createS3Client(
       awsConf.setUserAgentSuffix(parameters.getUserAgentSuffix());
     }
 
+    // Get the encryption method for this bucket.
+    S3AEncryptionMethods encryptionMethods =
+        getEncryptionAlgorithm(bucket, conf);
     try {
-      return buildAmazonS3Client(
-          awsConf,
-          parameters);
+      // If CSE is enabled then build a S3EncryptionClient.
+      if (S3AEncryptionMethods.CSE_KMS.getMethod()
+          .equals(encryptionMethods.getMethod())) {
+        return buildAmazonS3EncryptionClient(
+            awsConf,
+            parameters);
+      } else {
+        return buildAmazonS3Client(
+            awsConf,
+            parameters);
+      }
     } catch (SdkClientException e) {
       // SDK refused to build.
       throw translateException("creating AWS S3 client", uri.toString(), e);
     }
   }
 
+  /**
+   * Create an {@link AmazonS3} client of type
+   * {@link AmazonS3EncryptionV2} if CSE is enabled.
+   *
+   * @param awsConf    AWS configuration.
+   * @param parameters parameters.
+   *
+   * @return new AmazonS3 client.
+   * @throws IOException if lookupPassword() has any problem.
+   */
+  protected AmazonS3 buildAmazonS3EncryptionClient(
+      final ClientConfiguration awsConf,
+      final S3ClientCreationParameters parameters) throws IOException {
+
+    AmazonS3 client;
+    AmazonS3EncryptionClientV2Builder builder =
+        new AmazonS3EncryptionClientV2Builder();
+    Configuration conf = getConf();
+
+    // CSE-KMS Method
+    String kmsKeyId = getS3EncryptionKey(bucket, conf, true);
+    // Check if kmsKeyID is not null
+    Preconditions.checkArgument(!StringUtils.isBlank(kmsKeyId), "CSE-KMS "
+        + "method requires KMS key ID. Use " + S3_ENCRYPTION_KEY
+        + " property to set it. ");
+
+    EncryptionMaterialsProvider materialsProvider =
+        new KMSEncryptionMaterialsProvider(kmsKeyId);
+    builder.withEncryptionMaterialsProvider(materialsProvider);
+    //Configure basic params of a S3 builder.
+    configureBasicParams(builder, awsConf, parameters);
+
+    // Configuring endpoint.
+    AmazonS3EncryptionClientV2Builder.EndpointConfiguration epr
+        = createEndpointConfiguration(parameters.getEndpoint(),
+        awsConf, getConf().getTrimmed(AWS_REGION));
+    configureEndpoint(builder, epr);
+
+    // Create cryptoConfig.
+    CryptoConfigurationV2 cryptoConfigurationV2 =
+        new CryptoConfigurationV2(CryptoMode.AuthenticatedEncryption)
+            .withRangeGetMode(CryptoRangeGetMode.ALL);
+    if (epr != null) {
+      cryptoConfigurationV2
+          .withAwsKmsRegion(RegionUtils.getRegion(epr.getSigningRegion()));
+      LOG.debug("KMS region used: {}", cryptoConfigurationV2.getAwsKmsRegion());
+    }
+    builder.withCryptoConfiguration(cryptoConfigurationV2);
+    client = builder.build();
+    IGNORE_CSE_WARN.info("S3 client-side encryption enabled: Ignore S3-CSE "
+        + "Warnings.");
+
+    return client;
+  }
+
   /**
    * Use the Builder API to create an AWS S3 client.
    * <p>
@@ -137,41 +223,68 @@ protected AmazonS3 buildAmazonS3Client(
       final ClientConfiguration awsConf,
       final S3ClientCreationParameters parameters) {
     AmazonS3ClientBuilder b = AmazonS3Client.builder();
-    b.withCredentials(parameters.getCredentialSet());
-    b.withClientConfiguration(awsConf);
-    b.withPathStyleAccessEnabled(parameters.isPathStyleAccess());
+    configureBasicParams(b, awsConf, parameters);
+
+    // endpoint set up is a PITA
+    AwsClientBuilder.EndpointConfiguration epr
+        = createEndpointConfiguration(parameters.getEndpoint(),
+        awsConf, getConf().getTrimmed(AWS_REGION));
+    configureEndpoint(b, epr);
+    final AmazonS3 client = b.build();
+    return client;
+  }
+
+  /**
+   * A method to configure basic AmazonS3Builder parameters.
+   *
+   * @param builder    Instance of AmazonS3Builder used.
+   * @param awsConf    ClientConfiguration used.
+   * @param parameters Parameters used to set in the builder.
+   */
+  private void configureBasicParams(AmazonS3Builder builder,
+      ClientConfiguration awsConf, S3ClientCreationParameters parameters) {
+    builder.withCredentials(parameters.getCredentialSet());
+    builder.withClientConfiguration(awsConf);
+    builder.withPathStyleAccessEnabled(parameters.isPathStyleAccess());
 
     if (parameters.getMetrics() != null) {
-      b.withMetricsCollector(
+      builder.withMetricsCollector(
           new AwsStatisticsCollector(parameters.getMetrics()));
     }
     if (parameters.getRequestHandlers() != null) {
-      b.withRequestHandlers(
+      builder.withRequestHandlers(
           parameters.getRequestHandlers().toArray(new RequestHandler2[0]));
     }
     if (parameters.getMonitoringListener() != null) {
-      b.withMonitoringListener(parameters.getMonitoringListener());
+      builder.withMonitoringListener(parameters.getMonitoringListener());
     }
 
-    // endpoint set up is a PITA
-    AwsClientBuilder.EndpointConfiguration epr
-        = createEndpointConfiguration(parameters.getEndpoint(),
-        awsConf, getConf().getTrimmed(AWS_REGION));
+  }
+
+  /**
+   * A method to configure endpoint and Region for an AmazonS3Builder.
+   *
+   * @param builder Instance of AmazonS3Builder used.
+   * @param epr     EndpointConfiguration used to set in builder.
+   */
+  private void configureEndpoint(
+      AmazonS3Builder builder,
+      AmazonS3Builder.EndpointConfiguration epr) {
     if (epr != null) {
       // an endpoint binding was constructed: use it.
-      b.withEndpointConfiguration(epr);
+      builder.withEndpointConfiguration(epr);
     } else {
       // no idea what the endpoint is, so tell the SDK
       // to work it out at the cost of an extra HEAD request
-      b.withForceGlobalBucketAccessEnabled(true);
+      builder.withForceGlobalBucketAccessEnabled(true);
       // HADOOP-17771 force set the region so the build process doesn't halt.
       String region = getConf().getTrimmed(AWS_REGION, AWS_S3_CENTRAL_REGION);
       LOG.debug("fs.s3a.endpoint.region=\"{}\"", region);
       if (!region.isEmpty()) {
         // there's either an explicit region or we have fallen back
         // to the central one.
         LOG.debug("Using default endpoint; setting region to {}", region);
-        b.setRegion(region);
+        builder.setRegion(region);
       } else {
         // no region.
         // allow this if people really want it; it is OK to rely on this
@@ -180,8 +293,6 @@ protected AmazonS3 buildAmazonS3Client(
         LOG.debug(SDK_REGION_CHAIN_IN_USE);
       }
     }
-    final AmazonS3 client = b.build();
-    return client;
   }
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java

@@ -90,16 +90,18 @@
 public class Listing extends AbstractStoreOperation {
 
   private static final Logger LOG = S3AFileSystem.LOG;
+  private final boolean isCSEEnabled;
 
   static final FileStatusAcceptor ACCEPT_ALL_BUT_S3N =
       new AcceptAllButS3nDirs();
 
   private final ListingOperationCallbacks listingOperationCallbacks;
 
   public Listing(ListingOperationCallbacks listingOperationCallbacks,
-                 StoreContext storeContext) {
+      StoreContext storeContext) {
     super(storeContext);
     this.listingOperationCallbacks = listingOperationCallbacks;
+    this.isCSEEnabled = storeContext.isCSEEnabled();
   }
 
   /**
@@ -687,7 +689,7 @@ private boolean buildNextStatusBatch(S3ListResult objects) {
           S3AFileStatus status = createFileStatus(keyPath, summary,
                   listingOperationCallbacks.getDefaultBlockSize(keyPath),
                   getStoreContext().getUsername(),
-              summary.getETag(), null);
+              summary.getETag(), null, isCSEEnabled);
           LOG.debug("Adding: {}", status);
           stats.add(status);
           added++;
@@ -961,7 +963,7 @@ public AcceptFilesOnly(Path qualifiedPath) {
     public boolean accept(Path keyPath, S3ObjectSummary summary) {
       return !keyPath.equals(qualifiedPath)
           && !summary.getKey().endsWith(S3N_FOLDER_SUFFIX)
-          && !objectRepresentsDirectory(summary.getKey(), summary.getSize());
+          && !objectRepresentsDirectory(summary.getKey());
     }
 
     /**
@@ -1049,6 +1051,7 @@ public boolean accept(FileStatus status) {
     }
   }
 
+  @SuppressWarnings("unchecked")
   public static RemoteIterator<LocatedFileStatus> toLocatedFileStatusIterator(
       RemoteIterator<? extends LocatedFileStatus> iterator) {
     return (RemoteIterator < LocatedFileStatus >) iterator;