diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index 3ce753f18..098f2ba08 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -152,6 +152,25 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
   return credential;
 }
 
+/// \brief Reads the optional `storage-credentials` array shared by the
+/// LoadTable and LoadCredentials responses.
+Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
+    const nlohmann::json& json) {
+  std::vector<StorageCredential> credentials;
+  auto it = json.find(kStorageCredentials);
+  if (it == json.end() || it->is_null()) {
+    return credentials;
+  }
+  if (!it->is_array()) {
+    return JsonParseError("Cannot parse storage credentials from non-array");
+  }
+  for (const auto& entry : *it) {
+    ICEBERG_ASSIGN_OR_RAISE(auto credential, StorageCredentialFromJson(entry));
+    credentials.push_back(std::move(credential));
+  }
+  return credentials;
+}
+
 template <typename Value>
 Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
                                                      std::string_view key) {
@@ -738,19 +757,24 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
-  if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
-    if (!it->is_array()) {
-      return JsonParseError("Cannot parse storage credentials from non-array");
-    }
-    for (const auto& entry : *it) {
-      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
-      result.storage_credentials.push_back(std::move(cred));
-    }
-  }
+  ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials, StorageCredentialsFromJson(json));
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
 
+Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json) {
+  // Required here, unlike in LoadTable: reading a malformed response as "no
+  // credentials" would look like a refresh that succeeded and dropped them.
+  if (auto it = json.find(kStorageCredentials); it == json.end() || it->is_null()) {
+    return JsonParseError("Missing '{}'", kStorageCredentials);
+  }
+  LoadCredentialsResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
+  return response;
+}
+
 nlohmann::json ToJson(const ListNamespacesResponse& response) {
   nlohmann::json json;
   SetOptionalStringField(json, kNextPageToken, response.next_page_token);
diff --git a/src/iceberg/catalog/rest/json_serde_internal.h b/src/iceberg/catalog/rest/json_serde_internal.h
index 6e148e0d3..aedd2cd1a 100644
--- a/src/iceberg/catalog/rest/json_serde_internal.h
+++ b/src/iceberg/catalog/rest/json_serde_internal.h
@@ -71,6 +71,10 @@ template <>
 ICEBERG_REST_EXPORT Result<LoadTableResult> FromJson(const nlohmann::json& json);
 ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const LoadTableResult& model);
 
+// Response-only model: a client never serializes it, so no ToJson.
+ICEBERG_REST_EXPORT Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json);
+
 ICEBERG_REST_EXPORT Result<CreateTableRequest> CreateTableRequestFromJson(
     const nlohmann::json& json);
 template <>
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 9635aebef..e2aaf5f3e 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -20,6 +20,8 @@
 #include "iceberg/catalog/rest/rest_catalog.h"
 
 #include <memory>
+#include <mutex>
+#include <string>
 #include <string_view>
 #include <tuple>
 #include <unordered_map>
