Further fix for DENY permissions (#1834)

## Changes
- Legacy TACL migration logic currently grouped all permissions in a
single `GRANT` statement. Split this into 2 separate statements, one for
`DENY` and one for `GRANT`
- Added more unit tests, and added integration tests

### Tests
<!-- How is this tested? Please see the checklist below and also
describe any other relevant tests -->

- [x] manually tested
- [x] added unit tests
- [x] added integration tests
- [ ] verified on staging environment (screenshot attached)

Author: nkvuong

src/databricks/labs/ucx/hive_metastore/grants.py

@@ -120,11 +120,15 @@ def hive_grant_sql(self) -> list[str]:
         # See https://docs.databricks.com/en/sql/language-manual/security-grant.html
         statements = []
         actions = self.action_type.split(", ")
-        if "OWN" in actions:
-            actions.remove("OWN")
+        deny_actions = [action for action in actions if "DENIED" in action]
+        grant_actions = [action for action in actions if "DENIED" not in action]
+        if "OWN" in grant_actions:
+            grant_actions.remove("OWN")
             statements.append(self._set_owner_sql(object_type, object_key))
-        if actions:
-            statements.append(self._apply_grant_sql(", ".join(actions), object_type, object_key))
+        if grant_actions:
+            statements.append(self._apply_grant_sql(", ".join(grant_actions), object_type, object_key))
+        if deny_actions:
+            statements.append(self._apply_grant_sql(", ".join(deny_actions), object_type, object_key))
         return statements
 
     def hive_revoke_sql(self) -> str:

tests/integration/workspace_access/test_tacl.py

@@ -212,6 +212,7 @@ def test_permission_for_udfs(sql_backend, runtime_ctx, make_group_pair):
     sql_backend.execute(f"GRANT SELECT ON FUNCTION {udf_a.full_name} TO `{group.name_in_workspace}`")
     sql_backend.execute(f"ALTER FUNCTION {udf_a.full_name} OWNER TO `{group.name_in_workspace}`")
     sql_backend.execute(f"GRANT READ_METADATA ON FUNCTION {udf_b.full_name} TO `{group.name_in_workspace}`")
+    sql_backend.execute(f"DENY `SELECT` ON FUNCTION {udf_b.full_name} TO `{group.name_in_workspace}`")
 
     grants = runtime_ctx.grants_crawler
 
@@ -222,6 +223,7 @@ def test_permission_for_udfs(sql_backend, runtime_ctx, make_group_pair):
     assert f"{group.name_in_workspace}.{udf_a.full_name}:SELECT" in all_initial_grants
     assert f"{group.name_in_workspace}.{udf_a.full_name}:OWN" in all_initial_grants
     assert f"{group.name_in_workspace}.{udf_b.full_name}:READ_METADATA" in all_initial_grants
+    assert f"{group.name_in_workspace}.{udf_b.full_name}:DENIED_SELECT" in all_initial_grants
 
     tacl_support = TableAclSupport(grants, sql_backend)
     apply_tasks(tacl_support, [group])
@@ -234,7 +236,7 @@ def test_permission_for_udfs(sql_backend, runtime_ctx, make_group_pair):
     actual_udf_b_grants = defaultdict(set)
     for grant in grants.grants(catalog=schema.catalog_name, database=schema.name, udf=udf_b.name):
         actual_udf_b_grants[grant.principal].add(grant.action_type)
-    assert {"READ_METADATA"} == actual_udf_b_grants[group.name_in_account]
+    assert {"READ_METADATA", "DENIED_SELECT"} == actual_udf_b_grants[group.name_in_account]
 
 
 def test_verify_permission_for_udfs(sql_backend, runtime_ctx, make_group):

tests/unit/workspace_access/test_tacl.py

@@ -391,7 +391,7 @@ def test_tacl_udf_applier(mocker):
     assert validation_res
 
 
-def test_tacl_applier_multiple_actions(mocker):
+def test_tacl_applier_multiple_actions():
     sql_backend = MockBackend(
         rows={
             "SELECT \\* FROM hive_metastore.test.grants": UCX_GRANTS[
@@ -444,6 +444,61 @@ def test_tacl_applier_multiple_actions(mocker):
     assert validation_res
 
 
+def test_tacl_applier_deny_and_grant():
+    sql_backend = MockBackend(
+        rows={
+            "SELECT \\* FROM hive_metastore.test.grants": UCX_GRANTS[
+                ("abc", "SELECT", "catalog_a", "database_b", "table_c", None, None, False, False)
+            ],
+            "SHOW GRANTS ON TABLE catalog_a.database_b.table_c": SHOW_GRANTS[
+                ("account-abc", "READ_METADATA", "TABLE", "table_c"),
+                ("account-abc", "SELECT", "TABLE", "table_c"),
+                ("account-abc", "DENIED_MODIFY", "TABLE", "table_c"),
+            ],
+        }
+    )
+    tables_crawler = TablesCrawler(sql_backend, "test")
+    udf_crawler = UdfsCrawler(sql_backend, "test")
+    grants_crawler = GrantsCrawler(tables_crawler, udf_crawler)
+    table_acl_support = TableAclSupport(grants_crawler, sql_backend)
+
+    permissions = Permissions(
+        object_type="TABLE",
+        object_id="catalog_a.database_b.table_c",
+        raw=json.dumps(
+            {
+                "principal": "abc",
+                "action_type": "SELECT, READ_METADATA, DENIED_MODIFY",
+                "catalog": "catalog_a",
+                "database": "database_b",
+                "table": "table_c",
+            }
+        ),
+    )
+    grp = [
+        MigratedGroup(
+            id_in_workspace=None,
+            name_in_workspace="abc",
+            name_in_account="account-abc",
+            temporary_name="tmp-backup-abc",
+            members=None,
+            entitlements=None,
+            external_id=None,
+            roles=None,
+        )
+    ]
+    migration_state = MigrationState(grp)
+    task = table_acl_support.get_apply_task(permissions, migration_state)
+    validation_res = task()
+
+    assert [
+        "GRANT SELECT, READ_METADATA ON TABLE catalog_a.database_b.table_c TO `account-abc`",
+        "DENY `MODIFY` ON TABLE catalog_a.database_b.table_c TO `account-abc`",
+        "SHOW GRANTS ON TABLE catalog_a.database_b.table_c",
+    ] == sql_backend.queries
+    assert validation_res
+
+
 def test_tacl_applier_no_target_principal(mocker):
     sql_backend = MockBackend()
     table_acl_support = TableAclSupport(mocker.Mock(), sql_backend)