feat(rest): add opt-in client-only-properties to protect oauth2-serve…

src/iceberg/catalog/rest/auth/auth_properties.cc

@@ -36,6 +36,8 @@ std::pair<std::string, std::string> ParseCredential(const std::string& credentia
   return {credential.substr(0, colon_pos), credential.substr(colon_pos + 1)};
 }
 
+}  // namespace
+
 Result<std::string> ResolveOAuth2ServerUri(
     const std::unordered_map<std::string, std::string>& properties) {
   auto endpoint_it = properties.find(AuthProperties::kOAuth2ServerUri.key());
@@ -61,8 +63,6 @@ Result<std::string> ResolveOAuth2ServerUri(
   return base_uri + "/" + endpoint;
 }
 
-}  // namespace
-
 std::unordered_map<std::string, std::string> AuthProperties::optional_oauth_params()
     const {
   std::unordered_map<std::string, std::string> params;

src/iceberg/catalog/rest/auth/auth_properties.h

@@ -128,4 +128,20 @@ class ICEBERG_REST_EXPORT AuthProperties : public ConfigBase<AuthProperties> {
   std::optional<int64_t> expires_at_millis_;
 };
 
+/// \brief Resolve the OAuth2 server URI from a properties map.
+///
+/// This function resolves the OAuth2 server URI by examining the
+/// `oauth2-server-uri` key in the provided properties map. The resolution logic:
+/// - If the value starts with `http://` or `https://`, it is returned as-is
+///   (absolute URL).
+/// - If the value is empty, it is returned as-is.
+/// - Otherwise, the value is treated as a relative path and resolved against the
+///   `uri` key from the properties map. If the relative path starts with `/`, it
+///   is appended directly; otherwise, a `/` separator is inserted.
+///
+/// \param properties Properties map containing `oauth2-server-uri` and `uri` keys.
+/// \return Resolved OAuth2 server URI.
+Result<std::string> ResolveOAuth2ServerUri(
+    const std::unordered_map<std::string, std::string>& properties);
+
 }  // namespace iceberg::rest::auth

src/iceberg/catalog/rest/catalog_properties.h

@@ -77,6 +77,10 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
       "rest.client.connection-timeout-ms", 0};
   /// \brief HTTP socket/read timeout in milliseconds, matching Java REST clients.
   inline static Entry<int64_t> kClientSocketTimeoutMs{"rest.client.socket-timeout-ms", 0};
+  /// \brief Comma-separated keys that are resolved from client configuration
+  /// only; server-provided defaults and overrides for them are ignored.
+  inline static Entry<std::string> kClientOnlyProperties{"rest.client-only-properties",
+                                                         ""};
 
   /// \brief Create a default RestCatalogProperties instance.
   static RestCatalogProperties default_properties();

src/iceberg/catalog/rest/rest_catalog.cc

@@ -35,6 +35,7 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/catalog/rest/auth/auth_managers.h"
+#include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/constant.h"
@@ -481,8 +482,10 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   ICEBERG_ASSIGN_OR_RAISE(auto server_config,
                           FetchServerConfig(*init_client, *paths, config, *init_session));
 
