feat: add field_index_map and constant (#314)

* feat: add constants

* feat: add field_index_map

Author: BustDot

casbin/async_internal_enforcer.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 import copy
 
+from casbin.core_enforcer import CoreEnforcer
 from casbin.model import Model, FunctionMap
 from casbin.persist import Adapter
-from casbin.core_enforcer import CoreEnforcer
 from casbin.persist.adapters.async_file_adapter import AsyncFileAdapter
 
 
@@ -160,7 +160,6 @@ async def _update_policy(self, sec, ptype, old_rule, new_rule):
             return rule_updated
 
         if self.adapter and self.auto_save:
-
             result = await self.adapter.update_policy(sec, ptype, old_rule, new_rule)
             if result is False:
                 return False
@@ -289,3 +288,12 @@ async def _remove_filtered_policy_returns_effects(self, sec, ptype, field_index,
                 self.watcher.update()
 
         return rule_removed
+
+    async def get_field_index(self, ptype, field):
+        """gets the index of the field name."""
+        return self.model.get_field_index(ptype, field)
+
+    async def set_field_index(self, ptype, field, index):
+        """sets the index of the field name."""
+        assertion = self.model["p"][ptype]
+        assertion.field_index_map[field] = index

casbin/constant/__init__.py

@@ -0,0 +1,26 @@
+# Copyright 2023 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Index constants
+DOMAIN_INDEX = "dom"
+SUBJECT_INDEX = "sub"
+OBJECT_INDEX = "obj"
+PRIORITY_INDEX = "priority"
+
+# Effect constants
+ALLOW_OVERRIDE_EFFECT = "some(where (p_eft == allow))"
+DENY_OVERRIDE_EFFECT = "!some(where (p_eft == deny))"
+ALLOW_AND_DENY_EFFECT = "some(where (p_eft == allow)) && !some(where (p_eft == deny))"
+PRIORITY_EFFECT = "priority(p_eft) || deny"
+SUBJECT_PRIORITY_EFFECT = "subjectPriority(p_eft) || deny"

casbin/constant/constants.py

@@ -19,18 +19,25 @@
     PriorityEffector,
 )
 from .effector import Effector
+from ..constant.constants import (
+    ALLOW_OVERRIDE_EFFECT,
+    SUBJECT_PRIORITY_EFFECT,
+    PRIORITY_EFFECT,
+    DENY_OVERRIDE_EFFECT,
+    ALLOW_AND_DENY_EFFECT,
+)
 
 
 def get_effector(expr):
     """creates an effector based on the current policy effect expression"""
 
-    if expr == "some(where (p_eft == allow))":
+    if expr == ALLOW_OVERRIDE_EFFECT:
         return AllowOverrideEffector()
-    elif expr == "!some(where (p_eft == deny))":
+    elif expr == DENY_OVERRIDE_EFFECT:
         return DenyOverrideEffector()
-    elif expr == "some(where (p_eft == allow)) && !some(where (p_eft == deny))":
+    elif expr == ALLOW_AND_DENY_EFFECT:
         return AllowAndDenyEffector()
-    elif expr == "priority(p_eft) || deny" or expr == "subjectPriority(p_eft) || deny":
+    elif expr == PRIORITY_EFFECT or expr == SUBJECT_PRIORITY_EFFECT:
         return PriorityEffector()
     else:
         raise RuntimeError("unsupported effect")

casbin/effect/__init__.py

@@ -67,7 +67,6 @@ def _update_policy(self, sec, ptype, old_rule, new_rule):
             return rule_updated
 
         if self.adapter and self.auto_save:
-
             if self.adapter.update_policy(sec, ptype, old_rule, new_rule) is False:
                 return False
 
@@ -84,7 +83,6 @@ def _update_policies(self, sec, ptype, old_rules, new_rules):
             return rules_updated
 
         if self.adapter and self.auto_save:
-
             if self.adapter.update_policies(sec, ptype, old_rules, new_rules) is False:
                 return False
 
@@ -189,3 +187,12 @@ def _remove_filtered_policy_returns_effects(self, sec, ptype, field_index, *fiel
                 self.watcher.update()
 
         return rule_removed
+
+    def get_field_index(self, ptype, field):
+        """gets the index of the field name."""
+        return self.model.get_field_index(ptype, field)
+
+    def set_field_index(self, ptype, field, index):
+        """sets the index of the field name."""
+        assertion = self.model["p"][ptype]
+        assertion.field_index_map[field] = index

casbin/internal_enforcer.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import logging
+
 from casbin.model.policy_op import PolicyOp
 
 
@@ -26,6 +27,7 @@ def __init__(self):
         self.rm = None
         self.priority_index: int = -1
         self.policy_map: dict = {}
+        self.field_index_map: dict = {}
 
     def build_role_links(self, rm):
         self.rm = rm

casbin/model/assertion.py

@@ -12,16 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from . import Assertion
 from casbin import util, config
+from . import Assertion
 from .policy import Policy
 
 DEFAULT_DOMAIN = ""
 DEFAULT_SEPARATOR = "::"
 
 
 class Model(Policy):
-
     section_name_map = {
         "r": "request_definition",
         "p": "policy_definition",
@@ -207,3 +206,23 @@ def write_string(sec):
         s[-1] = s[-1].strip()
 
         return "".join(s)
+
+    def get_field_index(self, ptype, field):
+        """get_field_index gets the index of the field for a ptype in a policy,
+        return -1 if the field does not exist."""
+        assertion = self["p"][ptype]
+        if field in assertion.field_index_map:
+            return assertion.field_index_map[field]
+
+        pattern = f"{ptype}_{field}"
+        index = -1
+        for i, token in enumerate(assertion.tokens):
+            if token == pattern:
+                index = i
+                break
+
+        if index == -1:
+            return index
+
+        assertion.field_index_map[field] = index
+        return index

casbin/model/model.py

@@ -628,3 +628,12 @@ def build_incremental_role_links(self, op, ptype, rules):
 
     def new_enforce_context(self, suffix: str) -> "EnforceContext":
         return self._e.new_enforce_context(suffix)
+
+    def get_field_index(self, ptype, field):
+        """gets the index of the field name."""
+        return self._e.model.get_field_index(ptype, field)
+
+    def set_field_index(self, ptype, field, index):
+        """sets the index of the field name."""
+        assertion = self._e.model["p"][ptype]
+        assertion.field_index_map[field] = index

casbin/synced_enforcer.py

@@ -14,6 +14,7 @@
 from unittest import IsolatedAsyncioTestCase
 
 import casbin
+from casbin.constant.constants import DOMAIN_INDEX
 from tests.test_enforcer import get_examples, TestCaseBase
 
 
@@ -254,7 +255,6 @@ 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"),
@@ -405,6 +405,15 @@ def test_domain_match_model(self):
         self.assertTrue(e.enforce("bob", "domain2", "data2", "read"))
         self.assertTrue(e.enforce("bob", "domain2", "data2", "write"))
 
+    def test_set_field_index(self):
+        e = self.get_enforcer(
+            get_examples("rbac_with_domains_model.conf"),
+            get_examples("rbac_with_domains_policy.csv"),
+        )
+        self.assertEqual(e.get_field_index("p", DOMAIN_INDEX), 1)
+        e.set_field_index("p", DOMAIN_INDEX, 2)
+        self.assertEqual(e.get_field_index("p", DOMAIN_INDEX), 2)
+
 
 class TestRbacApiSynced(TestRbacApi):
     def get_enforcer(self, model=None, adapter=None):
@@ -686,7 +695,6 @@ async def test_enforce_implicit_permissions_api_with_domain(self):
         self.assertEqual(await e.get_implicit_permissions_for_user("bob", "domain1"), [])
 
     async 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"),
@@ -847,3 +855,12 @@ async def test_domain_match_model(self):
         self.assertFalse(e.enforce("bob", "domain2", "data1", "write"))
         self.assertTrue(e.enforce("bob", "domain2", "data2", "read"))
         self.assertTrue(e.enforce("bob", "domain2", "data2", "write"))
+
+    async def test_set_field_index(self):
+        e = self.get_enforcer(
+            get_examples("rbac_with_domains_model.conf"),
+            get_examples("rbac_with_domains_policy.csv"),
+        )
+        self.assertEqual(await e.get_field_index("p", DOMAIN_INDEX), 1)
+        await e.set_field_index("p", DOMAIN_INDEX, 2)
+        self.assertEqual(await e.get_field_index("p", DOMAIN_INDEX), 2)