1262282997

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/casbin/pycasbin/commit/<a class=hub.com/casbin/pycasbin/commit/<a class="double-link" href="https://git"><a class=hub.com/casbin/pycasbin/commit/<a class="double-link" href="https://git"><a class=hub.com/casbin/pycasbin/commit/<a class="double-link" href="https://git"><a class=hub.com/casbin/pycasbin/commit/667b08b8ef2451686013b80371505bbc1e041aa1">667b08b8e">&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/</a><a class="double-link" href="https://github.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/667b08b8ef2451686013b80371505bbc1e041aa1">667b08b8e</a><a href="https://github.com/casbin/pycasbin/commit/667b08b8ef2451686013b80371505bbc1e041aa1">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/casbin/pycasbin/commit/667b08b8ef2451686013b80371505bbc1e041aa1&quot;&gt;667b08b8e&lt;/a&gt;&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/667b08b8ef2451686013b80371505bbc1e041aa1&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/casbin/pycasbin/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/casbin/pycasbin/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&am... (continued)

Pull Request Pull Request #190: feat: add explicit priority support

Run Details

49 of 57 new or added lines in 5 files covered. (85.96%)

1690 of 2006 relevant lines covered (84.25%)

10.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.45

/casbin/core_enforcer.py

import logging, copy

from casbin.effect import Effector, get_effector, effect_to_bool
from casbin.model import Model, FunctionMap
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


class EnforceContext:
    """
    EnforceContext is used as the first element of the parameter "rvals" in method "enforce"
    """

    def __init__(self, rtype: str, ptype: str, etype: str, mtype: str):
        self.rtype: str = rtype
        self.ptype: str = ptype
        self.etype: str = etype
        self.mtype: str = mtype


