HIVE-30002 : Restrict Iceberg REST catalog table registration and drop-purge to authorized/contained locations

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -241,6 +241,10 @@ public String name() {
     return name;
   }
 
+  public FileIO io() {
+    return fileIO;
+  }
+
   @Override
   public boolean dropTable(TableIdentifier identifier, boolean purge) {
     if (!isValidIdentifier(identifier)) {
@@ -271,7 +275,7 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) {
       });
 
       if (purge && lastMetadata != null) {
-        CatalogUtil.dropTableData(ops.io(), lastMetadata);
+        CatalogUtil.dropTableData(new ScopedDeleteFileIO(ops.io(), lastMetadata.location()), lastMetadata);
       }
 
       LOG.info("Dropped table: {}", identifier);

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/ScopedDeleteFileIO.java

@@ -0,0 +1,91 @@
+/*
+ * 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.hive;
+
+import java.util.Map;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link FileIO} decorator used by {@link HiveCatalog#dropTable(org.apache.iceberg.catalog.TableIdentifier,
+ * boolean)} to fence purge deletions to files under a table's own location, regardless of what a table's
+ * metadata or manifests actually reference.
+ *
+ * <p>This does not implement {@link org.apache.iceberg.io.SupportsBulkOperations} or
+ * {@link org.apache.iceberg.io.SupportsPrefixOperations} even when the delegate does, so that
+ * {@code CatalogUtil.dropTableData} is forced to route every deletion through {@link #deleteFile(String)}.
+ */
+class ScopedDeleteFileIO implements FileIO {
+  private static final Logger LOG = LoggerFactory.getLogger(ScopedDeleteFileIO.class);
+
+  private final FileIO delegate;
+  private final String location;
+
+  ScopedDeleteFileIO(FileIO delegate, String location) {
+    this.delegate = delegate;
+    this.location = normalize(location);
+  }
+
+  @Override
+  public InputFile newInputFile(String path) {
+    return delegate.newInputFile(path);
+  }
+
+  @Override
+  public OutputFile newOutputFile(String path) {
+    return delegate.newOutputFile(path);
+  }
+
+  @Override
+  public void deleteFile(String path) {
+    if (!isContained(location, normalize(path))) {
+      LOG.warn("Skipping delete outside table location {}: {}", location, path);
+      return;
+    }
+    delegate.deleteFile(path);
+  }
+
+  @Override
+  public Map<String, String> properties() {
+    return delegate.properties();
+  }
+
+  @Override
+  public void initialize(Map<String, String> properties) {
+    delegate.initialize(properties);
+  }
+
+  @Override
+  public void close() {
+    delegate.close();
+  }
+
+  private static boolean isContained(String root, String candidate) {
+    return candidate.equals(root) || candidate.startsWith(root.endsWith("/") ? root : root + "/");
+  }
+
+  private static String normalize(String location) {
+    return new Path(location).toUri().normalize().toString();
+  }
+}

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

@@ -52,6 +52,8 @@
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
 import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.base.Splitter;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -298,10 +300,13 @@
   }
 
   private RESTResponse dropTable(Map<String, String> vars) {
+    TableIdentifier ident = identFromPathVars(vars);
     if (PropertyUtil.propertyAsBoolean(vars, "purgeRequested", false)) {
-      CatalogHandlers.purgeTable(catalog, identFromPathVars(vars));
+      String location = catalog.loadTable(ident).location();
+      icebergAuthorizer.validateDropTablePurge(catalogName, ident, location);
+      CatalogHandlers.purgeTable(catalog, ident);
     } else {
-      CatalogHandlers.dropTable(catalog, identFromPathVars(vars));
+      CatalogHandlers.dropTable(catalog, ident);
     }
     return null;
   }
@@ -320,6 +325,10 @@
   private LoadTableResponse registerTable(Map<String, String> vars, Object body) {
       Namespace namespace = namespaceFromPathVars(vars);
       RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
+      request.validate();
+      Map<String, String> namespaceMetadata = asNamespaceCatalog.loadNamespaceMetadata(namespace);
+      FileIO io = ((HiveCatalog) catalog).io();
+      icebergAuthorizer.validateRegisterTable(catalogName, namespace, namespaceMetadata, request, io);
       return castResponse(LoadTableResponse.class, CatalogHandlers.registerTable(catalog, namespace, request));
   }
 

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

