983693608

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/f96e92451e419b81f2b9e4c39723caad9891fb0c">f96e92451">&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/f96e92451e419b81f2b9e4c39723caad9891fb0c">f96e92451</a><a href="https://github.com/casbin/pycasbin/commit/f96e92451e419b81f2b9e4c39723caad9891fb0c">&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/f96e92451e419b81f2b9e4c39723caad9891fb0c&quot;&gt;f96e92451&lt;/a&gt;&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/f96e92451e419b81f2b9e4c39723caad9891fb0c&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 #172: fix: matching_func works with multiply match

Run Details

4 of 6 new or added lines in 1 file covered. (66.67%)

1577 of 1880 relevant lines covered (83.88%)

10.07 hits per line

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

90.24

/casbin/rbac/default_role_manager/role_manager.py

import logging

from casbin.rbac import RoleManager


class RoleManager(RoleManager):
    """provides a default implementation for the RoleManager interface"""

    all_roles = dict()
    max_hierarchy_level = 0

    def __init__(self, max_hierarchy_level=10):
        self.logger = logging.getLogger(__name__)
        self.all_roles = dict()
        self.max_hierarchy_level = max_hierarchy_level
        self.matching_func = None
        self.domain_matching_func = None
        self.has_pattern = False
        self.has_domain_pattern = False

    def add_matching_func(self, fn=None):
        self.has_pattern = True
        self.matching_func = fn

    def add_domain_matching_func(self, fn=None):
        self.has_domain_pattern = True
        self.domain_matching_func = fn

    def has_role(self, role):

        if not self.has_pattern and not self.has_domain_pattern:
            return role in self.all_roles.values()

        for known_role in list(self.all_roles.values()):
            if self.has_pattern:
                if not self.matching_func(role.name, known_role.name):
                    continue
            else:
                if not role.name == known_role.name:
                    continue

            if self.has_domain_pattern:
                if not self.domain_matching_func(role.domain, known_role.domain):
                    continue
            else:
                if not role.domain == known_role.domain:
                    continue
            return True

    def create_role(self, name, domain=""):
        role = Role(name, domain)
        if domain:
            key = domain + "::" + name
        else:
            key = name

        if key not in self.all_roles.keys():
            self.all_roles[key] = role

        return self.all_roles[key]

    def clear(self):
        self.all_roles.clear()

    def add_link(self, name1, name2, *domain):
        if len(domain) > 1:
            raise RuntimeError("error: domain should be 1 parameter")
        elif len(domain) == 1:
            domain = domain[0]
        else:
            domain = ""

        role1 = self.create_role(name1, domain)
        role2 = self.create_role(name2, domain)
        role1.add_role(role2)

        if self.has_pattern:
            for role in self.all_roles.values():
                if self.has_domain_pattern:
                    if not self.domain_matching_func(domain, role.domain):
                        continue
                else:
                    if domain != role.domain:
                        continue

                def duplicate_judge():
                    return role1.name != role.name and role2.name != role.name

                if (
                    match_error_handler(self.matching_func, role.name, role1.name)
                    or match_error_handler(self.matching_func, role1.name, role.name)
                    and duplicate_judge()
                ):
                    self.all_roles[role.get_key()].add_role(role1)

                if (
                    match_error_handler(self.matching_func, role.name, role2.name)
                    or match_error_handler(self.matching_func, role2.name, role.name)
                    and duplicate_judge()
                ):
                    self.all_roles[role2.get_key()].add_role(role)

    def delete_link(self, name1, name2, *domain):
        role1, role2 = two_role_domain_wrapper(self, name1, name2, domain)

        if not self.has_role(role1) or not self.has_role(role2):
            raise RuntimeError("error: name1 or name2 does not exist")

        role1.delete_role(role2)

    def has_link(self, name1, name2, *domain):
        if len(domain) > 1:
            raise RuntimeError("error: domain should be 1 parameter")
        elif len(domain) == 1:
            domain = domain[0]
        else:
            domain = ""

        role1, role2 = two_role_domain_wrapper(self, name1, name2, domain)

        if role1 == role2:
            return True

        if not self.has_role(role1) or not self.has_role(role2):
            return False

        if not self.has_pattern and not self.has_domain_pattern:
            return role1.has_role(role2, self.max_hierarchy_level, None, None)

        # Here is has_pattern logic.
        for role in self.all_roles.values():
            if self.has_domain_pattern:
                if not self.domain_matching_func(domain, role.domain):
                    continue
            else:
                if role.domain != domain:
                    continue

            def role_judge():
                if role.has_role(
                    role2,
                    self.max_hierarchy_level,
                    self.matching_func,
                    self.domain_matching_func,
                ):
                    return True
                return False

            if self.has_pattern:
                if self.matching_func(role1.name, role.name):
                    if role_judge():
                        return True
                    continue
            else:
                if role1.name == role.name:
                    if role_judge():
                        return True
                    continue
        return False

    def get_roles(self, name, *domain):
        """
        gets the roles that a subject inherits.
        domain is a prefix to the roles.
        """
        if len(domain) > 1:
            raise RuntimeError("error: domain should be 1 parameter")
        elif len(domain) == 1:
            domain = domain[0]
        else:
            domain = ""

        role = role_domain_wrapper(self, name, domain)

        if not self.has_role(role):
            return []

        roles = self.create_role(name, domain).get_roles()

        return roles

    def get_users(self, name, *domain):
        """
        gets the users that inherits a subject.
        domain is an unreferenced parameter here, may be used in other implementations.
        """
        target_role = role_domain_wrapper(self, name, domain)

        if not self.has_role(target_role):
            return []

        roles = []
        for role in self.all_roles.values():
            if role.has_direct_role(target_role):
                roles.append(role.name)

        return roles

    def print_roles(self):
        line = []
        for role in self.all_roles.values():
            text = role.to_string()
            if text:
                line.append(text)
        self.logger.info(", ".join(line))


