| /* |
| * 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 <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"; |
| |
| Result<std::string> RequiredNonEmptyOption(const std::map<std::string, std::string>& options, |
| const std::string& key) { |
| Result<std::string> value = OptionsUtils::GetNonEmptyValueFromMap(options, key); |
| if (!value.ok()) { |
| 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; |
| StringUtils::Trim(&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_)); |
| StringUtils::Trim(&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 |
