feat(io): support opt-in anonymous access for ADLS/Azure Blob storage

crates/iceberg/Cargo.toml

@@ -32,7 +32,7 @@ repository = { workspace = true }
 [features]
 default = ["storage-memory", "storage-fs", "storage-s3"]
 storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
-storage-azdls = ["opendal/services-azdls"]
+storage-azdls = ["opendal/services-azdls", "opendal/services-http"]
 storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]

crates/iceberg/public-api.txt

@@ -909,6 +909,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::StorageConfig
 pub fn iceberg::io::StorageConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub const iceberg::io::config::ADLS_ACCOUNT_KEY: &str
 pub const iceberg::io::config::ADLS_ACCOUNT_NAME: &str
+pub const iceberg::io::config::ADLS_ALLOW_ANONYMOUS: &str
 pub const iceberg::io::config::ADLS_AUTHORITY_HOST: &str
 pub const iceberg::io::config::ADLS_CLIENT_ID: &str
 pub const iceberg::io::config::ADLS_CLIENT_SECRET: &str
@@ -952,6 +953,7 @@ pub use iceberg::io::opendal::AwsCredential
 pub use iceberg::io::opendal::AwsCredentialLoad
 pub enum iceberg::io::opendal::OpenDalStorage
 pub iceberg::io::opendal::OpenDalStorage::Azdls
+pub iceberg::io::opendal::OpenDalStorage::Azdls::allow_anonymous: bool
 pub iceberg::io::opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc<opendal_service_azdls::config::AzdlsConfig>
 pub iceberg::io::opendal::OpenDalStorage::Azdls::configured_scheme: iceberg::io::storage::opendal::azdls::AzureStorageScheme
 pub iceberg::io::opendal::OpenDalStorage::Gcs
@@ -1048,6 +1050,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::opendal::Refreshable
 pub fn iceberg::io::opendal::RefreshableStorageFactory::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub enum iceberg::io::OpenDalStorage
 pub iceberg::io::OpenDalStorage::Azdls
+pub iceberg::io::OpenDalStorage::Azdls::allow_anonymous: bool
 pub iceberg::io::OpenDalStorage::Azdls::config: alloc::sync::Arc<opendal_service_azdls::config::AzdlsConfig>
 pub iceberg::io::OpenDalStorage::Azdls::configured_scheme: iceberg::io::storage::opendal::azdls::AzureStorageScheme
 pub iceberg::io::OpenDalStorage::Gcs
@@ -1459,6 +1462,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::StorageCredential
 pub fn iceberg::io::StorageCredential::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub const iceberg::io::ADLS_ACCOUNT_KEY: &str
 pub const iceberg::io::ADLS_ACCOUNT_NAME: &str
+pub const iceberg::io::ADLS_ALLOW_ANONYMOUS: &str
 pub const iceberg::io::ADLS_AUTHORITY_HOST: &str
 pub const iceberg::io::ADLS_CLIENT_ID: &str
 pub const iceberg::io::ADLS_CLIENT_SECRET: &str

crates/iceberg/src/io/storage/config/azdls.rs

@@ -48,6 +48,14 @@ pub const ADLS_CLIENT_SECRET: &str = "adls.client-secret";
 pub const ADLS_AUTHORITY_HOST: &str = "adls.authority-host";
 /// The endpoint of the storage account.
 pub const ADLS_ENDPOINT: &str = "adls.endpoint";
+/// Whether to allow genuinely anonymous, unsigned access when no credential
+/// is configured (e.g. a container with public read access). Mirrors
+/// `s3.allow-anonymous`; unlike that flag, this is required explicitly here
+/// rather than inferred from "no credentials configured", since the absence
+/// of an adls.* credential can also mean "rely on ambient Azure credentials
+/// (managed identity, Azure CLI, environment variables)", which this must not
+/// break.
+pub const ADLS_ALLOW_ANONYMOUS: &str = "adls.allow-anonymous";
 
 /// Azure Data Lake Storage configuration.
 ///

crates/iceberg/src/io/storage/opendal/azdls.rs

@@ -84,11 +84,12 @@ pub(crate) fn azdls_create_operator<'a>(
     absolute_path: &'a str,
     config: &AzdlsConfig,
     configured_scheme: &AzureStorageScheme,
