feat(S3): Add payload-signing-policy config (#12197)

Summary:
Some use-cases require the payload to be signed for additional security.
Add hive.s3.payload-signing-policy config to allow this.
Move away from deprecated S3Client API.

Pull Request resolved: #12197

Reviewed By: Yuhta

Differential Revision: D69254094

Pulled By: kgpai

fbshipit-source-id: b75c1e827db29520f3fabf4270af897860d68b2f

Author: majetideepak

velox/connectors/hive/storage_adapters/s3fs/S3Config.cpp

@@ -73,6 +73,8 @@ S3Config::S3Config(
       }
     }
   }
+  payloadSigningPolicy_ =
+      properties->get<std::string>(kS3PayloadSigningPolicy, "Never");
 }
 
 std::optional<std::string> S3Config::endpointRegion() const {

velox/connectors/hive/storage_adapters/s3fs/S3Config.h

@@ -47,6 +47,10 @@ class S3Config {
   /// Log granularity of AWS C++ SDK.
   static constexpr const char* kS3LogLevel = "hive.s3.log-level";
 
+  /// Payload signing policy.
+  static constexpr const char* kS3PayloadSigningPolicy =
+      "hive.s3.payload-signing-policy";
+
   /// S3FileSystem default identity.
   static constexpr const char* kDefaultS3Identity = "s3-default-identity";
 
@@ -220,8 +224,13 @@ class S3Config {
     return folly::to<bool>(value);
   }
 
+  std::string payloadSigningPolicy() const {
+    return payloadSigningPolicy_;
+  }
+
  private:
   std::unordered_map<Keys, std::optional<std::string>> config_;
+  std::string payloadSigningPolicy_;
 };
 
 } // namespace facebook::velox::filesystems

velox/connectors/hive/storage_adapters/s3fs/S3FileSystem.cpp

@@ -218,6 +218,21 @@ Aws::Utils::Logging::LogLevel inferS3LogLevel(std::string_view logLevel) {
   }
   return Aws::Utils::Logging::LogLevel::Fatal;
 }
