feat: get implicit permissions filter policies domain by matching function (#226)

Signed-off-by: timabilov <timabilov33@gmail.com>

Co-authored-by: timabilov <timabilov33@gmail.com>

Author: timabilov

casbin/enforcer.py

@@ -1,3 +1,5 @@
+from functools import partial
+
 from casbin.management_enforcer import ManagementEnforcer
 from casbin.util import join_slice, array_remove_duplicates, set_subtract
 
@@ -150,20 +152,26 @@ def get_implicit_permissions_for_user(
         get_permissions_for_user("alice") can only get: [["alice", "data2", "read"]].
         But get_implicit_permissions_for_user("alice") will get: [["admin", "data1", "read"], ["alice", "data2", "read"]].
 
-        Inherited roles can be matched by domain.
+        For given domain policies are filtered by corresponding domain matching function of DomainManager
+        Inherited roles can be matched by domain. For domain neutral policies set:
+         filter_policy_dom = False
+
         filter_policy_dom: bool - For given *domain*, policies will be filtered by domain as well. Default = True
         """
         roles = self.get_implicit_roles_for_user(user, domain)
 
         roles.insert(0, user)
 
         res = []
-        for role in roles:
-            if domain and filter_policy_dom:
-                permissions = self.get_permissions_for_user_in_domain(role, domain)
-            else:
-                permissions = self.get_permissions_for_user(role)
 
+        # policy domain should be matched by domain_match_fn of DomainManager
+        if domain:
+            domain = partial(self.get_role_manager().domain_matching_func, domain)
+
+        for role in roles:
+            permissions = self.get_permissions_for_user_in_domain(
+                role, domain if filter_policy_dom else ""
+            )
             res.extend(permissions)
 
         return res

examples/rbac_with_domain_and_policy_pattern_policy.csv

@@ -1,4 +1,5 @@
 p, admin, domain.*, data1, read
+p, user, domain.*, data3, read
 p, user, domain.1, data2, read
 p, user, domain.1, data2, write
 

tests/test_management_api.py

@@ -111,6 +111,7 @@ def test_get_policy_matching_function(self):
             e.get_policy(),
             [
                 ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.*", "data3", "read"],
                 ["user", "domain.1", "data2", "read"],
                 ["user", "domain.1", "data2", "write"],
             ],
@@ -127,21 +128,26 @@ def test_get_policy_matching_function(self):
             [["alice", "user", "*"]],
         )
 
-        # first p record matches to domain.3
+        # first and second p record matches to domain.3
         self.assertEqual(
             e.get_filtered_policy(1, partial(km2_fn, "domain.3")),
-            [["admin", "domain.*", "data1", "read"]],
-        )
-
-        # first and second p record should be matched to (.., domain.1, read)
-        self.assertEqual(
-            e.get_filtered_policy(1, partial(km2_fn, "domain.1"), "", "read"),
             [
                 ["admin", "domain.*", "data1", "read"],
-                ["user", "domain.1", "data2", "read"],
+                ["user", "domain.*", "data3", "read"],
             ],
         )
 
+        self.assertEqual(
+            sorted(e.get_filtered_policy(1, partial(km2_fn, "domain.1"), "", "read")),
+            sorted(
+                [
+                    ["admin", "domain.*", "data1", "read"],
+                    ["user", "domain.1", "data2", "read"],
+                    ["user", "domain.*", "data3", "read"],
+                ]
+            ),
+        )
+
     def test_get_policy_multiple_matching_functions(self):
         e = self.get_enforcer(
             get_examples("rbac_with_domain_and_policy_pattern_model.conf"),
@@ -152,6 +158,7 @@ def test_get_policy_multiple_matching_functions(self):
             e.get_policy(),
             [
                 ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.*", "data3", "read"],
                 ["user", "domain.1", "data2", "read"],
                 ["user", "domain.1", "data2", "write"],
             ],
@@ -160,30 +167,47 @@ def test_get_policy_multiple_matching_functions(self):
         km2_fn = casbin.util.key_match2_func
 
         self.assertEqual(
-            e.get_filtered_policy(
-                1, partial(km2_fn, "domain.2"), lambda a: "data" in a
+            sorted(
+                e.get_filtered_policy(
+                    1, partial(km2_fn, "domain.2"), lambda a: "data" in a
+                )
+            ),
+            sorted(
+                [
+                    ["admin", "domain.*", "data1", "read"],
+                    ["user", "domain.*", "data3", "read"],
+                ]
             ),
-            [["admin", "domain.*", "data1", "read"]],
         )
 
         self.assertEqual(
-            e.get_filtered_policy(
-                1, partial(km2_fn, "domain.1"), lambda a: "data" in a, "read"
+            sorted(
+                e.get_filtered_policy(
+                    1, partial(km2_fn, "domain.1"), lambda a: "data" in a, "read"
+                )
+            ),
+            sorted(
+                [
+                    ["admin", "domain.*", "data1", "read"],
+                    ["user", "domain.1", "data2", "read"],
+                    ["user", "domain.*", "data3", "read"],
+                ]
             ),
-            [
-                ["admin", "domain.*", "data1", "read"],
-                ["user", "domain.1", "data2", "read"],
-            ],
         )
 
         self.assertEqual(
-            e.get_filtered_policy(
-                1, partial(km2_fn, "domain.1"), "", "reading".startswith
+            sorted(
+                e.get_filtered_policy(
+                    1, partial(km2_fn, "domain.1"), "", "reading".startswith
+                )
+            ),
+            sorted(
+                [
+                    ["admin", "domain.*", "data1", "read"],
+                    ["user", "domain.1", "data2", "read"],
+                    ["user", "domain.*", "data3", "read"],
+                ]
             ),
-            [
-                ["admin", "domain.*", "data1", "read"],
-                ["user", "domain.1", "data2", "read"],
-            ],
         )
 
     def test_modify_policy_api(self):

tests/test_rbac_api.py

@@ -233,6 +233,43 @@ def test_enforce_implicit_permissions_api_with_domain(self):
         )
         self.assertEqual(e.get_implicit_permissions_for_user("bob", "domain1"), [])
 
+    def test_enforce_implicit_permissions_api_with_domain_matching_function(self):
+
+        e = self.get_enforcer(
+            get_examples("rbac_with_domain_and_policy_pattern_model.conf"),
+            get_examples("rbac_with_domain_and_policy_pattern_policy.csv"),
+        )
+
+        e.get_role_manager().add_domain_matching_func(casbin.util.key_match2_func)
+
+        self.assertEqual(
+            e.get_implicit_permissions_for_user("alice", "domain.3"),
+            [["user", "domain.*", "data3", "read"]],
+        )
+
+        self.assertEqual(
+            e.get_implicit_permissions_for_user("alice", "domain.1"),
+            [
+                ["user", "domain.*", "data3", "read"],
+                ["user", "domain.1", "data2", "read"],
+                ["user", "domain.1", "data2", "write"],
+            ],
+        )
+
+        self.assertEqual(
+            e.get_implicit_permissions_for_user("bob", "domain.3"),
+            [["admin", "domain.*", "data1", "read"]],
+        )
+
+        self.assertEqual(
+            e.get_implicit_permissions_for_user("bob", "domain.2"),
+            [],
+        )
+
+        self.assertEqual(
+            sorted(e.get_implicit_permissions_for_user("bob", "domain.1")), []
+        )
+
     def test_enforce_implicit_permissions_api_with_domain_ignore_domain_policies_filter(
         self,
     ):