Adding schema access rules to file based system access control

Cherry pick of trinodb/trino#3766

Co-authored-by: haldes <[email protected]>

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

@@ -64,10 +64,11 @@ contents:
 The config file is specified in JSON format.
 
 * It contains the rules defining which catalog can be accessed by which user (see Catalog Rules below).
+* The schema access rules defining which schema can be accessed by which user (see Schema Rules below).
 * The principal rules specifying what principals can identify as what users (see Principal Rules below).
 
-This plugin currently only supports catalog access control rules and principal
-rules. If you want to limit access on a system level in any other way, you
+This plugin currently supports catalog access control rules, schema access control rules
+and principal rules. If you want to limit access on a system level in any other way, you
 must implement a custom SystemAccessControl plugin
 (see :doc:`/develop/system-access-control`).
 
@@ -136,6 +137,48 @@ and deny all other access, you can use the following rules:
       ]
     }
 
+Schema Rules
+------------
+
+These rules allow you to grant ownership of a schema. Having ownership of an
+schema allows users to execute ``DROP SCHEMA``, ``ALTER SCHEMA`` and
+``CREATE SCHEMA``. The user is granted ownership of a schema, based on
+the first matching rule read from top to bottom. If no rule matches, ownership
+is not granted. Each rule is composed of the following fields:
+
+* ``user`` (optional): regex to match against user name. Defaults to ``.*``.
+* ``schema`` (optional): regex to match against schema name. Defaults to ``.*``.
+* ``owner`` (required): boolean indicating whether the user is to be considered
+  an owner of the schema. Defaults to ``false``.
+
+For example, to provide ownership of all schemas to user ``admin``, treat all
+users as owners of ``default`` schema and prevent user ``guest`` from ownership
+of any schema, you can use the following rules:
+
+.. code-block:: json
+    {
+      "catalogs": [
+        {
+          "allow": true
+        }
+      ],
+      "schemas": [
+        {
+          "user": "admin",
+          "schema": ".*",
+          "owner": true
+        },
+        {
+          "user": "guest",
+          "owner": false
+        },
+        {
+          "schema": "default",
+          "owner": true
+        }
+      ]
+    }
+
 Principal Rules
 ---------------
 

presto-main/src/main/java/com/facebook/presto/security/FileBasedSystemAccessControl.java

@@ -16,6 +16,7 @@
 import com.facebook.airlift.log.Logger;
 import com.facebook.presto.common.CatalogSchemaName;
 import com.facebook.presto.plugin.base.security.ForwardingSystemAccessControl;
+import com.facebook.presto.plugin.base.security.SchemaAccessControlRule;
 import com.facebook.presto.security.CatalogAccessControlRule.AccessMode;
 import com.facebook.presto.spi.CatalogSchemaTableName;
 import com.facebook.presto.spi.PrestoException;
@@ -77,11 +78,13 @@ public class FileBasedSystemAccessControl
 
     private final List<CatalogAccessControlRule> catalogRules;
     private final Optional<List<PrincipalUserMatchRule>> principalUserMatchRules;
+    private final Optional<List<SchemaAccessControlRule>> schemaRules;
 
-    private FileBasedSystemAccessControl(List<CatalogAccessControlRule> catalogRules, Optional<List<PrincipalUserMatchRule>> principalUserMatchRules)
+    private FileBasedSystemAccessControl(List<CatalogAccessControlRule> catalogRules, Optional<List<PrincipalUserMatchRule>> principalUserMatchRules, Optional<List<SchemaAccessControlRule>> schemaRules)
     {
         this.catalogRules = catalogRules;
         this.principalUserMatchRules = principalUserMatchRules;
+        this.schemaRules = schemaRules;
     }
 
     public static class Factory
@@ -144,7 +147,7 @@ private SystemAccessControl create(String configFileName)
                     Optional.of(Pattern.compile(".*")),
                     Optional.of(Pattern.compile("system"))));
 
