azure_storage/
clients.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
use crate::{
    authorization::{AuthorizationPolicy, StorageCredentialsInner},
    shared_access_signature::account_sas::{
        AccountSasPermissions, AccountSasResource, AccountSasResourceType,
        AccountSharedAccessSignature,
    },
    StorageCredentials,
};
use azure_core::{
    date,
    error::{Error, ErrorKind},
    headers::*,
    Body, ClientOptions, Method, Pipeline, Request, Url,
};
use std::{ops::Deref, sync::Arc};
use time::OffsetDateTime;

/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
pub const EMULATOR_ACCOUNT: &str = "devstoreaccount1";

/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
pub const EMULATOR_ACCOUNT_KEY: &str =
    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";

const AZURE_VERSION: HeaderValue = HeaderValue::from_static("2022-11-02");

#[derive(Debug, Clone, Copy)]
pub enum ServiceType {
    Blob,
    Queue,
    // File,
    Table,
    DataLake,
}

impl ServiceType {
    pub fn subdomain(&self) -> &str {
        match self {
            ServiceType::Blob => "blob",
            ServiceType::Queue => "queue",
            ServiceType::Table => "table",
            ServiceType::DataLake => "dfs",
        }
    }
}

pub async fn shared_access_signature(
    storage_credentials: &StorageCredentials,
    resource: AccountSasResource,
    resource_type: AccountSasResourceType,
    expiry: OffsetDateTime,
    permissions: AccountSasPermissions,
) -> Result<AccountSharedAccessSignature, Error> {
    let creds = storage_credentials.0.read().await;
    let StorageCredentialsInner::Key(account, key) = creds.deref() else {
        return Err(Error::message(
            ErrorKind::Credential,
            "Shared access signature generation - SAS can be generated with access_key clients",
        ));
    };

    Ok(AccountSharedAccessSignature::new(
        account.clone(),
        key.clone(),
        resource,
        resource_type,
        expiry,
        permissions,
    ))
}

pub fn finalize_request(
    url: Url,
    method: Method,
    headers: Headers,
    request_body: Option<Body>,
) -> Result<Request, Error> {
    let dt = OffsetDateTime::now_utc();
    let time = date::to_rfc1123(&dt);
    let mut request = Request::new(url, method);
    for (k, v) in headers {
        request.insert_header(k, v);
    }
    // let's add content length to avoid "chunking" errors.
    match request_body {
        Some(ref b) => request.insert_header(CONTENT_LENGTH, b.len().to_string()),
        None => request.insert_header(CONTENT_LENGTH, "0"),
    };
    request.insert_header(MS_DATE, time);
    request.insert_header(VERSION, AZURE_VERSION);
    if let Some(request_body) = request_body {
        request.set_body(request_body);
    } else {
        request.set_body(azure_core::EMPTY_BODY);
    };
    Ok(request)
}

/// Create a Pipeline from `ClientOptions`
pub fn new_pipeline_from_options(
    options: ClientOptions,
    credentials: StorageCredentials,
) -> Pipeline {
    let auth_policy: Arc<dyn azure_core::Policy> = Arc::new(AuthorizationPolicy::new(credentials));

    // The `AuthorizationPolicy` must be the **last** retry policy.
    // Policies can change the url and/or the headers, and the `AuthorizationPolicy`
    // must be able to inspect them or the resulting token will be invalid.
    let per_retry_policies = vec![
        Arc::new(options.timeout.clone()) as Arc<dyn azure_core::Policy>,
        auth_policy,
    ];

    Pipeline::new(
        option_env!("CARGO_PKG_NAME"),
        option_env!("CARGO_PKG_VERSION"),
        options,
        Vec::new(),
        per_retry_policies,
    )
}