diff --git a/changelog.d/azure_blob_api_version.enhancement.md b/changelog.d/azure_blob_api_version.enhancement.md new file mode 100644 index 0000000000000..1439c50314801 --- /dev/null +++ b/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 diff --git a/src/sinks/azure_blob/config.rs b/src/sinks/azure_blob/config.rs index 207bbb8497365..55c968227e3d9 100644 --- a/src/sinks/azure_blob/config.rs +++ b/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, + /// 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, + /// 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::(); } + #[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, connection_string: String, container_name: String, + api_version: Option, proxy: &crate::config::ProxyConfig, tls: Option, ) -> crate::Result> { @@ -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(()) +} diff --git a/src/sinks/azure_blob/integration_tests.rs b/src/sinks/azure_blob/integration_tests.rs index b3d97d72dca3b..30d3a02d88738 100644 --- a/src/sinks/azure_blob/integration_tests.rs +++ b/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(), ) diff --git a/src/sinks/azure_blob/test.rs b/src/sinks/azure_blob/test.rs index dc29923bf1c97..81ae5d75ada96 100644 --- a/src/sinks/azure_blob/test.rs +++ b/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(), diff --git a/src/sinks/azure_common/shared_key_policy.rs b/src/sinks/azure_common/shared_key_policy.rs index 38bd98aa27956..ff81c3951bb71 100644 --- a/src/sinks/azure_common/shared_key_policy.rs +++ b/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,7 +44,6 @@ use openssl::{hash::MessageDigest, pkey::PKey, sign::Signer}; pub struct SharedKeyAuthorizationPolicy { account_name: String, account_key: Vec, // decoded from base64 - storage_version: String, } impl SharedKeyAuthorizationPolicy { @@ -51,12 +51,7 @@ impl SharedKeyAuthorizationPolicy { /// /// - `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 { + pub fn new(account_name: String, account_key_b64: String) -> AzureResult { 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 { + fn build_string_to_sign(&self, req: &Request) -> AzureResult { 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], ) -> 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") + ); + } +} diff --git a/website/cue/reference/components/sinks/generated/azure_blob.cue b/website/cue/reference/components/sinks/generated/azure_blob.cue index a42c5ef4009d3..ad8e0675f25aa 100644 --- a/website/cue/reference/components/sinks/generated/azure_blob.cue +++ b/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