HADOOP-16445. Allow separate custom signing algorithms for S3 and DDB (apache#1332)

sidseth · deepakdamri · commit 15d78f48c955 · 2025-01-21T15:59:45.000+05:30
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
@@ -55,7 +55,8 @@ public AmazonS3 createS3Client(URI name,
       final AWSCredentialsProvider credentials,
       final String userAgentSuffix) throws IOException {
     Configuration conf = getConf();
-    final ClientConfiguration awsConf = S3AUtils.createAwsConf(getConf(), bucket);
+    final ClientConfiguration awsConf = S3AUtils
+        .createAwsConf(getConf(), bucket, Constants.AWS_SERVICE_IDENTIFIER_S3);
     if (!StringUtils.isEmpty(userAgentSuffix)) {
       awsConf.setUserAgentSuffix(userAgentSuffix);
     }
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/SignerManager.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/SignerManager.java
@@ -0,0 +1,99 @@
+/*
+ * 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;
+
+import com.amazonaws.auth.Signer;
+import com.amazonaws.auth.SignerFactory;
+import java.io.Closeable;
+import java.io.IOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import static org.apache.hadoop.fs.s3a.Constants.CUSTOM_SIGNERS;
+
+/**
+ * Class to handle custom signers.
+ */
+public class SignerManager implements Closeable {
+
+  private static final Logger LOG = LoggerFactory
+      .getLogger(SignerManager.class);
+
+
+  public SignerManager() {
+  }
+
+  /**
+   * Initialize custom signers and register them with the AWS SDK.
+   *
+   * @param conf Hadoop configuration
+   */
+  public void initCustomSigners(Configuration conf) {
+    String[] customSigners = conf.getTrimmedStrings(CUSTOM_SIGNERS);
+    if (customSigners == null || customSigners.length == 0) {
+      // No custom signers specified, nothing to do.
+      LOG.debug("No custom signers specified");
+      return;
+    }
+
+    for (String customSigner : customSigners) {
+      String[] parts = customSigner.split(":");
+      if (parts.length != 2) {
+        String message =
+            "Invalid format (Expected name:SignerClass) for CustomSigner: ["
+                + customSigner
+                + "]";
+        LOG.error(message);
+        throw new IllegalArgumentException(message);
+      }
+      maybeRegisterSigner(parts[0], parts[1], conf);
+    }
+  }
+
+  /*
+   * Make sure the signer class is registered once with the AWS SDK
+   */
+  private static void maybeRegisterSigner(String signerName,
+      String signerClassName, Configuration conf) {
+    try {
+      SignerFactory.getSignerByTypeAndService(signerName, null);
+    } catch (IllegalArgumentException e) {
+      // Signer is not registered with the AWS SDK.
+      // Load the class and register the signer.
+      Class<? extends Signer> clazz = null;
+      try {
+        clazz = (Class<? extends Signer>) conf.getClassByName(signerClassName);
+      } catch (ClassNotFoundException cnfe) {
+        throw new RuntimeException(String
+            .format("Signer class [%s] not found for signer [%s]",
+                signerClassName, signerName), cnfe);
+      }
+      LOG.debug("Registering Custom Signer - [{}->{}]", signerName,
+          clazz.getName());
+      synchronized (SignerManager.class) {
+        SignerFactory.registerSigner(signerName, clazz);
+      }
+    }
+  }
+
+  @Override
+  public void close() throws IOException {
+  }
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java
@@ -31,12 +31,14 @@
 import com.amazonaws.services.securitytoken.model.Credentials;
 import com.amazonaws.services.securitytoken.model.GetSessionTokenRequest;
 import com.google.common.base.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.Constants;
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.S3AUtils;
@@ -73,7 +75,8 @@ public static AWSSecurityTokenServiceClientBuilder builder(
       final Configuration conf,
       final String bucket,
       final AWSCredentialsProvider credentials) throws IOException {
-    final ClientConfiguration awsConf = S3AUtils.createAwsConf(conf, bucket);
+    final ClientConfiguration awsConf = S3AUtils.createAwsConf(conf, bucket,
+        Constants.AWS_SERVICE_IDENTIFIER_STS);
     String endpoint = conf.getTrimmed(DELEGATION_TOKEN_ENDPOINT,
         DEFAULT_DELEGATION_TOKEN_ENDPOINT);
     String region = conf.getTrimmed(DELEGATION_TOKEN_REGION,
@@ -99,7 +102,8 @@ public static AWSSecurityTokenServiceClientBuilder builder(
       final AWSCredentialsProvider credentials,
       final String stsEndpoint,
       final String stsRegion) throws IOException {
-    final ClientConfiguration awsConf = S3AUtils.createAwsConf(conf, bucket);
+    final ClientConfiguration awsConf = S3AUtils.createAwsConf(conf, bucket,
+        Constants.AWS_SERVICE_IDENTIFIER_STS);
     return builder(credentials, awsConf, stsEndpoint, stsRegion);
   }
 
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/SessionTokenBinding.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/SessionTokenBinding.java
@@ -36,6 +36,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.AWSCredentialProviderList;
+import org.apache.hadoop.fs.s3a.Constants;
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.S3ARetryPolicy;
@@ -301,7 +302,8 @@ private synchronized Optional<STSClientFactory.STSClient> maybeInitSTS()
 
       invoker = new Invoker(new S3ARetryPolicy(conf), LOG_EVENT);
       ClientConfiguration awsConf =
-          S3AUtils.createAwsConf(conf, uri.getHost());
+          S3AUtils.createAwsConf(conf, uri.getHost(),
+              Constants.AWS_SERVICE_IDENTIFIER_STS);
       AWSSecurityTokenService tokenService =
           STSClientFactory.builder(parentAuthChain,
               awsConf,
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBClientFactory.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBClientFactory.java
@@ -34,6 +34,7 @@
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.s3a.Constants;
 import org.apache.hadoop.fs.s3a.S3AUtils;
 
 import static org.apache.hadoop.fs.s3a.Constants.S3GUARD_DDB_REGION_KEY;
@@ -80,7 +81,8 @@ public AmazonDynamoDB createDynamoDBClient(String defaultRegion,
           "Should have been configured before usage");
 
       final Configuration conf = getConf();
-      final ClientConfiguration awsConf = S3AUtils.createAwsConf(conf, bucket);
+      final ClientConfiguration awsConf = S3AUtils
+          .createAwsConf(conf, bucket, Constants.AWS_SERVICE_IDENTIFIER_DDB);
 
       final String region = getRegion(conf, defaultRegion);
       LOG.debug("Creating DynamoDB client in region {}", region);
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java
@@ -22,6 +22,7 @@
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.S3ClientOptions;
 
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -30,12 +31,14 @@
 import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.apache.hadoop.fs.s3native.S3xLoginHelper;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.io.File;
 import java.net.URI;
 import java.security.PrivilegedExceptionAction;
@@ -612,4 +615,69 @@ public void testSecurityCredentialPropagationEndToEnd() throws Exception {
         "override,base");
   }
 
+  @Test(timeout = 10_000L)
+  public void testS3SpecificSignerOverride() throws IOException {
+    ClientConfiguration clientConfiguration = null;
+    Configuration config;
+
+    String signerOverride = "testSigner";
+    String s3SignerOverride = "testS3Signer";
+
+    // Default SIGNING_ALGORITHM, overridden for S3 only
+    config = new Configuration();
+    config.set(SIGNING_ALGORITHM_S3, s3SignerOverride);
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_S3);
+    Assert.assertEquals(s3SignerOverride,
+        clientConfiguration.getSignerOverride());
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_DDB);
+    Assert.assertNull(clientConfiguration.getSignerOverride());
+
+    // Configured base SIGNING_ALGORITHM, overridden for S3 only
+    config = new Configuration();
+    config.set(SIGNING_ALGORITHM, signerOverride);
+    config.set(SIGNING_ALGORITHM_S3, s3SignerOverride);
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_S3);
+    Assert.assertEquals(s3SignerOverride,
+        clientConfiguration.getSignerOverride());
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_DDB);
+    Assert
+        .assertEquals(signerOverride, clientConfiguration.getSignerOverride());
+  }
+
+  @Test(timeout = 10_000L)
+  public void testDdbSpecificSignerOverride() throws IOException {
+    ClientConfiguration clientConfiguration = null;
+    Configuration config;
+
+    String signerOverride = "testSigner";
+    String ddbSignerOverride = "testDdbSigner";
+
+    // Default SIGNING_ALGORITHM, overridden for S3
+    config = new Configuration();
+    config.set(SIGNING_ALGORITHM_DDB, ddbSignerOverride);
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_DDB);
+    Assert.assertEquals(ddbSignerOverride,
+        clientConfiguration.getSignerOverride());
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_S3);
+    Assert.assertNull(clientConfiguration.getSignerOverride());
+
+    // Configured base SIGNING_ALGORITHM, overridden for S3
+    config = new Configuration();
+    config.set(SIGNING_ALGORITHM, signerOverride);
+    config.set(SIGNING_ALGORITHM_DDB, ddbSignerOverride);
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_DDB);
+    Assert.assertEquals(ddbSignerOverride,
+        clientConfiguration.getSignerOverride());
+    clientConfiguration = S3AUtils
+        .createAwsConf(config, "dontcare", AWS_SERVICE_IDENTIFIER_S3);
+    Assert
+        .assertEquals(signerOverride, clientConfiguration.getSignerOverride());
+  }
 }
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ATemporaryCredentials.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ATemporaryCredentials.java
@@ -368,7 +368,7 @@ public <E extends Exception> E expectedSessionRequestFailure(
         DurationInfo ignored = new DurationInfo(LOG, "requesting credentials")) {
       Configuration conf = new Configuration(getContract().getConf());
       ClientConfiguration awsConf =
-          S3AUtils.createAwsConf(conf, null);
+          S3AUtils.createAwsConf(conf, null, AWS_SERVICE_IDENTIFIER_STS);
       return intercept(clazz, exceptionText,
           () -> {
             AWSSecurityTokenService tokenService =
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
@@ -619,7 +619,7 @@ public static MarshalledCredentials requestSessionCredentials(
     MarshalledCredentials sc = MarshalledCredentialBinding
         .requestSessionCredentials(
           buildAwsCredentialsProvider(conf),
-          S3AUtils.createAwsConf(conf, bucket),
+          S3AUtils.createAwsConf(conf, bucket, AWS_SERVICE_IDENTIFIER_STS),
           conf.getTrimmed(ASSUMED_ROLE_STS_ENDPOINT,
               DEFAULT_ASSUMED_ROLE_STS_ENDPOINT),
           conf.getTrimmed(ASSUMED_ROLE_STS_ENDPOINT_REGION,
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSignerManager.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSignerManager.java
