Merge pull request #47423 from ClickHouse/add-expiration-window-s3

Add expiration window for S3 credentials

docs/en/engines/table-engines/integrations/s3.md

@@ -150,6 +150,7 @@ The following settings can be specified in configuration file for given endpoint
 -   `use_environment_credentials` — If set to `true`, S3 client will try to obtain credentials from environment variables and [Amazon EC2](https://en.wikipedia.org/wiki/Amazon_Elastic_Compute_Cloud) metadata for given endpoint. Optional, default value is `false`.
 -   `region` — Specifies S3 region name. Optional.
 -   `use_insecure_imds_request` — If set to `true`, S3 client will use insecure IMDS request while obtaining credentials from Amazon EC2 metadata. Optional, default value is `false`.
+-   `expiration_window_seconds` — Grace period for checking if expiration-based credentials have expired. Optional, default value is `120`.
 -   `header` —  Adds specified HTTP header to a request to given endpoint. Optional, can be specified multiple times.
 -   `server_side_encryption_customer_key_base64` — If specified, required headers for accessing S3 objects with SSE-C encryption will be set. Optional.
 -   `max_single_read_retries` — The maximum number of attempts during single read. Default value is `4`. Optional.
@@ -166,6 +167,7 @@ The following settings can be specified in configuration file for given endpoint
         <!-- <region>us-west-1</region> -->
         <!-- <use_environment_credentials>false</use_environment_credentials> -->
         <!-- <use_insecure_imds_request>false</use_insecure_imds_request> -->
+        <!-- <expiration_window_seconds>120</expiration_window_seconds> -->
         <!-- <header>Authorization: Bearer SOME-TOKEN</header> -->
         <!-- <server_side_encryption_customer_key_base64>BASE64-ENCODED-KEY</server_side_encryption_customer_key_base64> -->
         <!-- <max_single_read_retries>4</max_single_read_retries> -->

docs/en/engines/table-engines/mergetree-family/mergetree.md

@@ -960,6 +960,7 @@ Optional parameters:
 -   `support_batch_delete` — This controls the check to see if batch deletes are supported. Set this to `false` when using Google Cloud Storage (GCS) as GCS does not support batch deletes and preventing the checks will prevent error messages in the logs.
 -   `use_environment_credentials` — Reads AWS credentials from the Environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY and AWS_SESSION_TOKEN if they exist. Default value is `false`.
 -   `use_insecure_imds_request` — If set to `true`, S3 client will use insecure IMDS request while obtaining credentials from Amazon EC2 metadata. Default value is `false`.
+-   `expiration_window_seconds` — Grace period for checking if expiration-based credentials have expired. Optional, default value is `120`.
 -   `proxy` — Proxy configuration for S3 endpoint. Each `uri` element inside `proxy` block should contain a proxy URL.
 -   `connect_timeout_ms` — Socket connect timeout in milliseconds. Default value is `10 seconds`.
 -   `request_timeout_ms` — Request timeout in milliseconds. Default value is `5 seconds`.

src/Backups/BackupIO_S3.cpp

@@ -11,6 +11,7 @@
 #include <IO/HTTPHeaderEntries.h>
 #include <IO/S3/copyS3File.h>
 #include <IO/S3/Client.h>
+#include <IO/S3/Credentials.h>
 
 #include <Poco/Util/AbstractConfiguration.h>
 
@@ -68,7 +69,9 @@ namespace
             settings.auth_settings.use_environment_credentials.value_or(
                 context->getConfigRef().getBool("s3.use_environment_credentials", false)),
             settings.auth_settings.use_insecure_imds_request.value_or(
-                context->getConfigRef().getBool("s3.use_insecure_imds_request", false)));
+                context->getConfigRef().getBool("s3.use_insecure_imds_request", false)),
+            settings.auth_settings.expiration_window_seconds.value_or(
+                context->getConfigRef().getUInt64("s3.expiration_window_seconds", S3::DEFAULT_EXPIRATION_WINDOW_SECONDS)));
     }
 
     Aws::Vector<Aws::S3::Model::Object> listObjects(S3::Client & client, const S3::URI & s3_uri, const String & file_name)

