Capturing any file and anonymous function grants

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

@@ -140,7 +140,8 @@ def _try_load(self):
 
     def _crawl(self) -> list[Grant]:
         """
-        Crawls and lists grants for all databases, tables, and views within hive_metastore.
+        Crawls and lists grants for all databases, tables,  views, any file
+        and anonymous function within hive_metastore.
 
         Returns:
             list[Grant]: A list of Grant objects representing the listed grants.
@@ -155,13 +156,16 @@ def _crawl(self) -> list[Grant]:
 
         Note:
         - The method assumes that the `_grants` method fetches grants based on the provided parameters (catalog,
-          database, table, view).
+          database, table, view, any file, anonymous function).
 
         Returns:
         list[Grant]: A list of Grant objects representing the grants found in hive_metastore.
         """
         catalog = "hive_metastore"
         tasks = [partial(self._grants, catalog=catalog)]
+        # Scanning ANY FILE and ANONYMOUS FUNCTION grants
+        tasks.append(partial(self._grants, catalog=catalog, any_file=True))
+        tasks.append(partial(self._grants, catalog=catalog, anonymous_function=True))
         # scan all databases, even empty ones
         for row in self._fetch(f"SHOW DATABASES FROM {catalog}"):
             tasks.append(partial(self._grants, catalog=catalog, database=row.databaseName))
@@ -239,7 +243,14 @@ def _grants(
         )
         try:
             grants = []
-            object_type_normalization = {"SCHEMA": "DATABASE", "CATALOG$": "CATALOG"}
+            # Added ANY FILE and ANONYMOUS FUNCTION in object_type_normalization
+            # to capture the same in grants. issue:#623
+            object_type_normalization = {
+                "SCHEMA": "DATABASE",
+                "CATALOG$": "CATALOG",
+                "ANY_FILE": "ANY FILE",
+                "ANONYMOUS_FUNCTION": "ANONYMOUS FUNCTION",
+            }
             for row in self._fetch(f"SHOW GRANTS ON {on_type} {key}"):
                 (principal, action_type, object_type, _) = row
                 if object_type in object_type_normalization:

tests/integration/workspace_access/test_tacl.py

@@ -9,6 +9,50 @@
 logger = logging.getLogger(__name__)
 
 
+def test_permission_for_files_anonymous_func(sql_backend, inventory_schema, make_group):
+    group_a = make_group()
+    group_b = make_group()
+    group_c = make_group()
+    group_d = make_group()
+
+    sql_backend.execute(f"GRANT READ_METADATA ON ANY FILE TO `{group_a.display_name}`")
+    sql_backend.execute(f"GRANT SELECT ON ANONYMOUS FUNCTION TO `{group_c.display_name}`")
+
+    tables = StaticTablesCrawler(sql_backend, inventory_schema, [])
+    grants = GrantsCrawler(tables)
+
+    tacl_support = TableAclSupport(grants, sql_backend)
+
+    migration_state = MigrationState(
+        [
+            MigratedGroup.partial_info(group_a, group_b),
+            MigratedGroup.partial_info(group_c, group_d),
+        ]
+    )
+    for crawler_task in tacl_support.get_crawler_tasks():
+        permission = crawler_task()
+        apply_task = tacl_support.get_apply_task(permission, migration_state)
+        if not apply_task:
+            continue
+        apply_task()
+
+    any_file_actual = {}
+    for any_file_grant in grants._grants(any_file=True):
+        any_file_actual[any_file_grant.principal] = any_file_grant.action_type
+
+    assert group_b.display_name in any_file_actual
+    assert any_file_actual[group_b.display_name] == "READ_METADATA"
+    assert any_file_actual[group_a.display_name] == any_file_actual[group_b.display_name]
+
+    anonymous_function_actual = {}
+    for ano_func_grant in grants._grants(anonymous_function=True):
+        anonymous_function_actual[ano_func_grant.principal] = ano_func_grant.action_type
+
+    assert group_d.display_name in anonymous_function_actual
+    assert anonymous_function_actual[group_d.display_name] == "SELECT"
+    assert anonymous_function_actual[group_c.display_name] == anonymous_function_actual[group_d.display_name]
+
+
 def test_owner_permissions_for_tables_and_schemas(sql_backend, inventory_schema, make_schema, make_table, make_group):
     group_a = make_group()
     group_b = make_group()