-  RestCatalogProperties final_config = RestCatalogProperties::FromMap(
-      MergeConfigs(server_config.defaults, config.configs(), server_config.overrides));
+  ICEBERG_ASSIGN_OR_RAISE(auto merged_config,
+                          MergeServerConfig(server_config.defaults,
+                                            server_config.overrides, config.configs()));
+  RestCatalogProperties final_config = RestCatalogProperties::FromMap(merged_config);
 
   std::unordered_set<Endpoint> endpoints;
   if (!server_config.endpoints.empty()) {
@@ -511,6 +514,30 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 
   ICEBERG_ASSIGN_OR_RAISE(auto client, client_factory(final_config));
   ICEBERG_PRECHECK(client != nullptr, "HTTP client factory returned null");
+
+  // Warn if oauth2-server-uri is neither explicitly configured nor protected from
+  // server overrides while using credential auth.
+  if (auto cred_it = final_config.configs().find(auth::AuthProperties::kCredential.key());
+      cred_it != final_config.configs().end() && !cred_it->second.empty()) {
+    auto client_uri_it =
+        config.configs().find(auth::AuthProperties::kOAuth2ServerUri.key());
+    bool endpoint_explicit =
+        client_uri_it != config.configs().end() && !client_uri_it->second.empty() &&
+        client_uri_it->second != auth::AuthProperties::kOAuth2ServerUri.value();
+    bool endpoint_protected =
+        ParseClientOnlyProperties(
+            config.Get(RestCatalogProperties::kClientOnlyProperties))
+            .contains(auth::AuthProperties::kOAuth2ServerUri.key());
+    if (!endpoint_explicit && !endpoint_protected) {
+      ICEBERG_LOG_WARN(
+          "OAuth2 server URI not explicitly configured; token endpoint will be derived "
+          "from catalog URI and may be affected by server overrides. Set '{}' explicitly "
+          "or add it to '{}' for stronger isolation.",
+          auth::AuthProperties::kOAuth2ServerUri.key(),
+          RestCatalogProperties::kClientOnlyProperties.key());
+    }
+  }
+
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(client, final_config.configs()));
 

src/iceberg/catalog/rest/rest_util.cc

@@ -20,11 +20,15 @@
 #include "iceberg/catalog/rest/rest_util.h"
 
 #include <format>
+#include <ranges>
 #include <unordered_set>
 
 #include <cpr/util.h>
 
+#include "iceberg/catalog/rest/auth/auth_properties.h"
+#include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/endpoint.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/util/macros.h"
 
@@ -135,6 +139,69 @@ std::unordered_map<std::string, std::string> MergeConfigs(
   return merged;
 }
 
