dwsmith1983 commented on code in PR #6059:
URL: https://github.com/apache/datafusion-comet/pull/6059#discussion_r4185939514
##########
native/core/src/parquet/objectstore/azure.rs:
##########
@@ -118,62 +472,611 @@ pub fn create_store(
.map(|(k, _)| k.as_ref())
.collect::<Vec<_>>()
);
+
+ let env: Vec<(String, String)> = env.collect();
+ validate_translated(
+ configs,
+ &translated,
+ account.as_deref(),
+ url_container,
+ oauth_keys,
+ env_token_file(&env).is_some(),
+ )?;
+ let store = build_builder(
+ url,
+ configs,
+ account.as_deref(),
+ 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 SAS fixed token 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, oauth_keys))
+ .or_else(|| unread_container_token_problem(configs, account,
container, translated))
+}
+
+/// A container-scoped SAS fixed token left unread
(`has_unread_container_fixed_token`) when
+/// nothing else supplies a credential. With the flag set the native scan
would select this
+/// token, while Hadoop, defaulting to SharedKey here, fails without an
account key, so the
+/// scan errors rather than build a store with no credential, which
`object_store` completes
+/// with the managed identity. The MSI provider class stands alone, so it
counts as a
+/// credential.
+fn unread_container_token_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<UrlContainer<'_>>,
+ 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
+ || !has_unread_container_fixed_token(configs, account, container)
+ {
+ return None;
+ }
+ let (key, _) = container_fixed_entry(configs, container?)?;
+ Some(format!(
+ "`{key}` is read only by Hadoop 3.4.2 and later; set
`{HADOOP_SAS_FIXED_TOKEN}` 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, _, _)| {
+ account_scoped_entry(configs, base, account)
+ .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_fixed_token) {
+ ", 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>,
+ 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 {
+ account_scoped_entry(configs, base, account)
+ };
+ entry
+ .filter(|(_, value)| {
+ !(*base == HADOOP_KEY_PROVIDER &&
is_simple_key_provider(value))
+ })
+ .map(|(key, _)| {
+ 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)
+}
+
+/// 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 SAS fixed token left unread
+/// (`has_unread_container_fixed_token`).
+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);
+ // `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
+ || has_unread_container_fixed_token(configs, account, container)
+ || 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.
+/// keys (`fs.azure.X.<account>.<endpoint-suffix>`) and the container-scoped
SAS keys
+/// (`fs.azure.sas.fixed.token.<container>.<host>`, when
`container_fixed_token` is set, and
+/// `fs.azure.sas.<container>.<account>`).
+/// Account-scoped keys win over global ones.
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 {
+ // 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) = account_scoped_value(configs, hadoop_base,
account) {
- out.push((*azure_key, value));
+ if !value.trim().is_empty() {
+ out.push((*azure_key, value));
+ }
}
}
// `fs.azure.account.oauth2.client.endpoint` is a full token URL of the
form
// `https://login.microsoftonline.com/<tenant>/oauth2/token`. object_store
wants the
// tenant id directly (`AuthorityId`), so extract it if AuthorityId hasn't
already
- // been set from `fs.azure.account.oauth2.msi.tenant`.
- let has_authority_id = out
- .iter()
- .any(|(k, _)| matches!(k, AzureConfigKey::AuthorityId));
- if !has_authority_id {
+ // been set from `fs.azure.account.oauth2.msi.tenant`. For client
credentials Hadoop
+ // posts to the endpoint itself, so its origin also becomes
`AuthorityHost` unless
+ // `msi.authority` set it. That holds only when object_store will use
client
+ // credentials: a client secret and no token file, since a token file
selects Workload
+ // Identity first. Otherwise the endpoint is not a token URL, so its host
must not
+ // receive another credential.
+ let has = |out: &[(AzureConfigKey, String)], wanted: AzureConfigKey| {
+ out.iter().any(|(k, _)| *k == wanted)
+ };
+ let has_authority_id = has(&out, AzureConfigKey::AuthorityId);
+ let endpoint_is_token_url = has(&out, AzureConfigKey::ClientSecret)
+ && !has(&out, AzureConfigKey::FederatedTokenFile)
+ && !has(&out, AzureConfigKey::AuthorityHost);
+ if !has_authority_id
+ && key_is_read(
+ configs,
+ account,
+ oauth_keys,
+ HADOOP_OAUTH_CLIENT_ENDPOINT,
+ AuthMechanism::OAuth,
+ )
+ {
if let Some(endpoint) = account_scoped_value(configs,
HADOOP_OAUTH_CLIENT_ENDPOINT, account)
{
if let Some(tenant) = tenant_from_oauth_endpoint(&endpoint) {
out.push((AzureConfigKey::AuthorityId, tenant));
+ if let Some(host) =
+ authority_host_from_oauth_endpoint(&endpoint).filter(|_|
endpoint_is_token_url)
Review Comment:
The `/gateway/oauth2/<tenant>/oauth2/v2.0/token` case from your later review
is fixed in 18455468a: the tenant is now the segment before the last `oauth2`,
so a prefix that itself contains `oauth2` is kept and the token URL matches the
endpoint. Your configuration is a test that records the token request, and
endpoints with only one `oauth2` segment are unchanged.
--
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]
