opendal_service_azblob/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use base64::Engine;
22use base64::prelude::BASE64_STANDARD;
23use http::StatusCode;
24use log::debug;
25use reqsign_azure_storage::DefaultCredentialProvider;
26use reqsign_azure_storage::RequestSigner;
27use reqsign_azure_storage::StaticCredentialProvider;
28use reqsign_core::Context;
29use reqsign_core::OsEnv;
30use reqsign_core::Signer;
31use reqsign_file_read_tokio::TokioFileRead;
32use sha2::Digest;
33use sha2::Sha256;
34
35use super::AZBLOB_SCHEME;
36use super::config::AzblobConfig;
37use super::copier::AzblobCopiers;
38use super::copier::new_azblob_copier;
39use super::core::AzblobCore;
40use super::core::constants::AZBLOB_COPY_MAX_BLOCK_SIZE;
41use super::core::constants::AZBLOB_COPY_MIN_BLOCK_SIZE;
42use super::core::constants::X_MS_META_PREFIX;
43use super::core::constants::X_MS_VERSION_ID;
44use super::core::parse_error;
45use super::deleter::AzblobDeleter;
46use super::lister::AzblobLister;
47use super::reader::*;
48use super::writer::AzblobWriter;
49use super::writer::AzblobWriters;
50use opendal_core::raw::*;
51use opendal_core::*;
52use opendal_service_azure_common::{
53    AzureStorageConfig as AzureConnectionConfig, AzureStorageService,
54    azure_account_name_from_endpoint, azure_config_from_connection_string,
55};
56
57const AZBLOB_BATCH_LIMIT: usize = 256;
58
59impl From<AzureConnectionConfig> for AzblobConfig {
60    fn from(value: AzureConnectionConfig) -> Self {
61        Self {
62            endpoint: value.endpoint,
63            account_name: value.account_name,
64            account_key: value.account_key,
65            sas_token: value.sas_token,
66            ..Default::default()
67        }
68    }
69}
70
71#[doc = include_str!("docs.md")]
72#[derive(Default)]
73pub struct AzblobBuilder {
74    pub(super) config: AzblobConfig,
75}
76
77impl Debug for AzblobBuilder {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct("AzblobBuilder")
80            .field("config", &self.config)
81            .finish_non_exhaustive()
82    }
83}
84
85impl AzblobBuilder {
86    /// Set root of this backend.
87    ///
88    /// All operations will happen under this root.
89    pub fn root(mut self, root: &str) -> Self {
90        self.config.root = if root.is_empty() {
91            None
92        } else {
93            Some(root.to_string())
94        };
95
96        self
97    }
98
99    /// Set container name of this backend.
100    pub fn container(mut self, container: &str) -> Self {
101        self.config.container = container.to_string();
102
103        self
104    }
105
106    /// Set endpoint of this backend
107    ///
108    /// Endpoint must be full uri, e.g.
109    ///
110    /// - Azblob: `https://accountname.blob.core.windows.net`
111    /// - Azurite: `http://127.0.0.1:10000/devstoreaccount1`
112    pub fn endpoint(mut self, endpoint: &str) -> Self {
113        if !endpoint.is_empty() {
114            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
115            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
116        }
117
118        self
119    }
120
121    /// Set account_name of this backend.
122    ///
123    /// - If account_name is set, we will take user's input first.
124    /// - If not, we will try to load it from environment.
125    pub fn account_name(mut self, account_name: &str) -> Self {
126        if !account_name.is_empty() {
127            self.config.account_name = Some(account_name.to_string());
128        }
129
130        self
131    }
132
133    /// Set account_key of this backend.
134    ///
135    /// - If account_key is set, we will take user's input first.
136    /// - If not, we will try to load it from environment.
137    pub fn account_key(mut self, account_key: &str) -> Self {
138        if !account_key.is_empty() {
139            self.config.account_key = Some(account_key.to_string());
140        }
141
142        self
143    }
144
145    /// Set encryption_key of this backend.
146    ///
147    /// # Args
148    ///
149    /// `v`: Base64-encoded key that matches algorithm specified in `encryption_algorithm`.
150    ///
151    /// # Note
152    ///
153    /// This function is the low-level setting for SSE related features.
154    ///
155    /// SSE related options should be set carefully to make them works.
156    /// Please use `server_side_encryption_with_*` helpers if even possible.
157    pub fn encryption_key(mut self, v: &str) -> Self {
158        if !v.is_empty() {
159            self.config.encryption_key = Some(v.to_string());
160        }
161
162        self
163    }
164
165    /// Set encryption_key_sha256 of this backend.
166    ///
167    /// # Args
168    ///
169    /// `v`: Base64-encoded SHA256 digest of the key specified in encryption_key.
170    ///
171    /// # Note
172    ///
173    /// This function is the low-level setting for SSE related features.
174    ///
175    /// SSE related options should be set carefully to make them works.
176    /// Please use `server_side_encryption_with_*` helpers if even possible.
177    pub fn encryption_key_sha256(mut self, v: &str) -> Self {
178        if !v.is_empty() {
179            self.config.encryption_key_sha256 = Some(v.to_string());
180        }
181
182        self
183    }
184
185    /// Set encryption_algorithm of this backend.
186    ///
187    /// # Args
188    ///
189    /// `v`: server-side encryption algorithm. (Available values: `AES256`)
190    ///
191    /// # Note
192    ///
193    /// This function is the low-level setting for SSE related features.
194    ///
195    /// SSE related options should be set carefully to make them works.
196    /// Please use `server_side_encryption_with_*` helpers if even possible.
197    pub fn encryption_algorithm(mut self, v: &str) -> Self {
198        if !v.is_empty() {
199            self.config.encryption_algorithm = Some(v.to_string());
200        }
201
202        self
203    }
204
205    /// Enable server side encryption with customer key.
206    ///
207    /// As known as: CPK
208    ///
209    /// # Args
210    ///
211    /// `key`: Base64-encoded SHA256 digest of the key specified in encryption_key.
212    ///
213    /// # Note
214    ///
215    /// Function that helps the user to set the server-side customer-provided encryption key, the key's SHA256, and the algorithm.
216    /// See [Server-side encryption with customer-provided keys (CPK)](https://learn.microsoft.com/en-us/azure/storage/blobs/encryption-customer-provided-keys)
217    /// for more info.
218    pub fn server_side_encryption_with_customer_key(mut self, key: &[u8]) -> Self {
219        // Only AES256 is supported for now
220        self.config.encryption_algorithm = Some("AES256".to_string());
221        self.config.encryption_key = Some(BASE64_STANDARD.encode(key));
222        let key_sha256 = Sha256::digest(key);
223        self.config.encryption_key_sha256 = Some(BASE64_STANDARD.encode(key_sha256));
224        self
225    }
226
227    /// Set sas_token of this backend.
228    ///
229    /// - If sas_token is set, we will take user's input first.
230    /// - If not, we will try to load it from environment.
231    ///
232    /// See [Grant limited access to Azure Storage resources using shared access signatures (SAS)](https://learn.microsoft.com/en-us/azure/storage/common/storage-sas-overview)
233    /// for more info.
234    pub fn sas_token(mut self, sas_token: &str) -> Self {
235        if !sas_token.is_empty() {
236            self.config.sas_token = Some(sas_token.to_string());
237        }
238
239        self
240    }
241
242    /// Deprecated: Azblob delete batch capability is enabled by default with Azure Blob's 256-operation batch limit.
243    #[deprecated(
244        since = "0.57.0",
245        note = "Azblob delete batch capability is enabled by default with Azure Blob's 256-operation batch limit and this option is no longer needed."
246    )]
247    pub fn batch_max_operations(self, _batch_max_operations: usize) -> Self {
248        self
249    }
250
251    /// Skip signature will skip loading credentials and signing requests.
252    pub fn skip_signature(mut self) -> Self {
253        self.config.skip_signature = true;
254        self
255    }
256
257    /// from_connection_string will make a builder from connection string
258    ///
259    /// connection string looks like:
260    ///
261    /// ```txt
262    /// DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;
263    /// AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;
264    /// BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;
265    /// QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;
266    /// TableEndpoint=http://127.0.0.1:10002/devstoreaccount1;
267    /// ```
268    ///
269    /// Or
270    ///
271    /// ```txt
272    /// DefaultEndpointsProtocol=https;
273    /// AccountName=storagesample;
274    /// AccountKey=<account-key>;
275    /// EndpointSuffix=core.chinacloudapi.cn;
276    /// ```
277    ///
278    /// For reference: [Configure Azure Storage connection strings](https://learn.microsoft.com/en-us/azure/storage/common/storage-configure-connection-string)
279    ///
280    /// # Note
281    ///
282    /// Connection strings can only configure the endpoint, account name and
283    /// authentication information. Users still need to configure container name.
284    pub fn from_connection_string(conn: &str) -> Result<Self> {
285        let config = azure_config_from_connection_string(conn, AzureStorageService::Blob)?;
286
287        Ok(AzblobConfig::from(config).into_builder())
288    }
289}
290
291impl Builder for AzblobBuilder {
292    type Config = AzblobConfig;
293
294    fn build(self) -> Result<impl Service> {
295        debug!("backend build started: {:?}", self);
296
297        let root = normalize_root(&self.config.root.unwrap_or_default());
298        debug!("backend use root {root}");
299
300        // Handle endpoint, region and container name.
301        let container = match self.config.container.is_empty() {
302            false => Ok(&self.config.container),
303            true => Err(Error::new(ErrorKind::ConfigInvalid, "container is empty")
304                .with_operation("Builder::build")
305                .with_context("service", AZBLOB_SCHEME)),
306        }?;
307        debug!("backend use container {}", container);
308
309        let endpoint = match &self.config.endpoint {
310            Some(endpoint) => Ok(endpoint.clone()),
311            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
312                .with_operation("Builder::build")
313                .with_context("service", AZBLOB_SCHEME)),
314        }?;
315        debug!("backend use endpoint {}", container);
316
317        let account_name = self
318            .config
319            .account_name
320            .clone()
321            .or_else(|| azure_account_name_from_endpoint(endpoint.as_str()));
322
323        if let Some(v) = &self.config.account_key {
324            // Validate that account_key can be decoded as base64
325            if let Err(e) = BASE64_STANDARD.decode(v) {
326                return Err(Error::new(
327                    ErrorKind::ConfigInvalid,
328                    format!("invalid account_key: cannot decode as base64: {e}"),
329                )
330                .with_operation("Builder::build")
331                .with_context("service", AZBLOB_SCHEME)
332                .with_context("key", "account_key"));
333            }
334        }
335
336        let encryption_key =
337            match &self.config.encryption_key {
338                None => None,
339                Some(v) => Some(build_header_value(v).map_err(|err| {
340                    err.with_context("key", "server_side_encryption_customer_key")
341                })?),
342            };
343
344        let encryption_key_sha256 = match &self.config.encryption_key_sha256 {
345            None => None,
346            Some(v) => Some(build_header_value(v).map_err(|err| {
347                err.with_context("key", "server_side_encryption_customer_key_sha256")
348            })?),
349        };
350
351        let encryption_algorithm = match &self.config.encryption_algorithm {
352            None => None,
353            Some(v) => {
354                if v == "AES256" {
355                    Some(build_header_value(v).map_err(|err| {
356                        err.with_context("key", "server_side_encryption_customer_algorithm")
357                    })?)
358                } else {
359                    return Err(Error::new(
360                        ErrorKind::ConfigInvalid,
361                        "encryption_algorithm value must be AES256",
362                    ));
363                }
364            }
365        };
366
367        let ctx = Context::new().with_file_read(TokioFileRead).with_env(OsEnv);
368
369        let mut credential = DefaultCredentialProvider::new();
370
371        if let (Some(account_name), Some(account_key)) =
372            (account_name.as_deref(), self.config.account_key.as_deref())
373        {
374            credential = credential.push_front(StaticCredentialProvider::new_shared_key(
375                account_name,
376                account_key,
377            ));
378        }
379
380        if let Some(sas_token) = self.config.sas_token.as_deref() {
381            credential = credential.push_front(StaticCredentialProvider::new_sas_token(sas_token));
382        }
383
384        let signer = Signer::new(
385            ctx,
386            credential,
387            RequestSigner::new().with_service_sas_permissions("racwd"),
388        );
389
390        let info = ServiceInfo::new(AZBLOB_SCHEME, &root, container);
391        let capability = Capability {
392            stat: true,
393            stat_with_if_match: true,
394            stat_with_if_none_match: true,
395
396            read: true,
397
398            read_with_if_match: true,
399            read_with_if_none_match: true,
400            read_with_override_content_disposition: true,
401            read_with_if_modified_since: true,
402            read_with_if_unmodified_since: true,
403
404            write: true,
405            write_can_append: true,
406            write_can_empty: true,
407            write_can_multi: true,
408            write_with_cache_control: true,
409            write_with_content_type: true,
410            write_with_if_not_exists: true,
411            write_with_if_none_match: true,
412            write_with_user_metadata: true,
413
414            delete: true,
415            delete_max_size: Some(AZBLOB_BATCH_LIMIT),
416
417            copy: true,
418            copy_with_if_not_exists: true,
419            copy_can_multi: true,
420            copy_multi_min_size: Some(AZBLOB_COPY_MIN_BLOCK_SIZE),
421            copy_multi_max_size: Some(AZBLOB_COPY_MAX_BLOCK_SIZE),
422
423            list: true,
424            list_with_recursive: true,
425
426            presign: self.config.sas_token.is_some(),
427            presign_stat: self.config.sas_token.is_some(),
428            presign_read: self.config.sas_token.is_some(),
429            presign_write: self.config.sas_token.is_some(),
430
431            shared: true,
432
433            ..Default::default()
434        };
435
436        Ok(AzblobBackend {
437            core: Arc::new(AzblobCore {
438                info,
439                capability,
440                root,
441                endpoint,
442                encryption_key,
443                encryption_key_sha256,
444                encryption_algorithm,
445                container: self.config.container.clone(),
446                skip_signature: self.config.skip_signature,
447                signer,
448            }),
449        })
450    }
451}
452
453/// Backend for azblob services.
454#[derive(Debug, Clone)]
455pub struct AzblobBackend {
456    pub(crate) core: Arc<AzblobCore>,
457}
458
459impl Service for AzblobBackend {
460    type Reader = oio::StreamReader<AzblobReader>;
461    type Writer = AzblobWriters;
462    type Lister = oio::PageLister<AzblobLister>;
463    type Deleter = oio::BatchDeleter<AzblobDeleter>;
464    type Copier = AzblobCopiers;
465
466    fn info(&self) -> ServiceInfo {
467        self.core.info.clone()
468    }
469
470    fn capability(&self) -> Capability {
471        self.core.capability
472    }
473
474    async fn create_dir(
475        &self,
476        _ctx: &OperationContext,
477        _path: &str,
478        _args: OpCreateDir,
479    ) -> Result<RpCreateDir> {
480        Err(Error::new(
481            ErrorKind::Unsupported,
482            "operation is not supported",
483        ))
484    }
485
486    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
487        let resp = self
488            .core
489            .azblob_get_blob_properties(ctx, path, &args)
490            .await?;
491
492        let status = resp.status();
493
494        match status {
495            StatusCode::OK => {
496                let headers = resp.headers();
497                let mut meta = parse_into_metadata(path, headers)?;
498                if let Some(version_id) = parse_header_to_str(headers, X_MS_VERSION_ID)? {
499                    meta.set_version(version_id);
500                }
501
502                let user_meta = parse_prefixed_headers(headers, X_MS_META_PREFIX);
503                if !user_meta.is_empty() {
504                    meta = meta.with_user_metadata(user_meta);
505                }
506
507                Ok(RpStat::new(meta))
508            }
509            _ => Err(parse_error(resp)),
510        }
511    }
512    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
513        let output: oio::StreamReader<AzblobReader> = {
514            Ok(oio::StreamReader::new(AzblobReader::new(
515                self.clone(),
516                ctx.clone(),
517                path,
518                args,
519            )))
520        }?;
521
522        Ok(output)
523    }
524
525    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
526        let output: AzblobWriters = {
527            let w = AzblobWriter::new(
528                self.core.clone(),
529                ctx.clone(),
530                args.clone(),
531                path.to_string(),
532            );
533            let w = if args.append() {
534                AzblobWriters::Two(oio::AppendWriter::new(w))
535            } else {
536                AzblobWriters::One(oio::BlockWriter::new(
537                    ctx.executor().clone(),
538                    w,
539                    args.concurrent(),
540                ))
541            };
542
543            Ok(w)
544        }?;
545
546        Ok(output)
547    }
548
549    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
550        let output: oio::BatchDeleter<AzblobDeleter> = {
551            Ok(oio::BatchDeleter::new(
552                AzblobDeleter::new(self.core.clone(), ctx.clone()),
553                self.core.capability.delete_max_size,
554            ))
555        }?;
556
557        Ok(output)
558    }
559
560    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
561        let output: oio::PageLister<AzblobLister> = {
562            let l = AzblobLister::new(
563                self.core.clone(),
564                ctx.clone(),
565                path.to_string(),
566                args.recursive(),
567                args.limit(),
568            );
569
570            Ok(oio::PageLister::new(l))
571        }?;
572
573        Ok(output)
574    }
575
576    fn copy(
577        &self,
578        ctx: &OperationContext,
579        from: &str,
580        to: &str,
581        args: OpCopy,
582        opts: OpCopier,
583    ) -> Result<Self::Copier> {
584        let output: AzblobCopiers = {
585            let copier = new_azblob_copier(self.core.clone(), ctx, from, to, args, opts)?;
586            Ok(copier)
587        }?;
588
589        Ok(output)
590    }
591
592    async fn rename(
593        &self,
594        _ctx: &OperationContext,
595        _from: &str,
596        _to: &str,
597        _args: OpRename,
598    ) -> Result<RpRename> {
599        Err(Error::new(
600            ErrorKind::Unsupported,
601            "operation is not supported",
602        ))
603    }
604
605    async fn presign(
606        &self,
607        ctx: &OperationContext,
608        path: &str,
609        args: OpPresign,
610    ) -> Result<RpPresign> {
611        let req = match args.operation() {
612            PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v),
613            PresignOperation::Read(range, v) => self.core.azblob_get_blob_request(path, *range, v),
614            PresignOperation::Write(_) => {
615                self.core
616                    .azblob_put_blob_request(path, None, &OpWrite::default(), Buffer::new())
617            }
618            PresignOperation::Delete(_) => Err(Error::new(
619                ErrorKind::Unsupported,
620                "operation is not supported",
621            )),
622            _ => Err(Error::new(
623                ErrorKind::Unsupported,
624                "presign operation is not supported",
625            )),
626        };
627
628        let req = req?;
629        let req = self.core.sign_query(ctx, req).await?;
630
631        let (parts, _) = req.into_parts();
632
633        Ok(RpPresign::new(PresignedRequest::new(
634            parts.method,
635            parts.uri,
636            parts.headers,
637        )))
638    }
639}
opendal_service_azblob/backend.rs

opendal_service_azblob/
backend.rs