class Role:
    """represents the data structure for a role in RBAC."""

    def __init__(self, name: str, domain: str = ""):
        self.name = name
        self.roles = []
        self.domain = domain

    def __eq__(self, other: "Role"):
        return (
            type(other) == type(self)
            and self.name == other.name
            and self.domain == other.domain
        )

    def __hash__(self):
        return hash(self.name + "::" + self.domain)

    def get_key(self):
        if self.domain:
            return self.domain + "::" + self.name
        return self.name

    def add_role(self, role: "Role"):
        if role in self.roles:
            return
        self.roles.append(role)

    def delete_role(self, role: "Role"):
        if role in self.roles:
            self.roles.remove(role)

    def has_role(
        self,
        role: "Role",
        hierarchy_level: int,
        matching_func=None,
        domain_matching_func=None,
    ):

        if self.has_direct_role(role, matching_func, domain_matching_func):
            return True
        if hierarchy_level <= 0:
            return False

        for knownRole in self.roles:
            if knownRole.has_role(
                role, hierarchy_level - 1, matching_func, domain_matching_func
            ):
                return True

        return False

    def has_direct_role(
        self, role: "Role", matching_func=None, domain_matching_func=None
    ):
        for known_role in self.roles:
            if matching_func:
                if not matching_func(role.name, known_role.name):
                    continue
            else:
                if not role.name == known_role.name:
                    continue

            if domain_matching_func:
                if not domain_matching_func(role.domain, known_role.domain):
                    continue
            else:
                if not role.domain == known_role.domain:
                    continue
            return True
        return False

    def to_string(self):
        if len(self.roles) == 0:
            return ""

        names = ", ".join(self.get_roles())

        if len(self.roles) == 1:
            return self.name + " < " + names
        else:
            return self.name + " < (" + names + ")"

    def get_roles(self):
        roles = []
        for role in self.roles:
            roles.append(role.name)

        return roles


def role_domain_wrapper(obj, name, domain):
    if type(domain) != str:
        if not domain or len(domain) == 0:
            domain = ""
        elif len(domain) == 1:
            domain = domain[0]
        elif len(domain) > 1:
            raise RuntimeError("error: domain should be 1 parameter")

    role = Role(name, domain)

    if not obj.has_role(role):
        return role
    return obj.create_role(name, domain)


def two_role_domain_wrapper(obj, name1, name2, domain):
    return role_domain_wrapper(obj, name1, domain), role_domain_wrapper(
        obj, name2, domain
    )


def match_error_handler(fn, key1, key2):
    try:
        return fn(key1, key2)
    except:
        return False

