1use std::fmt::Debug;
19use std::fmt::Formatter;
20use std::sync::Arc;
21
22use base64::prelude::BASE64_STANDARD;
23use base64::Engine;
24use http::Response;
25use http::StatusCode;
26use log::debug;
27use reqsign::AzureStorageConfig;
28use reqsign::AzureStorageLoader;
29use reqsign::AzureStorageSigner;
30use sha2::Digest;
31use sha2::Sha256;
32
33use super::core::constants::X_MS_META_PREFIX;
34use super::core::constants::X_MS_VERSION_ID;
35use super::core::AzblobCore;
36use super::delete::AzblobDeleter;
37use super::error::parse_error;
38use super::lister::AzblobLister;
39use super::writer::AzblobWriter;
40use super::writer::AzblobWriters;
41use super::DEFAULT_SCHEME;
42use crate::raw::*;
43use crate::services::AzblobConfig;
44use crate::*;
45const AZBLOB_BATCH_LIMIT: usize = 256;
46
47impl From<AzureStorageConfig> for AzblobConfig {
48 fn from(value: AzureStorageConfig) -> Self {
49 Self {
50 endpoint: value.endpoint,
51 account_name: value.account_name,
52 account_key: value.account_key,
53 sas_token: value.sas_token,
54 ..Default::default()
55 }
56 }
57}
58
59impl Configurator for AzblobConfig {
60 type Builder = AzblobBuilder;
61
62 #[allow(deprecated)]
63 fn into_builder(self) -> Self::Builder {
64 AzblobBuilder {
65 config: self,
66
67 http_client: None,
68 }
69 }
70}
71
72#[doc = include_str!("docs.md")]
73#[derive(Default, Clone)]
74pub struct AzblobBuilder {
75 config: AzblobConfig,
76
77 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
78 http_client: Option<HttpClient>,
79}
80
81impl Debug for AzblobBuilder {
82 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
83 let mut ds = f.debug_struct("AzblobBuilder");
84
85 ds.field("config", &self.config);
86
87 ds.finish()
88 }
89}
90
91impl AzblobBuilder {
92 pub fn root(mut self, root: &str) -> Self {
96 self.config.root = if root.is_empty() {
97 None
98 } else {
99 Some(root.to_string())
100 };
101
102 self
103 }
104
105 pub fn container(mut self, container: &str) -> Self {
107 self.config.container = container.to_string();
108
109 self
110 }
111
112 pub fn endpoint(mut self, endpoint: &str) -> Self {
119 if !endpoint.is_empty() {
120 self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
122 }
123
124 self
125 }
126
127 pub fn account_name(mut self, account_name: &str) -> Self {
132 if !account_name.is_empty() {
133 self.config.account_name = Some(account_name.to_string());
134 }
135
136 self
137 }
138
139 pub fn account_key(mut self, account_key: &str) -> Self {
144 if !account_key.is_empty() {
145 self.config.account_key = Some(account_key.to_string());
146 }
147
148 self
149 }
150
151 pub fn encryption_key(mut self, v: &str) -> Self {
164 if !v.is_empty() {
165 self.config.encryption_key = Some(v.to_string());
166 }
167
168 self
169 }
170
171 pub fn encryption_key_sha256(mut self, v: &str) -> Self {
184 if !v.is_empty() {
185 self.config.encryption_key_sha256 = Some(v.to_string());
186 }
187
188 self
189 }
190
191 pub fn encryption_algorithm(mut self, v: &str) -> Self {
204 if !v.is_empty() {
205 self.config.encryption_algorithm = Some(v.to_string());
206 }
207
208 self
209 }
210
211 pub fn server_side_encryption_with_customer_key(mut self, key: &[u8]) -> Self {
225 self.config.encryption_algorithm = Some("AES256".to_string());
227 self.config.encryption_key = Some(BASE64_STANDARD.encode(key));
228 self.config.encryption_key_sha256 =
229 Some(BASE64_STANDARD.encode(Sha256::digest(key).as_slice()));
230 self
231 }
232
233 pub fn sas_token(mut self, sas_token: &str) -> Self {
241 if !sas_token.is_empty() {
242 self.config.sas_token = Some(sas_token.to_string());
243 }
244
245 self
246 }
247
248 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
255 #[allow(deprecated)]
256 pub fn http_client(mut self, client: HttpClient) -> Self {
257 self.http_client = Some(client);
258 self
259 }
260
261 pub fn batch_max_operations(mut self, batch_max_operations: usize) -> Self {
263 self.config.batch_max_operations = Some(batch_max_operations);
264
265 self
266 }
267
268 pub fn from_connection_string(conn: &str) -> Result<Self> {
296 let config =
297 raw::azure_config_from_connection_string(conn, raw::AzureStorageService::Blob)?;
298
299 Ok(AzblobConfig::from(config).into_builder())
300 }
301}
302
303impl Builder for AzblobBuilder {
304 type Config = AzblobConfig;
305
306 fn build(self) -> Result<impl Access> {
307 debug!("backend build started: {:?}", &self);
308
309 let root = normalize_root(&self.config.root.unwrap_or_default());
310 debug!("backend use root {root}");
311
312 let container = match self.config.container.is_empty() {
314 false => Ok(&self.config.container),
315 true => Err(Error::new(ErrorKind::ConfigInvalid, "container is empty")
316 .with_operation("Builder::build")
317 .with_context("service", Scheme::Azblob)),
318 }?;
319 debug!("backend use container {}", &container);
320
321 let endpoint = match &self.config.endpoint {
322 Some(endpoint) => Ok(endpoint.clone()),
323 None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
324 .with_operation("Builder::build")
325 .with_context("service", Scheme::Azblob)),
326 }?;
327 debug!("backend use endpoint {}", &container);
328
329 #[cfg(target_arch = "wasm32")]
330 let mut config_loader = AzureStorageConfig::default();
331 #[cfg(not(target_arch = "wasm32"))]
332 let mut config_loader = AzureStorageConfig::default().from_env();
333
334 if let Some(v) = self
335 .config
336 .account_name
337 .clone()
338 .or_else(|| raw::azure_account_name_from_endpoint(endpoint.as_str()))
339 {
340 config_loader.account_name = Some(v);
341 }
342
343 if let Some(v) = self.config.account_key.clone() {
344 if let Err(e) = BASE64_STANDARD.decode(&v) {
346 return Err(Error::new(
347 ErrorKind::ConfigInvalid,
348 format!("invalid account_key: cannot decode as base64: {e}"),
349 )
350 .with_operation("Builder::build")
351 .with_context("service", Scheme::Azblob)
352 .with_context("key", "account_key"));
353 }
354 config_loader.account_key = Some(v);
355 }
356
357 if let Some(v) = self.config.sas_token.clone() {
358 config_loader.sas_token = Some(v);
359 }
360
361 let encryption_key =
362 match &self.config.encryption_key {
363 None => None,
364 Some(v) => Some(build_header_value(v).map_err(|err| {
365 err.with_context("key", "server_side_encryption_customer_key")
366 })?),
367 };
368
369 let encryption_key_sha256 = match &self.config.encryption_key_sha256 {
370 None => None,
371 Some(v) => Some(build_header_value(v).map_err(|err| {
372 err.with_context("key", "server_side_encryption_customer_key_sha256")
373 })?),
374 };
375
376 let encryption_algorithm = match &self.config.encryption_algorithm {
377 None => None,
378 Some(v) => {
379 if v == "AES256" {
380 Some(build_header_value(v).map_err(|err| {
381 err.with_context("key", "server_side_encryption_customer_algorithm")
382 })?)
383 } else {
384 return Err(Error::new(
385 ErrorKind::ConfigInvalid,
386 "encryption_algorithm value must be AES256",
387 ));
388 }
389 }
390 };
391
392 let cred_loader = AzureStorageLoader::new(config_loader);
393
394 let signer = AzureStorageSigner::new();
395
396 Ok(AzblobBackend {
397 core: Arc::new(AzblobCore {
398 info: {
399 let am = AccessorInfo::default();
400 am.set_scheme(DEFAULT_SCHEME)
401 .set_root(&root)
402 .set_name(container)
403 .set_native_capability(Capability {
404 stat: true,
405 stat_with_if_match: true,
406 stat_with_if_none_match: true,
407
408 read: true,
409
410 read_with_if_match: true,
411 read_with_if_none_match: true,
412 read_with_override_content_disposition: true,
413 read_with_if_modified_since: true,
414 read_with_if_unmodified_since: true,
415
416 write: true,
417 write_can_append: true,
418 write_can_empty: true,
419 write_can_multi: true,
420 write_with_cache_control: true,
421 write_with_content_type: true,
422 write_with_if_not_exists: true,
423 write_with_if_none_match: true,
424 write_with_user_metadata: true,
425
426 delete: true,
427 delete_max_size: Some(AZBLOB_BATCH_LIMIT),
428
429 copy: true,
430 copy_with_if_not_exists: true,
431
432 list: true,
433 list_with_recursive: true,
434
435 presign: self.config.sas_token.is_some(),
436 presign_stat: self.config.sas_token.is_some(),
437 presign_read: self.config.sas_token.is_some(),
438 presign_write: self.config.sas_token.is_some(),
439
440 shared: true,
441
442 ..Default::default()
443 });
444
445 #[allow(deprecated)]
447 if let Some(client) = self.http_client {
448 am.update_http_client(|_| client);
449 }
450
451 am.into()
452 },
453 root,
454 endpoint,
455 encryption_key,
456 encryption_key_sha256,
457 encryption_algorithm,
458 container: self.config.container.clone(),
459
460 loader: cred_loader,
461 signer,
462 }),
463 })
464 }
465}
466
467#[derive(Debug, Clone)]
469pub struct AzblobBackend {
470 core: Arc<AzblobCore>,
471}
472
473impl Access for AzblobBackend {
474 type Reader = HttpBody;
475 type Writer = AzblobWriters;
476 type Lister = oio::PageLister<AzblobLister>;
477 type Deleter = oio::BatchDeleter<AzblobDeleter>;
478
479 fn info(&self) -> Arc<AccessorInfo> {
480 self.core.info.clone()
481 }
482
483 async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
484 let resp = self.core.azblob_get_blob_properties(path, &args).await?;
485
486 let status = resp.status();
487
488 match status {
489 StatusCode::OK => {
490 let headers = resp.headers();
491 let mut meta = parse_into_metadata(path, headers)?;
492 if let Some(version_id) = parse_header_to_str(headers, X_MS_VERSION_ID)? {
493 meta.set_version(version_id);
494 }
495
496 let user_meta = parse_prefixed_headers(headers, X_MS_META_PREFIX);
497 if !user_meta.is_empty() {
498 meta = meta.with_user_metadata(user_meta);
499 }
500
501 Ok(RpStat::new(meta))
502 }
503 _ => Err(parse_error(resp)),
504 }
505 }
506
507 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
508 let resp = self.core.azblob_get_blob(path, args.range(), &args).await?;
509
510 let status = resp.status();
511 match status {
512 StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
513 _ => {
514 let (part, mut body) = resp.into_parts();
515 let buf = body.to_buffer().await?;
516 Err(parse_error(Response::from_parts(part, buf)))
517 }
518 }
519 }
520
521 async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
522 let w = AzblobWriter::new(self.core.clone(), args.clone(), path.to_string());
523 let w = if args.append() {
524 AzblobWriters::Two(oio::AppendWriter::new(w))
525 } else {
526 AzblobWriters::One(oio::BlockWriter::new(
527 self.core.info.clone(),
528 w,
529 args.concurrent(),
530 ))
531 };
532
533 Ok((RpWrite::default(), w))
534 }
535
536 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
537 Ok((
538 RpDelete::default(),
539 oio::BatchDeleter::new(AzblobDeleter::new(self.core.clone())),
540 ))
541 }
542
543 async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
544 let l = AzblobLister::new(
545 self.core.clone(),
546 path.to_string(),
547 args.recursive(),
548 args.limit(),
549 );
550
551 Ok((RpList::default(), oio::PageLister::new(l)))
552 }
553
554 async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
555 let resp = self.core.azblob_copy_blob(from, to, args).await?;
556
557 let status = resp.status();
558
559 match status {
560 StatusCode::ACCEPTED => Ok(RpCopy::default()),
561 _ => Err(parse_error(resp)),
562 }
563 }
564
565 async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
566 let req = match args.operation() {
567 PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v),
568 PresignOperation::Read(v) => {
569 self.core
570 .azblob_get_blob_request(path, BytesRange::default(), v)
571 }
572 PresignOperation::Write(_) => {
573 self.core
574 .azblob_put_blob_request(path, None, &OpWrite::default(), Buffer::new())
575 }
576 PresignOperation::Delete(_) => Err(Error::new(
577 ErrorKind::Unsupported,
578 "operation is not supported",
579 )),
580 };
581
582 let mut req = req?;
583
584 self.core.sign_query(&mut req).await?;
585
586 let (parts, _) = req.into_parts();
587
588 Ok(RpPresign::new(PresignedRequest::new(
589 parts.method,
590 parts.uri,
591 parts.headers,
592 )))
593 }
594}