@@ -30,6 +30,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.PrincipalType;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
@@ -44,10 +45,14 @@
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HiveOperationType;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.metastore.HiveMetaStoreAuthorizer;
+import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.hive.HiveHadoopUtil;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -64,7 +69,10 @@ class IcebergAuthorizer {
   @VisibleForTesting
   final Supplier<HiveAuthorizer> authorizerSupplier;
 
+  private final Configuration conf;
+
   IcebergAuthorizer(Configuration conf) {
+    this.conf = conf;
     final var classes = MetastoreConf.getTrimmedStringsVar(conf, MetastoreConf.ConfVars.PRE_EVENT_LISTENERS);
     if (classes.length == 0) {
       LOG.info("No pre-event listeners configured, skipping authorization checks");
@@ -100,7 +108,13 @@ class IcebergAuthorizer {
 
   @VisibleForTesting
   IcebergAuthorizer(Supplier<HiveAuthorizer> authorizerSupplier) {
+    this(authorizerSupplier, new Configuration(false));
+  }
+
+  @VisibleForTesting
+  IcebergAuthorizer(Supplier<HiveAuthorizer> authorizerSupplier, Configuration conf) {
     this.authorizerSupplier = authorizerSupplier;
+    this.conf = conf;
   }
 
   /**
@@ -161,4 +175,133 @@ void validateStageCreateTable(String catalogName, Namespace namespace, Map<Strin
       throw new IllegalStateException("Failed to check privileges stage-create", e);
     }
   }
+
+  /**
+   * Enforces authorization for REGISTER_TABLE. The request's {@code metadataLocation} is fetched with the
+   * catalog's shared, service-level {@link FileIO}, so both that location and the {@code location()} embedded in
+   * the metadata file it points to (which becomes the table's HMS {@code StorageDescriptor.location}, and is what
+   * a later purge trusts as its deletion root, see {@link #validateDropTablePurge}) must be authorized. Otherwise
+   * REGISTER_TABLE is an arbitrary-file-read primitive that returns any metadata file's contents to the caller.
+   *
+   * <p>When no {@code HiveAuthorizer} is configured, falls back to requiring both locations to be contained in
+   * the namespace's external or managed root, since there is no policy to otherwise decide whether the caller may
+   * read an arbitrary location with service credentials.
+   *
+   * @param catalogName the Hive catalog name
+   * @param namespace the Iceberg namespace
+   * @param namespaceMetadata the Iceberg namespace metadata
+   * @param request the register table request
+   * @param io the {@link FileIO} used to read the metadata file
+   * @throws ForbiddenException if a location is not authorized, or not contained in the namespace
+   * @throws IllegalStateException if the authorization plugin fails
+   */
+  void validateRegisterTable(String catalogName, Namespace namespace, Map<String, String> namespaceMetadata,
+      RegisterTableRequest request, FileIO io) {
+    Preconditions.checkArgument(namespace.levels().length == 1, "Hive does not support multi-level namespaces");
+    var databaseName = namespace.level(0);
+    var commandString = "register table " + request.name();
+    checkLocationAuthorized(catalogName, databaseName, namespaceMetadata, request.metadataLocation(), commandString);
+
+    var metadata = TableMetadataParser.read(io, request.metadataLocation());
+    checkLocationAuthorized(catalogName, databaseName, namespaceMetadata, metadata.location(), commandString);
+  }
+
+  /**
+   * Enforces authorization for DROP_TABLE with {@code purge=true}. Purge deletes every file referenced by the
+   * table's current metadata using the catalog's shared, service-level {@link FileIO}, so the location must be
+   * authorized like any other DFS_URI access.
+   *
+   * <p>Unlike {@link #validateRegisterTable}, there is no namespace-containment fallback here: the structural
+   * fence in {@code HiveCatalog.dropTable} already restricts purge deletions to files under the table's own
+   * location regardless of whether a {@code HiveAuthorizer} is configured, so a deployment without one relies on
+   * that fence rather than this check.
+   *
+   * @param catalogName the Hive catalog name
+   * @param identifier the table identifier being dropped
+   * @param location the table's current location
+   * @throws ForbiddenException if the location is not authorized
+   * @throws IllegalStateException if the authorization plugin fails
+   */
+  void validateDropTablePurge(String catalogName, TableIdentifier identifier, String location) {
+    var authorizer = authorizerSupplier.get();
+    if (authorizer == null) {
+      LOG.info("No pre-event listener is configured for catalog {}, skipping drop-table-purge authorization for {}",
+          catalogName, identifier);
+      return;
+    }
+
+    var inputs = Collections.singletonList(
+        new HivePrivilegeObject(HivePrivilegeObject.HivePrivilegeObjectType.DFS_URI, location));
+    var builder = new HiveAuthzContext.Builder();
+    builder.setCommandString("drop table " + identifier.name());
+    try {
+      authorizer.checkPrivileges(HiveOperationType.DROPTABLE, inputs, Collections.emptyList(), builder.build());
+    } catch (HiveAccessControlException e) {
+      throw new ForbiddenException(e, e.getMessage());
+    } catch (HiveAuthzPluginException e) {
+      throw new IllegalStateException("Failed to check privileges drop-table-purge", e);
+    }
+  }
+
+  private void checkLocationAuthorized(String catalogName, String databaseName,
+      Map<String, String> namespaceMetadata, String location, String commandString) {
+    var authorizer = authorizerSupplier.get();
+    if (authorizer == null) {
+      LOG.info("No pre-event listener is configured for catalog {}, falling back to namespace containment for {}",
+          catalogName, location);
+      checkContainedInNamespace(databaseName, namespaceMetadata, location);
+      return;
+    }
+
+    var inputs = Collections.singletonList(
+        new HivePrivilegeObject(HivePrivilegeObject.HivePrivilegeObjectType.DFS_URI, location));
+    var builder = new HiveAuthzContext.Builder();
+    builder.setCommandString(commandString);
+    try {
+      authorizer.checkPrivileges(HiveOperationType.CREATETABLE, inputs, Collections.emptyList(), builder.build());
+    } catch (HiveAccessControlException e) {
+      throw new ForbiddenException(e, e.getMessage());
+    } catch (HiveAuthzPluginException e) {
+      throw new IllegalStateException("Failed to check privileges for " + commandString, e);
+    }
+  }
+
+  private void checkContainedInNamespace(String databaseName, Map<String, String> namespaceMetadata,
+      String location) {
+    var externalRoot = namespaceMetadata.get("location");
+    if (externalRoot != null && isContained(externalRoot, location)) {
+      return;
+    }
+    if (isContained(managedNamespaceLocation(databaseName), location)) {
+      return;
+    }
+    throw new ForbiddenException(
+        "Location %s is not authorized and is not contained in namespace %s", location, databaseName);
+  }
+
+  private String managedNamespaceLocation(String databaseName) {
+    var warehouseLocation = conf.get(HiveConf.ConfVars.METASTORE_WAREHOUSE.varname);
+    Preconditions.checkNotNull(warehouseLocation, "Warehouse location is not set: hive.metastore.warehouse.dir=null");
+    if (warehouseLocation.endsWith("/")) {
+      warehouseLocation = warehouseLocation.substring(0, warehouseLocation.length() - 1);
+    }
+    return String.format("%s/%s.db", warehouseLocation, databaseName);
+  }
+
+  /**
+   * Checks whether {@code candidate} resolves under {@code root}. Both are resolved via {@link Path#toUri()} and
+   * {@link java.net.URI#normalize()}, which -- unlike {@link Path}'s own normalization -- actually collapses
+   * {@code .}/{@code ..} segments; a plain string-prefix comparison on unnormalized paths would let a location
+   * like {@code root/../../elsewhere} pass a naive check while actually resolving outside {@code root}.
+   */
+  private static boolean isContained(String root, String candidate) {
+    var normalizedRoot = normalize(root);
+    var normalizedCandidate = normalize(candidate);
+    return normalizedCandidate.equals(normalizedRoot)
+        || normalizedCandidate.startsWith(normalizedRoot.endsWith("/") ? normalizedRoot : normalizedRoot + "/");
+  }
+
+  private static String normalize(String location) {
+    return new Path(location).toUri().normalize().toString();
+  }
 }