class CoreEnforcer:
    """CoreEnforcer defines the core functionality of an enforcer."""

    model_path = ""
    model = None
    fm = None
    eft = None

    adapter = None
    watcher = None
    rm_map = None

    enabled = False
    auto_save = False
    auto_build_role_links = False

    def __init__(self, model=None, adapter=None):
        self.logger = logging.getLogger(__name__)
        if isinstance(model, str):
            if isinstance(adapter, str):
                self.init_with_file(model, adapter)
            else:
                self.init_with_adapter(model, adapter)
                pass
        else:
            if isinstance(adapter, str):
                raise RuntimeError("Invalid parameters for enforcer.")
            else:
                self.init_with_model_and_adapter(model, adapter)

    def init_with_file(self, model_path, policy_path):
        """initializes an enforcer with a model file and a policy file."""
        a = FileAdapter(policy_path)
        self.init_with_adapter(model_path, a)

    def init_with_adapter(self, model_path, adapter=None):
        """initializes an enforcer with a database adapter."""
        m = self.new_model(model_path)
        self.init_with_model_and_adapter(m, adapter)

        self.model_path = model_path

    def init_with_model_and_adapter(self, m, adapter=None):
        """initializes an enforcer with a model and a database adapter."""

        if (
            not isinstance(m, Model)
            or adapter is not None
            and not isinstance(adapter, Adapter)
        ):
            raise RuntimeError("Invalid parameters for enforcer.")

        self.adapter = adapter

        self.model = m
        m.logger = self.logger
        self.model.print_model()
        self.fm = FunctionMap.load_function_map()

        self._initialize()

        # Do not initialize the full policy when using a filtered adapter
        if self.adapter and not self.is_filtered():
            self.load_policy()

    def _initialize(self):
        self.rm_map = dict()
        self.eft = get_effector(self.model["e"]["e"].value)
        self.watcher = None

        self.enabled = True
        self.auto_save = True
        self.auto_build_role_links = True

        self.init_rm_map()

    @staticmethod
    def new_model(path="", text=""):
        """creates a model."""

        m = Model()
        if len(path) > 0:
            m.load_model(path)
        else:
            m.load_model_from_text(text)

        return m

    def load_model(self):
        """reloads the model from the model CONF file.
        Because the policy is attached to a model, so the policy is invalidated and needs to be reloaded by calling LoadPolicy().
        """

        self.model = self.new_model()
        self.model.load_model(self.model_path)
        self.model.print_model()
        self.fm = FunctionMap.load_function_map()

    def get_model(self):
        """gets the current model."""

        return self.model

    def set_model(self, m):
        """sets the current model."""

        self.model = m
        self.fm = FunctionMap.load_function_map()

    def get_adapter(self):
        """gets the current adapter."""

        return self.adapter

    def set_adapter(self, adapter):
        """sets the current adapter."""

        self.adapter = adapter

    def set_watcher(self, watcher):
        """sets the current watcher."""

        self.watcher = watcher
        pass

    def get_role_manager(self):
        """gets the current role manager."""
        return self.rm_map["g"]

    def get_named_role_manager(self, ptype):
        if ptype in self.rm_map.keys():
            return self.rm_map.get(ptype)
        raise ValueError("ptype not found")

    def set_role_manager(self, rm):
        """sets the current role manager."""
        self.rm_map["g"] = rm

    def set_named_role_manager(self, ptype, rm):
        self.rm_map[ptype] = rm

    def set_effector(self, eft):
        """sets the current effector."""
        self.eft = eft

    def clear_policy(self):
        """clears all policy."""

        self.model.clear_policy()

    def init_rm_map(self):
        if "g" in self.model.keys():
            for ptype in self.model["g"]:
                self.rm_map[ptype] = default_role_manager.RoleManager(10)

    def load_policy(self):
        """reloads the policy from file/database."""
        need_to_rebuild = False
        new_model = copy.copy(self.model)
        new_model.clear_policy()

        try:

            self.adapter.load_policy(new_model)

            new_model.sort_policies_by_priority()

            self.init_rm_map()
            self.model.print_policy()

            if self.auto_build_role_links:

                need_to_rebuild = True
                for rm in self.rm_map.values():
                    rm.clear()

                new_model.build_role_links(self.rm_map)
                self.build_role_links()

            self.model = new_model

        except:

            if self.auto_build_role_links and need_to_rebuild:
                self.build_role_links()

    def load_filtered_policy(self, filter):
        """reloads a filtered policy from file/database."""
        self.model.clear_policy()

        if not hasattr(self.adapter, "is_filtered"):
            raise ValueError("filtered policies are not supported by this adapter")

        self.adapter.load_filtered_policy(self.model, filter)

        self.model.sort_policies_by_priority()

        self.init_rm_map()
        self.model.print_policy()
        if self.auto_build_role_links:
            self.build_role_links()

    def load_increment_filtered_policy(self, filter):
        """LoadIncrementalFilteredPolicy append a filtered policy from file/database."""
        if not hasattr(self.adapter, "is_filtered"):
            raise ValueError("filtered policies are not supported by this adapter")

        self.adapter.load_filtered_policy(self.model, filter)
        self.model.print_policy()
        if self.auto_build_role_links:
            self.build_role_links()

    def is_filtered(self):
        """returns true if the loaded policy has been filtered."""

        return hasattr(self.adapter, "is_filtered") and self.adapter.is_filtered()

    def save_policy(self):
        if self.is_filtered():
            raise RuntimeError("cannot save a filtered policy")

        self.adapter.save_policy(self.model)

        if self.watcher:
            self.watcher.update()

    def enable_enforce(self, enabled=True):
        """changes the enforcing state of Casbin,
        when Casbin is disabled, all access will be allowed by the Enforce() function.
        """

        self.enabled = enabled

    def enable_auto_save(self, auto_save):
        """controls whether to save a policy rule automatically to the adapter when it is added or removed."""
        self.auto_save = auto_save

    def enable_auto_build_role_links(self, auto_build_role_links):
        """controls whether to rebuild the role inheritance relations when a role is added or deleted."""
        self.auto_build_role_links = auto_build_role_links

    def build_role_links(self):
        """manually rebuild the role inheritance relations."""

        for rm in self.rm_map.values():
            rm.clear()

        self.model.build_role_links(self.rm_map)

    def add_named_matching_func(self, ptype, fn):
        """add_named_matching_func add MatchingFunc by ptype RoleManager"""
        try:
            self.rm_map[ptype].add_matching_func(fn)
            return True
        except:
            return False

    def add_named_domain_matching_func(self, ptype, fn):
        """add_named_domain_matching_func add MatchingFunc by ptype to RoleManager"""
        if ptype in self.rm_map.keys():
            self.rm_map[ptype].add_domain_matching_func(fn)
            return True

        return False

    def new_enforce_context(self, suffix: str) -> EnforceContext:

        return EnforceContext(
            rtype="r" + suffix,
            ptype="p" + suffix,
            etype="e" + suffix,
            mtype="m" + suffix,
        )

    def enforce(self, *rvals):
        """decides whether a "subject" can access a "object" with the operation "action",
        input parameters are usually: (sub, obj, act).
        """
        result, _ = self.enforce_ex(*rvals)
        return result

    def enforce_ex(self, *rvals):
        """decides whether a "subject" can access a "object" with the operation "action",
        input parameters are usually: (sub, obj, act).
        return judge result with reason
        """

        rtype = "r"
        ptype = "p"
        etype = "e"
        mtype = "m"

        if not self.enabled:
            return [False, []]

        functions = self.fm.get_functions()

        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(rvals) != 0:
            if isinstance(rvals[0], EnforceContext):
                enforce_context = rvals[0]
                rtype = enforce_context.rtype
                ptype = enforce_context.ptype
                etype = enforce_context.etype
                mtype = enforce_context.mtype
                rvals = rvals[1:]

        if "m" not in self.model.keys():
            raise RuntimeError("model is undefined")

        if "m" not in self.model["m"].keys():
            raise RuntimeError("model is undefined")

        r_tokens = self.model["r"][rtype].tokens
        p_tokens = self.model["p"][ptype].tokens

        if len(r_tokens) != len(rvals):
            raise RuntimeError("invalid request size")

        exp_string = self.model["m"][mtype].value
        has_eval = util.has_eval(exp_string)
        if not has_eval:
            expression = self._get_expression(exp_string, functions)

        policy_effects = set()

        r_parameters = dict(zip(r_tokens, rvals))

        policy_len = len(self.model["p"][ptype].policy)

        explain_index = -1
        if not 0 == policy_len:
            for i, pvals in enumerate(self.model["p"][ptype].policy):
                if len(p_tokens) != len(pvals):
                    raise RuntimeError("invalid policy size")

                p_parameters = dict(zip(p_tokens, pvals))
                parameters = dict(r_parameters, **p_parameters)

                if util.has_eval(exp_string):
                    rule_names = util.get_eval_value(exp_string)
                    rules = [
                        util.escape_assertion(p_parameters[rule_name])
                        for rule_name in rule_names
                    ]
                    exp_with_rule = util.replace_eval(exp_string, rules)
                    expression = self._get_expression(exp_with_rule, functions)

                result = expression.eval(parameters)

                if isinstance(result, bool):
                    if not result:
                        policy_effects.add(Effector.INDETERMINATE)
                        continue
                elif isinstance(result, float):
                    if 0 == result:
                        policy_effects.add(Effector.INDETERMINATE)
                        continue
                else:
                    raise RuntimeError("matcher result should be bool, int or float")

                p_eft_key = ptype + "_eft"
                if p_eft_key in parameters.keys():
                    eft = parameters[p_eft_key]
                    if "allow" == eft:
                        policy_effects.add(Effector.ALLOW)
                    elif "deny" == eft:
                        policy_effects.add(Effector.DENY)
                    else:
                        policy_effects.add(Effector.INDETERMINATE)
                else:
                    policy_effects.add(Effector.ALLOW)

                if (
                    self.eft.intermediate_effect(policy_effects)
                    != Effector.INDETERMINATE
                ):
                    explain_index = i
                    break

        else:
            if has_eval:
                raise RuntimeError(
                    "please make sure rule exists in policy when using eval() in matcher"
                )

            parameters = r_parameters.copy()

            for token in self.model["p"][ptype].tokens:
                parameters[token] = ""

            result = expression.eval(parameters)

            if result:
                policy_effects.add(Effector.ALLOW)
            else:
                policy_effects.add(Effector.INDETERMINATE)

        final_effect = self.eft.final_effect(policy_effects)
        result = effect_to_bool(final_effect)

        # Log request.

        req_str = "Request: "
        req_str = req_str + ", ".join([str(v) for v in rvals])

        req_str = req_str + " ---> %s" % result
        if result:
            self.logger.info(req_str)
        else:
            # leaving this in error for now, if it's very noise this can be changed to info or debug
            self.logger.error(req_str)

        explain_rule = []
        if explain_index != -1 and explain_index < policy_len:
            explain_rule = self.model["p"][ptype].policy[explain_index]

        return result, explain_rule

    @staticmethod
    def _get_expression(expr, functions=None):
        expr = expr.replace("&&", "and")
        expr = expr.replace("||", "or")
        expr = expr.replace("!", "not")

        return SimpleEval(expr, functions)

