From 1917adcac1057ee9464ede84cea48247264df191 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 10:01:17 +0700 Subject: [PATCH] fix: honor the S3 profile name and file and Hadoop's addressing mode for custom endpoints The profile credentials provider ignored fs.s3a.auth.profile.name and fs.s3a.auth.profile.file, and fs.s3a.path.style.access was applied inverted, so every custom endpoint was addressed path-style whatever the flag said and virtual-hosted addressing was never produced. Carry the profile name and file into the SDK builder, derive the virtual-hosted flag from the path-style setting the way Hadoop does, rebuild the endpoint as bucket.host for virtual-hosted addressing while forcing path-style for IP-literal hosts as the AWS SDK does, and return the effective mode with the endpoint so the two cannot disagree. Closes #4245 Closes #2802 --- docs/source/user-guide/latest/datasources.md | 7 +- native/Cargo.lock | 1 + native/Cargo.toml | 1 + native/core/Cargo.toml | 1 + native/core/src/parquet/objectstore/s3.rs | 674 ++++++++++++++++--- 5 files changed, 603 insertions(+), 81 deletions(-) diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 97dbece0094..fe52d0552da 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -199,7 +199,7 @@ AWS credential providers can be configured using the `fs.s3a.aws.credentials.pro | `com.amazonaws.auth.InstanceProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider` | Access S3 using EC2 instance metadata service (IMDS) | None | | `com.amazonaws.auth.ContainerCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider`
`com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper` | Access S3 using ECS task credentials | None | | `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`
`software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider` | Authenticate using web identity token file | None | -| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | None | +| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | `fs.s3a.auth.profile.name` (optional), `fs.s3a.auth.profile.file` (optional); both apply only when this provider is configured | Multiple credential providers can be specified in a comma-separated list using the `fs.s3a.aws.credentials.provider` configuration, just as Hadoop AWS supports. If `fs.s3a.aws.credentials.provider` is not configured, Hadoop S3A's default credential provider chain will be used. All configuration options also support bucket-specific overrides using the pattern `fs.s3a.bucket.{bucket-name}.{option}`. @@ -216,6 +216,11 @@ Beyond credential providers, Comet's Parquet scan supports additional S3 configu All configuration options support bucket-specific overrides using the pattern `fs.s3a.bucket.{bucket-name}.{option}`. +`fs.s3a.path.style.access` selects how the bucket is placed in the request URL: virtual-hosted +addressing (the default, `false`) sends requests to `https://.`, while path-style +(`true`) sends them to `https:///`, which many S3-compatible services such as MinIO +require. An endpoint whose host is an IP address is always addressed path-style, as the AWS SDK does. + ### S3-Compliant Filesystem Schemes Some environments front an S3-compatible service (MinIO, Ceph RGW, Cloudflare R2, Wasabi, and diff --git a/native/Cargo.lock b/native/Cargo.lock index df5fd4ac14a..d65965913fb 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1963,6 +1963,7 @@ dependencies = [ "async-trait", "aws-config", "aws-credential-types", + "aws-runtime", "base64 0.23.1", "bytes", "comet-contrib-delta", diff --git a/native/Cargo.toml b/native/Cargo.toml index 1805a185e9d..2e86d8cc7f1 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -61,6 +61,7 @@ thiserror = "2" object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"] } url = "2.2" aws-config = "1.8.18" +aws-runtime = "1.9.2" aws-credential-types = "1.2.13" iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879" } iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 97ce3808cea..a76d6188f33 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -67,6 +67,7 @@ datafusion-comet-shuffle = { workspace = true } object_store = { workspace = true } url = { workspace = true } aws-config = { workspace = true } +aws-runtime = { workspace = true } aws-credential-types = { workspace = true } parking_lot = "0.12.5" # Optional Delta Lake contrib (enabled by the `contrib-delta` feature). Source lives diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index c10fc60816e..29187a3b7e8 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -18,7 +18,7 @@ use log::{debug, error}; use std::collections::HashMap; use std::sync::OnceLock; -use url::Url; +use url::{Host, Url}; use crate::cloud::s3::credential_bridge::{AccessMode, CometS3CredentialBridge}; use crate::execution::jni_api::get_runtime; @@ -34,6 +34,7 @@ use aws_credential_types::{ provider::{error::CredentialsError, ProvideCredentials}, Credentials, }; +use aws_runtime::env_config::file::{EnvConfigFileKind, EnvConfigFiles}; use object_store::{ aws::{AmazonS3Builder, AmazonS3ConfigKey, AwsCredential}, path::Path, @@ -239,24 +240,25 @@ fn extract_s3_config_options( s3_configs.insert(AmazonS3ConfigKey::Region, region.to_string()); } - // Extract and handle path style access (virtual hosted style) - let mut virtual_hosted_style_request = false; - if let Some(path_style) = get_config_trimmed(configs, bucket, "path.style.access") { - virtual_hosted_style_request = path_style.to_lowercase() == "true"; - s3_configs.insert( - AmazonS3ConfigKey::VirtualHostedStyleRequest, - virtual_hosted_style_request.to_string(), - ); - } + // Hadoop defaults fs.s3a.path.style.access to false, which means virtual-hosted addressing, + // and treats non-boolean text as that default. object_store expects the inverse flag. + let path_style_access = get_config_trimmed(configs, bucket, "path.style.access") + .is_some_and(|value| value.eq_ignore_ascii_case("true")); + let mut virtual_hosted_style_request = !path_style_access; - // Extract endpoint configuration and modify if virtual hosted style is enabled + // Extract endpoint configuration and shape it for the selected addressing style. The flag is + // taken from the normalized result so the endpoint and the flag never disagree. if let Some(endpoint) = get_config_trimmed(configs, bucket, "endpoint") { - let normalized_endpoint = - normalize_endpoint(endpoint, bucket, virtual_hosted_style_request); - if let Some(endpoint) = normalized_endpoint { - s3_configs.insert(AmazonS3ConfigKey::Endpoint, endpoint); + if let Some(normalized) = normalize_endpoint(endpoint, bucket, virtual_hosted_style_request) + { + virtual_hosted_style_request = normalized.virtual_hosted_style_request; + s3_configs.insert(AmazonS3ConfigKey::Endpoint, normalized.endpoint); } } + s3_configs.insert( + AmazonS3ConfigKey::VirtualHostedStyleRequest, + virtual_hosted_style_request.to_string(), + ); // Extract request payer configuration if let Some(requester_pays) = get_config_trimmed(configs, bucket, "requester.pays.enabled") { @@ -270,11 +272,21 @@ fn extract_s3_config_options( s3_configs } +/// An endpoint shaped for object_store together with the addressing mode it was shaped for. +#[derive(Debug, Clone, PartialEq)] +struct NormalizedEndpoint { + endpoint: String, + virtual_hosted_style_request: bool, +} + +/// Shapes a Hadoop `fs.s3a.endpoint` value into the endpoint object_store expects: for +/// virtual-hosted requests the bucket becomes the leading host label (`scheme://bucket.host[:port]`), +/// while for path-style requests object_store appends `/bucket` itself so the value passes through. fn normalize_endpoint( endpoint: &str, bucket: &str, virtual_hosted_style_request: bool, -) -> Option { +) -> Option { if endpoint.is_empty() { return None; } @@ -292,15 +304,36 @@ fn normalize_endpoint( endpoint.to_string() }; - if virtual_hosted_style_request { - if endpoint.ends_with("/") { - Some(format!("{endpoint}{bucket}")) - } else { - Some(format!("{endpoint}/{bucket}")) - } - } else { - Some(endpoint) // Avoid extra to_string() call since endpoint is already a String + let path_style = |endpoint: String| { + Some(NormalizedEndpoint { + endpoint, + virtual_hosted_style_request: false, + }) + }; + if !virtual_hosted_style_request { + return path_style(endpoint); } + + // Fall back to the endpoint as written when it cannot be parsed so object_store reports + // the malformed value instead of a mangled one + let Ok(url) = Url::parse(&endpoint) else { + return path_style(endpoint); + }; + // The AWS SDK endpoint rules address IP-literal hosts path-style since `bucket.127.0.0.1` is + // not a valid host. Hadoop does not special-case `localhost`, so neither does this. + let host = match url.host() { + Some(Host::Domain(host)) => host, + _ => return path_style(endpoint), + }; + let port = url + .port() + .map(|port| format!(":{port}")) + .unwrap_or_default(); + let path = url.path().trim_end_matches('/'); + Some(NormalizedEndpoint { + endpoint: format!("{}://{bucket}.{host}{port}{path}", url.scheme()), + virtual_hosted_style_request: true, + }) } fn get_config<'a>( @@ -323,6 +356,17 @@ pub(super) fn get_config_trimmed<'a>( get_config(configs, bucket, property).map(|s| s.trim()) } +/// Like [`get_config_trimmed`] but treats a blank value as unset. +fn get_non_empty_config( + configs: &HashMap, + bucket: &str, + property: &str, +) -> Option { + get_config_trimmed(configs, bucket, property) + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + /// Activation key (without `fs.s3a.` prefix) naming the vendor `CometS3CredentialProvider` FQCN. /// Per-bucket override is honored via [`get_config_trimmed`]. const PROVIDER_CLASS_PROPERTY: &str = "comet.credential.provider.class"; @@ -481,7 +525,10 @@ fn build_aws_credential_provider_metadata( } HADOOP_ASSUMED_ROLE => build_assume_role_credential_provider_metadata(configs, bucket), AWS_WEB_IDENTITY_V1 | AWS_WEB_IDENTITY => Ok(CredentialProviderMetadata::WebIdentity), - AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile), + AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { + name: get_non_empty_config(configs, bucket, "auth.profile.name"), + file: get_non_empty_config(configs, bucket, "auth.profile.file"), + }), _ => Err(object_store::Error::Generic { store: "S3", source: format!("Unsupported credential provider: {credential_provider_name}").into(), @@ -712,7 +759,10 @@ enum CredentialProviderMetadata { Imds, Environment, WebIdentity, - Profile, + Profile { + name: Option, + file: Option, + }, Static { is_valid: bool, access_key: String, @@ -735,7 +785,7 @@ impl CredentialProviderMetadata { CredentialProviderMetadata::Imds => "Imds", CredentialProviderMetadata::Environment => "Environment", CredentialProviderMetadata::WebIdentity => "WebIdentity", - CredentialProviderMetadata::Profile => "Profile", + CredentialProviderMetadata::Profile { .. } => "Profile", CredentialProviderMetadata::Static { .. } => "Static", CredentialProviderMetadata::AssumeRole { .. } => "AssumeRole", CredentialProviderMetadata::Chain(..) => "Chain", @@ -751,7 +801,17 @@ impl CredentialProviderMetadata { CredentialProviderMetadata::Imds => "Imds".to_string(), CredentialProviderMetadata::Environment => "Environment".to_string(), CredentialProviderMetadata::WebIdentity => "WebIdentity".to_string(), - CredentialProviderMetadata::Profile => "Profile".to_string(), + CredentialProviderMetadata::Profile { name, file } => { + let overrides: Vec = [("name", name), ("file", file)] + .into_iter() + .filter_map(|(key, value)| value.as_ref().map(|v| format!("{key}: {v}"))) + .collect(); + if overrides.is_empty() { + "Profile".to_string() + } else { + format!("Profile({})", overrides.join(", ")) + } + } CredentialProviderMetadata::Static { is_valid, .. } => { format!("Static(valid: {is_valid})") } @@ -814,11 +874,22 @@ impl CredentialProviderMetadata { .build(); Ok(Arc::new(credential_provider)) } - CredentialProviderMetadata::Profile => { - let credential_provider = ProfileFileCredentialsProvider::builder() - .configure(&ProviderConfig::with_default_region().await) - .build(); - Ok(Arc::new(credential_provider)) + CredentialProviderMetadata::Profile { name, file } => { + let mut builder = ProfileFileCredentialsProvider::builder() + .configure(&ProviderConfig::with_default_region().await); + if let Some(name) = name { + builder = builder.profile_name(name); + } + if let Some(file) = file { + // Hadoop's ProfileAWSCredentialsProvider loads fs.s3a.auth.profile.file as a + // credentials-format file and reads nothing else, so mirror that here + builder = builder.profile_files( + EnvConfigFiles::builder() + .with_file(EnvConfigFileKind::Credentials, file) + .build(), + ); + } + Ok(Arc::new(builder.build())) } CredentialProviderMetadata::Static { is_valid, @@ -974,6 +1045,20 @@ mod tests { self } + fn with_property(mut self, property: &str, value: &str) -> Self { + self.configs + .insert(format!("fs.s3a.{property}"), value.to_string()); + self + } + + fn with_bucket_property(mut self, bucket: &str, property: &str, value: &str) -> Self { + self.configs.insert( + format!("fs.s3a.bucket.{bucket}.{property}"), + value.to_string(), + ); + self + } + fn build(self) -> HashMap { self.configs } @@ -995,6 +1080,34 @@ mod tests { ); } + #[test] + #[cfg_attr(miri, ignore)] // AWS credential providers and object_store call foreign functions + fn test_create_store_with_custom_endpoint() { + // object_store must accept the flag and endpoint pair in both addressing modes, and + // create_store enables allow_http so an http endpoint is usable + let url = Url::parse("s3a://test-bucket/comet/data.parquet").unwrap(); + for path_style_access in ["false", "true"] { + let configs = TestConfigBuilder::new() + .with_credential_provider(HADOOP_ANONYMOUS) + .with_region("us-east-1") + .with_property("endpoint", "http://minio.internal:9000") + .with_property("path.style.access", path_style_access) + .build(); + let (_object_store, path) = + create_store(&url, &configs, Duration::from_secs(300)).unwrap(); + assert_eq!(path, Path::from("/comet/data.parquet")); + } + + // An IP-literal endpoint must build without path.style.access being set + let configs = TestConfigBuilder::new() + .with_credential_provider(HADOOP_ANONYMOUS) + .with_region("us-east-1") + .with_property("endpoint", "http://127.0.0.1:9000") + .build(); + let (_object_store, path) = create_store(&url, &configs, Duration::from_secs(300)).unwrap(); + assert_eq!(path, Path::from("/comet/data.parquet")); + } + #[test] fn test_get_config_trimmed() { let configs = TestConfigBuilder::new() @@ -1502,22 +1615,143 @@ mod tests { #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_profile_credential_provider() { + // (configured name, configured file, expected name, expected file) + let cases = [ + (None, None, None, None), + (Some("analytics"), None, Some("analytics"), None), + ( + None, + Some("/etc/aws/credentials"), + None, + Some("/etc/aws/credentials"), + ), + ( + Some("analytics"), + Some("/etc/aws/credentials"), + Some("analytics"), + Some("/etc/aws/credentials"), + ), + // Empty and blank values are treated as unset, other values are trimmed + (Some(""), Some(" "), None, None), + ( + Some(" analytics "), + Some(" /etc/aws/credentials "), + Some("analytics"), + Some("/etc/aws/credentials"), + ), + ]; for provider_name in [AWS_PROFILE, AWS_PROFILE_V1] { - let configs = TestConfigBuilder::new() - .with_credential_provider(provider_name) - .build(); + for (name, file, expected_name, expected_file) in cases { + let mut builder = TestConfigBuilder::new().with_credential_provider(provider_name); + if let Some(name) = name { + builder = builder.with_property("auth.profile.name", name); + } + if let Some(file) = file { + builder = builder.with_property("auth.profile.file", file); + } + let configs = builder.build(); + + let result = + build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: expected_name.map(str::to_string), + file: expected_file.map(str::to_string), + }, + "provider {provider_name}, name {name:?}, file {file:?}" + ); + } + } + } - let result = - build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) - .await - .unwrap(); - assert!(result.is_some(), "Should return a credential provider"); + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_profile_credential_provider_per_bucket_override() { + // Each key is overridden independently, so a bucket can replace just the name or just + // the file while the other key keeps its global value + let configs = TestConfigBuilder::new() + .with_credential_provider(AWS_PROFILE) + .with_property("auth.profile.name", "global-profile") + .with_property("auth.profile.file", "/etc/aws/global-credentials") + .with_bucket_property("name-bucket", "auth.profile.name", "bucket-profile") + .with_bucket_property( + "file-bucket", + "auth.profile.file", + "/etc/aws/bucket-credentials", + ) + .build(); - let test_provider = result.unwrap().metadata(); - assert_eq!(test_provider, CredentialProviderMetadata::Profile); + let cases = [ + ( + "name-bucket", + "bucket-profile", + "/etc/aws/global-credentials", + ), + ( + "file-bucket", + "global-profile", + "/etc/aws/bucket-credentials", + ), + ( + "other-bucket", + "global-profile", + "/etc/aws/global-credentials", + ), + ]; + for (bucket, expected_name, expected_file) in cases { + let result = build_credential_provider(&configs, bucket, Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: Some(expected_name.to_string()), + file: Some(expected_file.to_string()), + }, + "bucket {bucket}" + ); } } + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_profile_credential_provider_in_chain() { + let configs = TestConfigBuilder::new() + .with_credential_provider(&format!( + "{AWS_ENVIRONMENT},{AWS_PROFILE},{AWS_INSTANCE_PROFILE}" + )) + .with_property("auth.profile.name", "analytics") + .with_property("auth.profile.file", "/etc/aws/credentials") + .build(); + + let result = build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Chain(vec![ + CredentialProviderMetadata::Environment, + CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: Some("/etc/aws/credentials".to_string()), + }, + CredentialProviderMetadata::Imds, + ]) + ); + } + #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_hadoop_iam_instance_credential_provider() { @@ -1966,75 +2200,335 @@ mod tests { } #[test] - fn test_extract_s3_config_custom_endpoint() { - let cases = vec![ - ("custom.endpoint.com", "https://custom.endpoint.com"), - ("https://custom.endpoint.com", "https://custom.endpoint.com"), + fn test_normalize_endpoint_virtual_hosted_style() { + // Virtual-hosted addressing inserts the bucket as the leading host label. The scheme, + // port and any path suffix are preserved and a trailing slash is dropped. + let cases = [ + ( + "custom.endpoint.com", + "https://test-bucket.custom.endpoint.com", + ), + ( + "http://custom.endpoint.com", + "http://test-bucket.custom.endpoint.com", + ), + ( + "https://custom.endpoint.com/", + "https://test-bucket.custom.endpoint.com", + ), + ( + "http://minio.internal:9000", + "http://test-bucket.minio.internal:9000", + ), + ( + "https://custom.endpoint.com:8443/", + "https://test-bucket.custom.endpoint.com:8443", + ), ( "https://custom.endpoint.com/path/to/resource", - "https://custom.endpoint.com/path/to/resource", + "https://test-bucket.custom.endpoint.com/path/to/resource", + ), + ( + "https://custom.endpoint.com/path/to/resource/", + "https://test-bucket.custom.endpoint.com/path/to/resource", + ), + ( + "s3.us-west-2.amazonaws.com", + "https://test-bucket.s3.us-west-2.amazonaws.com", ), ]; - for (endpoint, configured_endpoint) in cases { - let mut configs = HashMap::new(); - configs.insert("fs.s3a.endpoint".to_string(), endpoint.to_string()); - let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + for (endpoint, expected) in cases { assert_eq!( - s3_configs.get(&AmazonS3ConfigKey::Endpoint), - Some(&configured_endpoint.to_string()) + normalize_endpoint(endpoint, "test-bucket", true), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: true, + }), + "endpoint {endpoint}" ); } + + assert_eq!( + normalize_endpoint("custom.endpoint.com", "my.dotted.bucket", true), + Some(NormalizedEndpoint { + endpoint: "https://my.dotted.bucket.custom.endpoint.com".to_string(), + virtual_hosted_style_request: true, + }) + ); } #[test] - fn test_extract_s3_config_custom_endpoint_with_virtual_hosted_style() { - let cases = vec![ + fn test_normalize_endpoint_path_style() { + // Path-style leaves the endpoint as configured apart from the https default, since + // object_store appends the bucket itself + let cases = [ + ("custom.endpoint.com", "https://custom.endpoint.com"), + ("http://custom.endpoint.com", "http://custom.endpoint.com"), ( - "custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com/", + "https://custom.endpoint.com/", ), + ("http://minio.internal:9000", "http://minio.internal:9000"), ( - "https://custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com:8443/", + "https://custom.endpoint.com:8443/", ), ( - "https://custom.endpoint.com/", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com/path/to/resource", + "https://custom.endpoint.com/path/to/resource", ), ( - "https://custom.endpoint.com/path/to/resource", - "https://custom.endpoint.com/path/to/resource/test-bucket", + "s3.us-west-2.amazonaws.com", + "https://s3.us-west-2.amazonaws.com", ), + ]; + for (endpoint, expected) in cases { + assert_eq!( + normalize_endpoint(endpoint, "test-bucket", false), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: false, + }), + "endpoint {endpoint}" + ); + } + + assert_eq!( + normalize_endpoint("custom.endpoint.com", "my.dotted.bucket", false), + Some(NormalizedEndpoint { + endpoint: "https://custom.endpoint.com".to_string(), + virtual_hosted_style_request: false, + }) + ); + } + + #[test] + fn test_normalize_endpoint_ip_host_forces_path_style() { + // The AWS SDK endpoint rules address IP-literal hosts path-style whatever the + // configuration says, since `bucket.127.0.0.1` is not a valid host + let cases = [ + ("http://127.0.0.1:9000", "http://127.0.0.1:9000"), + ("http://127.0.0.1", "http://127.0.0.1"), + ("127.0.0.1:9000", "https://127.0.0.1:9000"), + ("http://[::1]:9000", "http://[::1]:9000"), + ("https://[::1]", "https://[::1]"), + ("[::1]:9000", "https://[::1]:9000"), + ]; + for (endpoint, expected) in cases { + for virtual_hosted_style_request in [true, false] { + assert_eq!( + normalize_endpoint(endpoint, "test-bucket", virtual_hosted_style_request), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: false, + }), + "endpoint {endpoint}, requested virtual-hosted {virtual_hosted_style_request}" + ); + } + } + } + + #[test] + fn test_normalize_endpoint_skips_default_aws_endpoint() { + for virtual_hosted_style_request in [true, false] { + assert_eq!( + normalize_endpoint( + "s3.amazonaws.com", + "test-bucket", + virtual_hosted_style_request + ), + None + ); + assert_eq!( + normalize_endpoint("", "test-bucket", virtual_hosted_style_request), + None + ); + } + } + + #[test] + fn test_extract_s3_config_path_style_access() { + // Hadoop defaults fs.s3a.path.style.access to false (virtual-hosted) and, like + // Configuration.getBoolean, falls back to that default for non-boolean text + let cases = [ + (None, "true", "https://test-bucket.custom.endpoint.com"), ( - "https://custom.endpoint.com/path/to/resource/", - "https://custom.endpoint.com/path/to/resource/test-bucket", + Some("false"), + "true", + "https://test-bucket.custom.endpoint.com", + ), + ( + Some("yes"), + "true", + "https://test-bucket.custom.endpoint.com", ), + (Some("true"), "false", "https://custom.endpoint.com"), + (Some(" TRUE "), "false", "https://custom.endpoint.com"), ]; - for (endpoint, configured_endpoint) in cases { - let mut configs = HashMap::new(); - configs.insert("fs.s3a.endpoint".to_string(), endpoint.to_string()); - configs.insert("fs.s3a.path.style.access".to_string(), "true".to_string()); + for (path_style_access, expected_flag, expected_endpoint) in cases { + let mut builder = + TestConfigBuilder::new().with_property("endpoint", "custom.endpoint.com"); + if let Some(value) = path_style_access { + builder = builder.with_property("path.style.access", value); + } + let s3_configs = extract_s3_config_options(&builder.build(), "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&expected_flag.to_string()), + "path.style.access {path_style_access:?}" + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&expected_endpoint.to_string()), + "path.style.access {path_style_access:?}" + ); + } + } + + #[test] + fn test_extract_s3_config_ip_endpoint_forces_path_style() { + // With path.style.access unset an IP-literal endpoint stays path-style and the flag + // handed to object_store agrees with the unchanged endpoint + for endpoint in [ + "http://127.0.0.1:9000", + "http://127.0.0.1", + "http://[::1]:9000", + "http://[::1]", + ] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", endpoint) + .build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()), + "endpoint {endpoint}" + ); assert_eq!( s3_configs.get(&AmazonS3ConfigKey::Endpoint), - Some(&configured_endpoint.to_string()) + Some(&endpoint.to_string()), + "endpoint {endpoint}" ); } } #[test] - fn test_extract_s3_config_ignore_default_endpoint() { - let mut configs = HashMap::new(); - configs.insert( - "fs.s3a.endpoint".to_string(), - "s3.amazonaws.com".to_string(), - ); + fn test_extract_s3_config_http_endpoint_keeps_scheme() { + for (path_style_access, expected_endpoint) in [ + ("false", "http://test-bucket.minio.internal:9000"), + ("true", "http://minio.internal:9000"), + ] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", "http://minio.internal:9000") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&expected_endpoint.to_string()), + "path.style.access {path_style_access}" + ); + } + } + + #[test] + fn test_extract_s3_config_path_style_access_without_endpoint() { + // The flag is always handed to object_store so the default AWS endpoint follows the + // same addressing rule as a custom one + let configs = TestConfigBuilder::new().with_region("us-east-1").build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); - assert!(s3_configs.is_empty()); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); - configs.insert("fs.s3a.endpoint".to_string(), "".to_string()); + let configs = TestConfigBuilder::new() + .with_region("us-east-1") + .with_property("path.style.access", "true") + .build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); - assert!(s3_configs.is_empty()); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + } + + #[test] + fn test_extract_s3_config_per_bucket_overrides() { + // A bucket can override both the endpoint and the addressing flag, in either direction + let configs = TestConfigBuilder::new() + .with_property("endpoint", "global.endpoint.com") + .with_property("path.style.access", "true") + .with_bucket_property("vh-bucket", "endpoint", "http://bucket.endpoint.com:9000") + .with_bucket_property("vh-bucket", "path.style.access", "false") + .build(); + + let s3_configs = extract_s3_config_options(&configs, "vh-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"http://vh-bucket.bucket.endpoint.com:9000".to_string()) + ); + + let s3_configs = extract_s3_config_options(&configs, "other-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://global.endpoint.com".to_string()) + ); + + let configs = TestConfigBuilder::new() + .with_property("endpoint", "global.endpoint.com") + .with_property("path.style.access", "false") + .with_bucket_property("ps-bucket", "path.style.access", "true") + .build(); + + let s3_configs = extract_s3_config_options(&configs, "ps-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://global.endpoint.com".to_string()) + ); + + let s3_configs = extract_s3_config_options(&configs, "other-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://other-bucket.global.endpoint.com".to_string()) + ); + } + + #[test] + fn test_extract_s3_config_ignore_default_endpoint() { + for path_style_access in ["false", "true"] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", "s3.amazonaws.com") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + + let configs = TestConfigBuilder::new() + .with_property("endpoint", "") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + } } #[test] @@ -2059,6 +2553,26 @@ mod tests { "AssumeRole(role: arn:aws:iam::123456789012:role/test-role, session: test-session, base: Environment)" ); + // Test Profile provider with and without overrides + let profile_metadata = CredentialProviderMetadata::Profile { + name: None, + file: None, + }; + assert_eq!(profile_metadata.simple_string(), "Profile"); + let profile_metadata = CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: None, + }; + assert_eq!(profile_metadata.simple_string(), "Profile(name: analytics)"); + let profile_metadata = CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: Some("/etc/aws/credentials".to_string()), + }; + assert_eq!( + profile_metadata.simple_string(), + "Profile(name: analytics, file: /etc/aws/credentials)" + ); + // Test Chain provider let chain_metadata = CredentialProviderMetadata::Chain(vec![ CredentialProviderMetadata::Static {