src/Coordination/KeeperSnapshotManagerS3.cpp

@@ -7,6 +7,7 @@
 #include <Common/setThreadName.h>
 
 #include <IO/S3/getObjectInfo.h>
+#include <IO/S3/Credentials.h>
 #include <IO/WriteBufferFromS3.h>
 #include <IO/ReadBufferFromS3.h>
 #include <IO/ReadBufferFromFile.h>
@@ -103,7 +104,8 @@ void KeeperSnapshotManagerS3::updateS3Configuration(const Poco::Util::AbstractCo
             auth_settings.server_side_encryption_customer_key_base64,
             std::move(headers),
             auth_settings.use_environment_credentials.value_or(false),
-            auth_settings.use_insecure_imds_request.value_or(false));
+            auth_settings.use_insecure_imds_request.value_or(false),
+            auth_settings.expiration_window_seconds.value_or(S3::DEFAULT_EXPIRATION_WINDOW_SECONDS));
 
         auto new_client = std::make_shared<KeeperSnapshotManagerS3::S3Configuration>(std::move(new_uri), std::move(auth_settings), std::move(client));
 

src/Disks/ObjectStorages/S3/diskSettings.cpp

@@ -13,6 +13,7 @@
 #include <aws/core/client/DefaultRetryStrategy.h>
 #include <base/getFQDNOrHostName.h>
 #include <IO/S3Common.h>
+#include <IO/S3/Credentials.h>
 
 #include <Storages/StorageS3Settings.h>
 #include <Disks/ObjectStorages/S3/S3ObjectStorage.h>
@@ -152,7 +153,8 @@ std::unique_ptr<S3::Client> getClient(
         config.getString(config_prefix + ".server_side_encryption_customer_key_base64", ""),
         {},
         config.getBool(config_prefix + ".use_environment_credentials", config.getBool("s3.use_environment_credentials", false)),
-        config.getBool(config_prefix + ".use_insecure_imds_request", config.getBool("s3.use_insecure_imds_request", false)));
+        config.getBool(config_prefix + ".use_insecure_imds_request", config.getBool("s3.use_insecure_imds_request", false)),
+        config.getBool(config_prefix + ".expiration_window_seconds", config.getUInt64("s3.expiration_window_seconds", S3::DEFAULT_EXPIRATION_WINDOW_SECONDS)));
 }
 
 }

src/IO/S3/Client.cpp

