Add read-only access level to file-based system access control

* adds a `read-only` option to the file-based system access control
  mechanism
* three roles exist for `allow`: `all`, `read-only`, `none`
* legacy `allow` values of `true | false` map to `all` and `none`
* `allow` can be set to `all | read-only | none`

presto-docs/src/main/sphinx/security/built-in-system-access-control.rst

@@ -93,16 +93,23 @@ following fields:
 
 * ``user`` (optional): regex to match against user name. Defaults to ``.*``.
 * ``catalog`` (optional): regex to match against catalog name. Defaults to ``.*``.
-* ``allow`` (required): boolean indicating whether a user has access to the catalog
+* ``allow`` (required): string indicating whether a user has access to the catalog.
+  This value can be ``all``, ``read-only`` or ``none``, and defaults to ``none``.
+  Setting this value to ``read-only`` has the same behavior as the ``read-only``
+  system access control plugin.
 
 .. note::
 
     By default, all users have access to the ``system`` catalog. You can
     override this behavior by adding a rule.
 
+    Boolean ``true`` and ``false`` are also supported as legacy values for ``allow``,
+    to support backwards compatibility.  ``true`` maps to ``all``, and ``false`` maps to ``none``.
+
 For example, if you want to allow only the user ``admin`` to access the
 ``mysql`` and the ``system`` catalog, allow all users to access the ``hive``
-catalog, and deny all other access, you can use the following rules:
+catalog, allow the user ``alice`` read-only access to the ``postgresql`` catalog,
+and deny all other access, you can use the following rules:
 
 .. code-block:: json
 
@@ -111,15 +118,20 @@ catalog, and deny all other access, you can use the following rules:
         {
           "user": "admin",
           "catalog": "(mysql|system)",
-          "allow": true
+          "allow": "all"
         },
         {
           "catalog": "hive",
-          "allow": true
+          "allow": "all"
+        },
+        {
+          "user": "alice",
+          "catalog": "postgresql",
+          "allow": "read-only"
         },
         {
           "catalog": "system",
-          "allow": false
+          "allow": "none"
         }
       ]
     }

presto-main/src/test/java/io/prestosql/security/TestFileBasedSystemAccessControl.java

@@ -58,7 +58,7 @@ public class TestFileBasedSystemAccessControl
     private static final Identity bob = new Identity("bob", Optional.empty());
     private static final Identity admin = new Identity("admin", Optional.empty());
     private static final Identity nonAsciiUser = new Identity("\u0194\u0194\u0194", Optional.empty());
-    private static final Set<String> allCatalogs = ImmutableSet.of("secret", "open-to-all", "all-allowed", "alice-catalog", "allowed-absent", "\u0200\u0200\u0200");
+    private static final Set<String> allCatalogs = ImmutableSet.of("secret", "open-to-all", "all-allowed", "alice-catalog", "\u0200\u0200\u0200");
     private static final QualifiedObjectName aliceTable = new QualifiedObjectName("alice-catalog", "schema", "table");
     private static final QualifiedObjectName aliceView = new QualifiedObjectName("alice-catalog", "schema", "view");
     private static final CatalogSchemaName aliceSchema = new CatalogSchemaName("alice-catalog", "schema");
@@ -124,6 +124,24 @@ public void testCatalogOperations()
                 });
     }
 
+    @Test
+    public void testCatalogOperationsReadOnly()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "catalog_read_only.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    assertEquals(accessControlManager.filterCatalogs(admin, allCatalogs), allCatalogs);
+                    Set<String> aliceCatalogs = ImmutableSet.of("open-to-all", "alice-catalog", "all-allowed");
+                    assertEquals(accessControlManager.filterCatalogs(alice, allCatalogs), aliceCatalogs);
+                    Set<String> bobCatalogs = ImmutableSet.of("open-to-all", "all-allowed");
+                    assertEquals(accessControlManager.filterCatalogs(bob, allCatalogs), bobCatalogs);
+                    Set<String> nonAsciiUserCatalogs = ImmutableSet.of("open-to-all", "all-allowed", "\u0200\u0200\u0200");
+                    assertEquals(accessControlManager.filterCatalogs(nonAsciiUser, allCatalogs), nonAsciiUserCatalogs);
+                });
+    }
+
     @Test
     public void testSchemaOperations()
     {
@@ -171,6 +189,50 @@ public void testTableOperations()
         }));
     }
 
