fix(rest): improve REST catalog configuration and error handling

src/iceberg/CMakeLists.txt

@@ -24,6 +24,7 @@ set(ICEBERG_SOURCES
     arrow_row_builder.cc
     catalog/catalog_util.cc
     catalog/memory/in_memory_catalog.cc
+    catalog/rest/auth/auth_pinned_keys.cc
     catalog/session_catalog.cc
     catalog/session_context.cc
     data_table_scan.cc
@@ -151,6 +152,7 @@ set(ICEBERG_SOURCES
     util/decimal.cc
     util/gzip_internal.cc
     util/location_util.cc
+    util/log_sanitizer.cc
     util/murmurhash3_internal.cc
     util/property_util.cc
     util/retry_util.cc

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

@@ -0,0 +1,86 @@
+/*
+ * 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 "iceberg/catalog/rest/auth/auth_pinned_keys.h"
+
+#include "iceberg/logging/log_macros.h"
+
+namespace iceberg {
+
+const std::unordered_set<std::string>& AuthPinnedKeys::GetPinnedKeys() {
+  // Static set of auth-pinned keys that must be client-controlled.
+  // This is an exact, exhaustive list with no pattern matching.
+  //
+  // These keys control authentication routing and credentials.
+  // Allowing a server to override them would enable attacks like:
+  // - Redirecting OAuth2 credentials to an attacker-controlled endpoint
+  // - Replacing client credentials with attacker-controlled values
+  //
+  // Keys NOT in this list (safe for server override):
+  // - token-refresh-enabled, token-exchange-enabled (boolean flags)
+  // - scope, audience, resource (OAuth2 parameters, not endpoints)
+  // - rest.access-key-id, s3.access-key-id (public identifiers)
+  static const std::unordered_set<std::string> kPinnedKeys = {
+      // OAuth2 endpoint routing - redirecting this enables credential theft
+      "oauth2-server-uri",
+
+      // OAuth2 credentials - these ARE the secrets
+      "token",       // Bearer token
+      "credential",  // OAuth2 client_id:client_secret
+
+      // Basic Authentication credentials
+      "rest.auth.basic.username",
+      "rest.auth.basic.password",
+
+      // AWS SigV4 secrets (note: rest.access-key-id is NOT pinned - it's public)
+      "rest.secret-access-key",
+      "rest.session-token",
+  };
+  return kPinnedKeys;
+}
+
+bool AuthPinnedKeys::IsPinned(std::string_view key) {
+  const auto& pinned = GetPinnedKeys();
+  return pinned.contains(std::string(key));
+}
+
+std::unordered_map<std::string, std::string> AuthPinnedKeys::FilterServerConfig(
+    const std::unordered_map<std::string, std::string>& server_config,
+    bool log_filtered) {
+  std::unordered_map<std::string, std::string> filtered;
+  filtered.reserve(server_config.size());
+
+  for (const auto& [key, value] : server_config) {
+    if (IsPinned(key)) {
+      if (log_filtered) {
+        ICEBERG_LOG_WARN(
+            "Ignoring auth-pinned configuration key '{}' from server config; "
+            "this property must be resolved from client configuration only "
+            "(CVE-2024-XXXXX: server config override attack prevention)",
+            key);
+      }
+    } else {
+      filtered[key] = value;
+    }
+  }
+
+  return filtered;
+}
+
+}  // namespace iceberg

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

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <unordered_set>
+
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Utility for managing auth-pinned configuration keys.
+///
+/// Auth-pinned keys are configuration properties that control authentication
+/// routing and credentials. These keys MUST be resolved from client-side
+/// configuration only and MUST NOT be overridden by server-provided config
+/// responses, as doing so would allow a compromised or malicious server to
+/// redirect authentication credentials to an attacker-controlled endpoint.
+///
+/// This class provides filtering for server-provided configuration (e.g.,
+/// REST catalog ConfigResponse overrides) to prevent CVE-2024-XXXXX: OAuth2
+/// credential theft via server config override attack.
+///
+/// ## What is NOT pinned
+///
+/// The following keys are safe for servers to override and are NOT pinned:
+/// - `token-refresh-enabled`, `token-exchange-enabled` (boolean flags)
+/// - `scope`, `audience`, `resource` (OAuth2 scope parameters)
+/// - `rest.access-key-id`, `s3.access-key-id` (public identifiers, not secrets)
+///
+/// ## Example usage
+///
+/// \code
+///   // Filter server overrides before merging into client config
+///   auto safe_overrides = AuthPinnedKeys::FilterServerConfig(
+///       server_config.overrides);
+///
+///   RestCatalogProperties final_config = RestCatalogProperties::FromMap(
+///       MergeConfigs(client_config, safe_overrides));
+/// \endcode
+class ICEBERG_EXPORT AuthPinnedKeys {
+ public:
+  /// \brief Check if a configuration key is auth-pinned.
+  /// \param key The configuration key to check
+  /// \return true if the key must be client-controlled, false otherwise
+  ///
+  /// Auth-pinned keys include:
+  /// - OAuth2 endpoint: `oauth2-server-uri`
+  /// - OAuth2 credentials: `token`, `credential`
+  /// - Basic Auth: `rest.auth.basic.username`, `rest.auth.basic.password`
+  /// - SigV4 secrets: `rest.secret-access-key`, `rest.session-token`
+  static bool IsPinned(std::string_view key);
+
+  /// \brief Filter auth-pinned keys from server-provided configuration.
+  /// \param server_config The server configuration map to filter
+  /// \param log_filtered If true, log a warning for each filtered key
+  /// \return A new map with auth-pinned keys removed
+  ///
+  /// This method is intended for filtering ConfigResponse.defaults and
+  /// ConfigResponse.overrides before merging them into client configuration.
+  /// Filtered keys are removed to prevent server-side override attacks.
+  ///
+  /// Example:
+  /// \code
+  ///   // Prevent f009: OAuth2 credential theft via server config override
+  ///   auto safe_defaults = AuthPinnedKeys::FilterServerConfig(
+  ///       server_config.defaults);
+  ///   auto safe_overrides = AuthPinnedKeys::FilterServerConfig(
+  ///       server_config.overrides);
+  ///
+  ///   RestCatalogProperties final_config = RestCatalogProperties::FromMap(
+  ///       MergeConfigs(safe_defaults, client_config, safe_overrides));
+  /// \endcode
+  static std::unordered_map<std::string, std::string> FilterServerConfig(
+      const std::unordered_map<std::string, std::string>& server_config,
+      bool log_filtered = true);
+
+ private:
+  /// \brief Get the set of auth-pinned keys.
+  ///
+  /// This is an exact list with no pattern matching. New auth-related
+  /// properties that control credential routing should be added here.
+  static const std::unordered_set<std::string>& GetPinnedKeys();
+};
+
+}  // namespace iceberg

src/iceberg/catalog/rest/rest_catalog.cc

@@ -31,6 +31,7 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/catalog/rest/auth/auth_managers.h"
+#include "iceberg/catalog/rest/auth/auth_pinned_keys.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/constant.h"
@@ -443,8 +444,12 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   ICEBERG_ASSIGN_OR_RAISE(auto server_config,
                           FetchServerConfig(*paths, config, *init_session));
 
-  RestCatalogProperties final_config = RestCatalogProperties::FromMap(
-      MergeConfigs(server_config.defaults, config.configs(), server_config.overrides));
+  // Filter sensitive properties from server-provided configuration to prevent
+  // compromised catalog from redirecting OAuth2 credentials or overriding
+  // other authentication-sensitive settings.
+  RestCatalogProperties final_config = RestCatalogProperties::FromMap(MergeConfigs(
+      AuthPinnedKeys::FilterServerConfig(server_config.defaults), config.configs(),
+      AuthPinnedKeys::FilterServerConfig(server_config.overrides)));
 
   std::unordered_set<Endpoint> endpoints;
   if (!server_config.endpoints.empty()) {

src/iceberg/cpp20_compatibility_internal.h

@@ -181,6 +181,7 @@
 #include "iceberg/util/int128.h"
 #include "iceberg/util/lazy.h"
 #include "iceberg/util/location_util.h"
+#include "iceberg/util/log_sanitizer.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/partition_value_util.h"
 #include "iceberg/util/property_util.h"
@@ -226,6 +227,7 @@
 #ifdef ICEBERG_PUBLIC_HEADERS_REST
 #  include "iceberg/catalog/rest/auth/auth_manager.h"
 #  include "iceberg/catalog/rest/auth/auth_managers.h"
+#  include "iceberg/catalog/rest/auth/auth_pinned_keys.h"
 #  include "iceberg/catalog/rest/auth/auth_properties.h"
 #  include "iceberg/catalog/rest/auth/auth_session.h"
 #  include "iceberg/catalog/rest/auth/oauth2_util.h"

src/iceberg/json_serde.cc

@@ -1691,7 +1691,9 @@ Result<nlohmann::json> FromJsonString(const std::string& json_string) {
   auto json =
       nlohmann::json::parse(json_string, /*cb=*/nullptr, /*allow_exceptions=*/false);
   if (json.is_discarded()) [[unlikely]] {
-    return JsonParseError("Failed to parse JSON string: {}", json_string);
+    // Do not echo the raw body: it may be a catalog or token-endpoint response
+    // carrying credential material, and error messages routinely get logged.
+    return JsonParseError("Failed to parse JSON string of {} bytes", json_string.size());
   }
   return json;
 }