@@ -564,7 +564,8 @@ std::unique_ptr<S3::Client> ClientFactory::create( // NOLINT
     const String & server_side_encryption_customer_key_base64,
     HTTPHeaderEntries headers,
     bool use_environment_credentials,
-    bool use_insecure_imds_request)
+    bool use_insecure_imds_request,
+    uint64_t expiration_window_seconds)
 {
     PocoHTTPClientConfiguration client_configuration = cfg_;
     client_configuration.updateSchemeAndRegion();
@@ -592,7 +593,8 @@ std::unique_ptr<S3::Client> ClientFactory::create( // NOLINT
             client_configuration,
             std::move(credentials),
             use_environment_credentials,
-            use_insecure_imds_request);
+            use_insecure_imds_request,
+            expiration_window_seconds);
 
     client_configuration.retryStrategy = std::make_shared<Client::RetryStrategy>(std::move(client_configuration.retryStrategy));
     return Client::create(

src/IO/S3/Client.h

@@ -11,6 +11,7 @@
 #include <IO/S3/URI.h>
 #include <IO/S3/Requests.h>
 #include <IO/S3/PocoHTTPClient.h>
+#include <IO/S3/Credentials.h>
 
 #include <aws/core/Aws.h>
 #include <aws/core/client/DefaultRetryStrategy.h>
@@ -228,7 +229,8 @@ public:
         const String & server_side_encryption_customer_key_base64,
         HTTPHeaderEntries headers,
         bool use_environment_credentials,
-        bool use_insecure_imds_request);
+        bool use_insecure_imds_request,
+        uint64_t expiration_window_seconds = DEFAULT_EXPIRATION_WINDOW_SECONDS);
 
     PocoHTTPClientConfiguration createClientConfiguration(
         const String & force_region,

src/IO/S3/Credentials.cpp

@@ -21,6 +21,21 @@
 namespace DB::S3
 {
 
+namespace
+{
+
+bool areCredentialsEmptyOrExpired(const Aws::Auth::AWSCredentials & credentials, uint64_t expiration_window_seconds)
+{
+    if (credentials.IsEmpty())
+        return true;
+
+    const Aws::Utils::DateTime now = Aws::Utils::DateTime::Now();
+    return now >= credentials.GetExpiration() - std::chrono::seconds(expiration_window_seconds);
+}
+
+
+}
+
 AWSEC2MetadataClient::AWSEC2MetadataClient(const Aws::Client::ClientConfiguration & client_configuration, const char * endpoint_)
     : Aws::Internal::AWSHttpResourceClient(client_configuration)
     , endpoint(endpoint_)
@@ -270,8 +285,10 @@ void AWSInstanceProfileCredentialsProvider::refreshIfExpired()
     Reload();
 }
 
-AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider::AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider(DB::S3::PocoHTTPClientConfiguration & aws_client_configuration)
+AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider::AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider(
+    DB::S3::PocoHTTPClientConfiguration & aws_client_configuration, uint64_t expiration_window_seconds_)
     : logger(&Poco::Logger::get("AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider"))
+    , expiration_window_seconds(expiration_window_seconds_)
 {
     // check environment variables
     String tmp_region = Aws::Environment::GetEnv("AWS_DEFAULT_REGION");
@@ -388,16 +405,12 @@ void AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider::Reload()
 void AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider::refreshIfExpired()
 {
     Aws::Utils::Threading::ReaderLockGuard guard(m_reloadLock);
-    if (!credentials.IsExpiredOrEmpty())
-    {
+    if (!areCredentialsEmptyOrExpired(credentials, expiration_window_seconds))
         return;
-    }
 
     guard.UpgradeToWriterLock();
-    if (!credentials.IsExpiredOrEmpty()) // double-checked lock to avoid refreshing twice
-    {
+    if (!areCredentialsEmptyOrExpired(credentials, expiration_window_seconds)) // double-checked lock to avoid refreshing twice
         return;
-    }
 
     Reload();
 }
@@ -406,7 +419,8 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
         const DB::S3::PocoHTTPClientConfiguration & configuration,
         const Aws::Auth::AWSCredentials & credentials,
         bool use_environment_credentials,
-        bool use_insecure_imds_request)
+        bool use_insecure_imds_request,
+        uint64_t expiration_window_seconds)
 {
     auto * logger = &Poco::Logger::get("S3CredentialsProviderChain");
 
@@ -439,7 +453,7 @@ S3CredentialsProviderChain::S3CredentialsProviderChain(
                 configuration.for_disk_s3,
                 configuration.get_request_throttler,
                 configuration.put_request_throttler);
-            AddProvider(std::make_shared<AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider>(aws_client_configuration));
+            AddProvider(std::make_shared<AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider>(aws_client_configuration, expiration_window_seconds));
         }
 
         AddProvider(std::make_shared<Aws::Auth::EnvironmentAWSCredentialsProvider>());

src/IO/S3/Credentials.h

