feat(azure_blob): make storage API version configurable

changelog.d/azure_blob_api_version.enhancement.md

@@ -0,0 +1,4 @@
+Allow the Azure Blob sink to select the Azure Storage service API version used for requests,
+enabling compatibility with Azure Stack and other services that lag the public Azure API.
+
+authors: opencow

src/sinks/azure_blob/config.rs

@@ -11,6 +11,7 @@ use azure_core::{
 use azure_storage_blob::{BlobContainerClient, BlobContainerClientOptions};
 
 use bytes::Bytes;
+use chrono::NaiveDate;
 use futures::FutureExt;
 use snafu::Snafu;
 use tower::ServiceBuilder;
@@ -107,6 +108,15 @@ pub struct AzureBlobSinkConfig {
     #[configurable(metadata(docs::examples = "https://mylogstorage.blob.core.windows.net/"))]
     pub(super) blob_endpoint: Option<String>,
 
+    /// The Azure Blob Storage service API version to use for requests.
+    ///
+    /// This sets the `x-ms-version` request header. If unset, the version selected by the Azure
+    /// SDK is used. Setting an older version can be useful with Azure Stack or other storage
+    /// services that do not support the SDK's default version. The configured version must support
+    /// every operation and option used by this sink.
+    #[configurable(metadata(docs::examples = "2021-08-06"))]
+    pub(super) api_version: Option<String>,
+
     /// The Azure Blob Storage Account container name.
     #[configurable(metadata(docs::examples = "my-logs"))]
     pub(super) container_name: String,
@@ -202,6 +212,7 @@ impl GenerateConfig for AzureBlobSinkConfig {
             connection_string: Some(String::from("DefaultEndpointsProtocol=https;AccountName=some-account-name;AccountKey=some-account-key;").into()),
             account_name: None,
             blob_endpoint: None,
+            api_version: None,
             container_name: String::from("logs"),
             blob_prefix: default_blob_prefix(),
             blob_time_format: Some(String::from("%s")),
@@ -268,6 +279,7 @@ impl SinkConfig for AzureBlobSinkConfig {
             self.auth.clone(),
             connection_string.clone(),
             self.container_name.clone(),
+            self.api_version.clone(),
             cx.proxy(),
             self.tls.clone(),
         )
@@ -355,6 +367,49 @@ mod tests {
         crate::test_util::test_generate_config::<AzureBlobSinkConfig>();
     }
 
+    #[test]
+    fn api_version_overrides_sdk_default() {
+        let mut options = BlobContainerClientOptions::default();
+
+        set_api_version(&mut options, Some("2021-08-06")).unwrap();
+
+        assert_eq!(options.version, "2021-08-06");
+    }
+
+    #[test]
+    fn api_version_uses_sdk_default_when_unset() {
+        let mut options = BlobContainerClientOptions::default();
+        let sdk_default = options.version.clone();
+
+        set_api_version(&mut options, None).unwrap();
+
+        assert_eq!(options.version, sdk_default);
+    }
+
+    #[test]
+    fn api_version_rejects_invalid_date() {
+        let mut options = BlobContainerClientOptions::default();
+
+        let error = set_api_version(&mut options, Some("2021-13-40")).unwrap_err();
+
+        assert_eq!(
+            error.to_string(),
+            "Invalid Azure Blob Storage API version `2021-13-40`; expected YYYY-MM-DD"
+        );
+    }
+
+    #[test]
+    fn api_version_rejects_noncanonical_date() {
+        let mut options = BlobContainerClientOptions::default();
+
+        let error = set_api_version(&mut options, Some("2021-8-6")).unwrap_err();
+
+        assert_eq!(
+            error.to_string(),
+            "Invalid Azure Blob Storage API version `2021-8-6`; expected YYYY-MM-DD"
+        );
+    }
+
     #[test]
     fn confinement_rejects_unconfined_blob_prefix() {
         let template = Template::try_from("{{ tenant }}").unwrap();
@@ -501,6 +556,7 @@ pub async fn build_client(
     auth: Option<AzureAuthentication>,
     connection_string: String,
     container_name: String,
+    api_version: Option<String>,
     proxy: &crate::config::ProxyConfig,
     tls: Option<AzureBlobTlsConfig>,
 ) -> crate::Result<Arc<BlobContainerClient>> {
@@ -517,6 +573,7 @@ pub async fn build_client(
 
     // Prepare options; attach Shared Key policy if needed
     let mut options = BlobContainerClientOptions::default();
+    set_api_version(&mut options, api_version.as_deref())?;
     match (parsed.auth(), &auth) {
         (Auth::None, None) => {
             warn!("No authentication method provided, requests will be anonymous.");
@@ -533,13 +590,8 @@ pub async fn build_client(
         ) => {
             info!("Using Shared Key authentication.");
 
-            let policy = SharedKeyAuthorizationPolicy::new(
-                account_name,
-                account_key,
-                // Use an Azurite-supported storage service version
-                String::from("2025-11-05"),
-            )
-            .map_err(|e| format!("Failed to create SharedKey policy: {e}"))?;
+            let policy = SharedKeyAuthorizationPolicy::new(account_name, account_key)
+                .map_err(|e| format!("Failed to create SharedKey policy: {e}"))?;
             options
                 .client_options
                 .per_call_policies
@@ -630,3 +682,26 @@ pub async fn build_client(
         BlobContainerClient::new(url, credential, Some(options)).map_err(|e| format!("{e}"))?;
     Ok(Arc::new(client))
 }
+
+fn set_api_version(
+    options: &mut BlobContainerClientOptions,
+    api_version: Option<&str>,
+) -> crate::Result<()> {
+    if let Some(api_version) = api_version {
+        let bytes = api_version.as_bytes();
+        let is_canonical = bytes.len() == 10
+            && bytes.iter().enumerate().all(|(index, byte)| match index {
+                4 | 7 => *byte == b'-',
+                _ => byte.is_ascii_digit(),
+            });
+
+        if !is_canonical || NaiveDate::parse_from_str(api_version, "%Y-%m-%d").is_err() {
+            return Err(format!(
+                "Invalid Azure Blob Storage API version `{api_version}`; expected YYYY-MM-DD"
+            )
+            .into());
+        }
+        options.version = api_version.to_owned();
+    }
+    Ok(())
+}

src/sinks/azure_blob/integration_tests.rs

@@ -253,6 +253,7 @@ impl AzureBlobSinkConfig {
             connection_string: Some(format!("UseDevelopmentStorage=true;DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://{address}:10000/devstoreaccount1;QueueEndpoint=http://{address}:10001/devstoreaccount1;TableEndpoint=http://{address}:10002/devstoreaccount1;").into()),
             account_name: None,
             blob_endpoint: None,
+            api_version: Some("2021-08-06".into()),
             container_name: "logs".to_string(),
             blob_prefix: Default::default(),
             blob_time_format: None,
@@ -278,6 +279,7 @@ impl AzureBlobSinkConfig {
             connection_string: Some(format!("DefaultEndpointsProtocol=https;AccountName=devstoreaccount1;BlobEndpoint=https://{address}:14430/devstoreaccount1;QueueEndpoint=https://{address}:14431/devstoreaccount1;TableEndpoint=https://{address}:14432/devstoreaccount1;").into()),
             account_name: None,
             blob_endpoint: None,
+            api_version: None,
             container_name: "logs".to_string(),
             blob_prefix: Default::default(),
             blob_time_format: None,
@@ -307,6 +309,7 @@ impl AzureBlobSinkConfig {
                 .inner()
                 .to_string(),
             self.container_name.clone(),
+            self.api_version.clone(),
             &crate::config::ProxyConfig::default(),
             self.tls.clone(),
         )

src/sinks/azure_blob/test.rs

@@ -28,6 +28,7 @@ fn default_config(encoding: EncodingConfigWithFraming) -> AzureBlobSinkConfig {
         connection_string: Default::default(),
         account_name: Default::default(),
         blob_endpoint: Default::default(),
+        api_version: Default::default(),
         container_name: Default::default(),
         blob_prefix: Default::default(),
         blob_time_format: Default::default(),

src/sinks/azure_common/shared_key_policy.rs

@@ -13,9 +13,10 @@ use openssl::{hash::MessageDigest, pkey::PKey, sign::Signer};
 
 /// Shared Key authorization policy for Azure Blob Storage requests.
 ///
-/// This policy injects the required headers (x-ms-date, x-ms-version) if missing and
-/// adds the `Authorization: SharedKey {account}:{signature}` header. The signature
-/// is computed according to the "Authorize with Shared Key" rules for the Blob service:
+/// This policy injects the required x-ms-date header and adds the
+/// `Authorization: SharedKey {account}:{signature}` header. The x-ms-version header is supplied
+/// by the Azure SDK and signed without being changed. The signature is computed according to the
+/// "Authorize with Shared Key" rules for the Blob service:
 ///
 /// StringToSign =
 ///   VERB + "\n" +
@@ -43,20 +44,14 @@ use openssl::{hash::MessageDigest, pkey::PKey, sign::Signer};
 pub struct SharedKeyAuthorizationPolicy {
     account_name: String,
     account_key: Vec<u8>, // decoded from base64
-    storage_version: String,
 }
 
 impl SharedKeyAuthorizationPolicy {
     /// Create a new shared key policy.
     ///
     /// - `account_name`: The storage account name.
     /// - `account_key_b64`: Base64-encoded storage account key.
-    /// - `storage_version`: x-ms-version value to send (e.g. "2025-11-05").
-    pub fn new(
-        account_name: String,
-        account_key_b64: String,
-        storage_version: String,
-    ) -> AzureResult<Self> {
+    pub fn new(account_name: String, account_key_b64: String) -> AzureResult<Self> {
         let account_key = base64::decode(account_key_b64.as_bytes()).map_err(|e| {
             AzureError::with_message(
                 azure_core::error::ErrorKind::Other,
@@ -66,26 +61,16 @@ impl SharedKeyAuthorizationPolicy {
         Ok(Self {
             account_name,
             account_key,
-            storage_version,
         })
     }
 
-    fn ensure_ms_headers(&self, request: &mut Request) -> AzureResult<(String, String)> {
-        // Always set x-ms-date and x-ms-version explicitly to known values for signing.
+    fn ensure_ms_date_header(&self, request: &mut Request) {
         let now = OffsetDateTime::now_utc();
         let ms_date = to_rfc7231(&now);
-        request.insert_header("x-ms-date", ms_date.clone());
-        let ms_version = self.storage_version.clone();
-        request.insert_header("x-ms-version", ms_version.clone());
-        Ok((ms_date, ms_version))
+        request.insert_header("x-ms-date", ms_date);
     }
 
-    fn build_string_to_sign(
-        &self,
-        req: &Request,
-        ms_date: &str,
-        ms_version: &str,
-    ) -> AzureResult<String> {
+    fn build_string_to_sign(&self, req: &Request) -> AzureResult<String> {
         let method = req.method().as_str();
         let url = req.url();
 
@@ -106,6 +91,15 @@ impl SharedKeyAuthorizationPolicy {
             None
         };
 
+        for required_header in ["x-ms-date", "x-ms-version"] {
+            if header(required_header).is_none() {
+                return Err(AzureError::with_message(
+                    azure_core::error::ErrorKind::Other,
+                    format!("missing required {required_header} header"),
+                ));
+            }
+        }
+
         // Content-Encoding
         if let Some(v) = header("Content-Encoding") {
             s.push_str(v);
@@ -180,14 +174,6 @@ impl SharedKeyAuthorizationPolicy {
                     .push(value.as_str().trim().to_string());
             }
         }
-        // Ensure required headers are present (they should have been inserted).
-        xms.entry("x-ms-date".to_string())
-            .or_default()
-            .push(ms_date.to_string());
-        xms.entry("x-ms-version".to_string())
-            .or_default()
-            .push(ms_version.to_string());
-
         for (k, mut vals) in xms {
             vals.sort();
             vals.dedup();
@@ -239,10 +225,11 @@ impl Policy for SharedKeyAuthorizationPolicy {
         request: &mut Request,
         next: &[Arc<dyn Policy>],
     ) -> PolicyResult {
-        // Ensure required x-ms headers are present
-        let (ms_date, ms_version) = self.ensure_ms_headers(request)?;
+        // The Azure SDK sets x-ms-version before running policies. Add x-ms-date, then sign the
+        // exact headers that will be sent.
+        self.ensure_ms_date_header(request);
         // Build string to sign
-        let sts = self.build_string_to_sign(request, &ms_date, &ms_version)?;
+        let sts = self.build_string_to_sign(request)?;
         let signature = self.sign(&sts)?;
 
         // Authorization: SharedKey {account}:{signature}
@@ -289,3 +276,52 @@ fn append_canonicalized_resource(s: &mut String, account: &str, url: &Url) -> Az
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use azure_core::http::Method;
+
+    use super::*;
+
+    fn policy() -> SharedKeyAuthorizationPolicy {
+        SharedKeyAuthorizationPolicy::new("account".into(), "a2V5".into()).unwrap()
+    }
+
+    #[test]
+    fn signs_sdk_service_version_without_overwriting_it() {
+        let policy = policy();
+        let mut request = Request::new(
+            Url::parse("https://account.blob.core.windows.net/container/blob").unwrap(),
+            Method::Put,
+        );
+        request.insert_header("x-ms-version", "2021-08-06");
+
+        policy.ensure_ms_date_header(&mut request);
+        let string_to_sign = policy.build_string_to_sign(&request).unwrap();
+
+        let service_version = request
+            .headers()
+            .iter()
+            .find(|(name, _)| name.as_str().eq_ignore_ascii_case("x-ms-version"))
+            .map(|(_, value)| value.as_str());
+        assert_eq!(service_version, Some("2021-08-06"));
+        assert!(string_to_sign.contains("x-ms-version:2021-08-06\n"));
+    }
+
+    #[test]
+    fn rejects_request_without_sdk_service_version() {
+        let policy = policy();
+        let mut request = Request::new(
+            Url::parse("https://account.blob.core.windows.net/container/blob").unwrap(),
+            Method::Put,
+        );
+        policy.ensure_ms_date_header(&mut request);
+
+        let error = policy.build_string_to_sign(&request).unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("missing required x-ms-version header")
+        );
+    }
+}

website/cue/reference/components/sinks/generated/azure_blob.cue

@@ -37,6 +37,18 @@ generated: components: sinks: azure_blob: configuration: {
 			type: bool: {}
 		}
 	}
+	api_version: {
+		description: """
+			The Azure Blob Storage service API version to use for requests.
+
+			This sets the `x-ms-version` request header. If unset, the version selected by the Azure
+			SDK is used. Setting an older version can be useful with Azure Stack or other storage
+			services that do not support the SDK's default version. The configured version must support
+			every operation and option used by this sink.
+			"""
+		required: false
+		type: string: examples: ["2021-08-06"]
+	}
 	auth: {
 		description: "Azure service principal authentication."
 		required:    false