fix

Author: lucasfang

docs/source/user_guide/catalog.rst

@@ -404,6 +404,14 @@ the factory is called with are those of the accesses to authenticate, which is
 where an implementation reads its own configuration, such as the address of the
 token service, from.
 
+Merge the credentials into your file system's options with
+``CredentialProvider::MergeOptionsWithCredentials`` rather than by hand, so they are
+shaped the same way the built-in data token file system shapes them. The default
+overlays the credentials key by key over the base options, mirroring the Java
+client. A provider that knows the file system its credentials are for overrides
+``MergeOptionsWithCredentials`` to normalize option aliases or clear the stale
+bucket-scoped variants the fresh credentials replace.
+
 All factories share one identifier space, so an identifier a file system factory
 already takes — ``oss``, ``s3``, ``local``, ``jindo`` — would replace it; name the
 provider after where its credentials come from instead.

include/paimon/fs/credential_provider.h

@@ -46,9 +46,29 @@ class PAIMON_EXPORT CredentialProvider {
     virtual ~CredentialProvider() = default;
 
     /// Returns the credentials to sign an access with, reloading them when they are about
-    /// to expire. The keys are file system options, e.g. "fs.oss.accessKeyId" or
-    /// "fs.oss.securityToken", so a caller merges them over its own file system options.
+    /// to expire. The keys are file system options, so a caller merges them over its own
+    /// file system options with `MergeOptionsWithCredentials`.
     virtual Result<std::map<std::string, std::string>> GetCredentials() const = 0;
+
+    /// Merges the issued credentials into the file system options a delegate is built from.
+    /// This is the canonical way to shape credentials into the options of an access: a
+    /// caller that brings its own file system calls it so the credentials are applied the
+    /// same way the built-in data token file system applies them.
+    ///
+    /// The default overlays the credentials key by key over `base_options`, so they win
+    /// wherever they overlap, mirroring the Java client and staying scheme-agnostic. A
+    /// provider that knows the file system its credentials are for overrides this to
+    /// normalize option aliases or clear stale bucket-scoped variants the credentials
+    /// replace.
+    virtual std::map<std::string, std::string> MergeOptionsWithCredentials(
+        const std::map<std::string, std::string>& base_options,
+        const std::map<std::string, std::string>& credentials) const {
+        std::map<std::string, std::string> merged = base_options;
+        for (const auto& [key, value] : credentials) {
+            merged[key] = value;
+        }
+        return merged;
+    }
 };
 
 }  // namespace paimon

src/paimon/rest/rest_catalog.cpp

@@ -483,7 +483,7 @@ Result<std::shared_ptr<FileSystem>> RestCatalog::GetTableFileSystem(
     // from the catalog options, which every table agrees on, so a rotation rebuilds them.
     const std::map<std::string, std::string>& catalog_options = api_->GetMergedOptions();
     std::shared_ptr<RestCredentialProvider> provider =
-        std::make_shared<RestCredentialProvider>(api_, catalog_options, load_identifier);
+        std::make_shared<RestCredentialProvider>(api_, load_identifier);
     return std::make_shared<RestTokenFileSystem>(std::move(provider), catalog_options,
                                                  token_fs_cache_, fs_scheme_to_identifier_map_);
 }

src/paimon/rest/rest_credential_provider.cpp

@@ -18,19 +18,40 @@
 
 #include "paimon/rest/rest_credential_provider.h"
 
+#include <algorithm>
 #include <chrono>
 #include <functional>
 #include <mutex>
+#include <set>
 #include <utility>
 
 #include "paimon/catalog_options.h"
