persist: restore Azurite test coverage via Shared Key signing

The new Azure SDK dropped its built-in Azurite emulator support, so
this commit re-implements it as a per-try `Policy` that signs each
outgoing request with the Azure Storage "Shared Key" authentication
scheme.

- New `src/persist/src/azure/azurite.rs` submodule (cfg(test)):
  implements Shared Key HMAC-SHA256 signing per
  https://learn.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key.
  Builds the canonical string-to-sign (method, fixed HTTP headers in
  spec order, sorted `x-ms-*` headers, canonicalized resource), MACs
  it with the publicly-documented Azurite dev key, and attaches an
  `Authorization: SharedKey ...` header. Uses `insert_header` for
  `x-ms-date` / `x-ms-version` as well.

- `AzureBlobConfig::new`: the `account == AZURITE_ACCOUNT` branch now
  installs `SharedKeyPolicy` via `add_shared_key_policy` in test
  builds and unconditionally errors in production builds.

- `AzureBlob::open`: in test builds, if the client URL points at
  Azurite, create the container up front so freshly-started emulators
  work.

- `AzureBlobConfig::new_for_test`: restored to the real impl that
  reads the container name from the `MZ_PERSIST_EXTERNAL_STORAGE_
  TEST_AZURE_CONTAINER` env var and constructs a config pointing at
  `http://localhost:40111`.

- `hmac` added to `[dev-dependencies]`; `sha2` is already a runtime
  dep. Everything new is gated on `#[cfg(test)]`, so production
  builds are unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jasonhernandez

Cargo.lock

@@ -78,6 +78,7 @@ workspace = true
 features = ["asm"]
 
 [dev-dependencies]
+hmac.workspace = true
 mz-ore = { path = "../ore", default-features = false, features = ["test"] }
 serde_json.workspace = true
 tempfile.workspace = true

src/persist/Cargo.toml

@@ -11,6 +11,8 @@
 
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
+#[cfg(test)]
+use std::time::Duration;
 
 use anyhow::anyhow;
 use async_trait::async_trait;
@@ -24,11 +26,17 @@ use azure_storage_blob::models::{
 };
 use azure_storage_blob::{BlobContainerClient, BlobContainerClientOptions};
 use futures_util::{StreamExt, TryStreamExt};
+#[cfg(test)]
+use tracing::info;
 use url::Url;
+#[cfg(test)]
+use uuid::Uuid;
 
 use mz_dyncfg::ConfigSet;
 use mz_ore::bytes::SegmentedBytes;
 use mz_ore::cast::CastFrom;
+#[cfg(test)]
+use mz_ore::metrics::MetricsRegistry;
 
 use crate::cfg::BlobKnobs;
 use crate::error::Error;
@@ -110,14 +118,21 @@ impl AzureBlobConfig {
         });
 
         let client = if account == AZURITE_ACCOUNT {
-            // Azurite test support is restored in a follow-up commit (it
-            // needs the new SDK's `Policy` trait to implement Shared Key
-            // signing, which the old SDK provided built-in).
-            let _ = (url, container);
-            return Err(Error::from(
-                "Azurite emulator temporarily unsupported; restored in the \
-                 Azurite shared-key follow-up commit",
-            ));
+            #[cfg(test)]
+            {
+                info!("Connecting to Azure emulator");
+                azurite::add_shared_key_policy(&mut options);
+                let container_url = azurite::container_url(&url, &container)?;
+                BlobContainerClient::from_url(container_url, None, Some(options))
+                    .map_err(|e| Error::from(format!("azure container client: {e}")))?
+            }
+            #[cfg(not(test))]
+            {
+                let _ = (url, container);
+                return Err(Error::from(
+                    "Azurite emulator is only supported in test builds",
+                ));
+            }
         } else {
             let endpoint = format!("https://{account}.blob.core.windows.net/");
             if let Some(sas) = url.query() {
@@ -149,13 +164,61 @@ impl AzureBlobConfig {
     }
 
     /// Returns a new [AzureBlobConfig] for use in unit tests.
-    ///
-    /// Stubbed for now: Azurite requires Shared Key signing, which is
-    /// restored in a follow-up commit. Returning `None` makes all
-    /// Azurite-dependent tests skip.
     #[cfg(test)]
     pub fn new_for_test() -> Result<Option<Self>, Error> {
-        Ok(None)
+        struct TestBlobKnobs;
+        impl Debug for TestBlobKnobs {
+            fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+                f.debug_struct("TestBlobKnobs").finish_non_exhaustive()
+            }
+        }
+        impl BlobKnobs for TestBlobKnobs {
+            fn operation_timeout(&self) -> Duration {
+                Duration::from_secs(30)
+            }
+
+            fn operation_attempt_timeout(&self) -> Duration {
+                Duration::from_secs(10)
+            }
+
+            fn connect_timeout(&self) -> Duration {
+                Duration::from_secs(5)
+            }
+
+            fn read_timeout(&self) -> Duration {
+                Duration::from_secs(5)
+            }
+
+            fn is_cc_active(&self) -> bool {
+                false
+            }
+        }
+
+        let container_name = match std::env::var(Self::EXTERNAL_TESTS_AZURE_CONTAINER) {
+            Ok(container) => container,
+            Err(_) => {
+                assert!(
+                    !mz_ore::env::is_var_truthy("CI"),
+                    "CI is supposed to run this test but something has gone wrong!"
+                );
+                return Ok(None);
+            }
+        };
+
+        let prefix = Uuid::new_v4().to_string();
+        let metrics = S3BlobMetrics::new(&MetricsRegistry::new());
+
+        let config = AzureBlobConfig::new(
+            AZURITE_ACCOUNT.to_string(),
+            container_name.clone(),
+            prefix,
+            metrics,
+            Url::parse(&format!("http://localhost:40111/{}", container_name)).expect("valid url"),
+            Box::new(TestBlobKnobs),
+            Arc::new(ConfigSet::default()),
+        )?;
+
+        Ok(Some(config))
     }
 }
 
