HADOOP-19004. S3A: Support Authentication through HttpSigner API

Move to the new auth flow based signers for aws. * Implement a new Signer Initialization Chain
* Add a new instantiation method
* Add a new test
* Fix Reflection Code for SignerInitialization

(+steve changes)
* move instantiation code into SignerFactory
* CustomHttpSigner goes into production code for broader testing
* some tuning of property values and names of constants/methods.

Change-Id: I8f34af4bd958d631bcb03d7c073ed1da01cb8205

Author: HarshitGupta

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

@@ -1525,4 +1525,19 @@ private Constants() {
   public static final String STORE_CAPABILITY_AWS_V2 =
       "fs.s3a.capability.aws.v2";
 
+  /**
+   * Flag to switch to a v2 SDK HTTP signer. Value {@value}.
+   */
+  public static final String HTTP_SIGNER_ENABLED = "fs.s3a.http.signer.enabled";
+
+  /**
+   * Default value of {@link #HTTP_SIGNER_ENABLED}: {@value}.
+   */
+  public static final boolean HTTP_SIGNER_ENABLED_DEFAULT = false;
+
+  /**
+   * Classname of the http signer to use when {@link #HTTP_SIGNER_ENABLED}
+   * is true: {@value}.
+   */
+  public static String HTTP_SIGNER_CLASS_NAME = "fs.s3a.http.signer.class";
 }

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

@@ -32,7 +32,10 @@
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.http.auth.spi.scheme.AuthScheme;
+import software.amazon.awssdk.http.auth.spi.signer.HttpSigner;
 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3BaseClientBuilder;
@@ -52,10 +55,15 @@
 import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
 import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_DEFAULT_REGION;
 import static org.apache.hadoop.fs.s3a.Constants.CENTRAL_ENDPOINT;
+import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_CLASS_NAME;
+import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_ENABLED;
+import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_ENABLED_DEFAULT;
+import static org.apache.hadoop.fs.s3a.auth.SignerFactory.createHttpSigner;
 import static org.apache.hadoop.fs.s3a.impl.AWSHeaders.REQUESTER_PAYS_HEADER;
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SECURE_CONNECTIONS;
 import static org.apache.hadoop.fs.s3a.Constants.SECURE_CONNECTIONS;
 import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_S3;
