catalog_access_control.diff

diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/CatalogSystemTable.java b/presto-main/src/main/java/com/facebook/presto/connector/system/CatalogSystemTable.java
index 3c0e699..81d3544 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/CatalogSystemTable.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/CatalogSystemTable.java
@@ -27,6 +27,7 @@ import com.facebook.presto.spi.predicate.TupleDomain;
 import javax.inject.Inject;
 
 import java.util.Map;
+import java.util.Optional;
 
 import static com.facebook.presto.metadata.MetadataUtil.TableMetadataBuilder.tableMetadataBuilder;
 import static com.facebook.presto.spi.SystemTable.Distribution.SINGLE_COORDINATOR;
@@ -66,7 +67,7 @@ public class CatalogSystemTable
     public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
     {
         Builder table = InMemoryRecordSet.builder(CATALOG_TABLE);
-        for (Map.Entry<String, String> entry : metadata.getCatalogNames().entrySet()) {
+        for (Map.Entry<String, String> entry : metadata.getCatalogNames(Optional.of(session.getIdentity())).entrySet()) {
             table.addRow(entry.getKey(), entry.getValue());
         }
         return table.build().cursor();
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/CatalogJdbcTable.java b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/CatalogJdbcTable.java
index 285308b..c27ad88 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/CatalogJdbcTable.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/CatalogJdbcTable.java
@@ -25,6 +25,8 @@ import com.facebook.presto.spi.predicate.TupleDomain;
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 import static com.facebook.presto.metadata.MetadataUtil.TableMetadataBuilder.tableMetadataBuilder;
 import static com.facebook.presto.spi.type.VarcharType.createUnboundedVarcharType;
 import static java.util.Objects.requireNonNull;
@@ -56,7 +58,7 @@ public class CatalogJdbcTable
     public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
     {
         Builder table = InMemoryRecordSet.builder(METADATA);
-        for (String name : metadata.getCatalogNames().keySet()) {
+        for (String name : metadata.getCatalogNames(Optional.of(session.getIdentity())).keySet()) {
             table.addRow(name);
         }
         return table.build().cursor();
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/ColumnJdbcTable.java b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/ColumnJdbcTable.java
index 81595f3..cb11fee 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/ColumnJdbcTable.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/ColumnJdbcTable.java
@@ -115,7 +115,7 @@ public class ColumnJdbcTable
         Optional<String> tableFilter = stringFilter(constraint, 2);
 
         Builder table = InMemoryRecordSet.builder(METADATA);
-        for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
+        for (String catalog : filter(metadata.getCatalogNames(Optional.of(connectorSession.getIdentity())).keySet(), catalogFilter)) {
             QualifiedTablePrefix prefix = FilterUtil.tablePrefix(catalog, schemaFilter, tableFilter);
             for (Entry<QualifiedObjectName, List<ColumnMetadata>> entry : metadata.listTableColumns(session, prefix).entrySet()) {
                 addColumnRows(table, entry.getKey(), entry.getValue());
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/SchemaJdbcTable.java b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/SchemaJdbcTable.java
index 7cacded..95b26fe 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/SchemaJdbcTable.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/SchemaJdbcTable.java
@@ -68,7 +68,7 @@ public class SchemaJdbcTable
         Optional<String> catalogFilter = FilterUtil.stringFilter(constraint, 1);
 
         Builder table = InMemoryRecordSet.builder(METADATA);
-        for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
+        for (String catalog : filter(metadata.getCatalogNames(Optional.of(connectorSession.getIdentity())).keySet(), catalogFilter)) {
             for (String schema : metadata.listSchemaNames(session, catalog)) {
                 table.addRow(schema, catalog);
             }
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/TableJdbcTable.java b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/TableJdbcTable.java
index 692b989..88b68f8 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/TableJdbcTable.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/jdbc/TableJdbcTable.java
@@ -83,7 +83,7 @@ public class TableJdbcTable
         Optional<String> typeFilter = stringFilter(constraint, 3);
 
         Builder table = InMemoryRecordSet.builder(METADATA);
-        for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
+        for (String catalog : filter(metadata.getCatalogNames(Optional.of(connectorSession.getIdentity())).keySet(), catalogFilter)) {
             QualifiedTablePrefix prefix = tablePrefix(catalog, schemaFilter, tableFilter);
 
             if (FilterUtil.emptyOrEquals(typeFilter, "TABLE")) {
diff --git a/presto-main/src/main/java/com/facebook/presto/execution/RenameTableTask.java b/presto-main/src/main/java/com/facebook/presto/execution/RenameTableTask.java
index 4141fee..8e517c4 100644
--- a/presto-main/src/main/java/com/facebook/presto/execution/RenameTableTask.java
+++ b/presto-main/src/main/java/com/facebook/presto/execution/RenameTableTask.java
@@ -52,7 +52,7 @@ public class RenameTableTask
         }
 
         QualifiedObjectName target = createQualifiedObjectName(session, statement, statement.getTarget());
-        if (!metadata.getCatalogNames().containsKey(target.getCatalogName())) {
+        if (!metadata.getCatalogNames(Optional.of(session.getIdentity())).containsKey(target.getCatalogName())) {
             throw new SemanticException(MISSING_CATALOG, statement, "Target catalog '%s' does not exist", target.getCatalogName());
         }
         if (metadata.getTableHandle(session, target).isPresent()) {
diff --git a/presto-main/src/main/java/com/facebook/presto/metadata/Metadata.java b/presto-main/src/main/java/com/facebook/presto/metadata/Metadata.java
index 3760dd8..f442a7c 100644
--- a/presto-main/src/main/java/com/facebook/presto/metadata/Metadata.java
+++ b/presto-main/src/main/java/com/facebook/presto/metadata/Metadata.java
@@ -19,6 +19,7 @@ import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.Constraint;
 import com.facebook.presto.spi.block.BlockEncodingSerde;
 import com.facebook.presto.spi.predicate.TupleDomain;
+import com.facebook.presto.spi.security.Identity;
 import com.facebook.presto.spi.security.Privilege;
 import com.facebook.presto.spi.type.Type;
 import com.facebook.presto.spi.type.TypeManager;
@@ -198,7 +199,7 @@ public interface Metadata
      * @return Map of catalog name to connector id
      */
     @NotNull
-    Map<String, String> getCatalogNames();
+    Map<String, String> getCatalogNames(Optional<Identity> identity);
 
     /**
      * Get the names that match the specified table prefix (never null).
diff --git a/presto-main/src/main/java/com/facebook/presto/metadata/MetadataManager.java b/presto-main/src/main/java/com/facebook/presto/metadata/MetadataManager.java
index c700734..cd03bb9 100644
--- a/presto-main/src/main/java/com/facebook/presto/metadata/MetadataManager.java
+++ b/presto-main/src/main/java/com/facebook/presto/metadata/MetadataManager.java
@@ -15,6 +15,8 @@ package com.facebook.presto.metadata;
 
 import com.facebook.presto.Session;
 import com.facebook.presto.block.BlockEncodingManager;
+import com.facebook.presto.security.AccessControl;
+import com.facebook.presto.security.AllowAllAccessControl;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.ConnectorInsertTableHandle;
@@ -35,6 +37,8 @@ import com.facebook.presto.spi.connector.ConnectorMetadata;
 import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
 import com.facebook.presto.spi.predicate.NullableValue;
 import com.facebook.presto.spi.predicate.TupleDomain;
+import com.facebook.presto.spi.security.AccessDeniedException;
+import com.facebook.presto.spi.security.Identity;
 import com.facebook.presto.spi.security.Privilege;
 import com.facebook.presto.spi.type.Type;
 import com.facebook.presto.spi.type.TypeManager;
@@ -116,15 +120,17 @@ public class MetadataManager
     private final SessionPropertyManager sessionPropertyManager;
     private final TablePropertyManager tablePropertyManager;
     private final TransactionManager transactionManager;
+    private final AccessControl accessControl;
 
     public MetadataManager(FeaturesConfig featuresConfig,
             TypeManager typeManager,
             BlockEncodingSerde blockEncodingSerde,
             SessionPropertyManager sessionPropertyManager,
             TablePropertyManager tablePropertyManager,
-            TransactionManager transactionManager)
+            TransactionManager transactionManager,
+            AccessControl accessControl)
     {
-        this(featuresConfig, typeManager, createTestingViewCodec(), blockEncodingSerde, sessionPropertyManager, tablePropertyManager, transactionManager);
+        this(featuresConfig, typeManager, createTestingViewCodec(), blockEncodingSerde, sessionPropertyManager, tablePropertyManager, transactionManager, accessControl);
     }
 
     @Inject
@@ -134,7 +140,8 @@ public class MetadataManager
             BlockEncodingSerde blockEncodingSerde,
             SessionPropertyManager sessionPropertyManager,
             TablePropertyManager tablePropertyManager,
-            TransactionManager transactionManager)
+            TransactionManager transactionManager,
+            AccessControl accessControl)
     {
         functions = new FunctionRegistry(typeManager, blockEncodingSerde, featuresConfig);
         procedures = new ProcedureRegistry();
@@ -144,6 +151,7 @@ public class MetadataManager
         this.sessionPropertyManager = requireNonNull(sessionPropertyManager, "sessionPropertyManager is null");
         this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
         this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
+        this.accessControl = requireNonNull(accessControl, "accessControl is null");
 
         verifyComparableOrderableContract();
     }
@@ -155,7 +163,7 @@ public class MetadataManager
         SessionPropertyManager sessionPropertyManager = new SessionPropertyManager();
         BlockEncodingSerde blockEncodingSerde = new BlockEncodingManager(typeManager);
         TransactionManager transactionManager = createTestTransactionManager();
-        return new MetadataManager(featuresConfig, typeManager, blockEncodingSerde, sessionPropertyManager, new TablePropertyManager(), transactionManager);
+        return new MetadataManager(featuresConfig, typeManager, blockEncodingSerde, sessionPropertyManager, new TablePropertyManager(), transactionManager, new AllowAllAccessControl());
     }
 
     public synchronized void registerConnectorCatalog(String connectorId, String catalogName)
@@ -362,6 +370,8 @@ public class MetadataManager
     {
         requireNonNull(prefix, "prefix is null");
 
+        accessControl.checkCanAccessCatalog(session.getIdentity(), prefix.getCatalogName());
+
         String schemaNameOrNull = prefix.getSchemaName().orElse(null);
         Set<QualifiedObjectName> tables = new LinkedHashSet<>();
         for (ConnectorEntry entry : allConnectorsFor(prefix.getCatalogName())) {
@@ -378,6 +388,7 @@ public class MetadataManager
     public Optional<ColumnHandle> getSampleWeightColumnHandle(Session session, TableHandle tableHandle)
     {
         requireNonNull(tableHandle, "tableHandle is null");
+        accessControl.checkCanAccessCatalog(session.getIdentity(), tableHandle.getConnectorId());
         ConnectorEntry entry = lookupConnectorFor(tableHandle);
         ConnectorMetadata metadata = entry.getMetadata(session);
         ColumnHandle handle = metadata.getSampleWeightColumnHandle(session.toConnectorSession(entry.getCatalog()), tableHandle.getConnectorHandle());
@@ -589,11 +600,16 @@ public class MetadataManager
     }
 
     @Override
-    public Map<String, String> getCatalogNames()
+    public Map<String, String> getCatalogNames(Optional<Identity> identity)
     {
         ImmutableMap.Builder<String, String> catalogsMap = ImmutableMap.builder();
         for (Map.Entry<String, ConnectorEntry> entry : connectorsByCatalog.entrySet()) {
-            catalogsMap.put(entry.getKey(), entry.getValue().getConnectorId());
+            try {
+                identity.ifPresent(id -> accessControl.checkCanAccessCatalog(id, entry.getKey()));
+                catalogsMap.put(entry.getKey(), entry.getValue().getConnectorId());
+            }
+            catch (AccessDeniedException ignore) {
+            }
         }
         return catalogsMap.build();
     }
diff --git a/presto-main/src/main/java/com/facebook/presto/security/AccessControl.java b/presto-main/src/main/java/com/facebook/presto/security/AccessControl.java
index 12ff615..7b4a590 100644
--- a/presto-main/src/main/java/com/facebook/presto/security/AccessControl.java
+++ b/presto-main/src/main/java/com/facebook/presto/security/AccessControl.java
@@ -129,4 +129,10 @@ public interface AccessControl
      * @throws com.facebook.presto.spi.security.AccessDeniedException if not allowed
      */
     void checkCanSetCatalogSessionProperty(Identity identity, String catalogName, String propertyName);
+
+    /**
+     * Check if identity is allowed to set the specified catalog.
+     * @throws com.facebook.presto.spi.security.AccessDeniedException if not allowed
+     */
+    void checkCanAccessCatalog(Identity identity, String key);
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/security/AccessControlManager.java b/presto-main/src/main/java/com/facebook/presto/security/AccessControlManager.java
index 5ef1385..79c484d 100644
--- a/presto-main/src/main/java/com/facebook/presto/security/AccessControlManager.java
+++ b/presto-main/src/main/java/com/facebook/presto/security/AccessControlManager.java
@@ -362,6 +362,17 @@ public class AccessControlManager
         }
     }
 
+    @Override
+    public void checkCanAccessCatalog(Identity identity, String catalogName)
+    {
+        authorizationCheck(() -> systemAccessControl.get().checkCanAccessCatalog(identity, catalogName));
+
+        CatalogAccessControlEntry entry = catalogAccessControl.get(catalogName);
+        if (entry != null) {
+            authorizationCheck(() -> entry.getAccessControl().checkCanAccessCatalog(identity, catalogName));
+        }
+    }
+
     @Managed
     @Nested
     public CounterStat getAuthenticationSuccess()
diff --git a/presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java b/presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java
index a6069ae..842be0c 100644
--- a/presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java
+++ b/presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java
@@ -46,6 +46,7 @@ import org.weakref.jmx.guice.MBeanModule;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import static com.facebook.presto.server.PrestoSystemRequirements.verifyJvmRequirements;
@@ -150,7 +151,7 @@ public class PrestoServer
 
         // automatically build sources if not configured
         if (datasources.isEmpty()) {
-            Set<String> catalogs = metadata.getCatalogNames().keySet();
+            Set<String> catalogs = metadata.getCatalogNames(Optional.empty()).keySet();
             // if this is a dedicated coordinator, only add jmx
             if (serverConfig.isCoordinator() && !schedulerConfig.isIncludeCoordinator()) {
                 if (catalogs.contains("jmx")) {
diff --git a/presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java b/presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
index 072a67a..b28a0aa 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
@@ -534,7 +534,7 @@ class StatementAnalyzer
 
         Optional<TableHandle> tableHandle = metadata.getTableHandle(session, name);
         if (!tableHandle.isPresent()) {
-            if (!metadata.getCatalogNames().containsKey(name.getCatalogName())) {
+            if (!metadata.getCatalogNames(Optional.of(session.getIdentity())).containsKey(name.getCatalogName())) {
                 throw new SemanticException(MISSING_CATALOG, table, "Catalog %s does not exist", name.getCatalogName());
             }
             if (!metadata.listSchemaNames(session, name.getCatalogName()).contains(name.getSchemaName())) {
diff --git a/presto-main/src/main/java/com/facebook/presto/sql/rewrite/ShowQueriesRewrite.java b/presto-main/src/main/java/com/facebook/presto/sql/rewrite/ShowQueriesRewrite.java
index 86122cc..955034a 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/rewrite/ShowQueriesRewrite.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/rewrite/ShowQueriesRewrite.java
@@ -223,7 +223,7 @@ final class ShowQueriesRewrite
         @Override
         protected Node visitShowCatalogs(ShowCatalogs node, Void context)
         {
-            List<Expression> rows = metadata.getCatalogNames().keySet().stream()
+            List<Expression> rows = metadata.getCatalogNames(Optional.of(session.getIdentity())).keySet().stream()
                     .map(name -> row(new StringLiteral(name)))
                     .collect(toList());
 
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorAccessControl.java b/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorAccessControl.java
index 574b9e5..70a1a7c 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorAccessControl.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorAccessControl.java
@@ -130,4 +130,11 @@ public interface ConnectorAccessControl
      * @throws com.facebook.presto.spi.security.AccessDeniedException if not allowed
      */
     void checkCanRevokeTablePrivilege(Identity identity, Privilege privilege, SchemaTableName tableName);
+
+    /**
+     * Check if identity is allowed to revoke the specified privilege on the specified table from any user.
+     *
+     * @throws com.facebook.presto.spi.security.AccessDeniedException if not allowed
+     */
+    void checkCanAccessCatalog(Identity identity, String catalogName);
 }
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/security/SystemAccessControl.java b/presto-spi/src/main/java/com/facebook/presto/spi/security/SystemAccessControl.java
index e299ab0..abb01c8 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/security/SystemAccessControl.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/security/SystemAccessControl.java
@@ -30,4 +30,11 @@ public interface SystemAccessControl
      * @throws AccessDeniedException if not allowed
      */
     void checkCanSetSystemSessionProperty(Identity identity, String propertyName);
+
+    /**
+     * Check if identity is allowed to access the specified catalog.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    void checkCanAccessCatalog(Identity identity, String catalogName);
 }