+#include "paimon/common/utils/string_utils.h"
 
 namespace paimon {
 
 namespace {
 /// Declared here rather than taken from the OSS file system, which is an optional build
 /// component this module must not depend on.
 constexpr const char kOssEndpointOption[] = "fs.oss.endpoint";
+constexpr const char kOssOptionPrefix[] = "fs.oss.";
+constexpr const char kOssBucketPrefix[] = "fs.oss.bucket.";
+// The two OSS names for the STS token: the backend reads securityToken first and only
+// consults sessionToken when it is empty, so a token under either name has to clear the
+// catalog value under both.
+constexpr const char kOssSecurityTokenSuffix[] = "securityToken";
+constexpr const char kOssSessionTokenSuffix[] = "sessionToken";
+
+/// The suffix of a flat OSS option, i.e. what follows "fs.oss." for a key like
+/// "fs.oss.accessKeyId"; empty for a bucket-scoped ("fs.oss.bucket.<bucket>.<suffix>") or a
+/// non-OSS key.
+std::string FlatOssOptionSuffix(const std::string& key) {
+    if (!StringUtils::StartsWith(key, kOssOptionPrefix) ||
+        StringUtils::StartsWith(key, kOssBucketPrefix)) {
+        return "";
+    }
+    return key.substr(std::string(kOssOptionPrefix).size());
+}
 }  // namespace
 
 size_t RestToken::Hash::operator()(const RestToken& rest_token) const {
@@ -42,11 +63,9 @@ size_t RestToken::Hash::operator()(const RestToken& rest_token) const {
     return result;
 }
 
-RestCredentialProvider::RestCredentialProvider(
-    const std::shared_ptr<RestApi>& api, const std::map<std::string, std::string>& catalog_options,
-    const Identifier& identifier, Clock clock)
+RestCredentialProvider::RestCredentialProvider(const std::shared_ptr<RestApi>& api,
+                                               const Identifier& identifier, Clock clock)
     : api_(api),
-      catalog_options_(catalog_options),
       identifier_(identifier),
       clock_(std::move(clock)),
       logger_(Logger::GetLogger("RestCredentialProvider")) {}
@@ -60,13 +79,58 @@ bool RestCredentialProvider::ShouldRefresh() const {
     return token_->expires_at_millis - now_millis < RestApi::kTokenExpirationSafeTimeMillis;
 }
 
-std::map<std::string, std::string> RestCredentialProvider::ApplyDlfEndpointOverride(
-    const std::map<std::string, std::string>& token) const {
-    std::map<std::string, std::string> merged = token;
+std::map<std::string, std::string> RestCredentialProvider::MergeOptionsWithCredentials(
+    const std::map<std::string, std::string>& base_options,
+    const std::map<std::string, std::string>& credentials) const {
+    std::map<std::string, std::string> merged = base_options;
+    for (const auto& [key, value] : credentials) {
+        merged[key] = value;
+    }
+    // The OSS backend resolves a bucket-scoped option ("fs.oss.bucket.<b>.<suffix>") ahead of the
+    // flat one and reads "fs.oss.securityToken" ahead of its "fs.oss.sessionToken" alias, so a
+    // stale catalog value can shadow a credential the token just refreshed. Drop the stale
+    // variants of every credential the token supplies -- including both security-token aliases
+    // when it supplies either -- so the issued credential is the one that wins.
+    std::set<std::string> refreshed_suffixes;
+    bool refreshed_security_token = false;
+    for (const auto& [key, value] : credentials) {
+        std::string suffix = FlatOssOptionSuffix(key);
+        if (suffix.empty()) {
+            continue;
+        }
+        refreshed_suffixes.insert(suffix);
+        refreshed_security_token = refreshed_security_token || suffix == kOssSecurityTokenSuffix ||
+                                   suffix == kOssSessionTokenSuffix;
+    }
+    if (refreshed_security_token) {
+        refreshed_suffixes.insert(kOssSecurityTokenSuffix);
+        refreshed_suffixes.insert(kOssSessionTokenSuffix);
+    }
+    for (auto it = merged.begin(); it != merged.end();) {
+        const std::string& key = it->first;
+        // A credential the token itself supplies is authoritative and is never dropped, even when
+        // it is bucket-scoped or the alias of another key the token carries.
+        bool token_supplied = credentials.find(key) != credentials.end();
+        bool stale_bucket_scoped = !token_supplied &&
+                                   StringUtils::StartsWith(key, kOssBucketPrefix) &&
+                                   std::any_of(refreshed_suffixes.begin(), refreshed_suffixes.end(),
+                                               [&](const std::string& suffix) {
+                                                   return StringUtils::EndsWith(key, "." + suffix);
+                                               });
+        bool stale_security_token_alias =
+            !token_supplied && refreshed_security_token &&
+            (key == std::string(kOssOptionPrefix) + kOssSecurityTokenSuffix ||
+             key == std::string(kOssOptionPrefix) + kOssSessionTokenSuffix);
+        if (stale_bucket_scoped || stale_security_token_alias) {
+            it = merged.erase(it);
+        } else {
+            ++it;
+        }
+    }
     // The DLF OSS endpoint overrides the standard one, since the credentials are issued
     // for the DLF endpoint rather than for the endpoint the catalog was configured with.
-    auto dlf_oss_endpoint = catalog_options_.find(CatalogOptions::DLF_OSS_ENDPOINT);
-    if (dlf_oss_endpoint != catalog_options_.end() && !dlf_oss_endpoint->second.empty()) {
+    auto dlf_oss_endpoint = base_options.find(CatalogOptions::DLF_OSS_ENDPOINT);
+    if (dlf_oss_endpoint != base_options.end() && !dlf_oss_endpoint->second.empty()) {
         merged[kOssEndpointOption] = dlf_oss_endpoint->second;
     }
     return merged;
@@ -80,8 +144,7 @@ Status RestCredentialProvider::RefreshToken() const {
                     identifier_.ToString().c_str(),
                     static_cast<int64_t>(response.GetExpiresAtMillis()));
 
-    token_ =
-        RestToken{ApplyDlfEndpointOverride(response.GetToken()), response.GetExpiresAtMillis()};
+    token_ = RestToken{response.GetToken(), response.GetExpiresAtMillis()};
     return Status::OK();
 }
 

src/paimon/rest/rest_credential_provider.h

@@ -68,12 +68,9 @@ class RestCredentialProvider : public CredentialProvider {
 
     /// @param api Client of the catalog that issues the credentials. Shared because a
     ///            provider commonly outlives the catalog it was obtained from.
-    /// @param catalog_options Options the credentials are merged over.
     /// @param identifier The table the credentials are requested for.
     /// @param clock Source of the current time, overridable for tests.
-    RestCredentialProvider(const std::shared_ptr<RestApi>& api,
-                           const std::map<std::string, std::string>& catalog_options,
-                           const Identifier& identifier,
+    RestCredentialProvider(const std::shared_ptr<RestApi>& api, const Identifier& identifier,
                            Clock clock = std::chrono::system_clock::now);
 
     ~RestCredentialProvider() override = default;
@@ -89,6 +86,15 @@ class RestCredentialProvider : public CredentialProvider {
     /// source.
     Result<std::map<std::string, std::string>> GetCredentials() const override;
 
+    /// Merges the issued credentials over `base_options`, then corrects the OSS endpoint: the
+    /// credentials are issued for the catalog's DLF OSS endpoint, which overrides both the
+    /// endpoint the catalog was configured with and the one the server reported. The correction
+    /// lives here rather than in the token so the token stays a minimal cache key that carries
+    /// only the issued credentials.
+    std::map<std::string, std::string> MergeOptionsWithCredentials(
+        const std::map<std::string, std::string>& base_options,
+        const std::map<std::string, std::string>& credentials) const override;
+
  private:
     /// Reloads the credentials from the server. Called with the write lock of `mutex_`
     /// held.
@@ -97,17 +103,7 @@ class RestCredentialProvider : public CredentialProvider {
     /// Whether `token_` is absent or expires within the safe time.
     bool ShouldRefresh() const;
 
-    /// The issued `token` with the catalog's DLF OSS endpoint, when set, overriding the endpoint
-    /// the server reported: the credentials are issued for the DLF endpoint, not the one the
-    /// catalog was configured with. The token is otherwise left exactly as issued -- it is a file
-    /// system cache key and what `ValidToken()` serves, so it carries only the credentials, never
-    /// the whole catalog options; merging those over the token to build a delegate is the file
-    /// system's `MergeTokenOptions`.
-    std::map<std::string, std::string> ApplyDlfEndpointOverride(
-        const std::map<std::string, std::string>& token) const;
-
     std::shared_ptr<RestApi> api_;
-    std::map<std::string, std::string> catalog_options_;
     Identifier identifier_;
     Clock clock_;
     std::shared_ptr<Logger> logger_;

src/paimon/rest/rest_credential_provider_test.cpp

@@ -117,7 +117,7 @@ class RestCredentialProviderTest : public ::testing::Test {
         }
         std::shared_ptr<RestApi> shared_api(std::move(api).value());
         return std::make_shared<RestCredentialProvider>(
-            shared_api, catalog_options_, Identifier("db1", "t1"), [this] {
+            shared_api, Identifier("db1", "t1"), [this] {
                 return std::chrono::system_clock::time_point(
                     std::chrono::milliseconds(now_millis_.load()));
             });
@@ -194,7 +194,7 @@ TEST_F(RestCredentialProviderTest, ExpiredTokenReloadsOnEveryCall) {
     ASSERT_EQ(2, state_->request_count.load());
 }
 
-TEST_F(RestCredentialProviderTest, DlfEndpointOverridesTheServerEndpoint) {
+TEST_F(RestCredentialProviderTest, DlfEndpointOverridesTheServerEndpointOnMerge) {
     catalog_options_[CatalogOptions::DLF_OSS_ENDPOINT] = "dlf-endpoint";
     {
         std::lock_guard<std::mutex> lock(state_->mutex);
@@ -203,14 +203,20 @@ TEST_F(RestCredentialProviderTest, DlfEndpointOverridesTheServerEndpoint) {
     std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
     ASSERT_NE(nullptr, provider);
 
+    // the token is the cache key and stays exactly as issued, carrying no catalog secrets and
+    // not the endpoint correction, which is a merge concern
     ASSERT_OK_AND_ASSIGN(RestToken token, provider->ValidToken());
     ASSERT_EQ("ak-1", token.token.at("fs.oss.accessKeyId"));
-    // the endpoint the credentials were issued for wins over the one the server reported
-    ASSERT_EQ("dlf-endpoint", token.token.at(kOssEndpointOption));
+    ASSERT_EQ("server-endpoint", token.token.at(kOssEndpointOption));
     ASSERT_EQ(kExpiresAtMillis, token.expires_at_millis);
-    // the catalog options are not part of the token, so its secrets stay private
     ASSERT_EQ(0u, token.token.count(CatalogOptions::TOKEN));
     ASSERT_EQ(2u, token.token.size());
+
+    // merging shapes the credentials into the file system options: the endpoint the credentials
+    // were issued for wins over the one the server reported
+    Credentials merged = provider->MergeOptionsWithCredentials(catalog_options_, token.token);
+    ASSERT_EQ("ak-1", merged.at("fs.oss.accessKeyId"));
+    ASSERT_EQ("dlf-endpoint", merged.at(kOssEndpointOption));
 }
 
 TEST_F(RestCredentialProviderTest, EmptyDlfOssEndpointIsNotApplied) {
@@ -222,10 +228,92 @@ TEST_F(RestCredentialProviderTest, EmptyDlfOssEndpointIsNotApplied) {
     std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
     ASSERT_NE(nullptr, provider);
 
-    // an unset dlf endpoint leaves the endpoint the server reported alone
+    // an unset dlf endpoint leaves the endpoint the credentials carry alone through the merge
     ASSERT_OK_AND_ASSIGN(RestToken token, provider->ValidToken());
-    ASSERT_EQ("server-endpoint", token.token.at(kOssEndpointOption));
-    ASSERT_EQ(1u, token.token.size());
+    Credentials merged = provider->MergeOptionsWithCredentials(catalog_options_, token.token);
+    ASSERT_EQ("server-endpoint", merged.at(kOssEndpointOption));
+}
+
+TEST_F(RestCredentialProviderTest, IssuedCredentialsClearStaleBucketScopedCatalogVariants) {
+    // The OSS backend resolves a bucket-scoped option ahead of the flat one, so a stale
+    // bucket-scoped catalog value would shadow the flat credential the token just refreshed.
+    std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
+    ASSERT_NE(nullptr, provider);
+
+    Credentials base = {
+        {"fs.oss.accessKeyId", "catalog-ak"},
+        {"fs.oss.bucket.b.accessKeyId", "catalog-bucket-ak"},
+        {"unrelated", "kept"},
+    };
+    Credentials credentials = {{"fs.oss.accessKeyId", "token-ak"}};
+
+    Credentials merged = provider->MergeOptionsWithCredentials(base, credentials);
+    ASSERT_EQ("token-ak", merged.at("fs.oss.accessKeyId"));
+    ASSERT_EQ(0u, merged.count("fs.oss.bucket.b.accessKeyId"));
+    ASSERT_EQ("kept", merged.at("unrelated"));
+}
+
+TEST_F(RestCredentialProviderTest, SessionTokenCredentialClearsStaleSecurityTokenAliases) {
+    // A token carrying a fresh session token must clear a stale catalog security token -- the
+    // OSS backend reads securityToken first, so leaving it in place would pair the refreshed
+    // key pair with the old STS token and fail authentication. Both the flat and the
+    // bucket-scoped stale securityToken have to go.
+    std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
+    ASSERT_NE(nullptr, provider);
+
+    Credentials base = {
+        {"fs.oss.securityToken", "stale-sts"},
+        {"fs.oss.bucket.b.securityToken", "stale-bucket-sts"},
+    };
+    Credentials credentials = {{"fs.oss.accessKeyId", "token-ak"},
+                               {"fs.oss.accessKeySecret", "token-sk"},
+                               {"fs.oss.sessionToken", "fresh-sts"}};
+
+    Credentials merged = provider->MergeOptionsWithCredentials(base, credentials);
+    ASSERT_EQ("token-ak", merged.at("fs.oss.accessKeyId"));
+    ASSERT_EQ("token-sk", merged.at("fs.oss.accessKeySecret"));
+    ASSERT_EQ("fresh-sts", merged.at("fs.oss.sessionToken"));
+    ASSERT_EQ(0u, merged.count("fs.oss.securityToken"));
+    ASSERT_EQ(0u, merged.count("fs.oss.bucket.b.securityToken"));
+}
+
+TEST_F(RestCredentialProviderTest, SecurityTokenCredentialClearsStaleSessionTokenAlias) {
+    // The alias goes the other way too: a fresh securityToken clears a stale sessionToken so the
+    // backend cannot fall back to it.
+    std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
+    ASSERT_NE(nullptr, provider);
+
+    Credentials base = {{"fs.oss.sessionToken", "stale-sts"}};
+    Credentials credentials = {{"fs.oss.securityToken", "fresh-sts"}};
+
+    Credentials merged = provider->MergeOptionsWithCredentials(base, credentials);
+    ASSERT_EQ("fresh-sts", merged.at("fs.oss.securityToken"));
+    ASSERT_EQ(0u, merged.count("fs.oss.sessionToken"));
+}
+
+TEST_F(RestCredentialProviderTest, TokenSuppliedBucketScopedCredentialsArePreserved) {
+    // When the token itself carries a complete bucket-scoped credential set, those values are
+    // authoritative: the cleanup that a global credential of the same suffix would otherwise
+    // trigger must not erase the token's own bucket-scoped keys.
+    std::shared_ptr<RestCredentialProvider> provider = CreateProvider();
+    ASSERT_NE(nullptr, provider);
+
+    Credentials base = {};
+    Credentials credentials = {
+        {"fs.oss.accessKeyId", "token-ak"},
+        {"fs.oss.accessKeySecret", "token-sk"},
+        {"fs.oss.securityToken", "token-sts"},
+        {"fs.oss.bucket.b.accessKeyId", "token-bucket-ak"},
+        {"fs.oss.bucket.b.accessKeySecret", "token-bucket-sk"},
+        {"fs.oss.bucket.b.securityToken", "token-bucket-sts"},
+    };
+
+    Credentials merged = provider->MergeOptionsWithCredentials(base, credentials);
+    ASSERT_EQ("token-ak", merged.at("fs.oss.accessKeyId"));
+    ASSERT_EQ("token-sts", merged.at("fs.oss.securityToken"));
+    ASSERT_EQ("token-bucket-ak", merged.at("fs.oss.bucket.b.accessKeyId"));
+    ASSERT_EQ("token-bucket-sk", merged.at("fs.oss.bucket.b.accessKeySecret"));
+    ASSERT_EQ("token-bucket-sts", merged.at("fs.oss.bucket.b.securityToken"));
 }
 
 TEST_F(RestCredentialProviderTest, ForbiddenIsReportedToTheCaller) {