rws_storage/storage/
azure_blob.rs

1use super::azure_credentials::{AzureCredentialsProvider, Credential};
2use super::azure_signature;
3use super::{Storage, StorageError};
4use crate::http_client::{Client, Response};
5use std::env;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8/// `x-ms-version` sent on every request, regardless of auth scheme. Any
9/// reasonably recent stable version works for basic Put/Get/Delete Blob;
10/// bump only if a newer feature is needed.
11const API_VERSION: &str = "2021-08-06";
12
13/// Connection details for an Azure Blob Storage container.
14#[derive(Clone, Debug)]
15pub struct AzureBlobConfig {
16    pub account: String,
17    pub container: String,
18    /// Empty string means "no static key — auto-detect Managed Identity
19    /// (VM/AKS IMDS, or App Service/Container Apps identity endpoint)
20    /// instead". See [`AzureBlobStorage::new`].
21    pub account_key: String,
22    /// Scheme + host, e.g. `https://myaccount.blob.core.windows.net`.
23    /// Override for the Azurite local emulator or a private endpoint.
24    pub endpoint: String,
25}
26
27impl AzureBlobConfig {
28    /// Read configuration from environment variables.
29    ///
30    /// | Variable | Default |
31    /// |---|---|
32    /// | `RWS_AZURE_ACCOUNT` | **(required)** |
33    /// | `RWS_AZURE_CONTAINER` | **(required)** |
34    /// | `RWS_AZURE_ACCOUNT_KEY` | optional — falls back to Managed Identity when unset |
35    /// | `RWS_AZURE_ENDPOINT` | `https://{account}.blob.core.windows.net` |
36    pub fn from_env() -> Result<Self, StorageError> {
37        let account = env::var("RWS_AZURE_ACCOUNT")
38            .map_err(|_| StorageError::new("RWS_AZURE_ACCOUNT environment variable is required"))?;
39        let container = env::var("RWS_AZURE_CONTAINER")
40            .map_err(|_| StorageError::new("RWS_AZURE_CONTAINER environment variable is required"))?;
41        let account_key = env::var("RWS_AZURE_ACCOUNT_KEY").unwrap_or_default();
42        let endpoint =
43            env::var("RWS_AZURE_ENDPOINT").unwrap_or_else(|_| format!("https://{account}.blob.core.windows.net"));
44        Ok(AzureBlobConfig { account, container, account_key, endpoint })
45    }
46}
47
48/// Azure Blob Storage object storage.
49///
50/// Signs every request with the Shared Key HMAC-SHA256 scheme using the
51/// outbound HTTP client (`crate::http_client::Client`) — no Azure SDK
52/// dependency. Credentials come from `config.account_key` when non-empty;
53/// otherwise they're auto-detected from Managed Identity (App
54/// Service/Container Apps identity endpoint, or VM/AKS IMDS as a last
55/// resort), cached, and refreshed shortly before expiry. See
56/// `crate::storage::azure_credentials` and `crate::storage::azure_signature`.
57pub struct AzureBlobStorage {
58    config: AzureBlobConfig,
59    client: Client,
60    credentials: AzureCredentialsProvider,
61}
62
63impl AzureBlobStorage {
64    pub fn new(config: AzureBlobConfig) -> Self {
65        let account_key = non_empty(&config.account_key);
66        let credentials = AzureCredentialsProvider::detect(account_key);
67        AzureBlobStorage { config, client: Client::new(), credentials }
68    }
69
70    /// Shortcut for `AzureBlobStorage::new(AzureBlobConfig::from_env()?)`.
71    pub fn from_env() -> Result<Self, StorageError> {
72        Ok(AzureBlobStorage::new(AzureBlobConfig::from_env()?))
73    }
74
75    /// Percent-encoded `/{container}/{key}` — used both as the request path
76    /// and as the canonical resource signed by Shared Key, so the two always
77    /// match. Not shared with `aws_sigv4::uri_encode_path`: that helper only
78    /// compiles under the independently-gated `storage-s3` feature.
79    fn canonical_path(&self, key: &str) -> String {
80        format!("/{}/{}", self.config.container, uri_encode_key(key))
81    }
82
83    fn request_url(&self, key: &str) -> String {
84        format!("{}{}", self.config.endpoint.trim_end_matches('/'), self.canonical_path(key))
85    }
86
87    /// Builds the full header set for a request: `x-ms-date`, `x-ms-version`,
88    /// any caller-supplied extra `x-ms-*` headers (e.g. `x-ms-blob-type` on
89    /// `PUT`), and `Authorization` — either `SharedKey ...` (computed over
90    /// exactly this header set) or `Bearer {token}` for Managed Identity.
91    fn auth_headers(
92        &self,
93        method: &str,
94        key: &str,
95        content_type: &str,
96        content_length: usize,
97        extra_x_ms: &[(String, String)],
98    ) -> Result<Vec<(String, String)>, StorageError> {
99        let epoch_secs = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
100        let x_ms_date = azure_signature::rfc1123_date(epoch_secs);
101
102        let mut headers = vec![("x-ms-date".to_string(), x_ms_date), ("x-ms-version".to_string(), API_VERSION.to_string())];
103        headers.extend(extra_x_ms.iter().cloned());
104
105        let authorization = match self.credentials.get()? {
106            Credential::SharedKey(account_key) => azure_signature::sign(
107                method,
108                &self.config.account,
109                &self.canonical_path(key),
110                content_type,
111                content_length,
112                &account_key,
113                &headers,
114            )?,
115            Credential::Bearer(token) => format!("Bearer {token}"),
116        };
117        headers.push(("Authorization".to_string(), authorization));
118        Ok(headers)
119    }
120
121    fn error_from_response(action: &str, key: &str, resp: &Response) -> StorageError {
122        StorageError::new(format!("Azure Blob {action} '{key}' failed: HTTP {} {}", resp.status(), resp.text().unwrap_or_default()))
123    }
124}
125
126fn non_empty(s: &str) -> Option<String> {
127    if s.is_empty() { None } else { Some(s.to_string()) }
128}
129
130/// Percent-encodes a blob key, preserving `/` as a path separator — mirrors
131/// `aws_sigv4::uri_encode_path`'s rules (unreserved set: `A-Z a-z 0-9 - . _ ~`).
132fn uri_encode_key(key: &str) -> String {
133    key.split('/').map(uri_encode_segment).collect::<Vec<_>>().join("/")
134}
135
136fn uri_encode_segment(segment: &str) -> String {
137    let mut out = String::with_capacity(segment.len());
138    for b in segment.bytes() {
139        match b {
140            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => out.push(b as char),
141            _ => out.push_str(&format!("%{:02X}", b)),
142        }
143    }
144    out
145}
146
147impl Storage for AzureBlobStorage {
148    fn put(&self, key: &str, data: &[u8], content_type: &str) -> Result<String, StorageError> {
149        let url = self.request_url(key);
150        let extra = [("x-ms-blob-type".to_string(), "BlockBlob".to_string())];
151        let headers = self.auth_headers("PUT", key, content_type, data.len(), &extra)?;
152
153        let mut builder = self.client.put(&url).header("Content-Type", content_type);
154        for (name, value) in headers {
155            builder = builder.header(&name, &value);
156        }
157        let resp = builder
158            .body(data.to_vec())
159            .send()
160            .map_err(|e| StorageError::new(format!("Azure Blob PUT '{key}' failed: {e}")))?;
161        if !resp.is_success() {
162            return Err(Self::error_from_response("PUT", key, &resp));
163        }
164        Ok(key.to_string())
165    }
166
167    fn get(&self, key: &str) -> Result<Vec<u8>, StorageError> {
168        let url = self.request_url(key);
169        let headers = self.auth_headers("GET", key, "", 0, &[])?;
170
171        let mut builder = self.client.get(&url);
172        for (name, value) in headers {
173            builder = builder.header(&name, &value);
174        }
175        let resp = builder.send().map_err(|e| StorageError::new(format!("Azure Blob GET '{key}' failed: {e}")))?;
176        if !resp.is_success() {
177            return Err(Self::error_from_response("GET", key, &resp));
178        }
179        Ok(resp.bytes().to_vec())
180    }
181
182    fn delete(&self, key: &str) -> Result<(), StorageError> {
183        let url = self.request_url(key);
184        let headers = self.auth_headers("DELETE", key, "", 0, &[])?;
185
186        let mut builder = self.client.delete(&url);
187        for (name, value) in headers {
188            builder = builder.header(&name, &value);
189        }
190        let resp = builder.send().map_err(|e| StorageError::new(format!("Azure Blob DELETE '{key}' failed: {e}")))?;
191        // Azure returns 202 whether or not the blob existed — matches
192        // `Storage::delete`'s no-op-on-missing contract.
193        if !resp.is_success() {
194            return Err(Self::error_from_response("DELETE", key, &resp));
195        }
196        Ok(())
197    }
198
199    fn url(&self, key: &str) -> String {
200        self.request_url(key)
201    }
202}
203
204#[cfg(test)]
205mod tests;
rws_storage/storage/azure_blob.rs

rws_storage/storage/
azure_blob.rs