@@ -42,6 +44,7 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -60,6 +63,41 @@ namespace iceberg::rest {
 
 namespace {
 
+class RestStorageCredentialProvider final : public StorageCredentialProvider {
+ public:
+  RestStorageCredentialProvider(std::shared_ptr<HttpClient> client,
+                                std::shared_ptr<auth::AuthSession> session,
+                                std::string path)
+      : client_(std::move(client)),
+        session_(std::move(session)),
+        path_(std::move(path)) {}
+
+  Result<std::vector<StorageCredential>> Load() override {
+    // The provider can be shared by several FileIOs; keep fetches serialized.
+    std::lock_guard lock(mutex_);
+    ICEBERG_ASSIGN_OR_RAISE(const auto response,
+                            client_->Get(path_, /*params=*/{}, /*headers=*/{},
+                                         *TableErrorHandler::Instance(), *session_));
+    // Parse errors can contain credential data; return a fixed message.
+    auto json = FromJsonString(response.body());
+    if (!json.has_value()) {
+      return JsonParseError("Malformed LoadCredentials response");
+    }
+    auto result = LoadCredentialsResponseFromJson(*json);
+    if (!result.has_value()) {
+      return std::unexpected<Error>(
+          {.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
+    }
+    return std::move(result->storage_credentials);
+  }
+
+ private:
+  std::shared_ptr<HttpClient> client_;
+  std::shared_ptr<auth::AuthSession> session_;
+  std::string path_;
+  std::mutex mutex_;
+};
+
 /// \brief Get the default set of endpoints for backwards compatibility according to the
 /// iceberg rest spec.
 std::unordered_set<Endpoint> GetDefaultEndpoints() {
@@ -508,12 +546,45 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
                                      std::move(contextual_session));
 }
 
+std::shared_ptr<StorageCredentialProvider> RestCatalog::MakeStorageCredentialProvider(
+    const TableIdentifier& identifier,
+    std::shared_ptr<auth::AuthSession> table_session) const {
+  if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
+    // Not an error, but it surfaces much later as credentials expiring.
+    ICEBERG_LOG_DEBUG(
+        "Catalog does not advertise {}; vended credentials for '{}' will not be "
+        "refreshed",
+        Endpoint::TableCredentials().ToString(), ToString(identifier));
+    return nullptr;
+  }
+  auto path = paths_->Credentials(identifier);
+  if (!path.has_value()) {
+    ICEBERG_LOG_WARN(
+        "Cannot build the credentials path for '{}' ({}); its vended credentials "
+        "will not be refreshed",
+        ToString(identifier), path.error().message);
+    return nullptr;
+  }
+  auto client = client_;
+  auto credentials_path = std::move(path.value());
+  auto session = std::move(table_session);
+  return std::make_shared<RestStorageCredentialProvider>(
+      std::move(client), std::move(session), std::move(credentials_path));
+}
+
 Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
-    const SessionContext& /*context*/,
+    const SessionContext& /*context*/, const TableIdentifier& identifier,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) const {
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<auth::AuthSession> table_session) const {
   if (!table_config.empty() || !storage_credentials.empty()) {
-    return MakeTableFileIO(config_.configs(), table_config, storage_credentials);
+    // Only vended credentials expire, so only they need a provider.
+    std::shared_ptr<StorageCredentialProvider> provider;
+    if (!storage_credentials.empty()) {
+      provider = MakeStorageCredentialProvider(identifier, std::move(table_session));
+    }
+    return MakeTableFileIO(config_.configs(), table_config, storage_credentials,
+                           std::move(provider));
   }
 
   return file_io_;
@@ -772,11 +843,12 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
                           /*stage_create=*/true, *contextual_session));
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, std::move(table_session),
@@ -890,11 +962,12 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
     std::shared_ptr<auth::AuthSession> contextual_session) {
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session, table_io);
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
index 65b0b5eab..6f65e6d61 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -85,9 +85,17 @@ class ICEBERG_REST_EXPORT RestCatalog final
       std::shared_ptr<auth::AuthSession> contextual_session);
 
   Result<std::shared_ptr<FileIO>> TableFileIO(
-      const SessionContext& context,
+      const SessionContext& context, const TableIdentifier& identifier,
       const std::unordered_map<std::string, std::string>& table_config,
-      const std::vector<StorageCredential>& storage_credentials) const;
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<auth::AuthSession> table_session) const;
+
+  /// \brief Build a provider for this table's vended credentials.
+  ///
+  /// Returns nullptr when the catalog does not serve LoadCredentials.
+  std::shared_ptr<StorageCredentialProvider> MakeStorageCredentialProvider(
+      const TableIdentifier& identifier,
+      std::shared_ptr<auth::AuthSession> table_session) const;
 
   Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
                                                 auth::AuthSession& session) const;
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc
index cdc1204f1..4cd102704 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -21,11 +21,13 @@
 
 #include <string>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/resolving_file_io.h"
 #include "iceberg/util/macros.h"
 
@@ -57,7 +59,8 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) {
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<StorageCredentialProvider> provider) {
   const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
   ICEBERG_ASSIGN_OR_RAISE(
       auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
@@ -65,7 +68,16 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
   if (storage_credentials.empty()) {
     return io;
   } else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
-    ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
+    // First, so the FileIO never briefly holds credentials it cannot replace.
+    auto status = credentialed->InitializeStorageCredentials(storage_credentials,
+                                                             std::move(provider));
+    if (!status && !storage_credentials.empty() &&
+        status.error().kind == ErrorKind::kNotSupported) {
+      ICEBERG_LOG_WARN("Configured FileIO cannot refresh vended storage credentials: {}",
+                       status.error().message);
+      status = credentialed->SetStorageCredentials(storage_credentials);
+    }
+    ICEBERG_RETURN_UNEXPECTED(status);
   } else {
     return NotSupported("Configured FileIO does not support vended storage credentials");
   }
diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h
index e2316c3e8..5eb9f97fa 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -44,6 +44,7 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials);
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<StorageCredentialProvider> provider = nullptr);
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 20a59fa59..a47a8e851 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -209,6 +209,21 @@ using CreateTableResponse = LoadTableResult;
 /// \brief Alias of LoadTableResult used as the body of LoadTableResponse
 using LoadTableResponse = LoadTableResult;
 