+    @Test
+    public void testTableOperationsReadOnly()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "catalog_read_only.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    Set<SchemaTableName> aliceTables = ImmutableSet.of(new SchemaTableName("schema", "table"));
+                    assertEquals(accessControlManager.filterTables(new SecurityContext(transactionId, alice), "alice-catalog", aliceTables), aliceTables);
+                    assertEquals(accessControlManager.filterTables(new SecurityContext(transactionId, bob), "alice-catalog", aliceTables), ImmutableSet.of());
+
+                    accessControlManager.checkCanSelectFromColumns(new SecurityContext(transactionId, alice), aliceTable, ImmutableSet.of());
+                });
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateTable(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropTable(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanInsertIntoTable(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDeleteFromTable(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanAddColumns(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameColumn(new SecurityContext(transactionId, alice), aliceTable);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateTable(new SecurityContext(transactionId, bob), aliceTable);
+        }));
+    }
+
     @Test
     public void testViewOperations()
     {
@@ -193,6 +255,53 @@ public void testViewOperations()
         }));
     }
 
+    @Test
+    public void testEverythingImplemented()
+    {
+        assertAllMethodsOverridden(SystemAccessControl.class, FileBasedSystemAccessControl.class);
+    }
+
+    @Test
+    public void testViewOperationsReadOnly()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "catalog_read_only.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    accessControlManager.checkCanSelectFromColumns(new SecurityContext(transactionId, alice), aliceView, ImmutableSet.of());
+                    accessControlManager.checkCanSetCatalogSessionProperty(transactionId, alice, "alice-catalog", "property");
+                });
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateView(new SecurityContext(transactionId, alice), aliceView);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropView(new SecurityContext(transactionId, alice), aliceView);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateViewWithSelectFromColumns(new SecurityContext(transactionId, alice), aliceTable, ImmutableSet.of());
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateViewWithSelectFromColumns(new SecurityContext(transactionId, alice), aliceView, ImmutableSet.of());
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanGrantTablePrivilege(new SecurityContext(transactionId, alice), SELECT, aliceTable, new PrestoPrincipal(USER, "grantee"), true);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRevokeTablePrivilege(new SecurityContext(transactionId, alice), SELECT, aliceTable, new PrestoPrincipal(USER, "revokee"), true);
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateView(new SecurityContext(transactionId, bob), aliceView);
+        }));
+    }
+
     @Test
     public void testRefreshing()
             throws Exception
@@ -240,6 +349,18 @@ public void testRefreshing()
                 });
     }
 
+    @Test
+    public void testAllowModeIsRequired()
+    {
+        assertThrows(IllegalArgumentException.class, () -> newAccessControlManager(createTestTransactionManager(), "catalog_allow_unset.json"));
+    }
+
+    @Test
+    public void testAllowModeInvalidValue()
+    {
+        assertThrows(IllegalArgumentException.class, () -> newAccessControlManager(createTestTransactionManager(), "catalog_invalid_allow_value.json"));
+    }
+
     private AccessControlManager newAccessControlManager(TransactionManager transactionManager, String resourceName)
     {
         AccessControlManager accessControlManager = new AccessControlManager(transactionManager);

presto-main/src/test/resources/catalog.json

@@ -28,9 +28,6 @@
       "catalog": "alice-catalog",
       "allow": false
     },