1	import logging	12✔
2
3	from casbin.rbac import RoleManager	12✔
4
5
6	class RoleManager(RoleManager):	12✔
7	"""provides a default implementation for the RoleManager interface"""
8
9	all_roles = dict()	12✔
10	max_hierarchy_level = 0	12✔
11
12	def __init__(self, max_hierarchy_level=10):	12✔
13	self.logger = logging.getLogger(__name__)	12✔
14	self.all_roles = dict()	12✔
15	self.max_hierarchy_level = max_hierarchy_level	12✔
16	self.matching_func = None	12✔
17	self.domain_matching_func = None	12✔
18	self.has_pattern = False	12✔
19	self.has_domain_pattern = False	12✔
20
21	def add_matching_func(self, fn=None):	12✔
22	self.has_pattern = True	12✔
23	self.matching_func = fn	12✔
24
25	def add_domain_matching_func(self, fn=None):	12✔
26	self.has_domain_pattern = True	12✔
27	self.domain_matching_func = fn	12✔
28
29	def has_role(self, role):	12✔
30
31	if not self.has_pattern and not self.has_domain_pattern:	12✔
32	return role in self.all_roles.values()	12✔
33
34	for known_role in list(self.all_roles.values()):	12✔
35	if self.has_pattern:	12✔
36	if not self.matching_func(role.name, known_role.name):	12✔
37	continue	12✔
38	else:
39	if not role.name == known_role.name:	12✔
40	continue	12✔
41
42	if self.has_domain_pattern:	12✔
43	if not self.domain_matching_func(role.domain, known_role.domain):	12✔
44	continue	×
45	else:
46	if not role.domain == known_role.domain:	12✔
47	continue	×
48	return True	12✔
49
50	def create_role(self, name, domain=""):	12✔
51	role = Role(name, domain)	12✔
52	if domain:	12✔
53	key = domain + "::" + name	12✔
54	else:
55	key = name	12✔
56
57	if key not in self.all_roles.keys():	12✔
58	self.all_roles[key] = role	12✔
59
60	return self.all_roles[key]	12✔
61
62	def clear(self):	12✔
63	self.all_roles.clear()	12✔
64
65	def add_link(self, name1, name2, *domain):	12✔
66	if len(domain) > 1:	12✔
67	raise RuntimeError("error: domain should be 1 parameter")	×
68	elif len(domain) == 1:	12✔
69	domain = domain[0]	12✔
70	else:
71	domain = ""	12✔
72
73	role1 = self.create_role(name1, domain)	12✔
74	role2 = self.create_role(name2, domain)	12✔
75	role1.add_role(role2)	12✔
76
77	if self.has_pattern:	12✔
78	for role in self.all_roles.values():	12✔
79	if self.has_domain_pattern:	12✔
80	if not self.domain_matching_func(domain, role.domain):	×
81	continue	×
82	else:
83	if domain != role.domain:	12✔
84	continue	×
85
86	def duplicate_judge():	12✔
87	return role1.name != role.name and role2.name != role.name	12✔
88
89	if (	12✔
90	match_error_handler(self.matching_func, role.name, role1.name)
91	or match_error_handler(self.matching_func, role1.name, role.name)
92	and duplicate_judge()
93	):
94	self.all_roles[role.get_key()].add_role(role1)	12✔
95
96	if (	12✔
97	match_error_handler(self.matching_func, role.name, role2.name)
98	or match_error_handler(self.matching_func, role2.name, role.name)
99	and duplicate_judge()
100	):
101	self.all_roles[role2.get_key()].add_role(role)	12✔
102
103	def delete_link(self, name1, name2, *domain):	12✔
104	role1, role2 = two_role_domain_wrapper(self, name1, name2, domain)	12✔
105
106	if not self.has_role(role1) or not self.has_role(role2):	12✔
107	raise RuntimeError("error: name1 or name2 does not exist")	×
108
109	role1.delete_role(role2)	12✔
110
111	def has_link(self, name1, name2, *domain):	12✔
112	if len(domain) > 1:	12✔
113	raise RuntimeError("error: domain should be 1 parameter")	×
114	elif len(domain) == 1:	12✔
115	domain = domain[0]	12✔
116	else:
117	domain = ""	12✔
118
119	role1, role2 = two_role_domain_wrapper(self, name1, name2, domain)	12✔
120
121	if role1 == role2:	12✔
122	return True	12✔
123
124	if not self.has_role(role1) or not self.has_role(role2):	12✔
125	return False	12✔
126
127	if not self.has_pattern and not self.has_domain_pattern:	12✔
128	return role1.has_role(role2, self.max_hierarchy_level, None, None)	12✔
129
130	# Here is has_pattern logic.
131	for role in self.all_roles.values():	12✔
132	if self.has_domain_pattern:	12✔
133	if not self.domain_matching_func(domain, role.domain):	12✔
134	continue	×
135	else:
136	if role.domain != domain:	12✔
137	continue	×
138
139	def role_judge():	12✔
140	if role.has_role(	12✔
141	role2,
142	self.max_hierarchy_level,
143	self.matching_func,
144	self.domain_matching_func,
145	):
146	return True	12✔
147	return False	12✔
148
149	if self.has_pattern:	12✔
150	if self.matching_func(role1.name, role.name):	12✔
151	if role_judge():	12✔
152	return True	12✔
NEW 153	continue	×
154	else:
155	if role1.name == role.name:	12✔
156	if role_judge():	12✔
157	return True	12✔
NEW 158	continue	×
159	return False	12✔
160
161	def get_roles(self, name, *domain):	12✔
162	"""
163	gets the roles that a subject inherits.
164	domain is a prefix to the roles.
165	"""
166	if len(domain) > 1:	12✔
167	raise RuntimeError("error: domain should be 1 parameter")	×
168	elif len(domain) == 1:	12✔
169	domain = domain[0]	12✔
170	else:
171	domain = ""	12✔
172
173	role = role_domain_wrapper(self, name, domain)	12✔
174
175	if not self.has_role(role):	12✔
176	return []	12✔
177
178	roles = self.create_role(name, domain).get_roles()	12✔
179
180	return roles	12✔
181
182	def get_users(self, name, *domain):	12✔
183	"""
184	gets the users that inherits a subject.
185	domain is an unreferenced parameter here, may be used in other implementations.
186	"""
187	target_role = role_domain_wrapper(self, name, domain)	12✔
188
189	if not self.has_role(target_role):	12✔
190	return []	12✔
191
192	roles = []	12✔
193	for role in self.all_roles.values():	12✔
194	if role.has_direct_role(target_role):	12✔
195	roles.append(role.name)	12✔
196
197	return roles	12✔
198
199	def print_roles(self):	12✔
200	line = []	12✔
201	for role in self.all_roles.values():	12✔
202	text = role.to_string()	12✔
203	if text:	12✔
204	line.append(text)	12✔
205	self.logger.info(", ".join(line))	12✔
206
207
208	class Role:	12✔
209	"""represents the data structure for a role in RBAC."""
210
211	def __init__(self, name: str, domain: str = ""):	12✔
212	self.name = name	12✔
213	self.roles = []	12✔
214	self.domain = domain	12✔
215
216	def __eq__(self, other: "Role"):	12✔
217	return (	12✔
218	type(other) == type(self)
219	and self.name == other.name
220	and self.domain == other.domain
221	)
222
223	def __hash__(self):	12✔
224	return hash(self.name + "::" + self.domain)	×
225
226	def get_key(self):	12✔
227	if self.domain:	12✔
228	return self.domain + "::" + self.name	×
229	return self.name	12✔
230
231	def add_role(self, role: "Role"):	12✔
232	if role in self.roles:	12✔
233	return	12✔
234	self.roles.append(role)	12✔
235
236	def delete_role(self, role: "Role"):	12✔
237	if role in self.roles:	12✔
238	self.roles.remove(role)	12✔
239
240	def has_role(	12✔
241	self,
242	role: "Role",
243	hierarchy_level: int,
244	matching_func=None,
245	domain_matching_func=None,
246	):
247
248	if self.has_direct_role(role, matching_func, domain_matching_func):	12✔
249	return True	12✔
250	if hierarchy_level <= 0:	12✔
251	return False	12✔
252
253	for knownRole in self.roles:	12✔
254	if knownRole.has_role(	12✔
255	role, hierarchy_level - 1, matching_func, domain_matching_func
256	):
257	return True	12✔
258
259	return False	12✔
260
261	def has_direct_role(	12✔
262	self, role: "Role", matching_func=None, domain_matching_func=None
263	):
264	for known_role in self.roles:	12✔
265	if matching_func:	12✔
266	if not matching_func(role.name, known_role.name):	12✔
267	continue	12✔
268	else:
269	if not role.name == known_role.name:	12✔
270	continue	12✔
271
272	if domain_matching_func:	12✔
273	if not domain_matching_func(role.domain, known_role.domain):	12✔
274	continue	×
275	else:
276	if not role.domain == known_role.domain:	12✔
277	continue	12✔
278	return True	12✔
279	return False	12✔
280
281	def to_string(self):	12✔
282	if len(self.roles) == 0:	12✔
283	return ""	12✔
284
285	names = ", ".join(self.get_roles())	12✔
286
287	if len(self.roles) == 1:	12✔
288	return self.name + " < " + names	12✔
289	else:
290	return self.name + " < (" + names + ")"	12✔
291
292	def get_roles(self):	12✔
293	roles = []	12✔
294	for role in self.roles:	12✔
295	roles.append(role.name)	12✔
296
297	return roles	12✔
298
299
300	def role_domain_wrapper(obj, name, domain):	12✔
301	if type(domain) != str:	12✔
302	if not domain or len(domain) == 0:	12✔
303	domain = ""	12✔
304	elif len(domain) == 1:	12✔
305	domain = domain[0]	12✔
306	elif len(domain) > 1:	×
307	raise RuntimeError("error: domain should be 1 parameter")	×
308
309	role = Role(name, domain)	12✔
310
311	if not obj.has_role(role):	12✔
312	return role	12✔
313	return obj.create_role(name, domain)	12✔
314
315
316	def two_role_domain_wrapper(obj, name1, name2, domain):	12✔
317	return role_domain_wrapper(obj, name1, domain), role_domain_wrapper(	12✔
318	obj, name2, domain
319	)
320
321
322	def match_error_handler(fn, key1, key2):	12✔
323	try:	12✔
324	return fn(key1, key2)	12✔
325	except:	×
326	return False	×

casbin / pycasbin / 983693608

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