+
+// Supported values are "Always", "RequestDependent", "Never"(default).
+Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy inferPayloadSign(
+    std::string sign) {
+  // Convert to upper case.
+  std::transform(sign.begin(), sign.end(), sign.begin(), [](unsigned char c) {
+    return std::toupper(c);
+  });
+  if (sign == "ALWAYS") {
+    return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always;
+  } else if (sign == "REQUESTDEPENDENT") {
+    return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::RequestDependent;
+  }
+  return Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never;
+}
 } // namespace
 
 class S3WriteFile::Impl {
@@ -529,7 +544,7 @@ class S3FileSystem::Impl {
  public:
   Impl(const S3Config& s3Config) {
     VELOX_CHECK(getAwsInstance()->isInitialized(), "S3 is not initialized");
-    Aws::Client::ClientConfiguration clientConfig;
+    Aws::S3::S3ClientConfiguration clientConfig;
     if (s3Config.endpoint().has_value()) {
       clientConfig.endpointOverride = s3Config.endpoint().value();
     }
@@ -586,13 +601,14 @@ class S3FileSystem::Impl {
       clientConfig.retryStrategy = retryStrategy.value();
     }
 
+    clientConfig.useVirtualAddressing = s3Config.useVirtualAddressing();
+    clientConfig.payloadSigningPolicy =
+        inferPayloadSign(s3Config.payloadSigningPolicy());
+
     auto credentialsProvider = getCredentialsProvider(s3Config);
 
     client_ = std::make_shared<Aws::S3::S3Client>(
-        credentialsProvider,
-        clientConfig,
-        Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,
-        s3Config.useVirtualAddressing());
+        credentialsProvider, nullptr /* endpointProvider */, clientConfig);
     ++fileSystemCount;
   }
 

velox/connectors/hive/storage_adapters/s3fs/tests/CMakeLists.txt

@@ -12,8 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_executable(velox_s3file_test S3FileSystemTest.cpp S3UtilTest.cpp
-                                 S3ConfigTest.cpp)
+add_executable(velox_s3config_test S3ConfigTest.cpp)
+add_test(velox_s3config_test velox_s3config_test)
+target_link_libraries(
+  velox_s3config_test
+  velox_common_config
+  velox_s3fs
+  GTest::gtest
+  GTest::gtest_main)
+
+add_executable(velox_s3file_test S3FileSystemTest.cpp S3UtilTest.cpp)
 add_test(velox_s3file_test velox_s3file_test)
 target_link_libraries(
   velox_s3file_test

velox/connectors/hive/storage_adapters/s3fs/tests/S3ConfigTest.cpp

@@ -34,6 +34,7 @@ TEST(S3ConfigTest, defaultConfig) {
   ASSERT_EQ(s3Config.secretKey(), std::nullopt);
   ASSERT_EQ(s3Config.iamRole(), std::nullopt);
   ASSERT_EQ(s3Config.iamRoleSessionName(), "velox-session");
+  ASSERT_EQ(s3Config.payloadSigningPolicy(), "Never");
 }
 
 TEST(S3ConfigTest, overrideConfig) {
@@ -42,6 +43,7 @@ TEST(S3ConfigTest, overrideConfig) {
       {S3Config::baseConfigKey(S3Config::Keys::kSSLEnabled), "false"},
       {S3Config::baseConfigKey(S3Config::Keys::kUseInstanceCredentials),
        "true"},
+      {"hive.s3.payload-signing-policy", "RequestDependent"},
       {S3Config::baseConfigKey(S3Config::Keys::kEndpoint), "endpoint"},
       {S3Config::baseConfigKey(S3Config::Keys::kEndpointRegion), "region"},
       {S3Config::baseConfigKey(S3Config::Keys::kAccessKey), "access"},
@@ -59,6 +61,7 @@ TEST(S3ConfigTest, overrideConfig) {
   ASSERT_EQ(s3Config.secretKey(), std::optional("secret"));
   ASSERT_EQ(s3Config.iamRole(), std::optional("iam"));
   ASSERT_EQ(s3Config.iamRoleSessionName(), "velox");
+  ASSERT_EQ(s3Config.payloadSigningPolicy(), "RequestDependent");
 }
 
 TEST(S3ConfigTest, overrideBucketConfig) {
@@ -74,6 +77,7 @@ TEST(S3ConfigTest, overrideBucketConfig) {
       {S3Config::baseConfigKey(S3Config::Keys::kAccessKey), "access"},
       {S3Config::bucketConfigKey(S3Config::Keys::kAccessKey, bucket),
        "bucket-access"},
+      {"hive.s3.payload-signing-policy", "Always"},
       {S3Config::baseConfigKey(S3Config::Keys::kSecretKey), "secret"},
       {S3Config::bucketConfigKey(S3Config::Keys::kSecretKey, bucket),
        "bucket-secret"},
@@ -92,6 +96,7 @@ TEST(S3ConfigTest, overrideBucketConfig) {
   ASSERT_EQ(s3Config.secretKey(), std::optional("bucket-secret"));
   ASSERT_EQ(s3Config.iamRole(), std::optional("iam"));
   ASSERT_EQ(s3Config.iamRoleSessionName(), "velox");
+  ASSERT_EQ(s3Config.payloadSigningPolicy(), "Always");
 }
 
 } // namespace

velox/docs/configs.rst

@@ -696,8 +696,14 @@ Each query can override the config by setting corresponding query session proper
    * - hive.s3.log-level
      - string
      - FATAL
-     - **Allowed values:** "OFF", "FATAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"
+     - **Allowed values:** "OFF", "FATAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE".
        Granularity of logging generated by the AWS C++ SDK library.
+   * - hive.s3.payload-signing-policy
+     - string
+     - Never
+     - **Allowed values:** "Always", "RequestDependent", "Never".
+       When set to "Always", the payload checksum is included in the signature calculation.
+       When set to "RequestDependent", the payload checksum is included based on the value returned by "AmazonWebServiceRequest::SignBody()".
    * - hive.s3.iam-role
      - string
      -