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_match: true,
411            write_with_if_not_exists: true,
412            write_with_if_none_match: true,
413            write_with_user_metadata: true,
414
415            delete: true,
416            delete_max_size: Some(AZBLOB_BATCH_LIMIT),
417
418            copy: true,
419            copy_with_if_not_exists: true,
420            copy_can_multi: true,
421            copy_multi_min_size: Some(AZBLOB_COPY_MIN_BLOCK_SIZE),
422            copy_multi_max_size: Some(AZBLOB_COPY_MAX_BLOCK_SIZE),
423
424            list: true,
425            list_with_recursive: true,
426
427            presign: self.config.sas_token.is_some(),
428            presign_stat: self.config.sas_token.is_some(),
429            presign_read: self.config.sas_token.is_some(),
430            presign_write: self.config.sas_token.is_some(),
431
432            shared: true,
433
434            ..Default::default()
435        };
436
437        Ok(AzblobBackend {
438            core: Arc::new(AzblobCore {
439                info,
440                capability,
441                root,
442                endpoint,
443                encryption_key,
444                encryption_key_sha256,
445                encryption_algorithm,
446                container: self.config.container.clone(),
447                skip_signature: self.config.skip_signature,
448                signer,
449            }),
450        })
451    }
452}
453
454/// Backend for azblob services.
455#[derive(Debug, Clone)]
456pub struct AzblobBackend {
457    pub(crate) core: Arc<AzblobCore>,
458}
459
460impl Service for AzblobBackend {
461    type Reader = oio::StreamReader<AzblobReader>;
462    type Writer = AzblobWriters;
463    type Lister = oio::PageLister<AzblobLister>;
464    type Deleter = oio::BatchDeleter<AzblobDeleter>;
465    type Copier = AzblobCopiers;
466
467    fn info(&self) -> ServiceInfo {
468        self.core.info.clone()
469    }
470
471    fn capability(&self) -> Capability {
472        self.core.capability
473    }
474
475    async fn create_dir(
476        &self,
477        _ctx: &OperationContext,
478        _path: &str,
479        _args: OpCreateDir,
480    ) -> Result<RpCreateDir> {
481        Err(Error::new(
482            ErrorKind::Unsupported,
483            "operation is not supported",
484        ))
485    }
486
487    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
488        let resp = self
489            .core
490            .azblob_get_blob_properties(ctx, path, &args)
491            .await?;
492
493        let status = resp.status();
494
495        match status {
496            StatusCode::OK => {
497                let headers = resp.headers();
498                let mut meta = parse_into_metadata(path, headers)?;
499                if let Some(version_id) = parse_header_to_str(headers, X_MS_VERSION_ID)? {
500                    meta.set_version(version_id);
501                }
502
503                let user_meta = parse_prefixed_headers(headers, X_MS_META_PREFIX);
504                if !user_meta.is_empty() {
505                    meta = meta.with_user_metadata(user_meta);
506                }
507
508                Ok(RpStat::new(meta))
509            }
510            _ => Err(parse_error(resp)),
511        }
512    }
513    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
514        let output: oio::StreamReader<AzblobReader> = {
515            Ok(oio::StreamReader::new(AzblobReader::new(
516                self.clone(),
517                ctx.clone(),
518                path,
519                args,
520            )))
521        }?;
522
523        Ok(output)
524    }
525
526    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
527        let output: AzblobWriters = {
528            let w = AzblobWriter::new(
529                self.core.clone(),
530                ctx.clone(),
531                args.clone(),
532                path.to_string(),
533            );
534            let w = if args.append() {
535                AzblobWriters::Two(oio::AppendWriter::new(w))
536            } else {
537                AzblobWriters::One(oio::BlockWriter::new(
538                    ctx.executor().clone(),
539                    w,
540                    args.concurrent(),
541                ))
542            };
543
544            Ok(w)
545        }?;
546
547        Ok(output)
548    }
549
550    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
551        let output: oio::BatchDeleter<AzblobDeleter> = {
552            Ok(oio::BatchDeleter::new(
553                AzblobDeleter::new(self.core.clone(), ctx.clone()),
554                self.core.capability.delete_max_size,
555            ))
556        }?;
557
558        Ok(output)
559    }
560
561    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
562        let output: oio::PageLister<AzblobLister> = {
563            let l = AzblobLister::new(
564                self.core.clone(),
565                ctx.clone(),
566                path.to_string(),
567                args.recursive(),
568                args.limit(),
569            );
570
571            Ok(oio::PageLister::new(l))
572        }?;
573
574        Ok(output)
575    }
576
577    fn copy(
578        &self,
579        ctx: &OperationContext,
580        from: &str,
581        to: &str,
582        args: OpCopy,
583        opts: OpCopier,
584    ) -> Result<Self::Copier> {
585        let output: AzblobCopiers = {
586            let copier = new_azblob_copier(self.core.clone(), ctx, from, to, args, opts)?;
587            Ok(copier)
588        }?;
589
590        Ok(output)
591    }
592
593    async fn rename(
594        &self,
595        _ctx: &OperationContext,
596        _from: &str,
597        _to: &str,
598        _args: OpRename,
599    ) -> Result<RpRename> {
600        Err(Error::new(
601            ErrorKind::Unsupported,
602            "operation is not supported",
603        ))
604    }
605
606    async fn presign(
607        &self,
608        ctx: &OperationContext,
609        path: &str,
610        args: OpPresign,
611    ) -> Result<RpPresign> {
612        let req = match args.operation() {
613            PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v),
614            PresignOperation::Read(range, v) => self.core.azblob_get_blob_request(path, *range, v),
615            PresignOperation::Write(_) => {
616                self.core
617                    .azblob_put_blob_request(path, None, &OpWrite::default(), Buffer::new())
618            }
619            PresignOperation::Delete(_) => Err(Error::new(
620                ErrorKind::Unsupported,
621                "operation is not supported",
622            )),
623            _ => Err(Error::new(
624                ErrorKind::Unsupported,
625                "presign operation is not supported",
626            )),
627        };
628
629        let req = req?;
630        let req = self.core.sign_query(ctx, req).await?;
631
632        let (parts, _) = req.into_parts();
633
634        Ok(RpPresign::new(PresignedRequest::new(
635            parts.method,
636            parts.uri,
637            parts.headers,
638        )))
639    }
640}
opendal_service_azblob/backend.rs

opendal_service_azblob/
backend.rs