+/// \brief Response body of the LoadCredentials API.
+struct ICEBERG_REST_EXPORT LoadCredentialsResponse {
+  std::vector<StorageCredential> storage_credentials;
+
+  /// \brief Validates the LoadCredentialsResponse.
+  Status Validate() const {
+    for (const auto& credential : storage_credentials) {
+      ICEBERG_RETURN_UNEXPECTED(credential.Validate());
+    }
+    return {};
+  }
+
+  bool operator==(const LoadCredentialsResponse& other) const = default;
+};
+
 /// \brief Response body for listing namespaces.
 struct ICEBERG_REST_EXPORT ListNamespacesResponse {
   PageToken next_page_token;
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index e22e5cf21..4dd96530d 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -29,6 +29,7 @@
 #include <span>
 #include <string>
 #include <string_view>
+#include <utility>
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
@@ -198,6 +199,18 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
   /// By value because a concurrent install may replace them. An implementation
   /// that delegates may report what was installed on it.
   virtual std::vector<StorageCredential> credentials() const = 0;
+
+  /// \brief Install initial credentials and their refresh source before first use.
+  ///
+  /// The default implementation supports static credentials only.
+  virtual Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) {
+    if (provider) {
+      return NotSupported("Credential refresh is not supported");
+    }
+    return SetStorageCredentials(storage_credentials);
+  }
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 91be67eaf..ddf19b6fe 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -24,6 +24,7 @@
 #include <utility>
 
 #include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
@@ -44,13 +45,19 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   // without static keys may wait on the EC2 metadata service), which would
   // stall every other operation. Forwards all credentials; each implementation
   // applies the prefixes it understands.
-  auto load = [&](const std::vector<StorageCredential>& credentials)
+  auto load = [&](const std::vector<StorageCredential>& credentials,
+                  const std::shared_ptr<StorageCredentialProvider>& provider)
       -> Result<std::shared_ptr<FileIO>> {
     ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
                             FileIORegistry::Load(name, properties_));
