milvus-io · jiaqizho · Apr 27, 2026
diff --git a/cpp/include/milvus-storage/filesystem/azure/azure_cross_tenant_credential.h b/cpp/include/milvus-storage/filesystem/azure/azure_cross_tenant_credential.h
@@ -0,0 +1,96 @@
+// Copyright 2025 Zilliz
+//
+// 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.
+
+#pragma once
+
+#include <chrono>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+
+#include <azure/core/context.hpp>
+#include <azure/core/credentials/credentials.hpp>
+#include <azure/core/http/transport.hpp>
+
+namespace milvus_storage::fs {
+
+/// \brief Azure TokenCredential that performs the IMDS → AAD two-hop OAuth2
+/// federated client_assertion exchange to mint a customer-tenant Bearer.
+///
+/// The two-hop flow:
+///   1. GET 169.254.169.254/metadata/identity/oauth2/token
+///        ?api-version=2018-02-01
+///        &resource=api://AzureADTokenExchange
+///      → JWT signed by *our* tenant, audience=api://AzureADTokenExchange.
+///   2. POST https://login.microsoftonline.com/{customer_tenant}/oauth2/v2.0/token
+///      Body: client_id={customer_app}, scope=https://storage.azure.com/.default,
+///            grant_type=client_credentials, client_assertion={Step-1 JWT},
+///            client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer
+///      → Customer-tenant Bearer (~3600s lifetime).
+///
+/// This credential class is the C++ side mirror of `azure_federation.rs` in
+/// the Rust bridge: same protocol, same audience constants. We need both
+/// because the bridge handles Lance/Iceberg `plan_files` (Rust path) while
+/// this class handles Iceberg parquet data reads through the C++ Arrow
+/// Azure FS (since `IcebergFormatReader` calls `parquet::ParquetFormatReader`
+/// with an `arrow::fs::FileSystem`).
+///
+/// `ClientAssertionCredential` from Azure Identity 1.10+ would let us do
+/// this with just an assertion callback, but the version vendored via Conan
+/// (1.7.0-beta.3) doesn't have it yet. Subclassing `TokenCredential`
+/// directly is the portable path that works against any 1.x.
+///
+/// The cached bearer is refreshed when within `kRefreshOffset` seconds of
+/// expiry; concurrent `GetToken` calls that miss the cache may both fetch
+/// (no global mutex held during HTTP), but the cache stores whichever
+/// completes last — credential thrashing is bounded by the small number
+/// of in-flight FS operations rather than queue depth.
+class AzureCrossTenantCredential final : public Azure::Core::Credentials::TokenCredential {
+  public:
+  /// \param tenant_id  Customer's Entra ID tenant ID (the AAD authority).
+  /// \param client_id  Customer's App Registration client_id with a
+  ///                   Federated Identity Credential trusting our MI.
+  AzureCrossTenantCredential(std::string tenant_id, std::string client_id);
+
+  ~AzureCrossTenantCredential() override = default;
+
+  Azure::Core::Credentials::AccessToken GetToken(
+      Azure::Core::Credentials::TokenRequestContext const& tokenRequestContext,
+      Azure::Core::Context const& context) const override;
+
+  private:
+  /// Refresh the cached bearer when the remaining lifetime drops below this.
+  /// Mirrors `REFRESH_OFFSET_SECS` in `azure_federation.rs`.
+  static constexpr std::chrono::seconds kRefreshOffset{300};
+
+  /// Fetch step-1 MI assertion from IMDS. Returns the JWT string.
+  /// `transport` is reused across the two hops to share keep-alive.
+  std::string FetchManagedIdentityAssertion(Azure::Core::Http::HttpTransport& transport,
+                                            Azure::Core::Context const& context) const;
+
+  /// Fetch step-2 customer-tenant bearer. `mi_assertion` is the JWT from
+  /// step 1. Returns (bearer, expires_at).
+  std::pair<std::string, Azure::DateTime> ExchangeForStorageBearer(Azure::Core::Http::HttpTransport& transport,
+                                                                   std::string const& mi_assertion,
+                                                                   Azure::Core::Context const& context) const;
+
+  std::string tenant_id_;
+  std::string client_id_;
+
+  // Mutable state for cache. Mutex protects only cache entry, NOT the HTTP
+  // calls — those run unlocked so concurrent misses don't serialize.
+  mutable std::mutex cache_mu_;
+  mutable std::optional<Azure::Core::Credentials::AccessToken> cached_;
+};
+
+}  // namespace milvus_storage::fs
diff --git a/cpp/include/milvus-storage/filesystem/azure/azurefs.h b/cpp/include/milvus-storage/filesystem/azure/azurefs.h
@@ -215,6 +215,14 @@ struct ARROW_EXPORT AzureOptions {
   Status ConfigureCLICredential();
   Status ConfigureWorkloadIdentityCredential();
   Status ConfigureEnvironmentCredential();
+  /// \brief Cross-tenant Managed-Identity → AAD federated client_assertion.
+  ///
+  /// Uses our local IMDS-attached MI to request an `api://AzureADTokenExchange`
+  /// audience JWT, then exchanges it at the customer's tenant
+  /// (`{tenant_id}/oauth2/v2.0/token`) for a Bearer scoped to
+  /// `https://storage.azure.com/.default`. Customer-side prerequisite:
+  /// App Registration with a Federated Identity Credential trusting our MI.
+  Status ConfigureCrossTenantCredential(const std::string& tenant_id, const std::string& client_id);
 
   bool Equals(const AzureOptions& other) const;
 

diff --git a/cpp/include/milvus-storage/filesystem/fs.h b/cpp/include/milvus-storage/filesystem/fs.h
@@ -199,6 +199,15 @@ struct ArrowFileSystemConfig {
   // Target service account email to impersonate (e.g., "sa@project.iam.gserviceaccount.com")
   std::string gcp_target_service_account = "";
 
+  // Azure cross-tenant access (Managed Identity → customer-tenant bearer via
+  // OAuth2 federated client_assertion). When both are set, format readers
+  // emit bridge-private storage_options keys consumed by custom Lance/Iceberg
+  // providers; account_key/SAS are NOT used.
+  // - azure_client_id: customer's App Registration client_id
+  // - azure_tenant_id: customer's Entra ID tenant_id
+  std::string azure_client_id = "";
+  std::string azure_tenant_id = "";
+
   // Lifetime requested for cross-tenant temporary credentials, in seconds.
   // Shared across providers that mint short-lived tokens:
   //   - AWS STS AssumeRole: STS session length (valid range [900, 43200],

diff --git a/cpp/include/milvus-storage/properties.h b/cpp/include/milvus-storage/properties.h
@@ -101,6 +101,8 @@ struct PropertyInfo {
 
 // Cross-tenant access properties
 #define PROPERTY_FS_GCP_TARGET_SERVICE_ACCOUNT "fs.gcp_target_service_account"
+#define PROPERTY_FS_AZURE_CLIENT_ID "fs.azure_client_id"
+#define PROPERTY_FS_AZURE_TENANT_ID "fs.azure_tenant_id"
 
 // --- External Filesystem Properties ---
 // External filesystems are configured with properties following the pattern:

diff --git a/cpp/src/filesystem/azure/azure_cross_tenant_credential.cpp b/cpp/src/filesystem/azure/azure_cross_tenant_credential.cpp
@@ -0,0 +1,237 @@
+// Copyright 2025 Zilliz
+//
+// 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.
+
+#include "milvus-storage/filesystem/azure/azure_cross_tenant_credential.h"
+
+#include <cctype>
+#include <cstdio>
+#include <stdexcept>
+
+#include <azure/core/datetime.hpp>
+#include <azure/core/http/curl_transport.hpp>
+#include <azure/core/http/http.hpp>
+#include <azure/core/http/raw_response.hpp>
+#include <azure/core/io/body_stream.hpp>
+#include <azure/core/url.hpp>
+
+#include <folly/json/json.h>
+
+#include "milvus-storage/common/log.h"
+
+namespace milvus_storage::fs {
+
+namespace {
+
+// IMDS endpoint and audience are fixed by the Azure platform; not
+// configurable per VM. The audience `api://AzureADTokenExchange` is the
+// well-known audience that AAD's `oauth2/v2.0/token` accepts as a
+// `client_assertion` for federated identity credentials.
+constexpr char kImdsTokenUrl[] = "http://169.254.169.254/metadata/identity/oauth2/token";
+constexpr char kImdsApiVersion[] = "2018-02-01";
+constexpr char kMiAudience[] = "api://AzureADTokenExchange";
+
+// AAD authority host. Public Azure cloud only; sovereign clouds use
+// different hosts but cross-tenant FIC across sovereign boundaries is not a
+// supported scenario in this MVP.
+constexpr char kAadAuthority[] = "https://login.microsoftonline.com";
+
+// Scope for the customer-tenant Bearer. `.default` asks for whatever
+// permissions the customer's App Registration was granted on
+// `https://storage.azure.com/`.
+constexpr char kStorageScope[] = "https://storage.azure.com/.default";
+
+constexpr char kClientAssertionType[] = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer";
+
+/// Percent-encode a value for an `application/x-www-form-urlencoded` body.
+/// Keep alnum + `- _ . ~` literal; everything else percent-escaped. Equivalent
+/// to RFC 3986 unreserved.
+std::string UrlEncode(const std::string& s) {
+  std::string out;
+  out.reserve(s.size() * 11 / 10 + 2);
+  for (unsigned char c : s) {
+    if (std::isalnum(c) || c == '-' || c == '_' || c == '.' || c == '~') {
+      out.push_back(static_cast<char>(c));
+    } else {
+      char buf[4];
+      std::snprintf(buf, sizeof(buf), "%%%02X", c);
+      out.append(buf, 3);
+    }
+  }
+  return out;
+}
+
+/// Read full response body. Use `ExtractBodyStream` rather than `GetBody`
+/// because `CurlTransport` may return responses with the body still
+/// streaming and `GetBody` would be empty.
+std::string ReadResponseBodyAsString(Azure::Core::Http::RawResponse& response, Azure::Core::Context const& context) {
+  auto stream = response.ExtractBodyStream();
+  std::string body;
+  if (stream) {
+    auto bytes = stream->ReadToEnd(context);
+    body.assign(bytes.begin(), bytes.end());
+  }
+  // Some transport configurations populate `m_body` directly instead of
+  // setting a stream; fall back to that.
+  if (body.empty()) {
+    auto const& raw = response.GetBody();
+    body.assign(raw.begin(), raw.end());
+  }
+  return body;
+}
+
+}  // namespace
+
+AzureCrossTenantCredential::AzureCrossTenantCredential(std::string tenant_id, std::string client_id)
+    : Azure::Core::Credentials::TokenCredential("AzureCrossTenantCredential"),
+      tenant_id_(std::move(tenant_id)),
+      client_id_(std::move(client_id)) {}
+
+Azure::Core::Credentials::AccessToken AzureCrossTenantCredential::GetToken(
+    Azure::Core::Credentials::TokenRequestContext const& /*tokenRequestContext*/,
+    Azure::Core::Context const& context) const {
+  // Fast path: cached token still has comfortable lifetime.
+  {
+    std::lock_guard<std::mutex> g(cache_mu_);
+    if (cached_) {
+      // Both sides as Azure::DateTime so the time_point template params line
+      // up; DateTime inherits from system_clock::time_point but operator-
+      // doesn't pick up the unrelated raw time_point on the RHS.
+      auto remaining = cached_->ExpiresOn - Azure::DateTime(std::chrono::system_clock::now());
+      if (remaining > kRefreshOffset) {
+        return *cached_;
+      }
+    }
+  }
+
+  // Slow path: fetch fresh. We deliberately don't hold cache_mu_ across the
+  // HTTP calls — concurrent misses may both fetch, but only one bearer
+  // wins the cache slot. That is bounded waste; holding the mutex would
+  // serialize unrelated FS operations behind a 200-500ms AAD round-trip.
+  Azure::Core::Http::CurlTransport transport;
+  std::string mi_assertion;
+  std::pair<std::string, Azure::DateTime> exchanged;
+  try {
+    mi_assertion = FetchManagedIdentityAssertion(transport, context);
+    exchanged = ExchangeForStorageBearer(transport, mi_assertion, context);
+  } catch (const std::exception& e) {
+    throw Azure::Core::Credentials::AuthenticationException(std::string("AzureCrossTenantCredential failed: ") +
+                                                            e.what());
+  }
+
+  Azure::Core::Credentials::AccessToken fresh;
+  fresh.Token = std::move(exchanged.first);
+  fresh.ExpiresOn = exchanged.second;
+
+  {
+    std::lock_guard<std::mutex> g(cache_mu_);
+    cached_ = fresh;
+  }
+  return fresh;
+}
+
+std::string AzureCrossTenantCredential::FetchManagedIdentityAssertion(Azure::Core::Http::HttpTransport& transport,
+                                                                      Azure::Core::Context const& context) const {
+  Azure::Core::Url url(kImdsTokenUrl);
+  url.AppendQueryParameter("api-version", kImdsApiVersion);
+  url.AppendQueryParameter("resource", kMiAudience);
+
+  Azure::Core::Http::Request request(Azure::Core::Http::HttpMethod::Get, url);
+  request.SetHeader("Metadata", "true");
+
+  auto response = transport.Send(request, context);
+  if (!response) {
+    throw std::runtime_error("IMDS token request returned no response");
+  }
+  auto status = response->GetStatusCode();
+  if (status != Azure::Core::Http::HttpStatusCode::Ok) {
+    auto body = ReadResponseBodyAsString(*response, context);
+    throw std::runtime_error("IMDS token request failed status=" + std::to_string(static_cast<int>(status)) +
+                             " body=" + body);
+  }
+
+  auto body = ReadResponseBodyAsString(*response, context);
+  folly::dynamic parsed;
+  try {
+    parsed = folly::parseJson(body);
+  } catch (const std::exception& e) {
+    throw std::runtime_error(std::string("IMDS response was not valid JSON: ") + e.what());
+  }
+  auto access_token = parsed.getDefault("access_token", "").asString();
+  if (access_token.empty()) {
+    throw std::runtime_error("IMDS response missing access_token");
+  }
+  return access_token;
+}
+
+std::pair<std::string, Azure::DateTime> AzureCrossTenantCredential::ExchangeForStorageBearer(
+    Azure::Core::Http::HttpTransport& transport,
+    std::string const& mi_assertion,
+    Azure::Core::Context const& context) const {
+  // POST to https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token
+  // with form-urlencoded body. AAD requires:
+  //   client_id={customer_app}
+  //   scope=https://storage.azure.com/.default
+  //   grant_type=client_credentials
+  //   client_assertion={MI assertion JWT}
+  //   client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer
+  std::string token_url = std::string(kAadAuthority) + "/" + tenant_id_ + "/oauth2/v2.0/token";
+  Azure::Core::Url url(token_url);
+
+  std::string body_str;
+  body_str.reserve(2048);
+  body_str.append("client_id=").append(UrlEncode(client_id_));
+  body_str.append("&scope=").append(UrlEncode(kStorageScope));
+  body_str.append("&grant_type=client_credentials");
+  body_str.append("&client_assertion_type=").append(UrlEncode(kClientAssertionType));
+  body_str.append("&client_assertion=").append(UrlEncode(mi_assertion));
+
+  std::vector<uint8_t> body_bytes(body_str.begin(), body_str.end());
+  Azure::Core::IO::MemoryBodyStream body_stream(body_bytes.data(), body_bytes.size());
+
+  Azure::Core::Http::Request request(Azure::Core::Http::HttpMethod::Post, url, &body_stream);
+  request.SetHeader("Content-Type", "application/x-www-form-urlencoded");
+  request.SetHeader("Content-Length", std::to_string(body_bytes.size()));
+
+  auto response = transport.Send(request, context);
+  if (!response) {
+    throw std::runtime_error("AAD token exchange returned no response");
+  }
+  auto status = response->GetStatusCode();
+  if (status != Azure::Core::Http::HttpStatusCode::Ok) {
+    auto err_body = ReadResponseBodyAsString(*response, context);
+    throw std::runtime_error(
+        "AAD token exchange failed (the customer's App Registration likely has no "
+        "Federated Identity Credential trusting our MI, or the audience/issuer/subject "
+        "in the FIC don't match) tenant=" +
+        tenant_id_ + " client_id=" + client_id_ + " status=" + std::to_string(static_cast<int>(status)) +
+        " body=" + err_body);
+  }
+
+  auto resp_body = ReadResponseBodyAsString(*response, context);
+  folly::dynamic parsed;
+  try {
+    parsed = folly::parseJson(resp_body);
+  } catch (const std::exception& e) {
+    throw std::runtime_error(std::string("AAD response was not valid JSON: ") + e.what());
+  }
+  auto access_token = parsed.getDefault("access_token", "").asString();
+  if (access_token.empty()) {
+    throw std::runtime_error("AAD response missing access_token");
+  }
+  // expires_in is seconds-from-now. AAD always sends an int here for
+  // client_credentials grants.
+  int64_t expires_in = parsed.getDefault("expires_in", 3600).asInt();
+  auto expires_at = std::chrono::system_clock::now() + std::chrono::seconds(expires_in);
+  return {std::move(access_token), Azure::DateTime(expires_at)};
+}
+
+}  // namespace milvus_storage::fs
diff --git a/cpp/src/filesystem/azure/azure_fs_producer.cpp b/cpp/src/filesystem/azure/azure_fs_producer.cpp
@@ -51,7 +51,15 @@ arrow::Result<ArrowFileSystemPtr> AzureFileSystemProducer::Make() {
   }
   options.background_writes = config_.background_writes;
 
-  if (config_.use_iam) {
+  if (!config_.azure_client_id.empty() && !config_.azure_tenant_id.empty()) {
+    // Cross-tenant via Managed Identity. The customer's storage account is
+    // in their own tenant; we have no shared key. Our local IMDS-attached
+    // MI is exchanged at the customer's AAD authority for a Storage Bearer.
+    // See `AzureCrossTenantCredential` for the full two-hop protocol.
+    LOG_STORAGE_DEBUG_ << "Azure cross-tenant: tenant=" << config_.azure_tenant_id
+                       << " client=" << config_.azure_client_id;
+    ARROW_RETURN_NOT_OK(options.ConfigureCrossTenantCredential(config_.azure_tenant_id, config_.azure_client_id));
+  } else if (config_.use_iam) {
     const char* federated_token = getenv("AZURE_FEDERATED_TOKEN_FILE");
     if (federated_token != nullptr && strlen(federated_token) > 0) {
       // Workload Identity