2224317161

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/079bfcf82a10d06d57df17f121552f01b27f0b6d">079bfcf82<a href="https://github.com/casbin/pycasbin/commit/079bfcf82a10d06d57df17f121552f01b27f0b6d">&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/079bfcf82a10d06d57df17f121552f01b27f0b6d&quot;&gt;079bfcf82&lt;/a&gt;&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/079bfcf82a10d06d57df17f121552f01b27f0b6d&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/079bfcf82a10d06d57df17f121552f01&lt;/a&gt;b27f0b6d">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/<a class="double-link" href="https://github.com/casbin/pycasbin/commit/326a2049b23080d38359bfe4d78fe7c6d9bed00d">326a2049b</a><a href="https://github.com/casbin/pycasbin/commit/079bfcf82a10d06d57df17f121552f01b27f0b6d">&amp;quot;&amp;gt;326a2049b&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;326a2049b&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/079bfcf82a10d06d57df17f121552f01b27f0&lt;/a&gt;b6d&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/<a class="double-link" href="https://github.com/casbin/pycasbin/commit/f75702401">f75702401">f75702401</a>

Pull Request Pull Request #249: feat: update_filtered_policies & lint config file

Run Details

36 of 44 new or added lines in 11 files covered. (81.82%)

1864 of 2183 relevant lines covered (85.39%)

10.25 hits per line

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

84.57

/casbin/model/policy.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

DEFAULT_SEP = ","