-    if (!credentials.empty()) {
-      if (auto* credentialed = io->AsSupportsStorageCredentials()) {
-        ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
+    if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+      auto status = credentialed->InitializeStorageCredentials(credentials, provider);
+      if (!status && provider && status.error().kind == ErrorKind::kNotSupported) {
+        ICEBERG_LOG_WARN("FileIO '{}' cannot refresh vended storage credentials", name);
+        status = credentialed->SetStorageCredentials(credentials);
+      }
+      if (!status) {
+        return std::unexpected(status.error());
       }
     }
     return io;
@@ -59,6 +66,7 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   while (true) {
     uint64_t generation = 0;
     std::vector<StorageCredential> credentials;
+    std::shared_ptr<StorageCredentialProvider> provider;
     {
       std::shared_lock lock(mutex_);
       if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
@@ -66,13 +74,14 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
       }
       generation = credential_generation_;
       credentials = storage_credentials_;
+      provider = provider_;
     }
     // Declared before the lock, so a delegate that is not cached is torn down
     // only after the lock is released.
-    auto loaded = load(credentials);
+    auto loaded = load(credentials, provider);
     std::unique_lock lock(mutex_);
     if (generation != credential_generation_) {
-      continue;  // Credentials were replaced mid-load; load again with them.
+      continue;  // Replaced mid-load; load again with what is installed now.
     }
     if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
       return cached->second;
@@ -137,4 +146,17 @@ std::vector<StorageCredential> ResolvingFileIO::credentials() const {
   return storage_credentials_;
 }
 
+Status ResolvingFileIO::InitializeStorageCredentials(
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<StorageCredentialProvider> provider) {
+  std::unique_lock lock(mutex_);
+  if (!io_by_name_.empty()) {
+    return InvalidArgument("Storage credentials must be initialized before first use");
+  }
+  storage_credentials_ = storage_credentials;
+  provider_ = std::move(provider);
+  ++credential_generation_;
+  return {};
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index 9a5cf047e..05e11eb8f 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -40,8 +40,8 @@ namespace iceberg {
 
 /// \brief FileIO that resolves and caches implementations by registry name.
 ///
-/// Vended credentials are forwarded to every resolved implementation that
-/// supports them; each applies what it understands.
+/// Vended credentials and any credential provider are forwarded to every resolved
+/// implementation that supports them; each applies what it understands.
 class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
                                              public SupportsStorageCredentials {
  public:
@@ -62,8 +62,17 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& storage_credentials) override;
 
+  /// \brief Return the credentials installed on this resolver.
+  ///
+  /// Not necessarily the ones in use: a resolved implementation refreshes its
+  /// own without reporting back.
   std::vector<StorageCredential> credentials() const override;
 
+  /// \brief Install initial credentials before the first delegate is loaded.
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override;
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
  private:
@@ -74,6 +83,7 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   // Guards lazy resolution and credential state.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
+  std::shared_ptr<StorageCredentialProvider> provider_;
   // Bumped by every credential install, so a delegate loaded from an older set
   // never reaches the cache.
   uint64_t credential_generation_ = 0;
diff --git a/src/iceberg/storage_credential.h b/src/iceberg/storage_credential.h
index 746bad086..d21099615 100644
--- a/src/iceberg/storage_credential.h
+++ b/src/iceberg/storage_credential.h
@@ -24,6 +24,7 @@
 
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -48,4 +49,15 @@ struct ICEBERG_EXPORT StorageCredential {
   bool operator==(const StorageCredential& other) const = default;
 };
 
+/// \brief Source for loading the current vended credentials.
+///
+/// Implementations must be safe for concurrent calls.
+class ICEBERG_EXPORT StorageCredentialProvider {
+ public:
+  virtual ~StorageCredentialProvider() = default;
+
+  /// \brief Load the current credentials. Returns the whole vended list.
+  virtual Result<std::vector<StorageCredential>> Load() = 0;
+};
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc
index 2189b64a6..a3b194877 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -56,9 +56,28 @@ class RecordingFileIO : public FileIO {
   std::vector<std::vector<std::string>> deleted_batches;
 };
 
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+  explicit StaticStorageCredentialProvider(std::vector<StorageCredential> credentials)
+      : credentials_(std::move(credentials)) {}
+
+  Result<std::vector<StorageCredential>> Load() override { return credentials_; }
+
+ private:
+  std::vector<StorageCredential> credentials_;
+};
+
 class RecordingCredentialedFileIO : public RecordingFileIO,
                                     public SupportsStorageCredentials {
  public:
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override {
+    credentials_ = storage_credentials;
+    provider_ = std::move(provider);
+    return {};
+  }
+
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& storage_credentials) override {
     credentials_ = storage_credentials;
@@ -69,8 +88,19 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
 
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
+  Status Refresh() {
+    if (!provider_) {
+      return NotFound("no provider installed");
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto refreshed, provider_->Load());
+    return SetStorageCredentials(refreshed);
+  }
+
+  bool has_provider() const { return provider_ != nullptr; }
+
  private:
   std::vector<StorageCredential> credentials_;
+  std::shared_ptr<StorageCredentialProvider> provider_;
 };
 
 // File-scope recording state: registry factories are process-global, so they
@@ -318,4 +348,39 @@ TEST(ResolvingFileIOTest, LoadsWithoutTheLockAndDropsStaleDelegates) {
   EXPECT_EQ(last->credentials(), fresh);
 }
 
+TEST(ResolvingFileIOTest, ForwardsCredentialProviderToResolvedImplementations) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k2", "v2"}}}};
+  auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+  EXPECT_THAT(io.InitializeStorageCredentials(
+                  {{.prefix = "s3", .config = {{"k1", "v1"}}}}, provider),
+              IsOk());
+
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  ASSERT_TRUE(last_s3_io->has_provider());
+  EXPECT_THAT(last_s3_io->Refresh(), IsOk());
+  EXPECT_EQ(last_s3_io->credentials(), refreshed);
+}
+
+TEST(ResolvingFileIOTest, RejectsCredentialProviderAfterFirstUse) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+              IsOk());
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  EXPECT_FALSE(last_s3_io->has_provider());
+  EXPECT_EQ(s3_factory_calls, 1);
+
+  auto provider =
+      std::make_shared<StaticStorageCredentialProvider>(std::vector<StorageCredential>{});
+  EXPECT_THAT(io.InitializeStorageCredentials({}, provider),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(s3_factory_calls, 1);
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index 2fce9dacf..31a027ab4 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -50,11 +50,31 @@ class MockFileIO : public FileIO {
   Status DeleteFile(const std::string& /*file_location*/) override { return {}; }
 };
 
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+  explicit StaticStorageCredentialProvider(std::vector<StorageCredential> credentials)
+      : credentials_(std::move(credentials)) {}
+
+  Result<std::vector<StorageCredential>> Load() override { return credentials_; }
+
+ private:
+  std::vector<StorageCredential> credentials_;
+};
+
 std::vector<StorageCredential> captured_storage_credentials;
 std::unordered_map<std::string, std::string> captured_file_io_properties;
