1589226415

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/c9c9073a5cf9614e319fde15056b5e81aa368e9e">c9c9073a5">&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/c9c9073a5cf9614e319fde15056b5e81aa368e9e">c9c9073a5</a><a href="https://github.com/casbin/pycasbin/commit/c9c9073a5cf9614e319fde15056b5e81aa368e9e">&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/c9c9073a5cf9614e319fde15056b5e81aa368e9e&quot;&gt;c9c9073a5&lt;/a&gt;&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/c9c9073a5cf9614e319fde15056b5e81aa368e9e&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 #235: fix: disabled casbin should allow all requests

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

1839 of 2153 relevant lines covered (85.42%)

10.25 hits per line

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

85.35

/casbin/core_enforcer.py

# Copyright 2021 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.

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"]:
                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)

    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)

            self.model.sort_policies_by_subject_hierarchy()

            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 Exception as e:

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

            raise e

    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 [True, []]

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

casbin / pycasbin / 1589226415

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