+import static org.apache.hadoop.fs.s3a.impl.InternalConstants.AUTH_SCHEME_AWS_SIGV_4;
 
 
 /**
@@ -163,17 +171,26 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> Build
             .pathStyleAccessEnabled(parameters.isPathStyleAccess())
             .build();
 
-    return builder
+    S3BaseClientBuilder s3BaseClientBuilder = builder
         .overrideConfiguration(createClientOverrideConfiguration(parameters, conf))
         .credentialsProvider(parameters.getCredentialSet())
         .serviceConfiguration(serviceConfiguration);
+
+    if (conf.getBoolean(HTTP_SIGNER_ENABLED, HTTP_SIGNER_ENABLED_DEFAULT)) {
+      // use an http signer through an AuthScheme
+      final AuthScheme<AwsCredentialsIdentity> signer =
+          createHttpSigner(conf, AUTH_SCHEME_AWS_SIGV_4, HTTP_SIGNER_CLASS_NAME);
+      builder.putAuthScheme(signer);
+    }
+    return (BuilderT) s3BaseClientBuilder;
   }
 
   /**
    * Create an override configuration for an S3 client.
    * @param parameters parameter object
    * @param conf configuration object
    * @throws IOException any IOE raised, or translated exception
+   * @throws RuntimeException some failures creating an http signer
    * @return the override configuration
    */
   protected ClientOverrideConfiguration createClientOverrideConfiguration(

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

@@ -58,6 +58,8 @@
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CreateSessionRequest;
+import software.amazon.awssdk.services.s3.model.CreateSessionResponse;
 import software.amazon.awssdk.services.s3.model.GetBucketLocationRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
@@ -5751,6 +5753,10 @@ public StoreContext createStoreContext() {
         .build();
   }
 
+  public CreateSessionResponse createSessionInternal(CreateSessionRequest createSessionRequest){
+    return this.s3Client.createSession(createSessionRequest);
+  }
+
   /**
    * Create a marker tools operations binding for this store.
    * Auditing:

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

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.auth;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner;
+import software.amazon.awssdk.http.auth.spi.signer.AsyncSignRequest;
+import software.amazon.awssdk.http.auth.spi.signer.AsyncSignedRequest;
+import software.amazon.awssdk.http.auth.spi.signer.HttpSigner;
+import software.amazon.awssdk.http.auth.spi.signer.SignRequest;
+import software.amazon.awssdk.http.auth.spi.signer.SignedRequest;
+import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
+
+/**
+ * Custom signer that delegates to the AWS V4 signer.
+ * Logs at TRACE the string value of any request.
+ * This is in the production code to support testing the signer plugin mechansim.
+ * To use
+ * <pre>
+ *   fs.s3a.http.signer.enabled = true
+ *   fs.s3a.http.signer.class = org.apache.hadoop.fs.s3a.auth.CustomHttpSigner
+ * <pre>
+ */
+public final class CustomHttpSigner implements HttpSigner<AwsCredentialsIdentity> {
+  private static final Logger LOG = LoggerFactory
+      .getLogger(CustomHttpSigner.class);
+
+  /**
+   * The delegate signer.
+   */
+  private final HttpSigner<AwsCredentialsIdentity> delegateSigner;
+
+  public CustomHttpSigner() {
+    delegateSigner = AwsV4HttpSigner.create();
+  }
+
+  @Override
+  public SignedRequest sign(SignRequest<? extends AwsCredentialsIdentity>
+      request) {
+    LOG.trace("Signing request:{}", request.request());
+    return delegateSigner.sign(request);
+  }
+
+  @Override
+  public CompletableFuture<AsyncSignedRequest> signAsync(final AsyncSignRequest<? extends AwsCredentialsIdentity> request) {
+    LOG.trace("Signing async request:{}", request.request());
+    return delegateSigner.signAsync(request);
+  }
+}

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

@@ -29,12 +29,20 @@
 import software.amazon.awssdk.auth.signer.AwsS3V4Signer;
 import software.amazon.awssdk.core.signer.NoOpSigner;
 import software.amazon.awssdk.core.signer.Signer;
+import software.amazon.awssdk.http.auth.spi.scheme.AuthScheme;
+import software.amazon.awssdk.http.auth.spi.signer.HttpSigner;
+import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
+import software.amazon.awssdk.identity.spi.IdentityProvider;
+import software.amazon.awssdk.identity.spi.IdentityProviders;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.S3AUtils;
 import org.apache.hadoop.fs.s3a.impl.InstantiationIOException;
 
+import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_CLASS_NAME;
 import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.unavailable;
 import static org.apache.hadoop.util.Preconditions.checkArgument;
+import static org.apache.hadoop.util.Preconditions.checkState;
 
 /**
  * Signer factory used to register and create signers.
@@ -119,4 +127,61 @@ public static Signer createSigner(String signerType, String configKey) throws IO
 
     return signer;
   }
+
+  /**
+   * Create an auth scheme instance from an ID and a signer.
+   * @param schemeId scheme id
+   * @param signer signer
+   * @return the auth scheme
+   */
+  public static AuthScheme<AwsCredentialsIdentity> createAuthScheme(String schemeId,
+      HttpSigner<AwsCredentialsIdentity> signer) {
+    return new AuthScheme<AwsCredentialsIdentity>() {
+      @Override
+      public String schemeId() {
+        return schemeId;
+      }
+      @Override
+      public IdentityProvider<AwsCredentialsIdentity> identityProvider(IdentityProviders providers) {
+        return providers.identityProvider(AwsCredentialsIdentity.class);
+      }
+      @Override
+      public HttpSigner<AwsCredentialsIdentity> signer() {
+        return signer;
+      }
+    };
+  }
+
+  /**
+   * Create an auth scheme by looking up the signer class in the configuration,
+   * loading and instantiating it.
+   * @param conf configuration
+   * @param scheme scheme to bond to
+   * @param configKey configuration key
+   * @return the auth scheme
+   * @throws InstantiationIOException failure to instantiate
+   * @throws IllegalStateException if the signer class is not defined
+   * @throws RuntimeException other configuration problems
+   */
+  public static AuthScheme<AwsCredentialsIdentity> createHttpSigner(
+      Configuration conf, String scheme, String configKey) throws IOException {
+
+    final Class<? extends HttpSigner> clazz = conf.getClass(HTTP_SIGNER_CLASS_NAME,
+        null, HttpSigner.class);
+    checkState(clazz != null, "No http signer class defined in %s", configKey);
+    LOG.debug("Creating http signer {} from {}", clazz, configKey);
+    try {
+      return createAuthScheme(scheme, clazz.newInstance());
+
+    } catch (InstantiationException | IllegalAccessException e) {
+      throw new InstantiationIOException(
+          InstantiationIOException.Kind.InstantiationFailure,
+          null,
+          clazz.getName(),
+          HTTP_SIGNER_CLASS_NAME,
+          e.toString(),
+          e);
+    }
+  }
+
 }

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

@@ -105,6 +105,7 @@ private AWSClientConfig() {
    * @param awsServiceIdentifier service
    * @return the builder inited with signer, timeouts and UA.
    * @throws IOException failure.
+   * @throws RuntimeException some failures creating an http signer
    */
   public static ClientOverrideConfiguration.Builder createClientConfigBuilder(Configuration conf,
       String awsServiceIdentifier) throws IOException {

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

@@ -286,4 +286,10 @@ private InternalConstants() {
           FS_S3A_CREATE_PERFORMANCE_ENABLED,
           DIRECTORY_OPERATIONS_PURGE_UPLOADS,
           ENABLE_MULTI_DELETE));
+
+  /**
+   * AWS V4 Auth Scheme to use when creating signers.
+   */
+  public static final String AUTH_SCHEME_AWS_SIGV_4 = "aws.auth#sigv4";
+
 }