+    allow_anonymous: bool,
 ) -> Result<(opendal::Operator, &'a str)> {
     let path = absolute_path.parse::<AzureStoragePath>()?;
     match_path_with_config(&path, config, configured_scheme)?;
 
-    let op = azdls_config_build(config, &path)?;
+    let op = azdls_config_build(config, &path, allow_anonymous)?;
 
     // Paths to files in ADLS tend to be written in fully qualified form,
     // including their filesystem and account name.
@@ -201,7 +202,41 @@ fn match_path_with_config(
     Ok(())
 }
 
-fn azdls_config_build(config: &AzdlsConfig, path: &AzureStoragePath) -> Result<opendal::Operator> {
+fn azdls_config_build(
+    config: &AzdlsConfig,
+    path: &AzureStoragePath,
+    allow_anonymous: bool,
+) -> Result<opendal::Operator> {
+    // opendal's Azdls service (via reqsign-azure-storage) has no anonymous/
+    // unsigned mode: every request goes through a generic credential-provider
+    // chain whose `Credential` variants (SharedKey/SasToken/BearerToken) all
+    // require a non-empty secret to be considered valid, and the signer hard
+    // errors ("failed to load signing credential") if none can be found or
+    // validated -- unlike e.g. its S3 service, which exposes an explicit
+    // `skip_signature` bypass. So when no credential was configured at all,
+    // we can't get a genuinely anonymous request out of the Azdls service no
+    // matter what we pass it (an empty SAS token doesn't help either: it's
+    // treated as a *found but invalid* credential, which fails the same way).
+    //
+    // Public/anonymous-read Azure containers are plain, unauthenticated HTTPS
+    // (GET with a Range header for data, HEAD for stat) -- exactly what
+    // opendal's generic `Http` service already provides, with no built-in
+    // auth of its own unless a username/password/token is set. So route the
+    // no-credentials case there instead, scoped to the same account/container
+    // endpoint the Azdls builder would have used.
+    //
+    // This is opt-in via `allow_anonymous` (see `ADLS_ALLOW_ANONYMOUS`), not
+    // inferred from "no credentials configured": the absence of an explicit
+    // adls.* credential can also legitimately mean "rely on ambient Azure
+    // credentials" (managed identity, Azure CLI, environment variables, all
+    // tried by the Azdls service's own default credential chain), which this
+    // must not silently break for callers who never asked for anonymous
+    // access. Whenever `allow_anonymous` is false, this always goes through
+    // the real, signed Azdls service exactly as before.
+    if allow_anonymous {
+        return azdls_anonymous_operator_build(config, path);
+    }
+
     let mut builder = config.clone().into_builder();
 
     if config.endpoint.is_none() {
@@ -213,6 +248,27 @@ fn azdls_config_build(config: &AzdlsConfig, path: &AzureStoragePath) -> Result<o
     Ok(opendal::Operator::new(builder)?)
 }
 
+/// Builds a read-only, unauthenticated `opendal::Operator` (via opendal's
+/// generic `Http` service) against the same account/container endpoint the
+/// signed Azdls builder would have used. Only `read` and `stat` are
+/// supported, which matches Iceberg's own read-only file access pattern.
+fn azdls_anonymous_operator_build(
+    config: &AzdlsConfig,
+    path: &AzureStoragePath,
+) -> Result<opendal::Operator> {
+    let endpoint = config
+        .endpoint
+        .clone()
+        .unwrap_or_else(|| path.as_endpoint());
+    let mut http_config = opendal::services::HttpConfig::default();
+    http_config.endpoint = Some(endpoint);
+    http_config.root = Some(format!("/{}", path.filesystem));
+    // Deliberately leave username/password/token unset: opendal's Http
+    // service only adds an Authorization header when one of these is set, so
+    // leaving them unset is what makes this a genuinely unsigned request.
+    Ok(opendal::Operator::new(http_config.into_builder())?)
+}
+
 /// Represents a fully qualified path to blob/ file in Azure Storage.
 #[derive(Debug, PartialEq)]
 struct AzureStoragePath {
@@ -564,7 +620,7 @@ mod tests {
         ];
 
         for (name, input, expected) in test_cases {
-            let result = azdls_create_operator(input.0, &input.1, &input.2);
+            let result = azdls_create_operator(input.0, &input.1, &input.2, false);
             match expected {
                 Some((expected_filesystem, expected_path)) => {
                     assert!(result.is_ok(), "Test case {name} failed: {result:?}");
@@ -580,6 +636,45 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_azdls_create_operator_anonymous() {
+        let path = "abfss://myfs@myaccount.dfs.core.windows.net/path/to/file.parquet";
+        let config = AzdlsConfig::default();
+
+        // No credentials, no `allow_anonymous`: goes through the real (signed)
+        // Azdls service as before -- unaffected by this fix. Building the
+        // operator itself still succeeds (loading a credential is deferred
+        // until an actual request is signed), so this only demonstrates it's
+        // *not* the anonymous HTTP fallback, via the scheme name.
+        let (op, relative_path) =
+            azdls_create_operator(path, &config, &AzureStorageScheme::Abfss, false)
+                .expect("operator should build without a credential");
+        assert_eq!(relative_path, "/path/to/file.parquet");
+        assert_eq!(op.info().scheme(), "azdls");
+
+        // No credentials, `allow_anonymous=true`: routes to the anonymous
+        // HTTP fallback instead.
+        let (op, relative_path) =
+            azdls_create_operator(path, &config, &AzureStorageScheme::Abfss, true)
+                .expect("anonymous operator should build");
+        assert_eq!(relative_path, "/path/to/file.parquet");
+        assert_eq!(op.info().scheme(), "http");
+
+        // `allow_anonymous=true` takes priority even when a credential is
+        // also configured -- matching S3's `skip_signature`, which is checked
+        // unconditionally before any credential is loaded. Callers only ever
+        // set this alongside a real credential by mistake; since it's an
+        // explicit opt-in flag (never inferred), that mistake is on them.
+        let config_with_key = AzdlsConfig {
+            account_key: Some("secret".to_string()),
+            ..Default::default()
+        };
+        let (op, _) =
+            azdls_create_operator(path, &config_with_key, &AzureStorageScheme::Abfss, true)
+                .expect("anonymous operator should still build");
+        assert_eq!(op.info().scheme(), "http");
+    }
+
     #[test]
     fn test_azure_storage_path_parse() {
         let test_cases = vec![

crates/iceberg/src/io/storage/opendal/mod.rs

@@ -44,6 +44,8 @@ use url::Url;
 
 use crate::catalog::{NamespaceIdent, TableIdent};
 use crate::io::refreshable_storage::RefreshableOpenDalStorageBuilder;
+#[cfg(feature = "storage-azdls")]
+use crate::io::storage::config::ADLS_ALLOW_ANONYMOUS;
 use crate::io::storage::config::{PROP_METADATA_LOCATION, PROP_TABLE_IDENT};
 use crate::io::{
     FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig,
@@ -113,6 +115,13 @@ pub enum OpenDalStorageFactory {
     },
 }
 
+#[cfg(feature = "storage-azdls")]
+fn azdls_allow_anonymous(properties: &HashMap<String, String>) -> bool {
+    properties
+        .get(ADLS_ALLOW_ANONYMOUS)
+        .is_some_and(|v| crate::io::is_truthy(v.to_lowercase().as_str()))
+}
+
 #[typetag::serde(name = "OpenDalStorageFactory")]
 impl StorageFactory for OpenDalStorageFactory {
     #[allow(unused_variables)]
@@ -145,6 +154,7 @@ impl StorageFactory for OpenDalStorageFactory {
             OpenDalStorageFactory::Azdls { configured_scheme } => {
                 Ok(Arc::new(OpenDalStorage::Azdls {
                     configured_scheme: configured_scheme.clone(),
+                    allow_anonymous: azdls_allow_anonymous(config.props()),
                     config: azdls_config_parse(config.props().clone())?.into(),
                 }))
             }
@@ -216,6 +226,9 @@ pub enum OpenDalStorage {
         configured_scheme: AzureStorageScheme,
         /// Azure DLS configuration.
         config: Arc<AzdlsConfig>,
+        /// Whether to fall back to genuinely anonymous, unsigned access when
+        /// no credential is configured. See [`ADLS_ALLOW_ANONYMOUS`].
+        allow_anonymous: bool,
     },
     /// Wraps any storage with credential refresh capability.
     Refreshable {
@@ -259,9 +272,11 @@ impl OpenDalStorage {
             #[cfg(feature = "storage-azdls")]
             "abfss" | "abfs" | "wasbs" | "wasb" => {
                 let configured_scheme = scheme_str.parse::<AzureStorageScheme>()?;
+                let allow_anonymous = azdls_allow_anonymous(&props);
                 Ok(Self::Azdls {
                     config: azdls_config_parse(props)?.into(),
                     configured_scheme,
+                    allow_anonymous,
                 })
             }
             _ => Err(Error::new(
@@ -357,7 +372,8 @@ impl OpenDalStorage {
             OpenDalStorage::Azdls {
                 configured_scheme,
                 config,
-            } => azdls_create_operator(path, config, configured_scheme)?,
+                allow_anonymous,
+            } => azdls_create_operator(path, config, configured_scheme, *allow_anonymous)?,
             OpenDalStorage::Refreshable { backend } => {
                 let backend = backend.as_ref().ok_or_else(|| {
                     Error::new(