dwsmith1983 commented on code in PR #6059:
URL: https://github.com/apache/datafusion-comet/pull/6059#discussion_r4206439710
##########
native/core/src/parquet/objectstore/azure.rs:
##########
@@ -118,112 +505,819 @@ pub fn create_store(
.map(|(k, _)| k.as_ref())
.collect::<Vec<_>>()
);
+
+ let env: Vec<(String, String)> = env.collect();
+ validate_translated(
+ configs,
+ &translated,
+ account,
+ url_container,
+ oauth_keys,
+ env_token_file(&env).is_some(),
+ )?;
+ let store = build_builder(
+ url,
+ configs,
+ account,
+ url_container,
+ oauth_keys,
+ &translated,
+ env.into_iter(),
+ )
+ .build()?;
+ Ok((Box::new(store), path))
+}
+
+fn config_error(message: String) -> object_store::Error {
+ object_store::Error::Generic {
+ store: "MicrosoftAzure",
+ source: message.into(),
+ }
+}
+
+/// Reject a Hadoop configuration that `object_store` would silently build a
different
+/// identity from: a blank credential, an auth type or mechanism the native
scan cannot
+/// build, a named principal with no token file in Hadoop or the environment,
or a client
+/// secret or token file without the client id and tenant that complete it.
+/// `has_env_token_file` says whether `AZURE_FEDERATED_TOKEN_FILE` is set.
+fn validate_translated(
+ configs: &HashMap<String, String>,
+ translated: &[(AzureConfigKey, String)],
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+ has_env_token_file: bool,
+) -> Result<(), object_store::Error> {
+ let account_name = account.unwrap_or("<unknown>");
+ let fail = |reason: String| {
+ Err(config_error(format!(
+ "Hadoop configuration for account {account_name}: {reason}"
+ )))
+ };
+ if let Some(reason) = hadoop_problem(configs, account, container,
oauth_keys, translated) {
+ return fail(reason);
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let borrows_env_token_file = env_policy(configs, account, container,
oauth_keys, translated)
+ == EnvPolicy::TokenFileOnly
+ && !has(AzureConfigKey::FederatedTokenFile);
+ if borrows_env_token_file && !has_env_token_file {
+ return fail(format!(
+ "the principal named by the Hadoop keys needs a token file from \
+ `{HADOOP_WI_TOKEN_FILE}` or `{ENV_FEDERATED_TOKEN_FILE}`"
+ ));
+ }
+ let mechanism = if has(AzureConfigKey::ClientSecret) {
+ HADOOP_OAUTH_CLIENT_SECRET
+ } else if has(AzureConfigKey::FederatedTokenFile) {
+ HADOOP_WI_TOKEN_FILE
+ } else if borrows_env_token_file {
+ ENV_FEDERATED_TOKEN_FILE
+ } else {
+ return Ok(());
+ };
+ let mut missing = Vec::new();
+ if !has(AzureConfigKey::ClientId) {
+ missing.push(format!("`{HADOOP_OAUTH_CLIENT_ID}`"));
+ }
+ if !has(AzureConfigKey::AuthorityId) {
+ let tenant_keys: Vec<String> = [HADOOP_MSI_TENANT,
HADOOP_OAUTH_CLIENT_ENDPOINT]
+ .into_iter()
+ .filter(|base| key_is_read(configs, account, oauth_keys, base,
AuthMechanism::OAuth))
+ .map(|base| format!("`{base}`"))
+ .collect();
+ missing.push(tenant_keys.join(" or "));
+ }
+ if missing.is_empty() {
+ return Ok(());
+ }
+ fail(format!(
+ "`{mechanism}` also needs {}",
+ missing.join(" and ")
+ ))
+}
+
+/// Why the Hadoop keys cannot be built natively as configured, or `None` when
they can:
+/// a blank value first, then an explicit auth type the translated keys do not
satisfy,
+/// then a provider class they do not satisfy, then a key that selects a
mechanism with no
+/// native counterpart, then an unread container-scoped key with no other
credential.
+fn hadoop_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ blank_value_problem(configs, account, container, oauth_keys)
+ .or_else(|| auth_type_problem(configs, account, container, translated))
+ .or_else(|| provider_class_problem(configs, account, translated))
+ .or_else(|| key_provider_problem(configs, account, translated))
+ .or_else(|| unsupported_key_problem(configs, account, container,
oauth_keys))
+ .or_else(|| {
+ unread_container_key_problem(configs, account, container,
oauth_keys, translated)
+ })
+}
+
+/// A container-scoped key left unread (`unread_container_entry`) when nothing
else supplies
+/// a credential. With the flag set the native scan would read it, while
Hadoop 3.4.1 fails
+/// with no credential, so the scan errors rather than build a store with
none, which
+/// `object_store` completes with the managed identity. The MSI provider class
stands alone,
+/// so it counts as a credential, and so does a Workload Identity principal
named at account
+/// or global level (`names_workload_identity`), which Hadoop 3.4.1 completes
with its
+/// default token path. A key in `HADOOP_UNSUPPORTED_MECHANISM_KEYS` is
reported as that
+/// mechanism, since its account-scoped form is rejected too.
+fn unread_container_key_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ let msi_provider = active_provider_class(configs, account)
+ .is_some_and(|(_, class)| is_provider_class(&class,
HADOOP_MSI_PROVIDER_CLASS));
+ if has_mechanism_key(translated)
+ || msi_provider
+ || names_workload_identity(configs, account, translated)
+ {
+ return None;
+ }
+ let (base, key) = unread_container_entry(configs, account, container,
oauth_keys)?;
+ if is_unsupported_mechanism_key(base) {
+ return Some(unsupported_mechanism_message(&key));
+ }
+ Some(format!(
+ "`{key}` is read only by Hadoop 3.4.2 and later; set `{base}` or its
account-scoped \
+ form, or remove it"
+ ))
+}
+
+/// Whether the translated keys hold a credential that selects a mechanism on
its own.
+fn has_mechanism_key(translated: &[(AzureConfigKey, String)]) -> bool {
+ translated.iter().any(|(key, _)| {
+ matches!(
+ key,
+ AzureConfigKey::AccessKey
+ | AzureConfigKey::SasKey
+ | AzureConfigKey::FederatedTokenFile
+ | AzureConfigKey::ClientSecret
+ | AzureConfigKey::MsiEndpoint
+ )
+ })
+}
+
+/// Whether an explicit `SimpleKeyProvider` has the account key it reads.
Hadoop fails
+/// when that key is missing, so the scan must not proceed with another
credential.
+fn key_provider_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ if !mechanism_is_read(configs, account, AuthMechanism::SharedKey) {
+ return None;
+ }
+ let (key, class) = account_scoped_entry(configs, HADOOP_KEY_PROVIDER,
account)?;
+ let has_key = translated
+ .iter()
+ .any(|(key, _)| *key == AzureConfigKey::AccessKey);
+ (is_simple_key_provider(&class) && !has_key)
+ .then(|| format!("`{key}={class}` needs `{HADOOP_KEY}`"))
+}
+
+/// A blank SAS token or credential value, named by the exact key that holds
it.
+///
+/// Blank values are errors rather than absent, so a templated configuration
that
+/// substitutes an empty string fails loudly instead of silently using another
credential.
+/// The exceptions are the keys the active provider class accepts blank
+/// (`optional_provider_keys`): Hadoop reads them with a default or as plain
strings, so
+/// they are absent and the builder proceeds as it does when they are unset.
+fn blank_value_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+) -> Option<String> {
+ if let Some((key, value)) = active_sas_token(configs, account, container) {
+ if value.trim().is_empty() {
+ return Some(format!("`{key}` is blank"));
+ }
+ }
+ let optional = optional_provider_keys(configs, account);
+ HADOOP_CREDENTIAL_MAPPINGS
+ .iter()
+ .filter(|(base, _, mechanism)| key_is_read(configs, account,
oauth_keys, base, *mechanism))
+ .filter(|(base, _, _)| !optional.contains(base))
+ .find_map(|(base, _, _)| {
+ hadoop_entry(configs, base, account, container)
+ .filter(|(_, value)| value.trim().is_empty())
+ .map(|(key, _)| format!("`{key}` is blank"))
+ })
+}
+
+/// The keys the active provider class accepts blank, or none when there is no
provider
+/// class or one the scan rejects.
+fn optional_provider_keys(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+) -> &'static [&'static str] {
+ let Some((_, class)) = active_provider_class(configs, account) else {
+ return &[];
+ };
+ if is_provider_class(&class, HADOOP_MSI_PROVIDER_CLASS) {
+ HADOOP_MSI_OPTIONAL_KEYS
+ } else if is_provider_class(&class, HADOOP_WI_PROVIDER_CLASS) {
+ HADOOP_WI_OPTIONAL_KEYS
+ } else {
+ &[]
+ }
+}
+
+/// Whether an explicit `fs.azure.account.auth.type` is one the translated
keys satisfy.
+///
+/// Setting it is Hadoop choosing a mechanism, so it is validated rather than
ignored:
+/// `SharedKey` needs the account key, `OAuth` a provider the scan can build,
`SAS` a SAS
+/// token, `Custom` has no native counterpart and any other value is a typo.
+fn auth_type_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ let (key, value) = account_scoped_entry(configs, HADOOP_AUTH_TYPE,
account)?;
+ let auth_type = value.trim();
+ if auth_type.is_empty() {
+ return Some(format!("`{key}` is blank"));
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let setting = format!("`{key}={auth_type}`");
+ if auth_type.eq_ignore_ascii_case("SharedKey") {
+ return (!has(AzureConfigKey::AccessKey))
+ .then(|| format!("{setting} needs `{HADOOP_KEY}`"));
+ }
+ if auth_type.eq_ignore_ascii_case("OAuth") {
+ return oauth_problem(configs, account, translated, &setting);
+ }
+ if auth_type.eq_ignore_ascii_case("SAS") {
+ let container_form = if container.is_some_and(|c| c.container_scoped) {
+ ", its container-scoped form"
+ } else {
+ ""
+ };
+ return (!has(AzureConfigKey::SasKey)).then(|| {
+ format!(
+ "{setting} needs a SAS token from
`{HADOOP_SAS_FIXED_TOKEN}`{container_form} or \
+ the WASB container key; a SAS token provider class is not
supported by the \
+ native scan"
+ )
+ });
+ }
+ if auth_type.eq_ignore_ascii_case("Custom") {
+ return Some(format!(
+ "{setting} loads a custom token provider class, which the native
scan does not \
+ support"
+ ));
+ }
+ Some(format!(
+ "{setting} is not supported; the native scan supports
`{HADOOP_AUTH_TYPE}` values \
+ SharedKey, OAuth and SAS"
+ ))
+}
+
+/// Whether `fs.azure.account.auth.type=OAuth` can be satisfied: through the
provider class
+/// when one is set, otherwise through a translated secret or token file.
+fn oauth_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+ setting: &str,
+) -> Option<String> {
+ if active_provider_class(configs, account).is_some() {
+ return provider_class_problem(configs, account, translated);
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ if has(AzureConfigKey::ClientSecret) ||
has(AzureConfigKey::FederatedTokenFile) {
+ return None;
+ }
+ Some(format!(
+ "{setting} needs `{HADOOP_OAUTH_CLIENT_SECRET}` or
`{HADOOP_WI_TOKEN_FILE}`"
+ ))
+}
+
+/// Whether a `fs.azure.account.oauth.provider.type` class, validated whenever
it is set
+/// and OAuth is in use, names a token provider the native scan can
+/// satisfy: MSI stands alone, Workload Identity needs the client id and
tenant (the token
+/// file may still come from `AZURE_FEDERATED_TOKEN_FILE`), client credentials
need the
+/// secret, and any other class has no native counterpart.
+fn provider_class_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ let (provider_key, class) = active_provider_class(configs, account)?;
+ let class = class.trim();
+ if class.is_empty() {
+ return Some(format!("`{provider_key}` is blank"));
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let provider = format!("`{provider_key}={class}`");
+ if is_provider_class(class, HADOOP_MSI_PROVIDER_CLASS) {
+ return None;
+ }
+ if is_provider_class(class, HADOOP_WI_PROVIDER_CLASS) {
+ return (!workload_identity_named(configs, account,
translated)).then(|| {
+ format!("{provider} needs `{HADOOP_OAUTH_CLIENT_ID}` and
`{HADOOP_MSI_TENANT}`")
+ });
+ }
+ if is_provider_class(class, HADOOP_CLIENT_CREDS_PROVIDER_CLASS) {
+ return (!has(AzureConfigKey::ClientSecret))
+ .then(|| format!("{provider} needs
`{HADOOP_OAUTH_CLIENT_SECRET}`"));
+ }
+ Some(format!(
+ "{provider} is not a token provider the native scan supports"
+ ))
+}
+
+/// The first Hadoop key present that selects a mechanism with no native
counterpart,
+/// named exactly as the user set it. An explicit `SimpleKeyProvider` is the
default key
+/// provider, not a separate mechanism. The SAS token provider class is looked
up as
+/// `token_provider_entry` does, since Hadoop reads it through
`getTokenProviderClass`.
+fn unsupported_key_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+) -> Option<String> {
+ HADOOP_UNSUPPORTED_MECHANISM_KEYS
+ .iter()
+ .filter(|(base, mechanism)| key_is_read(configs, account, oauth_keys,
base, *mechanism))
+ .find_map(|(base, _)| {
+ let entry = if *base == HADOOP_SAS_TOKEN_PROVIDER_TYPE {
+ token_provider_entry(configs, base, account)
+ } else {
+ hadoop_entry(configs, base, account, container)
+ };
+ entry
+ .filter(|(_, value)| {
+ !(*base == HADOOP_KEY_PROVIDER &&
is_simple_key_provider(value))
+ })
+ .map(|(key, _)| unsupported_mechanism_message(&key))
+ })
+}
+
+/// Whether `base` is in `HADOOP_UNSUPPORTED_MECHANISM_KEYS`.
+fn is_unsupported_mechanism_key(base: &str) -> bool {
+ HADOOP_UNSUPPORTED_MECHANISM_KEYS
+ .iter()
+ .any(|(key, _)| *key == base)
+}
+
+/// The error for `key`, named exactly as the user set it, selecting a
mechanism with no
+/// native counterpart.
+fn unsupported_mechanism_message(key: &str) -> String {
+ format!("`{key}` selects an authentication mechanism the native scan does
not support")
+}
+
+/// Whether `class` is the Hadoop token provider with `simple_name`: the bare
simple name,
+/// or a qualified name whose last segment is exactly it, so that a lookalike
such as
+/// `com.attacker.EvilWorkloadIdentityTokenProvider` does not pass as the real
class. The
+/// value is trimmed because Hadoop reads it with `Configuration.getClass`,
which calls
+/// `getTrimmed` before loading the class.
+fn is_provider_class(class: &str, simple_name: &str) -> bool {
+ let class = class.trim();
+ class == simple_name
+ || class
+ .strip_suffix(simple_name)
+ .is_some_and(|prefix| prefix.ends_with('.'))
+}
+
+/// Whether Hadoop names the Workload Identity provider and the principal,
which leaves
+/// only the token file open for `AZURE_FEDERATED_TOKEN_FILE` to supply.
+fn workload_identity_named(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> bool {
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ active_provider_class(configs, account)
+ .is_some_and(|(_, class)| is_provider_class(&class,
HADOOP_WI_PROVIDER_CLASS))
+ && has(AzureConfigKey::ClientId)
+ && has(AzureConfigKey::AuthorityId)
+}
+
+/// Whether the translated keys name a Workload Identity principal: the client
id and
+/// tenant, with the Workload Identity provider class or no class at all. With
the
+/// container-scoped forms unread, these are the account-level and global
keys: under the class
+/// Hadoop 3.4.1 reads them and completes the principal with its default token
path, and without
+/// a class the named-identity rule applies, so an unread container-scoped key
beside them does
+/// not close the environment (`env_policy`) or fail the configuration
+/// (`unread_container_key_problem`).
+fn names_workload_identity(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> bool {
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ active_provider_class(configs, account)
+ .is_none_or(|(_, class)| is_provider_class(&class,
HADOOP_WI_PROVIDER_CLASS))
+ && has(AzureConfigKey::ClientId)
+ && has(AzureConfigKey::AuthorityId)
+}
+
+/// Process environment as UTF-8 `(key, value)` pairs, skipping entries that
are not UTF-8.
+fn env_pairs() -> impl Iterator<Item = (String, String)> {
+ std::env::vars_os()
+ .filter_map(|(k, v)| Some((k.to_str()?.to_string(),
v.to_str()?.to_string())))
+}
+
+/// Assemble the builder from the environment, the URL and the translated
Hadoop keys.
+///
+/// What the environment may contribute is decided by `env_policy`: everything
when the
+/// Hadoop keys name no identity, only the token file when they name a
principal, and
+/// nothing when they select a mechanism, so nothing ambient can outrank,
combine with or
+/// redirect the configured identity. `configs`, `account` and `container`
resolve the
+/// Hadoop keys that select a mechanism without translating to anything.
+fn build_builder(
+ url: &Url,
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+ translated: &[(AzureConfigKey, String)],
+ env: impl Iterator<Item = (String, String)>,
+) -> MicrosoftAzureBuilder {
+ let mut builder = MicrosoftAzureBuilder::new();
+ match env_policy(configs, account, container, oauth_keys, translated) {
+ EnvPolicy::All => builder = apply_env(builder, env),
+ EnvPolicy::TokenFileOnly => builder = apply_env_token_file(builder,
translated, env),
+ EnvPolicy::Nothing => {}
+ }
+ builder = builder.with_url(url.to_string());
for (key, value) in translated {
- builder = builder.with_config(key, value);
+ builder = builder.with_config(*key, value.clone());
}
+ builder
+}
- let store = builder.build()?;
- Ok((Box::new(store), path))
+/// How much of the environment the builder may read, decided by the Hadoop
keys.
+#[derive(Debug, PartialEq)]
+enum EnvPolicy {
+ /// Hadoop names neither an identity nor a mechanism: read it as
`from_env` does.
+ All,
+ /// Hadoop names a principal but not its token source: read
`AZURE_FEDERATED_TOKEN_FILE`
+ /// only, which is what AKS Workload Identity injects.
+ TokenFileOnly,
+ /// Hadoop selects a mechanism, or a configuration the scan rejects: read
nothing.
+ Nothing,
+}
+
+/// Decide `EnvPolicy` from the Hadoop keys.
+///
+/// A translated credential key, a provider class (the MSI one stands alone,
since Hadoop
+/// defaults its endpoint to IMDS without any translated key), an explicit
+/// `fs.azure.account.auth.type` or a rejected configuration selects the
mechanism. A
+/// client id, tenant or authority host, or the Workload Identity provider
with the client
+/// id and tenant (with or without `auth.type=OAuth`), names a principal and
leaves only the
+/// token file open. A partial mechanism, such as a token file alone, is never
completed
+/// from the environment, nor is a container-scoped key left unread
+/// (`unread_container_entry`), unless a Workload Identity principal is named
beside it
+/// (`names_workload_identity`), which leaves the token file open as usual.
+fn env_policy(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
+ translated: &[(AzureConfigKey, String)],
+) -> EnvPolicy {
+ let has_mechanism_key = has_mechanism_key(translated);
+ // A provider class or an explicit auth type is Hadoop choosing a
mechanism; only the
+ // Workload Identity provider with the client id and tenant leaves the
token file open.
+ let has_provider_class =
+ active_provider_class(configs, account).is_some_and(|(_, class)|
!class.trim().is_empty());
+ let has_auth_type = account_scoped_value(configs, HADOOP_AUTH_TYPE,
account).is_some();
+ let chooses_mechanism = (has_provider_class || has_auth_type)
+ && !workload_identity_named(configs, account, translated);
+ let unread_container_key = !names_workload_identity(configs, account,
translated)
+ && unread_container_entry(configs, account, container,
oauth_keys).is_some();
+ // `hadoop_problem` is checked here as well as in `validate_translated` so
that
+ // `build_builder` never borrows the environment even when called without
validation.
+ if has_mechanism_key
+ || chooses_mechanism
+ || unread_container_key
+ || hadoop_problem(configs, account, container, oauth_keys,
translated).is_some()
+ {
+ return EnvPolicy::Nothing;
+ }
+ // A named Workload Identity provider implies a translated client id and
tenant, so
+ // this covers it too.
+ let names_identity = translated.iter().any(|(key, _)| {
+ matches!(
+ key,
+ AzureConfigKey::ClientId | AzureConfigKey::AuthorityId |
AzureConfigKey::AuthorityHost
+ )
+ });
+ if names_identity {
+ EnvPolicy::TokenFileOnly
+ } else {
+ EnvPolicy::All
+ }
+}
+
+/// Borrow `AZURE_FEDERATED_TOKEN_FILE` alone, and only when Hadoop supplied
no token file.
+fn apply_env_token_file(
+ builder: MicrosoftAzureBuilder,
+ translated: &[(AzureConfigKey, String)],
+ env: impl Iterator<Item = (String, String)>,
+) -> MicrosoftAzureBuilder {
+ let has_token_file = translated
+ .iter()
+ .any(|(key, _)| *key == AzureConfigKey::FederatedTokenFile);
+ if has_token_file {
+ return builder;
+ }
+ let env: Vec<(String, String)> = env.collect();
+ match env_token_file(&env) {
+ Some(file) => builder.with_config(AzureConfigKey::FederatedTokenFile,
file),
+ None => builder,
+ }
+}
+
+/// The non-blank value of `AZURE_FEDERATED_TOKEN_FILE` in `env`, if any.
+fn env_token_file(env: &[(String, String)]) -> Option<String> {
+ env.iter()
+ .find(|(key, _)| key == ENV_FEDERATED_TOKEN_FILE)
+ .map(|(_, value)| value.clone())
+ .filter(|value| !value.trim().is_empty())
+}
+
+/// Apply the environment to `builder` the way
`MicrosoftAzureBuilder::from_env` does:
+/// every `AZURE_*` variable that parses as an `AzureConfigKey`, then the MSI
endpoint
+/// variable, which is applied last so it wins over `AZURE_MSI_ENDPOINT`
regardless of the
+/// order the environment is iterated in.
+fn apply_env(
+ mut builder: MicrosoftAzureBuilder,
+ env: impl Iterator<Item = (String, String)>,
+) -> MicrosoftAzureBuilder {
+ let mut msi_endpoint: Option<String> = None;
+ for (key, value) in env {
+ if key == MSI_ENDPOINT_ENV_KEY {
+ msi_endpoint = Some(value);
+ continue;
+ }
+ if !key.starts_with(AZURE_ENV_PREFIX) {
+ continue;
+ }
+ if let Ok(config_key) =
key.to_ascii_lowercase().parse::<AzureConfigKey>() {
+ builder = builder.with_config(config_key, value);
+ }
+ }
+ if let Some(endpoint) = msi_endpoint {
+ builder = builder.with_msi_endpoint(endpoint);
+ }
+ builder
}
/// Translate a Hadoop ABFS configuration map into `(AzureConfigKey, value)`
pairs.
///
-/// `account` and `container` are extracted from the URL and used to resolve
account-scoped
-/// keys (`fs.azure.X.<account>.<endpoint-suffix>`) and the SAS namespace
-/// (`fs.azure.sas.<container>.<account>`). Account-scoped keys win over
global ones.
+/// `account` and `container` are extracted from the URL and used to resolve
each key the
+/// way Hadoop does (`hadoop_entry`): the container-scoped form
`<key>.<container>.<host>`
+/// first for the keys Hadoop reads that way, when `container_scoped` is set,
then the
+/// account-scoped form `<key>.<account>` (with the account name of
`extract_account`), then
+/// the global key. The WASB container key
`fs.azure.sas.<container>.<account>` is the SAS
+/// fallback.
fn translate_hadoop_configs(
configs: &HashMap<String, String>,
account: Option<&str>,
- container: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ oauth_keys: OAuthKeys,
) -> Vec<(AzureConfigKey, String)> {
let mut out: Vec<(AzureConfigKey, String)> = Vec::new();
- let mappings: &[(&str, AzureConfigKey)] = &[
- (HADOOP_KEY, AzureConfigKey::AccessKey),
- (HADOOP_OAUTH_CLIENT_ID, AzureConfigKey::ClientId),
- (HADOOP_OAUTH_CLIENT_SECRET, AzureConfigKey::ClientSecret),
- (HADOOP_MSI_TENANT, AzureConfigKey::AuthorityId),
- (HADOOP_MSI_ENDPOINT, AzureConfigKey::MsiEndpoint),
- (HADOOP_MSI_AUTHORITY, AzureConfigKey::AuthorityHost),
- (HADOOP_WI_TOKEN_FILE, AzureConfigKey::FederatedTokenFile),
- ];
-
- for (hadoop_base, azure_key) in mappings {
- if let Some(value) = account_scoped_value(configs, hadoop_base,
account) {
- out.push((*azure_key, value));
+ // A blank value is left out so it never reaches the builder; validation
reports it
+ // as an error before the store is built.
+ for (hadoop_base, azure_key, mechanism) in HADOOP_CREDENTIAL_MAPPINGS {
+ if !key_is_read(configs, account, oauth_keys, hadoop_base, *mechanism)
{
+ continue;
+ }
+ if let Some((_, value)) = hadoop_entry(configs, hadoop_base, account,
container) {
+ if !value.trim().is_empty() {
+ out.push((*azure_key, value));
Review Comment:
The resolver now takes the token file the way Hadoop hands it to
`WorkloadIdentityTokenProvider`, through `getTrimmedPasswordString`, so `"
/tmp/token "` arrives as `/tmp/token`. `AbfsAuthParitySuite` pins it against
hadoop-azure's own `FileSystem` on every supported version
(`oauth-wi-padded-token-file`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
