feat: Implement OAuth2 authentication for REST catalog

Author: shuxu.li

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

@@ -180,11 +180,12 @@ class OAuth2AuthManager : public AuthManager {
     const auto& credential = credential_it->second;
     auto colon_pos = credential.find(':');
     if (colon_pos == std::string::npos) {
-      return InvalidArgument(
-          "Invalid OAuth2 credential format: expected 'client_id:client_secret'");
+      // No colon: treat entire string as client_secret with empty client_id.
+      ctx.client_secret = credential;
+    } else {
+      ctx.client_id = credential.substr(0, colon_pos);
+      ctx.client_secret = credential.substr(colon_pos + 1);
     }
-    ctx.client_id = credential.substr(0, colon_pos);
-    ctx.client_secret = credential.substr(colon_pos + 1);
 
     auto uri_it = properties.find(AuthProperties::kOAuth2ServerUri);
     if (uri_it != properties.end() && !uri_it->second.empty()) {

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

@@ -63,6 +63,8 @@ Result<OAuthTokenResponse> OAuthTokenResponseFromJsonString(const std::string& j
                           GetJsonValue<std::string>(json, kAccessToken));
   ICEBERG_ASSIGN_OR_RAISE(response.token_type,
                           GetJsonValue<std::string>(json, kTokenType));
+  // TODO(lishuxu): When implementing auto-refresh, extract exp claim from JWT if
+  // expires_in is missing.
   ICEBERG_ASSIGN_OR_RAISE(response.expires_in,
                           GetJsonValueOrDefault<int64_t>(json, kExpiresIn, 0));
   ICEBERG_ASSIGN_OR_RAISE(response.refresh_token,
@@ -80,9 +82,11 @@ Result<OAuthTokenResponse> FetchToken(HttpClient& client,
                                       const std::string& scope, AuthSession& session) {
   std::unordered_map<std::string, std::string> form_data{
       {std::string(kGrantType), std::string(kClientCredentials)},
-      {std::string(kClientId), client_id},
       {std::string(kClientSecret), client_secret},
   };
+  if (!client_id.empty()) {
+    form_data.emplace(std::string(kClientId), client_id);
+  }
   if (!scope.empty()) {
     form_data.emplace(std::string(kScope), scope);
   }

src/iceberg/test/auth_manager_test.cc

@@ -251,19 +251,24 @@ TEST_F(AuthManagerTest, OAuth2MissingCredentials) {
 }
 
 // Verifies OAuth2 fails with invalid credential format
-TEST_F(AuthManagerTest, OAuth2InvalidCredentialFormat) {
+// Verifies credential without colon is treated as client_secret only
+TEST_F(AuthManagerTest, OAuth2CredentialWithoutColon) {
   std::unordered_map<std::string, std::string> properties = {
       {AuthProperties::kAuthType, "oauth2"},
-      {AuthProperties::kOAuth2Credential, "no-colon-separator"},
+      {AuthProperties::kOAuth2Credential, "secret-only"},
+      {AuthProperties::kOAuth2Token, "my-static-token"},
       {"uri", "http://localhost:8181"},
   };
 
   auto manager_result = AuthManagers::Load("test-catalog", properties);
   ASSERT_THAT(manager_result, IsOk());
 
   auto session_result = manager_result.value()->CatalogSession(client_, properties);
-  EXPECT_THAT(session_result, IsError(ErrorKind::kInvalidArgument));
-  EXPECT_THAT(session_result, HasErrorMessage("client_id:client_secret"));
+  ASSERT_THAT(session_result, IsOk());
+
+  std::unordered_map<std::string, std::string> headers;
+  ASSERT_THAT(session_result.value()->Authenticate(headers), IsOk());
+  EXPECT_EQ(headers["Authorization"], "Bearer my-static-token");
 }
 
 // Verifies OAuthTokenResponse JSON parsing