@@ -17,6 +17,8 @@
 namespace DB::S3
 {
 
+inline static constexpr uint64_t DEFAULT_EXPIRATION_WINDOW_SECONDS = 120;
+
 class AWSEC2MetadataClient : public Aws::Internal::AWSHttpResourceClient
 {
     static constexpr char EC2_SECURITY_CREDENTIALS_RESOURCE[] = "/latest/meta-data/iam/security-credentials";
@@ -97,9 +99,11 @@ class AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider : public Aws::Auth::AWS
     /// See STSAssumeRoleWebIdentityCredentialsProvider.
 
 public:
-    explicit AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider(DB::S3::PocoHTTPClientConfiguration & aws_client_configuration);
+    explicit AwsAuthSTSAssumeRoleWebIdentityCredentialsProvider(
+        DB::S3::PocoHTTPClientConfiguration & aws_client_configuration, uint64_t expiration_window_seconds_);
 
     Aws::Auth::AWSCredentials GetAWSCredentials() override;
+
 protected:
     void Reload() override;
 
@@ -114,14 +118,19 @@ private:
     Aws::String token;
     bool initialized = false;
     Poco::Logger * logger;
+    uint64_t expiration_window_seconds;
 };
 
 class S3CredentialsProviderChain : public Aws::Auth::AWSCredentialsProviderChain
 {
 public:
-    S3CredentialsProviderChain(const DB::S3::PocoHTTPClientConfiguration & configuration, const Aws::Auth::AWSCredentials & credentials, bool use_environment_credentials, bool use_insecure_imds_request);
+    S3CredentialsProviderChain(
+        const DB::S3::PocoHTTPClientConfiguration & configuration,
+        const Aws::Auth::AWSCredentials & credentials,
+        bool use_environment_credentials,
+        bool use_insecure_imds_request,
+        uint64_t expiration_window_seconds);
 };
-
 }
 
 #endif

src/IO/S3Common.cpp

@@ -85,6 +85,10 @@ AuthSettings AuthSettings::loadFromConfig(const std::string & config_elem, const
     if (config.has(config_elem + ".use_insecure_imds_request"))
         use_insecure_imds_request = config.getBool(config_elem + ".use_insecure_imds_request");
 
+    std::optional<uint64_t> expiration_window_seconds;
+    if (config.has(config_elem + ".expiration_window_seconds"))
+        expiration_window_seconds = config.getUInt64(config_elem + ".expiration_window_seconds");
+
     HTTPHeaderEntries headers;
     Poco::Util::AbstractConfiguration::Keys subconfig_keys;
     config.keys(config_elem, subconfig_keys);
@@ -107,7 +111,8 @@ AuthSettings AuthSettings::loadFromConfig(const std::string & config_elem, const
         std::move(server_side_encryption_customer_key_base64),
         std::move(headers),
         use_environment_credentials,
-        use_insecure_imds_request
+        use_insecure_imds_request,
+        expiration_window_seconds
     };
 }
 
@@ -127,6 +132,7 @@ void AuthSettings::updateFrom(const AuthSettings & from)
     server_side_encryption_customer_key_base64 = from.server_side_encryption_customer_key_base64;
     use_environment_credentials = from.use_environment_credentials;
     use_insecure_imds_request = from.use_insecure_imds_request;
+    expiration_window_seconds = from.expiration_window_seconds;
 }
 
 }

src/IO/S3Common.h

@@ -84,6 +84,7 @@ struct AuthSettings
 
     std::optional<bool> use_environment_credentials;
     std::optional<bool> use_insecure_imds_request;
+    std::optional<uint64_t> expiration_window_seconds;
 
     bool operator==(const AuthSettings & other) const = default;
 

src/Storages/StorageS3.cpp

@@ -1266,7 +1266,8 @@ void StorageS3::updateConfiguration(ContextPtr ctx, StorageS3::Configuration & u
         upd.auth_settings.server_side_encryption_customer_key_base64,
         std::move(headers),
         upd.auth_settings.use_environment_credentials.value_or(ctx->getConfigRef().getBool("s3.use_environment_credentials", false)),
-        upd.auth_settings.use_insecure_imds_request.value_or(ctx->getConfigRef().getBool("s3.use_insecure_imds_request", false)));
+        upd.auth_settings.use_insecure_imds_request.value_or(ctx->getConfigRef().getBool("s3.use_insecure_imds_request", false)),
+        upd.auth_settings.expiration_window_seconds.value_or(ctx->getConfigRef().getUInt64("s3.expiration_window_seconds", S3::DEFAULT_EXPIRATION_WINDOW_SECONDS)));
 }
 
 void StorageS3::processNamedCollectionResult(StorageS3::Configuration & configuration, const NamedCollection & collection)