HADOOP-16477. S3 delegation token tests fail if fs.s3a.encryption.key set

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

@@ -457,7 +457,7 @@ public String toString() {
   @Retries.RetryTranslated
   public PutObjectResult putObject(PutObjectRequest putObjectRequest)
       throws IOException {
-    return retry("put",
+    return retry("Writing Object",
         putObjectRequest.getKey(), true,
         () -> owner.putObjectDirect(putObjectRequest));
   }
@@ -472,7 +472,7 @@ public PutObjectResult putObject(PutObjectRequest putObjectRequest)
   public UploadResult uploadObject(PutObjectRequest putObjectRequest)
       throws IOException {
     // no retry; rely on xfer manager logic
-    return retry("put",
+    return retry("Writing Object",
         putObjectRequest.getKey(), true,
         () -> owner.executePut(putObjectRequest, null));
   }

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

@@ -55,7 +55,7 @@ private RolePolicies() {
   /**
    * Arn for all KMS keys: {@value}.
    */
-  public static final String KMS_ALL_KEYS = "arn:aws:kms:*";
+  public static final String KMS_ALL_KEYS = "*";
 
   /**
    * This is used by S3 to generate a per-object encryption key and
@@ -68,7 +68,7 @@ private RolePolicies() {
    * Actions needed to read and write SSE-KMS data.
    */
   private static final String[] KMS_KEY_RW =
-      new String[]{KMS_DECRYPT, KMS_GENERATE_DATA_KEY};
+      new String[]{KMS_DECRYPT, KMS_GENERATE_DATA_KEY, KMS_ENCRYPT};
 
   /**
    * Actions needed to read SSE-KMS data.
@@ -81,7 +81,7 @@ private RolePolicies() {
    * SSE-KMS.
    */
   public static final Statement STATEMENT_ALLOW_SSE_KMS_RW =
-      statement(true, KMS_ALL_KEYS, KMS_KEY_RW);
+      statement(true, KMS_ALL_KEYS, KMS_ALL_OPERATIONS);
 
   /**
    * Statement to allow read access to KMS keys, so the ability

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/assumed_roles.md

@@ -46,9 +46,25 @@ have access to the appropriate KMS keys.
 Trying to learn how IAM Assumed Roles work by debugging stack traces from
 the S3A client is "suboptimal".
 
-### <a name="how_it_works"></a> How the S3A connector support IAM Assumed Roles.
+### <a name="how_it_works"></a> How the S3A connector supports IAM Assumed Roles.
 
-To use assumed roles, the client must be configured to use the
+
+The S3A connector support IAM Assumed Roles in two ways:
+
+1. Using the full credentials on the client to request credentials for a specific
+    role -credentials which are then used for all the store operations.
+    This can be used to verify that a specific role has the access permissions
+    you need, or to "su" into a role which has permissions that's the full
+    accounts does not directly qualify for -such as access to a KMS key.
+2. Using the full credentials to request role credentials which are then
+    propagated into a launched application as delegation tokens.
+    This extends the previous use as it allows the jobs to be submitted to a
+    shared cluster with the permissions of the requested role, rather than
+    those of the VMs/Containers of the deployed cluster.
+
+For Delegation Token integration, see (Delegation Tokens)[delegation_tokens.html]
+
+To for Assumed Role authentication, the client must be configured to use the
 *Assumed Role Credential Provider*, `org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider`,
 in the configuration option `fs.s3a.aws.credentials.provider`.
 
@@ -298,7 +314,7 @@ Without these permissions, tables cannot be created, destroyed or have their IO
 changed through the `s3guard set-capacity` call.
 The `dynamodb:Scan` permission is needed for `s3guard prune`
 
-The `dynamodb:CreateTable` permission is needed by a client it tries to
+The `dynamodb:CreateTable` permission is needed by a client when it tries to
 create the DynamoDB table on startup, that is
 `fs.s3a.s3guard.ddb.table.create` is `true` and the table does not already exist.
 
@@ -758,14 +774,51 @@ Make sure that all the read and write permissions are allowed for any bucket/pat
 to which data is being written to, and read permissions for all
 buckets read from.
 
+### <a name="access_denied_kms"></a> `AccessDeniedException` When working with KMS-encrypted data
+
 If the bucket is using SSE-KMS to encrypt data:
 
 1. The caller must have the `kms:Decrypt` permission to read the data.
-1. The caller needs `kms:Decrypt` and `kms:GenerateDataKey`.
+1. The caller needs `kms:Decrypt` and `kms:GenerateDataKey` to write data.
 
 Without permissions, the request fails *and there is no explicit message indicating
 that this is an encryption-key issue*.
 
+This problem is most obvious when you fail when writing data in a "Writing Object" operation.
+
+If the client does have write access to the bucket, verify that the caller has
+`kms:GenerateDataKey` permissions for the encryption key in use.
+
+```
+java.nio.file.AccessDeniedException: test/testDTFileSystemClient: Writing Object on test/testDTFileSystemClient:
+  com.amazonaws.services.s3.model.AmazonS3Exception: Access Denied (Service: Amazon S3; Status Code: 403; 
+  Error Code: AccessDenied; Request ID: E86544FF1D029857)
+
+    at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:243)
+    at org.apache.hadoop.fs.s3a.Invoker.once(Invoker.java:111)
+    at org.apache.hadoop.fs.s3a.Invoker.lambda$retry$4(Invoker.java:314)
+    at org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:406)
+    at org.apache.hadoop.fs.s3a.Invoker.retry(Invoker.java:310)
+    at org.apache.hadoop.fs.s3a.Invoker.retry(Invoker.java:285)
+    at org.apache.hadoop.fs.s3a.WriteOperationHelper.retry(WriteOperationHelper.java:150)
+    at org.apache.hadoop.fs.s3a.WriteOperationHelper.putObject(WriteOperationHelper.java:460)
+    at org.apache.hadoop.fs.s3a.S3ABlockOutputStream.lambda$putObject$0(S3ABlockOutputStream.java:438)
+    at org.apache.hadoop.util.SemaphoredDelegatingExecutor$CallableWithPermitRelease.call(SemaphoredDelegatingExecutor.java:219)
+    at org.apache.hadoop.util.SemaphoredDelegatingExecutor$CallableWithPermitRelease.call(SemaphoredDelegatingExecutor.java:219)
+    at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:125)
+    at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:57)
+    at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:78)
+    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
+    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
+    at java.lang.Thread.run(Thread.java:748)
+Caused by:  com.amazonaws.services.s3.model.AmazonS3Exception: Access Denied (Service: Amazon S3; Status Code: 403;
+            Error Code: AccessDenied; Request ID: E86544FF1D029857)
+```
+
+Note: the ability to read encrypted data in the store does not guarantee that the caller can encrypt new data.
+It is a separate permission.
+
+
 ### <a name="dynamodb_exception"></a> `AccessDeniedException` + `AmazonDynamoDBException`
 
 ```

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java

@@ -18,18 +18,25 @@
 
 package org.apache.hadoop.fs.s3a;
 
+import java.io.IOException;
+
 import com.amazonaws.services.s3.model.ObjectMetadata;
+import org.junit.Test;
+
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.net.util.Base64;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
-import org.junit.Test;
-
-import java.io.IOException;
+import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
 
 import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.*;
+import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_KEY;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfEncryptionTestsDisabled;
+import static org.apache.hadoop.fs.s3a.S3AUtils.getEncryptionAlgorithm;
 
 /**
  * Test whether or not encryption works by turning it on. Some checks
@@ -38,11 +45,18 @@
  */
 public abstract class AbstractTestS3AEncryption extends AbstractS3ATestBase {
 
+  protected static final String AWS_KMS_SSE_ALGORITHM = "aws:kms";
+
+  protected static final String SSE_C_ALGORITHM = "AES256";
+
   @Override
   protected Configuration createConfiguration() {
     Configuration conf = super.createConfiguration();
     S3ATestUtils.disableFilesystemCaching(conf);
-    conf.set(Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM,
+    removeBaseAndBucketOverrides(conf,
+        SERVER_SIDE_ENCRYPTION_ALGORITHM,
+        SERVER_SIDE_ENCRYPTION_KEY);
+    conf.set(SERVER_SIDE_ENCRYPTION_ALGORITHM,
             getSSEAlgorithm().getMethod());
     return conf;
   }
@@ -51,19 +65,46 @@ protected Configuration createConfiguration() {
       0, 1, 2, 3, 4, 5, 254, 255, 256, 257, 2 ^ 12 - 1
   };
 
+  protected void requireEncryptedFileSystem() {
+    skipIfEncryptionTestsDisabled(getFileSystem().getConf());
+  }
+
+  @Override
+  public void setup() throws Exception {
+    super.setup();
+    requireEncryptedFileSystem();
+  }
+
+  /**
+   * This examines how encryption settings propagate better.
+   * If the settings are actually in a JCEKS file, then the
+   * test override will fail; this is here to help debug the problem.
+   */
+  @Test
+  public void testEncryptionSettingPropagation() throws Throwable {
+    S3AFileSystem fs = getFileSystem();
+    S3AEncryptionMethods algorithm = getEncryptionAlgorithm(
+        fs.getBucket(), fs.getConf());
+    assertEquals("Configuration has wrong encryption algorithm",
+        getSSEAlgorithm(), algorithm);
+  }
+
   @Test
   public void testEncryption() throws Throwable {
+    requireEncryptedFileSystem();
+    validateEncrytionSecrets(getFileSystem().getEncryptionSecrets());
     for (int size: SIZES) {
       validateEncryptionForFilesize(size);
     }
   }
 
   @Test
   public void testEncryptionOverRename() throws Throwable {
-    skipIfEncryptionTestsDisabled(getConfiguration());
     Path src = path(createFilename(1024));
     byte[] data = dataset(1024, 'a', 'z');
     S3AFileSystem fs = getFileSystem();
+    EncryptionSecrets secrets = fs.getEncryptionSecrets();
+    validateEncrytionSecrets(secrets);
     writeDataset(fs, src, data, data.length, 1024 * 1024, true);
     ContractTestUtils.verifyFileContents(fs, src, data);
     Path dest = path(src.getName() + "-copy");
@@ -72,8 +113,19 @@ public void testEncryptionOverRename() throws Throwable {
     assertEncrypted(dest);
   }
 
+  /**
+   * Verify that the filesystem encryption secrets match expected.
+   * This makes sure that the settings have propagated properly.
+   * @param secrets encryption secrets of the filesystem.
+   */
+  protected void validateEncrytionSecrets(final EncryptionSecrets secrets) {
+    assertNotNull("No encryption secrets for filesystem", secrets);
+    S3AEncryptionMethods sseAlgorithm = getSSEAlgorithm();
+    assertEquals("Filesystem has wrong encryption algorithm",
+        sseAlgorithm, secrets.getEncryptionMethod());
+  }
+
   protected void validateEncryptionForFilesize(int len) throws IOException {
-    skipIfEncryptionTestsDisabled(getConfiguration());
     describe("Create an encrypted file of size " + len);
     String src = createFilename(len);
     Path path = writeThenReadFile(src, len);
@@ -98,15 +150,17 @@ protected void assertEncrypted(Path path) throws IOException {
     ObjectMetadata md = getFileSystem().getObjectMetadata(path);
     switch(getSSEAlgorithm()) {
     case SSE_C:
-      assertEquals("AES256", md.getSSECustomerAlgorithm());
+      assertNull("Metadata algorithm should have been null",
+          md.getSSEAlgorithm());
+      assertEquals("Wrong SSE-C algorithm", SSE_C_ALGORITHM, md.getSSECustomerAlgorithm());
       String md5Key = convertKeyToMd5();
-      assertEquals(md5Key, md.getSSECustomerKeyMd5());
+      assertEquals("getSSECustomerKeyMd5() wrong", md5Key, md.getSSECustomerKeyMd5());
       break;
     case SSE_KMS:
-      assertEquals("aws:kms", md.getSSEAlgorithm());
+      assertEquals(AWS_KMS_SSE_ALGORITHM, md.getSSEAlgorithm());
       //S3 will return full arn of the key, so specify global arn in properties
       assertEquals(this.getConfiguration().
-          getTrimmed(Constants.SERVER_SIDE_ENCRYPTION_KEY),
+          getTrimmed(SERVER_SIDE_ENCRYPTION_KEY),
           md.getSSEAwsKmsKeyId());
       break;
     default:
@@ -123,8 +177,8 @@ protected void assertEncrypted(Path path) throws IOException {
    * key
    */
   private String convertKeyToMd5() {
-    String base64Key = getConfiguration().getTrimmed(
-        Constants.SERVER_SIDE_ENCRYPTION_KEY
+    String base64Key = getFileSystem().getConf().getTrimmed(
+        SERVER_SIDE_ENCRYPTION_KEY
     );
     byte[] key = Base64.decodeBase64(base64Key);
     byte[] md5 =  DigestUtils.md5(key);

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAWSCredentialsProvider.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.nio.file.AccessDeniedException;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -38,9 +39,10 @@
 import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.fs.s3a.Constants.*;
-import static org.apache.hadoop.fs.s3a.S3ATestConstants.*;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.getCSVTestPath;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.S3AUtils.*;
+import static org.apache.hadoop.fs.s3a.auth.delegation.DelegationConstants.DELEGATION_TOKEN_BINDING;
 import static org.junit.Assert.*;
 
 /**
@@ -51,11 +53,11 @@ public class ITestS3AAWSCredentialsProvider {
       LoggerFactory.getLogger(ITestS3AAWSCredentialsProvider.class);
 
   @Rule
-  public Timeout testTimeout = new Timeout(1 * 60 * 1000);
+  public Timeout testTimeout = new Timeout(60_1000, TimeUnit.MILLISECONDS);
 
   @Test
   public void testBadConfiguration() throws IOException {
-    Configuration conf = new Configuration();
+    Configuration conf = createConf();
     conf.set(AWS_CREDENTIALS_PROVIDER, "no.such.class");
     try {
       createFailingFS(conf);
@@ -93,7 +95,7 @@ public void refresh() {
 
   @Test
   public void testBadCredentialsConstructor() throws Exception {
-    Configuration conf = new Configuration();
+    Configuration conf = createConf();
     conf.set(AWS_CREDENTIALS_PROVIDER,
         BadCredentialsProviderConstructor.class.getName());
     try {
@@ -103,6 +105,14 @@ public void testBadCredentialsConstructor() throws Exception {
     }
   }
 
+  protected Configuration createConf() {
+    Configuration conf = new Configuration();
+    removeBaseAndBucketOverrides(conf,
+        DELEGATION_TOKEN_BINDING,
+        AWS_CREDENTIALS_PROVIDER);
+    return conf;
+  }
+
   /**
    * Create a filesystem, expect it to fail by raising an IOException.
    * Raises an assertion exception if in fact the FS does get instantiated.