-            return new FileBasedSystemAccessControl(catalogRulesBuilder.build(), rules.getPrincipalUserMatchRules());
+            return new FileBasedSystemAccessControl(catalogRulesBuilder.build(), rules.getPrincipalUserMatchRules(), rules.getSchemaRules());
         }
     }
 
@@ -221,23 +224,23 @@ private boolean canAccessCatalog(Identity identity, String catalogName, AccessMo
     @Override
     public void checkCanCreateSchema(Identity identity, AccessControlContext context, CatalogSchemaName schema)
     {
-        if (!canAccessCatalog(identity, schema.getCatalogName(), ALL)) {
+        if (!isSchemaOwner(identity, schema)) {
             denyCreateSchema(schema.toString());
         }
     }
 
     @Override
     public void checkCanDropSchema(Identity identity, AccessControlContext context, CatalogSchemaName schema)
     {
-        if (!canAccessCatalog(identity, schema.getCatalogName(), ALL)) {
+        if (!isSchemaOwner(identity, schema)) {
             denyDropSchema(schema.toString());
         }
     }
 
     @Override
     public void checkCanRenameSchema(Identity identity, AccessControlContext context, CatalogSchemaName schema, String newSchemaName)
     {
-        if (!canAccessCatalog(identity, schema.getCatalogName(), ALL)) {
+        if (!isSchemaOwner(identity, schema) || !isSchemaOwner(identity, new CatalogSchemaName(schema.getCatalogName(), newSchemaName))) {
             denyRenameSchema(schema.toString(), newSchemaName);
         }
     }
@@ -385,4 +388,23 @@ public void checkCanRevokeTablePrivilege(Identity identity, AccessControlContext
             denyRevokeTablePrivilege(privilege.toString(), table.toString());
         }
     }
+
+    private boolean isSchemaOwner(Identity identity, CatalogSchemaName schema)
+    {
+        if (!canAccessCatalog(identity, schema.getCatalogName(), ALL)) {
+            return false;
+        }
+
+        if (!schemaRules.isPresent()) {
+            return true;
+        }
+
+        for (SchemaAccessControlRule rule : schemaRules.get()) {
+            Optional<Boolean> owner = rule.match(identity.getUser(), schema.getSchemaName());
+            if (owner.isPresent()) {
+                return owner.get();
+            }
+        }
+        return false;
+    }
 }

presto-main/src/main/java/com/facebook/presto/security/FileBasedSystemAccessControlRules.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.security;
 
+import com.facebook.presto.plugin.base.security.SchemaAccessControlRule;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
@@ -24,14 +25,17 @@ public class FileBasedSystemAccessControlRules
 {
     private final List<CatalogAccessControlRule> catalogRules;
     private final Optional<List<PrincipalUserMatchRule>> principalUserMatchRules;
+    private final Optional<List<SchemaAccessControlRule>> schemaRules;
 
     @JsonCreator
     public FileBasedSystemAccessControlRules(
             @JsonProperty("catalogs") Optional<List<CatalogAccessControlRule>> catalogRules,
-            @JsonProperty("principals") Optional<List<PrincipalUserMatchRule>> principalUserMatchRules)
+            @JsonProperty("principals") Optional<List<PrincipalUserMatchRule>> principalUserMatchRules,
+            @JsonProperty("schemas") Optional<List<SchemaAccessControlRule>> schemaRules)
     {
         this.catalogRules = catalogRules.map(ImmutableList::copyOf).orElse(ImmutableList.of());
         this.principalUserMatchRules = principalUserMatchRules.map(ImmutableList::copyOf);
+        this.schemaRules = schemaRules.map(ImmutableList::copyOf);
     }
 
     public List<CatalogAccessControlRule> getCatalogRules()
@@ -43,4 +47,9 @@ public Optional<List<PrincipalUserMatchRule>> getPrincipalUserMatchRules()
     {
         return principalUserMatchRules;
     }
+
+    public Optional<List<SchemaAccessControlRule>> getSchemaRules()
+    {
+        return schemaRules;
+    }
 }

presto-main/src/test/java/com/facebook/presto/security/TestFileBasedSystemAccessControl.java

@@ -169,6 +169,125 @@ public void testSchemaOperations()
         }));
     }
 
