REST: disallow overriding "credential" in table sessions

See #10256 for context.

This change disallows overriding the "credential" property
in table sessions, by introducing an allow-list of
auth-related properties that can be overridden in such
situations.

Only the "token" property and properties used to exchange
one token for another ("urn:ietf:params:oauth:token-type:*")
are now allowed.

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -68,6 +68,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.rest.auth.AuthConfig;
@@ -124,6 +125,13 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
           OAuth2Properties.SAML2_TOKEN_TYPE,
           OAuth2Properties.SAML1_TOKEN_TYPE);
 
+  // Auth-related properties that are allowed to be passed to the table session
+  private static final Set<String> TABLE_SESSION_ALLOW_LIST =
+      ImmutableSet.<String>builder()
+          .add(OAuth2Properties.TOKEN)
+          .addAll(TOKEN_PREFERENCE_ORDER)
+          .build();
+
   private final Function<Map<String, String>, RESTClient> clientBuilder;
   private final BiFunction<SessionContext, Map<String, String>, FileIO> ioBuilder;
   private Cache<String, AuthSession> sessions = null;
@@ -922,7 +930,14 @@ private FileIO tableFileIO(SessionContext context, Map<String, String> config) {
   }
 
   private AuthSession tableSession(Map<String, String> tableConf, AuthSession parent) {
-    Pair<String, Supplier<AuthSession>> newSession = newSession(tableConf, tableConf, parent);
+    Map<String, String> credentials = Maps.newHashMapWithExpectedSize(tableConf.size());
+    for (String prop : tableConf.keySet()) {
+      if (TABLE_SESSION_ALLOW_LIST.contains(prop)) {
+        credentials.put(prop, tableConf.get(prop));
+      }
+    }
+
+    Pair<String, Supplier<AuthSession>> newSession = newSession(credentials, tableConf, parent);
     if (null == newSession) {
       return parent;
     }

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

@@ -902,9 +902,9 @@ public void testTableCredential(String oauth2ServerUri) {
     testTableAuth(
         "catalog",
         ImmutableMap.of("urn:ietf:params:oauth:token-type:id_token", "id-token"),
-        ImmutableMap.of("credential", "table-user:secret"),
+        ImmutableMap.of("credential", "table-user:secret"), // will be ignored
+        ImmutableMap.of("Authorization", "Bearer token-exchange-token:sub=id-token,act=catalog"),
         ImmutableMap.of("Authorization", "Bearer token-exchange-token:sub=id-token,act=catalog"),
-        ImmutableMap.of("Authorization", "Bearer client-credentials-token:sub=table-user"),
         oauth2ServerUri);
   }
 
@@ -1354,9 +1354,9 @@ public void testTableAuth(
             eq(expectedContextHeaders),
             any());
 
-    // if the table returned a bearer token, there will be no token request
-    if (!tableConfig.containsKey("token")) {
-      // client credentials or token exchange to get a table token
+    // if the table returned a bearer token or a credential, there will be no token request
+    if (!tableConfig.containsKey("token") && !tableConfig.containsKey("credential")) {
+      // token exchange to get a table token
       Mockito.verify(adapter, times(1))
           .execute(
               eq(HTTPMethod.POST),
@@ -1368,38 +1368,40 @@ public void testTableAuth(
               any());
     }
 
-    // automatic refresh when metadata is accessed after commit
-    Mockito.verify(adapter)
-        .execute(
-            eq(HTTPMethod.GET),
-            eq("v1/namespaces/ns/tables/table"),
-            any(),
-            any(),
-            eq(LoadTableResponse.class),
-            eq(expectedTableHeaders),
-            any());
-
-    // load table from catalog
-    Mockito.verify(adapter)
-        .execute(
-            eq(HTTPMethod.GET),
-            eq("v1/namespaces/ns/tables/table"),
-            any(),
-            any(),
-            eq(LoadTableResponse.class),
-            eq(expectedContextHeaders),
-            any());
+    if (expectedContextHeaders.equals(expectedTableHeaders)) {
+      // load table from catalog + refresh loaded table
+      Mockito.verify(adapter, times(2))
+          .execute(
+              eq(HTTPMethod.GET),
+              eq("v1/namespaces/ns/tables/table"),
+              any(),
+              any(),
+              eq(LoadTableResponse.class),
+              eq(expectedTableHeaders),
+              any());
+    } else {
+      // load table from catalog
+      Mockito.verify(adapter)
+          .execute(
+              eq(HTTPMethod.GET),
+              eq("v1/namespaces/ns/tables/table"),
+              any(),
+              any(),
+              eq(LoadTableResponse.class),
+              eq(expectedContextHeaders),
+              any());
 
-    // refresh loaded table
-    Mockito.verify(adapter)
-        .execute(
-            eq(HTTPMethod.GET),
-            eq("v1/namespaces/ns/tables/table"),
-            any(),
-            any(),
-            eq(LoadTableResponse.class),
-            eq(expectedTableHeaders),
-            any());
+      // refresh loaded table
+      Mockito.verify(adapter)
+          .execute(
+              eq(HTTPMethod.GET),
+              eq("v1/namespaces/ns/tables/table"),
+              any(),
+              any(),
+              eq(LoadTableResponse.class),
+              eq(expectedTableHeaders),
+              any());
+    }
   }
 
   @ParameterizedTest