+std::shared_ptr<StorageCredentialProvider> captured_provider;
 
 class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredentials {
  public:
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override {
+    captured_storage_credentials = credentials;
+    captured_provider = std::move(provider);
+    return {};
+  }
+
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& credentials) override {
     captured_storage_credentials = credentials;
@@ -155,6 +175,34 @@ TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
   EXPECT_EQ(credentialed->credentials(), captured_storage_credentials);
 }
 
+TEST(RestFileIOTest, InstallsCredentialProviderOnlyForVendedCredentials) {
+  const std::string custom_impl = "rest-file-io-test-provider";
+  FileIORegistry::Register(
+      custom_impl,
+      {.create = [](const std::unordered_map<std::string, std::string>& /*properties*/)
+           -> Result<std::unique_ptr<FileIO>> {
+        return std::make_unique<MockCredentialedFileIO>();
+      }});
+  const std::unordered_map<std::string, std::string> table_config{
+      {"io-impl", custom_impl}};
+
+  captured_provider = nullptr;
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k", "v2"}}}};
+  auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+  auto result = MakeTableFileIO({}, table_config,
+                                {{.prefix = "s3", .config = {{"k", "v1"}}}}, provider);
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(captured_provider);
+  EXPECT_EQ(captured_provider, provider);
+  EXPECT_THAT(captured_provider->Load(), IsOk());
+  EXPECT_EQ(captured_provider->Load().value(), refreshed);
+
+  captured_provider = nullptr;
+  result = MakeTableFileIO({}, table_config, /*storage_credentials=*/{}, provider);
+  ASSERT_THAT(result, IsOk());
+  EXPECT_FALSE(captured_provider);
+}
+
 TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {
   captured_file_io_properties.clear();
   FileIORegistry::Register(
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index ec41e4a66..70b760b48 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -1229,6 +1229,32 @@ INSTANTIATE_TEST_SUITE_P(
       return info.param.test_name;
     });
 
+TEST(LoadCredentialsResponseTest, ParsesVendedCredentials) {
+  auto json = nlohmann::json::parse(
+      R"({"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest"}}]})");
+  auto response = LoadCredentialsResponseFromJson(json);
+  ASSERT_THAT(response, IsOk());
+  EXPECT_EQ(response->storage_credentials,
+            (std::vector<StorageCredential>{
+                {.prefix = "s3", .config = {{"s3.access-key-id", "AKIAtest"}}}}));
+}
+
+TEST(LoadCredentialsResponseTest, RejectsResponseWithoutCredentials) {
+  for (std::string_view body : {"{}", R"({"storage-credentials": null})"}) {
+    auto response = LoadCredentialsResponseFromJson(nlohmann::json::parse(body));
+    EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(response, HasErrorMessage("Missing 'storage-credentials'"));
+  }
+}
+
+TEST(LoadCredentialsResponseTest, RejectsNonArrayCredentials) {
+  auto response = LoadCredentialsResponseFromJson(
+      nlohmann::json::parse(R"({"storage-credentials":"oops"})"));
+  EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(response,
+              HasErrorMessage("Cannot parse storage credentials from non-array"));
+}
+
 DECLARE_ROUNDTRIP_TEST(CommitTableRequest)
 
 INSTANTIATE_TEST_SUITE_P(