+    @Test
+    public void testSchemaRulesForCheckCanCreateSchema()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "file-based-system-access-schema.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    accessControlManager.checkCanCreateSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "bob"));
+                    accessControlManager.checkCanCreateSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "bob"));
+                    accessControlManager.checkCanCreateSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "some-schema"));
+                    accessControlManager.checkCanCreateSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "bob"));
+                    accessControlManager.checkCanCreateSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "alice"));
+                });
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "alice"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "alice"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, bob, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, alice, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, admin, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanCreateSchema(transactionId, alice, context, new CatalogSchemaName("alice-catalog", "alice"));
+        }));
+    }
+
+    @Test
+    public void testSchemaRulesForCheckCanDropSchema()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "file-based-system-access-schema.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    accessControlManager.checkCanDropSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "bob"));
+                    accessControlManager.checkCanDropSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "bob"));
+                    accessControlManager.checkCanDropSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "bob"));
+                    accessControlManager.checkCanDropSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "alice"));
+                    accessControlManager.checkCanDropSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "some-schema"));
+                });
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "alice"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "alice"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, bob, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, alice, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, admin, context, new CatalogSchemaName("secret-catalog", "secret"));
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanDropSchema(transactionId, alice, context, new CatalogSchemaName("alice-catalog", "alice"));
+        }));
+    }
+
+    @Test
+    public void testSchemaRulesForCheckCanRenameSchema()
+    {
+        TransactionManager transactionManager = createTestTransactionManager();
+        AccessControlManager accessControlManager = newAccessControlManager(transactionManager, "file-based-system-access-schema.json");
+
+        transaction(transactionManager, accessControlManager)
+                .execute(transactionId -> {
+                    accessControlManager.checkCanRenameSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "bob"), "some-schema");
+                    accessControlManager.checkCanRenameSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "bob"), "some-schema");
+                    accessControlManager.checkCanRenameSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "bob"), "new-schema-name");
+                    accessControlManager.checkCanRenameSchema(transactionId, admin, context, new CatalogSchemaName("some-catalog", "alice"), "new-schema-name");
+                });
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, bob, context, new CatalogSchemaName("alice-catalog", "alice"), "new-schema-name");
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, bob, context, new CatalogSchemaName("bob-catalog", "alice"), "new-schema-name");
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, bob, context, new CatalogSchemaName("secret-catalog", "secret"), "new-schema-name");
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, alice, context, new CatalogSchemaName("secret-catalog", "secret"), "new-schema-name");
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, admin, context, new CatalogSchemaName("secret-catalog", "secret"), "new-schema-name");
+        }));
+
+        assertThrows(AccessDeniedException.class, () -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
+            accessControlManager.checkCanRenameSchema(transactionId, alice, context, new CatalogSchemaName("alice-catalog", "alice"), "new-schema-name");
+        }));
+    }
+
     @Test
     public void testSchemaOperationsReadOnly()
     {

presto-main/src/test/resources/file-based-system-access-schema.json

@@ -0,0 +1,33 @@
+{
+  "catalogs": [
+    {
+      "user": "alice",
+      "catalog": "alice-catalog",
+      "allow": "read-only"
+    },
+    {
+      "allow": true
+    }
+  ],
+  "schemas": [
+    {
+      "schema": "secret",
+      "owner": false
+    },
+    {
+      "user": "admin",
+      "schema": ".*",
+      "owner": true
+    },
+    {
+      "user": "bob",
+      "schema": "bob|some-schema",
+      "owner": true
+    },
+    {
+      "user": "alice",
+      "schema": "alice",
+      "owner": true
+    }
+  ]
+}

presto-main/src/test/resources/security-config-file-with-unknown-rules.json

@@ -1,9 +1,9 @@
 {
-  "schemas": [
+  "sessionProperties": [
     {
-      "user": "(alice|bob)",
-      "schema": "(default|pv)",
-      "owner": true
+      "user": "admin",
+      "property": "max_split_size",
+      "allow": true
     }
   ]
 }