feat(storage): global S3 credential-chain switches

Author: BohuTANG

src/common/storage/src/config.rs

@@ -44,6 +44,31 @@ pub struct StorageConfig {
     pub num_cpus: u64,
     pub allow_insecure: bool,
     pub params: StorageParams,
+    /// Global switches that affect AWS S3 credential chain behavior.
+    ///
+    /// Notes:
+    /// - These are runtime-only controls and are not persisted in meta.
+    /// - They apply to all S3 operators created in this process.
+    pub s3_disable_config_load: bool,
+    pub s3_disable_ec2_metadata: bool,
+}
+
+/// Runtime-only switches for AWS S3 credential chain behavior.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct S3CredentialChainConfig {
+    pub disable_config_load: bool,
+    pub disable_ec2_metadata: bool,
+}
+
+impl S3CredentialChainConfig {
+    pub fn init(cfg: S3CredentialChainConfig) -> databend_common_exception::Result<()> {
+        GlobalInstance::set(cfg);
+        Ok(())
+    }
+
+    pub fn try_get() -> Option<S3CredentialChainConfig> {
+        GlobalInstance::try_get()
+    }
 }
 
 // TODO: This config should be moved out of common-storage crate.

src/common/storage/src/operator.rs

@@ -67,6 +67,7 @@ use crate::http_client::get_storage_http_client;
 use crate::metrics_layer::METRICS_LAYER;
 use crate::operator_cache::get_operator_cache;
 use crate::runtime_layer::RuntimeLayer;
