object_store/azure/
client.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 super::credential::AzureCredential;
19use crate::azure::credential::*;
20use crate::azure::{AzureCredentialProvider, STORE};
21use crate::client::get::GetClient;
22use crate::client::header::{get_put_result, HeaderConfig};
23use crate::client::list::ListClient;
24use crate::client::retry::RetryExt;
25use crate::client::GetOptionsExt;
26use crate::multipart::PartId;
27use crate::path::DELIMITER;
28use crate::util::{deserialize_rfc1123, GetRange};
29use crate::{
30    Attribute, Attributes, ClientOptions, GetOptions, ListResult, ObjectMeta, Path, PutMode,
31    PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, RetryConfig, TagSet,
32};
33use async_trait::async_trait;
34use base64::prelude::BASE64_STANDARD;
35use base64::Engine;
36use bytes::{Buf, Bytes};
37use chrono::{DateTime, Utc};
38use hyper::http::HeaderName;
39use reqwest::{
40    header::{HeaderValue, CONTENT_LENGTH, IF_MATCH, IF_NONE_MATCH},
41    Client as ReqwestClient, Method, RequestBuilder, Response,
42};
43use serde::{Deserialize, Serialize};
44use snafu::{OptionExt, ResultExt, Snafu};
45use std::collections::HashMap;
46use std::sync::Arc;
47use std::time::Duration;
48use url::Url;
49
50const VERSION_HEADER: &str = "x-ms-version-id";
51const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-ms-meta-";
52static MS_CACHE_CONTROL: HeaderName = HeaderName::from_static("x-ms-blob-cache-control");
53static MS_CONTENT_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-content-type");
54static MS_CONTENT_DISPOSITION: HeaderName =
55    HeaderName::from_static("x-ms-blob-content-disposition");
56static MS_CONTENT_ENCODING: HeaderName = HeaderName::from_static("x-ms-blob-content-encoding");
57static MS_CONTENT_LANGUAGE: HeaderName = HeaderName::from_static("x-ms-blob-content-language");
58
59static TAGS_HEADER: HeaderName = HeaderName::from_static("x-ms-tags");
60
61/// A specialized `Error` for object store-related errors
62#[derive(Debug, Snafu)]
63#[allow(missing_docs)]
64pub(crate) enum Error {
65    #[snafu(display("Error performing get request {}: {}", path, source))]
66    GetRequest {
67        source: crate::client::retry::Error,
68        path: String,
69    },
70
71    #[snafu(display("Error performing put request {}: {}", path, source))]
72    PutRequest {
73        source: crate::client::retry::Error,
74        path: String,
75    },
76
77    #[snafu(display("Error performing delete request {}: {}", path, source))]
78    DeleteRequest {
79        source: crate::client::retry::Error,
80        path: String,
81    },
82
83    #[snafu(display("Error performing list request: {}", source))]
84    ListRequest { source: crate::client::retry::Error },
85
86    #[snafu(display("Error getting list response body: {}", source))]
87    ListResponseBody { source: reqwest::Error },
88
89    #[snafu(display("Got invalid list response: {}", source))]
90    InvalidListResponse { source: quick_xml::de::DeError },
91
92    #[snafu(display("Unable to extract metadata from headers: {}", source))]
93    Metadata {
94        source: crate::client::header::Error,
95    },
96
97    #[snafu(display("ETag required for conditional update"))]
98    MissingETag,
99
100    #[snafu(display("Error requesting user delegation key: {}", source))]
101    DelegationKeyRequest { source: crate::client::retry::Error },
102
103    #[snafu(display("Error getting user delegation key response body: {}", source))]
104    DelegationKeyResponseBody { source: reqwest::Error },
105
106    #[snafu(display("Got invalid user delegation key response: {}", source))]
107    DelegationKeyResponse { source: quick_xml::de::DeError },
108
109    #[snafu(display("Generating SAS keys with SAS tokens auth is not supported"))]
110    SASforSASNotSupported,
111
112    #[snafu(display("Generating SAS keys while skipping signatures is not supported"))]
113    SASwithSkipSignature,
114}
115
116impl From<Error> for crate::Error {
117    fn from(err: Error) -> Self {
118        match err {
119            Error::GetRequest { source, path }
120            | Error::DeleteRequest { source, path }
121            | Error::PutRequest { source, path } => source.error(STORE, path),
122            _ => Self::Generic {
123                store: STORE,
124                source: Box::new(err),
125            },
126        }
127    }
128}
129
130/// Configuration for [AzureClient]
131#[derive(Debug)]
132pub(crate) struct AzureConfig {
133    pub account: String,
134    pub container: String,
135    pub credentials: AzureCredentialProvider,
136    pub retry_config: RetryConfig,
137    pub service: Url,
138    pub is_emulator: bool,
139    pub skip_signature: bool,
140    pub disable_tagging: bool,
141    pub client_options: ClientOptions,
142}
143
144impl AzureConfig {
145    pub(crate) fn path_url(&self, path: &Path) -> Url {
146        let mut url = self.service.clone();
147        {
148            let mut path_mut = url.path_segments_mut().unwrap();
149            if self.is_emulator {
150                path_mut.push(&self.account);
151            }
152            path_mut.push(&self.container).extend(path.parts());
153        }
154        url
155    }
156    async fn get_credential(&self) -> Result<Option<Arc<AzureCredential>>> {
157        if self.skip_signature {
158            Ok(None)
159        } else {
160            Some(self.credentials.get_credential().await).transpose()
161        }
162    }
163}
164
165/// A builder for a put request allowing customisation of the headers and query string
166struct PutRequest<'a> {
167    path: &'a Path,
168    config: &'a AzureConfig,
169    payload: PutPayload,
170    builder: RequestBuilder,
171    idempotent: bool,
172}
173
174impl<'a> PutRequest<'a> {
175    fn header(self, k: &HeaderName, v: &str) -> Self {
176        let builder = self.builder.header(k, v);
177        Self { builder, ..self }
178    }
179
180    fn query<T: Serialize + ?Sized + Sync>(self, query: &T) -> Self {
181        let builder = self.builder.query(query);
182        Self { builder, ..self }
183    }
184
185    fn idempotent(self, idempotent: bool) -> Self {
186        Self { idempotent, ..self }
187    }
188
189    fn with_tags(mut self, tags: TagSet) -> Self {
190        let tags = tags.encoded();
191        if !tags.is_empty() && !self.config.disable_tagging {
192            self.builder = self.builder.header(&TAGS_HEADER, tags);
193        }
194        self
195    }
196
197    fn with_attributes(self, attributes: Attributes) -> Self {
198        let mut builder = self.builder;
199        let mut has_content_type = false;
200        for (k, v) in &attributes {
201            builder = match k {
202                Attribute::CacheControl => builder.header(&MS_CACHE_CONTROL, v.as_ref()),
203                Attribute::ContentDisposition => {
204                    builder.header(&MS_CONTENT_DISPOSITION, v.as_ref())
205                }
206                Attribute::ContentEncoding => builder.header(&MS_CONTENT_ENCODING, v.as_ref()),
207                Attribute::ContentLanguage => builder.header(&MS_CONTENT_LANGUAGE, v.as_ref()),
208                Attribute::ContentType => {
209                    has_content_type = true;
210                    builder.header(&MS_CONTENT_TYPE, v.as_ref())
211                }
212                Attribute::Metadata(k_suffix) => builder.header(
213                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
214                    v.as_ref(),
215                ),
216            };
217        }
218
219        if !has_content_type {
220            if let Some(value) = self.config.client_options.get_content_type(self.path) {
221                builder = builder.header(&MS_CONTENT_TYPE, value);
222            }
223        }
224        Self { builder, ..self }
225    }
226
227    async fn send(self) -> Result<Response> {
228        let credential = self.config.get_credential().await?;
229        let response = self
230            .builder
231            .header(CONTENT_LENGTH, self.payload.content_length())
232            .with_azure_authorization(&credential, &self.config.account)
233            .retryable(&self.config.retry_config)
234            .idempotent(self.idempotent)
235            .payload(Some(self.payload))
236            .send()
237            .await
238            .context(PutRequestSnafu {
239                path: self.path.as_ref(),
240            })?;
241
242        Ok(response)
243    }
244}
245
246#[derive(Debug)]
247pub(crate) struct AzureClient {
248    config: AzureConfig,
249    client: ReqwestClient,
250}
251
252impl AzureClient {
253    /// create a new instance of [AzureClient]
254    pub fn new(config: AzureConfig) -> Result<Self> {
255        let client = config.client_options.client()?;
256        Ok(Self { config, client })
257    }
258
259    /// Returns the config
260    pub fn config(&self) -> &AzureConfig {
261        &self.config
262    }
263
264    async fn get_credential(&self) -> Result<Option<Arc<AzureCredential>>> {
265        self.config.get_credential().await
266    }
267
268    fn put_request<'a>(&'a self, path: &'a Path, payload: PutPayload) -> PutRequest<'a> {
269        let url = self.config.path_url(path);
270        let builder = self.client.request(Method::PUT, url);
271
272        PutRequest {
273            path,
274            builder,
275            payload,
276            config: &self.config,
277            idempotent: false,
278        }
279    }
280
281    /// Make an Azure PUT request <https://docs.microsoft.com/en-us/rest/api/storageservices/put-blob>
282    pub async fn put_blob(
283        &self,
284        path: &Path,
285        payload: PutPayload,
286        opts: PutOptions,
287    ) -> Result<PutResult> {
288        let builder = self
289            .put_request(path, payload)
290            .with_attributes(opts.attributes)
291            .with_tags(opts.tags);
292
293        let builder = match &opts.mode {
294            PutMode::Overwrite => builder.idempotent(true),
295            PutMode::Create => builder.header(&IF_NONE_MATCH, "*"),
296            PutMode::Update(v) => {
297                let etag = v.e_tag.as_ref().context(MissingETagSnafu)?;
298                builder.header(&IF_MATCH, etag)
299            }
300        };
301
302        let response = builder.header(&BLOB_TYPE, "BlockBlob").send().await?;
303        Ok(get_put_result(response.headers(), VERSION_HEADER).context(MetadataSnafu)?)
304    }
305
306    /// PUT a block <https://learn.microsoft.com/en-us/rest/api/storageservices/put-block>
307    pub async fn put_block(
308        &self,
309        path: &Path,
310        part_idx: usize,
311        payload: PutPayload,
312    ) -> Result<PartId> {
313        let content_id = format!("{part_idx:20}");
314        let block_id = BASE64_STANDARD.encode(&content_id);
315
316        self.put_request(path, payload)
317            .query(&[("comp", "block"), ("blockid", &block_id)])
318            .idempotent(true)
319            .send()
320            .await?;
321
322        Ok(PartId { content_id })
323    }
324
325    /// PUT a block list <https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list>
326    pub async fn put_block_list(
327        &self,
328        path: &Path,
329        parts: Vec<PartId>,
330        opts: PutMultipartOpts,
331    ) -> Result<PutResult> {
332        let blocks = parts
333            .into_iter()
334            .map(|part| BlockId::from(part.content_id))
335            .collect();
336
337        let payload = BlockList { blocks }.to_xml().into();
338        let response = self
339            .put_request(path, payload)
340            .with_attributes(opts.attributes)
341            .with_tags(opts.tags)
342            .query(&[("comp", "blocklist")])
343            .idempotent(true)
344            .send()
345            .await?;
346
347        Ok(get_put_result(response.headers(), VERSION_HEADER).context(MetadataSnafu)?)
348    }
349
350    /// Make an Azure Delete request <https://docs.microsoft.com/en-us/rest/api/storageservices/delete-blob>
351    pub async fn delete_request<T: Serialize + ?Sized + Sync>(
352        &self,
353        path: &Path,
354        query: &T,
355    ) -> Result<()> {
356        let credential = self.get_credential().await?;
357        let url = self.config.path_url(path);
358
359        self.client
360            .request(Method::DELETE, url)
361            .query(query)
362            .header(&DELETE_SNAPSHOTS, "include")
363            .with_azure_authorization(&credential, &self.config.account)
364            .send_retry(&self.config.retry_config)
365            .await
366            .context(DeleteRequestSnafu {
367                path: path.as_ref(),
368            })?;
369
370        Ok(())
371    }
372
373    /// Make an Azure Copy request <https://docs.microsoft.com/en-us/rest/api/storageservices/copy-blob>
374    pub async fn copy_request(&self, from: &Path, to: &Path, overwrite: bool) -> Result<()> {
375        let credential = self.get_credential().await?;
376        let url = self.config.path_url(to);
377        let mut source = self.config.path_url(from);
378
379        // If using SAS authorization must include the headers in the URL
380        // <https://docs.microsoft.com/en-us/rest/api/storageservices/copy-blob#request-headers>
381        if let Some(AzureCredential::SASToken(pairs)) = credential.as_deref() {
382            source.query_pairs_mut().extend_pairs(pairs);
383        }
384
385        let mut builder = self
386            .client
387            .request(Method::PUT, url)
388            .header(&COPY_SOURCE, source.to_string())
389            .header(CONTENT_LENGTH, HeaderValue::from_static("0"));
390
391        if !overwrite {
392            builder = builder.header(IF_NONE_MATCH, "*");
393        }
394
395        builder
396            .with_azure_authorization(&credential, &self.config.account)
397            .retryable(&self.config.retry_config)
398            .idempotent(overwrite)
399            .send()
400            .await
401            .map_err(|err| err.error(STORE, from.to_string()))?;
402
403        Ok(())
404    }
405
406    /// Make a Get User Delegation Key request
407    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-user-delegation-key>
408    async fn get_user_delegation_key(
409        &self,
410        start: &DateTime<Utc>,
411        end: &DateTime<Utc>,
412    ) -> Result<UserDelegationKey> {
413        let credential = self.get_credential().await?;
414        let url = self.config.service.clone();
415
416        let start = start.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
417        let expiry = end.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
418
419        let mut body = String::new();
420        body.push_str("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<KeyInfo>\n");
421        body.push_str(&format!(
422            "\t<Start>{start}</Start>\n\t<Expiry>{expiry}</Expiry>\n"
423        ));
424        body.push_str("</KeyInfo>");
425
426        let response = self
427            .client
428            .request(Method::POST, url)
429            .body(body)
430            .query(&[("restype", "service"), ("comp", "userdelegationkey")])
431            .with_azure_authorization(&credential, &self.config.account)
432            .retryable(&self.config.retry_config)
433            .idempotent(true)
434            .send()
435            .await
436            .context(DelegationKeyRequestSnafu)?
437            .bytes()
438            .await
439            .context(DelegationKeyResponseBodySnafu)?;
440
441        let response: UserDelegationKey =
442            quick_xml::de::from_reader(response.reader()).context(DelegationKeyResponseSnafu)?;
443
444        Ok(response)
445    }
446
447    /// Creat an AzureSigner for generating SAS tokens (pre-signed urls).
448    ///
449    /// Depending on the type of credential, this will either use the account key or a user delegation key.
450    /// Since delegation keys are acquired ad-hoc, the signer aloows for signing multiple urls with the same key.
451    pub async fn signer(&self, expires_in: Duration) -> Result<AzureSigner> {
452        let credential = self.get_credential().await?;
453        let signed_start = chrono::Utc::now();
454        let signed_expiry = signed_start + expires_in;
455        match credential.as_deref() {
456            Some(AzureCredential::BearerToken(_)) => {
457                let key = self
458                    .get_user_delegation_key(&signed_start, &signed_expiry)
459                    .await?;
460                let signing_key = AzureAccessKey::try_new(&key.value)?;
461                Ok(AzureSigner::new(
462                    signing_key,
463                    self.config.account.clone(),
464                    signed_start,
465                    signed_expiry,
466                    Some(key),
467                ))
468            }
469            Some(AzureCredential::AccessKey(key)) => Ok(AzureSigner::new(
470                key.to_owned(),
471                self.config.account.clone(),
472                signed_start,
473                signed_expiry,
474                None,
475            )),
476            None => Err(Error::SASwithSkipSignature.into()),
477            _ => Err(Error::SASforSASNotSupported.into()),
478        }
479    }
480
481    #[cfg(test)]
482    pub async fn get_blob_tagging(&self, path: &Path) -> Result<Response> {
483        let credential = self.get_credential().await?;
484        let url = self.config.path_url(path);
485        let response = self
486            .client
487            .request(Method::GET, url)
488            .query(&[("comp", "tags")])
489            .with_azure_authorization(&credential, &self.config.account)
490            .send_retry(&self.config.retry_config)
491            .await
492            .context(GetRequestSnafu {
493                path: path.as_ref(),
494            })?;
495        Ok(response)
496    }
497}
498
499#[async_trait]
500impl GetClient for AzureClient {
501    const STORE: &'static str = STORE;
502
503    const HEADER_CONFIG: HeaderConfig = HeaderConfig {
504        etag_required: true,
505        last_modified_required: true,
506        version_header: Some(VERSION_HEADER),
507        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
508    };
509
510    /// Make an Azure GET request
511    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob>
512    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties>
513    async fn get_request(&self, path: &Path, options: GetOptions) -> Result<Response> {
514        // As of 2024-01-02, Azure does not support suffix requests,
515        // so we should fail fast here rather than sending one
516        if let Some(GetRange::Suffix(_)) = options.range.as_ref() {
517            return Err(crate::Error::NotSupported {
518                source: "Azure does not support suffix range requests".into(),
519            });
520        }
521
522        let credential = self.get_credential().await?;
523        let url = self.config.path_url(path);
524        let method = match options.head {
525            true => Method::HEAD,
526            false => Method::GET,
527        };
528
529        let mut builder = self
530            .client
531            .request(method, url)
532            .header(CONTENT_LENGTH, HeaderValue::from_static("0"))
533            .body(Bytes::new());
534
535        if let Some(v) = &options.version {
536            builder = builder.query(&[("versionid", v)])
537        }
538
539        let response = builder
540            .with_get_options(options)
541            .with_azure_authorization(&credential, &self.config.account)
542            .send_retry(&self.config.retry_config)
543            .await
544            .context(GetRequestSnafu {
545                path: path.as_ref(),
546            })?;
547
548        match response.headers().get("x-ms-resource-type") {
549            Some(resource) if resource.as_ref() != b"file" => Err(crate::Error::NotFound {
550                path: path.to_string(),
551                source: format!(
552                    "Not a file, got x-ms-resource-type: {}",
553                    String::from_utf8_lossy(resource.as_ref())
554                )
555                .into(),
556            }),
557            _ => Ok(response),
558        }
559    }
560}
561
562#[async_trait]
563impl ListClient for AzureClient {
564    /// Make an Azure List request <https://docs.microsoft.com/en-us/rest/api/storageservices/list-blobs>
565    async fn list_request(
566        &self,
567        prefix: Option<&str>,
568        delimiter: bool,
569        token: Option<&str>,
570        offset: Option<&str>,
571    ) -> Result<(ListResult, Option<String>)> {
572        assert!(offset.is_none()); // Not yet supported
573
574        let credential = self.get_credential().await?;
575        let url = self.config.path_url(&Path::default());
576
577        let mut query = Vec::with_capacity(5);
578        query.push(("restype", "container"));
579        query.push(("comp", "list"));
580
581        if let Some(prefix) = prefix {
582            query.push(("prefix", prefix))
583        }
584
585        if delimiter {
586            query.push(("delimiter", DELIMITER))
587        }
588
589        if let Some(token) = token {
590            query.push(("marker", token))
591        }
592
593        let response = self
594            .client
595            .request(Method::GET, url)
596            .query(&query)
597            .with_azure_authorization(&credential, &self.config.account)
598            .send_retry(&self.config.retry_config)
599            .await
600            .context(ListRequestSnafu)?
601            .bytes()
602            .await
603            .context(ListResponseBodySnafu)?;
604
605        let mut response: ListResultInternal =
606            quick_xml::de::from_reader(response.reader()).context(InvalidListResponseSnafu)?;
607        let token = response.next_marker.take();
608
609        Ok((to_list_result(response, prefix)?, token))
610    }
611}
612
613/// Raw / internal response from list requests
614#[derive(Debug, Clone, PartialEq, Deserialize)]
615#[serde(rename_all = "PascalCase")]
616struct ListResultInternal {
617    pub prefix: Option<String>,
618    pub max_results: Option<u32>,
619    pub delimiter: Option<String>,
620    pub next_marker: Option<String>,
621    pub blobs: Blobs,
622}
623
624fn to_list_result(value: ListResultInternal, prefix: Option<&str>) -> Result<ListResult> {
625    let prefix = prefix.unwrap_or_default();
626    let common_prefixes = value
627        .blobs
628        .blob_prefix
629        .into_iter()
630        .map(|x| Ok(Path::parse(x.name)?))
631        .collect::<Result<_>>()?;
632
633    let objects = value
634        .blobs
635        .blobs
636        .into_iter()
637        // Note: Filters out directories from list results when hierarchical namespaces are
638        // enabled. When we want directories, its always via the BlobPrefix mechanics,
639        // and during lists we state that prefixes are evaluated on path segment basis.
640        .filter(|blob| {
641            !matches!(blob.properties.resource_type.as_ref(), Some(typ) if typ == "directory")
642                && blob.name.len() > prefix.len()
643        })
644        .map(ObjectMeta::try_from)
645        .collect::<Result<_>>()?;
646
647    Ok(ListResult {
648        common_prefixes,
649        objects,
650    })
651}
652
653/// Collection of blobs and potentially shared prefixes returned from list requests.
654#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
655#[serde(rename_all = "PascalCase")]
656struct Blobs {
657    #[serde(default)]
658    pub blob_prefix: Vec<BlobPrefix>,
659    #[serde(rename = "Blob", default)]
660    pub blobs: Vec<Blob>,
661}
662
663/// Common prefix in list blobs response
664#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
665#[serde(rename_all = "PascalCase")]
666struct BlobPrefix {
667    pub name: String,
668}
669
670/// Details for a specific blob
671#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
672#[serde(rename_all = "PascalCase")]
673struct Blob {
674    pub name: String,
675    pub version_id: Option<String>,
676    pub is_current_version: Option<bool>,
677    pub deleted: Option<bool>,
678    pub properties: BlobProperties,
679    pub metadata: Option<HashMap<String, String>>,
680}
681
682impl TryFrom<Blob> for ObjectMeta {
683    type Error = crate::Error;
684
685    fn try_from(value: Blob) -> Result<Self> {
686        Ok(Self {
687            location: Path::parse(value.name)?,
688            last_modified: value.properties.last_modified,
689            size: value.properties.content_length as usize,
690            e_tag: value.properties.e_tag,
691            version: None, // For consistency with S3 and GCP which don't include this
692        })
693    }
694}
695
696/// Properties associated with individual blobs. The actual list
697/// of returned properties is much more exhaustive, but we limit
698/// the parsed fields to the ones relevant in this crate.
699#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
700#[serde(rename_all = "PascalCase")]
701struct BlobProperties {
702    #[serde(deserialize_with = "deserialize_rfc1123", rename = "Last-Modified")]
703    pub last_modified: DateTime<Utc>,
704    #[serde(rename = "Content-Length")]
705    pub content_length: u64,
706    #[serde(rename = "Content-Type")]
707    pub content_type: String,
708    #[serde(rename = "Content-Encoding")]
709    pub content_encoding: Option<String>,
710    #[serde(rename = "Content-Language")]
711    pub content_language: Option<String>,
712    #[serde(rename = "Etag")]
713    pub e_tag: Option<String>,
714    #[serde(rename = "ResourceType")]
715    pub resource_type: Option<String>,
716}
717
718#[derive(Debug, Clone, PartialEq, Eq)]
719pub(crate) struct BlockId(Bytes);
720
721impl BlockId {
722    pub fn new(block_id: impl Into<Bytes>) -> Self {
723        Self(block_id.into())
724    }
725}
726
727impl<B> From<B> for BlockId
728where
729    B: Into<Bytes>,
730{
731    fn from(v: B) -> Self {
732        Self::new(v)
733    }
734}
735
736impl AsRef<[u8]> for BlockId {
737    fn as_ref(&self) -> &[u8] {
738        self.0.as_ref()
739    }
740}
741
742#[derive(Default, Debug, Clone, PartialEq, Eq)]
743pub(crate) struct BlockList {
744    pub blocks: Vec<BlockId>,
745}
746
747impl BlockList {
748    pub fn to_xml(&self) -> String {
749        let mut s = String::new();
750        s.push_str("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<BlockList>\n");
751        for block_id in &self.blocks {
752            let node = format!(
753                "\t<Uncommitted>{}</Uncommitted>\n",
754                BASE64_STANDARD.encode(block_id)
755            );
756            s.push_str(&node);
757        }
758
759        s.push_str("</BlockList>");
760        s
761    }
762}
763
764#[derive(Debug, Clone, PartialEq, Deserialize)]
765#[serde(rename_all = "PascalCase")]
766pub(crate) struct UserDelegationKey {
767    pub signed_oid: String,
768    pub signed_tid: String,
769    pub signed_start: String,
770    pub signed_expiry: String,
771    pub signed_service: String,
772    pub signed_version: String,
773    pub value: String,
774}
775
776#[cfg(test)]
777mod tests {
778    use bytes::Bytes;
779
780    use super::*;
781
782    #[test]
783    fn deserde_azure() {
784        const S: &str = "<?xml version=\"1.0\" encoding=\"utf-8\"?>
785<EnumerationResults ServiceEndpoint=\"https://azureskdforrust.blob.core.windows.net/\" ContainerName=\"osa2\">
786    <Blobs>
787        <Blob>
788            <Name>blob0.txt</Name>
789            <Properties>
790                <Creation-Time>Thu, 01 Jul 2021 10:44:59 GMT</Creation-Time>
791                <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>
792                <Expiry-Time>Thu, 07 Jul 2022 14:38:48 GMT</Expiry-Time>
793                <Etag>0x8D93C7D4629C227</Etag>
794                <Content-Length>8</Content-Length>
795                <Content-Type>text/plain</Content-Type>
796                <Content-Encoding />
797                <Content-Language />
798                <Content-CRC64 />
799                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
800                <Cache-Control />
801                <Content-Disposition />
802                <BlobType>BlockBlob</BlobType>
803                <AccessTier>Hot</AccessTier>
804                <AccessTierInferred>true</AccessTierInferred>
805                <LeaseStatus>unlocked</LeaseStatus>
806                <LeaseState>available</LeaseState>
807                <ServerEncrypted>true</ServerEncrypted>
808            </Properties>
809            <Metadata><userkey>uservalue</userkey></Metadata>
810            <OrMetadata />
811        </Blob>
812        <Blob>
813            <Name>blob1.txt</Name>
814            <Properties>
815                <Creation-Time>Thu, 01 Jul 2021 10:44:59 GMT</Creation-Time>
816                <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>
817                <Etag>0x8D93C7D463004D6</Etag>
818                <Content-Length>8</Content-Length>
819                <Content-Type>text/plain</Content-Type>
820                <Content-Encoding />
821                <Content-Language />
822                <Content-CRC64 />
823                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
824                <Cache-Control />
825                <Content-Disposition />
826                <BlobType>BlockBlob</BlobType>
827                <AccessTier>Hot</AccessTier>
828                <AccessTierInferred>true</AccessTierInferred>
829                <LeaseStatus>unlocked</LeaseStatus>
830                <LeaseState>available</LeaseState>
831                <ServerEncrypted>true</ServerEncrypted>
832            </Properties>
833            <OrMetadata />
834        </Blob>
835        <Blob>
836            <Name>blob2.txt</Name>
837            <Properties>
838                <Creation-Time>Thu, 01 Jul 2021 10:44:59 GMT</Creation-Time>
839                <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>
840                <Etag>0x8D93C7D4636478A</Etag>
841                <Content-Length>8</Content-Length>
842                <Content-Type>text/plain</Content-Type>
843                <Content-Encoding />
844                <Content-Language />
845                <Content-CRC64 />
846                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
847                <Cache-Control />
848                <Content-Disposition />
849                <BlobType>BlockBlob</BlobType>
850                <AccessTier>Hot</AccessTier>
851                <AccessTierInferred>true</AccessTierInferred>
852                <LeaseStatus>unlocked</LeaseStatus>
853                <LeaseState>available</LeaseState>
854                <ServerEncrypted>true</ServerEncrypted>
855            </Properties>
856            <OrMetadata />
857        </Blob>
858    </Blobs>
859    <NextMarker />
860</EnumerationResults>";
861
862        let mut _list_blobs_response_internal: ListResultInternal =
863            quick_xml::de::from_str(S).unwrap();
864    }
865
866    #[test]
867    fn deserde_azurite() {
868        const S: &str = "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?>
869<EnumerationResults ServiceEndpoint=\"http://127.0.0.1:10000/devstoreaccount1\" ContainerName=\"osa2\">
870    <Prefix/>
871    <Marker/>
872    <MaxResults>5000</MaxResults>
873    <Delimiter/>
874    <Blobs>
875        <Blob>
876            <Name>blob0.txt</Name>
877            <Properties>
878                <Creation-Time>Thu, 01 Jul 2021 10:45:02 GMT</Creation-Time>
879                <Last-Modified>Thu, 01 Jul 2021 10:45:02 GMT</Last-Modified>
880                <Etag>0x228281B5D517B20</Etag>
881                <Content-Length>8</Content-Length>
882                <Content-Type>text/plain</Content-Type>
883                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
884                <BlobType>BlockBlob</BlobType>
885                <LeaseStatus>unlocked</LeaseStatus>
886                <LeaseState>available</LeaseState>
887                <ServerEncrypted>true</ServerEncrypted>
888                <AccessTier>Hot</AccessTier>
889                <AccessTierInferred>true</AccessTierInferred>
890                <AccessTierChangeTime>Thu, 01 Jul 2021 10:45:02 GMT</AccessTierChangeTime>
891            </Properties>
892        </Blob>
893        <Blob>
894            <Name>blob1.txt</Name>
895            <Properties>
896                <Creation-Time>Thu, 01 Jul 2021 10:45:02 GMT</Creation-Time>
897                <Last-Modified>Thu, 01 Jul 2021 10:45:02 GMT</Last-Modified>
898                <Etag>0x1DD959381A8A860</Etag>
899                <Content-Length>8</Content-Length>
900                <Content-Type>text/plain</Content-Type>
901                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
902                <BlobType>BlockBlob</BlobType>
903                <LeaseStatus>unlocked</LeaseStatus>
904                <LeaseState>available</LeaseState>
905                <ServerEncrypted>true</ServerEncrypted>
906                <AccessTier>Hot</AccessTier>
907                <AccessTierInferred>true</AccessTierInferred>
908                <AccessTierChangeTime>Thu, 01 Jul 2021 10:45:02 GMT</AccessTierChangeTime>
909            </Properties>
910        </Blob>
911        <Blob>
912            <Name>blob2.txt</Name>
913            <Properties>
914                <Creation-Time>Thu, 01 Jul 2021 10:45:02 GMT</Creation-Time>
915                <Last-Modified>Thu, 01 Jul 2021 10:45:02 GMT</Last-Modified>
916                <Etag>0x1FBE9C9B0C7B650</Etag>
917                <Content-Length>8</Content-Length>
918                <Content-Type>text/plain</Content-Type>
919                <Content-MD5>rvr3UC1SmUw7AZV2NqPN0g==</Content-MD5>
920                <BlobType>BlockBlob</BlobType>
921                <LeaseStatus>unlocked</LeaseStatus>
922                <LeaseState>available</LeaseState>
923                <ServerEncrypted>true</ServerEncrypted>
924                <AccessTier>Hot</AccessTier>
925                <AccessTierInferred>true</AccessTierInferred>
926                <AccessTierChangeTime>Thu, 01 Jul 2021 10:45:02 GMT</AccessTierChangeTime>
927            </Properties>
928        </Blob>
929    </Blobs>
930    <NextMarker/>
931</EnumerationResults>";
932
933        let _list_blobs_response_internal: ListResultInternal = quick_xml::de::from_str(S).unwrap();
934    }
935
936    #[test]
937    fn to_xml() {
938        const S: &str = "<?xml version=\"1.0\" encoding=\"utf-8\"?>
939<BlockList>
940\t<Uncommitted>bnVtZXJvMQ==</Uncommitted>
941\t<Uncommitted>bnVtZXJvMg==</Uncommitted>
942\t<Uncommitted>bnVtZXJvMw==</Uncommitted>
943</BlockList>";
944        let mut blocks = BlockList { blocks: Vec::new() };
945        blocks.blocks.push(Bytes::from_static(b"numero1").into());
946        blocks.blocks.push("numero2".into());
947        blocks.blocks.push("numero3".into());
948
949        let res: &str = &blocks.to_xml();
950
951        assert_eq!(res, S)
952    }
953
954    #[test]
955    fn test_delegated_key_response() {
956        const S: &str = r#"<?xml version="1.0" encoding="utf-8"?>
957<UserDelegationKey>
958    <SignedOid>String containing a GUID value</SignedOid>
959    <SignedTid>String containing a GUID value</SignedTid>
960    <SignedStart>String formatted as ISO date</SignedStart>
961    <SignedExpiry>String formatted as ISO date</SignedExpiry>
962    <SignedService>b</SignedService>
963    <SignedVersion>String specifying REST api version to use to create the user delegation key</SignedVersion>
964    <Value>String containing the user delegation key</Value>
965</UserDelegationKey>"#;
966
967        let _delegated_key_response_internal: UserDelegationKey =
968            quick_xml::de::from_str(S).unwrap();
969    }
970}
object_store/azure/client.rs

object_store/azure/
client.rs
