This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 6a2c90f0 feat(rest): support DLF authentication (#244)
6a2c90f0 is described below
commit 6a2c90f01a9029d037f29a82950811e30563859b
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Aug 25 13:14:48 2026 +0800
feat(rest): support DLF authentication (#244)
---
CMakeLists.txt | 6 +-
cmake_modules/arrow.diff | 7 +
cmake_modules/orc.diff | 9 +
docs/source/building.rst | 3 +-
docs/source/user_guide/catalog.rst | 42 +-
include/paimon/catalog_options.h | 31 +-
src/paimon/CMakeLists.txt | 8 +
src/paimon/common/catalog_options.cpp | 9 +
src/paimon/common/utils/http_client.cpp | 5 +
src/paimon/common/utils/http_client.h | 2 +
src/paimon/common/utils/options_utils.h | 13 +
src/paimon/common/utils/options_utils_test.cpp | 20 +
src/paimon/rest/dlf_auth.cpp | 811 +++++++++++++++++++++++++
src/paimon/rest/dlf_auth.h | 208 +++++++
src/paimon/rest/dlf_auth_test.cpp | 468 ++++++++++++++
src/paimon/rest/rest_api.cpp | 31 +-
src/paimon/rest/rest_api.h | 8 +-
src/paimon/rest/rest_auth.cpp | 12 +-
src/paimon/rest/rest_auth.h | 5 +
src/paimon/rest/rest_catalog_test.cpp | 8 +-
src/paimon/rest/rest_http_client.cpp | 13 +-
src/paimon/rest/rest_http_client.h | 12 +-
src/paimon/rest/rest_http_client_test.cpp | 19 +
23 files changed, 1716 insertions(+), 34 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index e99034dc..084e6bf0 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -68,7 +68,8 @@ option(PAIMON_ENABLE_LUMINA "Whether to enable lumina vector
index" OFF)
option(PAIMON_ENABLE_LUCENE "Whether to enable lucene index" OFF)
option(PAIMON_ENABLE_TANTIVY
"Whether to enable tantivy-fulltext global index (Rust FFI,
experimental)" OFF)
-option(PAIMON_ENABLE_REST "Whether to enable the rest catalog (requires
libcurl)" OFF)
+option(PAIMON_ENABLE_REST
+ "Whether to enable the rest catalog (requires libcurl and OpenSSL)" OFF)
if(PAIMON_ENABLE_ORC)
add_definitions(-DPAIMON_ENABLE_ORC)
endif()
@@ -79,6 +80,9 @@ endif()
if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
find_package(CURL REQUIRED)
endif()
+if(PAIMON_ENABLE_REST)
+ find_package(OpenSSL 1.1.0 REQUIRED)
+endif()
if(PAIMON_ENABLE_AVRO)
add_definitions(-DPAIMON_ENABLE_AVRO)
endif()
diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff
index b8b83517..bb71e9c3 100644
--- a/cmake_modules/arrow.diff
+++ b/cmake_modules/arrow.diff
@@ -15,6 +15,13 @@ diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake
b/cpp/cmake_modules/Thi
index 8cb3ec83f5..0765df8fa8 100644
--- a/cpp/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake
+@@ -814,5 +814,6 @@ if(DEFINED ENV{ARROW_THRIFT_URL})
+ set(THRIFT_SOURCE_URL "$ENV{ARROW_THRIFT_URL}")
+ else()
+ set_urls(THRIFT_SOURCE_URL
++
"https://archive.apache.org/dist/thrift/${ARROW_THRIFT_BUILD_VERSION}/thrift-${ARROW_THRIFT_BUILD_VERSION}.tar.gz";
+
"https://www.apache.org/dyn/closer.cgi?action=download&filename=/thrift/${ARROW_THRIFT_BUILD_VERSION}/thrift-${ARROW_THRIFT_BUILD_VERSION}.tar.gz";
+
"https://downloads.apache.org/thrift/${ARROW_THRIFT_BUILD_VERSION}/thrift-${ARROW_THRIFT_BUILD_VERSION}.tar.gz";
@@ -983,6 +983,11 @@ if(CMAKE_TOOLCHAIN_FILE)
list(APPEND EP_COMMON_CMAKE_ARGS
-DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE})
endif()
diff --git a/cmake_modules/orc.diff b/cmake_modules/orc.diff
index e4ca4e29..88742dbd 100644
--- a/cmake_modules/orc.diff
+++ b/cmake_modules/orc.diff
@@ -435,3 +435,12 @@ index 9b2c829c7..434841224 100644
set(ORC_FORMAT_VERSION "1.0.0")
set(LZ4_VERSION "1.10.0")
set(SNAPPY_VERSION "1.2.1")
+@@ -140,7 +142,7 @@ if(DEFINED ENV{ORC_FORMAT_URL})
+ set(ORC_FORMAT_SOURCE_URL "$ENV{ORC_FORMAT_URL}")
+ message(STATUS "Using ORC_FORMAT_URL: ${ORC_FORMAT_SOURCE_URL}")
+ else()
+- set(ORC_FORMAT_SOURCE_URL
"https://www.apache.org/dyn/closer.lua/orc/orc-format-${ORC_FORMAT_VERSION}/orc-format-${ORC_FORMAT_VERSION}.tar.gz?action=download";
)
++ set(ORC_FORMAT_SOURCE_URL
"https://archive.apache.org/dist/orc/orc-format-${ORC_FORMAT_VERSION}/orc-format-${ORC_FORMAT_VERSION}.tar.gz";
)
+ message(STATUS "Using DEFAULT URL: ${ORC_FORMAT_SOURCE_URL}")
+ endif()
+ ExternalProject_Add (orc-format_ep
diff --git a/docs/source/building.rst b/docs/source/building.rst
index 466d461b..32c7e67c 100644
--- a/docs/source/building.rst
+++ b/docs/source/building.rst
@@ -182,7 +182,8 @@ boolean flags to ``cmake``.
Linux ``x86_64``; see :ref:`cpp-building-platforms`.
* ``-DPAIMON_ENABLE_LUCENE=ON``: Support for Lucene full-text search indexes
* ``-DPAIMON_ENABLE_TANTIVY=ON``: Enable the experimental Tantivy full-text
index Rust FFI.
-* ``-DPAIMON_ENABLE_REST=ON``: Support for the REST catalog
(``metastore=rest``), requires the libcurl development package.
+* ``-DPAIMON_ENABLE_REST=ON``: Support for the REST catalog
+ (``metastore=rest``), requires the libcurl and OpenSSL development packages.
Third-party dependency source
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/docs/source/user_guide/catalog.rst
b/docs/source/user_guide/catalog.rst
index c57c50c1..b695044d 100644
--- a/docs/source/user_guide/catalog.rst
+++ b/docs/source/user_guide/catalog.rst
@@ -50,9 +50,25 @@ registered on the REST server. The catalog is configured
through the
* ``metastore``: must be ``rest`` to select the REST catalog.
* ``uri``: server url of the REST catalog server.
-* ``token.provider``: authentication provider of the REST catalog; currently
only
- ``bear`` is supported (the protocol's historical spelling of "bearer").
+* ``token.provider``: authentication provider of the REST catalog. ``bear``
+ (the protocol's historical spelling of "bearer") and ``dlf`` are supported.
* ``token``: token of the ``bear`` token provider.
+* ``dlf.region``: region used by DLF request signing. It is inferred from the
+ endpoint URI when omitted.
+* ``dlf.access-key-id`` and ``dlf.access-key-secret``: static DLF access key.
+* ``dlf.security-token``: optional STS security token used with a static access
+ key.
+* ``dlf.token-path``: path to a JSON file containing refreshable DLF
credentials.
+* ``dlf.token-loader``: refreshable credential loader. ``local_file`` reads
+ ``dlf.token-path`` and ``ecs`` obtains an STS token from an ECS RAM role.
+* ``dlf.token-ecs-metadata-url``: ECS RAM role metadata endpoint. It defaults
to
+ ``http://100.100.100.200/latest/meta-data/Ram/security-credentials/``.
+* ``dlf.token-ecs-role-name``: optional ECS RAM role name. The loader discovers
+ the role from the metadata endpoint when it is omitted.
+* ``dlf.signing-algorithm``: ``default`` selects DLF4-HMAC-SHA256 for DLF VPC
+ endpoints and ``openapi`` selects ROA HMAC-SHA1 for DlfNext OpenAPI
endpoints.
+ When omitted, an endpoint containing ``dlfnext`` selects ``openapi`` and
other
+ endpoints select ``default``.
* ``table-default.<key>``: table option defaults applied when a created table
left ``<key>`` unset.
* ``header.<name>``: sent as the ``<name>`` http header on every request to the
@@ -70,6 +86,25 @@ registered on the REST server. The catalog is configured
through the
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::Catalog> catalog,
paimon::Catalog::Create(/*root_path=*/"my_instance",
options));
+For DLF, configure one credential source. Static AK/SK credentials, an optional
+STS token, a refreshable local token file, and ECS RAM role credentials are
+supported. A local or ECS token has the Java-compatible JSON fields
+``AccessKeyId``, ``AccessKeySecret``, ``SecurityToken`` and ``Expiration``. The
+last field uses UTC ``yyyy-MM-dd'T'HH:mm:ss'Z'`` format. Refreshable
credentials
+are reloaded when less than one hour of validity remains.
+
+.. code-block:: cpp
+
+ std::map<std::string, std::string> options = {
+ {"metastore", "rest"},
+ {"uri", "https://dlfnext.cn-hangzhou.aliyuncs.com"},
+ {"token.provider", "dlf"},
+ {"dlf.access-key-id", "<access-key-id>"},
+ {"dlf.access-key-secret", "<access-key-secret>"},
+ // Optional for temporary credentials:
+ {"dlf.security-token", "<security-token>"},
+ };
+
On creation the catalog queries the server's ``/v1/config`` endpoint and merges
its response with the options above: the server's overrides win over the client
options, which in turn win over the server's defaults.
@@ -81,5 +116,4 @@ through the regular ``Catalog`` API, and table snapshots can
be listed through
The C++ REST catalog covers the database, table and snapshot operations of the
``Catalog`` API. The parts of the Java REST catalog that have no C++
counterpart
yet — altering a database or a table, views, functions, partitions, tags,
branch
-management and consumers — are not supported, and neither is the ``dlf`` token
-provider.
+management and consumers — are not supported.
diff --git a/include/paimon/catalog_options.h b/include/paimon/catalog_options.h
index f58a876c..e959483d 100644
--- a/include/paimon/catalog_options.h
+++ b/include/paimon/catalog_options.h
@@ -32,10 +32,37 @@ struct PAIMON_EXPORT CatalogOptions {
/// "token" - Token of the "bear" token provider of the REST catalog.
static const char TOKEN[];
- /// "token.provider" - Authentication provider of the REST catalog. Only
"bear" is
- /// supported ("bear" is the protocol's historical spelling of "bearer",
do not "fix" it).
+ /// "token.provider" - Authentication provider of the REST catalog.
Supported values are
+ /// "bear" (the protocol's historical spelling of "bearer") and "dlf".
static const char TOKEN_PROVIDER[];
+ /// "dlf.region" - Region used by DLF request signing. Inferred from URI
when absent.
+ static const char DLF_REGION[];
+
+ /// "dlf.token-path" - Path of a JSON file containing refreshable DLF
credentials.
+ static const char DLF_TOKEN_PATH[];
+
+ /// "dlf.access-key-id" - DLF access key id.
+ static const char DLF_ACCESS_KEY_ID[];
+
+ /// "dlf.access-key-secret" - DLF access key secret.
+ static const char DLF_ACCESS_KEY_SECRET[];
+
+ /// "dlf.security-token" - Optional STS security token used with a DLF
access key.
+ static const char DLF_SECURITY_TOKEN[];
+
+ /// "dlf.token-loader" - Refreshable DLF token loader ("ecs" or
"local_file").
+ static const char DLF_TOKEN_LOADER[];
+
+ /// "dlf.token-ecs-metadata-url" - ECS RAM role metadata endpoint.
+ static const char DLF_TOKEN_ECS_METADATA_URL[];
+
+ /// "dlf.token-ecs-role-name" - Optional ECS RAM role name.
+ static const char DLF_TOKEN_ECS_ROLE_NAME[];
+
+ /// "dlf.signing-algorithm" - DLF signer ("default" or "openapi").
+ static const char DLF_SIGNING_ALGORITHM[];
+
/// "table-default." - Prefix of the catalog options that provide table
option
/// defaults: "table-default.<key>=<value>" applies "<key>=<value>" to a
created
/// table when the caller left "<key>" unset.
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index a9810424..8b3a2953 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -196,6 +196,10 @@ if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
list(APPEND PAIMON_COMMON_SRCS common/utils/http_client.cpp)
set(PAIMON_CURL_LINK_LIBS CURL::libcurl)
endif()
+set(PAIMON_REST_LINK_LIBS)
+if(PAIMON_ENABLE_REST)
+ set(PAIMON_REST_LINK_LIBS OpenSSL::Crypto)
+endif()
if(PAIMON_ENABLE_S3)
list(APPEND PAIMON_COMMON_SRCS common/fs/object_store_file_system.cpp)
endif()
@@ -449,6 +453,7 @@ if(PAIMON_ENABLE_REST)
rest/resource_paths.cpp
rest/rest_api.cpp
rest/rest_auth.cpp
+ rest/dlf_auth.cpp
rest/rest_catalog.cpp
rest/rest_messages.cpp
rest/rest_util.cpp)
@@ -468,6 +473,7 @@ add_paimon_lib(paimon
Threads::Threads
RapidJSON
${PAIMON_CURL_LINK_LIBS}
+ ${PAIMON_REST_LINK_LIBS}
DataSketches
STATIC_LINK_LIBS
arrow
@@ -480,6 +486,7 @@ add_paimon_lib(paimon
RapidJSON
DataSketches
${PAIMON_CURL_LINK_LIBS}
+ ${PAIMON_REST_LINK_LIBS}
SHARED_LINK_FLAGS
${PAIMON_VERSION_SCRIPT_FLAGS})
@@ -965,6 +972,7 @@ if(PAIMON_BUILD_TESTS)
rest/rest_http_client_test.cpp
rest/mock_rest_server.cpp
rest/resource_paths_test.cpp
+ rest/dlf_auth_test.cpp
rest/rest_catalog_test.cpp
rest/rest_messages_test.cpp
rest/rest_util_test.cpp
diff --git a/src/paimon/common/catalog_options.cpp
b/src/paimon/common/catalog_options.cpp
index 6e89e80a..2722f06b 100644
--- a/src/paimon/common/catalog_options.cpp
+++ b/src/paimon/common/catalog_options.cpp
@@ -22,6 +22,15 @@ const char CatalogOptions::METASTORE[] = "metastore";
const char CatalogOptions::URI[] = "uri";
const char CatalogOptions::TOKEN[] = "token";
const char CatalogOptions::TOKEN_PROVIDER[] = "token.provider";
+const char CatalogOptions::DLF_REGION[] = "dlf.region";
+const char CatalogOptions::DLF_TOKEN_PATH[] = "dlf.token-path";
+const char CatalogOptions::DLF_ACCESS_KEY_ID[] = "dlf.access-key-id";
+const char CatalogOptions::DLF_ACCESS_KEY_SECRET[] = "dlf.access-key-secret";
+const char CatalogOptions::DLF_SECURITY_TOKEN[] = "dlf.security-token";
+const char CatalogOptions::DLF_TOKEN_LOADER[] = "dlf.token-loader";
+const char CatalogOptions::DLF_TOKEN_ECS_METADATA_URL[] =
"dlf.token-ecs-metadata-url";
+const char CatalogOptions::DLF_TOKEN_ECS_ROLE_NAME[] =
"dlf.token-ecs-role-name";
+const char CatalogOptions::DLF_SIGNING_ALGORITHM[] = "dlf.signing-algorithm";
const char CatalogOptions::TABLE_DEFAULT_OPTION_PREFIX[] = "table-default.";
} // namespace paimon
diff --git a/src/paimon/common/utils/http_client.cpp
b/src/paimon/common/utils/http_client.cpp
index e61c7faa..1542f58a 100644
--- a/src/paimon/common/utils/http_client.cpp
+++ b/src/paimon/common/utils/http_client.cpp
@@ -164,6 +164,9 @@ CurlHttpClient::~CurlHttpClient() = default;
Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
const HttpBodyConsumer& consumer)
const {
+ if (request.request_timeout_ms < 0) {
+ return Status::Invalid("HTTP request timeout must not be negative");
+ }
for (int32_t attempt = 0; attempt < kMaxAttempts; ++attempt) {
CURL* handle = impl_->Acquire();
if (handle == nullptr) {
@@ -184,6 +187,8 @@ Result<HttpResponse> CurlHttpClient::Execute(const
HttpRequest& request,
curl_easy_setopt(handle, CURLOPT_HTTPHEADER, headers);
curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT_MS, 30000L);
+ curl_easy_setopt(handle, CURLOPT_TIMEOUT_MS,
+ static_cast<long>(request.request_timeout_ms)); //
NOLINT(runtime/int)
curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, WriteCallback);
curl_easy_setopt(handle, CURLOPT_WRITEDATA, &context);
curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION, HeaderCallback);
diff --git a/src/paimon/common/utils/http_client.h
b/src/paimon/common/utils/http_client.h
index 0dd3e742..8b04a700 100644
--- a/src/paimon/common/utils/http_client.h
+++ b/src/paimon/common/utils/http_client.h
@@ -46,6 +46,8 @@ struct HttpRequest {
HttpMethod method = HttpMethod::GET;
std::string url;
HttpHeaders headers;
+ /// Overall request timeout in milliseconds; zero keeps libcurl's
no-timeout default.
+ int32_t request_timeout_ms = 0;
};
struct HttpResponse {
diff --git a/src/paimon/common/utils/options_utils.h
b/src/paimon/common/utils/options_utils.h
index 90b30b54..c2014007 100644
--- a/src/paimon/common/utils/options_utils.h
+++ b/src/paimon/common/utils/options_utils.h
@@ -76,6 +76,19 @@ class OptionsUtils {
return value.value();
}
+ template <typename T>
+ static Result<std::optional<T>> GetOptionalValueFromMap(
+ const std::map<std::string, std::string>& key_value_map, const
std::string& key) {
+ Result<T> value = GetValueFromMap<T>(key_value_map, key);
+ if (value.ok()) {
+ return std::optional<T>(value.value());
+ }
+ if (value.status().IsNotExist()) {
+ return std::optional<T>();
+ }
+ return value.status();
+ }
+
/// Fetch options with specific prefix and remove prefix for key.
static std::map<std::string, std::string> FetchOptionsWithPrefix(
const std::string& prefix, const std::map<std::string, std::string>&
options) {
diff --git a/src/paimon/common/utils/options_utils_test.cpp
b/src/paimon/common/utils/options_utils_test.cpp
index 7a09a985..d4641184 100644
--- a/src/paimon/common/utils/options_utils_test.cpp
+++ b/src/paimon/common/utils/options_utils_test.cpp
@@ -63,6 +63,26 @@ TEST(OptionsUtilsTest, TestGetValueFromMap) {
ASSERT_EQ(999, empty);
}
+TEST(OptionsUtilsTest, TestGetOptionalValueFromMap) {
+ const std::map<std::string, std::string> key_value_map = {
+ {"key_int", "10"}, {"key_empty", ""}, {"key_invalid", "ab"}};
+
+ ASSERT_OK_AND_ASSIGN(std::optional<int32_t> optional_value,
+
OptionsUtils::GetOptionalValueFromMap<int32_t>(key_value_map, "key_int"));
+ ASSERT_EQ(std::optional<int32_t>(10), optional_value);
+ ASSERT_OK_AND_ASSIGN(
+ std::optional<int32_t> optional_missing,
+ OptionsUtils::GetOptionalValueFromMap<int32_t>(key_value_map,
"key_nonexist"));
+ ASSERT_EQ(std::nullopt, optional_missing);
+ ASSERT_OK_AND_ASSIGN(
+ std::optional<std::string> optional_empty,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(key_value_map,
"key_empty"));
+ ASSERT_EQ(std::optional<std::string>(""), optional_empty);
+ ASSERT_TRUE(OptionsUtils::GetOptionalValueFromMap<int64_t>(key_value_map,
"key_invalid")
+ .status()
+ .IsInvalid());
+}
+
TEST(OptionsUtilsTest, TestFetchOptionsWithPrefix) {
std::map<std::string, std::string> options = {{"key1", "value1"},
{"test.key2", "value2"}};
auto new_options = OptionsUtils::FetchOptionsWithPrefix("test.", options);
diff --git a/src/paimon/rest/dlf_auth.cpp b/src/paimon/rest/dlf_auth.cpp
new file mode 100644
index 00000000..4c592f31
--- /dev/null
+++ b/src/paimon/rest/dlf_auth.cpp
@@ -0,0 +1,811 @@
+/*
+ * 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 "paimon/rest/dlf_auth.h"
+
+#include <openssl/evp.h>
+
+#include <array>
+#include <cctype>
+#include <climits>
+#include <ctime>
+#include <fstream>
+#include <iomanip>
+#include <limits>
+#include <regex>
+#include <set>
+#include <sstream>
+#include <string_view>
+#include <thread>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/catalog_options.h"
+#include "paimon/common/utils/options_utils.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/common/utils/url_utils.h"
+#include "paimon/common/utils/uuid.h"
+#include "paimon/rest/rest_http_client.h"
+#include "rapidjson/document.h"
+
+namespace paimon {
+
+namespace {
+
+constexpr int32_t kEcsMetadataRequestTimeoutMillis = 3 * 60 * 1000;
+
+constexpr int64_t kTokenExpirationSafeTimeMillis = 60 * 60 * 1000;
+constexpr size_t kMaxTokenResponseBytes = 1024 * 1024;
+constexpr const char kDefaultEcsMetadataUrl[] =
+ "http://100.100.100.200/latest/meta-data/Ram/security-credentials/";;
+
+constexpr const char kAuthorizationHeader[] = "Authorization";
+constexpr const char kContentMd5Header[] = "Content-MD5";
+constexpr const char kContentTypeHeader[] = "Content-Type";
+constexpr const char kDlfDateHeader[] = "x-dlf-date";
+constexpr const char kDlfSecurityTokenHeader[] = "x-dlf-security-token";
+constexpr const char kDlfVersionHeader[] = "x-dlf-version";
+constexpr const char kDlfContentSha256Header[] = "x-dlf-content-sha256";
+constexpr const char kUnsignedPayload[] = "UNSIGNED-PAYLOAD";
+constexpr const char kJsonMediaType[] = "application/json";
+
+constexpr const char kOpenApiDateHeader[] = "Date";
+constexpr const char kOpenApiAcceptHeader[] = "Accept";
+constexpr const char kOpenApiHostHeader[] = "Host";
+constexpr const char kAcsSignatureMethodHeader[] = "x-acs-signature-method";
+constexpr const char kAcsSignatureNonceHeader[] = "x-acs-signature-nonce";
+constexpr const char kAcsSignatureVersionHeader[] = "x-acs-signature-version";
+constexpr const char kAcsVersionHeader[] = "x-acs-version";
+constexpr const char kAcsSecurityTokenHeader[] = "x-acs-security-token";
+
+void TrimWhitespace(std::string* value) {
+ size_t begin = 0;
+ while (begin < value->size() && std::isspace(static_cast<unsigned
char>((*value)[begin]))) {
+ ++begin;
+ }
+ size_t end = value->size();
+ while (end > begin && std::isspace(static_cast<unsigned char>((*value)[end
- 1]))) {
+ --end;
+ }
+ *value = value->substr(begin, end - begin);
+}
+
+Result<std::string> RequiredNonEmptyOption(const std::map<std::string,
std::string>& options,
+ const std::string& key) {
+ Result<std::string> value =
OptionsUtils::GetValueFromMap<std::string>(options, key);
+ if (!value.ok()) {
+ if (!value.status().IsNotExist()) {
+ return value.status();
+ }
+ return Status::Invalid(fmt::format("option '{}' must be configured for
DLF auth", key));
+ }
+ if (value.value().empty()) {
+ return Status::Invalid(fmt::format("option '{}' must be configured for
DLF auth", key));
+ }
+ return value.value();
+}
+
+Result<std::string> RequiredJsonString(const rapidjson::Value& object, const
char* key) {
+ if (!object.HasMember(key) || !object[key].IsString() ||
object[key].GetStringLength() == 0) {
+ return Status::Invalid(fmt::format("DLF token field '{}' must be a
non-empty string", key));
+ }
+ return std::string(object[key].GetString(), object[key].GetStringLength());
+}
+
+Result<std::optional<std::string>> OptionalJsonString(const rapidjson::Value&
object,
+ const char* key) {
+ if (!object.HasMember(key) || object[key].IsNull()) {
+ return std::optional<std::string>();
+ }
+ if (!object[key].IsString()) {
+ return Status::Invalid(fmt::format("DLF token field '{}' must be a
string", key));
+ }
+ return std::optional<std::string>(
+ std::string(object[key].GetString(), object[key].GetStringLength()));
+}
+
+Result<std::tm> ToUtc(std::chrono::system_clock::time_point time) {
+ std::time_t seconds = std::chrono::system_clock::to_time_t(time);
+ std::tm utc{};
+ if (gmtime_r(&seconds, &utc) == nullptr) {
+ return Status::Invalid("failed to convert DLF signing time to UTC");
+ }
+ return utc;
+}
+
+Result<std::string> FormatDlfTime(std::chrono::system_clock::time_point time) {
+ PAIMON_ASSIGN_OR_RAISE(std::tm utc, ToUtc(time));
+ std::array<char, 32> buffer{};
+ if (std::strftime(buffer.data(), buffer.size(), "%Y%m%dT%H%M%SZ", &utc) ==
0) {
+ return Status::Invalid("failed to format DLF signing time");
+ }
+ return std::string(buffer.data());
+}
+
+Result<std::string> FormatRfc1123Time(std::chrono::system_clock::time_point
time) {
+ PAIMON_ASSIGN_OR_RAISE(std::tm utc, ToUtc(time));
+ static constexpr std::array<const char*, 7> kWeekdays = {"Sun", "Mon",
"Tue", "Wed",
+ "Thu", "Fri",
"Sat"};
+ static constexpr std::array<const char*, 12> kMonths = {
+ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct",
"Nov", "Dec"};
+ if (utc.tm_wday < 0 || utc.tm_wday >=
static_cast<int32_t>(kWeekdays.size()) ||
+ utc.tm_mon < 0 || utc.tm_mon >= static_cast<int32_t>(kMonths.size())) {
+ return Status::Invalid("failed to format DLF OpenAPI signing time");
+ }
+ return fmt::format("{}, {:02d} {} {:04d} {:02d}:{:02d}:{:02d} GMT",
kWeekdays[utc.tm_wday],
+ utc.tm_mday, kMonths[utc.tm_mon], utc.tm_year + 1900,
utc.tm_hour,
+ utc.tm_min, utc.tm_sec);
+}
+
+Result<int64_t> ParseExpiration(const std::string& expiration) {
+ static const std::regex kExpirationPattern(
+ "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$");
+ if (!std::regex_match(expiration, kExpirationPattern)) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+ std::tm utc{};
+ std::istringstream stream(expiration);
+ stream >> std::get_time(&utc, "%Y-%m-%dT%H:%M:%SZ");
+ if (stream.fail() || stream.peek() != std::char_traits<char>::eof()) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+ int32_t year = utc.tm_year;
+ int32_t month = utc.tm_mon;
+ int32_t day = utc.tm_mday;
+ int32_t hour = utc.tm_hour;
+ int32_t minute = utc.tm_min;
+ int32_t second = utc.tm_sec;
+ std::time_t timestamp = timegm(&utc);
+ std::tm verified{};
+ if (timestamp == static_cast<std::time_t>(-1) || gmtime_r(×tamp,
&verified) == nullptr) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+ if (verified.tm_year != year || verified.tm_mon != month ||
verified.tm_mday != day ||
+ verified.tm_hour != hour || verified.tm_min != minute ||
verified.tm_sec != second) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+ if (timestamp > std::numeric_limits<int64_t>::max() / 1000) {
+ return Status::Invalid("DLF token expiration is out of range");
+ }
+ return static_cast<int64_t>(timestamp) * 1000;
+}
+
+using Bytes = std::vector<uint8_t>;
+using EvpMdContext = std::unique_ptr<EVP_MD_CTX, decltype(&EVP_MD_CTX_free)>;
+using EvpPkey = std::unique_ptr<EVP_PKEY, decltype(&EVP_PKEY_free)>;
+
+Result<Bytes> Digest(const EVP_MD* digest, std::string_view data) {
+ EvpMdContext context(EVP_MD_CTX_new(), EVP_MD_CTX_free);
+ if (!context || EVP_DigestInit_ex(context.get(), digest, nullptr) != 1 ||
+ EVP_DigestUpdate(context.get(), data.data(), data.size()) != 1) {
+ return Status::IOError("failed to calculate DLF request digest");
+ }
+ Bytes output(EVP_MAX_MD_SIZE);
+ unsigned int output_size = 0;
+ if (EVP_DigestFinal_ex(context.get(), output.data(), &output_size) != 1) {
+ return Status::IOError("failed to calculate DLF request digest");
+ }
+ output.resize(output_size);
+ return output;
+}
+
+Result<Bytes> Hmac(const EVP_MD* digest, const Bytes& key, std::string_view
data) {
+ if (key.size() > static_cast<size_t>(INT_MAX)) {
+ return Status::Invalid("DLF signing key is too large");
+ }
+ EvpPkey signing_key(
+ EVP_PKEY_new_mac_key(EVP_PKEY_HMAC, nullptr, key.data(),
static_cast<int32_t>(key.size())),
+ EVP_PKEY_free);
+ EvpMdContext context(EVP_MD_CTX_new(), EVP_MD_CTX_free);
+ if (!signing_key || !context ||
+ EVP_DigestSignInit(context.get(), nullptr, digest, nullptr,
signing_key.get()) != 1 ||
+ EVP_DigestSignUpdate(context.get(), data.data(), data.size()) != 1) {
+ return Status::IOError("failed to calculate DLF request signature");
+ }
+ size_t output_size = 0;
+ if (EVP_DigestSignFinal(context.get(), nullptr, &output_size) != 1) {
+ return Status::IOError("failed to calculate DLF request signature");
+ }
+ Bytes output(output_size);
+ if (EVP_DigestSignFinal(context.get(), output.data(), &output_size) != 1) {
+ return Status::IOError("failed to calculate DLF request signature");
+ }
+ output.resize(output_size);
+ return output;
+}
+
+Bytes ToBytes(const std::string& value) {
+ return Bytes(value.begin(), value.end());
+}
+
+std::string HexEncode(const Bytes& value) {
+ static constexpr char kHex[] = "0123456789abcdef";
+ std::string encoded;
+ encoded.reserve(value.size() * 2);
+ for (uint8_t byte : value) {
+ encoded.push_back(kHex[byte >> 4]);
+ encoded.push_back(kHex[byte & 0x0f]);
+ }
+ return encoded;
+}
+
+Result<std::string> Base64Encode(const Bytes& value) {
+ if (value.size() > static_cast<size_t>(INT_MAX)) {
+ return Status::Invalid("DLF digest is too large to encode");
+ }
+ size_t capacity = 4 * ((value.size() + 2) / 3) + 1;
+ std::string encoded(capacity, '\0');
+ int32_t size = EVP_EncodeBlock(reinterpret_cast<unsigned
char*>(encoded.data()), value.data(),
+ static_cast<int32_t>(value.size()));
+ if (size < 0) {
+ return Status::IOError("failed to encode DLF request digest");
+ }
+ encoded.resize(static_cast<size_t>(size));
+ return encoded;
+}
+
+Result<std::string> Md5Base64(const std::string& value) {
+ PAIMON_ASSIGN_OR_RAISE(Bytes digest, Digest(EVP_md5(), value));
+ return Base64Encode(digest);
+}
+
+std::string Trimmed(const std::string& value) {
+ std::string trimmed = value;
+ TrimWhitespace(&trimmed);
+ return trimmed;
+}
+
+std::string DefaultCanonicalRequest(const RestAuthParameter& parameter,
+ const DlfRequestSigner::Headers& headers) {
+ std::string canonical = parameter.method + "\n" + parameter.resource_path
+ "\n";
+ bool first = true;
+ for (const auto& [key, value] : parameter.parameters) {
+ if (!first) {
+ canonical += "&";
+ }
+ canonical += Trimmed(key);
+ if (!value.empty()) {
+ canonical += "=" + Trimmed(value);
+ }
+ first = false;
+ }
+
+ static const std::set<std::string> kSignedHeaders = {
+ "content-md5", "content-type", "x-dlf-content-sha256",
+ "x-dlf-date", "x-dlf-version", "x-dlf-security-token"};
+ std::map<std::string, std::string> sorted_headers;
+ for (const auto& [key, value] : headers) {
+ std::string lower_key = StringUtils::ToLowerCase(key);
+ if (kSignedHeaders.count(lower_key) > 0) {
+ sorted_headers[lower_key] = Trimmed(value);
+ }
+ }
+ for (const auto& [key, value] : sorted_headers) {
+ canonical += "\n" + key + ":" + value;
+ }
+ auto content_iter = headers.find(kDlfContentSha256Header);
+ std::string content_sha =
+ content_iter == headers.end() ? std::string(kUnsignedPayload) :
content_iter->second;
+ return canonical + "\n" + content_sha;
+}
+
+Result<std::string> RequiredHeader(const DlfRequestSigner::Headers& headers,
+ const std::string& name) {
+ auto iter = headers.find(name);
+ if (iter == headers.end() || iter->second.empty()) {
+ return Status::Invalid(fmt::format("DLF signing header '{}' is
missing", name));
+ }
+ return iter->second;
+}
+
+std::string OpenApiCanonicalizedHeaders(const DlfRequestSigner::Headers&
headers) {
+ std::map<std::string, std::string> sorted;
+ for (const auto& [key, value] : headers) {
+ std::string lower_key = StringUtils::ToLowerCase(key);
+ if (StringUtils::StartsWith(lower_key, "x-acs-")) {
+ sorted[lower_key] = Trimmed(value);
+ }
+ }
+ std::string canonical;
+ for (const auto& [key, value] : sorted) {
+ canonical += key + ":" + value + "\n";
+ }
+ return canonical;
+}
+
+std::string OpenApiCanonicalizedResource(const RestAuthParameter& parameter) {
+ std::string resource = UrlUtils::DecodeString(parameter.resource_path);
+ if (parameter.parameters.empty()) {
+ return resource;
+ }
+ resource += "?";
+ bool first = true;
+ for (const auto& [key, value] : parameter.parameters) {
+ if (!first) {
+ resource += "&";
+ }
+ resource += key;
+ std::string decoded = UrlUtils::DecodeString(value);
+ if (!decoded.empty()) {
+ resource += "=" + decoded;
+ }
+ first = false;
+ }
+ return resource;
+}
+
+Result<std::string> GenerateNonce(std::chrono::system_clock::time_point now) {
+ std::string uuid;
+ if (!UUID::Generate(&uuid)) {
+ return Status::IOError("failed to generate DLF OpenAPI signing nonce");
+ }
+ int64_t millis =
+
std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+ std::ostringstream thread_id;
+ thread_id << std::this_thread::get_id();
+ return fmt::format("{}{}{}", uuid, millis, thread_id.str());
+}
+
+Result<std::unique_ptr<DlfRequestSigner>> CreateSigner(const std::string&
algorithm,
+ const std::string&
region) {
+ if (algorithm == DlfDefaultSigner::kIdentifier) {
+ return std::make_unique<DlfDefaultSigner>(region);
+ }
+ if (algorithm == DlfOpenApiSigner::kIdentifier) {
+ return std::make_unique<DlfOpenApiSigner>();
+ }
+ return Status::Invalid(fmt::format(
+ "unsupported DLF signing algorithm '{}', supported values are
'default' and 'openapi'",
+ algorithm));
+}
+
+} // namespace
+
+DlfToken::DlfToken(const std::string& access_key_id, const std::string&
access_key_secret,
+ const std::optional<std::string>& security_token,
+ const std::optional<int64_t>& expiration_at_millis)
+ : access_key_id_(access_key_id),
+ access_key_secret_(access_key_secret),
+ security_token_(security_token),
+ expiration_at_millis_(expiration_at_millis) {}
+
+Result<DlfToken> DlfToken::FromJson(const std::string& json) {
+ rapidjson::Document document;
+ document.Parse(json.data(), json.size());
+ if (document.HasParseError() || !document.IsObject()) {
+ return Status::Invalid("failed to parse DLF token JSON");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string access_key_id,
RequiredJsonString(document, "AccessKeyId"));
+ PAIMON_ASSIGN_OR_RAISE(std::string access_key_secret,
+ RequiredJsonString(document, "AccessKeySecret"));
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> security_token,
+ OptionalJsonString(document, "SecurityToken"));
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> expiration,
+ OptionalJsonString(document, "Expiration"));
+ std::optional<int64_t> expiration_at_millis;
+ if (expiration) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t parsed_expiration,
ParseExpiration(expiration.value()));
+ expiration_at_millis = parsed_expiration;
+ }
+ return DlfToken(access_key_id, access_key_secret, security_token,
expiration_at_millis);
+}
+
+bool DlfToken::ShouldRefresh(std::chrono::system_clock::time_point now) const {
+ if (!expiration_at_millis_) {
+ return false;
+ }
+ int64_t now_millis =
+
std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+ return expiration_at_millis_.value() - now_millis <
kTokenExpirationSafeTimeMillis;
+}
+
+DlfLocalFileTokenLoader::DlfLocalFileTokenLoader(const std::string&
token_file_path,
+ int32_t max_attempts,
+ std::chrono::milliseconds
retry_delay)
+ : token_file_path_(token_file_path), max_attempts_(max_attempts),
retry_delay_(retry_delay) {}
+
+Result<DlfToken> DlfLocalFileTokenLoader::LoadToken() {
+ if (token_file_path_.empty()) {
+ return Status::Invalid("DLF token file path is empty");
+ }
+ if (max_attempts_ <= 0 || retry_delay_.count() < 0) {
+ return Status::Invalid("invalid DLF token file retry configuration");
+ }
+ Status last_status = Status::Invalid("failed to load DLF token file");
+ for (int32_t attempt = 1; attempt <= max_attempts_; ++attempt) {
+ std::ifstream file(token_file_path_, std::ios::binary);
+ if (!file.is_open()) {
+ last_status = Status::IOError(
+ fmt::format("failed to read DLF token file '{}'",
token_file_path_));
+ } else {
+ std::string contents(kMaxTokenResponseBytes + 1, '\0');
+ file.read(contents.data(),
static_cast<std::streamsize>(contents.size()));
+ std::streamsize size = file.gcount();
+ if (file.bad()) {
+ last_status = Status::IOError(
+ fmt::format("failed to read DLF token file '{}'",
token_file_path_));
+ } else if (size >
static_cast<std::streamsize>(kMaxTokenResponseBytes)) {
+ last_status = Status::Invalid("DLF token file is too large");
+ } else {
+ contents.resize(static_cast<size_t>(size));
+ Result<DlfToken> token = DlfToken::FromJson(contents);
+ if (token.ok()) {
+ return token;
+ }
+ last_status = Status::Invalid("failed to parse DLF token
file");
+ }
+ }
+ if (attempt < max_attempts_) {
+ std::this_thread::sleep_for(retry_delay_ * attempt);
+ }
+ }
+ return last_status;
+}
+
+std::string DlfLocalFileTokenLoader::Description() const {
+ return token_file_path_;
+}
+
+DlfEcsTokenLoader::DlfEcsTokenLoader(const std::string& metadata_url,
+ const std::optional<std::string>&
role_name,
+ std::unique_ptr<HttpClient> http_client)
+ : metadata_url_(metadata_url), role_name_(role_name),
http_client_(std::move(http_client)) {}
+
+std::unique_ptr<DlfEcsTokenLoader> DlfEcsTokenLoader::Create(
+ const std::string& metadata_url, const std::optional<std::string>&
role_name) {
+ return std::make_unique<DlfEcsTokenLoader>(metadata_url, role_name,
+
std::make_unique<CurlHttpClient>());
+}
+
+Result<std::string> DlfEcsTokenLoader::Get(const std::string& url) const {
+ if (!http_client_) {
+ return Status::Invalid("DLF ECS metadata HTTP client is not
configured");
+ }
+ HttpRequest request;
+ request.url = url;
+ request.request_timeout_ms = kEcsMetadataRequestTimeoutMillis;
+ std::string body;
+ Result<HttpResponse> response =
+ http_client_->Execute(request, [&body](const char* data, int64_t size)
{
+ if (size < 0 || body.size() + static_cast<size_t>(size) >
kMaxTokenResponseBytes) {
+ return Status::Invalid("DLF ECS metadata response is too
large");
+ }
+ body.append(data, static_cast<size_t>(size));
+ return Status::OK();
+ });
+ if (!response.ok()) {
+ return Status::IOError("failed to request DLF credentials from ECS
metadata service: ",
+ response.status().message());
+ }
+ HttpResponse http_response = std::move(response).value();
+ if (http_response.status_code < 200 || http_response.status_code >= 300) {
+ return Status::IOError(fmt::format("DLF ECS metadata service returned
HTTP status {}",
+ http_response.status_code));
+ }
+ if (StringUtils::IsNullOrWhitespaceOnly(body)) {
+ return Status::Invalid("DLF ECS metadata service returned an empty
response");
+ }
+ return body;
+}
+
+Result<DlfToken> DlfEcsTokenLoader::LoadToken() {
+ if (metadata_url_.empty()) {
+ return Status::Invalid("DLF ECS metadata URL is empty");
+ }
+ if (!role_name_) {
+ PAIMON_ASSIGN_OR_RAISE(std::string role, Get(metadata_url_));
+ TrimWhitespace(&role);
+ if (role.empty()) {
+ return Status::Invalid("DLF ECS metadata service returned an empty
role name");
+ }
+ role_name_ = role;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string token_json, Get(metadata_url_ +
role_name_.value()));
+ Result<DlfToken> token = DlfToken::FromJson(token_json);
+ if (!token.ok()) {
+ return Status::Invalid("failed to parse DLF ECS token response");
+ }
+ return token;
+}
+
+std::string DlfEcsTokenLoader::Description() const {
+ return metadata_url_;
+}
+
+DlfDefaultSigner::DlfDefaultSigner(const std::string& region) :
region_(region) {}
+
+Result<DlfRequestSigner::Headers> DlfDefaultSigner::SignHeaders(
+ const std::string& body, std::chrono::system_clock::time_point now,
+ const std::optional<std::string>& security_token, const std::string& host)
const {
+ PAIMON_ASSIGN_OR_RAISE(std::string date_time, FormatDlfTime(now));
+ Headers headers = {{kDlfDateHeader, date_time},
+ {kDlfContentSha256Header, kUnsignedPayload},
+ {kDlfVersionHeader, "v1"}};
+ if (!body.empty()) {
+ PAIMON_ASSIGN_OR_RAISE(std::string content_md5, Md5Base64(body));
+ headers[kContentTypeHeader] = kJsonMediaType;
+ headers[kContentMd5Header] = content_md5;
+ }
+ if (security_token) {
+ headers[kDlfSecurityTokenHeader] = security_token.value();
+ }
+ return headers;
+}
+
+Result<std::string> DlfDefaultSigner::Authorization(const RestAuthParameter&
parameter,
+ const DlfToken& token,
const std::string& host,
+ const Headers&
sign_headers) const {
+ PAIMON_ASSIGN_OR_RAISE(std::string date_time, RequiredHeader(sign_headers,
kDlfDateHeader));
+ if (date_time.size() < 8) {
+ return Status::Invalid("DLF signing date is invalid");
+ }
+ std::string date = date_time.substr(0, 8);
+ std::string scope = fmt::format("{}/{}/DlfNext/aliyun_v4_request", date,
region_);
+ std::string canonical_request = DefaultCanonicalRequest(parameter,
sign_headers);
+ PAIMON_ASSIGN_OR_RAISE(Bytes canonical_hash, Digest(EVP_sha256(),
canonical_request));
+ std::string string_to_sign =
+ fmt::format("DLF4-HMAC-SHA256\n{}\n{}\n{}", date_time, scope,
HexEncode(canonical_hash));
+
+ PAIMON_ASSIGN_OR_RAISE(
+ Bytes date_key,
+ Hmac(EVP_sha256(), ToBytes("aliyun_v4" + token.GetAccessKeySecret()),
date));
+ PAIMON_ASSIGN_OR_RAISE(Bytes region_key, Hmac(EVP_sha256(), date_key,
region_));
+ PAIMON_ASSIGN_OR_RAISE(Bytes service_key, Hmac(EVP_sha256(), region_key,
"DlfNext"));
+ PAIMON_ASSIGN_OR_RAISE(Bytes signing_key, Hmac(EVP_sha256(), service_key,
"aliyun_v4_request"));
+ PAIMON_ASSIGN_OR_RAISE(Bytes signature, Hmac(EVP_sha256(), signing_key,
string_to_sign));
+ return fmt::format("DLF4-HMAC-SHA256 Credential={}/{},Signature={}",
token.GetAccessKeyId(),
+ scope, HexEncode(signature));
+}
+
+Result<DlfRequestSigner::Headers> DlfOpenApiSigner::SignHeaders(
+ const std::string& body, std::chrono::system_clock::time_point now,
+ const std::optional<std::string>& security_token, const std::string& host)
const {
+ if (host.empty()) {
+ return Status::Invalid("DLF OpenAPI signing host is empty");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string date, FormatRfc1123Time(now));
+ PAIMON_ASSIGN_OR_RAISE(std::string nonce, GenerateNonce(now));
+ Headers headers = {{kOpenApiDateHeader, date},
{kOpenApiAcceptHeader, kJsonMediaType},
+ {kOpenApiHostHeader, host},
{kAcsSignatureMethodHeader, "HMAC-SHA1"},
+ {kAcsSignatureNonceHeader, nonce},
{kAcsSignatureVersionHeader, "1.0"},
+ {kAcsVersionHeader, "2026-01-18"}};
+ if (!body.empty()) {
+ PAIMON_ASSIGN_OR_RAISE(std::string content_md5, Md5Base64(body));
+ headers[kContentMd5Header] = content_md5;
+ headers[kContentTypeHeader] = kJsonMediaType;
+ }
+ if (security_token) {
+ headers[kAcsSecurityTokenHeader] = security_token.value();
+ }
+ return headers;
+}
+
+Result<std::string> DlfOpenApiSigner::Authorization(const RestAuthParameter&
parameter,
+ const DlfToken& token,
const std::string& host,
+ const Headers&
sign_headers) const {
+ PAIMON_ASSIGN_OR_RAISE(std::string accept, RequiredHeader(sign_headers,
kOpenApiAcceptHeader));
+ PAIMON_ASSIGN_OR_RAISE(std::string date, RequiredHeader(sign_headers,
kOpenApiDateHeader));
+ std::string content_md5;
+ auto md5_iter = sign_headers.find(kContentMd5Header);
+ if (md5_iter != sign_headers.end()) {
+ content_md5 = md5_iter->second;
+ }
+ std::string content_type;
+ auto type_iter = sign_headers.find(kContentTypeHeader);
+ if (type_iter != sign_headers.end()) {
+ content_type = type_iter->second;
+ }
+ std::string string_to_sign =
+ parameter.method + "\n" + accept + "\n" + content_md5 + "\n" +
content_type + "\n" + date +
+ "\n" + OpenApiCanonicalizedHeaders(sign_headers) +
OpenApiCanonicalizedResource(parameter);
+ PAIMON_ASSIGN_OR_RAISE(Bytes signature,
+ Hmac(EVP_sha1(),
ToBytes(token.GetAccessKeySecret()), string_to_sign));
+ PAIMON_ASSIGN_OR_RAISE(std::string encoded_signature,
Base64Encode(signature));
+ return fmt::format("acs {}:{}", token.GetAccessKeyId(), encoded_signature);
+}
+
+DlfAuthProvider::DlfAuthProvider(std::unique_ptr<DlfTokenLoader> token_loader,
+ const std::optional<DlfToken>& token, const
std::string& host,
+ std::unique_ptr<DlfRequestSigner> signer,
Clock clock)
+ : token_loader_(std::move(token_loader)),
+ token_(token),
+ host_(host),
+ signer_(std::move(signer)),
+ clock_(std::move(clock)) {}
+
+Result<std::unique_ptr<DlfAuthProvider>> DlfAuthProvider::Create(
+ const std::map<std::string, std::string>& options) {
+ PAIMON_ASSIGN_OR_RAISE(std::string uri, RequiredNonEmptyOption(options,
CatalogOptions::URI));
+ std::string region;
+ PAIMON_ASSIGN_OR_RAISE(
+ std::optional<std::string> configured_region,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(options,
CatalogOptions::DLF_REGION));
+ if (configured_region) {
+ if (configured_region->empty()) {
+ return Status::Invalid("option 'dlf.region' must not be empty");
+ }
+ region = configured_region.value();
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(region, ParseRegionFromUri(uri));
+ }
+
+ std::string algorithm;
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> configured_algorithm,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options,
CatalogOptions::DLF_SIGNING_ALGORITHM));
+ if (configured_algorithm) {
+ algorithm = configured_algorithm.value();
+ } else {
+ algorithm = ParseSigningAlgorithmFromUri(uri);
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> loader_name,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options, CatalogOptions::DLF_TOKEN_LOADER));
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> token_path,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options, CatalogOptions::DLF_TOKEN_PATH));
+ if (loader_name) {
+ if (loader_name.value() == "ecs") {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string>
configured_metadata_url,
+
OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options,
CatalogOptions::DLF_TOKEN_ECS_METADATA_URL));
+ std::string metadata_url =
configured_metadata_url.value_or(kDefaultEcsMetadataUrl);
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> role_name,
+
OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options,
CatalogOptions::DLF_TOKEN_ECS_ROLE_NAME));
+ return FromTokenLoader(DlfEcsTokenLoader::Create(metadata_url,
role_name), uri, region,
+ algorithm, std::chrono::system_clock::now);
+ }
+ if (loader_name.value() == "local_file") {
+ PAIMON_ASSIGN_OR_RAISE(std::string path,
+ RequiredNonEmptyOption(options,
CatalogOptions::DLF_TOKEN_PATH));
+ return FromTokenLoader(
+ std::make_unique<DlfLocalFileTokenLoader>(path, 5,
std::chrono::seconds(1)), uri,
+ region, algorithm, std::chrono::system_clock::now);
+ }
+ return Status::NotImplemented(
+ fmt::format("unsupported DLF token loader '{}', supported values
are 'ecs' and "
+ "'local_file'",
+ loader_name.value()));
+ }
+ if (token_path) {
+ if (token_path->empty()) {
+ return Status::Invalid("option 'dlf.token-path' must not be
empty");
+ }
+ return
FromTokenLoader(std::make_unique<DlfLocalFileTokenLoader>(token_path.value(), 5,
+
std::chrono::seconds(1)),
+ uri, region, algorithm,
std::chrono::system_clock::now);
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> access_key_id,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options, CatalogOptions::DLF_ACCESS_KEY_ID));
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> access_key_secret,
+ OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options,
CatalogOptions::DLF_ACCESS_KEY_SECRET));
+ if (access_key_id && !access_key_id->empty() && access_key_secret &&
+ !access_key_secret->empty()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> security_token,
+
OptionsUtils::GetOptionalValueFromMap<std::string>(
+ options,
CatalogOptions::DLF_SECURITY_TOKEN));
+ DlfToken token(access_key_id.value(), access_key_secret.value(),
security_token,
+ std::nullopt);
+ return FromAccessKey(token, uri, region, algorithm,
std::chrono::system_clock::now);
+ }
+ return Status::Invalid("DLF token path or access key must be configured
for DLF auth");
+}
+
+Result<std::unique_ptr<DlfAuthProvider>> DlfAuthProvider::FromAccessKey(
+ const DlfToken& token, const std::string& uri, const std::string& region,
+ const std::string& signing_algorithm, Clock clock) {
+ if (token.GetAccessKeyId().empty() || token.GetAccessKeySecret().empty()) {
+ return Status::Invalid("DLF access key id and secret must not be
empty");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string host, ExtractHost(uri));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<DlfRequestSigner> signer,
+ CreateSigner(signing_algorithm, region));
+ return std::unique_ptr<DlfAuthProvider>(
+ new DlfAuthProvider(nullptr, token, host, std::move(signer),
std::move(clock)));
+}
+
+Result<std::unique_ptr<DlfAuthProvider>> DlfAuthProvider::FromTokenLoader(
+ std::unique_ptr<DlfTokenLoader> token_loader, const std::string& uri,
const std::string& region,
+ const std::string& signing_algorithm, Clock clock) {
+ if (!token_loader) {
+ return Status::Invalid("DLF token loader must not be null");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::string host, ExtractHost(uri));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<DlfRequestSigner> signer,
+ CreateSigner(signing_algorithm, region));
+ return std::unique_ptr<DlfAuthProvider>(new DlfAuthProvider(
+ std::move(token_loader), std::nullopt, host, std::move(signer),
std::move(clock)));
+}
+
+Result<DlfToken>
DlfAuthProvider::GetFreshToken(std::chrono::system_clock::time_point now) const
{
+ std::scoped_lock lock(token_mutex_);
+ if (token_ && !token_->ShouldRefresh(now)) {
+ return token_.value();
+ }
+ if (!token_loader_) {
+ return Status::Invalid("DLF credentials expired and no token loader is
configured");
+ }
+ PAIMON_ASSIGN_OR_RAISE(DlfToken loaded, token_loader_->LoadToken());
+ if (loaded.GetAccessKeyId().empty() ||
loaded.GetAccessKeySecret().empty()) {
+ return Status::Invalid("DLF token loader returned empty access key
credentials");
+ }
+ token_ = loaded;
+ return loaded;
+}
+
+Result<std::map<std::string, std::string>> DlfAuthProvider::MergeAuthHeader(
+ const std::map<std::string, std::string>& base_header,
+ const RestAuthParameter& parameter) const {
+ PAIMON_ASSIGN_OR_RAISE(DlfToken token, GetFreshToken(clock_()));
+ std::chrono::system_clock::time_point signing_time = clock_();
+ PAIMON_ASSIGN_OR_RAISE(
+ DlfRequestSigner::Headers sign_headers,
+ signer_->SignHeaders(parameter.data, signing_time,
token.GetSecurityToken(), host_));
+ PAIMON_ASSIGN_OR_RAISE(std::string authorization,
+ signer_->Authorization(parameter, token, host_,
sign_headers));
+ std::map<std::string, std::string> headers = base_header;
+ for (const auto& [key, value] : sign_headers) {
+ headers[key] = value;
+ }
+ headers[kAuthorizationHeader] = authorization;
+ return headers;
+}
+
+Result<std::string> DlfAuthProvider::ParseRegionFromUri(const std::string&
uri) {
+ static const std::regex
kRegionPattern("(?:pre-)?([a-z]+-[a-z]+(?:-[0-9]+)?)");
+ std::smatch match;
+ if (std::regex_search(uri, match, kRegionPattern) && match.size() > 1 &&
+ !match.str(1).empty()) {
+ return match.str(1);
+ }
+ return Status::Invalid(
+ "could not determine DLF region from option 'dlf.region' or REST
catalog URI");
+}
+
+std::string DlfAuthProvider::ParseSigningAlgorithmFromUri(const std::string&
uri) {
+ std::string lower_uri = StringUtils::ToLowerCase(uri);
+ return lower_uri.find("dlfnext") == std::string::npos ?
DlfDefaultSigner::kIdentifier
+ :
DlfOpenApiSigner::kIdentifier;
+}
+
+Result<std::string> DlfAuthProvider::ExtractHost(const std::string& uri) {
+ std::string host = RestHttpClient::NormalizeUri(uri);
+ std::string lower_uri = StringUtils::ToLowerCase(host);
+ if (StringUtils::StartsWith(lower_uri, "http://";)) {
+ host.erase(0, 7);
+ } else if (StringUtils::StartsWith(lower_uri, "https://";)) {
+ host.erase(0, 8);
+ }
+ size_t path = host.find('/');
+ if (path != std::string::npos) {
+ host.resize(path);
+ }
+ if (host.empty() || host.find('?') != std::string::npos ||
+ host.find('#') != std::string::npos || host.find('@') !=
std::string::npos) {
+ return Status::Invalid("could not determine DLF signing host from REST
catalog URI");
+ }
+ return host;
+}
+
+} // namespace paimon
diff --git a/src/paimon/rest/dlf_auth.h b/src/paimon/rest/dlf_auth.h
new file mode 100644
index 00000000..c10b655b
--- /dev/null
+++ b/src/paimon/rest/dlf_auth.h
@@ -0,0 +1,208 @@
+/*
+ * 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 <chrono>
+#include <cstdint>
+#include <functional>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "paimon/common/utils/http_client.h"
+#include "paimon/rest/rest_auth.h"
+
+namespace paimon {
+
+/// Access key credentials used to sign DLF REST requests.
+class DlfToken {
+ public:
+ DlfToken(const std::string& access_key_id, const std::string&
access_key_secret,
+ const std::optional<std::string>& security_token,
+ const std::optional<int64_t>& expiration_at_millis);
+
+ static Result<DlfToken> FromJson(const std::string& json);
+
+ const std::string& GetAccessKeyId() const {
+ return access_key_id_;
+ }
+
+ const std::string& GetAccessKeySecret() const {
+ return access_key_secret_;
+ }
+
+ const std::optional<std::string>& GetSecurityToken() const {
+ return security_token_;
+ }
+
+ const std::optional<int64_t>& GetExpirationAtMillis() const {
+ return expiration_at_millis_;
+ }
+
+ bool ShouldRefresh(std::chrono::system_clock::time_point now) const;
+
+ private:
+ std::string access_key_id_;
+ std::string access_key_secret_;
+ std::optional<std::string> security_token_;
+ std::optional<int64_t> expiration_at_millis_;
+};
+
+/// Loads refreshable DLF credentials.
+class DlfTokenLoader {
+ public:
+ virtual ~DlfTokenLoader() = default;
+
+ virtual Result<DlfToken> LoadToken() = 0;
+ virtual std::string Description() const = 0;
+};
+
+/// Loads a DLF STS token from a JSON file.
+class DlfLocalFileTokenLoader : public DlfTokenLoader {
+ public:
+ DlfLocalFileTokenLoader(const std::string& token_file_path, int32_t
max_attempts,
+ std::chrono::milliseconds retry_delay);
+
+ Result<DlfToken> LoadToken() override;
+ std::string Description() const override;
+
+ private:
+ std::string token_file_path_;
+ int32_t max_attempts_;
+ std::chrono::milliseconds retry_delay_;
+};
+
+/// Loads a DLF STS token from the Alibaba Cloud ECS metadata service.
+class DlfEcsTokenLoader : public DlfTokenLoader {
+ public:
+ DlfEcsTokenLoader(const std::string& metadata_url, const
std::optional<std::string>& role_name,
+ std::unique_ptr<HttpClient> http_client);
+
+ static std::unique_ptr<DlfEcsTokenLoader> Create(const std::string&
metadata_url,
+ const
std::optional<std::string>& role_name);
+
+ Result<DlfToken> LoadToken() override;
+ std::string Description() const override;
+
+ private:
+ Result<std::string> Get(const std::string& url) const;
+
+ std::string metadata_url_;
+ std::optional<std::string> role_name_;
+ std::unique_ptr<HttpClient> http_client_;
+};
+
+/// Signs a DLF REST request using one of the endpoint-specific algorithms.
+class DlfRequestSigner {
+ public:
+ using Headers = std::map<std::string, std::string>;
+
+ virtual ~DlfRequestSigner() = default;
+
+ virtual Result<Headers> SignHeaders(const std::string& body,
+ std::chrono::system_clock::time_point
now,
+ const std::optional<std::string>&
security_token,
+ const std::string& host) const = 0;
+
+ virtual Result<std::string> Authorization(const RestAuthParameter&
parameter,
+ const DlfToken& token, const
std::string& host,
+ const Headers& sign_headers)
const = 0;
+};
+
+/// DLF4-HMAC-SHA256 signer used by the default DLF VPC endpoint.
+class DlfDefaultSigner : public DlfRequestSigner {
+ public:
+ static constexpr const char* kIdentifier = "default";
+
+ explicit DlfDefaultSigner(const std::string& region);
+
+ Result<Headers> SignHeaders(const std::string& body,
std::chrono::system_clock::time_point now,
+ const std::optional<std::string>&
security_token,
+ const std::string& host) const override;
+
+ Result<std::string> Authorization(const RestAuthParameter& parameter,
const DlfToken& token,
+ const std::string& host,
+ const Headers& sign_headers) const
override;
+
+ private:
+ std::string region_;
+};
+
+/// ROA HMAC-SHA1 signer used by DlfNext OpenAPI endpoints.
+class DlfOpenApiSigner : public DlfRequestSigner {
+ public:
+ static constexpr const char* kIdentifier = "openapi";
+
+ Result<Headers> SignHeaders(const std::string& body,
std::chrono::system_clock::time_point now,
+ const std::optional<std::string>&
security_token,
+ const std::string& host) const override;
+
+ Result<std::string> Authorization(const RestAuthParameter& parameter,
const DlfToken& token,
+ const std::string& host,
+ const Headers& sign_headers) const
override;
+};
+
+/// Generates DLF authentication headers and refreshes expiring credentials.
+class DlfAuthProvider : public AuthProvider {
+ public:
+ using Clock = std::function<std::chrono::system_clock::time_point()>;
+
+ static Result<std::unique_ptr<DlfAuthProvider>> Create(
+ const std::map<std::string, std::string>& options);
+
+ static Result<std::unique_ptr<DlfAuthProvider>> FromAccessKey(
+ const DlfToken& token, const std::string& uri, const std::string&
region,
+ const std::string& signing_algorithm, Clock clock);
+
+ static Result<std::unique_ptr<DlfAuthProvider>> FromTokenLoader(
+ std::unique_ptr<DlfTokenLoader> token_loader, const std::string& uri,
+ const std::string& region, const std::string& signing_algorithm, Clock
clock);
+
+ Result<std::map<std::string, std::string>> MergeAuthHeader(
+ const std::map<std::string, std::string>& base_header,
+ const RestAuthParameter& parameter) const override;
+
+ bool AllowsRedirects() const override {
+ return false;
+ }
+
+ static Result<std::string> ParseRegionFromUri(const std::string& uri);
+ static std::string ParseSigningAlgorithmFromUri(const std::string& uri);
+ static Result<std::string> ExtractHost(const std::string& uri);
+
+ private:
+ DlfAuthProvider(std::unique_ptr<DlfTokenLoader> token_loader,
+ const std::optional<DlfToken>& token, const std::string&
host,
+ std::unique_ptr<DlfRequestSigner> signer, Clock clock);
+
+ Result<DlfToken> GetFreshToken(std::chrono::system_clock::time_point now)
const;
+
+ std::unique_ptr<DlfTokenLoader> token_loader_;
+ mutable std::optional<DlfToken> token_;
+ std::string host_;
+ std::unique_ptr<DlfRequestSigner> signer_;
+ Clock clock_;
+ mutable std::mutex token_mutex_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/rest/dlf_auth_test.cpp
b/src/paimon/rest/dlf_auth_test.cpp
new file mode 100644
index 00000000..afb9cec5
--- /dev/null
+++ b/src/paimon/rest/dlf_auth_test.cpp
@@ -0,0 +1,468 @@
+/*
+ * 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 "paimon/rest/dlf_auth.h"
+
+#include <atomic>
+#include <chrono>
+#include <fstream>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/catalog_options.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+namespace {
+
+using StringMap = std::map<std::string, std::string>;
+
+std::chrono::system_clock::time_point FixedTime() {
+ return std::chrono::system_clock::from_time_t(1744775086);
+}
+
+class SequenceTokenLoader : public DlfTokenLoader {
+ public:
+ explicit SequenceTokenLoader(
+ std::vector<DlfToken> tokens,
+ std::chrono::milliseconds load_delay = std::chrono::milliseconds(0))
+ : tokens_(std::move(tokens)), load_delay_(load_delay) {}
+
+ Result<DlfToken> LoadToken() override {
+ std::this_thread::sleep_for(load_delay_);
+ int32_t index = load_count_.fetch_add(1);
+ if (index >= static_cast<int32_t>(tokens_.size())) {
+ return Status::Invalid("test token loader exhausted");
+ }
+ return tokens_[index];
+ }
+
+ std::string Description() const override {
+ return "test sequence";
+ }
+
+ int32_t GetLoadCount() const {
+ return load_count_.load();
+ }
+
+ private:
+ std::vector<DlfToken> tokens_;
+ std::chrono::milliseconds load_delay_;
+ std::atomic<int32_t> load_count_{0};
+};
+
+class MockEcsHttpClient : public HttpClient {
+ public:
+ explicit MockEcsHttpClient(const std::string& metadata_url, bool
direct_token = false)
+ : metadata_url_(metadata_url), direct_token_(direct_token) {}
+
+ Result<HttpResponse> Execute(const HttpRequest& request,
+ const HttpBodyConsumer& consumer) const
override {
+ last_request_timeout_ms_.store(request.request_timeout_ms);
+ HttpResponse response;
+ response.status_code = 200;
+ std::string body;
+ if (request.url == metadata_url_) {
+ if (direct_token_) {
+ token_requests_.fetch_add(1);
+ body = R"({"AccessKeyId":"ecs-ak","AccessKeySecret":"ecs-sk",)"
+
R"("SecurityToken":"ecs-sts","Expiration":"2027-04-16T05:44:46Z"})";
+ } else {
+ role_requests_.fetch_add(1);
+ body = " test-role\n";
+ }
+ } else if (request.url == metadata_url_ + "test-role") {
+ token_requests_.fetch_add(1);
+ body = R"({"AccessKeyId":"ecs-ak","AccessKeySecret":"ecs-sk",)"
+
R"("SecurityToken":"ecs-sts","Expiration":"2027-04-16T05:44:46Z"})";
+ } else {
+ response.status_code = 404;
+ }
+ if (!body.empty()) {
+ PAIMON_RETURN_NOT_OK(consumer(body.data(),
static_cast<int64_t>(body.size())));
+ response.body_size = static_cast<int64_t>(body.size());
+ }
+ return response;
+ }
+
+ int32_t GetRoleRequestCount() const {
+ return role_requests_.load();
+ }
+
+ int32_t GetTokenRequestCount() const {
+ return token_requests_.load();
+ }
+
+ int64_t GetLastRequestTimeoutMillis() const {
+ return last_request_timeout_ms_.load();
+ }
+
+ private:
+ std::string metadata_url_;
+ bool direct_token_;
+ mutable std::atomic<int32_t> role_requests_{0};
+ mutable std::atomic<int32_t> token_requests_{0};
+ mutable std::atomic<int64_t> last_request_timeout_ms_{-1};
+};
+
+class FailingEcsHttpClient : public HttpClient {
+ public:
+ Result<HttpResponse> Execute(const HttpRequest&, const HttpBodyConsumer&)
const override {
+ return Status::IOError("connection refused");
+ }
+};
+
+} // namespace
+
+TEST(DlfDefaultSignerTest, SignsJavaCompatibleRequest) {
+ DlfDefaultSigner signer("cn-beijing");
+ const std::string body = R"({"name":"t1"})";
+ DlfToken token("YourAccessKeyId", "YourAccessKeySecret", "securityToken",
std::nullopt);
+ RestAuthParameter parameter =
+ RestAuthParameter::Create("POST", "/v1/wh/databases/db/tables",
+ {{"warehouse", "my instance"}, {"branch",
"main"}}, body);
+
+ ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers,
+ signer.SignHeaders(body, FixedTime(),
token.GetSecurityToken(), "unused"));
+ ASSERT_EQ("20250416T034446Z", headers.at("x-dlf-date"));
+ ASSERT_EQ("Od9T1x3c2+JusJPFMpXe9Q==", headers.at("Content-MD5"));
+ ASSERT_EQ("application/json", headers.at("Content-Type"));
+ ASSERT_EQ("UNSIGNED-PAYLOAD", headers.at("x-dlf-content-sha256"));
+ ASSERT_EQ("v1", headers.at("x-dlf-version"));
+ ASSERT_EQ("securityToken", headers.at("x-dlf-security-token"));
+
+ ASSERT_OK_AND_ASSIGN(std::string authorization,
+ signer.Authorization(parameter, token, "unused",
headers));
+ ASSERT_EQ(
+ "DLF4-HMAC-SHA256 Credential=YourAccessKeyId/20250416/cn-beijing/"
+ "DlfNext/aliyun_v4_request,Signature="
+ "22594f8bbb8bb0ec296ced6003b7ffdf7022a8ca3815da5b53090daa11a06558",
+ authorization);
+}
+
+TEST(DlfDefaultSignerTest, MatchesJavaGoldenAuthorization) {
+ // Mirrors Java DLFAuthSignatureTest#testGetAuthorization.
+ DlfDefaultSigner signer("cn-hangzhou");
+ const std::string body = R"({"name":"database","options":{"a":"b"}})";
+ DlfToken token("access-key-id", "access-key-secret", "securityToken",
std::nullopt);
+ RestAuthParameter parameter = RestAuthParameter::Create("POST",
"/v1/paimon/databases",
+ {{"k1", "v1"},
{"k2", "v2"}}, body);
+ const std::chrono::system_clock::time_point signing_time =
+ std::chrono::system_clock::from_time_t(1701605532);
+
+ ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers,
+ signer.SignHeaders(body, signing_time,
token.GetSecurityToken(), "host"));
+ ASSERT_OK_AND_ASSIGN(std::string authorization,
+ signer.Authorization(parameter, token, "host",
headers));
+ ASSERT_EQ(
+ "DLF4-HMAC-SHA256 Credential=access-key-id/20231203/cn-hangzhou/"
+ "DlfNext/aliyun_v4_request,Signature="
+ "c72caf1d40b55b1905d891ee3e3de48a2f8bebefa7e39e4f277acc93c269c5e3",
+ authorization);
+}
+
+TEST(DlfDefaultSignerTest, OmitsBodyAndSecurityTokenHeadersWhenAbsent) {
+ DlfDefaultSigner signer("cn-hangzhou");
+ ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers,
+ signer.SignHeaders("", FixedTime(), std::nullopt,
"unused"));
+ ASSERT_EQ(3, headers.size());
+ ASSERT_EQ(0, headers.count("Content-MD5"));
+ ASSERT_EQ(0, headers.count("Content-Type"));
+ ASSERT_EQ(0, headers.count("x-dlf-security-token"));
+}
+
+TEST(DlfOpenApiSignerTest, SignsJavaCompatibleRequest) {
+ DlfOpenApiSigner signer;
+ const std::string body =
R"({"CategoryName":"test","CategoryType":"UNSTRUCTURED"})";
+ const std::string host = "dlfnext.cn-beijing.aliyuncs.com";
+ DlfToken token("YourAccessKeyId", "YourAccessKeySecret", "securityToken",
std::nullopt);
+ RestAuthParameter parameter =
+ RestAuthParameter::Create("POST",
"/llm-p2e4XXXXXXXXsvtn/datacenter/category", {}, body);
+
+ ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers,
+ signer.SignHeaders(body, FixedTime(),
token.GetSecurityToken(), host));
+ headers["x-acs-signature-nonce"] = "ef34aae7-7bd2-413d-a541-680cd2c48538";
+ ASSERT_EQ("Wed, 16 Apr 2025 03:44:46 GMT", headers.at("Date"));
+ ASSERT_EQ("q2qaEcR4P47+Z7CUzHRTBw==", headers.at("Content-MD5"));
+ ASSERT_EQ("application/json", headers.at("Accept"));
+ ASSERT_EQ("application/json", headers.at("Content-Type"));
+ ASSERT_EQ(host, headers.at("Host"));
+ ASSERT_EQ("HMAC-SHA1", headers.at("x-acs-signature-method"));
+ ASSERT_EQ("1.0", headers.at("x-acs-signature-version"));
+ ASSERT_EQ("2026-01-18", headers.at("x-acs-version"));
+ ASSERT_EQ("securityToken", headers.at("x-acs-security-token"));
+
+ ASSERT_OK_AND_ASSIGN(std::string authorization,
+ signer.Authorization(parameter, token, host,
headers));
+ ASSERT_EQ("acs YourAccessKeyId:wX4CDPSCtfgYkxdK9tJIO3ez5VI=",
authorization);
+}
+
+TEST(DlfOpenApiSignerTest, DecodesPathAndQueryValuesBeforeSigning) {
+ DlfOpenApiSigner signer;
+ DlfToken token("ak", "sk", std::nullopt, std::nullopt);
+ RestAuthParameter parameter = RestAuthParameter::Create(
+ "GET", "/v1/%24snapshots", {{"z", ""}, {"name", "hello world"}}, "");
+ ASSERT_OK_AND_ASSIGN(DlfRequestSigner::Headers headers,
+ signer.SignHeaders("", FixedTime(), std::nullopt,
"host"));
+ headers["x-acs-signature-nonce"] = "fixed-nonce";
+
+ ASSERT_OK_AND_ASSIGN(std::string authorization,
+ signer.Authorization(parameter, token, "host",
headers));
+ ASSERT_EQ("acs ak:vD+M7291KKoOTvt2gQuD6jPDlw8=", authorization);
+ ASSERT_EQ(0, headers.count("Content-MD5"));
+ ASSERT_EQ(0, headers.count("Content-Type"));
+}
+
+TEST(DlfTokenTest, ParsesExpirationAndRefreshBoundary) {
+ ASSERT_OK_AND_ASSIGN(
+ DlfToken token,
+
DlfToken::FromJson(R"({"AccessKeyId":"ak","AccessKeySecret":"sk","SecurityToken":"sts",)"
+
R"("Expiration":"2025-04-16T05:44:46Z","Ignored":"value"})"));
+ ASSERT_EQ("ak", token.GetAccessKeyId());
+ ASSERT_EQ("sk", token.GetAccessKeySecret());
+ ASSERT_EQ(std::optional<std::string>("sts"), token.GetSecurityToken());
+ ASSERT_EQ(std::optional<int64_t>(1744782286000),
token.GetExpirationAtMillis());
+ ASSERT_FALSE(token.ShouldRefresh(FixedTime() + std::chrono::hours(1)));
+ ASSERT_TRUE(
+ token.ShouldRefresh(FixedTime() + std::chrono::hours(1) +
std::chrono::milliseconds(1)));
+
+ DlfToken permanent("ak", "sk", std::nullopt, std::nullopt);
+ ASSERT_FALSE(permanent.ShouldRefresh(FixedTime() +
std::chrono::hours(100000)));
+}
+
+TEST(DlfTokenTest, ParseFailureDoesNotLeakCredentials) {
+ const std::string secret = "STSSECRET_AKID_9999";
+ Status status =
+ DlfToken::FromJson(R"({"AccessKeyId":"ak","AccessKeySecret":")" +
secret).status();
+ ASSERT_FALSE(status.ok());
+ ASSERT_EQ(std::string::npos, status.ToString().find(secret));
+}
+
+TEST(DlfLocalFileTokenLoaderTest, LoadsTokenAndRedactsMalformedContent) {
+ std::unique_ptr<UniqueTestDirectory> test_dir =
UniqueTestDirectory::Create();
+ ASSERT_NE(nullptr, test_dir);
+ const std::string path = test_dir->Str() + "/dlf-token.json";
+ {
+ std::ofstream file(path);
+ file << R"({"AccessKeyId":"file-ak","AccessKeySecret":"file-sk",)"
+
R"("SecurityToken":"file-sts","Expiration":"2027-04-16T05:44:46Z"})";
+ }
+ DlfLocalFileTokenLoader loader(path, 1, std::chrono::milliseconds(0));
+ ASSERT_OK_AND_ASSIGN(DlfToken token, loader.LoadToken());
+ ASSERT_EQ("file-ak", token.GetAccessKeyId());
+ ASSERT_EQ(std::optional<std::string>("file-sts"),
token.GetSecurityToken());
+
+ std::map<std::string, std::string> options = {
+ {CatalogOptions::URI, "https://cn-hangzhou-vpc.dlf.aliyuncs.com"},
+ {CatalogOptions::TOKEN_PROVIDER, "dlf"},
+ {CatalogOptions::DLF_TOKEN_PATH, path},
+ {CatalogOptions::DLF_ACCESS_KEY_ID, "ignored-ak"},
+ {CatalogOptions::DLF_ACCESS_KEY_SECRET, "ignored-sk"}};
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<AuthProvider> provider,
AuthProvider::Create(options));
+ RestAuthParameter parameter = RestAuthParameter::Create("GET",
"/v1/config", {}, "");
+ ASSERT_OK_AND_ASSIGN(StringMap headers, provider->MergeAuthHeader({},
parameter));
+ ASSERT_NE(std::string::npos,
headers.at("Authorization").find("Credential=file-ak/"));
+
+ options[CatalogOptions::DLF_TOKEN_LOADER] = "local_file";
+ ASSERT_OK_AND_ASSIGN(provider, AuthProvider::Create(options));
+ ASSERT_OK(provider->MergeAuthHeader({}, parameter));
+
+ const std::string secret = "FILE_SECRET_9999";
+ {
+ std::ofstream file(path);
+ file << R"({"AccessKeyId":"ak","AccessKeySecret":")" << secret;
+ }
+ Status status = loader.LoadToken().status();
+ ASSERT_FALSE(status.ok());
+ ASSERT_EQ(std::string::npos, status.ToString().find(secret));
+}
+
+TEST(DlfEcsTokenLoaderTest, DiscoversRoleOnceAndRefreshesToken) {
+ const std::string metadata_url = "http://100.100.100.200/metadata/";;
+ auto http_client = std::make_unique<MockEcsHttpClient>(metadata_url);
+ MockEcsHttpClient* http_client_ptr = http_client.get();
+ DlfEcsTokenLoader loader(metadata_url, std::nullopt,
std::move(http_client));
+
+ ASSERT_OK_AND_ASSIGN(DlfToken first, loader.LoadToken());
+ ASSERT_OK_AND_ASSIGN(DlfToken second, loader.LoadToken());
+ ASSERT_EQ("ecs-ak", first.GetAccessKeyId());
+ ASSERT_EQ("ecs-ak", second.GetAccessKeyId());
+ ASSERT_EQ(1, http_client_ptr->GetRoleRequestCount());
+ ASSERT_EQ(2, http_client_ptr->GetTokenRequestCount());
+ ASSERT_EQ(180000, http_client_ptr->GetLastRequestTimeoutMillis());
+}
+
+TEST(DlfEcsTokenLoaderTest, ExplicitEmptyRoleUsesMetadataUrlAsTokenEndpoint) {
+ const std::string metadata_url = "http://100.100.100.200/metadata/token";;
+ auto http_client = std::make_unique<MockEcsHttpClient>(metadata_url,
/*direct_token=*/true);
+ MockEcsHttpClient* http_client_ptr = http_client.get();
+ DlfEcsTokenLoader loader(metadata_url, std::string(""),
std::move(http_client));
+
+ ASSERT_OK_AND_ASSIGN(DlfToken token, loader.LoadToken());
+ ASSERT_EQ("ecs-ak", token.GetAccessKeyId());
+ ASSERT_EQ(0, http_client_ptr->GetRoleRequestCount());
+ ASSERT_EQ(1, http_client_ptr->GetTokenRequestCount());
+ ASSERT_EQ(180000, http_client_ptr->GetLastRequestTimeoutMillis());
+}
+
+TEST(DlfEcsTokenLoaderTest, PreservesTransportFailureDetail) {
+ DlfEcsTokenLoader loader("http://100.100.100.200/metadata/token";,
std::string(""),
+ std::make_unique<FailingEcsHttpClient>());
+ Status status = loader.LoadToken().status();
+ ASSERT_NOK_WITH_MSG(status, "failed to request DLF credentials from ECS
metadata service");
+ ASSERT_NOK_WITH_MSG(status, "connection refused");
+}
+
+TEST(DlfAuthProviderTest, RefreshesWithinSafeWindow) {
+ std::atomic<int64_t> now_seconds{1744775086};
+ std::vector<DlfToken> tokens;
+ tokens.emplace_back("ak-1", "sk-1", std::nullopt, 1744782286000);
+ tokens.emplace_back("ak-2", "sk-2", std::nullopt, std::nullopt);
+ auto loader = std::make_unique<SequenceTokenLoader>(tokens);
+ SequenceTokenLoader* loader_ptr = loader.get();
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<DlfAuthProvider> provider,
+ DlfAuthProvider::FromTokenLoader(
+ std::move(loader), "https://cn-beijing-vpc.dlf.aliyuncs.com";,
"cn-beijing", "default",
+ [&] { return
std::chrono::system_clock::from_time_t(now_seconds.load()); }));
+ RestAuthParameter parameter = RestAuthParameter::Create("GET",
"/v1/config", {}, "");
+
+ ASSERT_OK_AND_ASSIGN(StringMap first, provider->MergeAuthHeader({},
parameter));
+ ASSERT_NE(std::string::npos,
first.at("Authorization").find("Credential=ak-1/"));
+ ASSERT_OK(provider->MergeAuthHeader({}, parameter));
+ ASSERT_EQ(1, loader_ptr->GetLoadCount());
+
+ now_seconds.fetch_add(3601);
+ ASSERT_OK_AND_ASSIGN(StringMap refreshed, provider->MergeAuthHeader({},
parameter));
+ ASSERT_NE(std::string::npos,
refreshed.at("Authorization").find("Credential=ak-2/"));
+ ASSERT_EQ(2, loader_ptr->GetLoadCount());
+}
+
+TEST(DlfAuthProviderTest, ConcurrentFirstUseLoadsTokenOnce) {
+ auto loader = std::make_unique<SequenceTokenLoader>(
+ std::vector<DlfToken>{
+ DlfToken("concurrent-ak", "concurrent-sk", std::nullopt,
std::nullopt)},
+ std::chrono::milliseconds(10));
+ SequenceTokenLoader* loader_ptr = loader.get();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<DlfAuthProvider> provider,
+ DlfAuthProvider::FromTokenLoader(std::move(loader),
+
"https://cn-beijing-vpc.dlf.aliyuncs.com";,
+ "cn-beijing",
"default", FixedTime));
+ RestAuthParameter parameter = RestAuthParameter::Create("GET",
"/v1/config", {}, "");
+ std::vector<std::thread> threads;
+ std::vector<Status> statuses;
+ std::mutex statuses_mutex;
+ for (int32_t i = 0; i < 16; ++i) {
+ threads.emplace_back([&] {
+ Status status = provider->MergeAuthHeader({}, parameter).status();
+ std::scoped_lock lock(statuses_mutex);
+ statuses.push_back(status);
+ });
+ }
+ for (std::thread& thread : threads) {
+ thread.join();
+ }
+ ASSERT_EQ(16, statuses.size());
+ for (const Status& status : statuses) {
+ ASSERT_OK(status);
+ }
+ ASSERT_EQ(1, loader_ptr->GetLoadCount());
+}
+
+TEST(DlfAuthProviderTest, SelectsEndpointSignerAndCredentialSource) {
+ std::map<std::string, std::string> options = {
+ {CatalogOptions::URI, "https://dlfnext.cn-hangzhou.aliyuncs.com"},
+ {CatalogOptions::TOKEN_PROVIDER, "dlf"},
+ {CatalogOptions::DLF_ACCESS_KEY_ID, "ak"},
+ {CatalogOptions::DLF_ACCESS_KEY_SECRET, "sk"},
+ {CatalogOptions::DLF_SECURITY_TOKEN, "sts"}};
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<AuthProvider> provider,
AuthProvider::Create(options));
+ RestAuthParameter parameter = RestAuthParameter::Create("GET",
"/v1/config", {}, "");
+ ASSERT_OK_AND_ASSIGN(StringMap headers,
provider->MergeAuthHeader({{"Authorization", "old"},
+
{"x-acs-version", "old"},
+
{"custom-header", "kept"}},
+
parameter));
+ ASSERT_NE(std::string::npos, headers.at("Authorization").find("acs ak:"));
+ ASSERT_EQ("2026-01-18", headers.at("x-acs-version"));
+ ASSERT_EQ("sts", headers.at("x-acs-security-token"));
+ ASSERT_EQ("kept", headers.at("custom-header"));
+ ASSERT_FALSE(provider->AllowsRedirects());
+
+ BearTokenAuthProvider bear_provider("token");
+ ASSERT_TRUE(bear_provider.AllowsRedirects());
+
+ ASSERT_EQ("openapi",
DlfAuthProvider::ParseSigningAlgorithmFromUri(options.at("uri")));
+ ASSERT_EQ("default", DlfAuthProvider::ParseSigningAlgorithmFromUri(
+ "https://cn-hangzhou-vpc.dlf.aliyuncs.com";));
+ ASSERT_OK_AND_ASSIGN(std::string region,
+
DlfAuthProvider::ParseRegionFromUri(options.at("uri")));
+ ASSERT_EQ("cn-hangzhou", region);
+ ASSERT_OK_AND_ASSIGN(std::string host,
+
DlfAuthProvider::ExtractHost("https://example.com:8443/prefix";));
+ ASSERT_EQ("example.com:8443", host);
+}
+
+TEST(DlfAuthProviderTest, NormalizesSigningHostLikeTransport) {
+ std::map<std::string, std::string> options = {
+ {CatalogOptions::URI, " https://dlfnext.cn-hangzhou.aliyuncs.com "},
+ {CatalogOptions::TOKEN_PROVIDER, "dlf"},
+ {CatalogOptions::DLF_ACCESS_KEY_ID, "ak"},
+ {CatalogOptions::DLF_ACCESS_KEY_SECRET, "sk"}};
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<AuthProvider> provider,
AuthProvider::Create(options));
+ RestAuthParameter parameter = RestAuthParameter::Create("GET",
"/v1/config", {}, "");
+ ASSERT_OK_AND_ASSIGN(StringMap headers, provider->MergeAuthHeader({},
parameter));
+ ASSERT_EQ("dlfnext.cn-hangzhou.aliyuncs.com", headers.at("Host"));
+}
+
+TEST(DlfAuthProviderTest, RejectsIncompleteOrUnknownConfiguration) {
+ const std::map<std::string, std::string> base = {
+ {CatalogOptions::URI, "https://dlfnext.cn-hangzhou.aliyuncs.com"},
+ {CatalogOptions::TOKEN_PROVIDER, "dlf"}};
+ ASSERT_NOK_WITH_MSG(AuthProvider::Create(base).status(), "token path or
access key");
+
+ std::map<std::string, std::string> options = base;
+ options[CatalogOptions::DLF_TOKEN_LOADER] = "ecs";
+ options[CatalogOptions::DLF_TOKEN_ECS_METADATA_URL] = "http://metadata/";;
+ options[CatalogOptions::DLF_TOKEN_ECS_ROLE_NAME] = "role";
+ ASSERT_OK(AuthProvider::Create(options));
+
+ options = base;
+ options[CatalogOptions::DLF_TOKEN_LOADER] = "unknown";
+ ASSERT_NOK_WITH_MSG(AuthProvider::Create(options).status(), "unsupported
DLF token loader");
+
+ options = base;
+ options[CatalogOptions::DLF_ACCESS_KEY_ID] = "ak";
+ options[CatalogOptions::DLF_ACCESS_KEY_SECRET] = "sk";
+ options[CatalogOptions::DLF_SIGNING_ALGORITHM] = "unknown";
+ ASSERT_NOK_WITH_MSG(AuthProvider::Create(options).status(),
+ "unsupported DLF signing algorithm");
+
+ options[CatalogOptions::DLF_SIGNING_ALGORITHM] = "default";
+ options[CatalogOptions::URI] = "http://127.0.0.1:8080";;
+ ASSERT_NOK_WITH_MSG(AuthProvider::Create(options).status(), "DLF region");
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/rest/rest_api.cpp b/src/paimon/rest/rest_api.cpp
index 393a0938..437484ef 100644
--- a/src/paimon/rest/rest_api.cpp
+++ b/src/paimon/rest/rest_api.cpp
@@ -81,11 +81,12 @@ Result<std::unique_ptr<RestApi>> RestApi::Create(const
std::map<std::string, std
RestAuthParameter::Create("GET", ResourcePaths::Config(),
query_params, "");
PAIMON_ASSIGN_OR_RAISE(StringMap headers,
auth_provider->MergeAuthHeader(base_headers,
auth_parameter));
- PAIMON_ASSIGN_OR_RAISE(
- RestHttpClient::Response response,
- client->Execute("GET", ResourcePaths::Config(), query_params,
headers, ""));
+ bool follow_redirects = auth_provider->AllowsRedirects();
+ PAIMON_ASSIGN_OR_RAISE(RestHttpClient::Response response,
+ client->Execute("GET", ResourcePaths::Config(),
query_params,
+ headers, "", follow_redirects));
if (!response.IsSuccessful()) {
- return ErrorToStatus(response);
+ return ErrorToStatus(response, follow_redirects);
}
ConfigResponse config;
PAIMON_RETURN_NOT_OK(ParseResponseBody(response.body,
ResourcePaths::Config(), &config));
@@ -109,7 +110,19 @@ Result<std::unique_ptr<RestApi>> RestApi::Create(const
std::map<std::string, std
ResourcePaths(prefix)));
}
-Status RestApi::ErrorToStatus(const RestHttpClient::Response& response) {
+Status RestApi::ErrorToStatus(const RestHttpClient::Response& response, bool
follow_redirects) {
+ if (!follow_redirects && response.code >= 300 && response.code <
HttpStatus::kBadRequest) {
+ std::string message = fmt::format(
+ "rest endpoint returned redirect status {}, which is not followed
for "
+ "signed requests",
+ response.code);
+ std::string request_id = RestUtil::ExtractRequestId(response.headers);
+ if (request_id != RestUtil::kUnknownRequestId) {
+ message += fmt::format(" requestId:{}", request_id);
+ }
+ return Status::IOError(message).WithDetail(
+ std::make_shared<RestErrorDetail>(response.code));
+ }
// The code of the parsed error body takes precedence over the http
status, which
// a gateway may have rewritten.
int64_t code = response.code;
@@ -188,10 +201,12 @@ Result<RestHttpClient::Response> RestApi::Execute(
}
PAIMON_ASSIGN_OR_RAISE(StringMap headers,
auth_provider_->MergeAuthHeader(request_headers,
auth_parameter));
- PAIMON_ASSIGN_OR_RAISE(RestHttpClient::Response response,
- client_->Execute(method, path, query_params,
headers, body));
+ bool follow_redirects = auth_provider_->AllowsRedirects();
+ PAIMON_ASSIGN_OR_RAISE(
+ RestHttpClient::Response response,
+ client_->Execute(method, path, query_params, headers, body,
follow_redirects));
if (!response.IsSuccessful()) {
- return ErrorToStatus(response);
+ return ErrorToStatus(response, follow_redirects);
}
return response;
}
diff --git a/src/paimon/rest/rest_api.h b/src/paimon/rest/rest_api.h
index c2394ced..f23fe135 100644
--- a/src/paimon/rest/rest_api.h
+++ b/src/paimon/rest/rest_api.h
@@ -105,9 +105,11 @@ class RestApi {
/// Maps a non-successful http response to a status: 404 becomes
`NotExist`, 409
/// becomes `Exist`, 400 becomes `Invalid`, 501 becomes `NotImplemented`
and the
- /// other codes become `IOError`. The status carries a `RestErrorDetail`
with the
- /// mapped code.
- static Status ErrorToStatus(const RestHttpClient::Response& response);
+ /// other codes become `IOError`. A redirect returned while
`follow_redirects` is
+ /// false is reported as deliberately rejected for a signed request. The
status
+ /// carries a `RestErrorDetail` with the mapped code.
+ static Status ErrorToStatus(const RestHttpClient::Response& response,
+ bool follow_redirects = true);
private:
RestApi(std::unique_ptr<RestHttpClient> client,
std::unique_ptr<AuthProvider> auth_provider,
diff --git a/src/paimon/rest/rest_auth.cpp b/src/paimon/rest/rest_auth.cpp
index 65d6d755..30a1869a 100644
--- a/src/paimon/rest/rest_auth.cpp
+++ b/src/paimon/rest/rest_auth.cpp
@@ -20,6 +20,7 @@
#include "paimon/catalog_options.h"
#include "paimon/common/utils/string_utils.h"
#include "paimon/common/utils/url_utils.h"
+#include "paimon/rest/dlf_auth.h"
namespace paimon {
@@ -52,8 +53,8 @@ Result<std::unique_ptr<AuthProvider>> AuthProvider::Create(
return Status::Invalid(fmt::format("option '{}' must be configured for
the rest catalog",
CatalogOptions::TOKEN_PROVIDER));
}
- // Matched leniently in lower case; other clients may match the provider
name
- // case-sensitively, so only the exact "bear" spelling is portable.
+ // Matched leniently in lower case; other clients may match provider names
+ // case-sensitively, so the exact "bear" and "dlf" spellings are portable.
std::string provider = StringUtils::ToLowerCase(provider_iter->second);
if (provider == "bear") {
auto token_iter = options.find(CatalogOptions::TOKEN);
@@ -64,8 +65,11 @@ Result<std::unique_ptr<AuthProvider>> AuthProvider::Create(
}
return std::make_unique<BearTokenAuthProvider>(token_iter->second);
}
- return Status::NotImplemented(
- fmt::format("unsupported token provider: {}, only 'bear' is supported
for now", provider));
+ if (provider == "dlf") {
+ return DlfAuthProvider::Create(options);
+ }
+ return Status::NotImplemented(fmt::format(
+ "unsupported token provider: {}, supported providers are 'bear' and
'dlf'", provider));
}
} // namespace paimon
diff --git a/src/paimon/rest/rest_auth.h b/src/paimon/rest/rest_auth.h
index 22da08d9..72120d24 100644
--- a/src/paimon/rest/rest_auth.h
+++ b/src/paimon/rest/rest_auth.h
@@ -52,6 +52,11 @@ class AuthProvider {
const std::map<std::string, std::string>& base_header,
const RestAuthParameter& parameter) const = 0;
+ /// Whether the transport may follow a redirect without regenerating auth
headers.
+ virtual bool AllowsRedirects() const {
+ return true;
+ }
+
/// Creates the provider configured by `CatalogOptions::TOKEN_PROVIDER`.
static Result<std::unique_ptr<AuthProvider>> Create(
const std::map<std::string, std::string>& options);
diff --git a/src/paimon/rest/rest_catalog_test.cpp
b/src/paimon/rest/rest_catalog_test.cpp
index fda03cc4..f1bb93aa 100644
--- a/src/paimon/rest/rest_catalog_test.cpp
+++ b/src/paimon/rest/rest_catalog_test.cpp
@@ -414,7 +414,7 @@ TEST_F(RestCatalogTest, CreateRejectsInvalidOptions) {
ASSERT_NOK_WITH_MSG(CreateRestCatalog().status(), "'token.provider' must
be configured");
options_ = valid_options;
- options_[CatalogOptions::TOKEN_PROVIDER] = "dlf";
+ options_[CatalogOptions::TOKEN_PROVIDER] = "unsupported";
Status unsupported_provider = CreateRestCatalog().status();
ASSERT_TRUE(unsupported_provider.IsNotImplemented()) <<
unsupported_provider.ToString();
ASSERT_NOK_WITH_MSG(unsupported_provider, "unsupported token provider");
@@ -991,6 +991,12 @@ TEST(RestApiErrorTest, ErrorToStatus) {
ASSERT_NOK_WITH_MSG(RestApi::ErrorToStatus(response), "rest request failed
with code 429");
response.code = 418;
ASSERT_NOK_WITH_MSG(RestApi::ErrorToStatus(response), "rest request failed
with code 418");
+
+ response.code = 302;
+ Status signed_redirect = RestApi::ErrorToStatus(response,
/*follow_redirects=*/false);
+ ASSERT_NOK_WITH_MSG(signed_redirect, "redirect status 302");
+ ASSERT_NOK_WITH_MSG(signed_redirect, "not followed for signed requests");
+ ASSERT_EQ(302,
checked_pointer_cast<RestErrorDetail>(signed_redirect.detail())->GetCode());
}
TEST(RestApiErrorTest, MalformedSuccessBodyFails) {
diff --git a/src/paimon/rest/rest_http_client.cpp
b/src/paimon/rest/rest_http_client.cpp
index 52e6417e..10695818 100644
--- a/src/paimon/rest/rest_http_client.cpp
+++ b/src/paimon/rest/rest_http_client.cpp
@@ -249,7 +249,7 @@ std::string RestHttpClient::BuildQueryString(
Result<RestHttpClient::Response> RestHttpClient::ExecuteOnce(
const std::string& method, const std::string& url,
const std::map<std::string, std::string>& headers, const std::string& body,
- bool* transport_retriable) const {
+ bool follow_redirects, bool* transport_retriable) const {
CURL* curl = handle_pool_->Acquire();
if (curl == nullptr) {
return Status::IOError("failed to create curl handle");
@@ -271,10 +271,11 @@ Result<RestHttpClient::Response>
RestHttpClient::ExecuteOnce(
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response.body);
curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, WriteHeaderCallback);
curl_easy_setopt(curl, CURLOPT_HEADERDATA, &response.headers);
- // Follow redirects transparently, restricted to http(s) targets. Without
+ // Redirects can be disabled for auth headers whose signatures are bound
to the
+ // original request. Otherwise they are restricted to http(s) targets.
Without
// CURLOPT_POSTREDIR a 301/302 would replay a body-carrying request as a
bodyless
// GET. A 303 is left to become a GET, which is what it is defined to mean.
- curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
+ curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, follow_redirects ? 1L : 0L);
curl_easy_setopt(curl, CURLOPT_MAXREDIRS, 50L);
curl_easy_setopt(curl, CURLOPT_POSTREDIR,
static_cast<long>(CURL_REDIR_POST_301 | //
NOLINT(runtime/int)
@@ -399,7 +400,8 @@ std::optional<int64_t>
RestHttpClient::GetRetryDelayMs(int32_t execution_count,
Result<RestHttpClient::Response> RestHttpClient::Execute(
const std::string& method, const std::string& path,
const std::map<std::string, std::string>& query_params,
- const std::map<std::string, std::string>& headers, const std::string&
body) const {
+ const std::map<std::string, std::string>& headers, const std::string& body,
+ bool follow_redirects) const {
if (method != "GET" && method != "POST" && method != "DELETE") {
return Status::Invalid(fmt::format("unsupported http method: {}",
method));
}
@@ -423,7 +425,8 @@ Result<RestHttpClient::Response> RestHttpClient::Execute(
while (true) {
execution_count++;
bool transport_retriable = false;
- Result<Response> result = ExecuteOnce(method, url, headers, body,
&transport_retriable);
+ Result<Response> result =
+ ExecuteOnce(method, url, headers, body, follow_redirects,
&transport_retriable);
bool retriable;
if (result.ok()) {
retriable = IsRetriableCode(result.value().code);
diff --git a/src/paimon/rest/rest_http_client.h
b/src/paimon/rest/rest_http_client.h
index 07712795..412e9853 100644
--- a/src/paimon/rest/rest_http_client.h
+++ b/src/paimon/rest/rest_http_client.h
@@ -50,8 +50,8 @@ struct HttpStatus {
/// response header (delta-seconds or HTTP-date form). A backoff sleep is
bounded by
/// `retry_max_delay_ms` and the whole request by `retry_timeout_ms`; a
`Retry-After`
/// beyond the remaining budget stops retrying rather than shortening the
sleep.
-/// Redirects to http(s) targets are followed transparently, keeping the
method and
-/// body of POST/DELETE requests.
+/// Redirects to http(s) targets are followed by default, keeping the method
and body
+/// of POST/DELETE requests; callers can disable them for request-bound
signatures.
class RestHttpClient {
public:
struct Config {
@@ -96,11 +96,12 @@ class RestHttpClient {
/// final response, which may carry a non-2xx code, or an error status
when the
/// request could not be transported at all. Only transient transport
errors (an
/// established connection breaking mid-request or a truncated response
body) are
- /// retried; every other transport failure fails immediately.
+ /// retried; every other transport failure fails immediately.
`follow_redirects`
+ /// must be false when authentication headers are bound to the original
request.
Result<Response> Execute(const std::string& method, const std::string&
path,
const std::map<std::string, std::string>&
query_params,
const std::map<std::string, std::string>& headers,
- const std::string& body) const;
+ const std::string& body, bool follow_redirects =
true) const;
const std::string& GetBaseUri() const {
return base_uri_;
@@ -135,7 +136,8 @@ class RestHttpClient {
/// may be retried; see `Execute` for which kinds are not.
Result<Response> ExecuteOnce(const std::string& method, const std::string&
url,
const std::map<std::string, std::string>&
headers,
- const std::string& body, bool*
transport_retriable) const;
+ const std::string& body, bool
follow_redirects,
+ bool* transport_retriable) const;
std::optional<int64_t> GetRetryDelayMs(int32_t execution_count, const
Response* response,
int64_t remaining_budget_ms) const;
diff --git a/src/paimon/rest/rest_http_client_test.cpp
b/src/paimon/rest/rest_http_client_test.cpp
index 34d07fe6..82ea9e2a 100644
--- a/src/paimon/rest/rest_http_client_test.cpp
+++ b/src/paimon/rest/rest_http_client_test.cpp
@@ -489,6 +489,25 @@ TEST(RestHttpClientTest, RedirectIsFollowed) {
ASSERT_EQ(0, response.headers.count("location"));
}
+TEST(RestHttpClientTest, RedirectCanBeDisabledForSignedRequests) {
+ std::atomic<int32_t> request_count{0};
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<MockRestServer> server,
+ MockRestServer::Start([&](const
MockRestServer::Request& request) {
+ request_count++;
+ MockRestServer::Response response;
+ response.code = 302;
+ response.headers["Location"] = "/v1/config";
+ return response;
+ }));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RestHttpClient> client,
+ RestHttpClient::Create(server->GetBaseUri()));
+ ASSERT_OK_AND_ASSIGN(RestHttpClient::Response response,
+ client->Execute("GET", "/old", {},
{{"x-acs-security-token", "secret"}},
+ "", /*follow_redirects=*/false));
+ ASSERT_EQ(302, response.code);
+ ASSERT_EQ(1, request_count.load());
+}
+
TEST(RestHttpClientTest, PostRedirectKeepsMethodAndBody) {
std::mutex mutex;
MockRestServer::Request last_request;