+use crate::config::S3CredentialChainConfig;
 
 static METRIC_OPENDAL_RETRIES_COUNT: LazyLock<FamilyCounter<Vec<(&'static str, String)>>> =
     LazyLock::new(|| register_counter_family("opendal_retries_count"));
@@ -416,9 +417,18 @@ fn init_s3_operator(cfg: &StorageS3Config) -> Result<impl Builder> {
         builder = builder.default_storage_class(cfg.storage_class.to_string().as_ref())
     }
 
-    // Disable credential loader
+    // `disable_credential_loader` disables the ambient credential chain entirely.
+    // When it is enabled, only explicit credentials are used and requests can fall back to unsigned.
     if cfg.disable_credential_loader {
         builder = builder.disable_config_load().disable_ec2_metadata();
+    } else if let Some(global) = S3CredentialChainConfig::try_get() {
+        // Apply global limits to the credential chain.
+        if global.disable_config_load {
+            builder = builder.disable_config_load();
+        }
+        if global.disable_ec2_metadata {
+            builder = builder.disable_ec2_metadata();
+        }
     }
 
     // If credential loading is disabled and no credentials are provided, use unsigned requests.
@@ -569,6 +579,10 @@ impl DataOperator {
         conf: &StorageConfig,
         spill_params: Option<StorageParams>,
     ) -> databend_common_exception::Result<()> {
+        S3CredentialChainConfig::init(S3CredentialChainConfig {
+            disable_config_load: conf.s3_disable_config_load,
+            disable_ec2_metadata: conf.s3_disable_ec2_metadata,
+        })?;
         GlobalInstance::set(Self::try_create(conf, spill_params).await?);
 
         Ok(())

src/common/storage/src/stage.rs

@@ -90,12 +90,12 @@ impl StageFileInfo {
 
 pub fn init_stage_operator(stage_info: &StageInfo) -> Result<Operator> {
     if stage_info.stage_type == StageType::External {
-        // External S3 stages don't load credentials by default; `role_arn` opts into assume-role.
+        // External S3 stages disallow the ambient credential chain by default.
+        // `role_arn` opts into using the credential chain as the source credential.
         let storage = match stage_info.stage_params.storage.clone() {
             StorageParams::S3(mut cfg) => {
-                if cfg.role_arn.is_empty() {
-                    cfg.disable_credential_loader = true;
-                }
+                let allow_credential_chain = !cfg.role_arn.is_empty();
+                cfg.disable_credential_loader = !allow_credential_chain;
                 StorageParams::S3(cfg)
             }
             v => v,

src/query/config/src/config.rs

@@ -542,6 +542,9 @@ impl From<InnerStorageConfig> for StorageConfig {
             v => unreachable!("{v:?} should not be used as storage backend"),
         }
 
+        cfg.s3.disable_config_load = inner.s3_disable_config_load;
+        cfg.s3.disable_ec2_metadata = inner.s3_disable_ec2_metadata;
+
         cfg
     }
 }
@@ -587,6 +590,8 @@ impl TryInto<InnerStorageConfig> for StorageConfig {
         Ok(InnerStorageConfig {
             num_cpus: self.storage_num_cpus,
             allow_insecure: self.allow_insecure,
+            s3_disable_config_load: self.s3.disable_config_load,
+            s3_disable_ec2_metadata: self.s3.disable_ec2_metadata,
             params: {
                 match self.typ.as_str() {
                     "azblob" => {
@@ -931,6 +936,14 @@ pub struct S3StorageConfig {
     #[clap(long = "storage-s3-enable-virtual-host-style", value_name = "VALUE")]
     pub enable_virtual_host_style: bool,
 
+    /// Disable loading credentials from env/shared config/web identity.
+    #[clap(long = "storage-s3-disable-config-load")]
+    pub disable_config_load: bool,
+
+    /// Disable EC2 instance metadata credential provider (IMDS).
+    #[clap(long = "storage-s3-disable-ec2-metadata")]
+    pub disable_ec2_metadata: bool,
+
     #[clap(long = "storage-s3-role-arn", value_name = "VALUE", default_value_t)]
     #[serde(rename = "role_arn")]
     pub s3_role_arn: String,
@@ -963,6 +976,8 @@ impl Debug for S3StorageConfig {
             .field("bucket", &self.bucket)
             .field("root", &self.root)
             .field("enable_virtual_host_style", &self.enable_virtual_host_style)
+            .field("disable_config_load", &self.disable_config_load)
+            .field("disable_ec2_metadata", &self.disable_ec2_metadata)
             .field("role_arn", &self.s3_role_arn)
             .field("external_id", &self.s3_external_id)
             .field("access_key_id", &mask_string(&self.access_key_id, 3))
@@ -987,6 +1002,8 @@ impl From<InnerStorageS3Config> for S3StorageConfig {
             root: inner.root,
             master_key: inner.master_key,
             enable_virtual_host_style: inner.enable_virtual_host_style,
+            disable_config_load: false,
+            disable_ec2_metadata: false,
             s3_role_arn: inner.role_arn,
             s3_external_id: inner.external_id,
             storage_class: inner.storage_class,

src/query/service/tests/it/storages/testdata/configs_table_basic.txt

@@ -267,6 +267,8 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | 'storage' | 'oss.server_side_encryption_key_id'                  | ''                                                                                                                                                                                                        | ''       |
 | 'storage' | 's3.access_key_id'                                   | ''                                                                                                                                                                                                        | ''       |
 | 'storage' | 's3.bucket'                                          | ''                                                                                                                                                                                                        | ''       |
+| 'storage' | 's3.disable_config_load'                             | 'false'                                                                                                                                                                                                   | ''       |
+| 'storage' | 's3.disable_ec2_metadata'                            | 'false'                                                                                                                                                                                                   | ''       |
 | 'storage' | 's3.enable_virtual_host_style'                       | 'false'                                                                                                                                                                                                   | ''       |
 | 'storage' | 's3.endpoint_url'                                    | 'https://s3.amazonaws.com'                                                                                                                                                                                | ''       |
 | 'storage' | 's3.external_id'                                     | ''                                                                                                                                                                                                        | ''       |
@@ -293,4 +295,3 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | 'storage' | 'webhdfs.user_name'                                  | ''                                                                                                                                                                                                        | ''       |
 +-----------+------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------+
 
-

src/query/sql/src/planner/binder/ddl/stage.rs

@@ -90,24 +90,32 @@ impl Binder {
                 )
                 .await?;
 
-                // External S3 stages don't load credentials by default; `role_arn` opts into assume-role.
-                let stage_storage = match stage_storage {
+                // Validate the input config using the same credential-chain policy as runtime stage access.
+                let validate_storage = match stage_storage.clone() {
                     StorageParams::S3(mut cfg) => {
-                        if cfg.role_arn.is_empty() {
-                            cfg.disable_credential_loader = true;
-                        }
+                        let allow_credential_chain = !cfg.role_arn.is_empty();
+                        cfg.disable_credential_loader = !allow_credential_chain;
                         StorageParams::S3(cfg)
                     }
                     v => v,
                 };
 
                 // Check the storage params via init operator.
-                let _ = init_operator(&stage_storage).map_err(|err| {
+                let _ = init_operator(&validate_storage).map_err(|err| {
                     ErrorCode::InvalidConfig(format!(
                         "Input storage config for stage is invalid: {err:?}"
                     ))
                 })?;
 
+                // Persist stage configs without embedding runtime-only credential-chain policy.
+                let stage_storage = match stage_storage {
+                    StorageParams::S3(mut cfg) => {
+                        cfg.disable_credential_loader = false;
+                        StorageParams::S3(cfg)
+                    }
+                    v => v,
+                };
+
                 StageInfo::new_external_stage(stage_storage, true).with_stage_name(stage_name)
             }
         };

src/query/sql/src/planner/binder/location.rs

@@ -25,7 +25,6 @@ use databend_common_ast::ast::Connection;
 use databend_common_ast::ast::UriLocation;
 use databend_common_base::runtime::ThreadTracker;
 use databend_common_catalog::table_context::TableContext;
-use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_meta_app::storage::S3StorageClass;
 use databend_common_meta_app::storage::STORAGE_GCS_DEFAULT_ENDPOINT;
@@ -173,16 +172,13 @@ fn parse_s3_params(l: &mut UriLocation, root: String) -> Result<StorageParams> {
     }
     .to_string();
 
-    // If role_arn is empty and we don't allow allow insecure, we should disable credential loader.
-    let mut disable_credential_loader =
-        role_arn.is_empty() && !GlobalConfig::instance().storage.allow_insecure;
-
-    // If we are in history table scope, we allow credential loader to be enabled
+    // Allow credential chain only when it is explicitly needed (assume-role) or for history tables.
     let in_history_table_scope = ThreadTracker::capture_log_settings()
         .is_some_and(|settings| settings.level == LevelFilter::Off);
+    let allow_credential_chain = !role_arn.is_empty() || in_history_table_scope;
+    let disable_credential_loader = !allow_credential_chain;
     if in_history_table_scope {
-        info!("Enable credential loader for history tables");
-        disable_credential_loader = false;
+        info!("Allow credential chain for history tables");
     }
 
     let sp = StorageParams::S3(StorageS3Config {