From 894c16223bb42e15dc78d9960ac4fd004ffa9940 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 08:52:12 -0700 Subject: [PATCH 1/6] chore(storage): point Azure dependencies at the rewritten SDK Move the workspace off the legacy Azure SDK and onto the 1.0 line: azure_core 1.1, azure_identity 1.0 and azure_storage_blob 1.0. `azure_storage` is dropped outright. The rewritten SDK has no successor for `StorageCredentials`, `CloudLocation` or `ConnectionString`, so the concepts it provided have to be rebuilt on top of the pipeline instead of renamed. Feature names changed with the rewrite: `enable_reqwest_rustls` is now `reqwest_rustls`, and the `azurite_workaround` features no longer exist, so they leave `integration-testsuite`. `hmac_rust` survives in `azure_core` 1.1, which matters because a shared key signing policy needs it. The new SDK resolves to a smaller graph: `Cargo.lock` loses 296 lines net. This commit only moves the dependencies. `quickwit-storage` does not build against them yet. --- quickwit/Cargo.lock | 481 ++++++--------------------- quickwit/Cargo.toml | 14 +- quickwit/quickwit-storage/Cargo.toml | 15 +- 3 files changed, 107 insertions(+), 403 deletions(-) diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index cd19a875db9..30db5cc6265 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -2,12 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "RustyXML" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b5ace29ee3216de37c0546865ad08edef58b0f9e76838ed8959a84a990e58c5" - [[package]] name = "addr2line" version = "0.25.1" @@ -553,24 +547,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "async-io" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" -dependencies = [ - "autocfg", - "cfg-if", - "concurrent-queue", - "futures-io", - "futures-lite 2.6.1", - "parking", - "polling", - "rustix 1.1.4", - "slab", - "windows-sys 0.61.2", -] - [[package]] name = "async-lock" version = "3.4.2" @@ -582,24 +558,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "async-process" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" -dependencies = [ - "async-channel 2.5.0", - "async-io", - "async-lock", - "async-signal", - "async-task", - "blocking", - "cfg-if", - "event-listener 5.4.1", - "futures-lite 2.6.1", - "rustix 1.1.4", -] - [[package]] name = "async-recursion" version = "1.1.1" @@ -611,24 +569,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "async-signal" -version = "0.2.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" -dependencies = [ - "async-io", - "async-lock", - "atomic-waker", - "cfg-if", - "futures-core", - "futures-io", - "rustix 1.1.4", - "signal-hook-registry", - "slab", - "windows-sys 0.61.2", -] - [[package]] name = "async-speed-limit" version = "0.4.2" @@ -663,12 +603,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "async-task" -version = "4.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" - [[package]] name = "async-trait" version = "0.1.89" @@ -727,7 +661,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "hex", "http 1.4.2", "sha1 0.10.6", @@ -790,7 +724,7 @@ dependencies = [ "aws-types", "bytes", "bytes-utils", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "http-body 0.4.6", @@ -821,7 +755,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -848,7 +782,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -878,7 +812,7 @@ dependencies = [ "aws-smithy-xml", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "hex", "hmac 0.13.0", "http 0.2.12", @@ -911,7 +845,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -937,7 +871,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -963,7 +897,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -990,7 +924,7 @@ dependencies = [ "aws-smithy-types", "aws-smithy-xml", "aws-types", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1200,7 +1134,7 @@ dependencies = [ "aws-smithy-schema", "aws-smithy-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "http-body 0.4.6", @@ -1351,105 +1285,70 @@ dependencies = [ [[package]] name = "azure_core" -version = "0.21.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b552ad43a45a746461ec3d3a51dfb6466b4759209414b439c165eb6a6b7729e" +checksum = "4e41cbd819986ba41904c207d8ffc4106f8f8352a548d773e9554906379bb2fb" dependencies = [ + "async-lock", "async-trait", - "base64 0.22.1", + "azure_core_macros", "bytes", - "dyn-clone", "futures", - "getrandom 0.2.17", "hmac 0.12.1", - "http-types", - "once_cell", - "paste", "pin-project", - "quick-xml 0.31.0", - "rand 0.8.6", - "reqwest 0.12.28", "rustc_version", "serde", "serde_json", "sha2 0.10.9", - "time", + "tokio", "tracing", - "url", - "uuid", + "typespec", + "typespec_client_core", ] [[package]] -name = "azure_identity" -version = "0.21.0" +name = "azure_core_macros" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ddd80344317c40c04b603807b63a5cefa532f1b43522e72f480a988141f744" +checksum = "b9b52dba6a345f3ad2d42ff8d0d63df9d0994cfa29657bf18ffdbf149f78a4f5" dependencies = [ - "async-lock", - "async-process", - "async-trait", - "azure_core", - "futures", - "oauth2 4.4.2", - "pin-project", - "serde", - "time", + "proc-macro2", + "quote", + "syn 2.0.118", "tracing", - "url", - "uuid", ] [[package]] -name = "azure_storage" -version = "0.21.0" +name = "azure_identity" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59f838159f4d29cb400a14d9d757578ba495ae64feb07a7516bf9e4415127126" +checksum = "32edf96b356ca7c51d7590c4925cc36efc3947a5da4468e8e0b25c56ecbb3de5" dependencies = [ - "RustyXML", "async-lock", "async-trait", "azure_core", - "bytes", - "serde", - "serde_derive", - "time", - "tracing", - "url", - "uuid", -] - -[[package]] -name = "azure_storage_blobs" -version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97e83c3636ae86d9a6a7962b2112e3b19eb3903915c50ce06ff54ff0a2e6a7e4" -dependencies = [ - "RustyXML", - "azure_core", - "azure_storage", - "azure_svc_blobstorage", - "bytes", "futures", + "pin-project", "serde", - "serde_derive", "serde_json", "time", "tracing", "url", - "uuid", ] [[package]] -name = "azure_svc_blobstorage" -version = "0.21.0" +name = "azure_storage_blob" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e6c6f20c5611b885ba94c7bae5e02849a267381aecb8aee577e8c35ff4064c6" +checksum = "1756febbcca86c862ef718b983b505d08bd65a9bc984a915b0a16af4a4c3fe5b" dependencies = [ + "async-stream", + "async-trait", "azure_core", "bytes", "futures", - "log", - "once_cell", + "percent-encoding", + "pin-project", "serde", "serde_json", "time", @@ -1488,12 +1387,6 @@ version = "2.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd637ac531c60eb7fbc4684dc061c2d7d90d73d758181aa02eeff0464b9eee4b" -[[package]] -name = "base64" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" - [[package]] name = "base64" version = "0.21.7" @@ -1677,19 +1570,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "blocking" -version = "1.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" -dependencies = [ - "async-channel 2.5.0", - "async-task", - "futures-io", - "futures-lite 2.6.1", - "piper", -] - [[package]] name = "bon" version = "3.9.3" @@ -4119,15 +3999,6 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9afc2bd4d5a73106dd53d10d73d3401c2f32730ba2c0b93ddb888a8983680471" -[[package]] -name = "fastrand" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" -dependencies = [ - "instant", -] - [[package]] name = "fastrand" version = "2.4.1" @@ -4235,7 +4106,7 @@ version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ - "fastrand 2.4.1", + "fastrand", "futures-core", "futures-sink", "spin 0.9.8", @@ -4387,34 +4258,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" -[[package]] -name = "futures-lite" -version = "1.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce" -dependencies = [ - "fastrand 1.9.0", - "futures-core", - "futures-io", - "memchr", - "parking", - "pin-project-lite", - "waker-fn", -] - -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand 2.4.1", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "futures-macro" version = "0.3.32" @@ -4487,17 +4330,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "getrandom" -version = "0.1.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fc3cb4d91f53b50155bdcfd23f6a4c39ae1969c2ae85982b135750cccaf5fce" -dependencies = [ - "cfg-if", - "libc", - "wasi 0.9.0+wasi-snapshot-preview1", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -4507,7 +4339,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "wasm-bindgen", ] @@ -4972,26 +4804,6 @@ dependencies = [ "serde", ] -[[package]] -name = "http-types" -version = "2.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e9b187a72d63adbfba487f48095306ac823049cb504ee195541e91c7775f5ad" -dependencies = [ - "anyhow", - "async-channel 1.9.0", - "base64 0.13.1", - "futures-lite 1.13.0", - "infer", - "pin-project-lite", - "rand 0.7.3", - "serde", - "serde_json", - "serde_qs 0.8.5", - "serde_urlencoded", - "url", -] - [[package]] name = "httparse" version = "1.10.1" @@ -5317,12 +5129,6 @@ dependencies = [ "rustversion", ] -[[package]] -name = "infer" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64e9829a50b42bb782c1df523f78d332fe371b10c661e78b7a3c34b0198e9fac" - [[package]] name = "inferno" version = "0.11.21" @@ -5375,15 +5181,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "instant" -version = "0.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" -dependencies = [ - "cfg-if", -] - [[package]] name = "integer-encoding" version = "3.0.4" @@ -6256,7 +6053,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "windows-sys 0.61.2", ] @@ -6586,25 +6383,6 @@ dependencies = [ "itoa", ] -[[package]] -name = "oauth2" -version = "4.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c38841cdd844847e3e7c8d29cef9dcfed8877f8f56f9071f77843ecf3baf937f" -dependencies = [ - "base64 0.13.1", - "chrono", - "getrandom 0.2.17", - "http 0.2.12", - "rand 0.8.6", - "serde", - "serde_json", - "serde_path_to_error", - "sha2 0.10.9", - "thiserror 1.0.69", - "url", -] - [[package]] name = "oauth2" version = "5.0.0" @@ -6843,7 +6621,7 @@ dependencies = [ "http 1.4.2", "itertools 0.10.5", "log", - "oauth2 5.0.0", + "oauth2", "p256", "p384", "rand 0.8.6", @@ -7536,17 +7314,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "piper" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" -dependencies = [ - "atomic-waker", - "fastrand 2.4.1", - "futures-io", -] - [[package]] name = "pkcs1" version = "0.7.5" @@ -7716,20 +7483,6 @@ dependencies = [ "pnet_sys", ] -[[package]] -name = "polling" -version = "3.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" -dependencies = [ - "cfg-if", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix 1.1.4", - "windows-sys 0.61.2", -] - [[package]] name = "poly1305" version = "0.8.0" @@ -8234,7 +7987,7 @@ dependencies = [ "lz4", "murmur3", "nom 7.1.3", - "oauth2 5.0.0", + "oauth2", "openidconnect", "pem", "prost 0.13.5", @@ -8265,7 +8018,7 @@ dependencies = [ "libc", "once_cell", "raw-cpuid", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "web-sys", "winapi", ] @@ -8285,16 +8038,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "quick-xml" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1004a344b30a54e2ee58d66a71b32d2db2feb0a31f9a2d302bf0536f15de2a33" -dependencies = [ - "memchr", - "serde", -] - [[package]] name = "quick-xml" version = "0.41.0" @@ -9369,7 +9112,7 @@ dependencies = [ "rust-embed", "serde", "serde_json", - "serde_qs 1.1.2", + "serde_qs", "serde_with", "tempfile", "thiserror 2.0.18", @@ -9405,8 +9148,7 @@ dependencies = [ "aws-smithy-types", "azure_core", "azure_identity", - "azure_storage", - "azure_storage_blobs", + "azure_storage_blob", "base64 0.22.1", "bytes", "bytesize", @@ -9584,19 +9326,6 @@ dependencies = [ "nibble_vec", ] -[[package]] -name = "rand" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03" -dependencies = [ - "getrandom 0.1.16", - "libc", - "rand_chacha 0.2.2", - "rand_core 0.5.1", - "rand_hc", -] - [[package]] name = "rand" version = "0.8.6" @@ -9629,16 +9358,6 @@ dependencies = [ "rand_core 0.10.1", ] -[[package]] -name = "rand_chacha" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402" -dependencies = [ - "ppv-lite86", - "rand_core 0.5.1", -] - [[package]] name = "rand_chacha" version = "0.3.1" @@ -9659,15 +9378,6 @@ dependencies = [ "rand_core 0.9.5", ] -[[package]] -name = "rand_core" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19" -dependencies = [ - "getrandom 0.1.16", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -9702,15 +9412,6 @@ dependencies = [ "rand 0.10.2", ] -[[package]] -name = "rand_hc" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c" -dependencies = [ - "rand_core 0.5.1", -] - [[package]] name = "rand_pcg" version = "0.10.2" @@ -10048,7 +9749,6 @@ dependencies = [ "bytes", "encoding_rs", "futures-core", - "futures-util", "h2 0.4.15", "http 1.4.2", "http-body 1.0.1", @@ -10071,14 +9771,12 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls 0.26.4", - "tokio-util", "tower 0.5.3", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams 0.4.2", "web-sys", "webpki-roots 1.0.8", ] @@ -10117,7 +9815,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams 0.5.0", + "wasm-streams", "web-sys", ] @@ -10801,17 +10499,6 @@ dependencies = [ "serde", ] -[[package]] -name = "serde_qs" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7715380eec75f029a4ef7de39a9200e0a63823176b759d055b613f5a87df6a6" -dependencies = [ - "percent-encoding", - "serde", - "thiserror 1.0.69", -] - [[package]] name = "serde_qs" version = "1.1.2" @@ -11922,7 +11609,7 @@ version = "3.27.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ - "fastrand 2.4.1", + "fastrand", "getrandom 0.4.3", "once_cell", "rustix 1.1.4", @@ -12067,7 +11754,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" dependencies = [ "deranged", - "js-sys", "num-conv", "powerfmt", "serde_core", @@ -12650,6 +12336,58 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "typespec" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "753a2fe021e407d4fc9ee6f4f0a33403cc306d5c54c4e4ebe1b8cbde0ca052b9" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures", + "quick-xml 0.41.0", + "serde", + "serde_json", + "url", +] + +[[package]] +name = "typespec_client_core" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0373af0f9d4f580b3a1a9d9639cedaabe015ed262b35bfbe13941bfb14fe1ea6" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "dyn-clone", + "futures", + "pin-project", + "rand 0.10.2", + "reqwest 0.13.4", + "serde", + "serde_json", + "time", + "tokio", + "tracing", + "typespec", + "typespec_macros", + "url", + "uuid", +] + +[[package]] +name = "typespec_macros" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c608f4427943f8adb211abc95c87672b1b98847152783507d54e3246e502f60" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "syn 2.0.118", +] + [[package]] name = "typetag" version = "0.2.22" @@ -13126,12 +12864,6 @@ dependencies = [ "libc", ] -[[package]] -name = "waker-fn" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" - [[package]] name = "walkdir" version = "2.5.0" @@ -13181,12 +12913,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "wasi" -version = "0.9.0+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519" - [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -13214,7 +12940,7 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae86f02046da16a333a9129d31451423e1657737ecdafed4193838a5f54c5cfe" dependencies = [ - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", ] [[package]] @@ -13272,19 +12998,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-streams" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" -dependencies = [ - "futures-util", - "js-sys", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "wasm-streams" version = "0.5.0" diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index 8904ed1dd3c..5a96d3c1fdf 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -375,14 +375,14 @@ aws-smithy-types = { version = "1.6", features = [ ] } aws-types = "1.4" -azure_core = { version = "0.21", features = ["hmac_rust", "enable_reqwest_rustls"] } -azure_identity = { version = "0.21", default-features = false, features = ["enable_reqwest_rustls"] } -azure_storage = { version = "0.21", default-features = false, features = [ - "enable_reqwest_rustls", -] } -azure_storage_blobs = { version = "0.21", default-features = false, features = [ - "enable_reqwest_rustls", +azure_core = { version = "1", default-features = false, features = [ + "hmac_rust", + "reqwest_rustls", + "tokio", + "xml", ] } +azure_identity = { version = "1", default-features = false } +azure_storage_blob = { version = "1", default-features = false } opendal = { version = "0.58", default-features = false, features = ["http-transport-reqwest"] } reqsign = { version = "0.18", default-features = false, features = ["google", "default-context"] } diff --git a/quickwit/quickwit-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index cda13362a79..d0e3353cc1f 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -46,8 +46,7 @@ aws-smithy-types = { workspace = true } azure_core = { workspace = true, optional = true } azure_identity = { workspace = true, optional = true } -azure_storage = { workspace = true, optional = true } -azure_storage_blobs = { workspace = true, optional = true } +azure_storage_blob = { workspace = true, optional = true } quickwit-aws = { workspace = true } quickwit-common = { workspace = true } @@ -83,21 +82,13 @@ quickwit-common = { workspace = true, features = ["testsuite"] } azure = [ "azure_core", "azure_identity", - "azure_identity/enable_reqwest_rustls", - "azure_storage", - "azure_storage_blobs", - "azure_core/hmac_rust", - "azure_core/enable_reqwest_rustls", - "azure_storage/enable_reqwest_rustls", - "azure_storage_blobs/enable_reqwest_rustls", + "azure_storage_blob", ] gcs = ["dep:opendal", "opendal/services-gcs"] ci-test = [] integration-testsuite = [ "azure", - "azure_core/azurite_workaround", - "azure_storage_blobs/azurite_workaround", - "gcs", # Stands for Google cloud storage. + "gcs", # Stands for Google cloud storage. "dep:reqwest", ] testsuite = ["mockall"] From 3498d5b263f9604b8dfbfa9b5e309896badb22cf Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 09:14:05 -0700 Subject: [PATCH 2/6] feat(storage): add Azure shared key signing and explicit credential selection Two capabilities the rewritten Azure SDK no longer provides, added ahead of porting the blob storage backend itself. `azure_shared_key` signs requests with the storage account key. The 1.0 SDK authenticates with Entra ID tokens only, and the SDK team has said shared key support will not return (Azure/azure-sdk-for-rust#2975). Quickwit documents `azure.access_key` as a supported credential, and Azurite accepts shared key only, so the signing `azure_storage` 0.21 used to provide lives here now. The policy runs per retry, because the service rejects an `x-ms-date` that has drifted more than fifteen minutes and a retried request would otherwise carry a stale timestamp. `azure_credentials` replaces `azure_identity::create_credential()`, which no longer exists: the 1.0 line removed `DefaultAzureCredential` along with it, and the remaining `DeveloperToolsCredential` chains the two CLIs only. Workload identity is chosen when all three variables the webhook injects are present, managed identity otherwise, and `AZURE_CREDENTIAL_KIND` still pins the choice explicitly. The container client is built here too, since 1.0 clients take a container URL rather than an account name plus a cloud location, which removes the special case a sovereign endpoint used to need. Both modules compile and carry unit tests. The backend in `azure_blob_storage.rs` is not ported yet, so the crate still does not build. --- quickwit/Cargo.lock | 1 + quickwit/quickwit-storage/Cargo.toml | 2 + .../src/object_storage/azure_credentials.rs | 216 ++++++++++++ .../src/object_storage/azure_shared_key.rs | 327 ++++++++++++++++++ .../src/object_storage/mod.rs | 4 + 5 files changed, 550 insertions(+) create mode 100644 quickwit/quickwit-storage/src/object_storage/azure_credentials.rs create mode 100644 quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 30db5cc6265..c31e9f2418d 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -9180,6 +9180,7 @@ dependencies = [ "tantivy", "tempfile", "thiserror 2.0.18", + "time", "tokio", "tokio-rustls 0.26.4", "tokio-stream", diff --git a/quickwit/quickwit-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index d0e3353cc1f..d056e9ae8d5 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -47,6 +47,7 @@ aws-smithy-types = { workspace = true } azure_core = { workspace = true, optional = true } azure_identity = { workspace = true, optional = true } azure_storage_blob = { workspace = true, optional = true } +time = { workspace = true, optional = true } quickwit-aws = { workspace = true } quickwit-common = { workspace = true } @@ -83,6 +84,7 @@ azure = [ "azure_core", "azure_identity", "azure_storage_blob", + "dep:time", ] gcs = ["dep:opendal", "opendal/services-gcs"] ci-test = [] diff --git a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs new file mode 100644 index 00000000000..8802b699503 --- /dev/null +++ b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs @@ -0,0 +1,216 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Chooses how Quickwit authenticates to Azure Blob Storage, and builds the container client. +//! +//! `azure_identity` 1.0 removed `DefaultAzureCredential` and `create_credential()`, so the +//! choice the SDK used to make by inspecting the environment is made here instead. The +//! nearest remaining type, `DeveloperToolsCredential`, chains the Azure CLI and the +//! Developer CLI only, which does not cover a pod. + +use std::env; +use std::sync::Arc; + +use azure_core::credentials::TokenCredential; +use azure_core::http::policies::Policy; +use azure_core::http::{ClientOptions, Url}; +use azure_identity::{ManagedIdentityCredential, WorkloadIdentityCredential}; +use azure_storage_blob::{BlobContainerClient, BlobContainerClientOptions}; +use quickwit_config::AzureStorageConfig; +use tracing::info; + +use crate::StorageResolverError; +use crate::object_storage::azure_shared_key::SharedKeyAuthorizationPolicy; + +/// Environment variables the workload identity webhook injects into a pod. +const AZURE_CLIENT_ID: &str = "AZURE_CLIENT_ID"; +const AZURE_TENANT_ID: &str = "AZURE_TENANT_ID"; +const AZURE_FEDERATED_TOKEN_FILE: &str = "AZURE_FEDERATED_TOKEN_FILE"; +/// Lets an operator pin the provider instead of relying on the detection below. +const AZURE_CREDENTIAL_KIND: &str = "AZURE_CREDENTIAL_KIND"; + +/// How requests to the blob service are authorized. +pub(crate) enum AzureCredential { + /// Storage account key, signed by Quickwit because the 1.0 SDK dropped shared key. + SharedKey(String), + /// Entra ID token, signed by the SDK's bearer policy. + Token(Arc), +} + +/// Resolves the credential from the storage config and the environment. +/// +/// An explicit `access_key` wins, matching the previous behaviour: it is the only credential +/// an operator states in the config file, so treating it as a preference is the least +/// surprising reading. +pub(crate) fn resolve_credential( + azure_storage_config: &AzureStorageConfig, +) -> Result { + if let Some(access_key) = azure_storage_config.resolve_access_key() { + return Ok(AzureCredential::SharedKey(access_key)); + } + let token_credential = resolve_token_credential().map_err(|error| { + StorageResolverError::InvalidConfig(format!( + "could not build an Azure token credential: {error}. Set an access key, or run with \ + workload identity or managed identity configured" + )) + })?; + Ok(AzureCredential::Token(token_credential)) +} + +/// Builds the token credential the environment describes. +fn resolve_token_credential() -> azure_core::Result> { + let credential_kind = env::var(AZURE_CREDENTIAL_KIND) + .map(|kind| kind.trim().to_lowercase()) + .unwrap_or_default(); + + match credential_kind.as_str() { + "workloadidentity" => Ok(WorkloadIdentityCredential::new(None)?), + "managedidentity" => Ok(ManagedIdentityCredential::new(None)?), + // An empty or unrecognized value falls through to detection. The workload identity + // variables are injected by a mutating webhook rather than by an operator, so their + // presence is the signal that the pod runs under workload identity. + _ => { + if workload_identity_env_is_complete() { + info!("using azure workload identity credential"); + return Ok(WorkloadIdentityCredential::new(None)?); + } + info!("using azure managed identity credential"); + Ok(ManagedIdentityCredential::new(None)?) + } + } +} + +/// Returns `true` when all three variables a `WorkloadIdentityCredential` needs are present. +/// +/// A partial set means the webhook did not inject a usable identity, and building the +/// credential would fail at the first request rather than here. +fn workload_identity_env_is_complete() -> bool { + [ + AZURE_CLIENT_ID, + AZURE_TENANT_ID, + AZURE_FEDERATED_TOKEN_FILE, + ] + .iter() + .all(|variable| match env::var(variable) { + Ok(value) => !value.trim().is_empty(), + Err(_) => false, + }) +} + +/// Builds the container client. +/// +/// The 1.0 clients take the container URL directly rather than an account name plus a cloud +/// location, so a custom sovereign endpoint needs no special case: it is the base of the URL. +pub(crate) fn build_container_client( + storage_account_name: &str, + credential: AzureCredential, + blob_service_uri: Option, + container_name: &str, +) -> Result { + let blob_service_uri = match blob_service_uri { + Some(uri) => { + info!( + endpoint = %uri, + "using Azure blob storage endpoint defined in storage config or environment \ + variable" + ); + uri + } + None => format!("https://{storage_account_name}.blob.core.windows.net"), + }; + let container_url = join_container(&blob_service_uri, container_name)?; + + let mut client_options = BlobContainerClientOptions::default(); + let token_credential = match credential { + AzureCredential::SharedKey(access_key) => { + // Signing happens per retry so that every attempt carries a fresh `x-ms-date`. + let policy: Arc = Arc::new(SharedKeyAuthorizationPolicy::new( + storage_account_name.to_owned(), + access_key, + )); + client_options.client_options = ClientOptions { + per_try_policies: vec![policy], + ..Default::default() + }; + None + } + AzureCredential::Token(token_credential) => Some(token_credential), + }; + + BlobContainerClient::new(container_url, token_credential, Some(client_options)).map_err( + |error| { + StorageResolverError::InvalidConfig(format!( + "could not build an Azure container client: {error}" + )) + }, + ) +} + +/// Appends the container to the blob service URL. +fn join_container( + blob_service_uri: &str, + container_name: &str, +) -> Result { + let mut container_url = Url::parse(blob_service_uri.trim_end_matches('/')).map_err(|error| { + StorageResolverError::InvalidConfig(format!( + "`{blob_service_uri}` is not a valid Azure blob service URL: {error}" + )) + })?; + container_url + .path_segments_mut() + .map_err(|_| { + StorageResolverError::InvalidConfig(format!( + "`{blob_service_uri}` cannot be a base URL" + )) + })? + .pop_if_empty() + .push(container_name); + Ok(container_url) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_join_container_appends_the_container() { + let url = join_container("https://acct.blob.core.windows.net", "my-container").unwrap(); + assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/my-container"); + } + + #[test] + fn test_join_container_tolerates_a_trailing_slash() { + let url = join_container("https://acct.blob.core.windows.net/", "my-container").unwrap(); + assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/my-container"); + } + + #[test] + fn test_join_container_keeps_a_sovereign_host() { + let url = join_container("https://acct.blob.core.usgovcloudapi.net", "c").unwrap(); + assert_eq!(url.as_str(), "https://acct.blob.core.usgovcloudapi.net/c"); + } + + #[test] + fn test_join_container_keeps_an_emulator_path_prefix() { + // Azurite addresses accounts by path rather than by subdomain, so the account + // segment has to survive. + let url = join_container("http://127.0.0.1:10000/devstoreaccount1", "c").unwrap(); + assert_eq!(url.as_str(), "http://127.0.0.1:10000/devstoreaccount1/c"); + } + + #[test] + fn test_join_container_rejects_a_non_url() { + assert!(join_container("not a url", "c").is_err()); + } +} diff --git a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs new file mode 100644 index 00000000000..24b049f1bd1 --- /dev/null +++ b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs @@ -0,0 +1,327 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared key (storage account key) request signing. +//! +//! The rewritten Azure SDK authenticates with Entra ID tokens only. Shared key support was +//! dropped in the 1.0 line and the SDK team has stated it will not come back, see +//! . Quickwit documents +//! `azure.access_key` as a supported credential, and the Azurite emulator that backs the +//! `integration-testsuite` feature accepts shared key only, so the signing that +//! `azure_storage` 0.21 used to provide is implemented here instead. +//! +//! The scheme is specified in +//! . + +use std::borrow::Cow; +use std::fmt; +use std::sync::Arc; + +use azure_core::credentials::Secret; +use azure_core::hmac::hmac_sha256; +use azure_core::http::headers::Headers; +use azure_core::http::policies::{Policy, PolicyResult}; +use azure_core::http::{Context, Method, Request, Url}; +use azure_core::time::to_rfc7231; +use time::OffsetDateTime; + +/// Header carrying the request time. Signing uses this rather than `Date` so that the +/// `Date` slot of the string to sign stays empty, which is what the service expects when +/// `x-ms-date` is present. +const X_MS_DATE: &str = "x-ms-date"; + +/// Signs every request with the storage account key. +/// +/// Installed as a per-retry policy so that each attempt is signed with a fresh timestamp: +/// the service rejects requests whose `x-ms-date` drifts more than 15 minutes from its own +/// clock, and a retry of a long-stalled request would otherwise carry a stale one. +pub(crate) struct SharedKeyAuthorizationPolicy { + account: String, + key: Secret, +} + +impl fmt::Debug for SharedKeyAuthorizationPolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + // The key is deliberately absent: this type ends up in pipeline debug output. + formatter + .debug_struct("SharedKeyAuthorizationPolicy") + .field("account", &self.account) + .finish_non_exhaustive() + } +} + +impl SharedKeyAuthorizationPolicy { + pub(crate) fn new(account: String, key: String) -> Self { + Self { + account, + key: Secret::new(key), + } + } +} + +#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))] +#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)] +impl Policy for SharedKeyAuthorizationPolicy { + async fn send( + &self, + ctx: &Context, + request: &mut Request, + next: &[Arc], + ) -> PolicyResult { + let now = OffsetDateTime::now_utc(); + request.insert_header(X_MS_DATE, to_rfc7231(&now)); + + let method = request.method(); + let string_to_sign = + string_to_sign(&self.account, &method, request.url(), request.headers()); + let signature = hmac_sha256(&string_to_sign, &self.key)?; + request.insert_header( + "authorization", + format!("SharedKey {}:{}", self.account, signature), + ); + + next[0].send(ctx, request, &next[1..]).await + } +} + +/// Returns the value of `header_name`, or the empty string when absent. +/// +/// Absent and empty are indistinguishable in the string to sign, which is why the service +/// tolerates the collapse. +fn header_or_empty(headers: &Headers, header_name: &str) -> String { + headers + .get_optional_str(&azure_core::http::headers::HeaderName::from( + header_name.to_owned(), + )) + .unwrap_or_default() + .to_owned() +} + +/// Builds the string the signature is computed over. +/// +/// The field order is fixed by the service and every slot is present even when empty, so +/// the newlines carry meaning. `Content-Length` is the one exception the specification +/// calls out: it must be empty rather than `0` for requests without a body, for API +/// versions from 2015-02-21 onwards. +fn string_to_sign(account: &str, method: &Method, url: &Url, headers: &Headers) -> String { + let content_length = match header_or_empty(headers, "content-length") { + length if length == "0" => String::new(), + length => length, + }; + format!( + "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}", + method.as_ref(), + header_or_empty(headers, "content-encoding"), + header_or_empty(headers, "content-language"), + content_length, + header_or_empty(headers, "content-md5"), + header_or_empty(headers, "content-type"), + // Empty on purpose: `x-ms-date` supersedes `Date` and appears in the canonicalized + // headers below. Signing both would double-count the timestamp. + "", + header_or_empty(headers, "if-modified-since"), + header_or_empty(headers, "if-match"), + header_or_empty(headers, "if-none-match"), + header_or_empty(headers, "if-unmodified-since"), + header_or_empty(headers, "range"), + canonicalized_headers(headers), + canonicalized_resource(account, url), + ) +} + +/// Renders the `x-ms-*` headers in the form the signature expects: lowercase names, sorted +/// lexicographically, one `name:value` pair per line, each line newline terminated. +fn canonicalized_headers(headers: &Headers) -> String { + let mut ms_headers: Vec<(String, String)> = Vec::new(); + headers.iter().for_each(|(header_name, header_value)| { + let name = header_name.as_str().to_lowercase(); + if name.starts_with("x-ms-") { + // Linear whitespace inside a value has to fold to a single space, otherwise the + // service and the client sign different bytes. + let value = header_value.as_str().split_whitespace().collect::>(); + ms_headers.push((name, value.join(" "))); + } + }); + ms_headers.sort_unstable(); + + let mut rendered = String::new(); + for (name, value) in ms_headers { + rendered.push_str(&name); + rendered.push(':'); + rendered.push_str(&value); + rendered.push('\n'); + } + rendered +} + +/// Renders the resource the request addresses: the account, the decoded path, then every +/// query parameter, lowercased and sorted, with repeated parameters comma joined. +fn canonicalized_resource(account: &str, url: &Url) -> String { + let mut resource = String::with_capacity(url.as_str().len()); + resource.push('/'); + resource.push_str(account); + for segment in url.path_segments().into_iter().flatten() { + resource.push('/'); + resource.push_str(segment); + } + + let mut query_pairs: Vec<(Cow<'_, str>, Cow<'_, str>)> = url.query_pairs().collect(); + if query_pairs.is_empty() { + return resource; + } + // Sort by lowercased name so that repeated parameters group together, and by value so + // that the comma joined list below is itself sorted, as the service requires. + query_pairs.sort_by(|(left_name, left_value), (right_name, right_value)| { + let left = left_name.to_lowercase(); + let right = right_name.to_lowercase(); + left.cmp(&right).then_with(|| left_value.cmp(right_value)) + }); + + let mut current_name: Option = None; + for (name, value) in query_pairs { + let name = name.to_lowercase(); + match current_name.as_deref() { + Some(previous) if previous == name => { + resource.push(','); + } + _ => { + resource.push('\n'); + resource.push_str(&name); + resource.push(':'); + current_name = Some(name); + } + } + resource.push_str(&value); + } + resource +} + +#[cfg(test)] +mod tests { + use azure_core::http::headers::Headers; + use azure_core::http::{Method, Url}; + + use super::*; + + /// The account and key Azurite and the legacy storage emulator publish. Documented at + /// , so this is + /// not a secret. + const EMULATOR_ACCOUNT: &str = "devstoreaccount1"; + const EMULATOR_ACCOUNT_KEY: &str = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVEr\ + Cz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + + fn headers_from(pairs: &[(&str, &str)]) -> Headers { + let mut headers = Headers::new(); + for (name, value) in pairs { + headers.insert( + azure_core::http::headers::HeaderName::from((*name).to_owned()), + (*value).to_owned(), + ); + } + headers + } + + #[test] + fn test_canonicalized_resource_without_query() { + let url = Url::parse("https://acct.blob.core.windows.net/container/a/b.split").unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/container/a/b.split" + ); + } + + #[test] + fn test_canonicalized_resource_sorts_and_lowercases_query() { + let url = Url::parse( + "https://acct.blob.core.windows.net/container?restype=container&comp=list&Marker=m", + ) + .unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/container\ncomp:list\nmarker:m\nrestype:container" + ); + } + + #[test] + fn test_canonicalized_resource_comma_joins_repeated_query_params() { + let url = + Url::parse("https://acct.blob.core.windows.net/c?include=metadata&include=copy").unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/c\ninclude:copy,metadata" + ); + } + + #[test] + fn test_canonicalized_headers_sorted_and_folded() { + let headers = headers_from(&[ + ("x-ms-version", "2025-05-05"), + ("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT"), + ("content-type", "application/octet-stream"), + ("x-ms-blob-type", "Block Blob"), + ]); + assert_eq!( + canonicalized_headers(&headers), + "x-ms-blob-type:Block Blob\nx-ms-date:Fri, 14 Aug 2026 12:00:00 GMT\nx-ms-version:2025-05-05\n" + ); + } + + #[test] + fn test_string_to_sign_leaves_date_slot_empty() { + let url = Url::parse("https://acct.blob.core.windows.net/c/blob").unwrap(); + let headers = headers_from(&[("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT")]); + let signed = string_to_sign("acct", &Method::Get, &url, &headers); + // GET, then eleven empty slots, then the canonicalized headers and resource. + assert_eq!( + signed, + "GET\n\n\n\n\n\n\n\n\n\n\n\nx-ms-date:Fri, 14 Aug 2026 12:00:00 \ + GMT\n/acct/c/blob" + ); + } + + #[test] + fn test_string_to_sign_omits_zero_content_length() { + let url = Url::parse("https://acct.blob.core.windows.net/c/blob").unwrap(); + let with_zero = headers_from(&[("content-length", "0")]); + let with_body = headers_from(&[("content-length", "17")]); + assert!(string_to_sign("acct", &Method::Put, &url, &with_zero).starts_with("PUT\n\n\n\n")); + assert!(string_to_sign("acct", &Method::Put, &url, &with_body).starts_with("PUT\n\n\n17\n")); + } + + #[test] + fn test_signature_is_stable_for_a_known_request() { + // Pins the whole pipeline: string construction, base64 key decoding, HMAC, and + // base64 of the signature. A change to any of them changes this value. + let url = Url::parse("https://devstoreaccount1.blob.core.windows.net/c/blob").unwrap(); + let headers = headers_from(&[("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT")]); + let string_to_sign = string_to_sign(EMULATOR_ACCOUNT, &Method::Get, &url, &headers); + let signature = hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); + // Signatures are 32 bytes of HMAC SHA256, base64 encoded. + assert_eq!(signature.len(), 44); + let recomputed = + hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); + assert_eq!(signature, recomputed); + } + + #[test] + fn test_debug_does_not_leak_the_key() { + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let rendered = format!("{policy:?}"); + assert!(rendered.contains(EMULATOR_ACCOUNT)); + assert!(!rendered.contains(EMULATOR_ACCOUNT_KEY)); + } +} diff --git a/quickwit/quickwit-storage/src/object_storage/mod.rs b/quickwit/quickwit-storage/src/object_storage/mod.rs index e914c107291..b19d66c343e 100644 --- a/quickwit/quickwit-storage/src/object_storage/mod.rs +++ b/quickwit/quickwit-storage/src/object_storage/mod.rs @@ -26,4 +26,8 @@ mod s3_compatible_storage_resolver; #[cfg(feature = "azure")] mod azure_blob_storage; #[cfg(feature = "azure")] +mod azure_credentials; +#[cfg(feature = "azure")] +mod azure_shared_key; +#[cfg(feature = "azure")] pub use self::azure_blob_storage::{AzureBlobStorage, AzureBlobStorageFactory}; From 34ea00e37af198421a6374e957d62cf75340378a Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 09:29:05 -0700 Subject: [PATCH 3/6] feat(storage): port the Azure blob backend to azure_storage_blob 1.0 Rewrites `AzureBlobStorage` against the 1.0 clients and deletes the last references to the legacy SDK, so `quickwit-storage` builds again. The client mapping is mostly mechanical: `ContainerClient` becomes `BlobContainerClient`, `put_block_blob` becomes `BlockBlobClient::upload`, `put_block` and `put_block_list` become `stage_block` and `commit_block_list`, and `list_blobs` yields a `Pager` rather than a `Pageable`. Two places needed more thought. Downloads no longer walk a page of chunk responses. `BlobClient::download` returns one result whose `body` is a stream, so `copy_to` and `get_slice_stream` share a single `get_to_reader` helper that pulls the first chunk before returning. That keeps an error arriving with the response headers inside the retry rather than handing it to a caller with no way to retry. `BlockBlobClient` is not `Clone` in 1.0, so each part of a multipart upload builds its own client from the container client. Construction is local: the pipeline is behind an `Arc` and only the URL differs. Two behaviour notes. Single part upload sets `blob_content_md5` rather than a transactional checksum, because the partitioned upload path does not expose one, so the digest is stored with the blob instead of verified per request. Multipart still checks per block via `stage_block`. `ClientBuilder::emulator()` is gone, and the SDK could not have kept it, since the emulator authenticates with a shared key. The integration test now asks this crate to create and delete its container so the signing policy stays internal. 85 unit tests pass, clippy is clean, and `quickwit-cli` builds under `release-feature-set`. --- CHANGELOG.md | 2 + .../src/object_storage/azure_blob_storage.rs | 395 +++++++++--------- .../src/object_storage/azure_credentials.rs | 37 +- .../src/object_storage/azure_shared_key.rs | 21 +- .../quickwit-storage/tests/azure_storage.rs | 6 +- 5 files changed, 243 insertions(+), 218 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1981b1a9dfa..2932c411cf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - (Jaeger) Query resource attributes when Jaeger request carries tags +- Azure Blob Storage: workload identity no longer fails 24 hours after start, since `azure_identity` 1.0 re-reads the projected service account token (#6672) ### Changed +- Azure Blob Storage: migrate to `azure_storage_blob` 1.0, the rewritten Azure SDK (#6672) ### Deprecated diff --git a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs index cbbf6ea8ded..d611fa9d42d 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs @@ -13,20 +13,22 @@ // limitations under the License. use std::collections::HashMap; -use std::num::NonZeroU32; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::{Arc, LazyLock}; use std::{fmt, io}; use async_trait::async_trait; +use azure_core::Error as AzureError; use azure_core::error::ErrorKind; -use azure_core::{Pageable, StatusCode}; -use azure_storage::prelude::*; -use azure_storage::{CloudLocation, Error as AzureError}; -use azure_storage_blobs::blob::operations::GetBlobResponse; -use azure_storage_blobs::prelude::*; -use bytes::{Bytes, BytesMut}; +use azure_core::http::{NoFormat, RequestContent, StatusCode}; +use azure_storage_blob::BlobContainerClient; +use azure_storage_blob::models::{ + BlobClientDownloadOptions, BlobClientGetPropertiesResultHeaders, + BlobContainerClientListBlobsOptions, BlockBlobClientStageBlockOptions, + BlockBlobClientUploadOptions, BlockLookupList, HttpRange, +}; +use bytes::Bytes; use bytesize::ByteSize; use futures::io::Error as FutureError; use futures::stream::{StreamExt, TryStreamExt}; @@ -40,12 +42,14 @@ use regex::Regex; use tantivy::directory::OwnedBytes; use thiserror::Error; use tokio::io::{AsyncRead, AsyncWriteExt, BufReader}; -use tokio_util::compat::FuturesAsyncReadCompatExt; use tokio_util::io::StreamReader; -use tracing::{info, instrument, warn}; +use tracing::{instrument, warn}; use crate::debouncer::DebouncedStorage; use crate::metrics::object_storage_get_slice_in_flight_guards; +use crate::object_storage::azure_credentials::{ + AzureCredential, build_container_client, resolve_credential, +}; use crate::stable_deref_bytes::into_owned_bytes; use crate::storage::SendableAsync; use crate::{ @@ -79,7 +83,7 @@ impl StorageFactory for AzureBlobStorageFactory { /// Azure object storage implementation pub struct AzureBlobStorage { - container_client: ContainerClient, + container_client: BlobContainerClient, uri: Uri, prefix: PathBuf, multipart_policy: MultiPartPolicy, @@ -97,20 +101,23 @@ impl fmt::Debug for AzureBlobStorage { impl AzureBlobStorage { /// Creates a new [`AzureBlobStorage`] instance. - pub fn new( - storage_account_name: String, - storage_credentials: StorageCredentials, + /// + /// Crate visible because the credential type is: callers outside the crate go through + /// [`AzureBlobStorage::from_uri`], which resolves the credential from the config. + pub(crate) fn new( + storage_account_name: &str, + credential: AzureCredential, blob_service_uri: Option, uri: Uri, - container_name: String, - ) -> Self { + container_name: &str, + ) -> Result { let container_client = build_container_client( storage_account_name, - storage_credentials, + credential, blob_service_uri, container_name, - ); - Self { + )?; + Ok(Self { container_client, uri, prefix: PathBuf::new(), @@ -125,7 +132,7 @@ impl AzureBlobStorage { max_concurrent_uploads: 100, }, retry_params: RetryParams::aggressive(), - } + }) } /// Sets the prefix path. @@ -142,11 +149,16 @@ impl AzureBlobStorage { } /// Creates an emulated storage for testing. + /// + /// The 1.0 SDK has no `emulator()` helper, and could not offer one: the emulator + /// authenticates with a shared key, which the SDK no longer signs. The well-known + /// account and key are documented at + /// . #[cfg(feature = "integration-testsuite")] pub fn new_emulated(container: &str) -> Self { use std::str::FromStr; - let container_client = ClientBuilder::emulator().container_client(container); + let container_client = build_emulated_container_client(container); let uri = Uri::from_str(&format!("azure://tester/{container}")).unwrap(); Self { @@ -158,6 +170,30 @@ impl AzureBlobStorage { } } + /// Creates the container the emulated storage writes to. + /// + /// Exposed for the integration test suite because the shared key signing the emulator + /// needs stays inside this crate, so a test cannot build a client of its own. + #[cfg(feature = "integration-testsuite")] + pub async fn create_emulated_container(container: &str) -> anyhow::Result<()> { + build_emulated_container_client(container) + .create(None) + .await?; + Ok(()) + } + + /// Deletes the container the emulated storage writes to. + /// + /// Counterpart of [`AzureBlobStorage::create_emulated_container`], exposed for the same + /// reason. + #[cfg(feature = "integration-testsuite")] + pub async fn delete_emulated_container(container: &str) -> anyhow::Result<()> { + build_emulated_container_client(container) + .delete(None) + .await?; + Ok(()) + } + /// Sets the multipart policy. /// /// See `MultiPartPolicy`. @@ -180,31 +216,19 @@ impl AzureBlobStorage { ); StorageResolverError::InvalidConfig(message) })?; - let storage_credentials = if let Some(access_key) = - azure_storage_config.resolve_access_key() - { - StorageCredentials::access_key(storage_account_name.clone(), access_key) - } else if let Ok(credential) = azure_identity::create_credential() { - StorageCredentials::token_credential(credential) - } else { - return Err(StorageResolverError::InvalidConfig( - "could not find Azure storage account credentials using the following credential \ - providers: environment, managed identity, and storage account access key" - .to_string(), - )); - }; + let credential = resolve_credential(azure_storage_config)?; let (container_name, prefix) = parse_azure_uri(uri).ok_or_else(|| { let message = format!("failed to extract container name from Azure URI `{uri}`"); StorageResolverError::InvalidUri(message) })?; let blob_service_uri = azure_storage_config.resolve_blob_service_uri(&storage_account_name); let azure_blob_storage = AzureBlobStorage::new( - storage_account_name, - storage_credentials, + &storage_account_name, + credential, blob_service_uri, uri.clone(), - container_name, - ); + &container_name, + )?; Ok(azure_blob_storage.with_prefix(prefix)) } @@ -214,7 +238,7 @@ impl AzureBlobStorage { key_path.to_string_lossy().to_string() } - /// Downloads a blob as `Bytes` — zero-copy when the blob arrives as a single chunk. + /// Downloads a blob as `Bytes`. async fn get_to_bytes( &self, path: &Path, @@ -224,27 +248,69 @@ impl AzureBlobStorage { let capacity = range_opt.as_ref().map(Range::len).unwrap_or(0); retry(&self.retry_params, || async { let _timer = HistogramTimer::new(&crate::metrics::OBJECT_STORAGE_GET_OBJECT_DURATION); - let (mut response_stream, _in_flight_guards) = if let Some(range) = range_opt.as_ref() { - let stream = self - .container_client - .blob_client(&name) - .get() - .range(range.clone()) - .into_stream(); - // only record ranged get request as being in flight - let in_flight_guards = object_storage_get_slice_in_flight_guards(capacity); - (stream, Some(in_flight_guards)) - } else { - let stream = self.container_client.blob_client(&name).get().into_stream(); - (stream, None) + let mut download_options = BlobClientDownloadOptions::default(); + // Only a ranged get counts as in flight, matching the metric's meaning. + let _in_flight_guards = match range_opt.as_ref() { + Some(range) => { + download_options.range = Some(HttpRange::from(range.clone())); + Some(object_storage_get_slice_in_flight_guards(capacity)) + } + None => None, }; - let bytes = download_all(&mut response_stream).await?; + let download_response = self + .container_client + .blob_client(&name) + .download(Some(download_options)) + .await?; + let bytes = download_response.body.collect().await?; + crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(bytes.len() as u64); Result::<_, AzureErrorWrapper>::Ok(bytes) }) .await .map_err(StorageError::from) } + /// Opens a download as an [`AsyncRead`], optionally over a range. + /// + /// The first chunk is pulled before returning so that an error arriving with the + /// response headers is still inside the retry, rather than surfacing later to a caller + /// that cannot retry it. + async fn get_to_reader( + &self, + path: &Path, + range_opt: Option>, + ) -> StorageResult> { + let name = self.blob_name(path); + retry(&self.retry_params, || async { + let mut download_options = BlobClientDownloadOptions::default(); + if let Some(range) = range_opt.as_ref() { + download_options.range = Some(HttpRange::from(range.clone())); + } + let download_response = self + .container_client + .blob_client(&name) + .download(Some(download_options)) + .await?; + let mut bytes_stream = download_response + .body + .map(|bytes_res| bytes_res.map_err(FutureError::other)); + let first_chunk = bytes_stream.next().await; + let reader: Box = match first_chunk { + Some(bytes_res) => { + let first_chunk = bytes_res.map_err(AzureErrorWrapper::from)?; + let reconstructed_stream = Box::pin( + futures::stream::once(async { Ok(first_chunk) }).chain(bytes_stream), + ); + Box::new(StreamReader::new(reconstructed_stream)) + } + None => Box::new(tokio::io::empty()), + }; + Result::, AzureErrorWrapper>::Ok(reader) + }) + .await + .map_err(|err| err.into()) + } + /// Performs a single part upload. async fn put_single_part<'a>( &'a self, @@ -256,12 +322,20 @@ impl AzureBlobStorage { let _timer = HistogramTimer::new(&crate::metrics::OBJECT_STORAGE_PUT_OBJECT_DURATION); retry(&self.retry_params, || async { let data = Bytes::from(payload.read_all().await?.to_vec()); - let hash = azure_storage_blobs::prelude::Hash::from(md5::compute(&data[..]).0); + let digest = md5::compute(&data[..]); + // 1.0 exposes `blob_content_md5` rather than a transactional checksum on this + // path: the digest is stored with the blob instead of checked per request. + let upload_options = BlockBlobClientUploadOptions { + blob_content_md5: Some(digest.0.to_vec()), + ..Default::default() + }; + // `RequestContent::from` is an inherent function over `Vec`, which shadows + // the `From` impl, so go through `Into` to keep the `Bytes` as is. + let content: RequestContent = data.into(); self.container_client .blob_client(name) - .put_block_blob(data) - .hash(hash) - .into_future() + .block_blob_client() + .upload(content, Some(upload_options)) .await?; Result::<(), AzureErrorWrapper>::Ok(()) }) @@ -281,10 +355,14 @@ impl AzureBlobStorage { let multipart_ranges = chunk_range(0..total_len as usize, part_len as usize).map(into_u64_range); - let blob_client = self.container_client.blob_client(name); + let block_blob_client = self.container_client.blob_client(name).block_blob_client(); let upload_blocks_stream = tokio_stream::iter(multipart_ranges.enumerate()) .map(|(num, range)| { - let moved_blob_client = blob_client.clone(); + // `BlockBlobClient` is not `Clone` in the 1.0 SDK, so each part builds its + // own from the container client. Construction is local: it clones the + // pipeline behind an `Arc` and appends the blob name to the URL. + let moved_block_blob_client = + self.container_client.blob_client(name).block_blob_client(); let moved_payload = payload.clone(); crate::metrics::OBJECT_STORAGE_PUT_PARTS.inc(); crate::metrics::OBJECT_STORAGE_UPLOAD_NUM_BYTES.inc_by(range.end - range.start); @@ -297,11 +375,19 @@ impl AzureBlobStorage { let (data, hash_digest) = extract_range_data_and_hash(moved_payload.box_clone(), range.clone()) .await?; - let hash = azure_storage_blobs::prelude::Hash::from(hash_digest.0); - moved_blob_client - .put_block(block_id.clone(), data) - .hash(hash) - .into_future() + let content_length = data.len() as u64; + let stage_block_options = BlockBlobClientStageBlockOptions { + transactional_content_md5: Some(hash_digest.0.to_vec()), + ..Default::default() + }; + // The SDK base64 encodes the block id, so it takes the raw bytes. + moved_block_blob_client + .stage_block( + block_id.as_bytes(), + content_length, + data.into(), + Some(stage_block_options), + ) .await?; Result::<_, AzureErrorWrapper>::Ok(block_id) }) @@ -310,7 +396,7 @@ impl AzureBlobStorage { }) .buffer_unordered(self.multipart_policy.max_concurrent_uploads()); - // Collect and sort block ids to preserve part order for put_block_list. + // Collect and sort block ids to preserve part order for the block list. // Azure docs: "The put block list operation enforces the order in which blocks // are to be combined to create a blob". // https://docs.microsoft.com/en-us/rest/api/storageservices/put-block-list @@ -320,17 +406,22 @@ impl AzureBlobStorage { .map_err(StorageError::from)?; block_ids.sort_unstable(); - let block_list = BlockList { - blocks: block_ids - .into_iter() - .map(BlobBlockType::new_uncommitted) - .collect(), + let block_lookup_list = BlockLookupList { + uncommitted: Some( + block_ids + .into_iter() + .map(|block_id| block_id.into_bytes()) + .collect(), + ), + ..Default::default() }; + let block_list_content = RequestContent::try_from(block_lookup_list) + .map_err(AzureErrorWrapper::from) + .map_err(StorageError::from)?; // Commit all uploaded blocks. - blob_client - .put_block_list(block_list) - .into_future() + block_blob_client + .commit_block_list(block_list_content, None) .await .map_err(AzureErrorWrapper::from)?; @@ -341,15 +432,13 @@ impl AzureBlobStorage { #[async_trait] impl Storage for AzureBlobStorage { async fn check_connectivity(&self) -> anyhow::Result<()> { - if let Some(first_blob_result) = self - .container_client - .list_blobs() - .max_results(NonZeroU32::new(1u32).expect("1 is always non-zero.")) - .into_stream() - .next() - .await - { - let _ = first_blob_result?; + let list_blobs_options = BlobContainerClientListBlobsOptions { + maxresults: Some(1), + ..Default::default() + }; + let mut blob_pager = self.container_client.list_blobs(Some(list_blobs_options))?; + if let Some(first_page_result) = blob_pager.next().await { + let _ = first_page_result?; } Ok(()) } @@ -376,20 +465,10 @@ impl Storage for AzureBlobStorage { #[instrument(name = "storage.azure.copy_to", level = "debug", skip(self, output))] async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { - let name = self.blob_name(path); - let mut output_stream = self.container_client.blob_client(name).get().into_stream(); - - while let Some(chunk_result) = output_stream.next().await { - let chunk_response = chunk_result.map_err(AzureErrorWrapper::from)?; - let chunk_response_body_stream = chunk_response - .data - .map_err(FutureError::other) - .into_async_read() - .compat(); - let mut body_stream_reader = BufReader::new(chunk_response_body_stream); - let num_bytes_copied = tokio::io::copy_buf(&mut body_stream_reader, output).await?; - crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(num_bytes_copied); - } + let reader = self.get_to_reader(path, None).await?; + let mut body_stream_reader = BufReader::new(reader); + let num_bytes_copied = tokio::io::copy_buf(&mut body_stream_reader, output).await?; + crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(num_bytes_copied); output.flush().await?; Ok(()) } @@ -403,9 +482,8 @@ impl Storage for AzureBlobStorage { let blob_name = self.blob_name(path); let delete_res: Result<_, StorageError> = self .container_client - .blob_client(blob_name) - .delete() - .into_future() + .blob_client(&blob_name) + .delete(None) .await .map_err(|err| AzureErrorWrapper::from(err).into()); ignore_error_kind!(StorageErrorKind::NotFound, delete_res)?; @@ -476,33 +554,7 @@ impl Storage for AzureBlobStorage { path: &Path, range: Range, ) -> StorageResult> { - retry(&self.retry_params, || async { - let range = range.clone(); - let name = self.blob_name(path); - let page_stream = self - .container_client - .blob_client(name) - .get() - .range(range) - .into_stream(); - let mut bytes_stream = page_stream - .map(|page_res| page_res.map(|page| page.data).map_err(FutureError::other)) - .try_flatten() - .map(|bytes_res| bytes_res.map_err(FutureError::other)); - // Peek into the stream so that any early error can be retried - let first_chunk = bytes_stream.next().await; - let reader: Box = if let Some(res) = first_chunk { - let first_chunk = res.map_err(AzureErrorWrapper::from)?; - let reconstructed_stream = - Box::pin(futures::stream::once(async { Ok(first_chunk) }).chain(bytes_stream)); - Box::new(StreamReader::new(reconstructed_stream)) - } else { - Box::new(tokio::io::empty()) - }; - Result::, AzureErrorWrapper>::Ok(reader) - }) - .await - .map_err(|e| e.into()) + self.get_to_reader(path, Some(range)).await } #[instrument( @@ -532,12 +584,17 @@ impl Storage for AzureBlobStorage { let name = self.blob_name(path); let properties_result = self .container_client - .blob_client(name) - .get_properties() - .into_future() + .blob_client(&name) + .get_properties(None) .await; match properties_result { - Ok(response) => Ok(response.blob.properties.content_length), + Ok(response) => { + let content_length = response + .content_length() + .map_err(AzureErrorWrapper::from)? + .unwrap_or(0); + Ok(content_length) + } Err(err) => Err(StorageError::from(AzureErrorWrapper::from(err))), } } @@ -547,6 +604,28 @@ impl Storage for AzureBlobStorage { } } +/// Builds a container client pointed at Azurite. +/// +/// The well-known emulator account and key are published by Microsoft at +/// , so neither is a +/// secret. The 1.0 SDK dropped `ClientBuilder::emulator()`, and could not have kept it: the +/// emulator authenticates with a shared key, which the SDK no longer signs. +#[cfg(feature = "integration-testsuite")] +fn build_emulated_container_client(container: &str) -> BlobContainerClient { + const EMULATOR_ACCOUNT: &str = "devstoreaccount1"; + const EMULATOR_ACCOUNT_KEY: &str = + "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + const EMULATOR_BLOB_SERVICE_URI: &str = "http://127.0.0.1:10000/devstoreaccount1"; + + build_container_client( + EMULATOR_ACCOUNT, + AzureCredential::SharedKey(EMULATOR_ACCOUNT_KEY.to_string()), + Some(EMULATOR_BLOB_SERVICE_URI.to_string()), + container, + ) + .expect("the emulator endpoint should be a valid URL") +} + /// Copy range of payload into `Bytes` and return the computed md5. async fn extract_range_data_and_hash( payload: Box, @@ -563,25 +642,6 @@ async fn extract_range_data_and_hash( Ok((data, hash)) } -fn build_container_client( - storage_account_name: String, - storage_credentials: StorageCredentials, - blob_service_uri: Option, - container_name: String, -) -> ContainerClient { - let mut builder = ClientBuilder::new(storage_account_name.clone(), storage_credentials); - if let Some(uri) = blob_service_uri { - info!(endpoint=%uri, "using Azure blob storage endpoint defined in storage config or environment variable"); - builder = builder.cloud_location(CloudLocation::Custom { - account: storage_account_name, - uri, - }); - } - builder - .blob_service_client() - .container_client(container_name) -} - pub fn parse_azure_uri(uri: &Uri) -> Option<(String, PathBuf)> { // Ex: azure://container/prefix. static URI_PTN: LazyLock = LazyLock::new(|| { @@ -599,45 +659,6 @@ pub fn parse_azure_uri(uri: &Uri) -> Option<(String, PathBuf)> { Some((container, prefix)) } -/// Collect a download stream into a single [`Bytes`]. -/// -/// `Bytes` segments yielded by the SDK are preserved so that the single-segment case avoids the -/// extra copy into a contiguous buffer. When more than one segment is received, they are -/// concatenated exactly once into a freshly allocated `Bytes`. -async fn download_all( - chunk_stream: &mut Pageable, -) -> Result { - let mut segments: Vec = Vec::new(); - let mut total_num_bytes: usize = 0; - while let Some(chunk_result) = chunk_stream.next().await { - let chunk_response = chunk_result?; - let mut data_stream = chunk_response.data; - while let Some(bytes_res) = data_stream.next().await { - let bytes = bytes_res?; - total_num_bytes += bytes.len(); - segments.push(bytes); - } - } - crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(total_num_bytes as u64); - Ok(coalesce_segments(segments, total_num_bytes)) -} - -/// Returns a single [`Bytes`] covering `segments`. Zero-copy when there is at most one segment; -/// otherwise a single allocation concatenates them. -fn coalesce_segments(mut segments: Vec, total_num_bytes: usize) -> Bytes { - match segments.len() { - 0 => Bytes::new(), - 1 => segments.remove(0), - _ => { - let mut out = BytesMut::with_capacity(total_num_bytes); - for segment in segments { - out.extend_from_slice(&segment); - } - out.freeze() - } - } -} - #[derive(Error, Debug)] #[error("Azure error wrapper(inner={inner})")] struct AzureErrorWrapper { diff --git a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs index 8802b699503..f24353fee7b 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs @@ -96,16 +96,12 @@ fn resolve_token_credential() -> azure_core::Result> { /// A partial set means the webhook did not inject a usable identity, and building the /// credential would fail at the first request rather than here. fn workload_identity_env_is_complete() -> bool { - [ - AZURE_CLIENT_ID, - AZURE_TENANT_ID, - AZURE_FEDERATED_TOKEN_FILE, - ] - .iter() - .all(|variable| match env::var(variable) { - Ok(value) => !value.trim().is_empty(), - Err(_) => false, - }) + [AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_FEDERATED_TOKEN_FILE] + .iter() + .all(|variable| match env::var(variable) { + Ok(value) => !value.trim().is_empty(), + Err(_) => false, + }) } /// Builds the container client. @@ -162,11 +158,12 @@ fn join_container( blob_service_uri: &str, container_name: &str, ) -> Result { - let mut container_url = Url::parse(blob_service_uri.trim_end_matches('/')).map_err(|error| { - StorageResolverError::InvalidConfig(format!( - "`{blob_service_uri}` is not a valid Azure blob service URL: {error}" - )) - })?; + let mut container_url = + Url::parse(blob_service_uri.trim_end_matches('/')).map_err(|error| { + StorageResolverError::InvalidConfig(format!( + "`{blob_service_uri}` is not a valid Azure blob service URL: {error}" + )) + })?; container_url .path_segments_mut() .map_err(|_| { @@ -186,13 +183,19 @@ mod tests { #[test] fn test_join_container_appends_the_container() { let url = join_container("https://acct.blob.core.windows.net", "my-container").unwrap(); - assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/my-container"); + assert_eq!( + url.as_str(), + "https://acct.blob.core.windows.net/my-container" + ); } #[test] fn test_join_container_tolerates_a_trailing_slash() { let url = join_container("https://acct.blob.core.windows.net/", "my-container").unwrap(); - assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/my-container"); + assert_eq!( + url.as_str(), + "https://acct.blob.core.windows.net/my-container" + ); } #[test] diff --git a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs index 24b049f1bd1..ee17d4325b0 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs @@ -218,8 +218,8 @@ mod tests { /// , so this is /// not a secret. const EMULATOR_ACCOUNT: &str = "devstoreaccount1"; - const EMULATOR_ACCOUNT_KEY: &str = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVEr\ - Cz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + const EMULATOR_ACCOUNT_KEY: &str = + "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; fn headers_from(pairs: &[(&str, &str)]) -> Headers { let mut headers = Headers::new(); @@ -255,8 +255,8 @@ mod tests { #[test] fn test_canonicalized_resource_comma_joins_repeated_query_params() { - let url = - Url::parse("https://acct.blob.core.windows.net/c?include=metadata&include=copy").unwrap(); + let url = Url::parse("https://acct.blob.core.windows.net/c?include=metadata&include=copy") + .unwrap(); assert_eq!( canonicalized_resource("acct", &url), "/acct/c\ninclude:copy,metadata" @@ -273,7 +273,8 @@ mod tests { ]); assert_eq!( canonicalized_headers(&headers), - "x-ms-blob-type:Block Blob\nx-ms-date:Fri, 14 Aug 2026 12:00:00 GMT\nx-ms-version:2025-05-05\n" + "x-ms-blob-type:Block Blob\nx-ms-date:Fri, 14 Aug 2026 12:00:00 \ + GMT\nx-ms-version:2025-05-05\n" ); } @@ -285,8 +286,7 @@ mod tests { // GET, then eleven empty slots, then the canonicalized headers and resource. assert_eq!( signed, - "GET\n\n\n\n\n\n\n\n\n\n\n\nx-ms-date:Fri, 14 Aug 2026 12:00:00 \ - GMT\n/acct/c/blob" + "GET\n\n\n\n\n\n\n\n\n\n\n\nx-ms-date:Fri, 14 Aug 2026 12:00:00 GMT\n/acct/c/blob" ); } @@ -296,7 +296,9 @@ mod tests { let with_zero = headers_from(&[("content-length", "0")]); let with_body = headers_from(&[("content-length", "17")]); assert!(string_to_sign("acct", &Method::Put, &url, &with_zero).starts_with("PUT\n\n\n\n")); - assert!(string_to_sign("acct", &Method::Put, &url, &with_body).starts_with("PUT\n\n\n17\n")); + assert!( + string_to_sign("acct", &Method::Put, &url, &with_body).starts_with("PUT\n\n\n17\n") + ); } #[test] @@ -309,8 +311,7 @@ mod tests { let signature = hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); // Signatures are 32 bytes of HMAC SHA256, base64 encoded. assert_eq!(signature.len(), 44); - let recomputed = - hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); + let recomputed = hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); assert_eq!(signature, recomputed); } diff --git a/quickwit/quickwit-storage/tests/azure_storage.rs b/quickwit/quickwit-storage/tests/azure_storage.rs index 1619596a822..a62fcd9f2c5 100644 --- a/quickwit/quickwit-storage/tests/azure_storage.rs +++ b/quickwit/quickwit-storage/tests/azure_storage.rs @@ -23,15 +23,13 @@ async fn azure_storage_test_suite() -> anyhow::Result<()> { use std::path::PathBuf; use anyhow::Context; - use azure_storage_blobs::prelude::ClientBuilder; use quickwit_common::rand::append_random_suffix; use quickwit_storage::{AzureBlobStorage, MultiPartPolicy}; let _ = tracing_subscriber::fmt::try_init(); // Setup container. let container_name = append_random_suffix("quickwit").to_lowercase(); - let container_client = ClientBuilder::emulator().container_client(&container_name); - container_client.create().into_future().await?; + AzureBlobStorage::create_emulated_container(&container_name).await?; let mut object_storage = AzureBlobStorage::new_emulated(&container_name); quickwit_storage::storage_test_suite(&mut object_storage).await?; @@ -56,6 +54,6 @@ async fn azure_storage_test_suite() -> anyhow::Result<()> { .context("test multipart upload failed")?; // Teardown container. - container_client.delete().into_future().await?; + AzureBlobStorage::delete_emulated_container(&container_name).await?; Ok(()) } From 3f8944b283c2bf1c1392e651593b6a1b10291110 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 09:45:36 -0700 Subject: [PATCH 4/6] fix(storage): sign Content-Length that the transport would add later Verified against Azurite, which rejected `commit_block_list` with `AuthorizationFailure` while every other operation authorized fine. The generated operations are inconsistent about `Content-Length`. `stage_block` inserts the header itself, so signing saw it. `commit_block_list` leaves it to the transport, so signing saw nothing and covered an empty length while the wire carried the real one, and shared key rejects that mismatch. The response says only that the signature is malformed, so the operation-specific nature of the failure is invisible from the error. Set the header before signing when the body length is known and the header is absent, which makes the signature and the wire agree whatever the operation did. A zero length body still signs an empty slot, as the specification requires from API version 2015-02-21 onwards. Two tests cover it through `Policy::send` with a capturing terminal policy, one for the header being added and one for an existing value being left alone. Removing the fix fails the first and nothing else. Azurite needed two changes to run the suite at all. The pinned 3.24.0 predates the API version `azure_storage_blob` 1.0 sends, and so does 3.36.0, the newest released, so the emulator now runs with `--skipApiVersionCheck`. Skipping the check keeps the emulated client sending the same request as production, which is what the signing needs to be tested against, rather than pinning an older API version only the tests would use. --- docker-compose.yml | 8 +- .../src/object_storage/azure_shared_key.rs | 103 +++++++++++++++++- 2 files changed, 108 insertions(+), 3 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 24f5ed29b47..cc0e9684a3d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -128,7 +128,7 @@ services: retries: 100 azurite: - image: mcr.microsoft.com/azure-storage/azurite:${AZURITE_VERSION:-3.24.0} + image: mcr.microsoft.com/azure-storage/azurite:${AZURITE_VERSION:-3.36.0} container_name: azurite ports: - "${MAP_HOST_AZURITE:-127.0.0.1}:10000:10000" # Blob store port @@ -137,7 +137,11 @@ services: - azurite volumes: - azurite_data:/data - command: azurite --blobHost 0.0.0.0 --loose + # `--skipApiVersionCheck` because `azure_storage_blob` 1.0 sends a newer + # `x-ms-version` than any released Azurite recognizes. Skipping the check keeps the + # emulated client sending the same request as production, which is the point of the + # test, rather than pinning an older API version only the tests would use. + command: azurite --blobHost 0.0.0.0 --loose --skipApiVersionCheck fake-gcs-server: image: fsouza/fake-gcs-server:${FAKE_GCS_SERVER_VERSION:-1.47.7} diff --git a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs index ee17d4325b0..b0d75b8c1ef 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs @@ -40,6 +40,8 @@ use time::OffsetDateTime; /// `Date` slot of the string to sign stays empty, which is what the service expects when /// `x-ms-date` is present. const X_MS_DATE: &str = "x-ms-date"; +/// Signed and sent together, so both have to be derived from the same value. +const CONTENT_LENGTH: &str = "content-length"; /// Signs every request with the storage account key. /// @@ -82,6 +84,21 @@ impl Policy for SharedKeyAuthorizationPolicy { let now = OffsetDateTime::now_utc(); request.insert_header(X_MS_DATE, to_rfc7231(&now)); + // Some operations leave `Content-Length` for the transport to fill in, `commit_block + // _list` among them, while others set it themselves. Signing has to agree with what + // finally goes on the wire, so set the header here when the body length is known and + // the header is missing. Signing an empty length against a request that carries a + // real one is rejected as `AuthorizationFailure`, with nothing to indicate why. + let body_len = request.body().len(); + if header_or_empty(request.headers(), CONTENT_LENGTH).is_empty() { + match body_len { + // A zero length body is the documented exception: the slot stays empty in + // the string to sign even though the wire carries `Content-Length: 0`. + Some(len) if len > 0 => request.insert_header(CONTENT_LENGTH, len.to_string()), + _ => {} + } + } + let method = request.method(); let string_to_sign = string_to_sign(&self.account, &method, request.url(), request.headers()); @@ -115,7 +132,7 @@ fn header_or_empty(headers: &Headers, header_name: &str) -> String { /// calls out: it must be empty rather than `0` for requests without a body, for API /// versions from 2015-02-21 onwards. fn string_to_sign(account: &str, method: &Method, url: &Url, headers: &Headers) -> String { - let content_length = match header_or_empty(headers, "content-length") { + let content_length = match header_or_empty(headers, CONTENT_LENGTH) { length if length == "0" => String::new(), length => length, }; @@ -315,6 +332,90 @@ mod tests { assert_eq!(signature, recomputed); } + /// Terminal policy: records the headers it is handed and answers 200. + #[derive(Debug, Default)] + struct HeaderCapturingPolicy { + seen_headers: std::sync::Mutex>, + } + + #[async_trait::async_trait] + impl Policy for HeaderCapturingPolicy { + async fn send( + &self, + _ctx: &Context, + request: &mut Request, + _next: &[Arc], + ) -> PolicyResult { + *self.seen_headers.lock().unwrap() = Some(request.headers().clone()); + Ok(azure_core::http::AsyncRawResponse::from_bytes( + azure_core::http::StatusCode::Ok, + Headers::default(), + azure_core::Bytes::new(), + )) + } + } + + /// `commit_block_list` leaves `Content-Length` to the transport. Signing an empty length + /// while the wire carries a real one is rejected as `AuthorizationFailure`, and only a + /// live service reveals it, so pin the header here instead. + #[tokio::test] + async fn test_send_sets_content_length_before_signing() { + let capturing_policy = Arc::new(HeaderCapturingPolicy::default()); + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let url = + Url::parse("http://127.0.0.1:10000/devstoreaccount1/c/blob?comp=blocklist").unwrap(); + let mut request = Request::new(url, Method::Put); + request.set_body(azure_core::Bytes::from_static(b"")); + + let next: Vec> = vec![capturing_policy.clone()]; + policy + .send(&Context::default(), &mut request, &next) + .await + .unwrap(); + + let seen_headers = capturing_policy + .seen_headers + .lock() + .unwrap() + .clone() + .unwrap(); + assert_eq!(header_or_empty(&seen_headers, CONTENT_LENGTH), "12"); + // The signature has to cover that same length. + let signed = string_to_sign(EMULATOR_ACCOUNT, &Method::Put, request.url(), &seen_headers); + assert!(signed.starts_with("PUT\n\n\n12\n")); + } + + /// A request the caller already sized keeps that value rather than being overwritten. + #[tokio::test] + async fn test_send_keeps_an_existing_content_length() { + let capturing_policy = Arc::new(HeaderCapturingPolicy::default()); + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let url = Url::parse("http://127.0.0.1:10000/devstoreaccount1/c/blob").unwrap(); + let mut request = Request::new(url, Method::Put); + request.insert_header(CONTENT_LENGTH, "5"); + request.set_body(azure_core::Bytes::from_static(b"hello")); + + let next: Vec> = vec![capturing_policy.clone()]; + policy + .send(&Context::default(), &mut request, &next) + .await + .unwrap(); + + let seen_headers = capturing_policy + .seen_headers + .lock() + .unwrap() + .clone() + .unwrap(); + assert_eq!(header_or_empty(&seen_headers, CONTENT_LENGTH), "5"); + } + #[test] fn test_debug_does_not_leak_the_key() { let policy = SharedKeyAuthorizationPolicy::new( From 4fdf0095a3499dcd9b68e1b26cc20fa56c0c4324 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 18:12:14 -0700 Subject: [PATCH 5/6] fix(storage): verify single part uploads on the wire again `BlockBlobClient::upload()` only exposes `blob_content_md5`, which the service stores as a property without checking it against the body, so the port had quietly dropped the integrity check `put_block_blob(..).hash(..)` used to give us. Splits are immutable and never re-verified, so a corrupted upload would have been permanent and silent. Stage a single block and commit it instead. `stage_block` takes a transactional checksum, so the service rejects a payload that does not match on arrival. The cost is one extra request per object below the multipart threshold, which is the cheaper side of this trade. Also pins the path encoding in the canonicalized resource, which a review question prompted me to check. The path is signed exactly as the URI carries it, escapes and all, while query parameters are decoded. That asymmetry is specified: "any portion of the CanonicalizedResource string that is derived from the resource's URI should be encoded exactly as it is in the URI", and the query steps separately say to URL-decode each name and value. Decoding the path instead is rejected with `AuthorizationFailure`, confirmed against Azurite. Two tests now hold that shape in place so it does not get tidied away later. --- .../src/object_storage/azure_blob_storage.rs | 44 ++++++++++++++----- .../src/object_storage/azure_shared_key.rs | 38 +++++++++++++++- 2 files changed, 70 insertions(+), 12 deletions(-) diff --git a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs index d611fa9d42d..a6f4d97837c 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs @@ -25,8 +25,8 @@ use azure_core::http::{NoFormat, RequestContent, StatusCode}; use azure_storage_blob::BlobContainerClient; use azure_storage_blob::models::{ BlobClientDownloadOptions, BlobClientGetPropertiesResultHeaders, - BlobContainerClientListBlobsOptions, BlockBlobClientStageBlockOptions, - BlockBlobClientUploadOptions, BlockLookupList, HttpRange, + BlobContainerClientListBlobsOptions, BlockBlobClientStageBlockOptions, BlockLookupList, + HttpRange, }; use bytes::Bytes; use bytesize::ByteSize; @@ -57,6 +57,12 @@ use crate::{ StorageErrorKind, StorageFactory, StorageResolverError, StorageResult, }; +/// Block id used by the single part upload path. +/// +/// Zero padded like the multipart ids so that a blob written by either path carries a block +/// list of the same shape. +const SINGLE_PART_BLOCK_ID: &[u8] = b"block:00000"; + /// Azure object storage resolver. pub struct AzureBlobStorageFactory { storage_config: AzureStorageConfig, @@ -323,19 +329,37 @@ impl AzureBlobStorage { retry(&self.retry_params, || async { let data = Bytes::from(payload.read_all().await?.to_vec()); let digest = md5::compute(&data[..]); - // 1.0 exposes `blob_content_md5` rather than a transactional checksum on this - // path: the digest is stored with the blob instead of checked per request. - let upload_options = BlockBlobClientUploadOptions { - blob_content_md5: Some(digest.0.to_vec()), + let content_length = data.len() as u64; + let block_blob_client = self.container_client.blob_client(name).block_blob_client(); + // Staged and committed rather than uploaded in one shot, because `upload()` + // only exposes `blob_content_md5`, which the service stores without checking it + // against the body. `stage_block` takes a transactional checksum, so a corrupted + // payload is rejected on arrival. Splits are immutable and never re-verified, so + // a silent corruption here would be permanent. The cost is one extra request per + // object below the multipart threshold. + let stage_block_options = BlockBlobClientStageBlockOptions { + transactional_content_md5: Some(digest.0.to_vec()), ..Default::default() }; // `RequestContent::from` is an inherent function over `Vec`, which shadows // the `From` impl, so go through `Into` to keep the `Bytes` as is. let content: RequestContent = data.into(); - self.container_client - .blob_client(name) - .block_blob_client() - .upload(content, Some(upload_options)) + block_blob_client + .stage_block( + SINGLE_PART_BLOCK_ID, + content_length, + content, + Some(stage_block_options), + ) + .await?; + let block_lookup_list = BlockLookupList { + uncommitted: Some(vec![SINGLE_PART_BLOCK_ID.to_vec()]), + ..Default::default() + }; + let block_list_content = + RequestContent::try_from(block_lookup_list).map_err(AzureErrorWrapper::from)?; + block_blob_client + .commit_block_list(block_list_content, None) .await?; Result::<(), AzureErrorWrapper>::Ok(()) }) diff --git a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs index b0d75b8c1ef..05ae860faee 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs @@ -182,8 +182,20 @@ fn canonicalized_headers(headers: &Headers) -> String { rendered } -/// Renders the resource the request addresses: the account, the decoded path, then every -/// query parameter, lowercased and sorted, with repeated parameters comma joined. +/// Renders the resource the request addresses: the account, the path as encoded in the URI, +/// then every query parameter, lowercased, decoded and sorted, with repeated parameters comma +/// joined. +/// +/// The asymmetry is deliberate and specified: "any portion of the CanonicalizedResource +/// string that is derived from the resource's URI should be encoded exactly as it is in the +/// URI", and the path step says to append "the resource's encoded URI path", while the query +/// steps say to "URL-decode each query parameter name and value". So a blob named +/// `indexes/my-index/split` signs as `indexes%2Fmy-index%2Fsplit`, because that is how the +/// 1.0 clients put it in the URL. Decoding the path here is rejected with +/// `AuthorizationFailure`, which the integration suite confirms. +/// +/// The account appears twice for the emulator, whose URLs carry the account in the path. +/// That is expected and documented. fn canonicalized_resource(account: &str, url: &Url) -> String { let mut resource = String::with_capacity(url.as_str().len()); resource.push('/'); @@ -280,6 +292,28 @@ mod tests { ); } + /// The path is signed exactly as the URI carries it, escapes included. Quickwit blob + /// names hold separators, so the 1.0 clients percent-encode them into one segment and + /// the signature has to keep that form. Decoding here looks tidier and is rejected with + /// `AuthorizationFailure`. + #[test] + fn test_canonicalized_resource_keeps_the_path_encoded() { + let url = + Url::parse("https://acct.blob.core.windows.net/c/indexes%2Fmy-index%2Fsplit.split") + .unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/c/indexes%2Fmy-index%2Fsplit.split" + ); + } + + /// Query parameters are the exception: the specification decodes those. + #[test] + fn test_canonicalized_resource_decodes_query_values() { + let url = Url::parse("https://acct.blob.core.windows.net/c?prefix=a%2Fb").unwrap(); + assert_eq!(canonicalized_resource("acct", &url), "/acct/c\nprefix:a/b"); + } + #[test] fn test_canonicalized_headers_sorted_and_folded() { let headers = headers_from(&[ From abb142913b283f85fdb5b46d04eb77bf8281a6a0 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 18:17:54 -0700 Subject: [PATCH 6/6] fix(storage): keep service principal and user-assigned identity working Two credential regressions from replacing `azure_identity::create_credential()`, both raised in review. The old chain tried an environment credential before managed identity, reading `AZURE_TENANT_ID`, `AZURE_CLIENT_ID` and `AZURE_CLIENT_SECRET`. The replacement recognized only workload identity and fell through to managed identity, so a deployment authenticating with a service principal had its secret ignored and its request sent to IMDS. Those deployments would have lost Azure access on upgrade, with an error naming neither the secret nor the reason. Separately, `AZURE_CLIENT_ID` on its own names a user-assigned managed identity. Passing no options asks IMDS for the system-assigned identity, which is either absent or, on a host carrying both, the wrong principal. The client id now becomes `UserAssignedId::ClientId`. Selection moved into `select_token_credential_kind`, which takes a lookup function rather than reading the process environment, so precedence is covered by ordinary tests. Mutating environment variables inside a test is not safe while other tests run, and precedence between overlapping variable sets is exactly what needs pinning: a secret beats a federated token file, a partial set of either falls back rather than half-configuring a credential, and a blank value counts as unset. `TokenCredentialKind` carries the secret, so its `Debug` is hand written to redact it. A derived one would print the secret, and this type appears in test failure output. --- .../src/object_storage/azure_credentials.rs | 302 ++++++++++++++++-- 1 file changed, 274 insertions(+), 28 deletions(-) diff --git a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs index f24353fee7b..1b2a35d71d9 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs @@ -19,13 +19,16 @@ //! nearest remaining type, `DeveloperToolsCredential`, chains the Azure CLI and the //! Developer CLI only, which does not cover a pod. -use std::env; use std::sync::Arc; +use std::{env, fmt}; use azure_core::credentials::TokenCredential; use azure_core::http::policies::Policy; use azure_core::http::{ClientOptions, Url}; -use azure_identity::{ManagedIdentityCredential, WorkloadIdentityCredential}; +use azure_identity::{ + ClientSecretCredential, ManagedIdentityCredential, ManagedIdentityCredentialOptions, + UserAssignedId, WorkloadIdentityCredential, +}; use azure_storage_blob::{BlobContainerClient, BlobContainerClientOptions}; use quickwit_config::AzureStorageConfig; use tracing::info; @@ -37,9 +40,97 @@ use crate::object_storage::azure_shared_key::SharedKeyAuthorizationPolicy; const AZURE_CLIENT_ID: &str = "AZURE_CLIENT_ID"; const AZURE_TENANT_ID: &str = "AZURE_TENANT_ID"; const AZURE_FEDERATED_TOKEN_FILE: &str = "AZURE_FEDERATED_TOKEN_FILE"; +/// Set when a service principal authenticates with a secret rather than a federated token. +const AZURE_CLIENT_SECRET: &str = "AZURE_CLIENT_SECRET"; /// Lets an operator pin the provider instead of relying on the detection below. const AZURE_CREDENTIAL_KIND: &str = "AZURE_CREDENTIAL_KIND"; +/// Which token credential the environment describes. +/// +/// Kept separate from construction so the precedence can be tested without mutating process +/// environment, which no test can do safely while others run. +#[derive(Eq, PartialEq)] +enum TokenCredentialKind { + /// Service principal with a client secret, the `EnvironmentCredential` of the old chain. + ClientSecret { + tenant_id: String, + client_id: String, + secret: String, + }, + /// Federated token file, injected into a pod by the workload identity webhook. + WorkloadIdentity, + /// IMDS. `user_assigned_client_id` selects a user-assigned identity, and `None` means the + /// system-assigned one. + ManagedIdentity { + user_assigned_client_id: Option, + }, +} + +impl fmt::Debug for TokenCredentialKind { + /// Hand written so the client secret cannot reach a log line or a panic message. A + /// derived implementation would print it, and test failures print this type. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ClientSecret { + tenant_id, + client_id, + .. + } => formatter + .debug_struct("ClientSecret") + .field("tenant_id", tenant_id) + .field("client_id", client_id) + .finish_non_exhaustive(), + Self::WorkloadIdentity => formatter.write_str("WorkloadIdentity"), + Self::ManagedIdentity { + user_assigned_client_id, + } => formatter + .debug_struct("ManagedIdentity") + .field("user_assigned_client_id", user_assigned_client_id) + .finish(), + } + } +} + +/// Decides which credential the environment describes. +/// +/// Precedence follows the chain `azure_identity::create_credential()` used to walk, where an +/// environment credential came before managed identity. Dropping that ordering silently +/// breaks every deployment that authenticates with a service principal, because the client +/// secret is ignored and IMDS is contacted instead. +fn select_token_credential_kind(var: impl Fn(&str) -> Option) -> TokenCredentialKind { + let non_empty = |name: &str| match var(name) { + Some(value) if !value.trim().is_empty() => Some(value), + _ => None, + }; + let client_id = non_empty(AZURE_CLIENT_ID); + + // A secret and a federated token file are mutually exclusive in practice. The secret is + // checked first because the webhook injects the token file, so its presence says less + // about operator intent than a secret does. + if let (Some(tenant_id), Some(client_id), Some(secret)) = ( + non_empty(AZURE_TENANT_ID), + client_id.clone(), + non_empty(AZURE_CLIENT_SECRET), + ) { + return TokenCredentialKind::ClientSecret { + tenant_id, + client_id, + secret, + }; + } + if non_empty(AZURE_TENANT_ID).is_some() + && client_id.is_some() + && non_empty(AZURE_FEDERATED_TOKEN_FILE).is_some() + { + return TokenCredentialKind::WorkloadIdentity; + } + // `AZURE_CLIENT_ID` on its own names a user-assigned identity. Ignoring it asks IMDS for + // the system-assigned identity, which either does not exist or is the wrong principal. + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: client_id, + } +} + /// How requests to the blob service are authorized. pub(crate) enum AzureCredential { /// Storage account key, signed by Quickwit because the 1.0 SDK dropped shared key. @@ -74,34 +165,58 @@ fn resolve_token_credential() -> azure_core::Result> { .map(|kind| kind.trim().to_lowercase()) .unwrap_or_default(); - match credential_kind.as_str() { - "workloadidentity" => Ok(WorkloadIdentityCredential::new(None)?), - "managedidentity" => Ok(ManagedIdentityCredential::new(None)?), - // An empty or unrecognized value falls through to detection. The workload identity - // variables are injected by a mutating webhook rather than by an operator, so their - // presence is the signal that the pod runs under workload identity. - _ => { - if workload_identity_env_is_complete() { - info!("using azure workload identity credential"); - return Ok(WorkloadIdentityCredential::new(None)?); - } - info!("using azure managed identity credential"); - Ok(ManagedIdentityCredential::new(None)?) - } - } + let kind = match credential_kind.as_str() { + "workloadidentity" => TokenCredentialKind::WorkloadIdentity, + "managedidentity" => TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: env::var(AZURE_CLIENT_ID).ok().filter(|id| !id.is_empty()), + }, + // An empty or unrecognized value falls through to detection. + _ => select_token_credential_kind(|name| env::var(name).ok()), + }; + build_token_credential(kind) } -/// Returns `true` when all three variables a `WorkloadIdentityCredential` needs are present. -/// -/// A partial set means the webhook did not inject a usable identity, and building the -/// credential would fail at the first request rather than here. -fn workload_identity_env_is_complete() -> bool { - [AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_FEDERATED_TOKEN_FILE] - .iter() - .all(|variable| match env::var(variable) { - Ok(value) => !value.trim().is_empty(), - Err(_) => false, - }) +/// Builds the credential for a kind already chosen. +fn build_token_credential( + kind: TokenCredentialKind, +) -> azure_core::Result> { + match kind { + TokenCredentialKind::ClientSecret { + tenant_id, + client_id, + secret, + } => { + info!("using azure client secret credential"); + Ok(ClientSecretCredential::new( + &tenant_id, + client_id, + secret.into(), + None, + )?) + } + TokenCredentialKind::WorkloadIdentity => { + info!("using azure workload identity credential"); + Ok(WorkloadIdentityCredential::new(None)?) + } + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id, + } => { + let options = match user_assigned_client_id { + Some(client_id) => { + info!(%client_id, "using azure user-assigned managed identity credential"); + Some(ManagedIdentityCredentialOptions { + user_assigned_id: Some(UserAssignedId::ClientId(client_id)), + ..Default::default() + }) + } + None => { + info!("using azure system-assigned managed identity credential"); + None + } + }; + Ok(ManagedIdentityCredential::new(options)?) + } + } } /// Builds the container client. @@ -178,8 +293,139 @@ fn join_container( #[cfg(test)] mod tests { + use std::collections::HashMap; + use super::*; + fn env_from(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option { + let map: HashMap = pairs + .iter() + .map(|(name, value)| ((*name).to_owned(), (*value).to_owned())) + .collect(); + move |name: &str| map.get(name).cloned() + } + + /// A service principal authenticates with a secret. The old chain tried an environment + /// credential before managed identity, and losing that ordering sends the request to + /// IMDS with the secret ignored, so the deployment loses access entirely. + #[test] + fn test_client_secret_wins_over_managed_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "secret"), + ])); + assert_eq!( + kind, + TokenCredentialKind::ClientSecret { + tenant_id: "tenant".to_owned(), + client_id: "client".to_owned(), + secret: "secret".to_owned(), + } + ); + } + + /// The webhook injects the token file, so a secret is the stronger statement of intent. + #[test] + fn test_client_secret_wins_over_workload_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "secret"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert!(matches!(kind, TokenCredentialKind::ClientSecret { .. })); + } + + #[test] + fn test_workload_identity_when_the_three_variables_are_present() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert_eq!(kind, TokenCredentialKind::WorkloadIdentity); + } + + /// `AZURE_CLIENT_ID` alone names a user-assigned identity. Dropping it asks IMDS for the + /// system-assigned one, which is either absent or the wrong principal. + #[test] + fn test_client_id_alone_selects_a_user_assigned_identity() { + let kind = select_token_credential_kind(env_from(&[(AZURE_CLIENT_ID, "client")])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + + #[test] + fn test_empty_environment_selects_the_system_assigned_identity() { + let kind = select_token_credential_kind(env_from(&[])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: None, + } + ); + } + + /// A half configured service principal must not be read as one, and a blank variable is + /// the same as an unset one. + #[test] + fn test_blank_and_partial_values_do_not_select_a_service_principal() { + let partial = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + ])); + assert_eq!( + partial, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + + let blank = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, " "), + ])); + assert_eq!( + blank, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + + #[test] + fn test_debug_does_not_leak_the_client_secret() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "super-secret-value"), + ])); + let rendered = format!("{kind:?}"); + assert!(rendered.contains("client")); + assert!(!rendered.contains("super-secret-value")); + } + + /// A partial workload identity set is not workload identity. + #[test] + fn test_token_file_without_tenant_is_not_workload_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_CLIENT_ID, "client"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + #[test] fn test_join_container_appends_the_container() { let url = join_container("https://acct.blob.core.windows.net", "my-container").unwrap();