HIVE-29020: Support OAuth 2 in Iceberg REST Catalog

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -983,8 +983,8 @@ public enum ConfVars {
                 "  NOSASL:  Raw transport" +
                 "  JWT:  JSON Web Token authentication via JWT token. Only supported in Http/Https mode"),
     THRIFT_METASTORE_AUTHENTICATION_JWT_JWKS_URL("metastore.authentication.jwt.jwks.url",
-        "hive.metastore.authentication.jwt.jwks.url", "", "File URL from where URLBasedJWKSProvider "
-        + "in metastore server will try to load JWKS to match a JWT sent in HTTP request header. Used only when "
+        "hive.metastore.authentication.jwt.jwks.url", "", "File URL from where "
+        + "metastore server will try to load JWKS to match a JWT sent in HTTP request header. Used only when "
         + "Hive metastore server is running in JWT auth mode"),
     METASTORE_CUSTOM_AUTHENTICATION_CLASS("metastore.custom.authentication.class",
             "hive.metastore.custom.authentication.class",
@@ -1873,8 +1873,56 @@ public enum ConfVars {
             " positive value will be used as-is."
     ),
     CATALOG_SERVLET_AUTH("metastore.catalog.servlet.auth",
-        "hive.metastore.catalog.servlet.auth", "jwt", new StringSetValidator("none", "simple", "jwt"),
-        "HMS Catalog servlet authentication method (none, simple, or jwt)."
+        "hive.metastore.catalog.servlet.auth", "jwt", new StringSetValidator("none", "simple", "jwt", "oauth2"),
+        "HMS Catalog servlet authentication method (none, simple, jwt, or oauth2)."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_ISSUER("metastore.catalog.servlet.auth.oauth2.issuer",
+        "hive.metastore.catalog.servlet.auth.oauth2.issuer", "",
+        "The authorization server's identifier, which is a URL. This is required when you use " +
+        "metastore.catalog.servlet.auth=oauth2"
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_AUDIENCE("metastore.catalog.servlet.auth.oauth2.audience",
+        "hive.metastore.catalog.servlet.auth.oauth2.audience", "",
+        "The acceptable name in the audience(aud) claim. This is required when you use " +
+        "metastore.catalog.servlet.auth=oauth2"
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_VALIDATION_METHOD("metastore.catalog.servlet.auth.oauth2.validation.method",
+        "hive.metastore.catalog.servlet.auth.oauth2.validation.method", "jwt",
+        new StringSetValidator("jwt", "introspection"),
+        "How to evaluate an access token. When your authorization server issues opaque tokens or you need " +
+        "to consider additional security requirements such as token revocations, use introspection."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_CLIENT_ID("metastore.catalog.servlet.auth.oauth2.client.id",
+        "hive.metastore.catalog.servlet.auth.oauth2.client.id", "",
+        "The client ID to authenticate HMS, as a resource server, to the introspection endpoint. This is required to " +
+        "use metastore.catalog.servlet.auth.oauth2.validation.method=introspection."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_CLIENT_SECRET("metastore.catalog.servlet.auth.oauth2.client.secret",
+        "hive.metastore.catalog.servlet.auth.oauth2.client.secret", "",
+        "The client secret to authenticate HMS, as a resource server, to the introspection endpoint. This is " +
+        "required to use metastore.catalog.servlet.auth.oauth2.validation.method=introspection."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_INTROSPECTION_CACHE_EXPIRY(
+        "metastore.catalog.servlet.auth.oauth2.introspection.cache.expiry",
+        "hive.metastore.catalog.servlet.auth.oauth2.introspection.cache.expiry", 60, TimeUnit.SECONDS,
+        "The expiry time of the token introspection cache. Set to 0 to disable caching."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_INTROSPECTION_CACHE_SIZE(
+        "metastore.catalog.servlet.auth.oauth2.introspection.cache.num",
+        "hive.metastore.catalog.servlet.auth.oauth2.introspection.cache.num", 1000L,
+        "The number of entries of the token introspection cache."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_PRINCIPAL_MAPPER_REGEX_FIELD(
+        "metastore.catalog.servlet.auth.oauth2.principal.mapper.regex.username.field",
+        "hive.metastore.catalog.servlet.auth.oauth2.principal.mapper.regex.username.field", "sub",
+        "The claim name including a username. This is effective when you use RegexPrincipalMapper. For example, if " +
+            "you want to resolve a user name from the email claim, set this to email."
+    ),
+    CATALOG_SERVLET_AUTH_OAUTH2_PRINCIPAL_MAPPER_REGEX_PATTERN(
+        "metastore.catalog.servlet.auth.oauth2.principal.mapper.regex.username.pattern",
+        "hive.metastore.catalog.servlet.auth.oauth2.principal.mapper.regex.username.pattern", "(.*)",
+        "The pattern to extract a user name. This is effective when you use RegexPrincipalMapper. For example, if " +
+        "you want to extract a user name from the local part of the email claim, set this to (.*)@example.com."
     ),
     ICEBERG_CATALOG_SERVLET_PATH("metastore.iceberg.catalog.servlet.path",
         "hive.metastore.iceberg.catalog.servlet.path", "iceberg",

standalone-metastore/metastore-rest-catalog/pom.xml

@@ -224,6 +224,16 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.keycloak</groupId>
+      <artifactId>keycloak-admin-client</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java

@@ -71,7 +71,6 @@
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.iceberg.rest.responses.LoadViewResponse;
-import org.apache.iceberg.rest.responses.OAuthTokenResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PropertyUtil;
@@ -104,15 +103,6 @@ public class HMSCatalogAdapter implements RESTClient {
           .put(CommitStateUnknownException.class, 500)
           .buildOrThrow();
 
-  private static final String URN_OAUTH_TOKEN_EXCHANGE = "urn:ietf:params:oauth:grant-type:token-exchange";
-  private static final String URN_OAUTH_ACCESS_TOKEN = "urn:ietf:params:oauth:token-type:access_token";
-  private static final String GRANT_TYPE = "grant_type";
-  private static final String CLIENT_CREDENTIALS = "client_credentials";
-  private static final String BEARER = "Bearer";
-  private static final String CLIENT_ID = "client_id";
-  private static final String ACTOR_TOKEN = "actor_token";
-  private static final String SUBJECT_TOKEN = "subject_token";
-
   private final Catalog catalog;
   private final SupportsNamespaces asNamespaceCatalog;
   private final ViewCatalog asViewCatalog;
@@ -127,8 +117,6 @@ public HMSCatalogAdapter(Catalog catalog) {
   }
 
   enum Route {
-    TOKENS(HTTPMethod.POST, "v1/oauth/tokens", null),
-    SEPARATE_AUTH_TOKENS_URI(HTTPMethod.POST, "https://auth-server.com/token", null),
     CONFIG(HTTPMethod.GET, "v1/config", null),
     LIST_NAMESPACES(HTTPMethod.GET, ResourcePaths.V1_NAMESPACES, null),
     CREATE_NAMESPACE(HTTPMethod.POST, ResourcePaths.V1_NAMESPACES, CreateNamespaceRequest.class),
@@ -240,35 +228,6 @@ private ConfigResponse config() {
     return castResponse(ConfigResponse.class, ConfigResponse.builder().withEndpoints(endpoints).build());
   }
 
-  private OAuthTokenResponse tokens(Object body) {
-    @SuppressWarnings("unchecked")
-    Map<String, String> request = (Map<String, String>) castRequest(Map.class, body);
-    String grantType = request.get(GRANT_TYPE);
-    switch (grantType) {
-      case CLIENT_CREDENTIALS:
-        return OAuthTokenResponse.builder()
-            .withToken("client-credentials-token:sub=" + request.get(CLIENT_ID))
-            .withIssuedTokenType(URN_OAUTH_ACCESS_TOKEN)
-            .withTokenType(BEARER)
-            .build();
-
-      case URN_OAUTH_TOKEN_EXCHANGE:
-        String actor = request.get(ACTOR_TOKEN);
-        String token =
-            String.format(
-                "token-exchange-token:sub=%s%s",
-                request.get(SUBJECT_TOKEN), actor != null ? ",act=" + actor : "");
-        return OAuthTokenResponse.builder()
-            .withToken(token)
-            .withIssuedTokenType(URN_OAUTH_ACCESS_TOKEN)
-            .withTokenType(BEARER)
-            .build();
-
-      default:
-        throw new UnsupportedOperationException("Unsupported grant_type: " + grantType);
-    }
-  }
-
   private ListNamespacesResponse listNamespaces(Map<String, String> vars) {
     Namespace namespace;
     if (vars.containsKey("parent")) {
@@ -469,9 +428,6 @@ private <T extends RESTResponse> T handleRequest(
       counter.inc();
     }
     switch (route) {
-      case TOKENS:
-        return (T) tokens(body);
-
       case CONFIG:
         return (T) config();
 

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.rest;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import javax.servlet.http.HttpServlet;
@@ -100,7 +102,9 @@ private Catalog createCatalog() {
    */
   private HttpServlet createServlet(Catalog catalog) {
     String authType = MetastoreConf.getVar(configuration, ConfVars.CATALOG_SERVLET_AUTH);
-    ServletSecurity security = new ServletSecurity(AuthType.fromString(authType), configuration);
+    // Iceberg REST client uses "catalog" by default
+    List<String> scopes = Collections.singletonList("catalog");
+    ServletSecurity security = new ServletSecurity(AuthType.fromString(authType), configuration, req -> scopes);
     return security.proxy(new HMSCatalogServlet(new HMSCatalogAdapter(catalog)));
   }
 

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogServlet.java

@@ -20,8 +20,6 @@
 package org.apache.iceberg.rest;
 
 import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.Reader;
 import java.io.UncheckedIOException;
 import java.util.Map;
 import java.util.Optional;
@@ -31,7 +29,6 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.relocated.com.google.common.io.CharStreams;
 import org.apache.iceberg.rest.HMSCatalogAdapter.Route;
 import org.apache.iceberg.rest.HTTPRequest.HTTPMethod;
 import org.apache.iceberg.rest.responses.ErrorResponse;
@@ -152,10 +149,6 @@ static ServletRequestContext from(HttpServletRequest request) throws IOException
       if (route.requestClass() != null) {
         requestBody =
             RESTObjectMapper.mapper().readValue(request.getReader(), route.requestClass());
-      } else if (route == Route.TOKENS) {
-        try (Reader reader = new InputStreamReader(request.getInputStream())) {
-          requestBody = RESTUtil.decodeFormData(CharStreams.toString(reader));
-        }
       }
 
       Map<String, String> queryParams =

standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestRESTCatalogOAuth2Jwt.java

@@ -0,0 +1,83 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.rest;
+
+import java.util.Map;
+import org.apache.hadoop.hive.metastore.ServletSecurity.AuthType;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreCheckinTest;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.rest.extension.HiveRESTCatalogServerExtension;
+import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+@Category(MetastoreCheckinTest.class)
+class TestRESTCatalogOAuth2Jwt extends BaseRESTCatalogTests {
+  @RegisterExtension
+  private static final HiveRESTCatalogServerExtension REST_CATALOG_EXTENSION =
+      HiveRESTCatalogServerExtension.builder(AuthType.OAUTH2).build();
+
+  @Override
+  protected Map<String, String> getDefaultClientConfiguration() {
+    return Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "oauth2-server-uri", REST_CATALOG_EXTENSION.getOAuth2TokenEndpoint(),
+        "credential", REST_CATALOG_EXTENSION.getOAuth2ClientCredential()
+    );
+  }
+
+  @Test
+  void testWithAccessToken() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "token", REST_CATALOG_EXTENSION.getOAuth2AccessToken()
+    );
+    Assertions.assertFalse(RCKUtils.initCatalogClient(properties).listNamespaces().isEmpty());
+  }
+
+  @Test
+  void testWithWrongCredential() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "oauth2-server-uri", REST_CATALOG_EXTENSION.getOAuth2TokenEndpoint(),
+        "credential", "dummy:dummy"
+    );
+    NotAuthorizedException error = Assertions.assertThrows(NotAuthorizedException.class,
+        () -> RCKUtils.initCatalogClient(properties));
+    Assertions.assertEquals("Not authorized: invalid_client: Invalid client or Invalid client credentials",
+        error.getMessage());
+  }
+
+  @Test
+  void testWithWrongAccessToken() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "token", "invalid"
+    );
+    NotAuthorizedException error = Assertions.assertThrows(NotAuthorizedException.class,
+        () -> RCKUtils.initCatalogClient(properties));
+    Assertions.assertEquals("Not authorized: Authentication error: Invalid JWT serialization: Missing dot delimiter(s)",
+        error.getMessage());
+  }
+}

standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestRESTCatalogOAuth2TokenIntrospection.java

@@ -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.
+ */
+
+package org.apache.iceberg.rest;
+
+import static org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.CATALOG_SERVLET_AUTH_OAUTH2_VALIDATION_METHOD;
+
+import java.util.Map;
+import org.apache.hadoop.hive.metastore.ServletSecurity.AuthType;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreCheckinTest;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.rest.extension.HiveRESTCatalogServerExtension;
+import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+@Category(MetastoreCheckinTest.class)
+class TestRESTCatalogOAuth2TokenIntrospection extends BaseRESTCatalogTests {
+  @RegisterExtension
+  private static final HiveRESTCatalogServerExtension REST_CATALOG_EXTENSION =
+      HiveRESTCatalogServerExtension.builder(AuthType.OAUTH2)
+          .configure(CATALOG_SERVLET_AUTH_OAUTH2_VALIDATION_METHOD.getVarname(), "introspection").build();
+
+  @Override
+  protected Map<String, String> getDefaultClientConfiguration() {
+    return Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "oauth2-server-uri", REST_CATALOG_EXTENSION.getOAuth2TokenEndpoint(),
+        "credential", REST_CATALOG_EXTENSION.getOAuth2ClientCredential()
+    );
+  }
+
+  @Test
+  void testWithAccessToken() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "token", REST_CATALOG_EXTENSION.getOAuth2AccessToken()
+    );
+    Assertions.assertFalse(RCKUtils.initCatalogClient(properties).listNamespaces().isEmpty());
+  }
+
+  @Test
+  void testWithWrongCredential() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "oauth2-server-uri", REST_CATALOG_EXTENSION.getOAuth2TokenEndpoint(),
+        "credential", "dummy:dummy"
+    );
+    NotAuthorizedException error = Assertions.assertThrows(NotAuthorizedException.class,
+        () -> RCKUtils.initCatalogClient(properties));
+    Assertions.assertEquals("Not authorized: invalid_client: Invalid client or Invalid client credentials",
+        error.getMessage());
+  }
+
+  @Test
+  void testWithWrongAccessToken() {
+    Map<String, String> properties = Map.of(
+        "uri", REST_CATALOG_EXTENSION.getRestEndpoint(),
+        "rest.auth.type", "oauth2",
+        "token", "invalid"
+    );
+    NotAuthorizedException error = Assertions.assertThrows(NotAuthorizedException.class,
+        () -> RCKUtils.initCatalogClient(properties));
+    Assertions.assertEquals("Not authorized: Authentication error: The token is not active",
+        error.getMessage());
+  }
+}