dwsmith1983 commented on code in PR #6059:
URL: https://github.com/apache/datafusion-comet/pull/6059#discussion_r4104556834
##########
native/core/src/parquet/objectstore/azure.rs:
##########
@@ -118,12 +308,440 @@ 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(),
+ container.as_deref(),
+ env_token_file(&env).is_some(),
+ )?;
+ let store = build_builder(
+ url,
+ configs,
+ account.as_deref(),
+ container.as_deref(),
+ &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<&str>,
+ 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,
translated) {
+ return fail(reason);
+ }
+ let has = |wanted: AzureConfigKey| translated.iter().any(|(key, _)| *key
== wanted);
+ let borrows_env_token_file = env_policy(configs, account, container,
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) {
+ missing.push(format!(
+ "`{HADOOP_MSI_TENANT}` or `{HADOOP_OAUTH_CLIENT_ENDPOINT}`"
+ ));
+ }
+ 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.
+fn hadoop_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<&str>,
+ translated: &[(AzureConfigKey, String)],
+) -> Option<String> {
+ blank_value_problem(configs, account, container)
+ .or_else(|| auth_type_problem(configs, account, translated))
+ .or_else(|| provider_class_problem(configs, account, translated))
+ .or_else(|| unsupported_key_problem(configs, account))
+}
+
+/// 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 exception is the client id and tenant under `MsiTokenProvider`: Hadoop
accepts
+/// empty strings there, and a system-assigned identity sets them that way, so
they are
+/// absent and the builder proceeds to the managed identity endpoint with no
client id.
+fn blank_value_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+ container: Option<&str>,
+) -> Option<String> {
+ if let Some((key, value)) = active_sas_token(configs, account, container) {
+ if value.trim().is_empty() {
+ let fallback = if key.starts_with(HADOOP_SAS_FIXED_TOKEN) {
+ String::new()
+ } else {
+ format!(
+ "; `{HADOOP_SAS_FIXED_TOKEN}` is not used as a fallback
when a \
+ container-scoped SAS key is set"
+ )
+ };
+ return Some(format!("`{key}` is blank{fallback}"));
+ }
+ }
+ let msi_provider = active_provider_class(configs, account)
+ .is_some_and(|(_, class)| is_provider_class(&class,
HADOOP_MSI_PROVIDER_CLASS));
+ HADOOP_CREDENTIAL_MAPPINGS
+ .iter()
+ .filter(|(_, _, mechanism)| mechanism_is_read(configs, account,
*mechanism))
+ .filter(|(base, _, _)| !(msi_provider &&
HADOOP_MSI_OPTIONAL_KEYS.contains(base)))
+ .find_map(|(base, _, _)| {
+ account_scoped_entry(configs, base, account)
+ .filter(|(_, value)| value.trim().is_empty())
+ .map(|(key, _)| format!("`{key}` is blank"))
+ })
+}
+
+/// 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>,
+ 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") {
+ return (!has(AzureConfigKey::SasKey)).then(|| {
+ format!(
+ "{setting} needs `{HADOOP_SAS_FIXED_TOKEN}`; 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.
+fn unsupported_key_problem(
+ configs: &HashMap<String, String>,
+ account: Option<&str>,
+) -> Option<String> {
+ HADOOP_UNSUPPORTED_MECHANISM_KEYS
+ .iter()
+ .filter(|(_, mechanism)| mechanism_is_read(configs, account,
*mechanism))
Review Comment:
Done. Once `fs.azure.account.oauth.provider.type` resolves to
`ClientCredsTokenProvider`, `MsiTokenProvider` or
`WorkloadIdentityTokenProvider`, only the OAuth keys that class reads in
`AbfsConfiguration.getTokenProvider` are translated or checked. Client
credentials read the client endpoint, id and secret. MSI reads the MSI
endpoint, tenant, client id and authority. Workload Identity reads the
authority, tenant, client id and token file. Every other
`fs.azure.account.oauth2.*` key is inactive for that account. It is neither
validated nor handed to the builder, so an unused `user.password` or a blank
`msi.endpoint` no longer rejects a complete client-credentials configuration.
With no provider class, or one the scan rejects anyway, every OAuth key is
still read as before.
Scoping the translation as well closes the same gap on the builder side.
Under client credentials the tenant now comes from `client.endpoint` alone, as
in Hadoop, so a global `msi.tenant` kept for another provider no longer
replaces it. Under MSI an unused client secret no longer turns the managed
identity into a client-secret credential. Under Workload Identity an unused MSI
endpoint no longer stops the token file from being taken from
`AZURE_FEDERATED_TOKEN_FILE`.
A `ClientCredsTokenProvider` configuration with a client id and secret but
no `fs.azure.account.oauth2.client.endpoint`, whose tenant came only from
`fs.azure.account.oauth2.msi.tenant`, used to build and now fails, which
matches Hadoop, where `AbfsConfiguration.getTokenProvider` reads the endpoint
with `getMandatoryPasswordString(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT)` for
that class.
The new tests cover each of those cases and a shared configuration carrying
user-password, refresh-token and MSI settings beside an active
client-credentials or Workload Identity provider. A blank secret under
`ClientCredsTokenProvider` itself is still rejected.
--
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]