class Policy:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.model = {}

    def __getitem__(self, item):
        return self.model.get(item)

    def __setitem__(self, key, value):
        self.model[key] = value

    def keys(self):
        return self.model.keys()

    def values(self):
        return self.model.values()

    def items(self):
        return self.model.items()

    def build_role_links(self, rm_map):
        """initializes the roles in RBAC."""

        if "g" not in self.keys():
            return

        for ptype, ast in self["g"].items():
            rm = rm_map[ptype]
            ast.build_role_links(rm)

    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 print_policy(self):
        """Log using info"""

        self.logger.info("Policy:")
        for sec in ["p", "g"]:
            if sec not in self.keys():
                continue

            for key, ast in self[sec].items():
                self.logger.info("{} : {} : {}".format(key, ast.value, ast.policy))

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

        for sec in ["p", "g"]:
            if sec not in self.keys():
                continue

            for key in self[sec].keys():
                self[sec][key].policy = []

    def get_policy(self, sec, ptype):
        """gets all rules in a policy."""

        return self[sec][ptype].policy

    def get_filtered_policy(self, sec, ptype, field_index, *field_values):
        """gets rules based on field filters from a policy."""
        return [
            rule
            for rule in self[sec][ptype].policy
            if all(
                (callable(value) and value(rule[field_index + i]))
                or (value == "" or rule[field_index + i] == value)
                for i, value in enumerate(field_values)
            )
        ]

    def has_policy(self, sec, ptype, rule):
        """determines whether a model has the specified policy rule."""
        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False

        return rule in self[sec][ptype].policy

    def add_policy(self, sec, ptype, rule):
        """adds a policy rule to the model."""
        assertion = self[sec][ptype]
        if not self.has_policy(sec, ptype, rule):
            assertion.policy.append(rule)
        else:
            return False

        if sec == "p" and assertion.priority_index >= 0:
            try:
                idx_insert = int(rule[assertion.priority_index])

                i = len(assertion.policy) - 1
                for i in range(i, 0, -1):

                    try:
                        idx = int(assertion.policy[i - 1][assertion.priority_index])
                    except Exception as e:
                        print(e)

                    if idx > idx_insert:
                        assertion.policy[i] = assertion.policy[i - 1]
                    else:
                        break

                assertion.policy[i] = rule
                assertion.policy_map[DEFAULT_SEP.join(rule)] = i

            except Exception as e:
                print(e)

        assertion.policy_map[DEFAULT_SEP.join(rule)] = len(assertion.policy) - 1
        return True

    def add_policies(self, sec, ptype, rules):
        """adds policy rules to the model."""

        for rule in rules:
            if self.has_policy(sec, ptype, rule):
                return False

        for rule in rules:
            self[sec][ptype].policy.append(rule)

        return True

    def update_policy(self, sec, ptype, old_rule, new_rule):
        """update a policy rule from the model."""

        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False

        ast = self[sec][ptype]

        if old_rule in ast.policy:
            rule_index = ast.policy.index(old_rule)
        else:
            return False

        if "p_priority" in ast.tokens:
            priority_index = ast.tokens.index("p_priority")
            if old_rule[priority_index] == new_rule[priority_index]:
                ast.policy[rule_index] = new_rule
            else:
                raise Exception("New rule should have the same priority with old rule.")
        else:
            ast.policy[rule_index] = new_rule

        return True

    def update_policies(self, sec, ptype, old_rules, new_rules):
        """update policy rules from the model."""

        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False
        if len(old_rules) != len(new_rules):
            return False

        ast = self[sec][ptype]
        old_rules_index = []

        for old_rule in old_rules:
            if old_rule in ast.policy:
                old_rules_index.append(ast.policy.index(old_rule))
            else:
                return False

        if "p_priority" in ast.tokens:
            priority_index = ast.tokens.index("p_priority")
            for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):
                if old_rule[priority_index] == new_rule[priority_index]:
                    ast.policy[idx] = new_rule
                else:
                    raise Exception(
                        "New rule should have the same priority with old rule."
                    )
        else:
            for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):
                ast.policy[idx] = new_rule

        return True

    def remove_policy(self, sec, ptype, rule):
        """removes a policy rule from the model."""
        if not self.has_policy(sec, ptype, rule):
            return False

        self[sec][ptype].policy.remove(rule)

        return rule not in self[sec][ptype].policy

    def remove_policies(self, sec, ptype, rules):
        """RemovePolicies removes policy rules from the model."""

        for rule in rules:
            if not self.has_policy(sec, ptype, rule):
                return False
            self[sec][ptype].policy.remove(rule)
            if rule in self[sec][ptype].policy:
                return False

        return True

    def remove_policies_with_effected(self, sec, ptype, rules):
        effected = []
        for rule in rules:
            if self.has_policy(sec, ptype, rule):
                effected.append(rule)
                self.remove_policy(sec, ptype, rule)

        return effected

    def remove_filtered_policy_returns_effects(
        self, sec, ptype, field_index, *field_values
    ):
        """
        remove_filtered_policy_returns_effects removes policy rules based on field filters from the model.
        """
        tmp = []
        effects = []

        if len(field_values) == 0:
            return []
        if sec not in self.keys():
            return []
        if ptype not in self[sec]:
            return []

        for rule in self[sec][ptype].policy:
            if all(
                value == "" or rule[field_index + i] == value
                for i, value in enumerate(field_values)
            ):
                effects.append(rule)
            else:
                tmp.append(rule)

        self[sec][ptype].policy = tmp

        return effects

    def remove_filtered_policy(self, sec, ptype, field_index, *field_values):
        """removes policy rules based on field filters from the model."""
        tmp = []
        res = False

        if sec not in self.keys():
            return res
        if ptype not in self[sec]:
            return res

        for rule in self[sec][ptype].policy:
            if all(
                value == "" or rule[field_index + i] == value
                for i, value in enumerate(field_values)
            ):
                res = True
            else:
                tmp.append(rule)

        self[sec][ptype].policy = tmp

        return res

    def get_values_for_field_in_policy(self, sec, ptype, field_index):
        """gets all values for a field for all rules in a policy, duplicated values are removed."""
        values = []
        if sec not in self.keys():
            return values
        if ptype not in self[sec]:
            return values

        for rule in self[sec][ptype].policy:
            value = rule[field_index]
            if value not in values:
                values.append(value)

        return values

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	12✔
16
17	DEFAULT_SEP = ","	12✔
18
19
20	class Policy:	12✔
21	def __init__(self):	12✔
22	self.logger = logging.getLogger(__name__)	12✔
23	self.model = {}	12✔
24
25	def __getitem__(self, item):	12✔
26	return self.model.get(item)	12✔
27
28	def __setitem__(self, key, value):	12✔
29	self.model[key] = value	12✔
30
31	def keys(self):	12✔
32	return self.model.keys()	12✔
33
34	def values(self):	12✔
35	return self.model.values()	×
36
37	def items(self):	12✔
38	return self.model.items()	12✔
39
40	def build_role_links(self, rm_map):	12✔
41	"""initializes the roles in RBAC."""
42
43	if "g" not in self.keys():	12✔
44	return	12✔
45
46	for ptype, ast in self["g"].items():	12✔
47	rm = rm_map[ptype]	12✔
48	ast.build_role_links(rm)	12✔
49
50	def build_incremental_role_links(self, rm, op, sec, ptype, rules):	12✔
51	if sec == "g":	12✔
52	self[sec].get(ptype).build_incremental_role_links(rm, op, rules)	12✔
53
54	def print_policy(self):	12✔
55	"""Log using info"""
56
57	self.logger.info("Policy:")	12✔
58	for sec in ["p", "g"]:	12✔
59	if sec not in self.keys():	12✔
60	continue	12✔
61
62	for key, ast in self[sec].items():	12✔
63	self.logger.info("{} : {} : {}".format(key, ast.value, ast.policy))	12✔
64
65	def clear_policy(self):	12✔
66	"""clears all current policy."""
67
68	for sec in ["p", "g"]:	12✔
69	if sec not in self.keys():	12✔
70	continue	12✔
71
72	for key in self[sec].keys():	12✔
73	self[sec][key].policy = []	12✔
74
75	def get_policy(self, sec, ptype):	12✔
76	"""gets all rules in a policy."""
77
78	return self[sec][ptype].policy	12✔
79
80	def get_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
81	"""gets rules based on field filters from a policy."""
82	return [	12✔
83	rule
84	for rule in self[sec][ptype].policy
85	if all(
86	(callable(value) and value(rule[field_index + i]))
87	or (value == "" or rule[field_index + i] == value)
88	for i, value in enumerate(field_values)
89	)
90	]
91
92	def has_policy(self, sec, ptype, rule):	12✔
93	"""determines whether a model has the specified policy rule."""
94	if sec not in self.keys():	12✔
95	return False	×
96	if ptype not in self[sec]:	12✔
97	return False	×
98
99	return rule in self[sec][ptype].policy	12✔
100
101	def add_policy(self, sec, ptype, rule):	12✔
102	"""adds a policy rule to the model."""
103	assertion = self[sec][ptype]	12✔
104	if not self.has_policy(sec, ptype, rule):	12✔
105	assertion.policy.append(rule)	12✔
106	else:
107	return False	×
108
109	if sec == "p" and assertion.priority_index >= 0:	12✔
110	try:	12✔
111	idx_insert = int(rule[assertion.priority_index])	12✔
112
113	i = len(assertion.policy) - 1	12✔
114	for i in range(i, 0, -1):	12✔
115
116	try:	12✔
117	idx = int(assertion.policy[i - 1][assertion.priority_index])	12✔
118	except Exception as e:	×
119	print(e)	×
120
121	if idx > idx_insert:	12✔
122	assertion.policy[i] = assertion.policy[i - 1]	12✔
123	else:
124	break	12✔
125
126	assertion.policy[i] = rule	12✔
127	assertion.policy_map[DEFAULT_SEP.join(rule)] = i	12✔
128
129	except Exception as e:	×
130	print(e)	×
131
132	assertion.policy_map[DEFAULT_SEP.join(rule)] = len(assertion.policy) - 1	12✔
133	return True	12✔
134
135	def add_policies(self, sec, ptype, rules):	12✔
136	"""adds policy rules to the model."""
137
138	for rule in rules:	12✔
139	if self.has_policy(sec, ptype, rule):	12✔
140	return False	×
141
142	for rule in rules:	12✔
143	self[sec][ptype].policy.append(rule)	12✔
144
145	return True	12✔
146
147	def update_policy(self, sec, ptype, old_rule, new_rule):	12✔
148	"""update a policy rule from the model."""
149
150	if sec not in self.keys():	12✔
151	return False	×
152	if ptype not in self[sec]:	12✔
153	return False	×
154
155	ast = self[sec][ptype]	12✔
156
157	if old_rule in ast.policy:	12✔
158	rule_index = ast.policy.index(old_rule)	12✔
159	else:
160	return False	×
161
162	if "p_priority" in ast.tokens:	12✔
163	priority_index = ast.tokens.index("p_priority")	12✔
164	if old_rule[priority_index] == new_rule[priority_index]:	12✔
165	ast.policy[rule_index] = new_rule	12✔
166	else:
167	raise Exception("New rule should have the same priority with old rule.")	×
168	else:
169	ast.policy[rule_index] = new_rule	12✔
170
171	return True	12✔
172
173	def update_policies(self, sec, ptype, old_rules, new_rules):	12✔
174	"""update policy rules from the model."""
175
176	if sec not in self.keys():	12✔
177	return False	×
178	if ptype not in self[sec]:	12✔
179	return False	×
180	if len(old_rules) != len(new_rules):	12✔
181	return False	×
182
183	ast = self[sec][ptype]	12✔
184	old_rules_index = []	12✔
185
186	for old_rule in old_rules:	12✔
187	if old_rule in ast.policy:	12✔
188	old_rules_index.append(ast.policy.index(old_rule))	12✔
189	else:
190	return False	×
191
192	if "p_priority" in ast.tokens:	12✔
193	priority_index = ast.tokens.index("p_priority")	12✔
194	for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):	12✔
195	if old_rule[priority_index] == new_rule[priority_index]:	12✔
196	ast.policy[idx] = new_rule	12✔
197	else:
NEW 198	raise Exception(	×
199	"New rule should have the same priority with old rule."
200	)
201	else:
202	for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):	12✔
203	ast.policy[idx] = new_rule	12✔
204
205	return True	12✔
206
207	def remove_policy(self, sec, ptype, rule):	12✔
208	"""removes a policy rule from the model."""
209	if not self.has_policy(sec, ptype, rule):	12✔
210	return False	12✔
211
212	self[sec][ptype].policy.remove(rule)	12✔
213
214	return rule not in self[sec][ptype].policy	12✔
215
216	def remove_policies(self, sec, ptype, rules):	12✔
217	"""RemovePolicies removes policy rules from the model."""
218
219	for rule in rules:	12✔
220	if not self.has_policy(sec, ptype, rule):	12✔
221	return False	×
222	self[sec][ptype].policy.remove(rule)	12✔
223	if rule in self[sec][ptype].policy:	12✔
224	return False	×
225
226	return True	12✔
227
228	def remove_policies_with_effected(self, sec, ptype, rules):	12✔
229	effected = []	12✔
230	for rule in rules:	12✔
231	if self.has_policy(sec, ptype, rule):	12✔
232	effected.append(rule)	12✔
233	self.remove_policy(sec, ptype, rule)	12✔
234
235	return effected	12✔
236
237	def remove_filtered_policy_returns_effects(	12✔
238	self, sec, ptype, field_index, *field_values
239	):
240	"""
241	remove_filtered_policy_returns_effects removes policy rules based on field filters from the model.
242	"""
243	tmp = []	12✔
244	effects = []	12✔
245
246	if len(field_values) == 0:	12✔
247	return []	×
248	if sec not in self.keys():	12✔
249	return []	×
250	if ptype not in self[sec]:	12✔
251	return []	×
252
253	for rule in self[sec][ptype].policy:	12✔
254	if all(	12✔
255	value == "" or rule[field_index + i] == value
256	for i, value in enumerate(field_values)
257	):
258	effects.append(rule)	12✔
259	else:
260	tmp.append(rule)	12✔
261
262	self[sec][ptype].policy = tmp	12✔
263
264	return effects	12✔
265
266	def remove_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
267	"""removes policy rules based on field filters from the model."""
268	tmp = []	12✔
269	res = False	12✔
270
271	if sec not in self.keys():	12✔
272	return res	×
273	if ptype not in self[sec]:	12✔
274	return res	×
275
276	for rule in self[sec][ptype].policy:	12✔
277	if all(	12✔
278	value == "" or rule[field_index + i] == value
279	for i, value in enumerate(field_values)
280	):
281	res = True	12✔
282	else:
283	tmp.append(rule)	12✔
284
285	self[sec][ptype].policy = tmp	12✔
286
287	return res	12✔
288
289	def get_values_for_field_in_policy(self, sec, ptype, field_index):	12✔
290	"""gets all values for a field for all rules in a policy, duplicated values are removed."""
291	values = []	12✔
292	if sec not in self.keys():	12✔
293	return values	×
294	if ptype not in self[sec]:	12✔
295	return values	×
296
297	for rule in self[sec][ptype].policy:	12✔
298	value = rule[field_index]	12✔
299	if value not in values:	12✔
300	values.append(value)	12✔
301
302	return values	12✔

casbin / pycasbin / 2224317161

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