-    {
-      "catalog": "allowed-absent"
-    },
     {
       "user": "\u0194\u0194\u0194",
       "catalog": "\u0200\u0200\u0200",

presto-main/src/test/resources/catalog_allow_unset.json

@@ -0,0 +1,7 @@
+{
+  "catalogs": [
+    {
+      "catalog": "allowed-absent"
+    }
+  ]
+}

presto-main/src/test/resources/catalog_invalid_allow_value.json

@@ -0,0 +1,8 @@
+{
+  "catalogs": [
+    {
+      "catalog": "invalid-allow-value",
+      "allow": "not-a-valid-value"
+    }
+  ]
+}

presto-main/src/test/resources/catalog_read_only.json

@@ -0,0 +1,37 @@
+{
+  "catalogs": [
+    {
+      "user": "admin",
+      "catalog": ".*",
+      "allow": "read-only"
+    },
+    {
+      "catalog": "secret",
+      "allow": "none"
+    },
+    {
+      "user": ".*",
+      "catalog": "open-to-all",
+      "allow": "read-only"
+    },
+    {
+      "catalog": "all-allowed",
+      "allow": "read-only"
+    },
+    {
+      "user": "alice",
+      "catalog": "alice-catalog",
+      "allow": "read-only"
+    },
+    {
+      "user": "bob",
+      "catalog": "alice-catalog",
+      "allow": "none"
+    },
+    {
+      "user": "\u0194\u0194\u0194",
+      "catalog": "\u0200\u0200\u0200",
+      "allow": "read-only"
+    }
+  ]
+}

presto-plugin-toolkit/src/main/java/io/prestosql/plugin/base/security/CatalogAccessControlRule.java

@@ -15,35 +15,90 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
 
+import java.util.Locale;
+import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 
 import static java.util.Objects.requireNonNull;
 
 public class CatalogAccessControlRule
 {
-    private final boolean allow;
+    private final AccessMode accessMode;
     private final Optional<Pattern> userRegex;
     private final Optional<Pattern> catalogRegex;
 
     @JsonCreator
     public CatalogAccessControlRule(
-            @JsonProperty("allow") boolean allow,
+            @JsonProperty("allow") AccessMode accessMode,
             @JsonProperty("user") Optional<Pattern> userRegex,
             @JsonProperty("catalog") Optional<Pattern> catalogRegex)
     {
-        this.allow = allow;
+        this.accessMode = requireNonNull(accessMode, "accessMode is null");
         this.userRegex = requireNonNull(userRegex, "userRegex is null");
         this.catalogRegex = requireNonNull(catalogRegex, "catalogRegex is null");
     }
 
-    public Optional<Boolean> match(String user, String catalog)
+    public Optional<AccessMode> match(String user, String catalog)
     {
         if (userRegex.map(regex -> regex.matcher(user).matches()).orElse(true) &&
                 catalogRegex.map(regex -> regex.matcher(catalog).matches()).orElse(true)) {
-            return Optional.of(allow);
+            return Optional.of(accessMode);
         }
         return Optional.empty();
     }
+
+    public enum AccessMode
+    {
+        ALL("all"),
+        READ_ONLY("read-only"),
+        NONE("none");
+
+        private static final Map<String, AccessMode> modeByName = Maps.uniqueIndex(ImmutableList.copyOf(AccessMode.values()), AccessMode::toString);
+
+        private final String stringValue;
+
+        AccessMode(String stringValue)
+        {
+            this.stringValue = requireNonNull(stringValue, "stringValue is null");
+        }
+
+        @JsonValue
+        @Override
+        public String toString()
+        {
+            return stringValue;
+        }
+
+        @JsonCreator
+        public static AccessMode fromJson(Object value)
+        {
+            if (Boolean.TRUE.equals(value)) {
+                return ALL;
+            }
+            if (Boolean.FALSE.equals(value)) {
+                return NONE;
+            }
+            if (value instanceof String) {
+                AccessMode accessMode = modeByName.get(((String) value).toLowerCase(Locale.US));
+                if (accessMode != null) {
+                    return accessMode;
+                }
+            }
+
+            throw new IllegalArgumentException("Unknown " + AccessMode.class.getSimpleName() + ": " + value);
+        }
+
+        boolean implies(AccessMode other)
+        {
+            if (this == ALL && other == READ_ONLY) {
+                return true;
+            }
+            return this == other;
+        }
+    }
 }