1	import logging, copy	12✔
2
3	from casbin.effect import Effector, get_effector, effect_to_bool	12✔
4	from casbin.model import Model, FunctionMap	12✔
5	from casbin.persist import Adapter	12✔
6	from casbin.persist.adapters import FileAdapter	12✔
7	from casbin.rbac import default_role_manager	12✔
8	from casbin.util import generate_g_function, SimpleEval, util	12✔
9
10
11	class EnforceContext:	12✔
12	"""
13	EnforceContext is used as the first element of the parameter "rvals" in method "enforce"
14	"""
15
16	def __init__(self, rtype: str, ptype: str, etype: str, mtype: str):	12✔
17	self.rtype: str = rtype	12✔
18	self.ptype: str = ptype	12✔
19	self.etype: str = etype	12✔
20	self.mtype: str = mtype	12✔
21
22
23	class CoreEnforcer:	12✔
24	"""CoreEnforcer defines the core functionality of an enforcer."""
25
26	model_path = ""	12✔
27	model = None	12✔
28	fm = None	12✔
29	eft = None	12✔
30
31	adapter = None	12✔
32	watcher = None	12✔
33	rm_map = None	12✔
34
35	enabled = False	12✔
36	auto_save = False	12✔
37	auto_build_role_links = False	12✔
38
39	def __init__(self, model=None, adapter=None):	12✔
40	self.logger = logging.getLogger(__name__)	12✔
41	if isinstance(model, str):	12✔
42	if isinstance(adapter, str):	12✔
43	self.init_with_file(model, adapter)	12✔
44	else:
45	self.init_with_adapter(model, adapter)	12✔
46	pass	12✔
47	else:
48	if isinstance(adapter, str):	×
49	raise RuntimeError("Invalid parameters for enforcer.")	×
50	else:
51	self.init_with_model_and_adapter(model, adapter)	×
52
53	def init_with_file(self, model_path, policy_path):	12✔
54	"""initializes an enforcer with a model file and a policy file."""
55	a = FileAdapter(policy_path)	12✔
56	self.init_with_adapter(model_path, a)	12✔
57
58	def init_with_adapter(self, model_path, adapter=None):	12✔
59	"""initializes an enforcer with a database adapter."""
60	m = self.new_model(model_path)	12✔
61	self.init_with_model_and_adapter(m, adapter)	12✔
62
63	self.model_path = model_path	12✔
64
65	def init_with_model_and_adapter(self, m, adapter=None):	12✔
66	"""initializes an enforcer with a model and a database adapter."""
67
68	if (	12✔
69	not isinstance(m, Model)
70	or adapter is not None
71	and not isinstance(adapter, Adapter)
72	):
73	raise RuntimeError("Invalid parameters for enforcer.")	×
74
75	self.adapter = adapter	12✔
76
77	self.model = m	12✔
78	m.logger = self.logger	12✔
79	self.model.print_model()	12✔
80	self.fm = FunctionMap.load_function_map()	12✔
81
82	self._initialize()	12✔
83
84	# Do not initialize the full policy when using a filtered adapter
85	if self.adapter and not self.is_filtered():	12✔
86	self.load_policy()	12✔
87
88	def _initialize(self):	12✔
89	self.rm_map = dict()	12✔
90	self.eft = get_effector(self.model["e"]["e"].value)	12✔
91	self.watcher = None	12✔
92
93	self.enabled = True	12✔
94	self.auto_save = True	12✔
95	self.auto_build_role_links = True	12✔
96
97	self.init_rm_map()	12✔
98
99	@staticmethod	12✔
100	def new_model(path="", text=""):	12✔
101	"""creates a model."""
102
103	m = Model()	12✔
104	if len(path) > 0:	12✔
105	m.load_model(path)	12✔
106	else:
107	m.load_model_from_text(text)	12✔
108
109	return m	12✔
110
111	def load_model(self):	12✔
112	"""reloads the model from the model CONF file.
113	Because the policy is attached to a model, so the policy is invalidated and needs to be reloaded by calling LoadPolicy().
114	"""
115
116	self.model = self.new_model()	12✔
117	self.model.load_model(self.model_path)	12✔
118	self.model.print_model()	12✔
119	self.fm = FunctionMap.load_function_map()	12✔
120
121	def get_model(self):	12✔
122	"""gets the current model."""
123
124	return self.model	12✔
125
126	def set_model(self, m):	12✔
127	"""sets the current model."""
128
129	self.model = m	12✔
130	self.fm = FunctionMap.load_function_map()	12✔
131
132	def get_adapter(self):	12✔
133	"""gets the current adapter."""
134
135	return self.adapter	12✔
136
137	def set_adapter(self, adapter):	12✔
138	"""sets the current adapter."""
139
140	self.adapter = adapter	×
141
142	def set_watcher(self, watcher):	12✔
143	"""sets the current watcher."""
144
145	self.watcher = watcher	×
146	pass	×
147
148	def get_role_manager(self):	12✔
149	"""gets the current role manager."""
150	return self.rm_map["g"]	12✔
151
152	def get_named_role_manager(self, ptype):	12✔
153	if ptype in self.rm_map.keys():	×
154	return self.rm_map.get(ptype)	×
155	raise ValueError("ptype not found")	×
156
157	def set_role_manager(self, rm):	12✔
158	"""sets the current role manager."""
159	self.rm_map["g"] = rm	×
160
161	def set_named_role_manager(self, ptype, rm):	12✔
162	self.rm_map[ptype] = rm	×
163
164	def set_effector(self, eft):	12✔
165	"""sets the current effector."""
166	self.eft = eft	×
167
168	def clear_policy(self):	12✔
169	"""clears all policy."""
170
171	self.model.clear_policy()	×
172
173	def init_rm_map(self):	12✔
174	if "g" in self.model.keys():	12✔
175	for ptype in self.model["g"]:	12✔
176	self.rm_map[ptype] = default_role_manager.RoleManager(10)	12✔
177
178	def load_policy(self):	12✔
179	"""reloads the policy from file/database."""
180	need_to_rebuild = False	12✔
181	new_model = copy.copy(self.model)	12✔
182	new_model.clear_policy()	12✔
183
184	try:	12✔
185
186	self.adapter.load_policy(new_model)	12✔
187
188	new_model.sort_policies_by_priority()	12✔
189
190	self.init_rm_map()	12✔
191	self.model.print_policy()	12✔
192
193	if self.auto_build_role_links:	12✔
194
195	need_to_rebuild = True	12✔
196	for rm in self.rm_map.values():	12✔
197	rm.clear()	12✔
198
199	new_model.build_role_links(self.rm_map)	12✔
200	self.build_role_links()	12✔
201
202	self.model = new_model	12✔
203
NEW 204	except:	×
205
NEW 206	if self.auto_build_role_links and need_to_rebuild:	×
NEW 207	self.build_role_links()	×
208
209	def load_filtered_policy(self, filter):	12✔
210	"""reloads a filtered policy from file/database."""
211	self.model.clear_policy()	12✔
212
213	if not hasattr(self.adapter, "is_filtered"):	12✔
214	raise ValueError("filtered policies are not supported by this adapter")	12✔
215
216	self.adapter.load_filtered_policy(self.model, filter)	12✔
217
218	self.model.sort_policies_by_priority()	12✔
219
220	self.init_rm_map()	12✔
221	self.model.print_policy()	12✔
222	if self.auto_build_role_links:	12✔
223	self.build_role_links()	12✔
224
225	def load_increment_filtered_policy(self, filter):	12✔
226	"""LoadIncrementalFilteredPolicy append a filtered policy from file/database."""
227	if not hasattr(self.adapter, "is_filtered"):	12✔
228	raise ValueError("filtered policies are not supported by this adapter")	×
229
230	self.adapter.load_filtered_policy(self.model, filter)	12✔
231	self.model.print_policy()	12✔
232	if self.auto_build_role_links:	12✔
233	self.build_role_links()	12✔
234
235	def is_filtered(self):	12✔
236	"""returns true if the loaded policy has been filtered."""
237
238	return hasattr(self.adapter, "is_filtered") and self.adapter.is_filtered()	12✔
239
240	def save_policy(self):	12✔
241	if self.is_filtered():	12✔
242	raise RuntimeError("cannot save a filtered policy")	12✔
243
244	self.adapter.save_policy(self.model)	12✔
245
246	if self.watcher:	12✔
247	self.watcher.update()	×
248
249	def enable_enforce(self, enabled=True):	12✔
250	"""changes the enforcing state of Casbin,
251	when Casbin is disabled, all access will be allowed by the Enforce() function.
252	"""
253
254	self.enabled = enabled	×
255
256	def enable_auto_save(self, auto_save):	12✔
257	"""controls whether to save a policy rule automatically to the adapter when it is added or removed."""
258	self.auto_save = auto_save	×
259
260	def enable_auto_build_role_links(self, auto_build_role_links):	12✔
261	"""controls whether to rebuild the role inheritance relations when a role is added or deleted."""
262	self.auto_build_role_links = auto_build_role_links	×
263
264	def build_role_links(self):	12✔
265	"""manually rebuild the role inheritance relations."""
266
267	for rm in self.rm_map.values():	12✔
268	rm.clear()	12✔
269
270	self.model.build_role_links(self.rm_map)	12✔
271
272	def add_named_matching_func(self, ptype, fn):	12✔
273	"""add_named_matching_func add MatchingFunc by ptype RoleManager"""
274	try:	12✔
275	self.rm_map[ptype].add_matching_func(fn)	12✔
276	return True	12✔
277	except:	×
278	return False	×
279
280	def add_named_domain_matching_func(self, ptype, fn):	12✔
281	"""add_named_domain_matching_func add MatchingFunc by ptype to RoleManager"""
282	if ptype in self.rm_map.keys():	×
283	self.rm_map[ptype].add_domain_matching_func(fn)	×
284	return True	×
285
286	return False	×
287
288	def new_enforce_context(self, suffix: str) -> EnforceContext:	12✔
289
290	return EnforceContext(	12✔
291	rtype="r" + suffix,
292	ptype="p" + suffix,
293	etype="e" + suffix,
294	mtype="m" + suffix,
295	)
296
297	def enforce(self, *rvals):	12✔
298	"""decides whether a "subject" can access a "object" with the operation "action",
299	input parameters are usually: (sub, obj, act).
300	"""
301	result, _ = self.enforce_ex(*rvals)	12✔
302	return result	12✔
303
304	def enforce_ex(self, *rvals):	12✔
305	"""decides whether a "subject" can access a "object" with the operation "action",
306	input parameters are usually: (sub, obj, act).
307	return judge result with reason
308	"""
309
310	rtype = "r"	12✔
311	ptype = "p"	12✔
312	etype = "e"	12✔
313	mtype = "m"	12✔
314
315	if not self.enabled:	12✔
316	return [False, []]	×
317
318	functions = self.fm.get_functions()	12✔
319
320	if "g" in self.model.keys():	12✔
321	for key, ast in self.model["g"].items():	12✔
322	rm = ast.rm	12✔
323	functions[key] = generate_g_function(rm)	12✔
324
325	if len(rvals) != 0:	12✔
326	if isinstance(rvals[0], EnforceContext):	12✔
327	enforce_context = rvals[0]	12✔
328	rtype = enforce_context.rtype	12✔
329	ptype = enforce_context.ptype	12✔
330	etype = enforce_context.etype	12✔
331	mtype = enforce_context.mtype	12✔
332	rvals = rvals[1:]	12✔
333
334	if "m" not in self.model.keys():	12✔
335	raise RuntimeError("model is undefined")	×
336
337	if "m" not in self.model["m"].keys():	12✔
338	raise RuntimeError("model is undefined")	×
339
340	r_tokens = self.model["r"][rtype].tokens	12✔
341	p_tokens = self.model["p"][ptype].tokens	12✔
342
343	if len(r_tokens) != len(rvals):	12✔
344	raise RuntimeError("invalid request size")	×
345
346	exp_string = self.model["m"][mtype].value	12✔
347	has_eval = util.has_eval(exp_string)	12✔
348	if not has_eval:	12✔
349	expression = self._get_expression(exp_string, functions)	12✔
350
351	policy_effects = set()	12✔
352
353	r_parameters = dict(zip(r_tokens, rvals))	12✔
354
355	policy_len = len(self.model["p"][ptype].policy)	12✔
356
357	explain_index = -1	12✔
358	if not 0 == policy_len:	12✔
359	for i, pvals in enumerate(self.model["p"][ptype].policy):	12✔
360	if len(p_tokens) != len(pvals):	12✔
361	raise RuntimeError("invalid policy size")	×
362
363	p_parameters = dict(zip(p_tokens, pvals))	12✔
364	parameters = dict(r_parameters, **p_parameters)	12✔
365
366	if util.has_eval(exp_string):	12✔
367	rule_names = util.get_eval_value(exp_string)	12✔
368	rules = [	12✔
369	util.escape_assertion(p_parameters[rule_name])
370	for rule_name in rule_names
371	]
372	exp_with_rule = util.replace_eval(exp_string, rules)	12✔
373	expression = self._get_expression(exp_with_rule, functions)	12✔
374
375	result = expression.eval(parameters)	12✔
376
377	if isinstance(result, bool):	12✔
378	if not result:	12✔
379	policy_effects.add(Effector.INDETERMINATE)	12✔
380	continue	12✔
381	elif isinstance(result, float):	×
382	if 0 == result:	×
383	policy_effects.add(Effector.INDETERMINATE)	×
384	continue	×
385	else:
386	raise RuntimeError("matcher result should be bool, int or float")	×
387
388	p_eft_key = ptype + "_eft"	12✔
389	if p_eft_key in parameters.keys():	12✔
390	eft = parameters[p_eft_key]	12✔
391	if "allow" == eft:	12✔
392	policy_effects.add(Effector.ALLOW)	12✔
393	elif "deny" == eft:	12✔
394	policy_effects.add(Effector.DENY)	12✔
395	else:
396	policy_effects.add(Effector.INDETERMINATE)	12✔
397	else:
398	policy_effects.add(Effector.ALLOW)	12✔
399
400	if (	12✔
401	self.eft.intermediate_effect(policy_effects)
402	!= Effector.INDETERMINATE
403	):
404	explain_index = i	12✔
405	break	12✔
406
407	else:
408	if has_eval:	12✔
409	raise RuntimeError(	×
410	"please make sure rule exists in policy when using eval() in matcher"
411	)
412
413	parameters = r_parameters.copy()	12✔
414
415	for token in self.model["p"][ptype].tokens:	12✔
416	parameters[token] = ""	12✔
417
418	result = expression.eval(parameters)	12✔
419
420	if result:	12✔
421	policy_effects.add(Effector.ALLOW)	12✔
422	else:
423	policy_effects.add(Effector.INDETERMINATE)	12✔
424
425	final_effect = self.eft.final_effect(policy_effects)	12✔
426	result = effect_to_bool(final_effect)	12✔
427
428	# Log request.
429
430	req_str = "Request: "	12✔
431	req_str = req_str + ", ".join([str(v) for v in rvals])	12✔
432
433	req_str = req_str + " ---> %s" % result	12✔
434	if result:	12✔
435	self.logger.info(req_str)	12✔
436	else:
437	# leaving this in error for now, if it's very noise this can be changed to info or debug
438	self.logger.error(req_str)	12✔
439
440	explain_rule = []	12✔
441	if explain_index != -1 and explain_index < policy_len:	12✔
442	explain_rule = self.model["p"][ptype].policy[explain_index]	12✔
443
444	return result, explain_rule	12✔
445
446	@staticmethod	12✔
447	def _get_expression(expr, functions=None):	12✔
448	expr = expr.replace("&&", "and")	12✔
449	expr = expr.replace("\|\|", "or")	12✔
450	expr = expr.replace("!", "not")	12✔
451
452	return SimpleEval(expr, functions)	12✔

casbin / pycasbin / 1262282997

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous