feat: add temporal role model (#320)

Author: BustDot

casbin/core_enforcer.py

@@ -20,7 +20,7 @@
 from casbin.persist import Adapter
 from casbin.persist.adapters import FileAdapter
 from casbin.rbac import default_role_manager
-from casbin.util import generate_g_function, SimpleEval, util
+from casbin.util import generate_g_function, SimpleEval, util, generate_conditional_g_function
 from casbin.util.log import configure_logging
 
 
@@ -47,6 +47,7 @@ class CoreEnforcer:
     adapter = None
     watcher = None
     rm_map = None
+    cond_rm_map = None
 
     enabled = False
     auto_save = False
@@ -104,6 +105,7 @@ def init_with_model_and_adapter(self, m, adapter=None):
 
     def _initialize(self):
         self.rm_map = dict()
+        self.cond_rm_map = dict()
         self.eft = get_effector(self.model["e"]["e"].value)
         self.watcher = None
 
@@ -192,10 +194,26 @@ def init_rm_map(self):
         if "g" in self.model.keys():
             for ptype in self.model["g"]:
                 assertion = self.model["g"][ptype]
-                if assertion.value.count("_") == 2:
-                    self.rm_map[ptype] = default_role_manager.RoleManager(10)
-                else:
-                    self.rm_map[ptype] = default_role_manager.DomainManager(10)
+                if ptype in self.rm_map:
+                    rm = self.rm_map[ptype]
+                    rm.clear()
+                    continue
+
+                if len(assertion.tokens) <= 2 and len(assertion.params_tokens) == 0:
+                    assertion.rm = default_role_manager.RoleManager(10)
+                    self.rm_map[ptype] = assertion.rm
+
+                if len(assertion.tokens) <= 2 and len(assertion.params_tokens) != 0:
+                    assertion.cond_rm = default_role_manager.ConditionalRoleManager(10)
+                    self.cond_rm_map[ptype] = assertion.cond_rm
+
+                if len(assertion.tokens) > 2:
+                    if len(assertion.params_tokens) == 0:
+                        assertion.rm = default_role_manager.DomainManager(10)
+                        self.rm_map[ptype] = assertion.rm
+                    else:
+                        assertion.cond_rm = default_role_manager.ConditionalDomainManager(10)
+                        self.cond_rm_map[ptype] = assertion.cond_rm
 
     def load_policy(self):
         """reloads the policy from file/database."""
@@ -216,8 +234,13 @@ def load_policy(self):
                 need_to_rebuild = True
                 for rm in self.rm_map.values():
                     rm.clear()
+                if len(self.rm_map) != 0:
+                    new_model.build_role_links(self.rm_map)
 
-                new_model.build_role_links(self.rm_map)
+                for crm in self.cond_rm_map.values():
+                    crm.clear()
+                if len(self.cond_rm_map) != 0:
+                    new_model.build_conditional_role_links(self.cond_rm_map)
 
             self.model = new_model
 
@@ -313,6 +336,40 @@ def add_named_domain_matching_func(self, ptype, fn):
 
         return False
 
+    def add_named_link_condition_func(self, ptype, user, role, fn):
+        """Add condition function fn for Link userName->roleName,
+        when fn returns true, Link is valid, otherwise invalid"""
+        if ptype in self.cond_rm_map:
+            rm = self.cond_rm_map[ptype]
+            rm.add_link_condition_func(user, role, fn)
+            return True
+        return False
+
+    def add_named_domain_link_condition_func(self, ptype, user, role, domain, fn):
+        """Add condition function fn for Link userName-> {roleName, domain},
+        when fn returns true, Link is valid, otherwise invalid"""
+        if ptype in self.cond_rm_map:
+            rm = self.cond_rm_map[ptype]
+            rm.add_domain_link_condition_func(user, role, domain, fn)
+            return True
+        return False
+
+    def set_named_link_condition_func_params(self, ptype, user, role, *params):
+        """Sets the parameters of the condition function fn for Link userName->roleName"""
+        if ptype in self.cond_rm_map:
+            rm = self.cond_rm_map[ptype]
+            rm.set_link_condition_func_params(user, role, *params)
+            return True
+        return False
+
+    def set_named_domain_link_condition_func_params(self, ptype, user, role, domain, *params):
+        """Sets the parameters of the condition function fn for Link userName->{roleName, domain}"""
+        if ptype in self.cond_rm_map:
+            rm = self.cond_rm_map[ptype]
+            rm.set_domain_link_condition_func_params(user, role, domain, *params)
+            return True
+        return False
+
     def new_enforce_context(self, suffix: str) -> EnforceContext:
         return EnforceContext(
             rtype="r" + suffix,
@@ -346,8 +403,10 @@ def enforce_ex(self, *rvals):
 
         if "g" in self.model.keys():
             for key, ast in self.model["g"].items():
-                rm = ast.rm
-                functions[key] = generate_g_function(rm)
+                if len(self.rm_map) != 0:
+                    functions[key] = generate_g_function(ast.rm)
+                if len(self.cond_rm_map) != 0:
+                    functions[key] = generate_conditional_g_function(ast.cond_rm)
 
         if len(rvals) != 0:
             if isinstance(rvals[0], EnforceContext):

casbin/management_enforcer.py

@@ -238,6 +238,10 @@ def add_named_grouping_policy(self, ptype, *params):
 
         if self.auto_build_role_links:
             self.model.build_incremental_role_links(self.rm_map[ptype], PolicyOp.Policy_add, "g", ptype, rules)
+            if ptype in self.cond_rm_map:
+                self.model.build_incremental_conditional_role_links(
+                    self.cond_rm_map[ptype], PolicyOp.Policy_add, "g", ptype, rules
+                )
         return rule_added
 
     def add_named_grouping_policies(self, ptype, rules):

casbin/model/assertion.py

@@ -23,8 +23,10 @@ def __init__(self):
         self.key = ""
         self.value = ""
         self.tokens = []
+        self.params_tokens = []
         self.policy = []
         self.rm = None
+        self.cond_rm = None
         self.priority_index: int = -1
         self.policy_map: dict = {}
         self.field_index_map: dict = {}
@@ -62,3 +64,48 @@ def build_incremental_role_links(self, rm, op, rules):
                 rm.delete_link(rule[0], rule[1], *rule[2:])
             else:
                 raise TypeError("Invalid operation: " + str(op))
+
+    def build_incremental_conditional_role_links(self, cond_rm, op, rules):
+        self.cond_rm = cond_rm
+        count = self.value.count("_")
+        if count < 2:
+            raise RuntimeError('the number of "_" in role definition should be at least 2')
+
+        for rule in rules:
+            if len(rule) < count:
+                raise TypeError("grouping policy elements do not meet role definition")
+            if len(rule) > count:
+                rule = rule[:count]
+
+            domain_rule = rule[2 : len(self.tokens)]
+
+            if op == PolicyOp.Policy_add:
+                self.add_conditional_role_link(rule, domain_rule)
+            elif op == PolicyOp.Policy_remove:
+                self.cond_rm.delete_link(rule[0], rule[1], *rule[2:])
+            else:
+                raise TypeError("Invalid operation: " + str(op))
+
+    def build_conditional_role_links(self, cond_rm):
+        self.cond_rm = cond_rm
+        count = self.value.count("_")
+        if count < 2:
+            raise RuntimeError('the number of "_" in role definition should be at least 2')
+        for rule in self.policy:
+            if len(rule) < count:
+                raise TypeError("grouping policy elements do not meet role definition")
+            if len(rule) > count:
+                rule = rule[:count]
+
+            domain_rule = rule[2 : len(self.tokens)]
+
+            self.add_conditional_role_link(rule, domain_rule)
+
+    def add_conditional_role_link(self, rule, domain_rule):
+        if not domain_rule:
+            self.cond_rm.add_link(rule[0], rule[1])
+            self.cond_rm.set_link_condition_func_params(rule[0], rule[1], *rule[len(self.tokens) :])
+        else:
+            domain = domain_rule[0]
+            self.cond_rm.add_link(rule[0], rule[1], domain)
+            self.cond_rm.set_domain_link_condition_func_params(rule[0], rule[1], domain, *rule[len(self.tokens) :])

casbin/model/function.py

@@ -35,6 +35,7 @@ def load_function_map():
         fm.add_function("regexMatch", util.regex_match_func)
         fm.add_function("ipMatch", util.ip_match_func)
         fm.add_function("globMatch", util.glob_match_func)
+        fm.add_function("timeMatch", util.time_match_func)
 
         return fm
 

casbin/model/model.py

@@ -11,13 +11,15 @@
 # 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.
+import re
 
 from casbin import util, config
 from . import Assertion
 from .policy import Policy
 
 DEFAULT_DOMAIN = ""
 DEFAULT_SEPARATOR = "::"
+PARAMS_REGEX = re.compile(r"\((.*?)\)")
 
 
 class Model(Policy):
@@ -34,6 +36,18 @@ def _load_assertion(self, cfg, sec, key):
 
         return self.add_def(sec, key, value)
 
+    def get_params_token(self, value):
+        """get_params_token Get params_token from Assertion.value"""
+        # Find the matching string using the regular expression
+        params_string = PARAMS_REGEX.search(value)
+
+        if params_string is None:
+            return []
+
+        # Extract the captured group (inside parentheses) and split it by commas
+        params_string = params_string.group(1)
+        return [param.strip() for param in params_string.split(",")]
+
     def add_def(self, sec, key, value):
         if value == "":
             return
@@ -46,6 +60,10 @@ def add_def(self, sec, key, value):
             ast.tokens = ast.value.split(",")
             for i, token in enumerate(ast.tokens):
                 ast.tokens[i] = key + "_" + token.strip()
+        elif "g" == sec:
+            ast.params_tokens = self.get_params_token(ast.value)
+            ast.tokens = ast.value.split(",")
+            ast.tokens = ast.tokens[: len(ast.tokens) - len(ast.params_tokens)]
         else:
             ast.value = util.remove_comments(util.escape_assertion(ast.value))
 

casbin/model/policy.py

@@ -51,6 +51,20 @@ def build_incremental_role_links(self, rm, op, sec, ptype, rules):
         if sec == "g":
             self[sec].get(ptype).build_incremental_role_links(rm, op, rules)
 
+    def build_incremental_conditional_role_links(self, cond_rm, op, sec, ptype, rules):
+        if sec == "g":
+            return self[sec].get(ptype).build_incremental_conditional_role_links(cond_rm, op, rules)
+        return None
+
+    def build_conditional_role_links(self, cond_rm_map):
+        if "g" not in self.keys():
+            return
+        self.print_policy()
+        for ptype, ast in self["g"].items():
+            cond_rm = cond_rm_map.get(ptype)
+            if cond_rm:
+                ast.build_conditional_role_links(cond_rm)
+
     def print_policy(self):
         """Log using info"""
 

casbin/rbac/__init__.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .role_manager import RoleManager
+from .role_manager import RoleManager, ConditionalRoleManager

casbin/rbac/default_role_manager/__init__.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .role_manager import RoleManager, DomainManager
+from .role_manager import RoleManager, DomainManager, ConditionalRoleManager, ConditionalDomainManager