+std::unordered_set<std::string> ParseClientOnlyProperties(std::string_view value) {
+  constexpr std::string_view kWhitespace = " \t\n\r\f\v";
+
+  std::unordered_set<std::string> result;
+  for (auto part : std::views::split(value, ',')) {
+    std::string_view key(part.begin(), part.end());
+    if (auto first = key.find_first_not_of(kWhitespace);
+        first != std::string_view::npos) {
+      key = key.substr(first, key.find_last_not_of(kWhitespace) - first + 1);
+      result.emplace(key);
+    }
+  }
+  return result;
+}
+
+Result<std::unordered_map<std::string, std::string>> MergeServerConfig(
+    const std::unordered_map<std::string, std::string>& server_defaults,
+    const std::unordered_map<std::string, std::string>& server_overrides,
+    const std::unordered_map<std::string, std::string>& client_configs) {
+  // Make mutable copies of server configs
+  auto defaults = server_defaults;
+  auto overrides = server_overrides;
+
+  // Parse client-only properties list from client config only
+  std::string_view client_only_value;
+  if (auto it = client_configs.find(RestCatalogProperties::kClientOnlyProperties.key());
+      it != client_configs.end()) {
+    client_only_value = it->second;
+  }
+  const auto client_only = ParseClientOnlyProperties(client_only_value);
+
+  // Remove rest.client-only-properties itself from server configs
+  defaults.erase(RestCatalogProperties::kClientOnlyProperties.key());
+  overrides.erase(RestCatalogProperties::kClientOnlyProperties.key());
+
+  // Remove client-only keys from server configs
+  for (const auto& key : client_only) {
+    if (auto it = defaults.find(key); it != defaults.end()) {
+      ICEBERG_LOG_INFO("Ignoring server-provided '{}' in defaults: listed in {}", key,
+                       RestCatalogProperties::kClientOnlyProperties.key());
+      defaults.erase(it);
+    }
+    if (auto it = overrides.find(key); it != overrides.end()) {
+      ICEBERG_LOG_INFO("Ignoring server-provided '{}' in overrides: listed in {}", key,
+                       RestCatalogProperties::kClientOnlyProperties.key());
+      overrides.erase(it);
+    }
+  }
+
+  // If oauth2-server-uri is protected, resolve it to an absolute URL before merging
+  auto merged_client_configs = client_configs;
+  if (client_only.contains(auth::AuthProperties::kOAuth2ServerUri.key())) {
+    ICEBERG_ASSIGN_OR_RAISE(auto resolved_uri,
+                            auth::ResolveOAuth2ServerUri(client_configs));
+    if (!resolved_uri.empty()) {
+      merged_client_configs[auth::AuthProperties::kOAuth2ServerUri.key()] =
+          std::move(resolved_uri);
+    }
+  }
+
+  return MergeConfigs(defaults, merged_client_configs, overrides);
+}
+
 std::string GetStandardReasonPhrase(int32_t status_code) {
   switch (status_code) {
     case 100:

src/iceberg/catalog/rest/rest_util.h

@@ -92,6 +92,32 @@ ICEBERG_REST_EXPORT std::unordered_map<std::string, std::string> MergeConfigs(
     const std::unordered_map<std::string, std::string>& client_configs,
     const std::unordered_map<std::string, std::string>& server_overrides);
 
+/// \brief Parse the rest.client-only-properties value.
+///
+/// \details Parses a comma-separated list of property keys. Each key is taken from
+/// client configuration only; server-provided values for them are ignored. Leading
+/// and trailing whitespace is stripped from each key, and empty entries are ignored.
+/// \param value The raw string value of rest.client-only-properties.
+/// \return A set of property keys to protect from server override.
+ICEBERG_REST_EXPORT std::unordered_set<std::string> ParseClientOnlyProperties(
+    std::string_view value);
+
+/// \brief Resolve the effective catalog configuration from server-provided
+/// defaults/overrides and client configuration.
+///
+/// \details Keys listed in rest.client-only-properties are taken from client
+/// configuration only: server-provided values for them are ignored. If
+/// oauth2-server-uri is listed, it is resolved to an absolute URL before merging
+/// to prevent indirect override via the uri property.
+/// \param server_defaults Default properties provided by the server.
+/// \param server_overrides Override properties enforced by the server.
+/// \param client_configs Configuration properties from the client.
+/// \return The merged configuration map with client-only properties protected.
+ICEBERG_REST_EXPORT Result<std::unordered_map<std::string, std::string>>
+MergeServerConfig(const std::unordered_map<std::string, std::string>& server_defaults,
+                  const std::unordered_map<std::string, std::string>& server_overrides,
+                  const std::unordered_map<std::string, std::string>& client_configs);
+
 /// \brief Get the standard HTTP reason phrase for a status code.
 ///
 /// \details Returns the standard English reason phrase for common HTTP status codes.

src/iceberg/test/rest_catalog_unit_test.cc

@@ -29,6 +29,7 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/catalog.h"
+#include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/error_handlers.h"
@@ -76,6 +77,13 @@ class MockHttpClient : public HttpClient {
                const ErrorHandler& error_handler, auth::AuthSession& session),
               (override));
 
+  MOCK_METHOD(Result<HttpResponse>, PostForm,
+              (const std::string& path,
+               (const std::unordered_map<std::string, std::string>&)form_data,
+               (const std::unordered_map<std::string, std::string>&)headers,
+               const ErrorHandler& error_handler, auth::AuthSession& session),
+              (override));
+
   MOCK_METHOD(Result<HttpResponse>, Delete,
               (const std::string& path,
                (const std::unordered_map<std::string, std::string>&)params,
@@ -559,6 +567,107 @@ TEST_P(RestCatalogScanLifetimeTest, ScanKeepsCredentialRefreshAlive) {
 
 INSTANTIATE_TEST_SUITE_P(ScanModes, RestCatalogScanLifetimeTest, ::testing::Bool());
 
+// --------------------------------------------------------------------------
+// Client-only properties tests
+// --------------------------------------------------------------------------
+
+class RestCatalogClientOnlyPropertiesTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    mock_client_ = std::make_shared<MockHttpClient>();
+    FileIORegistry::Register(
+        "rest-catalog-unit-test-noop",
+        {.create = [](const std::unordered_map<std::string, std::string>&)
+             -> Result<std::unique_ptr<FileIO>> {
+          return std::make_unique<NoOpFileIO>();
+        }});
+  }
+
+  // Runs RestCatalog::Make against a server that returns `server_config` from
+  // /v1/config, and returns the configurations handed to the HTTP client
+  // factory. The second entry is the merged configuration.
+  std::vector<RestCatalogProperties> MakeWithServerConfig(
+      const CatalogConfig& server_config,
+      std::unordered_map<std::string, std::string> client_properties) {
+    client_properties["io-impl"] = "rest-catalog-unit-test-noop";
+    EXPECT_CALL(*mock_client_, Get(::testing::EndsWith("/v1/config"), _, _, _, _))
+        .WillOnce(Return(HttpResponse(200, ToJson(server_config).dump())));
+    EXPECT_CALL(*mock_client_, PostForm(_, _, _, _, _))
+        .WillRepeatedly(
+            [](const std::string&, const std::unordered_map<std::string, std::string>&,
+               const std::unordered_map<std::string, std::string>&, const ErrorHandler&,
+               auth::AuthSession&) -> Result<HttpResponse> {
+              return HttpResponse(200, R"({"access_token":"tok","token_type":"bearer"})");
+            });
+
+    auto config = RestCatalogProperties::FromMap(std::move(client_properties));
+    std::vector<RestCatalogProperties> captured;
+    auto result =
+        RestCatalog::Make(config, /*metrics_executor=*/nullptr,
+                          [&captured, this](const RestCatalogProperties& properties) {
+                            captured.push_back(properties);
+                            return mock_client_;
+                          });
+    EXPECT_THAT(result, IsOk());
+    return captured;
+  }
+
+  std::shared_ptr<MockHttpClient> mock_client_;
+};
+
+TEST_F(RestCatalogClientOnlyPropertiesTest,
+       ServerOverrideAppliesWhenNoClientOnlyPropertiesAreSet) {
+  auto configs = MakeWithServerConfig(
+      CatalogConfig{.defaults = {},
+                    .overrides = {{"oauth2-server-uri", "https://server.example/token"}},
+                    .endpoints = {}},
+      {{"uri", "https://catalog.example"}, {"credential", "client:secret"}});
+
+  ASSERT_EQ(configs.size(), 2);
+  // Upstream REST behavior is unchanged when the list is empty.
+  EXPECT_EQ(
+      configs[1].configs().at(std::string(auth::AuthProperties::kOAuth2ServerUri.key())),
+      "https://server.example/token");
+}
+
+TEST_F(RestCatalogClientOnlyPropertiesTest,
+       ClientOnlyPropertiesProtectsExplicitOAuth2ServerUri) {
+  auto configs = MakeWithServerConfig(
+      CatalogConfig{.defaults = {},
+                    .overrides = {{"oauth2-server-uri", "https://server.example/token"}},
+                    .endpoints = {}},
+      {{"uri", "https://catalog.example"},
+       {"credential", "client:secret"},
+       {"oauth2-server-uri", "https://idp.example/token"},
+       {"rest.client-only-properties", "oauth2-server-uri"}});
+
+  ASSERT_EQ(configs.size(), 2);
+  EXPECT_EQ(
+      configs[1].configs().at(std::string(auth::AuthProperties::kOAuth2ServerUri.key())),
+      "https://idp.example/token");
+}
+
+TEST_F(RestCatalogClientOnlyPropertiesTest,
+       ClientOnlyPropertiesProtectsDefaultOAuth2ServerUriFromIndirectOverride) {
+  auto configs = MakeWithServerConfig(
+      CatalogConfig{.defaults = {},
+                    .overrides = {{"uri", "https://collector.example"},
+                                  {"oauth2-server-uri", "https://server.example/token"}},
+                    .endpoints = {}},
+      {{"uri", "https://catalog.example"},
+       {"credential", "client:secret"},
+       {"rest.client-only-properties", "oauth2-server-uri"}});
+
+  ASSERT_EQ(configs.size(), 2);
+  // The endpoint is pinned against the client URI before merging, so a server
+  // override of `uri` cannot move it.
+  EXPECT_EQ(
+      configs[1].configs().at(std::string(auth::AuthProperties::kOAuth2ServerUri.key())),
+      "https://catalog.example/v1/oauth/tokens");
+  // `uri` itself is still overridable.
+  EXPECT_EQ(configs[1].Get(RestCatalogProperties::kUri), "https://collector.example");
+}
+
 }  // namespace
 
 }  // namespace iceberg::rest