@@ -190,11 +253,24 @@ impl Debug for AzureBlob {
 
 impl AzureBlob {
     /// Opens the given location for non-exclusive read-write access.
-    // `open` is intentionally `async`: the follow-up commit adds a
-    // `#[cfg(test)]` Azurite container-create call that awaits. The
-    // production path currently performs no awaits.
-    #[allow(clippy::unused_async)]
+    // `open` is async only for the `#[cfg(test)]` Azurite container-create
+    // call below; the production path performs no awaits.
+    #[cfg_attr(not(test), allow(clippy::unused_async))]
     pub async fn open(config: AzureBlobConfig) -> Result<Self, ExternalError> {
+        // In tests we may be pointed at a freshly-spun-up Azurite instance
+        // where the container hasn't been created yet.
+        #[cfg(test)]
+        {
+            if azurite::is_emulator_url(config.client.url()) {
+                if let Err(error) = config.client.create(None).await {
+                    info!(
+                        ?error,
+                        "failed to create emulator container; this is expected on repeat runs"
+                    );
+                }
+            }
+        }
+
         let ret = AzureBlob {
             metrics: config.metrics,
             client: config.client,
@@ -388,6 +464,9 @@ enum PreSizedBuffer {
     Unknown(SegmentedBytes),
 }
 
+#[cfg(test)]
+mod azurite;
+
 #[cfg(test)]
 mod tests {
     use tracing::info;

src/persist/src/azure.rs

@@ -0,0 +1,187 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Azurite-only Shared Key request signing.
+//!
+//! Azurite doesn't accept Entra ID / OAuth tokens, only the Azure Storage
+//! "Shared Key" authentication scheme. The new Azure SDK dropped its
+//! built-in emulator support, so we install a `Policy` on the test client
+//! that signs each outgoing request.
+//!
+//! Spec: <https://learn.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key>
+//!
+//! This module is `#[cfg(test)]`-gated and never compiled into production
+//! builds; the Azurite dev key is a publicly documented constant.
+
+use std::fmt;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use azure_core::base64;
+use azure_core::http::policies::{Policy, PolicyResult};
+use azure_core::http::{Context, Request};
+use hmac::{Hmac, Mac};
+use sha2::Sha256;
+use url::Url;
+
+use crate::error::Error;
+
+use super::{AZURITE_ACCOUNT, BlobContainerClientOptions};
+
+/// Default Azurite account key. Publicly documented, hard-coded.
+const AZURITE_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// Build a container URL for Azurite from the emulator URL provided by
+/// the test harness.
+///
+/// Azurite expects URLs of the form
+/// `http://{host}:{port}/devstoreaccount1/{container}`.
+pub fn container_url(url: &Url, container: &str) -> Result<Url, Error> {
+    let host = url
+        .host_str()
+        .ok_or_else(|| Error::from(format!("Azurite URL missing host: {url}")))?;
+    let port = url
+        .port()
+        .ok_or_else(|| Error::from(format!("Azurite URL missing port: {url}")))?;
+    Url::parse(&format!(
+        "http://{host}:{port}/{AZURITE_ACCOUNT}/{container}"
+    ))
+    .map_err(|e| Error::from(format!("invalid Azurite URL: {e}")))
+}
+
+pub fn is_emulator_url(url: &Url) -> bool {
+    url.path()
+        .trim_start_matches('/')
+        .starts_with(AZURITE_ACCOUNT)
+}
+
+pub fn add_shared_key_policy(options: &mut BlobContainerClientOptions) {
+    options
+        .client_options
+        .per_try_policies
+        .push(Arc::new(SharedKeyPolicy));
+}
+
+pub struct SharedKeyPolicy;
+
+impl fmt::Debug for SharedKeyPolicy {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("SharedKeyPolicy")
+    }
+}
+
+#[async_trait]
+impl Policy for SharedKeyPolicy {
+    async fn send(
+        &self,
+        ctx: &Context,
+        request: &mut Request,
+        next: &[Arc<dyn Policy>],
+    ) -> PolicyResult {
+        // `insert_header` stores the value with `'static` bounds, so we
+        // have to hand it an owned `String`.
+        let date = azure_core::time::to_rfc7231(&azure_core::time::OffsetDateTime::now_utc());
+        request.insert_header("x-ms-date", date);
+        request.insert_header("x-ms-version", "2024-08-04");
+
+        let string_to_sign = compute_signature(request);
+        let key = base64::decode(AZURITE_KEY).expect("valid base64 Azurite key");
+        let mut mac = Hmac::<Sha256>::new_from_slice(&key).expect("HMAC accepts any key");
+        mac.update(string_to_sign.as_bytes());
+        let auth = base64::encode(mac.finalize().into_bytes());
+        request.insert_header(
+            "authorization",
+            format!("SharedKey {AZURITE_ACCOUNT}:{auth}"),
+        );
+
+        next[0].send(ctx, request, &next[1..]).await
+    }
+}
+
+/// Build the canonical string-to-sign per Azure Shared Key spec.
+fn compute_signature(request: &Request) -> String {
+    let method = match request.method() {
+        azure_core::http::Method::Delete => "DELETE",
+        azure_core::http::Method::Get => "GET",
+        azure_core::http::Method::Head => "HEAD",
+        azure_core::http::Method::Patch => "PATCH",
+        azure_core::http::Method::Post => "POST",
+        azure_core::http::Method::Put => "PUT",
+        // `Method` is `#[non_exhaustive]` — unreachable today, but fall
+        // back to the method's debug string if the SDK adds a new verb.
+        other => return format!("{other:?}"),
+    };
+
+    let headers = request.headers();
+    let header = |name: &str| {
+        headers
+            .get_optional_str(&azure_core::http::headers::HeaderName::from(
+                name.to_owned(),
+            ))
+            .unwrap_or("")
+    };
+
+    // Content-Length is signed as empty when zero per spec.
+    let content_length = {
+        let v = header("content-length");
+        if v == "0" { "" } else { v }
+    };
+
+    // x-ms-* headers, lowercased, sorted, joined as `name:value\n`.
+    let mut x_ms_headers: Vec<(String, String)> = headers
+        .iter()
+        .filter_map(|(name, value)| {
+            let n = name.as_str().to_ascii_lowercase();
+            if n.starts_with("x-ms-") {
+                Some((n, value.as_str().trim().to_owned()))
+            } else {
+                None
+            }
+        })
+        .collect();
+    x_ms_headers.sort_by(|a, b| a.0.cmp(&b.0));
+    let canonicalized_headers = x_ms_headers
+        .iter()
+        .map(|(n, v)| format!("{n}:{v}"))
+        .collect::<Vec<_>>()
+        .join("\n");
+
+    // Canonicalized resource: `/account/path` + sorted query params
+    // joined as `\nname:value`.
+    let url = request.url();
+    let mut canonicalized_resource = format!("/{AZURITE_ACCOUNT}{}", url.path());
+    let mut query_pairs: Vec<(String, String)> = url
+        .query_pairs()
+        .map(|(k, v)| (k.to_ascii_lowercase(), v.into_owned()))
+        .collect();
+    query_pairs.sort_by(|a, b| a.0.cmp(&b.0));
+    for (k, v) in query_pairs {
+        canonicalized_resource.push_str(&format!("\n{k}:{v}"));
+    }
+
+    [
+        method,
+        header("content-encoding"),
+        header("content-language"),
+        content_length,
+        header("content-md5"),
+        header("content-type"),
+        // `Date` is empty because we always send `x-ms-date` instead.
+        "",
+        header("if-modified-since"),
+        header("if-match"),
+        header("if-none-match"),
+        header("if-unmodified-since"),
+        header("range"),
+        &canonicalized_headers,
+        &canonicalized_resource,
+    ]
+    .join("\n")
+}