Add catalog denylist config for Thrift API

metacat-common-server/src/main/java/com/netflix/metacat/common/server/properties/Config.java

@@ -221,6 +221,13 @@ public interface Config {
      */
     String getTagServiceUserAdmin();
 
+    /**
+     * Catalog denylist for the HMS Thrift API.
+     *
+     * @return denylisted catalog names
+     */
+    Set<String> getThriftCatalogDenylist();
+
     /**
      * Thrift server max worker threads.
      *
@@ -658,4 +665,3 @@ public interface Config {
      */
     boolean isDefinitionMetadataSelectForUpdateEnabled();
 }
-

metacat-common-server/src/main/java/com/netflix/metacat/common/server/properties/DefaultConfigImpl.java

@@ -280,6 +280,14 @@ public String getTagServiceUserAdmin() {
         return this.metacatProperties.getTagService().getUserAdmin();
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Set<String> getThriftCatalogDenylist() {
+        return this.metacatProperties.getThrift().getCatalogDenylist();
+    }
+
     /**
      * {@inheritDoc}
      */

metacat-common-server/src/main/java/com/netflix/metacat/common/server/properties/ThriftProperties.java

@@ -19,6 +19,9 @@
 
 import lombok.Data;
 
+import java.util.HashSet;
+import java.util.Set;
+
 /**
  * Properties related to Thrift.
  *
@@ -28,6 +31,7 @@
 @Data
 //TODO: This should be in the Thrift module
 public class ThriftProperties {
+    private Set<String> catalogDenylist = new HashSet<>();
     private int serverMaxWorkerThreads = 100;
     private int serverSocketClientTimeoutInSeconds = 60;
 }

metacat-thrift/src/main/java/com/netflix/metacat/thrift/CatalogThriftHiveMetastore.java

@@ -43,6 +43,7 @@
 import com.netflix.metacat.common.exception.MetacatAlreadyExistsException;
 import com.netflix.metacat.common.exception.MetacatNotFoundException;
 import com.netflix.metacat.common.exception.MetacatPreconditionFailedException;
+import com.netflix.metacat.common.exception.MetacatUnAuthorizedException;
 import com.netflix.metacat.common.server.api.v1.MetacatV1;
 import com.netflix.metacat.common.server.api.v1.PartitionV1;
 import com.netflix.metacat.common.server.monitoring.Metrics;
@@ -1834,6 +1835,9 @@ private TException unimplemented(final String methodName, final Object[] args) t
 
     private <R> R requestWrapper(final String methodName, final Object[] args, final ThriftSupplier<R> supplier)
         throws TException {
+        if (config.getThriftCatalogDenylist().contains(catalogName)) {
+            throw new MetacatUnAuthorizedException("HMS Thrift API is denied for catalog: " + catalogName);
+        }
         final long start = registry.clock().wallTime();
         registry.counter(registry.createId(Metrics.CounterThrift.getMetricName() + "." + methodName)).increment();
         try {

metacat-thrift/src/test/groovy/com/netflix/metacat/thrift/CatalogThriftHiveMetastoreSpec.groovy

@@ -19,11 +19,13 @@ import com.google.common.collect.Lists
 import com.google.common.collect.Maps
 import com.netflix.metacat.common.QualifiedName
 import com.netflix.metacat.common.dto.*
+import com.netflix.metacat.common.exception.MetacatUnAuthorizedException
 import com.netflix.metacat.common.server.api.v1.MetacatV1
 import com.netflix.metacat.common.server.api.v1.PartitionV1
 import com.netflix.metacat.common.server.properties.Config
 import com.netflix.metacat.common.server.util.MetacatContextManager
 import com.netflix.spectator.api.Clock
+import com.netflix.spectator.api.DefaultRegistry
 import com.netflix.spectator.api.Timer
 import com.netflix.spectator.api.Counter
 import com.netflix.spectator.api.Id
@@ -45,7 +47,48 @@ class CatalogThriftHiveMetastoreSpec extends Specification {
 
     def setup() {
         MetacatContextManager.context = catalogServerContext
+        config.getThriftCatalogDenylist() >> Collections.emptySet()
+    }
+
+    def 'denylisted catalog rejects #operation without calling Metacat'() {
+        given:
+        def metastore = new CatalogThriftHiveMetastore(
+                config, hiveConverters, metacatV1, partitionV1, catalogName, new DefaultRegistry())
+
+        when:
+        metastore."$operation"(*arguments)
 
+        then:
+        def exception = thrown(MetacatUnAuthorizedException)
+        exception.message == 'HMS Thrift API is denied for catalog: testcatalogname'
+        config.getThriftCatalogDenylist() >> ['testcatalogname'].toSet()
+        0 * metacatV1._
+        0 * partitionV1._
+
+        where:
+        operation           | arguments
+        'get_all_databases' | []
+        'create_database'   | [new Database(name: 'db1')]
+        'get_table'         | ['db1', 't1']
+        'get_partitions'    | ['db1', 't1', (short) 10]
+        'set_ugi'           | ['user', []]
+    }
+
+    def 'catalog access is allowed when catalog denylist is #catalogDenylist'() {
+        given:
+        def metastore = new CatalogThriftHiveMetastore(
+                config, hiveConverters, metacatV1, partitionV1, catalogName, new DefaultRegistry())
+
+        when:
+        def databases = metastore.get_all_databases()
+
+        then:
+        databases == ['db1']
+        config.getThriftCatalogDenylist() >> catalogDenylist.toSet()
+        1 * metacatV1.getCatalog('testcatalogname', true, false) >> new CatalogDto(databases: ['db1'])
+
+        where:
+        catalogDenylist << [[], ['othercatalog']]
     }
 
     def 'test abort_txn'() {