loonfs_objectstore/
abs.rs

1//! Azure Blob Storage provider.
2
3use super::{ByteRange, ObjectBody, ObjectMetadata, ObjectStore, PutMode};
4use crate::object_store::Result;
5use crate::secret::SecretString;
6use crate::store_io_runtime::StoreIoRuntime;
7use crate::{ObjectStoreError, ProviderObjectStore, ProviderObjectStoreConfig};
8use async_trait::async_trait;
9use bytes::Bytes;
10use futures::stream::BoxStream;
11use object_store::azure::MicrosoftAzureBuilder;
12use std::sync::Arc;
13
14/// Supplies explicit credentials and key scoping for the native Azure Blob adapter.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct AzureAbsStoreConfig {
17    /// Storage account used for both request addressing and shared-key signing.
18    pub account_name: String,
19    /// Blob container that acts as the LoonFS object-store root.
20    pub container_name: String,
21    /// Shared account key; empty or whitespace-only credentials are rejected.
22    pub access_key: SecretString,
23    /// Azure-compatible service endpoint override, or `None` for the public Azure endpoint.
24    pub endpoint_url: Option<String>,
25    /// Logical prefix prepended to every key, or `None` to use the container root.
26    pub key_prefix: Option<String>,
27}
28
29/// Azure Blob Storage through its native API.
30///
31/// Authentication intentionally has one path: the caller must provide an
32/// account name and account access key explicitly. The adapter does not use SAS
33/// tokens, bearer tokens, managed identity, Azure CLI credentials, or ambient
34/// environment fallback.
35#[derive(Debug)]
36pub struct AzureAbsStore {
37    inner: ProviderObjectStore,
38    /// Keeps the HTTP IO runtime alive for the provider client's lifetime;
39    /// the connector inside the client holds only a handle onto it.
40    _io_runtime: StoreIoRuntime,
41}
42
43impl AzureAbsStore {
44    /// Builds a native Azure Blob adapter with its own bounded HTTP runtime.
45    ///
46    /// Construction fails for blank account, container, key, or endpoint
47    /// values, an invalid key prefix, runtime initialization, or provider-client configuration.
48    pub fn new(config: AzureAbsStoreConfig) -> Result<Self> {
49        if config.account_name.trim().is_empty() {
50            return Err(ObjectStoreError::Configuration(
51                "account name must not be empty".to_owned(),
52            ));
53        }
54        if config.container_name.trim().is_empty() {
55            return Err(ObjectStoreError::Configuration(
56                "container name must not be empty".to_owned(),
57            ));
58        }
59        if config.access_key.expose().trim().is_empty() {
60            return Err(ObjectStoreError::Configuration(
61                "access key must not be empty".to_owned(),
62            ));
63        }
64
65        let io_runtime = StoreIoRuntime::new()?;
66        let mut builder = MicrosoftAzureBuilder::new()
67            .with_http_connector(io_runtime.connector())
68            .with_client_options(crate::provider_object_store::provider_client_options())
69            .with_retry(crate::provider_object_store::provider_retry_config())
70            .with_account(config.account_name)
71            .with_container_name(config.container_name)
72            .with_access_key(config.access_key.expose());
73        if let Some(endpoint_url) = config.endpoint_url {
74            let endpoint_url = endpoint_url.trim();
75            if endpoint_url.is_empty() {
76                return Err(ObjectStoreError::Configuration(
77                    "endpoint url must not be empty".to_owned(),
78                ));
79            }
80            let endpoint_url = normalize_http_endpoint_scheme(endpoint_url);
81            if endpoint_url.starts_with("http://") {
82                builder = builder.with_allow_http(true);
83            }
84            builder = builder.with_endpoint(endpoint_url);
85        }
86
87        let provider = Arc::new(
88            builder
89                .build()
90                .map_err(|err| ObjectStoreError::Configuration(err.to_string()))?,
91        );
92        let inner = ProviderObjectStore::new(
93            Arc::clone(&provider) as Arc<dyn object_store::ObjectStore>,
94            Some(provider),
95            ProviderObjectStoreConfig {
96                key_prefix: config.key_prefix,
97            },
98        )?;
99        Ok(Self {
100            inner,
101            _io_runtime: io_runtime,
102        })
103    }
104}
105
106fn normalize_http_endpoint_scheme(endpoint_url: &str) -> String {
107    match endpoint_url.split_once("://") {
108        Some((scheme, rest)) if scheme.eq_ignore_ascii_case("http") => format!("http://{rest}"),
109        Some((scheme, rest)) if scheme.eq_ignore_ascii_case("https") => format!("https://{rest}"),
110        _ => endpoint_url.to_owned(),
111    }
112}
113
114#[async_trait]
115impl ObjectStore for AzureAbsStore {
116    async fn head(&self, key: &str) -> Result<Option<ObjectMetadata>> {
117        self.inner.head(key).await
118    }
119
120    async fn get_with_metadata(&self, key: &str) -> Result<Option<ObjectBody>> {
121        self.inner.get_with_metadata(key).await
122    }
123
124    async fn get(&self, key: &str, range: Option<ByteRange>) -> Result<Option<Bytes>> {
125        self.inner.get(key, range).await
126    }
127
128    async fn put(&self, key: &str, bytes: Bytes, mode: PutMode) -> Result<ObjectMetadata> {
129        self.inner.put(key, bytes, mode).await
130    }
131
132    async fn delete(&self, key: &str) -> Result<()> {
133        self.inner.delete(key).await
134    }
135
136    fn list_prefix_stream(&self, prefix: &str) -> BoxStream<'static, Result<String>> {
137        self.inner.list_prefix_stream(prefix)
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::{AzureAbsStore, AzureAbsStoreConfig};
144    use crate::ObjectStore;
145    use crate::ObjectStoreError;
146
147    const AZURITE_ACCOUNT_KEY: &str =
148        "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
149
150    #[test]
151    fn access_key_is_required() {
152        let error = AzureAbsStore::new(AzureAbsStoreConfig {
153            account_name: "account".to_owned(),
154            container_name: "container".to_owned(),
155            access_key: " ".into(),
156            endpoint_url: None,
157            key_prefix: None,
158        })
159        .expect_err("blank access key should be rejected");
160
161        assert!(
162            matches!(error, ObjectStoreError::Configuration(message) if message.contains("access key"))
163        );
164    }
165
166    #[test]
167    fn http_endpoint_is_allowed_for_emulator() {
168        AzureAbsStore::new(AzureAbsStoreConfig {
169            account_name: "devstoreaccount1".to_owned(),
170            container_name: "container".to_owned(),
171            access_key: AZURITE_ACCOUNT_KEY.into(),
172            endpoint_url: Some("http://127.0.0.1:10000/devstoreaccount1".to_owned()),
173            key_prefix: None,
174        })
175        .expect("construct azure store with HTTP endpoint");
176    }
177
178    #[test]
179    fn http_endpoint_scheme_is_case_insensitive_for_emulator() {
180        AzureAbsStore::new(AzureAbsStoreConfig {
181            account_name: "devstoreaccount1".to_owned(),
182            container_name: "container".to_owned(),
183            access_key: AZURITE_ACCOUNT_KEY.into(),
184            endpoint_url: Some("HTTP://127.0.0.1:10000/devstoreaccount1".to_owned()),
185            key_prefix: None,
186        })
187        .expect("construct azure store with uppercase HTTP endpoint");
188    }
189
190    #[tokio::test]
191    async fn invalid_keys_are_rejected_before_compare_tokens() {
192        let store = AzureAbsStore::new(AzureAbsStoreConfig {
193            account_name: "devstoreaccount1".to_owned(),
194            container_name: "container".to_owned(),
195            access_key: AZURITE_ACCOUNT_KEY.into(),
196            endpoint_url: None,
197            key_prefix: Some("tenant-a".to_owned()),
198        })
199        .expect("construct azure store");
200
201        let error = store
202            .compare_and_swap(
203                "../escape",
204                "not-an-etag",
205                bytes::Bytes::from_static(br#"{"seq":1}"#),
206            )
207            .await
208            .expect_err("invalid key should be rejected before provider request");
209
210        assert!(matches!(
211            error,
212            ObjectStoreError::InvalidKey { object_key, .. } if object_key == "../escape"
213        ));
214    }
215}
loonfs_objectstore/abs.rs

loonfs_objectstore/
abs.rs
