feat: support matching functions to filter policies (#222)

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

casbin/model/policy.py

@@ -69,7 +69,8 @@ def get_filtered_policy(self, sec, ptype, field_index, *field_values):
             rule
             for rule in self[sec][ptype].policy
             if all(
-                value == "" or rule[field_index + i] == value
+                (callable(value) and value(rule[field_index + i]))
+                or (value == "" or rule[field_index + i] == value)
                 for i, value in enumerate(field_values)
             )
         ]

examples/rbac_with_domain_and_policy_pattern_model.conf

@@ -0,0 +1,14 @@
+[request_definition]
+r = sub, dom, obj, act
+
+[policy_definition]
+p = sub, dom, obj, act
+
+[role_definition]
+g = _, _, _
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = g(r.sub, p.sub, r.dom) && keyMatch2(r.dom, p.dom) && r.obj == p.obj && r.act == p.act

examples/rbac_with_domain_and_policy_pattern_policy.csv

@@ -0,0 +1,6 @@
+p, admin, domain.*, data1, read
+p, user, domain.1, data2, read
+p, user, domain.1, data2, write
+
+g, alice, user, *
+g, bob, admin, domain.3

examples/rbac_with_domains_policy.csv

@@ -1,6 +1,6 @@
-p, admin, domain1, data1, read
-p, admin, domain1, data1, write
-p, admin, domain2, data2, read
-p, admin, domain2, data2, write
-g, alice, admin, domain1
+p, admin, domain1, data1, read
+p, admin, domain1, data1, write
+p, admin, domain2, data2, read
+p, admin, domain2, data2, write
+g, alice, admin, domain1
 g, bob, admin, domain2

tests/test_management_api.py

@@ -1,3 +1,5 @@
+from functools import partial
+
 import casbin
 from tests.test_enforcer import get_examples, TestCaseBase
 
@@ -99,6 +101,91 @@ def test_get_policy_api(self):
         self.assertTrue(e.has_grouping_policy(["alice", "data2_admin"]))
         self.assertFalse(e.has_grouping_policy(["bob", "data2_admin"]))
 
+    def test_get_policy_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"),
+        )
+
+        self.assertEqual(
+            e.get_policy(),
+            [
+                ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.1", "data2", "read"],
+                ["user", "domain.1", "data2", "write"],
+            ],
+        )
+
+        km2_fn = casbin.util.key_match2_func
+        self.assertEqual(
+            e.get_filtered_grouping_policy(2, partial(km2_fn, "domain.3")),
+            [["alice", "user", "*"], ["bob", "admin", "domain.3"]],
+        )
+
+        self.assertEqual(
+            e.get_filtered_grouping_policy(2, partial(km2_fn, "domain.1")),
+            [["alice", "user", "*"]],
+        )
+
+        # first 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"],
+            ],
+        )
+
+    def test_get_policy_multiple_matching_functions(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"),
+        )
+
+        self.assertEqual(
+            e.get_policy(),
+            [
+                ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.1", "data2", "read"],
+                ["user", "domain.1", "data2", "write"],
+            ],
+        )
+
+        km2_fn = casbin.util.key_match2_func
+
+        self.assertEqual(
+            e.get_filtered_policy(
+                1, partial(km2_fn, "domain.2"), lambda a: "data" in a
+            ),
+            [["admin", "domain.*", "data1", "read"]],
+        )
+
+        self.assertEqual(
+            e.get_filtered_policy(
+                1, partial(km2_fn, "domain.1"), lambda a: "data" in a, "read"
+            ),
+            [
+                ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.1", "data2", "read"],
+            ],
+        )
+
+        self.assertEqual(
+            e.get_filtered_policy(
+                1, partial(km2_fn, "domain.1"), "", "reading".startswith
+            ),
+            [
+                ["admin", "domain.*", "data1", "read"],
+                ["user", "domain.1", "data2", "read"],
+            ],
+        )
+
     def test_modify_policy_api(self):
         e = self.get_enforcer(
             get_examples("rbac_model.conf"),