entrouvert
/
veridic
Archived
17
0
Fork 0
Code Issues Pull Requests Releases Activity
This repository has been archived on 2023-02-21. You can view files and clone it, but cannot push or open issues or pull requests.
veridic/acs/core.py

3724 lines
131 KiB
Python
Raw Blame History

'''
VERIDIC - Towards a centralized access control system
Copyright (C) 2011 Mikael Ates
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
'''
import logging
import re
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned
from django.contrib.contenttypes.models import ContentType
from django.contrib.auth.models import User
from django.db import transaction
from models import UserAlias, Role, AcsObject, View, Action, Activity, \
Namespace, AcsPermission, AcsAbacPermission, Policy
#from signals import attributes_call
logger = logging.getLogger('acs')
'''
Core decision functions
'''
def isAuthorizedRBAC0(who, what, how):
if not who or not what or not how:
return None
p = None
t_who = ContentType.objects.get_for_model(who)
t_what = ContentType.objects.get_for_model(what)
t_how = ContentType.objects.get_for_model(how)
try:
p = AcsPermission.objects.get(content_type_who__pk=t_who.id,
object_id_who=who.id,
content_type_what__pk=t_what.id,
object_id_what=what.id,
content_type_how__pk=t_how.id,
object_id_how=how.id,
)
except MultipleObjectsReturned:
raise Exception('This should never happen')
except ObjectDoesNotExist:
pass
if p:
'''Access granted'''
return p
'''We look at the graph role but making
a recursive research is not relevant because:
- we have to prevent loops
- we don't want an exhaustive research
(we want to stop the research at the first permission found)
Then, we try role one by one in the subtree
and stack the roles already processed to prevent loops.
'''
'''Stack or roles to check'''
r = []
'''List of roles already checked'''
control = []
'''Look at the roles of who'''
if isinstance(who, UserAlias):
roles = Role.objects.filter(users__id=who.id)
'''Init stack'''
for role in roles:
r.append(role)
elif isinstance(who, Role):
r.append(who)
else:
logger.error(('isAuthorizedRBAC0: Unknown type for what: %s')
% who.__class__.__name__)
return None
'''Limit the number of roles to check'''
if not settings.ROLE_GRAPH_LIMIT:
limit = 0
logger.warning('isAuthorizedRBAC0: Not Limit to research')
else:
limit = settings.ROLE_GRAPH_LIMIT
#logger.info('isAuthorizedRBAC0: Limit is at %s' %limit)
t_who = ContentType.objects.get_for_model(Role)
i = 0
while r:
for role in r:
try:
p = AcsPermission.objects.get(content_type_who__pk=t_who.id,
object_id_who=role.id,
content_type_what__pk=t_what.id,
object_id_what=what.id,
content_type_how__pk=t_how.id,
object_id_how=how.id,
)
except MultipleObjectsReturned:
raise Exception('This should never happen')
except ObjectDoesNotExist:
pass
if p:
'''Access granted'''
return p
i = i + 1
if i == limit and limit != 0:
logger.warning(('isAuthorizedRBAC0: Limit of %s reached')
% limit)
return None
tmp = []
for role in r:
control.append(role)
for it in role.roles.all():
if it not in control:
'''Avoid loop'''
tmp.append(it)
r = tmp
'''No permission found'''
return None
def isAuthorizedRBAC1(who, what, how):
if not who or not what or not how:
return None
p = None
p = isAuthorizedRBAC0(who, what, how)
if p:
return p
a = []
control = []
if isinstance(how, Action):
activities = Activity.objects.filter(actions__id=how.id)
for activity in activities:
a.append(activity)
elif isinstance(how, Activity):
a.append(how)
else:
logger.error(('isAuthorizedRBAC1: Unknown type for what: %s')
% how.__class__.__name__)
return None
'''Limit the number of activities to check'''
limit = 0
if not settings.ACTIVITY_GRAPH_LIMIT:
logger.warning('isAuthorizedRBAC1: Not Limit to research')
else:
limit = settings.ACTIVITY_GRAPH_LIMIT
#logger.info('isAuthorizedRBAC1: Limit is at %s' %limit)
i = 0
while a and (limit == 0 or i < limit):
for activity in a:
p = isAuthorizedRBAC0(who, what, activity)
if p:
return p
i = i + 1
if i == limit and limit != 0:
logger.warning(('isAuthorizedRBAC1: Limit of %s reached')
% limit)
return None
tmp = []
for activity in a:
control.append(activity)
for it in Activity.objects.filter(activities__id=activity.id):
if it not in control:
tmp.append(it)
a = tmp
return None
def isAuthorizedRBAC2(who, what, how):
if not who or not what or not how:
return None
p = None
p = isAuthorizedRBAC1(who, what, how)
if p:
return p
v = []
control = []
if isinstance(what, AcsObject):
views = View.objects.filter(acs_objects__id=what.id)
for view in views:
v.append(view)
elif isinstance(what, View):
v.append(what)
elif isinstance(what, UserAlias):
views = View.objects.filter(users__id=what.id)
for view in views:
v.append(view)
elif isinstance(what, Role):
views = View.objects.filter(roles__id=what.id)
for view in views:
v.append(view)
elif isinstance(what, Action):
views = View.objects.filter(actions__id=what.id)
for view in views:
v.append(view)
elif isinstance(what, Activity):
views = View.objects.filter(activities__id=what.id)
for view in views:
v.append(view)
else:
logger.error('isAuthorizedRBAC2: Unknown type')
return None
'''Limit the number of views to check'''
limit = 0
if not settings.VIEW_GRAPH_LIMIT:
logger.warning('isAuthorizedRBAC2: Not Limit to research')
else:
limit = settings.VIEW_GRAPH_LIMIT
#logger.info('isAuthorizedRBAC2: Limit is at %s' %limit)
i = 0
while v and (limit == 0 or i < limit):
for view in v:
p = isAuthorizedRBAC1(who, view, how)
if p:
return p
i = i + 1
if i == limit and limit != 0:
logger.warning(('isAuthorizedRBAC2: Limit of %s reached')
% limit)
return None
tmp = []
for view in v:
# Append the list of views already checked
control.append(view)
for it in View.objects.filter(views__id=view.id):
if it not in control:
'''Avoid loop'''
tmp.append(it)
v = tmp
return None
'''
RBAC and ABAC policies requesting.
* Output
The function returns a tuple (decision, rule/permission, error code)
- decision is a boolean
- rule/permission is new rule computed when decision is False, else it is
the persmission that permitted that the decision be True
- See error codes below
* @parameters
By default @who is a User instance, @what is an Object instance and @how
is an Action instance.
Deprecated (cf bellow): Use @role=True if @who is a Role instance
Use @view=True if @what is a View instance
Use @activity=True if @how is a Activity instance
@who, @what and @how are the name attribute of instances
@namespace is the name of a namespace instance
@no_rule_returned is False by default. The process of building a new
rule is ressource consuming. This process can be desactivated with
option no_rule_returned = True
@no_attribute_signal is False by default. Use to not send the signal to
obtain attributes.
* Error codes:
0 : No error
TBD
* Preface
1- Reminder on aliases
Requester is the user accessing ACS. The username is equal to the alias in
default.
Who is the subject of the request, it is described by the alias in a
policy.
Requester == who means that default_alias(requester) == default_alias(who)
2- difference between ABAC and RBAC permissions
An RBAC permission set with who is a role means that at the evaluation,
the access is granted if who is that role, a subject (a user or a role)
has that role (directly or by hierarchy)
With an ABAC rule, a request with a role as subject can never be granted,
but can only serve to obtain a new rule with that role has subject or
other permissions with roles as who included in that role
Note that an ABAC rule with who that is a role is strictly equivalent to
a rule on anybody with in the ABAC rule a PredicateRole at the root with
a AND statement with the rest of the logical expression.
Then an ABAC permission set with who is a role is not a way to authorize a
role (only possible with classic RABC rules) but a way to indicate that
they are candidate permissions for users that have that role.
In fact it is not very useful to ask if a role may have an access.
It does not make to say "does the role/function can access what/how?".
The right way to ask the question is "User lambda that eventually I don't
but who has the role/function X can access what/how?".
Then with ABAC rule we do not handle when the request is about a role.
It explains why we have removed the possibility to set an ABAC rule on a
role as who.
- When who is None/Anybody and the rule is only a predicate of having a
role, it is a classic RBAC permission.
- Idem with a rule on attributes, a classic ABAC rule.
- We can mix both. It is a way to give a logical expression on roles.
- Who is a user and no rule, classic IBAC rule.
- Who is a user and a rule, this user can have an access if the rule is
satisfied
* Requester != who
Requester != None
Who != None
A requester different from who request the policy to know if who can do
how on what.
The requester must be authorized in that policy for administration on who,
what and how.
RBAC permissions are checked. Then ABAC permissions are checked.
Attributes can be retreived from different sources. who has a default
alias that is the username that is mapped with the user identifiers wihtin
the different identity repositories (sources) configured.
A signal is send to make external functions provide attributes.
This is the preferred way the attributes should be provided from sources.
An option allows to say to not send this signal
no_attribute_signal = True, at False by default.
Attributes can also be given in parameters of the functions.
A validity period may be provided per attribute or per source. When it is
the case, these attributes are cached to be reused in other session.
These are stored is a special source called cached_attributes.
This function always tries to load attributes from this source.
Attributes are provided with dictionnaries as follows:
data = []
attr = {}
attr['name'] = 'certificate_type'
attr['namespace'] = 'ACS-ABAC'
attr['values'] = ('eID',)
attr['expiration_date'] = '2011-09-05T11:48:24'
data.append(attr)
attr = {}
attr['name'] = \
'http://schemas.xmlsoap.org/ws/2005/05/identity/claims/dateofbirth'
attr['namespace'] = \
'http://schemas.xmlsoap.org/ws/2005/05/identity/claims'
attr['values'] = ('22', )
data.append(attr)
dic = {}
dic[source_name] = data
*All attributes are loaded in an object profile used for the evaluation.*
If the access is granted, the function return un tuple containing 'True',
the permission that allowed the positive answer and error code at 0.
If the access is denied, the function return un tuple containing 'False',
the ABAC rule necessary to grant the access and error code at 0.
Obtaining a new ABAC rule can only be done with ABAC permissions.
The rule returned can be build from any ABAC permission that is on what to
do how.
The process of building a new rule is ressource consuming, this process
can be desactivated with the option no_rule_returned = True, at False by
default.
If who is a role. If there is a matching RBAC rule, the decision is True.
If there is an ABAC rule set on anybody with a unique predicate role on
that role, the decission is True.
Else a new rule is returned made of permissions set on anybody.
* Requester != who
Requester == None
Who != None
Unauthorized access to request the policy on somebody is not authorized.
This request is rejected with an error code
* Requester != who
Requester != None
Who == None
This is supported to only to make a requester authorized for
administration on what and how to obtain a rule that anybody may satisfy
to do how on what.
This is not compatible with no_rule_returned = True
* Requester == who != None
This mode means that the user is currently using an application coupled
with ACS. This application and ACS share the user session.
Checking administration rights does not have sense here.
Except this, this functions the same way as the first case.
However, attributes are also check in the session:
request.session['attributes']
An option allows to say to not check attributes in the session
no_session_attributes = True, at False by default.
This mode also always try to load attributes from the cached_attributes
source
The rule returned can only contain predicates of ABAC rules set on
anybody or on who or on a role that who has.
Differently from when requester != who and the requester is checked for
administration, new rule can not be set from any ABAC permission matching
what and how. Indeed, since the administration rights are not check, the
access to the policy is given on ABAC permissions that have as subject
anybody, the requester or a role that the requester has.
* Requester == who == None
This mode means that the user is currently using an application coupled
with ACS. This application and ACS share the user session.
Checking administration rights does not have sense here neither.
The requester/subject being 'anonymous', identity attributes can't be
retreived from sources.
Contextual attributes (time for instance) but it is not supported yet.
Signals may be used for contextual attributes.
Attributes in parameters and in the session can be used.
Example of scenario:
- The user is redirected on an IdP proxy with an access request from a SP.
- The user is not able to authenticate against the proxy.
- The proxy asks the policy in that mode with no attributes, a new rule from
permissions set on anybody are returned.
- The attributes can only come from external sources and be brought by
the user, i.e. certificates.
- The proxy make the user provide certificates to from the ABAC rule.
- Then the proxy parse the certificates and builds a dictionnary. The proxy
may also try to authenticate the user from certificates. if yes,
previously provided certificates may be loaded from cached_attributes.
- Either the dictionnary is put in the session, or given in parameter of
this function.
- The policy is requested.
- If the access is not granted, a new rule is returned.
- If the access is granted, before responding to the SP, the proxy may
create an account giving authentication credentials to the user or using
one of the certificates as an authentication credential.
- Then the proxy associates the living-more-than-a-session certified
attributes to this profile. The proxy is in charge to add by itself add
attributes in the cached_attributes source.
- Next time, the user will be able to authenticate and benefit from the
cached_attributes source
'''
'''
The function is not able to send the list of attributes necessary with the
signal.
A new rule may be returned and another function will have to grab the
attributes and put them in the session or in parameter
'''
'''
A profile contains data assertions.
A profile may be built for an anonymous user. Then a routine must be used
to clean the db.
Else, a user has a profile which contain all the attributes provided.
Attributes may be providing with an expiration date (ISO 8601 format).
Then, we look at the user profile if existing and at loading, every
outdated are removed.
If is expected to check an authorization with an existing profil an no
other attributes, it is enough to call is_authorized_by_names_with_abac
with no attributes in parameter nor in session and
no_attribute_signal=True
If a new rule is returned, it is enough to ask only attributes missing
from that rule.
However, the function won't send in the signal the list of necessary
attributes.
'''
def is_authorized_by_names_with_abac(requestor_name, who_name, what_name,
how_name, namespace_name,
view=False, activity=False, request=None, attributes=None,
no_rule_returned=False,
no_attribute_signal=False):
if not what_name or not how_name or not namespace_name:
logger.error('is_authorized_by_names_with_abac: \
a parameters is missing')
return (False, None, -1)
ns = None
try:
ns = Namespace.objects.get(name=namespace_name)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
unable to find the namespace object')
return (False, None, -4)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple namespaces with name %s' %namespace_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
Namespace %s - Error due to %s' % (namespace_name, err))
return (False, None, -5)
policy = get_policy_from_namespace(ns)
if not policy:
logger.critical('is_authorized_by_names_with_abac: \
Not policy found for %s' % ns)
return (False, None, -5)
what = None
whats_regexp = None
if view:
try:
what = View.objects.get(name=what_name, namespace=ns)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
unable to find the view object')
return (False, None, -2)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple views with name %s' %what_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
View %s - Error due to %s' % (what_name, err))
return (False, None, -5)
else:
whats_regexp = get_objects_from_regex(what_name, ns)
try:
what = AcsObject.objects.get(name=what_name, namespace=ns)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
unable to find the object object')
if not whats_regexp:
return (False, None, -2)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple objects with name %s' %what_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
UserAlias %s - Error due to %s' % (what_name, err))
return (False, None, -5)
how = None
if activity:
try:
how = Activity.objects.get(name=how_name, namespace=ns)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
unable to find the activity object')
return (False, None, -3)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple activities with name %s' %how_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
Activity %s - Error due to %s' % (how_name, err))
return (False, None, -5)
else:
try:
how = Action.objects.get(name=how_name, namespace=ns)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
unable to find the action object')
return (False, None, -3)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple actions with name %s' %how_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
Action %s - Error due to %s' % (how_name, err))
return (False, None, -5)
requestor = None
if requestor_name:
try:
requestor = User.objects.get(username=requestor_name)
if not set_default_alias(requestor):
logger.critical('is_authorized_by_names_with_abac: \
unable to handle default alias of %s' % requestor)
return (False, None, -6)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
the requestor with name %s does not exists' % requestor_name)
return (False, None, -6)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple user with name %s' % requestor_name)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
User %s - Error due to %s' % (requestor_name, err))
return (False, None, -5)
who = None
if who_name:
who = None
try:
who = UserAlias.objects.get(alias=who_name, namespace=ns)
except ObjectDoesNotExist:
logger.error('is_authorized_by_names_with_abac: \
the user with name %s does not exists' % who_name)
return (False, None, -7)
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names_with_abac: \
Multiple users with name %s' %who)
return (False, None, -5)
except Exception, err:
logger.critical('is_authorized_by_names_with_abac: \
UserAlias %s - Error due to %s' % (who, err))
return (False, None, -5)
if not requestor and who:
logger.error('is_authorized_by_names_with_abac: \
requestor is unknwon and who %s is provided' % who)
return (False, None, -8)
'''
The requester is different from who that means that the requester must
be authorized on the parameters of its request
'''
if requestor and (not who or not who.user or who.user != requestor):
logger.debug("The requester is different from who that means that the requester must \
be authorized on the parameters of its request")
administration = Action.objects.get(name='administration')
if who:
p = isAuthorizedRBAC2(set_default_alias(requestor), who, administration)
if not is_policy_user_administrator(requestor, policy) and not p:
logger.debug('is_authorized_by_names_with_abac: %s is not authorized on %s' \
% (requestor, who))
return (False, None, -11)
if what:
p = isAuthorizedRBAC2(set_default_alias(requestor), what, administration)
if not is_policy_object_creator(requestor, policy) and not p:
logger.debug('is_authorized_by_names_with_abac: %s is not authorized on %s' \
% (requestor, what))
return (False, None, -9)
if whats_regexp:
for o in whats_regexp:
p = isAuthorizedRBAC2(set_default_alias(requestor), o, administration)
if not is_policy_object_creator(requestor, policy) and not p:
logger.warning('is_authorized_by_names_with_abac: %s is not authorized on %s \
This object is removed from permission lookup!' \
% (requestor, o))
whats_regexp.pop(o)
p = isAuthorizedRBAC2(set_default_alias(requestor), how, administration)
if not is_policy_action_creator(requestor, policy) and not p:
logger.debug('is_authorized_by_names_with_abac: %s is not authorized on %s' \
% (requestor, how))
return (False, None, -10)
'''
RBAC permission checking
'''
if who and what:
p = isAuthorizedRBAC2(who, what, how)
if p:
return (True, p, 0)
if who and whats_regexp:
for o in whats_regexp:
p = isAuthorizedRBAC2(who, o, how)
if p:
return (True, p, 0)
'''
ABAC permission checking
'''
'''
Attribute loading in profile object
'''
# from abac.core import load_profile_by_dic, load_or_create_user_profile
from attribute_aggregator.core import load_or_create_user_profile
if who.user:
profile = load_or_create_user_profile(user=who.user)
else:
# profile = load_or_create_user_profile(user=None)
profile = load_or_create_user_profile()
if not profile:
logger.critical('is_authorized_by_names_with_abac: \
Error to create or load profile for %s' % who)
return (False, None, -13)
logger.debug('is_authorized_by_names_with_abac: profile %s returned' \
% profile)
if attributes:
logger.debug('is_authorized_by_names_with_abac: \
load attributes %s in parameters' \
% attributes)
# load_profile_by_dic(profile, attributes)
profile.load_by_dic(attributes)
'''
The requester is who, that means that the user is logged on
the application and request the policy.
Attributes can be grabbed from local sources, the django session and
in parameters.
Requester is also considered to be who, if requester and who are None.
It means that the user is anonymous - not been authenticated.
Attributes can be grabbed from the django session and in parameters.
'''
if (not requestor and not who) \
or (requestor and who and requestor == who):
if request and getattr(request, 'session', None) and 'attributes' in request.session:
# load_profile_by_dic(profile, request.session['attributes'])
logger.debug('is_authorized_by_names_with_abac: \
load attributes %s in parameters' \
% request.session['attributes'])
profile.load_by_dic(request.session['attributes'])
# if who and not no_attribute_signal:
# if not who.user:
# attributes_provided = attributes_call.send(sender=None,
# request=request, user=who)
# else:
# attributes_provided = attributes_call.send(sender=None,
# request=request, user=who.user)
# logger.info('is_authorized_by_names_with_abac: signal attributes_call sent')
# attributes = {}
# for attrs in attributes_provided:
# logger.info('is_authorized_by_names_with_abac: attributes_call connected to function %s' % \
# attrs[0].__name__)
# logger.info('is_authorized_by_names_with_abac: attributes provided are %s' %str(attrs[1]))
# load_profile_by_dic(profile, attrs[1])
'''
Attribute profile filled by the signal only works with a User instance
'''
if who.user and not no_attribute_signal:
logger.debug('is_authorized_by_names_with_abac: \
load attributes in bulk for %s' % who.user)
profile.load_greedy()
logger.debug('is_authorized_by_names_with_abac: The profile is %s' % profile)
'''
All the permissions with and view or an activity containing the what
and how
'''
'''
There are no ABAC permissions for administration and then what is only
AcsObject or View
'''
what_list = []
if whats_regexp:
for o in whats_regexp:
views = View.objects.filter(acs_objects__id=o.id)
for view in views:
if view.namespace == ns and not view in what_list:
what_list.append(view)
if what and isinstance(what, AcsObject):
views = View.objects.filter(acs_objects__id=what.id)
for view in views:
if view.namespace == ns and not view in what_list:
what_list.append(view)
elif what and isinstance(what, View):
if what.namespace == ns:
what_list.append(what)
# elif isinstance(what, UserAlias):
# views = View.objects.filter(users__id=what.id)
# for view in views:
# if view.namespace == ns:
# what_list.append(view)
# elif isinstance(what, Role):
# views = View.objects.filter(roles__id=what.id)
# for view in views:
# if view.namespace == ns:
# what_list.append(view)
# elif isinstance(what, Action):
# views = View.objects.filter(actions__id=what.id)
# for view in views:
# if view.namespace == ns:
# what_list.append(view)
# elif isinstance(what, Activity):
# views = View.objects.filter(activities__id=what.id)
# for view in views:
# if view.namespace == ns:
# what_list.append(view)
'''Limit the number of views to check'''
limit = 0
if not settings.VIEW_GRAPH_LIMIT:
logger.warning('is_authorized_by_names_with_abac: Not Limit to research')
else:
limit = settings.VIEW_GRAPH_LIMIT
#logger.info('is_authorized_by_names_with_abac: Limit is at %s' %limit)
i = 0
while i < len(what_list) and (limit == 0 or i < limit):
for it in View.objects.filter(views__id=what_list[i].id):
if it not in what_list and it.namespace == ns:
what_list.append(it)
i = i + 1
if what:
what_list.append(what)
for o in whats_regexp:
what_list.append(o)
how_list = []
if isinstance(how, Action):
activities = Activity.objects.filter(actions__id=how.id)
for activity in activities:
if activity.namespace == ns:
how_list.append(activity)
elif isinstance(how, Activity):
activities = Activity.objects.filter(activities__id=how.id)
for activity in activities:
if activity.namespace == ns:
how_list.append(activity)
'''Limit the number of activities to check'''
limit = 0
if not settings.ACTIVITY_GRAPH_LIMIT:
logger.warning('is_authorized_by_names_with_abac: Not Limit to research')
else:
limit = settings.ACTIVITY_GRAPH_LIMIT
#logger.info('is_authorized_by_names_with_abac: Limit is at %s' %limit)
i = 0
while i < len(how_list) and (limit == 0 or i < limit):
for it in Activity.objects.filter(activities__id=how_list[i].id):
if it not in how_list:
how_list.append(it)
i = i + 1
how_list.append(how)
permissions = []
for what in what_list:
for how in how_list:
t_what = ContentType.objects.get_for_model(what)
t_how = ContentType.objects.get_for_model(how)
'''Permissions for anybody'''
permissions = permissions + list(AcsAbacPermission.objects.filter(who=None,
content_type_what__pk=t_what.id,
object_id_what=what.id,
content_type_how__pk=t_how.id,
object_id_how=how.id))
'''Permissions specific for a user'''
if who:
permissions = permissions + list(AcsAbacPermission.objects.filter(who=who,
content_type_what__pk=t_what.id,
object_id_what=what.id,
content_type_how__pk=t_how.id,
object_id_how=how.id))
if not permissions:
logger.debug('is_authorized_by_names_with_abac: Access denied - no permission found')
return (False, None, 0)
logger.debug('is_authorized_by_names_with_abac: permissions %s' % str([str(x) for x in permissions]))
from abac.core import check_predicates, \
make_new_rule_from_missing_predicates
from acs.abac.logic import evaluation, return_sorted_variables_to_truth
new_rule = ''
dic_of_new_rules = []
for permission in permissions:
if not getattr(permission, 'rule', None) or permission.rule == '':
logger.debug('is_authorized_by_names_with_abac: The permission %s has no rule - access granted!' % permission)
return (True, permission, 0)
logger.debug('is_authorized_by_names_with_abac: permission %s' % permission)
l = check_predicates(permission.rule, profile)
logger.debug('is_authorized_by_names_with_abac: result on %s is %s' % (permission.rule.expression, l))
res = evaluation(permission.rule.expression, l)
if res:
return (True, permission, 0)
elif not no_rule_returned:
logger.debug('is_authorized_by_names_with_abac: permission not successful')
missing_p = return_sorted_variables_to_truth(permission.rule.expression, l)
logger.debug('is_authorized_by_names_with_abac: missing predicates %s' % missing_p)
r = make_new_rule_from_missing_predicates(missing_p)
if new_rule != '':
if not r in dic_of_new_rules:
new_rule += '|' + r
else:
new_rule = r
dic_of_new_rules.append(r)
logger.debug('is_authorized_by_names_with_abac: access denied')
profile.delete()
if new_rule != '':
logger.debug('is_authorized_by_names_with_abac: new rule %s' % new_rule)
return (False, new_rule, 0)
else:
return (False, None, 0)
def is_authorized_by_names(requester, who, what, how, namespace,
role=False, view=False, activity=False):
'''
By default who is a User, what is an Object and how is an Action
Use role=True if who is a Role, etc.
who, what and how are the name attribute of instances
namespace is the name of a namespace instance
return codes:
1 : Authorization granted
0 : Authorization denied
-1 : who not found
-2 : what not found
-3 : how not found
-4 : namespace not found
-5 : error processing request
-6 : requester not found
-7 : requester not authorized on who
-8 : requester not authorized on what
-9 : requester not authorized on how
'''
user = None
try:
user = User.objects.get(username=requester)
if not set_default_alias(user):
return -6
except ObjectDoesNotExist:
return -6
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple user with name %s' %requester)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
User %s - Error due to %s' % (requester, err))
return -5
ns = None
try:
ns = Namespace.objects.get(name=namespace)
except ObjectDoesNotExist:
return -4
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple namespaces with name %s' %namespace)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
Namespace %s - Error due to %s' % (namespace, err))
return -5
policy = get_policy_from_namespace(ns)
if not policy:
logger.critical('is_authorized_by_names: \
Not policy found for %s' % ns)
return -5
who_o = None
if role:
try:
who_o = Role.objects.get(name=who, namespace=ns)
except ObjectDoesNotExist:
return -1
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple roles with name %s' %who)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
Role %s - Error due to %s' % (who, err))
return -5
else:
try:
who_o = UserAlias.objects.get(alias=who, namespace=ns)
except ObjectDoesNotExist:
return -1
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple users with name %s' %who)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
UserAlias %s - Error due to %s' % (who, err))
return -5
what_o = None
if view:
try:
what_o = View.objects.get(name=what, namespace=ns)
except ObjectDoesNotExist:
return -2
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple views with name %s' %what)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
View %s - Error due to %s' % (what, err))
return -5
else:
try:
what_o = AcsObject.objects.get(name=what, namespace=ns)
except ObjectDoesNotExist:
return -2
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple objects with name %s' %what)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
UserAlias %s - Error due to %s' % (what, err))
return -5
how_o = None
if activity:
try:
how_o = Activity.objects.get(name=how, namespace=ns)
except ObjectDoesNotExist:
return -3
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple activities with name %s' %how)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
Activity %s - Error due to %s' % (how, err))
return -5
else:
try:
how_o = Action.objects.get(name=how, namespace=ns)
except ObjectDoesNotExist:
return -3
except MultipleObjectsReturned:
logger.critical('is_authorized_by_names: \
Multiple actions with name %s' %how)
return -5
except Exception, err:
logger.critical('is_authorized_by_names: \
Action %s - Error due to %s' % (how, err))
return -5
administration = Action.objects.get(name='administration')
p = isAuthorizedRBAC2(set_default_alias(user), who_o, administration)
if not is_policy_user_administrator(user, policy) and not p:
logger.debug('is_authorized_by_names: %s is not authorized on %s' \
% (user, who_o))
return -7
p = isAuthorizedRBAC2(set_default_alias(user), what_o, administration)
if not is_policy_object_creator(user, policy) and not p:
logger.debug('is_authorized_by_names: %s is not authorized on %s' \
% (user, what_o))
return -8
p = isAuthorizedRBAC2(set_default_alias(user), how_o, administration)
if not is_policy_action_creator(user, policy) and not p:
logger.debug('is_authorized_by_names: %s is not authorized on %s' \
% (user, how_o))
return -9
p = isAuthorizedRBAC2(who_o, what_o, how_o)
if p:
logger.debug('is_authorized_by_names: %s - %s - %s - access granted' \
% (who, what, how))
return 1
else:
logger.debug('is_authorized_by_names: %s - %s - %s - access denied' \
% (who, what, how))
return 0
def is_authorized_with_object_in_regexp(requester, who, what, how, namespace,
role=False, activity=False):
'''
Acs object with new field regex - If filled it is a regexp field
A regexp object is conceptually a view
'''
user = None
try:
user = User.objects.get(username=requester)
if not set_default_alias(user):
return -6
except ObjectDoesNotExist:
return -6
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple user with name %s' %requester)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
User %s - Error due to %s' % (requester, err))
return -5
ns = None
try:
ns = Namespace.objects.get(name=namespace)
except ObjectDoesNotExist:
return -4
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple namespaces with name %s' %namespace)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
Namespace %s - Error due to %s' % (namespace, err))
return -5
policy = get_policy_from_namespace(ns)
if not policy:
logger.critical('is_authorized_with_object_in_regexp: \
Not policy found for %s' % ns)
return -5
who_o = None
if role:
try:
who_o = Role.objects.get(name=who, namespace=ns)
except ObjectDoesNotExist:
return -1
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple roles with name %s' %who)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
Role %s - Error due to %s' % (who, err))
return -5
else:
try:
who_o = UserAlias.objects.get(alias=who, namespace=ns)
except ObjectDoesNotExist:
return -1
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple users with name %s' %who)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
UserAlias %s - Error due to %s' % (who, err))
return -5
how_o = None
if activity:
try:
how_o = Activity.objects.get(name=how, namespace=ns)
except ObjectDoesNotExist:
return -3
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple activities with name %s' %how)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
Activity %s - Error due to %s' % (how, err))
return -5
else:
try:
how_o = Action.objects.get(name=how, namespace=ns)
except ObjectDoesNotExist:
return -3
except MultipleObjectsReturned:
logger.critical('is_authorized_with_object_in_regexp: \
Multiple actions with name %s' %how)
return -5
except Exception, err:
logger.critical('is_authorized_with_object_in_regexp: \
Action %s - Error due to %s' % (how, err))
return -5
administration = Action.objects.get(name='administration')
p = isAuthorizedRBAC2(set_default_alias(user), who_o, administration)
if not is_policy_user_administrator(user, policy) and not p:
logger.debug('is_authorized_with_object_in_regexp: \
%s is not authorized on %s' % (user, who_o))
return -7
p = isAuthorizedRBAC2(set_default_alias(user), how_o, administration)
if not is_policy_action_creator(user, policy) and not p:
logger.debug('is_authorized_with_object_in_regexp: \
%s is not authorized on %s' % (user, how_o))
return -9
objects = get_objects_from_regex(what, ns)
if not objects:
logger.debug('is_authorized_with_object_in_regexp: \
No matching regexp with %s' % what)
return -2
for o in objects:
p = isAuthorizedRBAC2(who_o, o, how_o)
if p:
p2 = isAuthorizedRBAC2(set_default_alias(user), o, administration)
if p2 or is_policy_object_creator(user, policy):
logger.debug('is_authorized_with_object_in_regexp: \
%s - %s - %s - access granted' % (who, o, how))
return 1
logger.debug('is_authorized_with_object_in_regexp: \
%s - %s - %s - access denied' % (who, what, how))
return 0
def get_objects_from_regex(value, namespace):
if not value or not namespace:
return []
objects = AcsObject.objects.filter(namespace=namespace)
objects_found = []
for o in objects:
if o.regex:
r = None
try:
r = re.compile(o.regex)
if r.search(value):
logger.debug('get_objects_from_regex: \
%s match the regex %s of %s' % (value, o.regex, o))
objects_found.append(o)
except Exception, err:
logger.error('get_objects_from_regex: \
invalid regex %s of %s due to %s' % (o.regex, o, err))
return objects_found
def is_valid_regex(regex):
try:
logger.debug('is_valid_regex: regex is %s' % regex)
re.compile(regex)
logger.debug('is_valid_regex: valid regex')
return True
except Exception, err:
logger.debug('is_valid_regex: invalid regex due to %s' % err)
return False
'''
''''''
Administrators management
''''''
'''
'''Achtung: every administration rights are set on the aliases in default
All the admin roles and views are in default.
To separate them, we check the namespace
Admin roles only take aliases in default.
To separate admin roles and views created to administrate
a policy we tag it with one or multiple policies.
Every policy administrator should be able
to create admin views tagged for a policy.
Then they can include into any entities they want of that namespace.
'''
def is_root_administrator(user):
'''Root administrators'''
alias = set_default_alias(user)
try:
Role.objects.get(name='root_admin_role', users__id=alias.id)
return True
except:
return False
def set_root_administrator(requester, target):
if not requester:
return -1
if not is_root_administrator(requester):
return -2
try:
admin_role = Role.objects.get(name='root_admin_role')
if not set_default_alias(target):
return -3
alias = set_default_alias(target)
if not alias in admin_role.users.all():
admin_role.users.add(alias)
logger.debug('set_root_administrator: \
%s added to the root admin role' % target)
return 0
else:
logger.debug('set_root_administrator: \
%s already a root admin' % target)
return -4
except Exception, err:
logger.error('set_root_administrator: \
failed to add %s to the root admin role due to %s' \
% (target, err))
return -5
def remove_root_administrator(requester, target):
if not requester:
return -1
if not is_root_administrator(requester):
return -2
if requester.id == target.id:
return -3
try:
admin_role = Role.objects.get(name='root_admin_role')
alias = set_default_alias(target)
if alias in admin_role.users.all():
admin_role.users.remove(alias)
logger.debug('remove_root_administrator: \
%s is not a root admin anymore' % target)
return 0
else:
logger.debug('remove_root_administrator: \
%s is not a root admin' % target)
return -4
except Exception, err:
logger.error('remove_root_administrator: \
failed to remove %s to the root admin role due to %s' \
% (target, err))
return -5
def return_root_administrators():
'''
A root administrator has a system view
that is not included in other system view.
There could be multiple root administrators.
'''
try:
root_role = Role.objects.get(name='root_admin_role')
return root_role.users.all()
except:
return []
def is_user_administrator(user):
'''User administrators: User admin role or root adminnistrators'''
if is_root_administrator(user):
return True
alias = set_default_alias(user)
try:
Role.objects.get(name='root_user_administrator_role',
users__id=alias.id)
return True
except:
return False
def is_abac_administrator(user):
if is_root_administrator(user):
return True
alias = set_default_alias(user)
try:
Role.objects.get(name='root_abac_administrator_role',
users__id=alias.id)
return True
except:
return False
def get_user_administrator_role():
try:
return Role.objects.get(name='root_user_administrator_role')
except:
return None
def get_abac_administrator_role():
try:
return Role.objects.get(name='root_abac_administrator_role')
except:
return None
'''
Each policy has a root admin role.
Every policy root admin should be granted through this role.
No direct administration permission of the user on the policy root admin view
should be set.
These functions can only be handled by policy root administrators
(that includes root administrators).
'''
def is_policy_root_administrator(target, policy):
if not target or not policy or not policy.admin_role:
logger.debug('is_policy_root_administrator: \
Missing argument')
return -1
if is_root_administrator(target):
logger.debug('is_policy_root_administrator: \
%s is a policy root admin (by root admin)' % target)
return 1
if not set_default_alias(target):
logger.debug('is_policy_root_administrator: \
%s do not pass check for admin' % target)
return -3
alias = set_default_alias(target)
admins = []
try:
admins = policy.admin_role.users.all()
except Exception, err:
logger.error('is_policy_root_administrator: \
error due to %s' % err)
return -2
if alias in admins:
logger.debug('is_policy_root_administrator: \
%s is a policy root admin' % target)
return 1
else:
logger.debug('is_policy_root_administrator: \
%s is not a policy root admin' % target)
return 0
def set_policy_root_administrator(requester, target, policy):
if not requester or not target or not policy or not policy.admin_role:
logger.debug('set_root_policy_administrator: \
Missing argument')
return -1
if not is_policy_root_administrator(requester, policy):
logger.debug('set_root_policy_administrator: \
requester %s is not a root admin' % requester)
return -2
if not set_default_alias(target):
logger.debug('set_root_policy_administrator: \
%s do not pass check for admin' % target)
return -3
alias = set_default_alias(target)
if alias in policy.admin_role.users.all():
logger.debug('set_root_policy_administrator: \
%s already policy root admin' % target)
return -4
try:
policy.admin_role.users.add(alias)
logger.debug('set_root_policy_administrator: \
%s set policy root admin' % target)
return 0
except Exception, err:
logger.error('set_root_policy_administrator: \
error due to %s' % err)
return -5
def remove_policy_root_administrator(requester, target, policy):
if not requester or not target or not policy or not policy.admin_role:
logger.debug('remove_policy_root_administrator: \
Missing argument')
return -1
if not is_policy_root_administrator(requester, policy):
logger.debug('remove_policy_root_administrator: \
requester %s is not a root admin' % requester)
return -2
if not set_default_alias(target):
logger.debug('remove_policy_root_administrator: \
%s do not pass check for admin' % target)
return -3
alias = set_default_alias(target)
if not alias in policy.admin_role.users.all():
logger.debug('remove_policy_root_administrator: \
%s is not a policy root admin' % target)
return -4
try:
policy.admin_role.users.remove(alias)
logger.debug('remove_policy_root_administrator: \
%s set policy root admin' % target)
return 0
except Exception, err:
logger.error('remove_policy_root_administrator: \
error due to %s' % err)
return -5
'''A policy administrator'''
'''A policy administrator is at least:
- In default:
- a root administrator
- Per policy:
- a policy admin
- has an administration permission in a policy
- has a role which has an administration permission in a policy
'''
def is_policy_administrator(user, policy):
if is_policy_user_administrator(user, policy) or \
is_policy_abac_administrator(user, policy) or \
is_policy_object_creator(user, policy) or \
is_policy_action_creator(user, policy):
return True
how = Action.objects.get(name='administration')
t_how = ContentType.objects.get_for_model(how)
alias = set_default_alias(user)
if alias:
t_who = ContentType.objects.get_for_model(alias)
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=alias.id,
content_type_how__pk=t_how.id,
object_id_how=how.id)
if perms:
return True
'''
We do not authorize to mix admin roles and not admin roles
We do not authorize to add to a role an alias not in that namespace
Then, it is enough to know if a user is a member of an admin role
in this namespace to say that it is an administrator
'''
try:
for r in policy.admin_roles.all():
if alias in r.users.all():
return True
except:
return False
return False
'''
An administrator is one or many:
- In default:
- a root administrator
- a user administrator
- an abac administrator
- In any policy:
- a policy admin
- has an administration permission in a policy
- has a role which has an administration permission in a policy
'''
def is_administrator(user):
if is_user_administrator(user):
return True
if is_abac_administrator(user):
return True
policies = Policy.objects.all()
for p in policies:
if is_policy_administrator(user, p):
return True
return False
def is_policy_user_administrator(user, policy):
if is_root_administrator(user) \
or is_user_administrator(user) \
or is_policy_root_administrator(user, policy):
return True
alias = set_default_alias(user)
if not alias:
return False
if alias in policy.user_admin_role.users.all():
return True
return False
def is_policy_abac_administrator(user, policy):
if is_root_administrator(user) \
or is_abac_administrator(user):
return True
alias = set_default_alias(user)
if not alias:
return False
if alias in policy.abac_admin_role.users.all():
return True
return False
def is_policy_object_creator(user, policy):
if is_root_administrator(user) \
or is_policy_root_administrator(user, policy):
return True
alias = set_default_alias(user)
if not alias:
return False
if alias in policy.object_creator_role.users.all():
return True
return False
def is_policy_action_creator(user, policy):
if is_root_administrator(user) \
or is_policy_root_administrator(user, policy):
return True
alias = set_default_alias(user)
if not alias:
return False
if alias in policy.action_creator_role.users.all():
return True
return False
'''
A user self administrator can give its permissions to other users or
group of users
'''
def is_self_admin(alias):
try:
if alias.self_admin == True:
return True
return False
except:
return False
def switch_self_admin(alias, policy):
try:
ua = UserAlias.objects.get(alias=alias, namespace=policy.namespace)
if ua.self_admin:
ua.self_admin = False
else:
ua.self_admin = True
ua.save()
return True
except:
return False
def switch_self_admin_by_id(alias_id):
try:
ua = UserAlias.objects.get(id=alias_id)
if ua.self_admin:
ua.self_admin = False
else:
ua.self_admin = True
ua.save()
return True
except:
return False
def all_users_self_admin_in_policy(user, policy):
if not user or not policy \
or not is_policy_user_administrator(user, policy):
return False
for user in User.objects.all():
try:
ua = UserAlias.objects.get(user=user, namespace=policy.namespace)
ua.self_admin = True
ua.save()
except:
pass
return True
def at_least_one_user_to_admin(user, policy):
items = []
items = return_list_users_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_user_administrator(user, policy):
for a in UserAlias.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_role_to_admin(user, policy):
items = []
items = return_list_roles_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_user_administrator(user, policy):
for a in Role.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_object_to_admin(user, policy):
items = []
items = return_list_objects_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_object_creator(user, policy):
for a in AcsObject.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_view_to_admin(user, policy):
items = []
items = return_list_views_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_object_creator(user, policy):
for a in View.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_action_to_admin(user, policy):
items = []
items = return_list_actions_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_action_creator(user, policy):
for a in Action.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_activity_to_admin(user, policy):
items = []
items = \
return_list_activities_authorized_for_admin(set_default_alias(user))
items = filter_list_in_namespace(items, policy.namespace)
if is_policy_action_creator(user, policy):
for a in Activity.objects.filter(namespace=policy.namespace):
items.append(a)
if items:
return True
return False
def at_least_one_permission_to_set(user, policy):
if (at_least_one_user_to_admin(user, policy) \
or at_least_one_role_to_admin(user, policy)) \
and (at_least_one_object_to_admin(user, policy) \
or at_least_one_view_to_admin(user, policy)) \
and (at_least_one_action_to_admin(user, policy) \
or at_least_one_activity_to_admin(user, policy)):
return True
return False
def at_least_one_permission_to_remove(user, policy):
administration = Action.objects.get(name='administration')
permissions = AcsPermission.objects.all()
items = []
for p in permissions:
if is_in_policy(p.who, policy) \
and is_in_policy(p.what, policy) \
and is_in_policy(p.how, policy):
perm = isAuthorizedRBAC2(set_default_alias(user),
p.who, administration)
if is_policy_user_administrator(user, policy) or perm:
perm = isAuthorizedRBAC2(set_default_alias(user),
p.what, administration)
if is_policy_object_creator(user, policy) or perm:
perm = isAuthorizedRBAC2(set_default_alias(user),
p.how, administration)
if is_policy_action_creator(user, policy) or perm:
items.append(p)
if items:
return True
return False
def at_least_one_abac_permission_to_set(user, policy):
if is_policy_abac_administrator(user, policy) \
and (at_least_one_object_to_admin(user, policy) \
or at_least_one_view_to_admin(user, policy)) \
and (at_least_one_action_to_admin(user, policy) \
or at_least_one_activity_to_admin(user, policy)):
return True
return False
def at_least_one_abac_permission_to_admin(user, policy):
if not is_policy_abac_administrator(user, policy):
return False
administration = Action.objects.get(name='administration')
permissions = AcsAbacPermission.objects.all()
items = []
for p in permissions:
if is_in_policy(p.what, policy) \
and is_in_policy(p.how, policy):
perm = isAuthorizedRBAC2(set_default_alias(user),
p.what, administration)
if is_policy_object_creator(user, policy) or perm:
perm = isAuthorizedRBAC2(set_default_alias(user),
p.how, administration)
if is_policy_action_creator(user, policy) or perm:
items.append(p)
if items:
return True
return False
'''
Every user need to have an alias in Default,
necessary for administration permissions.
When we deal with administration of, we deal with users and not alias.
To administrate ACS you need a user to login. However, administration
permissions a set on the default alias. The default alias is the username.
Conversely in normal permissions (into policies), we use aliases.
'''
def set_default_alias(user):
ns = Namespace.objects.get(name='Default')
try:
ua, res = UserAlias.objects.get_or_create(alias=user.username,
user=user, namespace=ns)
return ua
except Exception, err:
logger.critical('set_default_alias: Unable to get \
or create default alias due to %s' % err)
return None
'''Check acs is initialized
- Test what is created in the initialization script
- At leat one root user
'''
def check_acs_initialized():
'''
Test that the application is intialized -
Very loose test to be fast
Implement a depth test
'''
if return_root_administrators():
return True
logger.error('check_acs_initialized: No root administrator')
return False
'''
''''''
Aliases management
''''''
'''
def get_aliases_of_user(user):
return UserAlias.objects.filter(user=user)
def get_aliases_of_user_without_default(user):
aliases_tmp = get_aliases_of_user(user)
aliases = []
default = set_default_alias(user)
if not default:
return aliases_tmp
for a in aliases_tmp:
if a.id != default.id:
aliases.append(a)
return aliases
def get_aliases_in_policy(policy):
return UserAlias.objects.filter(namespace=policy.namespace)
def get_alias_in_policy(user, policy):
try:
alias = UserAlias.objects.get(user=user, namespace=policy.namespace)
return alias
except:
return None
def get_alias_in_policy_from_namespace(user, namespace):
try:
alias = UserAlias.objects.get(user=user, namespace=namespace)
return alias
except:
return None
def synchronize_users_in_policy(requester, policy):
if not requester or not policy:
return -1
if not is_policy_user_administrator(requester, policy):
return -2
for user in User.objects.all():
try:
ua = UserAlias.objects.get(user=user, namespace=policy.namespace)
logger.info('synchronize_users_in_policy: \
User %s already has the alias %s' \
% (user, ua.alias))
except ObjectDoesNotExist:
try:
ua = UserAlias(user=user, namespace=policy.namespace,
alias=user.username)
ua.save()
logger.debug('synchronize_users_in_policy: \
add alias %s for user %s' % (ua.alias, user))
except Exception, err:
logger.critical('synchronize_users_in_policy: \
Error creating an alias in %s for user %s' \
% (policy, user))
pass
except Exception, err:
logger.critical('synchronize_users_in_policy: \
Error raised when in reading alias %s of user %s' \
% (err, user))
pass
return 0
'''
''''''
Policies management
''''''
'''
def is_in_policy(item, policy):
if item and item.namespace and policy and policy.namespace and \
item.namespace.id == policy.namespace.id:
return True
return False
def is_default_alias(alias):
ns = Namespace.objects.get(name='Default')
if alias.namespace.id == ns.id:
return True
return False
def filter_list_in_namespace(liste, namespace):
l = []
if not liste or not liste[0]:
return l
for e in liste:
if e.namespace.id == namespace.id:
l.append(e)
return l
def permission_in_namespace(permission, namespace):
if permission and namespace and permission.who and permission.what and \
permission.how and permission.who.namespace and \
permission.what.namespace and permission.how.namespace and \
permission.who.namespace.id == permission.what.namespace.id \
== permission.how.namespace.id == namespace.id:
return True
return False
def abac_permission_in_namespace(permission, namespace):
if permission and namespace and permission.what \
and permission.how and permission.what.namespace and \
permission.how.namespace and permission.what.namespace.id == \
permission.how.namespace.id == namespace.id:
return True
return False
def check_permission(user, policy, permission):
administration = Action.objects.get(name='administration')
if not is_in_policy(permission.who, policy) \
or not is_in_policy(permission.what, policy) \
or not is_in_policy(permission.how, policy):
return False
default_alias = set_default_alias(user)
if not default_alias:
return False
if not is_policy_user_administrator(user, policy):
p = isAuthorizedRBAC2(set_default_alias(user),
permission.who, administration)
if not p:
return False
if not is_policy_object_creator(user, policy):
p = isAuthorizedRBAC2(set_default_alias(user),
permission.what, administration)
if not p:
return False
if not is_policy_action_creator(user, policy):
p = isAuthorizedRBAC2(set_default_alias(user),
permission.how, administration)
if not p:
return False
return True
def check_abac_permission(user, policy, permission):
administration = Action.objects.get(name='administration')
if not is_policy_abac_administrator(user, policy):
return False
if not is_in_policy(permission.what, policy) \
or not is_in_policy(permission.how, policy):
return False
default_alias = set_default_alias(user)
if not default_alias:
return False
if not is_policy_object_creator(user, policy):
p = isAuthorizedRBAC2(set_default_alias(user),
permission.what, administration)
if not p:
return False
if not is_policy_action_creator(user, policy):
p = isAuthorizedRBAC2(set_default_alias(user),
permission.how, administration)
if not p:
return False
return True
def get_authorized_as_admin_policies(user):
'''All the policies the user has the admin role of this policy'''
items = []
policies = Policy.objects.all()
for p in policies:
if is_policy_root_administrator(p, user):
items.append(p)
return items
def get_policy_from_namespace(namespace):
try:
p = Policy.objects.get(namespace=namespace)
return p
except:
return None
def is_admin_permission_in_policy(permission, policy):
'''
Who is an alias or an admin role, both in default
If admin role, must be in policy.admin_roles
What is an admin view in default or any object in the policy namespace
If admin view, must be in policy.admin_views
'''
default_ns = Namespace.objects.get(name='Default')
if isinstance(permission.who, Role) and \
not permission.who in policy.admin_roles.all():
return False
elif permission.who.namespace.id != default_ns.id:
return False
if permission.what in policy.admin_views.all():
return True
elif permission.what.namespace.id == policy.namespace.id:
return True
return False
@transaction.commit_manually
def create_policy(name, user_creator, namespace=None, restore=True):
'''
Check the user_creator is authorized to create a policy:
only root administrators
input:
name: string for the policy name
user_creator: alias in default of the creator that must be root
admininistrator
namespace: string for the policy namespace, if not given,
name is taken
restore: boolean
The option restore, by default True, consider that if one of the
necessary object it is not a problem.
Then a broken policy can be restored
With restore=False, we expect that all the necessary objects be
created freshly, then an error code is returned if an object
already exists
output:
(policy created/restored, boolean if created)
or raise Exception
'The user creator is not a root admin'
'Missing name or user creator'
'Multiple policy objects with the same name already exist'
'Multiple namespace objects with the same name already exist'
'Multiple policy admin view objects with the same name already
exist'
'Multiple policy admin role objects with the same name already
exist'
'Multiple policy user admin role objects with the same name
already exist'
'Multiple policy object creator role objects with the same
name already exist'
'Multiple policy action creator role objects with the same
name already exist'
restore at False:
'Object policy already exists'
'Object namespace already exists'
'Object policy admin view already exists'
'Object policy admin role already exists'
'Object policy user admin role already exists'
'Object policy abac admin role already exists'
'Object policy object creator role already exists'
'Object policy action creator role already exists'
'Policy permission already exists'
'''
policy_created = False
try:
if not is_root_administrator(user_creator):
logger.debug('create_policy: \
Only root administrators can create a policy')
raise Exception('The user creator is not a root admin')
if not name or not user_creator:
logger.debug('create_policy: Missing data to create a policy')
raise Exception('Missing name or user creator')
try:
Policy.objects.get(name=name)
logger.error('create_policy: \
A policy with the name %s already exists' % name)
if not restore:
raise Exception('Object policy already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple policies with the name %s exist' % name)
raise Exception('Multiple policy objects with the same name \
already exist')
if not namespace:
namespace = name
try:
Namespace.objects.get(name=namespace)
logger.error('create_policy: \
A namespace with the name %s already exists' % namespace)
if not restore:
raise Exception('Object namespace already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple namespaces with the name %s exist' % namespace)
raise Exception('Multiple namespace objects with the same name \
already exist')
view_name = name + '_policy_admin_view'
try:
View.objects.get(name=view_name)
if not restore:
raise Exception('Object policy admin view already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple views %s already exist' % view_name)
raise Exception('Multiple policy admin view objects with the \
same name already exist')
'''
Add a role with a permission on that view to make easier
the administration
'''
role_name = name + '_policy_admin_role'
try:
Role.objects.get(name=role_name)
logger.error('create_policy: The role %s already exists' \
% role_name)
if not restore:
raise Exception('Object policy admin role already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple roles %s already exist' % role_name)
raise Exception('Multiple policy admin role objects with the \
same name already exist')
user_admin_role_name = name + '_policy_user_admin_role'
try:
Role.objects.get(name=user_admin_role_name)
logger.error('create_policy: \
The role %s already exists' % user_admin_role_name)
if not restore:
raise Exception(\
'Object policy user admin role already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple roles %s already exist' % user_admin_role_name)
raise Exception('Multiple policy user admin role objects with \
the same name already exist')
abac_admin_role_name = name + '_policy_abac_admin_role'
try:
Role.objects.get(name=abac_admin_role_name)
logger.error('create_policy: \
The role %s already exists' % abac_admin_role_name)
if not restore:
raise Exception(\
'Object policy abac admin role already exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple roles %s already exist' % abac_admin_role_name)
raise Exception('Multiple policy abac admin role objects with \
the same name already exist')
object_creator_role_name = name + '_policy_object_creator_role'
try:
Role.objects.get(name=object_creator_role_name)
logger.error('create_policy: \
The role %s already exists' % object_creator_role_name)
if not restore:
raise Exception('Object policy object creator role already \
exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple roles %s already exist' % object_creator_role_name)
raise Exception('Multiple policy object creator role objects \
with the same name already exist')
action_creator_role_name = name + '_policy_action_creator_role'
try:
Role.objects.get(name=object_creator_role_name)
logger.error('create_policy: \
The role %s already exists' % action_creator_role_name)
if not restore:
raise Exception('Object policy action creator role already \
exists')
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('create_policy: \
Multiple roles %s already exist' % action_creator_role_name)
raise Exception('Multiple policy action creator role objects \
with the same name already exist')
admin_view = None
try:
admin_view, created = View.objects.get_or_create(name=view_name)
admin_view.save()
if created:
logger.debug('create_policy: creation of %s' % view_name)
else:
logger.debug('create_policy: object %s already exists' \
% view_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create view with name %s due to %s' \
% view_name)
raise err
admin_role = None
try:
admin_role, created = Role.objects.get_or_create(name=role_name)
admin_role.save()
if created:
logger.debug('create_policy: creation of %s' % role_name)
else:
logger.debug('create_policy: object %s already exists' \
% role_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create role with name %s due to %s' \
% (role_name, err))
raise err
user_admin_role = None
try:
user_admin_role, created = \
Role.objects.get_or_create(name=user_admin_role_name)
user_admin_role.save()
if created:
logger.debug('create_policy: creation of %s' \
% user_admin_role_name)
else:
logger.debug('create_policy: object %s already exists' \
% user_admin_role_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create role with name %s due to %s' \
% (user_admin_role_name, err))
raise err
abac_admin_role = None
try:
abac_admin_role, created = \
Role.objects.get_or_create(name=abac_admin_role_name)
abac_admin_role.save()
if created:
logger.debug('create_policy: creation of %s' \
% abac_admin_role_name)
else:
logger.debug('create_policy: object %s already exists' \
% abac_admin_role_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create role with name %s due to %s' \
% (abac_admin_role_name, err))
raise err
object_creator_role = None
try:
object_creator_role, created = \
Role.objects.get_or_create(name=object_creator_role_name)
object_creator_role.save()
if created:
logger.debug('create_policy: creation of %s' \
% object_creator_role_name)
else:
logger.debug('create_policy: object %s already exists' \
% object_creator_role_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create role with name %s due to %s' \
% (object_creator_role_name, err))
raise err
action_creator_role = None
try:
action_creator_role, created = \
Role.objects.get_or_create(name=action_creator_role_name)
action_creator_role.save()
if created:
logger.debug('create_policy: creation of %s' \
% action_creator_role_name)
else:
logger.debug('create_policy: object %s already exists' \
% action_creator_role_name)
except Exception, err:
logger.critical('create_policy: \
Unable to create role with name %s due to %s' \
% (action_creator_role_name, err))
raise err
ns = None
try:
ns, created = Namespace.objects.get_or_create(name=namespace)
ns.save()
if created:
logger.debug('create_policy: creation of %s' % namespace)
else:
logger.debug('create_policy: object %s already exists' \
% namespace)
except Exception, err:
logger.critical('create_policy: \
Unable to create namespace with name %s due to %s' \
% (namespace, err))
raise err
how = Action.objects.get(name='administration')
t_who = ContentType.objects.get_for_model(admin_role)
t_what = ContentType.objects.get_for_model(admin_view)
t_how = ContentType.objects.get_for_model(how)
try:
logger.debug('create_policy: %s %s %s %s %s %s'
% (t_who.id, admin_role.id,
t_what.id, admin_view.id,
t_how.id, how.id))
p = AcsPermission.objects.get(content_type_who__pk=t_who.id,
object_id_who=admin_role.id,
content_type_what__pk=t_what.id,
object_id_what=admin_view.id,
content_type_how__pk=t_how.id,
object_id_how=how.id)
if not restore:
logger.debug('create_policy: already existing permission %s' \
% p)
raise Exception('Policy permission already exists')
logger.debug('create_policy: existing permission %s' % p)
except ObjectDoesNotExist:
p = AcsPermission(who=admin_role, what=admin_view, how=how)
p.save()
logger.debug('create_policy: creation of permission %s' % p)
except Exception, err:
logger.critical('create_policy: \
Unable to create the admin permission due to %s' % err)
raise err
try:
policy, created = Policy.objects.get_or_create(name=name,
namespace=ns,
admin_view=admin_view,
admin_role=admin_role,
user_admin_role=user_admin_role,
abac_admin_role=abac_admin_role,
object_creator_role=object_creator_role,
action_creator_role=action_creator_role)
policy.save()
if created:
policy_created = True
logger.debug('create_policy: creation of %s' % policy)
else:
logger.debug('create_policy: object %s already exists' \
% policy)
except Exception, err:
logger.critical('create_policy: \
Unable to create policy with name %s due to %s' \
% (name, err))
raise err
'''
Add the policy view and admin role
to the admin view of the creator
Creator is a root admin only for now.
So we can push that on the user admin view
or direclty in the global admin view.
We don't manage permissions on the policy and namespace objects
the are cheched according to the permission
on the policy admin view
'''
root_view = View.objects.get(name='root_admin_view')
if not admin_view in root_view.views.all():
root_view.views.add(admin_view)
logger.debug('create_policy: View %s added to %s' \
% (admin_view, root_view))
if not admin_role in root_view.roles.all():
root_view.roles.add(admin_role)
logger.debug('create_policy: Role %s added to %s' \
% (admin_role, root_view))
except Exception, err:
transaction.rollback()
logger.critical('create_policy: all creation canceled due to %s'
% err)
raise err
else:
transaction.commit()
logger.debug('create_policy: everything successfully performed')
return (policy, policy_created)
@transaction.commit_manually
def remove_policy(name, user_creator):
'''
Check the user_creator is authorized to create a policy:
only root administrators
'''
if not is_root_administrator(user_creator):
logger.debug('remove_policy: \
Only root administrators can remove a policy')
return False
policy = None
try:
policy = Policy.objects.get(name=name)
except:
logger.debug('remove_policy: \
Unable to find the policy with the name %s' % name)
return False
try:
'''Remove permissions in the namespace'''
for it in AcsPermission.objects.all():
if it.who and it.what and it.how and \
it.who.namespace.id == policy.namespace.id \
or it.what.namespace.id == policy.namespace.id \
or it.how.namespace.id == policy.namespace.id:
logger.debug('remove_policy: Removing %s' % it)
it.delete()
'''Remove all objects in this namespace'''
for it in Action.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
for it in Activity.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
for it in AcsObject.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
for it in View.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
for it in Role.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
for it in UserAlias.objects.filter(namespace=policy.namespace):
logger.debug('remove_policy: Removing %s' % it)
it.delete()
t_what = ContentType.objects.get_for_model(policy.admin_view)
perms = AcsPermission.objects.filter(content_type_what__pk=t_what.id,
object_id_what=policy.admin_view.id)
for it in perms:
logger.debug('remove_policy: Removing %s' % it)
it.delete()
t_who = ContentType.objects.get_for_model(policy.admin_role)
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=policy.admin_role.id)
for it in perms:
logger.debug('remove_policy: Removing %s' % it)
it.delete()
'''Remove permissions on the admin objects'''
perms = AcsPermission.objects.filter()
'''Remove root admin role and view, namespace and policy objects'''
logger.debug('remove_policy: Removing policy.admin_role %s' \
% policy.admin_role)
policy.admin_role.delete()
logger.debug('remove_policy: Removing policy.admin_view %s' \
% policy.admin_view)
policy.admin_view.delete()
logger.debug('remove_policy: Removing policy.namespace %s' \
% policy.namespace)
policy.namespace.delete()
logger.debug('remove_policy: Removing user admin role %s' \
% policy.user_admin_role)
policy.user_admin_role.delete()
logger.debug('remove_policy: Removing abac admin role %s' \
% policy.abac_admin_role)
policy.user_admin_role.delete()
logger.debug('remove_policy: Removing object creator role %s' \
% policy.object_creator_role)
policy.object_creator_role.delete()
logger.debug('remove_policy: Removing action creator role %s' \
% policy.action_creator_role)
policy.action_creator_role.delete()
logger.debug('remove_policy: Removing admin roles %s' \
% policy.admin_roles)
for r in policy.admin_roles.all():
logger.debug('remove_policy: Removing admin role %s' % r)
r.delete()
logger.debug('remove_policy: Removing admin views %s' \
% policy.admin_views)
for v in policy.admin_views.all():
logger.debug('remove_policy: Removing admin view %s' % v)
v.delete()
logger.debug('remove_policy: Removing policy %s' % policy)
policy.delete()
except Exception, err:
transaction.rollback()
logger.critical('remove_policy: all deletion canceled due to %s'
% err)
return False
else:
transaction.commit()
logger.debug('remove_policy: all deletion successfully performed')
return True
'''
''''''
Manage policy
''''''
'''
def add_role(requester, name, policy):
if not requester or not name or not policy:
logger.error('add_role: Missing argument')
return -1
if not is_policy_user_administrator(requester, policy):
logger.error('add_role: requester not authorized')
return -2
try:
role, created = Role.objects.get_or_create(name=name,
namespace=policy.namespace)
try:
policy.admin_view.roles.add(role)
if not created:
logger.error('add_role: existing role %s' % role)
return role
logger.debug('add_role: role %s created' % role)
return role
except Exception, err:
logger.error('add_role: error adding role in policy admin view \
due to %s' % err)
return -3
except Exception, err:
logger.error('add_role: error getting or creating role \
due to %s' % err)
return -4
def mod_role(requester, target_role, policy,
users_added=[], users_removed=[],
roles_added=[], roles_removed=[]):
if not requester or not target_role:
logger.error('mod_role: Missing argument')
return -1
if is_authorized_on_alias_or_role(requester, target_role, policy):
logger.error('mod_role: requester not authorized on role %s' \
% target_role)
return -2
for user in users_removed:
if is_authorized_on_alias_or_role(requester, user, policy):
logger.error('mod_role: requester not authorized on user %s' \
% user)
return -2
try:
target_role.users.remove(user)
except Exception, err:
logger.critical('mod_role: unable to remove %s of %s due to %s' \
% (user, target_role, err))
return -3
for user in users_added:
if is_authorized_on_alias_or_role(requester, user, policy):
logger.error('mod_role: requester not authorized on user %s' \
% user)
return -2
try:
target_role.users.add(user)
except Exception, err:
logger.critical('mod_role: unable to add %s of %s due to %s' \
% (user, target_role, err))
return -4
for role in roles_removed:
if is_authorized_on_alias_or_role(requester, role, policy):
logger.error('mod_role: requester not authorized on role %s' \
% role)
return -2
try:
target_role.roles.remove(role)
except Exception, err:
logger.critical('mod_role: unable to remove %s of %s due to %s' \
% (role, target_role, err))
return -5
for role in roles_added:
if is_authorized_on_alias_or_role(requester, role, policy):
logger.error('mod_role: requester not authorized on role %s' \
% role)
return -2
try:
target_role.roles.add(role)
except Exception, err:
logger.critical('mod_role: unable to add %s of %s due to %s' \
% (role, target_role, err))
return -6
logger.debug('mod_role: role %s successfully modified' % target_role)
return 0
def is_authorized_on_alias_or_role(requester, item, policy):
if not requester or not item or not policy:
logger.error('is_authorized_on_alias_or_role: Missing argument')
return -1
administration = Action.objects.get(name='administration')
if not is_in_policy(item, policy):
logger.error('is_authorized_on_alias_or_role: % not in %s' \
% (item, policy))
return -2
if is_policy_user_administrator(requester, policy):
logger.debug('is_authorized_on_alias_or_role: %s authorized on %s by \
policy root admin role' % (requester, item))
return 0
p = isAuthorizedRBAC2(set_default_alias(requester),
item, administration)
if p:
logger.debug('is_authorized_on_alias_or_role: \
%s authorized on %s by permission' % (requester, item))
return 0
logger.debug('is_authorized_on_alias_or_role: \
%s not authorized on %s' % (requester, item))
return -1
def add_object(requester, name, policy, regex=None):
if not requester or not name or not policy:
logger.error('add_object: Missing argument')
return -1
if not is_policy_object_creator(requester, policy):
logger.error('add_object: requester not authorized')
return -2
try:
o = None
created = None
if regex:
if not is_valid_regex(regex):
return -5
o, created = AcsObject.objects.get_or_create(name=name,
namespace=policy.namespace, regex=regex)
else:
o, created = AcsObject.objects.get_or_create(name=name,
namespace=policy.namespace)
try:
policy.admin_view.acs_objects.add(o)
if not created:
logger.error('add_object: existing object %s' % o)
return o
logger.debug('add_object: object %s created' % o)
return o
except Exception, err:
logger.error('add_object: error adding object in policy admin \
view due to %s' % err)
return -3
except Exception, err:
logger.error('add_object: error getting or creating object \
due to %s' % err)
return -4
def add_view(requester, name, policy):
if not requester or not name or not policy:
logger.error('add_view: Missing argument')
return -1
if not is_policy_object_creator(requester, policy):
logger.error('add_view: requester not authorized')
return -2
try:
o = None
created = None
o, created = View.objects.get_or_create(name=name,
namespace=policy.namespace)
try:
policy.admin_view.views.add(o)
if not created:
logger.error('add_view: existing view %s' % o)
return o
logger.debug('add_view: view %s created' % o)
return o
except Exception, err:
logger.error('add_view: error adding view in policy admin \
view due to %s' % err)
return -3
except Exception, err:
logger.error('add_view: error getting or creating view \
due to %s' % err)
return -4
def mod_view(requester, target_view, policy,
users_added=[], users_removed=[],
roles_added=[], roles_removed=[],
objects_added=[], objects_removed=[],
views_added=[], views_removed=[],
actions_added=[], actions_removed=[],
activities_added=[], activities_removed=[]):
if not requester or not target_view:
logger.error('mod_view: Missing argument')
return -1
if is_authorized_on_object_or_view(requester, target_view, policy):
logger.error('mod_view: requester not authorized on view %s' \
% target_view)
return -2
'''
If not admin view, only objects and views
'''
if target_view in policy.admin_views.all() \
and (users_added or users_removed or roles_added \
or roles_removed or actions_added or actions_removed \
or activities_added or activities_removed):
logger.error('mod_view: you cannot add users, roles, actions or \
activities to normal views')
return -15
for user in users_removed:
if is_authorized_on_alias_or_role(requester, user, policy):
logger.error('mod_role: requester not authorized on user %s' \
% user)
return -2
try:
target_view.users.remove(user)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (user, target_view, err))
return -3
for user in users_added:
if is_authorized_on_alias_or_role(requester, user, policy):
logger.error('mod_role: requester not authorized on user %s' \
% user)
return -2
try:
target_view.users.add(user)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (user, target_view, err))
return -4
for role in roles_removed:
if is_authorized_on_alias_or_role(requester, role, policy):
logger.error('mod_role: requester not authorized on role %s' \
% role)
return -2
try:
target_view.roles.remove(role)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (role, target_view, err))
return -5
for role in roles_added:
if is_authorized_on_alias_or_role(requester, role, policy):
logger.error('mod_role: requester not authorized on role %s' \
% role)
return -2
try:
target_view.roles.add(role)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (role, target_view, err))
return -6
for o in objects_removed:
if is_authorized_on_object_or_view(requester, o, policy):
logger.error('mod_role: requester not authorized on object %s' \
% o)
return -2
try:
target_view.acs_objects.remove(o)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (o, target_view, err))
return -7
for o in objects_added:
if is_authorized_on_object_or_view(requester, o, policy):
logger.error('mod_role: requester not authorized on object %s' \
% o)
return -2
try:
target_view.acs_objects.add(o)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (o, target_view, err))
return -8
for view in views_removed:
if is_authorized_on_object_or_view(requester, view, policy):
logger.error('mod_view: requester not authorized on view %s' \
% view)
return -2
try:
target_view.views.remove(view)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (view, target_view, err))
return -9
for view in views_added:
if is_authorized_on_object_or_view(requester, view, policy):
logger.error('mod_view: requester not authorized on view %s' \
% view)
return -2
try:
target_view.views.add(view)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (view, target_view, err))
return -10
for action in actions_removed:
if is_authorized_on_action_or_activity(requester, action, policy):
logger.error('mod_view: requester not authorized on action %s' \
% action)
return -2
try:
target_view.actions.remove(action)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (action, target_view, err))
return -11
for action in actions_added:
if is_authorized_on_action_or_activity(requester, action, policy):
logger.error('mod_view: requester not authorized on action %s' \
% action)
return -2
try:
target_view.actions.add(action)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (action, target_view, err))
return -12
for activity in activities_removed:
if is_authorized_on_action_or_activity(requester, activity, policy):
logger.error('mod_view: requester not authorized on activity %s' \
% activity)
return -2
try:
target_view.activities.remove(activity)
except Exception, err:
logger.critical('mod_view: unable to remove %s of %s due to %s' \
% (activity, target_view, err))
return -13
for activity in activities_added:
if is_authorized_on_action_or_activity(requester, activity, policy):
logger.error('mod_view: requester not authorized on activity %s' \
% activity)
return -2
try:
target_view.activities.add(activity)
except Exception, err:
logger.critical('mod_view: unable to add %s of %s due to %s' \
% (activity, target_view, err))
return -14
logger.debug('mod_view: view %s successfully modified' % target_view)
return 0
def is_authorized_on_object_or_view(requester, item, policy):
if not requester or not item or not policy:
logger.error('is_authorized_on_object_or_view: Missing argument')
return -1
administration = Action.objects.get(name='administration')
if not is_in_policy(item, policy):
logger.error('is_authorized_on_object_or_view: % not in %s' \
% (item, policy))
return -2
if is_policy_object_creator(requester, policy):
logger.debug('is_authorized_on_object_or_view: %s authorized on %s \
by policy root admin role' % (requester, item))
return 0
p = isAuthorizedRBAC2(set_default_alias(requester),
item, administration)
if p:
logger.debug('is_authorized_on_object_or_view: \
%s authorized on %s by permission' % (requester, item))
return 0
logger.debug('is_authorized_on_object_or_view: \
%s not authorized on %s' % (requester, item))
return -1
def add_action(requester, name, policy):
if not requester or not name or not policy:
logger.error('add_action: Missing argument')
return -1
if not is_policy_object_creator(requester, policy):
logger.error('add_action: requester not authorized')
return -2
try:
action, created = Action.objects.get_or_create(name=name,
namespace=policy.namespace)
try:
policy.admin_view.actions.add(action)
if not created:
logger.error('add_action: existing action %s' % action)
return action
logger.debug('add_action: action %s created' % action)
return action
except Exception, err:
logger.error('add_action: error adding action in policy admin \
view due to %s' % err)
return -3
except Exception, err:
logger.error('add_action: error getting or creating action \
due to %s' % err)
return -4
def add_activity(requester, name, policy):
if not requester or not name or not policy:
logger.error('add_activity: Missing argument')
return -1
if not is_policy_action_creator(requester, policy):
logger.error('add_activity: requester not authorized')
return -2
try:
o = None
created = None
o, created = Activity.objects.get_or_create(name=name,
namespace=policy.namespace)
try:
policy.admin_view.activities.add(o)
if not created:
logger.error('add_activity: existing activity %s' % o)
return o
logger.debug('add_activity: activity %s created' % o)
return o
except Exception, err:
logger.error('add_activity: error adding activity in policy admin \
view due to %s' % err)
return -3
except Exception, err:
logger.error('add_activity: error getting or creating activity \
due to %s' % err)
return -4
def mod_activity(requester, target_activity, policy,
actions_added=[], actions_removed=[],
activities_added=[], activities_removed=[]):
if not requester or not target_activity:
logger.error('mod_activity: Missing argument')
return -1
if is_authorized_on_action_or_activity(requester, target_activity, policy):
logger.error('mod_activity: requester not authorized on activity %s' \
% target_activity)
return -2
for action in actions_removed:
if is_authorized_on_action_or_activity(requester, action, policy):
logger.error('mod_activity: requester not authorized on action %s' \
% action)
return -2
try:
target_activity.actions.remove(action)
except Exception, err:
logger.critical('mod_activity: unable to remove %s of %s due to %s' \
% (action, target_activity, err))
return -11
for action in actions_added:
if is_authorized_on_action_or_activity(requester, action, policy):
logger.error('mod_activity: requester not authorized on action %s' \
% action)
return -2
try:
target_activity.actions.add(action)
except Exception, err:
logger.critical('mod_activity: unable to add %s of %s due to %s' \
% (action, target_activity, err))
return -12
for activity in activities_removed:
if is_authorized_on_action_or_activity(requester, activity, policy):
logger.error('mod_activity: requester not authorized on activity %s' \
% activity)
return -2
try:
target_activity.activities.remove(activity)
except Exception, err:
logger.critical('mod_activity: unable to remove %s of %s due to %s' \
% (activity, target_activity, err))
return -13
for activity in activities_added:
if is_authorized_on_action_or_activity(requester, activity, policy):
logger.error('mod_activity: requester not authorized on activity %s' \
% activity)
return -2
try:
target_activity.activities.add(activity)
except Exception, err:
logger.critical('mod_activity: unable to add %s of %s due to %s' \
% (activity, target_activity, err))
return -14
logger.debug('mod_activity: activity %s successfully modified' % target_activity)
return 0
def is_authorized_on_action_or_activity(requester, item, policy):
if not requester or not item or not policy:
logger.error('is_authorized_on_action_or_activity: Missing argument')
return -1
administration = Action.objects.get(name='administration')
if not is_in_policy(item, policy):
logger.error('is_authorized_on_action_or_activity: % not in %s' \
% (item, policy))
return -2
if is_policy_action_creator(requester, policy):
logger.debug('is_authorized_on_action_or_activity: %s authorized on \
%s by policy root admin role' % (requester, item))
return 0
p = isAuthorizedRBAC2(set_default_alias(requester),
item, administration)
if p:
logger.debug('is_authorized_on_action_or_activity: \
%s authorized on %s by permission' % (requester, item))
return 0
logger.debug('is_authorized_on_action_or_activity: \
%s not authorized on %s' % (requester, item))
return -1
def add_permission(requester, policy, who, what, how, delegable=False):
if not requester or not policy or not who or not what or not how:
logger.error('add_permission: Missing argument')
return -1
if is_authorized_on_alias_or_role(requester, who, policy) < 0:
return -2
if is_authorized_on_object_or_view(requester, what, policy) < 0:
return -3
if is_authorized_on_action_or_activity(requester, how, policy) < 0:
return -4
try:
t_who = ContentType.objects.get_for_model(who)
t_what = ContentType.objects.get_for_model(what)
t_how = ContentType.objects.get_for_model(how)
p = \
AcsPermission.objects.get(content_type_who__pk=t_who.id,
object_id_who=who.id,
content_type_what__pk=t_what.id,
object_id_what=what.id,
content_type_how__pk=t_how.id,
object_id_how=how.id)
logger.error('add_permission: \
permission %s already exists' % p)
return p
except ObjectDoesNotExist:
try:
p = AcsPermission(who=who, what=what, how=how,
delegable=delegable)
p.save()
logger.debug('add_permission: \
%s added permission %s' % (requester, p))
return p
except Exception, err:
logger.debug('add_permission: \
Error adding permission by %s due to %s' % (requester, err))
return -5
except Exception, err:
logger.debug('add_permission: \
Error looking for permission by %s due to %s' % (requester, err))
return -6
'''
''''''
Helper functions
''''''
'''
def get_from_pk(type_entity, pk):
'''
Return entity from entity_type and its primary key (id).
'''
if type_entity == 'User':
try:
return User.objects.get(id=pk)
except:
return None
elif type_entity == 'UserAlias':
try:
return UserAlias.objects.get(id=pk)
except:
return None
elif type_entity == 'Role':
try:
return Role.objects.get(id=pk)
except:
return None
elif type_entity == 'AcsObject':
try:
return AcsObject.objects.get(id=pk)
except:
return None
elif type_entity == 'View':
try:
return View.objects.get(id=pk)
except:
return None
elif type_entity == 'Action':
try:
return Action.objects.get(id=pk)
except:
return None
elif type_entity == 'Activity':
try:
return Activity.objects.get(id=pk)
except:
return None
return None
def get_from_name(type_entity, name, namespace=None):
'''
Return entity from entity_type and its name and namespace.
'''
if not namespace:
namespace = Namespace.objects.get(name='Default')
if type_entity == 'UserAlias':
try:
return UserAlias.objects.get(name=name, namespace=namespace)
except:
return None
elif type_entity == 'Role':
try:
return Role.objects.get(name=name, namespace=namespace)
except:
return None
elif type_entity == 'AcsObject':
try:
return AcsObject.objects.get(name=name, namespace=namespace)
except:
return None
elif type_entity == 'View':
try:
return View.objects.get(name=name, namespace=namespace)
except:
return None
elif type_entity == 'Action':
try:
return Action.objects.get(name=name, namespace=namespace)
except:
return None
elif type_entity == 'Activity':
try:
return Activity.objects.get(name=name, namespace=namespace)
except:
return None
return None
def is_entity_who(entity):
return isinstance(entity, (UserAlias, Role))
def is_entity_what(entity):
return isinstance(entity, (AcsObject, View))
def is_entity_what_admin(entity):
return isinstance(entity,
(AcsObject, View, UserAlias, Role, Action, Activity))
def is_entity_how(entity):
return isinstance(entity, (Action, Activity))
'''
Here we return all the views on which the user has admin rights.
It includes:
- (1) Permissions direclty set to this user on a view
- (2) Permissions set to the roles of this user on a view
- (3) all the subviews
@user: It is the user alias.
'''
def return_list_views_authorized_for_admin(user):
items = []
administration = Action.objects.get(name='administration')
'''(1)'''
t_who = ContentType.objects.get_for_model(UserAlias)
t_what = ContentType.objects.get_for_model(View)
t_how = ContentType.objects.get_for_model(Action)
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=user.id,
content_type_what__pk=t_what.id,
content_type_how__pk=t_how.id,
object_id_how=administration.id)
for p in perms:
if p.what not in items:
items.append(p.what)
'''(2)'''
stack = stack_of_roles_from_user(user)
t_who = ContentType.objects.get_for_model(Role)
for role in stack:
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=role.id,
content_type_what__pk=t_what.id,
content_type_how__pk=t_how.id,
object_id_how=administration.id)
for p in perms:
if p.what not in items:
items.append(p.what)
'''(3)'''
return stack_of_views_from_list(items)
'''
Here we return all the objects of a type on which the user has admin rights.
It includes:
- (1) Permissions direclty set to this user
- (2) Permissions set to the roles of this user
- (3) The objects in views on which the user has admin rights
@user: It is the user alias.
'''
def return_list_items_authorized_for_admin(user, content_type):
items = []
administration = Action.objects.get(name='administration')
'''(1)'''
t_who = ContentType.objects.get_for_model(UserAlias)
t_how = ContentType.objects.get_for_model(Action)
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=user.id,
content_type_what__pk=content_type.id,
content_type_how__pk=t_how.id,
object_id_how=administration.id)
for p in perms:
if p.what not in items:
items.append(p.what)
'''(2)'''
stack = stack_of_roles_from_user(user)
t_who = ContentType.objects.get_for_model(Role)
for role in stack:
# All the roles, the role has direct permission on it
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=role.id,
content_type_what__pk=content_type.id,
content_type_how__pk=t_how.id,
object_id_how=administration.id)
for p in perms:
if p.what not in items:
items.append(p.what)
'''(3)'''
stack = return_list_views_authorized_for_admin(user)
if content_type == ContentType.objects.get_for_model(UserAlias):
for view in stack:
for it in view.users.all():
if it not in items:
items.append(it)
elif content_type == ContentType.objects.get_for_model(Role):
for view in stack:
for it in view.roles.all():
if it not in items:
items.append(it)
elif content_type == ContentType.objects.get_for_model(AcsObject):
for view in stack:
for it in view.acs_objects.all():
if it not in items:
items.append(it)
elif content_type == ContentType.objects.get_for_model(Action):
for view in stack:
for it in view.actions.all():
if it not in items and it.id != administration.id:
items.append(it)
elif content_type == ContentType.objects.get_for_model(Activity):
for view in stack:
for it in view.activities.all():
if it not in items:
items.append(it)
else:
raise Exception
return items
def return_list_users_authorized_for_admin(user):
return return_list_items_authorized_for_admin(user,
ContentType.objects.get_for_model(UserAlias))
def return_list_roles_authorized_for_admin(user):
return return_list_items_authorized_for_admin(user,
ContentType.objects.get_for_model(Role))
def return_list_objects_authorized_for_admin(user):
return return_list_items_authorized_for_admin(user,
ContentType.objects.get_for_model(AcsObject))
def return_list_actions_authorized_for_admin(user):
return return_list_items_authorized_for_admin(user,
ContentType.objects.get_for_model(Action))
def return_list_activities_authorized_for_admin(user):
return return_list_items_authorized_for_admin(user,
ContentType.objects.get_for_model(Activity))
def list_of_direct_roles_of_a_user(user):
items = []
roles = Role.objects.all()
for role in roles:
if user in role.users.all():
if role not in items:
items.append(role)
return items
def stack_of_roles_from_role(role):
stack_treated = []
roles_stack = []
roles_stack.append(role)
while roles_stack:
stack_tmp = []
for role in roles_stack:
roles_stack.remove(role)
stack_treated.append(role)
news = role.roles.all()
for r in news:
if r not in stack_treated and \
r not in roles_stack:
stack_tmp.append(r)
roles_stack += stack_tmp
return stack_treated
def stack_of_roles_from_user(user):
stack_treated = []
roles_stack = list_of_direct_roles_of_a_user(user)
while roles_stack:
stack_tmp = []
for role in roles_stack:
roles_stack.remove(role)
stack_treated.append(role)
news = role.roles.all()
for r in news:
if r not in stack_treated and \
r not in roles_stack:
stack_tmp.append(r)
roles_stack += stack_tmp
return stack_treated
def stack_of_views_from_view(view):
stack_treated = []
views_stack = []
views_stack.append(view)
while views_stack:
stack_tmp = []
for view in views_stack:
views_stack.remove(view)
stack_treated.append(view)
news = view.views.all()
for v in news:
if v not in stack_treated and \
v not in views_stack:
stack_tmp.append(v)
views_stack += stack_tmp
return stack_treated
def stack_of_views_from_list(list_views):
stack_treated = []
views_stack = list_views
while views_stack:
stack_tmp = []
for view in views_stack:
views_stack.remove(view)
stack_treated.append(view)
news = view.views.all()
for v in news:
if v not in stack_treated and \
v not in views_stack:
stack_tmp.append(v)
views_stack += stack_tmp
return stack_treated
def stack_of_activities_from_activity(activity):
stack_treated = []
activities_stack = []
activities_stack.append(activity)
while activities_stack:
stack_tmp = []
for activity in activities_stack:
activities_stack.remove(activity)
stack_treated.append(activity)
news = activity.activities.all()
for v in news:
if v not in stack_treated and \
v not in activities_stack:
stack_tmp.append(v)
activities_stack += stack_tmp
return stack_treated
def is_permission_delegable(permission):
'''
TODO: Check expiracy date also
'''
if permission.delegable:
return True
return False
def is_permission_delegated(permission):
if permission.permission_source:
return True
return False
def is_permission_delegated_by_user(alias, permission):
if alias.self_admin and alias.delegated_permissions and \
permission in alias.delegated_permissions.all():
return True
return False
def return_all_permissions_delegatable_by_objects(alias, policy):
'''
return values:
[] : permissions
raise
User not authorized for self administration
'''
if not is_self_admin(alias):
raise Exception('User not authorized for self administration')
logger.critical('test %s %s' % (alias, policy))
permissions = []
'''(1)'''
t_who = ContentType.objects.get_for_model(UserAlias)
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=alias.id)
for p in perms:
if is_permission_delegable(p):
permissions.append(p)
'''(2)'''
stack = stack_of_roles_from_user(alias)
t_who = ContentType.objects.get_for_model(Role)
for role in stack:
perms = AcsPermission.objects.filter(content_type_who__pk=t_who.id,
object_id_who=role.id)
for p in perms:
if is_permission_delegable(p):
permissions.append(p)
logger.critical('permissions %s' % permissions)
return permissions
def return_all_permissions_delegatable_by_names(alias, policy):
'''
by names means get objets by string names
return values:
[] : permissions
raise
Policy not found
Multiple policies found
User not found
Multiple users found
User not authorized for self administration
Unknown error
'''
policy = Policy.objects.get(policy=policy)
alias = UserAlias.objects.get(alias=alias, namespace=policy.namespace)
return return_all_permissions_delegatable_by_objects(alias, policy)
@transaction.commit_manually
def add_permission_delegated_by_objects(policy, grantor, grantee,
permission_source, what, how,
delegable=False, expiration_date=None):
'''
Grantor is a user
Grantee is a user or a role
return values:
None or the new permission
raise
Grantor not in policy
Grantee not in policy
What not in policy
How not in policy
Grantor not authorized for self administration
Grantor not authorized on this permission
A permission cannot be delegated from the source permission
What not in source permission
How not in source permission
'''
if grantor.namespace.id != policy.namespace.id:
raise Exception('Grantor not in policy')
if grantee.namespace.id != policy.namespace.id:
raise Exception('Grantee not in policy')
if not is_self_admin(grantor):
raise Exception('User not authorized for self administration')
grantor_authz = False
if permission_source:
if isinstance(permission_source.who, UserAlias) and \
permission_source.who.id == grantor.id:
grantor_authz = True
if not grantor_authz and \
isinstance(permission_source.who, Role):
roles = stack_of_roles_from_user(grantor)
if permission_source.who in roles:
grantor_authz = True
if not grantor_authz:
raise Exception('Grantor not authorized on this permission')
if not is_permission_delegable(permission_source):
raise Exception('A permission cannot be delegated from the source \
permission')
what_in_permission = False
if what:
if what.namespace.id != policy.namespace.id:
raise Exception('What not in policy')
if isinstance(permission_source.what, AcsObject) and \
isinstance(what, AcsObject) and \
permission_source.what.id == what.id:
what_in_permission = True
if not what_in_permission and \
not isinstance(permission_source.what, AcsObject):
'''
permission_source.what is a view
what is a view or an object
'''
views = stack_of_views_from_view(permission_source.what)
for v in views:
if not what_in_permission:
if isinstance(what, View) and what.id == v.id:
what_in_permission = True
elif isinstance(what, AcsObject) and \
what in v.acs_objects.all():
what_in_permission = True
if not what_in_permission:
raise Exception('What not in source permission')
how_in_permission = False
if how:
if how.namespace.id != policy.namespace.id:
raise Exception('How not in policy')
if isinstance(permission_source.how, Action) and \
isinstance(how, Action) and \
permission_source.how.id == how.id:
how_in_permission = True
if not how_in_permission and \
not isinstance(permission_source.how, Action):
'''
permission_source.how is a view
how is a view or an object
'''
activities = \
stack_of_activities_from_activity(permission_source.how)
for v in activities:
if not how_in_permission:
if isinstance(how, Activity) and how.id == v.id:
how_in_permission = True
elif isinstance(how, Action) and \
how in v.actions.all():
how_in_permission = True
if not how_in_permission:
raise Exception('How not in source permission')
'''
Check that expiration_date is <= to the source permission
expiration_date
'''
p = None
try:
p = AcsPermission(who=grantee, what=what, how=how,
permission_source=permission_source, delegable=delegable,
expiration_date=expiration_date)
p.save()
grantor.delegated_permissions.add(p)
except Exception, err:
transaction.rollback()
logger.error('add_permission_delegated_by_objects: error due to %s'
% err)
raise err
else:
transaction.commit()
logger.debug('create_policy: all creation successfully performed')
return p
@transaction.commit_manually
def add_permission_delegated_by_names(policy, grantor, grantee,
permission_source, what, what_type, how, how_type,
delegable=False, expiration_date=None):
'''
return values:
True or False
raise
Policy not found
Multiple policies found
Grantor not found
Multiple grantors found
Grantee not found
Multiple grantees found
Source permision not found
Multiple source permisions found
What not found
Multiple whats found
How not found
Multiple hows found
Unknown error
'''
#TODO: grab objects
policy = None
grantor = None
grantee = None
permission_source = None
what = None
how = None
return add_permission_delegated_by_objects(policy, grantor, grantee,
permission_source, what, how,
delegable, expiration_date)
def get_delegated_permissions(user):
return user.delegated_permissions.all()
def return_all_derived_delegated_permissions(permission):
stack = []
stack_tmp = []
stack_tmp.append(permission)
while stack_tmp:
p = stack_tmp.pop()
if not p in stack:
stack.append(p)
perms = AcsPermission.objects.filter(permission_source=p)
for perm in perms:
if not perm in stack_tmp:
stack_tmp.append(perm)
return stack
View Git Blame Copy Permalink