495 lines
19 KiB
Python
495 lines
19 KiB
Python
'''
|
|
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
|
|
|
|
from django.contrib.auth import REDIRECT_FIELD_NAME
|
|
from django.contrib.auth.forms import AuthenticationForm
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.decorators.cache import never_cache
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.http import HttpResponseRedirect
|
|
from django.shortcuts import render_to_response
|
|
from django.contrib import messages
|
|
from django.conf import settings
|
|
from django.contrib.auth import login as auth_login
|
|
from django.contrib.auth import logout as auth_logout
|
|
from django.utils.translation import ugettext as _
|
|
from django.template import RequestContext
|
|
|
|
from core import at_least_one_permission_to_remove, \
|
|
is_policy_action_creator, is_policy_object_creator, \
|
|
is_policy_user_administrator, is_policy_abac_administrator, \
|
|
check_acs_initialized, \
|
|
is_user_administrator, is_abac_administrator, \
|
|
is_policy_root_administrator, \
|
|
create_policy, is_root_administrator, remove_policy, \
|
|
is_policy_administrator, \
|
|
at_least_one_role_to_admin, at_least_one_object_to_admin, \
|
|
at_least_one_view_to_admin, at_least_one_activity_to_admin, \
|
|
at_least_one_action_to_admin, \
|
|
at_least_one_permission_to_set, \
|
|
at_least_one_abac_permission_to_set, \
|
|
at_least_one_abac_permission_to_admin, \
|
|
get_alias_in_policy, is_self_admin, set_default_alias
|
|
|
|
from decorators import prevent_access_to_not_root_administrators
|
|
|
|
from models import Policy
|
|
|
|
from abac.models import Source
|
|
|
|
logger = logging.getLogger('acs')
|
|
|
|
login_url = settings.LOGIN_URL
|
|
root_url = settings.ROOT_URL
|
|
|
|
|
|
'''
|
|
''''''
|
|
Login/Logout views
|
|
''''''
|
|
'''
|
|
|
|
|
|
@csrf_exempt
|
|
@never_cache
|
|
def login(request):
|
|
form = None
|
|
if request.method == "POST":
|
|
form = AuthenticationForm(data=request.POST)
|
|
if form.is_valid():
|
|
auth_login(request, form.get_user())
|
|
if REDIRECT_FIELD_NAME in request.session:
|
|
return \
|
|
HttpResponseRedirect(request.session[REDIRECT_FIELD_NAME])
|
|
return HttpResponseRedirect(root_url)
|
|
elif request.method == "GET":
|
|
if REDIRECT_FIELD_NAME in request.GET:
|
|
request.session[REDIRECT_FIELD_NAME] = \
|
|
request.GET[REDIRECT_FIELD_NAME]
|
|
form = AuthenticationForm()
|
|
|
|
return render_to_response('login.html',
|
|
{'form': form, },
|
|
context_instance=RequestContext(request))
|
|
|
|
|
|
@login_required
|
|
def logout(request):
|
|
auth_logout(request)
|
|
return HttpResponseRedirect('/login')
|
|
|
|
|
|
'''
|
|
''''''
|
|
Homepage
|
|
''''''
|
|
'''
|
|
|
|
|
|
@login_required
|
|
def index(request):
|
|
tpl_parameters = {}
|
|
|
|
if not check_acs_initialized():
|
|
messages.add_message(request, messages.ERROR,
|
|
_('The ACS application is not initialized. \
|
|
Contact your administrator.'))
|
|
return render_to_response('index.html',
|
|
tpl_parameters,
|
|
context_instance=RequestContext(request))
|
|
|
|
request.session['policy'] = None
|
|
|
|
'''
|
|
Here check that the user can at least delegate
|
|
(Special model licence) for not AcsAdmin
|
|
'''
|
|
#list_services['delegate_permissions'] = \
|
|
# "Delegate some of your permission"
|
|
#list_services['remove_delegations'] = "Remove your delegations"
|
|
|
|
list_authorized_services = {}
|
|
list_power_services = {}
|
|
list_user_mgmt_services = {}
|
|
|
|
if is_root_administrator(request.user):
|
|
authz_services = {}
|
|
policies = Policy.objects.all()
|
|
authz_services['create_policy_view'] = 'Add a policy'
|
|
if policies:
|
|
authz_services['delete_policy_view'] = 'Remove a policy'
|
|
list_authorized_services['Create an access control policy'] = \
|
|
authz_services
|
|
|
|
policies = Policy.objects.all()
|
|
plist = {}
|
|
self_list = {}
|
|
for p in policies:
|
|
url = 'mod_policy?id=' + str(p.id)
|
|
if is_policy_administrator(request.user, p):
|
|
plist[url] = p.name
|
|
alias = get_alias_in_policy(request.user, p)
|
|
if alias and is_self_admin(alias):
|
|
self_list[url] = p.name
|
|
if plist:
|
|
list_authorized_services['Administrate an access control policy'] = \
|
|
plist
|
|
|
|
if self_list:
|
|
list_authorized_services['Manage permission delegation'] = \
|
|
self_list
|
|
|
|
if is_user_administrator(request.user):
|
|
list_power_services['Generic user management'] = \
|
|
{'add_user': "Add a user",
|
|
'list_users': "Modify or delete a user",
|
|
'add_abac_source': "Add a generic source of attributes",
|
|
'add_abac_ldap_source': "Add a LDAP source of attributes"}
|
|
sources = Source.objects.all()
|
|
if sources:
|
|
list_power_services['Generic user management']['list_abac_sources'] = \
|
|
'Modify a source of attributes'
|
|
if policies or sources:
|
|
list_user_mgmt_services['list_users_for_aliases'] = \
|
|
'Manage user aliases or \
|
|
enable users for permission delegation'
|
|
# list_user_mgmt_services['declare_all_users_in_policy'] = \
|
|
# 'Declare all known users in a policy \
|
|
# (username will be taken as an alias)'
|
|
|
|
if is_root_administrator(request.user):
|
|
list_user_mgmt_services['manage_root_administrators'] = \
|
|
'Add or remove root administrators'
|
|
list_user_mgmt_services['manage_user_administrators'] = \
|
|
'Add or remove user administrators'
|
|
list_user_mgmt_services['manage_abac_administrators'] = \
|
|
'Add or remove abac administrators'
|
|
if list_user_mgmt_services:
|
|
list_power_services['User management in ACS'] = \
|
|
list_user_mgmt_services
|
|
|
|
tpl_parameters['power_services'] = list_power_services
|
|
tpl_parameters['administration_services'] = list_authorized_services
|
|
|
|
list_other_services = {}
|
|
# list_other_services['ask_decision'] = "Ask for a decision"
|
|
# list_other_services['graph'] = \
|
|
# "Display your administration and access control policies (graph)"
|
|
list_exploitation_services = {}
|
|
if list_other_services:
|
|
list_exploitation_services['Consult policies'] = list_other_services
|
|
|
|
tpl_parameters['exploitation_services'] = list_exploitation_services
|
|
tpl_parameters['username'] = request.user.username
|
|
if is_root_administrator(request.user):
|
|
tpl_parameters['special_role'] = _('You are a root administrator')
|
|
elif is_user_administrator(request.user):
|
|
tpl_parameters['special_role'] = _('You are a user administrator')
|
|
elif is_abac_administrator(request.user):
|
|
tpl_parameters['special_role'] = _('You are an abac administrator')
|
|
return render_to_response('index.html',
|
|
tpl_parameters,
|
|
context_instance=RequestContext(request))
|
|
|
|
|
|
'''
|
|
''''''
|
|
General policy management views
|
|
''''''
|
|
'''
|
|
|
|
|
|
@csrf_exempt
|
|
@prevent_access_to_not_root_administrators
|
|
def create_policy_view(request):
|
|
tpl_parameters = {}
|
|
if request.method == 'POST':
|
|
if 'cancel' in request.POST:
|
|
messages.add_message(request, messages.INFO,
|
|
_('Operation canceled'))
|
|
return HttpResponseRedirect(root_url)
|
|
|
|
logger.debug('create_policy: proccess to create a new policy')
|
|
|
|
if not 'name' in request.POST or not request.POST['name']:
|
|
messages.add_message(request, messages.ERROR,
|
|
_('A name is required'))
|
|
return render_to_response('create_policy.html',
|
|
tpl_parameters,
|
|
context_instance=RequestContext(request))
|
|
|
|
name = request.POST['name']
|
|
namespace = None
|
|
if not 'namespace' in request.POST \
|
|
or not request.POST['namespace']:
|
|
namespace = request.POST['name']
|
|
else:
|
|
namespace = request.POST['namespace']
|
|
|
|
policy = None
|
|
created = False
|
|
try:
|
|
policy, created = create_policy(name, request.user, namespace)
|
|
except Exception, err:
|
|
logger.error('create_policy: Unable to create policy due to %s' \
|
|
% err)
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Unable to create the policy. \
|
|
Contact your administrator'))
|
|
if created:
|
|
messages.add_message(request, messages.INFO,
|
|
_('Policy %s successfully created') % policy)
|
|
else:
|
|
messages.add_message(request, messages.INFO,
|
|
_('Already existing policy %s, successfully restored if \
|
|
necessary') % policy)
|
|
return HttpResponseRedirect(root_url)
|
|
|
|
return render_to_response('create_policy.html',
|
|
tpl_parameters, context_instance=RequestContext(request))
|
|
|
|
|
|
@csrf_exempt
|
|
@prevent_access_to_not_root_administrators
|
|
def delete_policy_view(request):
|
|
'''Get avec un policy id, on supprime, sinon on liste'''
|
|
tpl_parameters = {}
|
|
|
|
if request.method == 'GET':
|
|
if 'id' in request.GET:
|
|
p = None
|
|
try:
|
|
p = Policy.objects.get(id=request.GET['id'])
|
|
res = remove_policy(p.name, request.user)
|
|
if res:
|
|
messages.add_message(request, messages.INFO,
|
|
_('Policy successfully deleted'))
|
|
else:
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Unable to delete policy %s') %p)
|
|
except:
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Policy not found'))
|
|
|
|
policies = Policy.objects.all()
|
|
plist = []
|
|
for p in policies:
|
|
url = 'delete_policy_view?id=%s' %p.id
|
|
plist.append((p, url))
|
|
tpl_parameters['policies'] = plist
|
|
tpl_parameters['backlink'] = root_url
|
|
return render_to_response('remove_policy.html',
|
|
tpl_parameters, context_instance=RequestContext(request))
|
|
|
|
|
|
@csrf_exempt
|
|
def mod_policy(request):
|
|
tpl_parameters = {}
|
|
|
|
if request.method == 'GET':
|
|
if not 'id' in request.GET:
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Unknown policy'))
|
|
return HttpResponseRedirect(root_url)
|
|
|
|
policy = None
|
|
try:
|
|
policy = Policy.objects.get(id=request.GET['id'])
|
|
except:
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Unknown policy'))
|
|
return HttpResponseRedirect(root_url)
|
|
|
|
if not is_policy_administrator(request.user, policy) and \
|
|
not is_self_admin(get_alias_in_policy(request.user, policy)):
|
|
messages.add_message(request, messages.ERROR,
|
|
_('You are not authorized to access %s') %policy)
|
|
return HttpResponseRedirect(root_url)
|
|
|
|
request.session['policy'] = policy
|
|
|
|
tpl_parameters['policy'] = policy
|
|
|
|
list_authorized_services = {}
|
|
|
|
if is_self_admin(get_alias_in_policy(request.user, policy)):
|
|
list_authorized_services['Manage your permissions delegations'] \
|
|
= {'list_delegated_permissions': \
|
|
'List the permissions you have already delegated',
|
|
'list_accesses':
|
|
'All the accesses you can delegate'}
|
|
|
|
if set_default_alias(request.user):
|
|
'''Not just a self admin'''
|
|
|
|
list_user_services = {}
|
|
list_abac_services = {}
|
|
list_object_services = {}
|
|
list_action_services = {}
|
|
list_services = {}
|
|
list_other_services = {}
|
|
|
|
if is_policy_user_administrator(request.user, policy):
|
|
list_user_services['list_aliases_in_policy'] = \
|
|
"Manage user aliases in the policy and \
|
|
enable users for permission delegation"
|
|
list_user_services['synchronize_users_in_policy_view'] = \
|
|
"Synchronize all users in this policy"
|
|
list_user_services['synchronize_users_in_policy_view'] = \
|
|
"Synchronize all users in this policy"
|
|
list_user_services['all_users_self_admin'] = \
|
|
"All users in this policy are set self administrators"
|
|
list_user_services['add_role'] = "Add a role"
|
|
list_other_services['graph?type_graph=whole_policy'] = \
|
|
"Display the whole policy"
|
|
|
|
if is_policy_abac_administrator(request.user, policy):
|
|
if at_least_one_abac_permission_to_set(request.user, policy):
|
|
list_abac_services['add_abac_permission'] = \
|
|
"Create an ABAC permission"
|
|
if at_least_one_abac_permission_to_admin(request.user,
|
|
policy):
|
|
list_abac_services['list_abac_permissions'] = \
|
|
"Modify or delete an ABAC permission"
|
|
|
|
if at_least_one_role_to_admin(request.user, policy):
|
|
list_user_services['list_roles'] = "Modify or delete a role"
|
|
|
|
if is_policy_object_creator(request.user, policy):
|
|
list_object_services['add_object'] = "Add an object"
|
|
list_object_services['add_view'] = "Add a view"
|
|
if at_least_one_object_to_admin(request.user, policy):
|
|
list_object_services['list_objects'] = \
|
|
"Rename or delete an object"
|
|
if at_least_one_view_to_admin(request.user, policy):
|
|
list_object_services['list_views'] = "Modify or delete a view"
|
|
|
|
if is_policy_action_creator(request.user, policy):
|
|
list_action_services['add_action'] = "Add an action"
|
|
list_action_services['add_activity'] = "Add an activity"
|
|
if at_least_one_action_to_admin(request.user, policy):
|
|
list_action_services['list_actions'] = \
|
|
"Rename or delete an action"
|
|
if at_least_one_activity_to_admin(request.user, policy):
|
|
list_action_services['list_activities'] = \
|
|
"Modify or delete an activity"
|
|
|
|
if at_least_one_permission_to_set(request.user, policy):
|
|
list_services['add_permission'] = "Add a permission"
|
|
|
|
|
|
if at_least_one_permission_to_remove(request.user, policy):
|
|
list_services['list_permissions'] = "Delete a permission"
|
|
list_other_services['ask_decision'] = "Ask for a decision"
|
|
list_other_services['ask_decision_regex'] = \
|
|
"Ask for a decision with a regex"
|
|
|
|
if list_other_services:
|
|
list_authorized_services['Request the policy'] = \
|
|
list_other_services
|
|
|
|
if list_user_services:
|
|
list_authorized_services['Users and Roles management'] = \
|
|
list_user_services
|
|
if list_abac_services:
|
|
list_authorized_services[\
|
|
'Attribute based access control management'] = \
|
|
list_abac_services
|
|
if list_object_services:
|
|
list_authorized_services['Objects and Views management'] = \
|
|
list_object_services
|
|
if list_action_services:
|
|
list_authorized_services[ \
|
|
'Actions and Activities management'] = \
|
|
list_action_services
|
|
if list_services:
|
|
list_authorized_services['Manage permissions'] = list_services
|
|
|
|
list_power_services = {}
|
|
|
|
'''
|
|
We only authorize the root policy administrators
|
|
to give administration rights to other user.
|
|
We could easily let not root administrator give an
|
|
administration rights on the entities they can administrate
|
|
but it complexifies the administration.
|
|
'''
|
|
if is_policy_root_administrator(request.user, policy):
|
|
list_power_services[\
|
|
'Manage policy administrators using root roles'] = \
|
|
{'manage_policy_administrators?type=root':
|
|
'Manage root policy administrators',
|
|
'manage_policy_administrators?type=user':
|
|
'Manage users and roles administrators of the policy',
|
|
'manage_policy_administrators?type=abac':
|
|
'Manage ABAC administrators of the policy',
|
|
'manage_policy_administrators?type=object':
|
|
'Manage objects and views creators',
|
|
'manage_policy_administrators?type=action':
|
|
'Manage actions and activities creators'}
|
|
|
|
list_power_services['Manage administration views'] = \
|
|
{'add_admin_view': "Add an administration view",
|
|
'list_admin_views':
|
|
"Manage administration rights using admin views"}
|
|
list_power_services['Manage administration roles'] = \
|
|
{'add_admin_role': "Add an administration role",
|
|
'list_admin_roles':
|
|
"Manage administration rights using admin roles"}
|
|
list_power_services['Manage administration permissions'] = \
|
|
{'add_admin_permission':
|
|
"Add an admininistration permission",
|
|
'list_admin_permissions':
|
|
"Remove an admininistration permission"}
|
|
|
|
if list_power_services:
|
|
tpl_parameters['power_services'] = list_power_services
|
|
tpl_parameters['username'] = request.user.username
|
|
if is_policy_root_administrator(request.user, policy):
|
|
tpl_parameters['special_roles'] = \
|
|
[_('You are a root administrator of this policy')]
|
|
else:
|
|
l = []
|
|
if is_policy_user_administrator(request.user, policy):
|
|
l.append(_('User and Roles administrator'))
|
|
if is_policy_abac_administrator(request.user, policy):
|
|
l.append(_('ABAC administrator'))
|
|
if is_policy_object_creator(request.user, policy):
|
|
l.append(_('Objects and Views administrator'))
|
|
if is_policy_action_creator(request.user, policy):
|
|
l.append(_('Actions and Activities administrator'))
|
|
if l:
|
|
tpl_parameters['special_roles'] = l
|
|
|
|
if list_authorized_services:
|
|
tpl_parameters['administration_services'] = \
|
|
list_authorized_services
|
|
|
|
tpl_parameters['username'] = request.user.username
|
|
|
|
return render_to_response('mod_policy.html',
|
|
tpl_parameters,
|
|
context_instance=RequestContext(request))
|
|
|
|
messages.add_message(request, messages.ERROR,
|
|
_('Unknown policy'))
|
|
return HttpResponseRedirect(root_url)
|