src/iceberg/test/CMakeLists.txt

@@ -125,6 +125,7 @@ add_iceberg_test(json_serde_test
 add_iceberg_test(util_test
                  USE_DATA
                  SOURCES
+                 auth_pinned_keys_test.cc
                  base64_test.cc
                  bucket_util_test.cc
                  cache_test.cc
@@ -133,20 +134,21 @@ add_iceberg_test(util_test
                  data_file_set_test.cc
                  decimal_test.cc
                  endian_test.cc
+                 executor_util_test.cc
                  expected_test.cc
                  file_io_test.cc
                  formatter_test.cc
-                 stream_test.cc
                  lazy_test.cc
                  location_util_test.cc
+                 log_sanitizer_test.cc
                  math_util_internal_test.cc
-                 executor_util_test.cc
-                 roaring_position_bitmap_test.cc
                  position_delete_index_test.cc
                  position_delete_range_consumer_test.cc
                  property_util_test.cc
                  resolving_file_io_test.cc
                  retry_util_test.cc
+                 roaring_position_bitmap_test.cc
+                 stream_test.cc
                  string_util_test.cc
                  struct_like_set_test.cc
                  task_group_test.cc
@@ -321,8 +323,9 @@ if(ICEBERG_BUILD_REST)
   add_rest_iceberg_test(rest_catalog_test
                         SOURCES
                         auth_manager_test.cc
-                        error_handlers_test.cc
                         endpoint_test.cc
+                        error_handlers_test.cc
+                        rest_catalog_security_test.cc
                         rest_file_io_test.cc
                         rest_json_serde_test.cc
                         rest_metrics_reporter_test.cc