SteNicholas commented on code in PR #244:
URL: https://github.com/apache/paimon-cpp/pull/244#discussion_r3849088346
##########
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";
Review Comment:
**Scope: this looks like an unrelated change riding along.**
Commit `bb1f37d` ("build: use archives for retired Apache dependencies")
changes `cmake_modules/arrow.diff` and `cmake_modules/orc.diff`, which have
nothing to do with DLF authentication, and the PR description does not mention
it at all (it says "No storage-format or REST protocol changes" and stops
there). `AGENTS.md` asks to keep a change focused and to leave third-party
patches alone unless the task requires it.
Could this be split into its own PR? If it has to stay here because CI
cannot build without it, please at least say so in the description.
One small thing to double-check either way: inserting this hunk at line 814
leaves the *new-side* line numbers of the following pre-existing `@@ -983,6
+983,11 @@` hunk stale by one. `git apply` matches on content so this normally
still applies, but it is worth confirming once in a clean build that the Arrow
patch is still accepted.
##########
src/paimon/rest/dlf_auth_test.cpp:
##########
@@ -0,0 +1,431 @@
+/*
+ * 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};
+};
+
+} // namespace
+
+TEST(DlfDefaultSignerTest, SignsJavaCompatibleRequest) {
+ DlfDefaultSigner signer("cn-beijing");
+ const std::string body = R"({"name":"t1"})";
+ DlfToken token("YourAccessKeyId", "YourAccessKeySecret", "securityToken");
+ 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",
Review Comment:
**Please add Java's golden vector so cross-language compatibility is
actually pinned.**
The expected strings here are newly minted for this PR. I verified
out-of-band that they are consistent with Java's algorithm, but nothing in the
suite locks that: a canonicalization regression (signed-header set, the
empty-query newline, parameter trimming) would keep these tests green while
every real request started coming back 403.
Porting `DLFAuthSignatureTest#testGetAuthorization` verbatim would fix that,
and it passes against this implementation as-is:
- method `POST`, path `/v1/paimon/databases`
- parameters `k1=v1`, `k2=v2`
- body `{"name":"database","options":{"a":"b"}}`
- region `cn-hangzhou`, `x-dlf-date` `20231203T121212Z`
- AK `access-key-id` / `access-key-secret`, security token `securityToken`
expected:
```
DLF4-HMAC-SHA256
Credential=access-key-id/20231203/cn-hangzhou/DlfNext/aliyun_v4_request,Signature=c72caf1d40b55b1905d891ee3e3de48a2f8bebefa7e39e4f277acc93c269c5e3
```
A matching golden case for `DlfOpenApiSigner` would be good too, though Java
has no published vector for it, so the one here is fine as the reference.
##########
src/paimon/rest/dlf_auth.h:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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 = std::nullopt,
+ const std::optional<int64_t>& expiration_at_millis =
std::nullopt);
+
+ 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:
+ explicit DlfLocalFileTokenLoader(
+ const std::string& token_file_path, int32_t max_attempts = 5,
+ std::chrono::milliseconds retry_delay = std::chrono::seconds(1));
+
+ 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 =
std::chrono::system_clock::now);
+
+ 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 = std::chrono::system_clock::now);
+
+ 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 {
Review Comment:
**A 3xx from a DLF endpoint now surfaces as a misleading error.**
With redirects disabled, a `302` is no longer followed, so it reaches
`RestApi::ErrorToStatus`, misses every `case` and falls through to `default`,
producing:
```
rest request failed with code 302: unparsable error response body (http
status 302)
```
That reads like a malformed server response, when the actual cause is that
we deliberately refused to follow a redirect. Since this PR is what introduces
the non-following behaviour, it would be good to add an explicit branch in
`ErrorToStatus` (or a check on the caller side) for a 3xx when the provider
reports `AllowsRedirects() == false`, saying something like "the endpoint
returned a redirect, which is not followed for signed requests".
The behaviour itself is right — just the diagnostics need to catch up with
it.
##########
src/paimon/rest/dlf_auth.cpp:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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/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);
+}
+
+std::optional<std::string> FindOption(const std::map<std::string,
std::string>& options,
+ const std::string& key) {
+ auto iter = options.find(key);
+ if (iter == options.end()) {
+ return std::nullopt;
+ }
+ return iter->second;
+}
+
+Result<std::string> RequiredNonEmptyOption(const std::map<std::string,
std::string>& options,
+ const std::string& key) {
+ std::optional<std::string> value = FindOption(options, 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 defined(_WIN32)
+ if (gmtime_s(&utc, &seconds) != 0) {
+ return Status::Invalid("failed to convert DLF signing time to UTC");
+ }
+#else
+ if (gmtime_r(&seconds, &utc) == nullptr) {
+ return Status::Invalid("failed to convert DLF signing time to UTC");
+ }
+#endif
+ 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);
Review Comment:
**The `_WIN32` branches in this file are dead code and cannot work as
written.**
`timegm` is called unconditionally here, but it does not exist on MSVC (the
equivalent is `_mkgmtime`). So the `#if defined(_WIN32)` / `gmtime_s` branches
at L128-136 and L184-192 cannot actually make this file compile on Windows —
they only give the impression that Windows is handled.
Meanwhile `docs/source/building.rst` states "Windows is not supported for
now", and `include/paimon/visibility.h` is the only other `_WIN32` user in the
tree, so this is not an established pattern here either.
Suggest dropping the `_WIN32` branches and just using `gmtime_r` + `timegm`,
consistent with the rest of the repository. (Or complete them with `_mkgmtime`,
if Windows support is actually intended — but that seems out of scope for this
PR.)
##########
src/paimon/rest/dlf_auth.cpp:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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/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);
+}
+
+std::optional<std::string> FindOption(const std::map<std::string,
std::string>& options,
+ const std::string& key) {
+ auto iter = options.find(key);
+ if (iter == options.end()) {
+ return std::nullopt;
+ }
+ return iter->second;
+}
+
+Result<std::string> RequiredNonEmptyOption(const std::map<std::string,
std::string>& options,
+ const std::string& key) {
+ std::optional<std::string> value = FindOption(options, 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 defined(_WIN32)
+ if (gmtime_s(&utc, &seconds) != 0) {
+ return Status::Invalid("failed to convert DLF signing time to UTC");
+ }
+#else
+ if (gmtime_r(&seconds, &utc) == nullptr) {
+ return Status::Invalid("failed to convert DLF signing time to UTC");
+ }
+#endif
+ 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 defined(_WIN32)
+ if (timestamp == static_cast<std::time_t>(-1) || gmtime_s(&verified,
×tamp) != 0) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+#else
+ if (timestamp == static_cast<std::time_t>(-1) || gmtime_r(×tamp,
&verified) == nullptr) {
+ return Status::Invalid("invalid DLF token expiration");
+ }
+#endif
+ 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");
Review Comment:
**The underlying transport error is dropped, which makes ECS failures very
hard to diagnose.**
DNS failure, connection refused, TLS error and the three-minute timeout all
collapse into this one message, so an operator on an ECS instance has no way to
tell "no RAM role attached" from "metadata service unreachable" from "timed
out".
The secret in this exchange is the *response body*, not the URL or libcurl's
error string — the metadata endpoint is a fixed link-local address, and `Get()`
already refuses to echo the body. Java keeps the message here
(`DLFECSTokenLoader.getRole` wraps it as `"get role failed, error : " +
e.getMessage()`).
Suggest propagating the failure detail, e.g. `return Status::IOError("failed
to request DLF credentials from ECS metadata service: ",
response.status().message());`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
