[feature](storage) Support native GCP IAM authentication

be/CMakeLists.txt

@@ -692,7 +692,7 @@ set(DORIS_LINK_LIBS
 )
 
 set(absl_DIR ${THIRDPARTY_DIR}/lib/cmake/absl)
-find_package(absl)
+find_package(absl CONFIG REQUIRED PATHS ${absl_DIR} NO_DEFAULT_PATH)
 
 # COMMON_THIRDPARTY are thirdparty dependencies that can run on all platform
 # When adding new dependencies, If you don’t know if it can run on all platforms,
@@ -722,6 +722,14 @@ add_definitions(-DUSE_DORIS_HADOOP_HDFS)
 if (absl_FOUND)
     set(COMMON_THIRDPARTY
         ${COMMON_THIRDPARTY}
+        # Public dependencies of the statically linked google-cloud-cpp common library.
+        absl::base
+        absl::memory
+        absl::optional
+        absl::span
+        absl::str_format
+        absl::time
+        absl::variant
         absl::cord
         absl::cord_internal
         absl::cordz_functions

be/cmake/thirdparty.cmake

@@ -69,6 +69,9 @@ add_thirdparty(lz4)
 add_thirdparty(lzo2)
 add_thirdparty(thrift)
 add_thirdparty(thriftnb)
+add_thirdparty(google_cloud_cpp_oauth2)
+add_thirdparty(google_cloud_cpp_rest_internal)
+add_thirdparty(google_cloud_cpp_common)
 add_thirdparty(crc32c)
 
 add_thirdparty(libevent_core LIBNAME "lib/libevent_core.a")

be/src/io/fs/gcs_signed_url_provider.cpp

@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "io/fs/gcs_signed_url_provider.h"
+
+#include <fmt/format.h>
+#include <rapidjson/document.h>
+
+#include <exception>
+#include <string_view>
+#include <utility>
+
+#include "cpp/obj-client/auth/gcp/gcp_token_provider.h"
+#include "cpp/obj-client/auth/gcp/gcs_signed_url.h"
+#include "cpp/sync_point.h"
+#include "service/http/http_client.h"
+#include "util/url_coding.h"
+
+namespace doris::io {
+namespace {
+
+constexpr std::string_view IAM_CREDENTIALS_ENDPOINT =
+        "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/";
+
+std::string iam_error_message(const rapidjson::Document& document) {
+    if (!document.IsObject() || !document.HasMember("error") || !document["error"].IsObject()) {
+        return "unparseable error response";
+    }
+    const auto& error = document["error"];
+    std::string message;
+    if (error.HasMember("status") && error["status"].IsString()) {
+        message = error["status"].GetString();
+    }
+    if (error.HasMember("message") && error["message"].IsString()) {
+        if (!message.empty()) {
+            message.append(": ");
+        }
+        message.append(error["message"].GetString());
+    }
+    if (message.empty()) {
+        return "error response did not contain status or message";
+    }
+    constexpr size_t MAX_ERROR_MESSAGE_SIZE = 1024;
+    if (message.size() > MAX_ERROR_MESSAGE_SIZE) {
+        message.resize(MAX_ERROR_MESSAGE_SIZE);
+        message.append("...");
+    }
+    return message;
+}
+
+bool is_unreserved(unsigned char c) {
+    return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' ||
+           c == '_' || c == '.' || c == '~';
+}
+
+std::string percent_encode(std::string_view value) {
+    constexpr char HEX[] = "0123456789ABCDEF";
+    std::string encoded;
+    encoded.reserve(value.size());
+    for (unsigned char c : value) {
+        if (is_unreserved(c)) {
+            encoded.push_back(static_cast<char>(c));
+            continue;
+        }
+        encoded.push_back('%');
+        encoded.push_back(HEX[c >> 4]);
+        encoded.push_back(HEX[c & 0x0F]);
+    }
+    return encoded;
+}
+
+Status call_iam_sign_blob(std::string_view access_token, std::string_view service_account,
+                          std::string_view string_to_sign, int64_t request_timeout_ms,
+                          const std::string& ca_cert_file_path, std::string* signature) {
+    std::string encoded_payload;
+    base64_encode(std::string(string_to_sign), &encoded_payload);
+    std::string request_body = fmt::format(R"({{"payload":"{}"}})", encoded_payload);
+    std::string endpoint =
+            std::string(IAM_CREDENTIALS_ENDPOINT) + percent_encode(service_account) + ":signBlob";
+
+    HttpClient client;
+    // Keep the response body for non-2xx replies so IAM permission and
+    // service-account errors are actionable to operators.
+    RETURN_IF_ERROR(client.init(endpoint, false, HttpClient::AuthTokenMode::NONE));
+    if (!ca_cert_file_path.empty()) {
+        RETURN_IF_ERROR(client.set_ca_cert_file(ca_cert_file_path));
+    }
+    client.set_authorization("Bearer " + std::string(access_token));
+    client.set_content_type("application/json");
+    client.set_timeout_ms(request_timeout_ms > 0 ? request_timeout_ms : 10000);
+
+    TEST_SYNC_POINT_RETURN_WITH_VALUE("GcsV4Signer::sign_blob", Status::OK(), access_token,
+                                      service_account, string_to_sign, signature);
+    std::string response;
+    RETURN_IF_ERROR(client.execute_post_request(request_body, &response));
+
+    rapidjson::Document document;
+    document.Parse(response.data(), response.size());
+    const auto http_status = client.get_http_status();
+    if (http_status < 200 || http_status >= 300) {
+        return Status::HttpError("IAM signBlob failed with HTTP {}: {}", http_status,
+                                 iam_error_message(document));
+    }
+    if (document.HasParseError() || !document.IsObject() || !document.HasMember("signedBlob") ||
+        !document["signedBlob"].IsString()) {
+        return Status::InternalError("IAM signBlob returned an invalid response");
+    }
+    if (!base64_decode(document["signedBlob"].GetString(), signature)) {
+        return Status::InternalError("IAM signBlob returned an invalid base64 signature");
+    }
+    return Status::OK();
+}
+
+} // namespace
+
+Status resolve_gcs_signer_email(const GcpCredentialConfig& credential,
+                                const GcpTokenProvider& token_provider, std::string* signer_email) {
+    *signer_email = credential.impersonation_service_account.empty()
+                            ? token_provider.get_service_account_email()
+                            : credential.impersonation_service_account;
+    if (!is_valid_gcp_service_account_email(*signer_email)) {
+        return Status::InvalidArgument(
+                "GCS V4 signing requires a service account identity; configure "
+                "gs.impersonation_service_account for credentials without one");
+    }
+    return Status::OK();
+}
+
+Status generate_gcs_v4_signed_url(const GcsV4SignedUrlProviderOptions& options,
+                                  const GcpCredentialConfig& credential,
+                                  const std::shared_ptr<GcpTokenProvider>& token_provider,
+                                  std::string* signed_url) {
+    if (signed_url == nullptr) {
+        return Status::InvalidArgument("signed_url output must not be null");
+    }
+    signed_url->clear();
+
+    if (token_provider == nullptr) {
+        return Status::InternalError("GCS V4 signing token provider is not initialized");
+    }
+
+    try {
+        auto token = token_provider->get_token();
+        if (!token.has_value()) {
+            return Status::InternalError("failed to obtain OAuth token for IAM signBlob");
+        }
+
+        // Resolve identity from the same credentials that supplied the token. DEFAULT may
+        // select a VM, a key file, or another ADC source; never infer it from VM presence.
+        std::string signer_email;
+        RETURN_IF_ERROR(resolve_gcs_signer_email(credential, *token_provider, &signer_email));
+
+        auto result = build_gcs_v4_signed_url(
+                {.endpoint = options.endpoint,
+                 .bucket = options.bucket,
+                 .key = options.key,
+                 .signer_email = signer_email,
+                 .expiration_secs = options.expiration_secs},
+                std::chrono::system_clock::now(), [&](std::string_view string_to_sign) {
+                    std::string signature;
+                    auto status = call_iam_sign_blob(*token, signer_email, string_to_sign,
+                                                     options.request_timeout_ms,
+                                                     options.ca_cert_file_path, &signature);
+                    if (!status.ok()) {
+                        return GcsSignBlobResult {.error = status.to_string()};
+                    }
+                    return GcsSignBlobResult {.signature = std::move(signature)};
+                });
+        if (!result.ok()) {
+            return Status::InternalError("failed to build GCS V4 signed URL: {}", result.error);
+        }
+        *signed_url = std::move(result.signed_url);
+        return Status::OK();
+    } catch (const std::exception& e) {
+        return Status::InternalError("failed to initialize GCS URL signer: {}", e.what());
+    }
+}
+
+} // namespace doris::io

be/src/io/fs/gcs_signed_url_provider.h

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "common/status.h"
+#include "cpp/obj-client/auth/gcp/gcp_auth.h"
+
+namespace doris {
+class GcpTokenProvider;
+}
+
+namespace doris::io {
+
+struct GcsV4SignedUrlProviderOptions {
+    std::string endpoint;
+    std::string bucket;
+    std::string key;
+    int64_t expiration_secs = 0;
+    int64_t request_timeout_ms = 10000;
+    std::string ca_cert_file_path;
+};
+
+// Resolve the identity selected by ADC or explicit impersonation, without falling back
+// to an unrelated VM service account when the selected credentials have no identity.
+Status resolve_gcs_signer_email(const GcpCredentialConfig& credential,
+                                const GcpTokenProvider& token_provider, std::string* signer_email);
+
+// Resolves the signing service account, obtains an OAuth token, and invokes
+// IAM Credentials signBlob before delegating URL construction to common auth.
+Status generate_gcs_v4_signed_url(const GcsV4SignedUrlProviderOptions& options,
+                                  const GcpCredentialConfig& credential,
+                                  const std::shared_ptr<GcpTokenProvider>& token_provider,
+                                  std::string* signed_url);
+
+} // namespace doris::io

be/src/io/fs/s3_file_system.cpp

@@ -85,6 +85,7 @@ Status ObjClientHolder::reset(const S3ClientConf& conf) {
     {
         std::shared_lock lock(_mtx);
         reset_conf = _conf;
+        reset_conf.provider = conf.provider;
         reset_conf.ak = conf.ak;
         reset_conf.sk = conf.sk;
         reset_conf.token = conf.token;
@@ -98,6 +99,7 @@ Status ObjClientHolder::reset(const S3ClientConf& conf) {
         reset_conf.role_arn = conf.role_arn;
         reset_conf.external_id = conf.external_id;
         reset_conf.cred_provider_type = conf.cred_provider_type;
+        reset_conf.credential = conf.credential;
 
         // Compare full-field equality of the merged conf, not get_hash(): the hash is
         // an XOR of crc32s and distinct configurations can collide, which would skip a
@@ -444,7 +446,9 @@ Status S3FileSystem::download_impl(const Path& remote_file, const Path& local_fi
 // whether to return a public endpoint.
 std::string S3FileSystem::generate_presigned_url(const Path& path, int64_t expiration_secs,
                                                  bool is_public_endpoint) const {
-    std::string key = fmt::format("{}/{}", _prefix, path.native());
+    // An empty prefix must not add a separator absent from the uploaded object key.
+    std::string key =
+            _prefix.empty() ? path.native() : fmt::format("{}/{}", _prefix, path.native());
     std::shared_ptr<ObjStorageClient> client;
     if (is_public_endpoint &&
         _client->s3_client_conf().endpoint.ends_with(OSS_PRIVATE_ENDPOINT_SUFFIX)) {

be/src/runtime/runtime_state.cpp

@@ -599,6 +599,11 @@ std::string RuntimeState::get_error_log_file_path() {
     auto presigned_url =
             s3_error_fs->generate_presigned_url(remote_error_log_file_path, EXPIRATION_SECONDS,
                                                 config::use_public_endpoint_for_error_log);
+    if (presigned_url.empty()) {
+        LOG(WARNING) << "Fail to generate presigned URL for S3 error log; retain local path "
+                     << local_error_log_file_path;
+        return local_error_log_file_path;
+    }
     std::lock_guard<std::mutex> load_lock(_load_error_log_lock);
     _error_log_file_path = std::move(presigned_url);
     return _error_log_file_path;

be/src/service/http/http_client.cpp

@@ -34,6 +34,7 @@
 #include "common/cast_set.h"
 #include "common/config.h"
 #include "common/status.h"
+#include "cpp/sync_point.h"
 #include "io/fs/local_file_system.h"
 #include "runtime/cluster_info.h"
 #include "runtime/exec_env.h"
@@ -356,7 +357,8 @@ void HttpClient::reset() {
     _method = GET;
 }
 
-Status HttpClient::init(const std::string& url, bool set_fail_on_error) {
+Status HttpClient::init(const std::string& url, bool set_fail_on_error,
+                        AuthTokenMode auth_token_mode) {
     reset();
     if (_curl == nullptr) {
         _curl = curl_easy_init();
@@ -426,9 +428,20 @@ Status HttpClient::init(const std::string& url, bool set_fail_on_error) {
         return Status::InternalError("fail to set CURLOPT_URL");
     }
 
+    if (auth_token_mode == AuthTokenMode::CLUSTER) {
 #ifndef BE_TEST
-    set_auth_token(ExecEnv::GetInstance()->cluster_info()->curr_auth_token);
+        set_auth_token(ExecEnv::GetInstance()->cluster_info()->curr_auth_token);
 #endif
+    }
+    return Status::OK();
+}
+
+Status HttpClient::set_ca_cert_file(const std::string& path) {
+    auto code = curl_easy_setopt(_curl, CURLOPT_CAINFO, path.c_str());
+    if (code != CURLE_OK) {
+        return Status::InternalError("failed to set CA certificate file: {}", _to_errmsg(code));
+    }
+    TEST_SYNC_POINT_CALLBACK("HttpClient::set_ca_cert_file", &path);
     return Status::OK();
 }
 

be/src/service/http/http_client.h

@@ -44,6 +44,8 @@ struct HttpTransferMetrics {
 // Helper class to access HTTP resource
 class HttpClient {
 public:
+    enum class AuthTokenMode { CLUSTER, NONE };
+
     HttpClient();
     ~HttpClient();
 
@@ -56,7 +58,8 @@ class HttpClient {
 
     // this function must call before other function,
     // you can call this multiple times to reuse this object
-    Status init(const std::string& url, bool set_fail_on_error = true);
+    Status init(const std::string& url, bool set_fail_on_error = true,
+                AuthTokenMode auth_token_mode = AuthTokenMode::CLUSTER);
 
     // Release request-owned options and callbacks while retaining reusable connections.
     void reset();
@@ -95,7 +98,9 @@ class HttpClient {
         curl_easy_setopt(_curl, CURLOPT_COPYPOSTFIELDS, post_body.c_str());
     }
 
-    // Currently, only fake SSL configurations are supported
+    // Override the trusted CA bundle after init(), which resets curl options.
+    Status set_ca_cert_file(const std::string& path);
+
     void use_untrusted_ssl() {
         curl_easy_setopt(_curl, CURLOPT_SSL_VERIFYPEER, 0L);
         curl_easy_setopt(_curl, CURLOPT_SSL_VERIFYHOST, 0L);