entrouvert
/
veridic
Archived
17
0
Fork 0
Code Issues Pull Requests Releases Activity

Add support of LDAP sources of attributes 

- New model for LDAP sources
    - LDAP attributes are provided with the attribute signal
    - Multiple sources supported
    - Add of a new source with the Web interface
    - User alias per source support added
    - Define alias per user per source with the Web interface
    - Accept that the aliases within a source are the User instances username
      attributes (will be usefull when the authentication on LDAP will be added)
This commit is contained in:
Mikaël Ates 2011-09-07 14:03:56 +02:00
parent ee505be1e7
commit 6796211d01
12 changed files with 613 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
acs/abac/core.py
View File

@ -25,6 +25,8 @@ import time
from django.db import transaction
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned
from acs.models import Namespace, UserAlias
from acs.abac.models import *
from acs.xacml.constants import *
@ -574,7 +576,8 @@ def add_assertion_to_profile(profile, source, definition, values,
'''
a = None
logger.debug('add_assertion_to_profile: Add assertion to profile %s \
- (%s, %s, %s)' % (profile, source, definition, str(values)))
- (%s, %s, %s)' % (profile, source, definition, values))
try:
data = None
try:
@ -1043,3 +1046,44 @@ def remove_rule(rule):
else:
transaction.commit()
logger.debug('remove_rule: rule deleted')
def get_identifier_in_source(user, source):
if not user:
logger.error('get_identifier_in_source: no user provided')
return None
if not source:
logger.error('get_identifier_in_source: no source provided')
return None
'''
If the source is used for authentication and the option
is_auth_backend is selected, we consider that the username is the
identifier in the ldap, i.e. the dn
We may do better with django_to_ldap_username() of
http://packages.python.org/django-auth-ldap/
'''
if source.is_auth_backend:
return user.username
ns = None
try:
ns = Namespace.objects.get(name=source.name)
except ObjectDoesNotExist:
logger.error('get_identifier_in_source: no corresponding namespace')
return None
except MultipleObjectsReturned:
logger.critical('get_identifier_in_source: multiple namespaces \
corresponding to source %s' % source)
return None
try:
ua = UserAlias.objects.get(user=user, namespace=ns)
logger.debug('get_identifier_in_source: user has alias %s in source \
%s' % (ua.alias, source.name))
return ua.alias
except ObjectDoesNotExist:
logger.error('get_identifier_in_source: user has no correspondiong \
alias for that source')
return None
except MultipleObjectsReturned:
logger.critical('get_identifier_in_source: multiple aliases of user \
%s for the source %s' % (user, source))
return None

29
acs/abac/models.py
View File

@ -27,21 +27,30 @@ from django.utils.translation import ugettext as _
from acs.xacml.constants import *
SOURCE_TYPE = (
('DIRECT', _('Direct trusted source')),
('ANCHOR', _('Trust anchor')))
class Source(models.Model):
name = models.CharField(max_length = 100, unique=True)
public_key_or_ssl_certificate = models.TextField(blank=True)
type_source = models.CharField(
max_length = 60, choices = SOURCE_TYPE,
verbose_name = '',
default = 'DIRECT')
def __unicode__(self):
return 'Source %s of type %s' % (self.name, self.type_source)
return 'Source %s' % self.name
def get_source_instance(self):
try:
return self.ldapsource
except:
pass
return None
class LdapSource(Source):
server = models.CharField(max_length=100, unique=True)
user = models.CharField(max_length=100, blank=True, null=True)
password = models.CharField(max_length=100, blank=True, null=True)
base = models.CharField(max_length=100)
port = models.IntegerField(default=389)
ldaps = models.BooleanField(default=False)
certificate = models.TextField(blank=True)
is_auth_backend = models.BooleanField(default=False)
class AttributeDefinition(models.Model):

126
acs/alias_mgmt_views.py
View File

@ -39,6 +39,8 @@ from core import get_alias_in_policy, \
from models import UserAlias, Namespace, Policy
from abac.models import Source, LdapSource
from decorators import prevent_access_to_not_user_administrators, \
check_policy_in_session, \
check_authorized_on_users_and_roles
@ -116,7 +118,8 @@ def list_users_for_aliases(request):
tpl_parameters = \
{'list_any': users,
'authz_policies': plist,
'title': 'Pick a policy to synchronize or a user to administrate',
'title': \
'User aliases administration and self-administration enabling',
'type_entity': 'user',
'backlink': root_url}
return render_to_response('list_users_for_aliases.html', tpl_parameters,
@ -276,29 +279,66 @@ def list_aliases(request, pk=None):
messages.add_message(request, messages.ERROR,
_('Unknown user with id %s') %request.GET['id'])
return HttpResponseRedirect('/list_users_for_aliases')
sources = None
try:
sources = Source.objects.all()
except Exception, err:
logger.error('list_aliases: An error occurred looking for \
sources: %s' % err)
logger.debug('list_aliases: sources found %s' % sources)
source_names = [source.name for source in sources]
'''
Aliases in policies
'''
aliases = get_aliases_of_user_without_default(user)
namespaces_in = []
namespaces_all = []
namespaces_not_in = []
list_any = []
for a in aliases:
namespaces_in.append(a.namespace)
if is_self_admin(a):
list_any.append((a, True))
else:
list_any.append((a, False))
if a.namespace.name not in source_names:
namespaces_in.append(a.namespace)
if is_self_admin(a):
list_any.append((a, True))
else:
list_any.append((a, False))
policies = Policy.objects.all()
for p in policies:
if is_user_administrator(request.user) \
or is_policy_root_administrator(request.user, p):
namespaces_all.append(p.namespace)
namespaces_not_in = list(set(namespaces_all) - set(namespaces_in))
'''
Aliases in sources
'''
aliases_sources = []
for source in sources:
if not (isinstance(source.get_source_instance(), LdapSource) \
and source.get_source_instance().is_auth_backend):
ns = None
try:
ns = Namespace.objects.get(name=source.name)
except Exception, err:
logger.error('list_aliases: An error occurred \
looking for the namespace of %s: %s\
sources: %s' % (source.name, err))
else:
try:
u = UserAlias.objects.get(user=user, namespace=ns)
aliases_sources.append((ns, u))
except:
aliases_sources.append((ns, None))
back_url = '/list_aliases?id=%s' %pk
tpl_parameters = {'namespaces_not_in': namespaces_not_in,
'list_any': list_any,
'title': _('Add or remove an alias of %s') %user,
'user': user,
'type_entity': 'alias',
'aliases_sources': aliases_sources,
'back_url': back_url,
'backlink': 'list_users_for_aliases'}
return render_to_response('list_aliases_and_add.html',
@ -372,7 +412,7 @@ def set_user_in_policy_from_home(request):
already has the alias %s in this namespace' %(user, a))
messages.add_message(request, messages.ERROR,
_('The user already has the alias %s in this namespace') %a)
return HttpResponseRedirect('mod_policy?id=' + str(policy.id))
return HttpResponseRedirect(backurl)
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
@ -382,7 +422,7 @@ def set_user_in_policy_from_home(request):
_('The user already has multiple aliases in this namespace'))
return HttpResponseRedirect(backurl)
if not 'alias' in request.POST and not request.POST['alias']:
if not 'alias' in request.POST or not request.POST['alias']:
alias = user.username
else:
alias = request.POST['alias']
@ -407,6 +447,76 @@ def set_user_in_policy_from_home(request):
HttpResponseRedirect('/list_users_for_aliases')
@csrf_exempt
@prevent_access_to_not_user_administrators
def set_user_in_source(request):
'''The admin must have admin right on this user'''
if request.method == 'POST':
if not 'alias' in request.POST or not 'user_id' in request.POST \
or not 'namespace_id' in request.POST:
logger.error('set_user_in_namespace_from_home: \
missing data from %s' %request.user)
messages.add_message(request, messages.ERROR, _('Invalid action'))
HttpResponseRedirect('/list_users_for_aliases')
user = None
try:
user = User.objects.get(id=request.POST['user_id'])
except:
logger.error('set_user_in_source: \
unknown user with id %s' %request.POST['user_id'])
messages.add_message(request, messages.ERROR, _('Unkown user'))
HttpResponseRedirect('/list_users_for_aliases')
backurl = '/list_aliases?id=' + request.POST['user_id']
ns = None
try:
ns = Namespace.objects.get(id=request.POST['namespace_id'])
except:
logger.error('set_user_in_source: \
unknown namespace with id %s' %request.POST['namespace_id'])
messages.add_message(request, messages.ERROR,
_('Unkown namespace'))
return HttpResponseRedirect(backurl)
'''Create alias'''
try:
a = UserAlias.objects.get(user=user, namespace=ns)
logger.error('set_user_in_source: the user %s \
already has the alias %s in this namespace' %(user, a))
messages.add_message(request, messages.ERROR,
_('The user already has the alias %s in this namespace') %a)
return HttpResponseRedirect(backurl)
except ObjectDoesNotExist:
pass
except MultipleObjectsReturned:
logger.critical('set_user_in_source: \
the user has mulitple aliases in this namespace')
messages.add_message(request, messages.ERROR,
_('The user already has multiple aliases in this namespace'))
return HttpResponseRedirect(backurl)
if not 'alias' in request.POST or not request.POST['alias']:
alias = user.username
else:
alias = request.POST['alias']
ua = None
try:
ua = UserAlias(alias=alias, user=user, namespace=ns)
ua.save()
except:
messages.add_message(request, messages.ERROR,
_('Unable to add %s for %s in %s') %(alias, user, ns))
return HttpResponseRedirect(backurl)
messages.add_message(request, messages.INFO,
_('User %s added in %s') %(ua, ns))
return HttpResponseRedirect(backurl)
messages.add_message(request, messages.ERROR, _('Unknown action'))
HttpResponseRedirect('/list_users_for_aliases')
@csrf_exempt
@check_policy_in_session
@check_authorized_on_users_and_roles

26
acs/core.py
View File

@ -677,9 +677,9 @@ def is_authorized_by_names_with_abac(requestor_name, who_name, what_name,
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 who.user != requestor):
logger.debug("The requester %s is different from who %s that means that the requester must \
be authorized on the parameters of its request" % (requestor, who.user))
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')
p = isAuthorizedRBAC2(set_default_alias(requestor), what, administration)
if not is_policy_object_creator(requestor, policy) and not p:
@ -698,20 +698,16 @@ def is_authorized_by_names_with_abac(requestor_name, who_name, what_name,
% (requestor, who))
return (False, None, -11)
if not who.user:
# We should remove the possibility to have a UserAlias not pointing to a
# user
logger.critical('is_authorized_by_names_with_abac: \
No user associated with that alias %s' % who)
return (False, None, -12)
'''
Attribute loading in profile object
'''
from abac.core import load_profile_by_dic, load_or_create_user_profile
profile = load_or_create_user_profile(user=who.user)
if who.user:
profile = load_or_create_user_profile(user=who.user)
else:
profile = load_or_create_user_profile(user=None)
if not profile:
logger.critical('is_authorized_by_names_with_abac: \
Error to create or load profile for %s' % who)
@ -738,8 +734,12 @@ def is_authorized_by_names_with_abac(requestor_name, who_name, what_name,
load_profile_by_dic(profile, request.session['attributes'])
if who and not no_attribute_signal:
attributes_provided = attributes_call.send(sender=None,
request=request, user=who)
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 = {}

111
acs/deletion_view.py
View File

@ -17,6 +17,7 @@
along with this program. If not, see <http://www.gnu.org/licenses/>.
'''
import logging
from django.http import HttpResponseRedirect
@ -41,7 +42,11 @@ from alias_mgmt_views import list_aliases
from utils_views import get_policy_from_session
from models import UserAlias, Role, AcsObject, View, Action, Activity, \
Namespace, AcsPermission, Policy
Namespace, AcsPermission, Policy, AcsAbacPermission
from abac.models import Source, PredicateRequired, PredicateComparison
from acs.abac.core import remove_rule
from decorators import prevent_access_to_normal_users
@ -96,7 +101,7 @@ def del_any(request):
'''
Check authorization to delete this item
The speccial user administrator role is needed because here we
The special user administrator role is needed because here we
manage users even if they are not declared in ACS (empowered)
'''
if not is_user_administrator(request.user):
@ -389,6 +394,57 @@ def del_any(request):
ContentType.objects.get_for_model(administration).id,
object_id_how=administration.id))
elif type_entity == 'source':
if not is_user_administrator(request.user):
messages.add_message(request, messages.ERROR,
_('You are not authorized to delete a source of attributes'))
return list_any(request, type_entity)
try:
item = Source.objects.get(id=pk)
except:
messages.add_message(request, messages.ERROR,
_('Unable to find the source'))
return list_any(request, type_entity)
'''
Grab all ABAC permissions that have a predicate with that
source
'''
ats = item.attachedsource_set.all()
preds_req = []
for at in ats:
tmp = \
PredicateRequired.objects.filter(definition=at.assertion)
for t in tmp:
if t not in preds_req:
preds_req.append(t)
preds_cmp = []
for at in ats:
tmp = \
PredicateComparison.objects.filter(operand1=at.assertion)
for t in tmp:
if t not in preds_cmp:
preds_cmp.append(t)
tmp = \
PredicateComparison.objects.filter(operand2=at.assertion)
for t in tmp:
if t not in preds_cmp:
preds_cmp.append(t)
p = []
for it in preds_req:
tmp = AcsAbacPermission.objects.filter(rule=it.rule)
for t in tmp:
if t not in p:
p.append(t)
for it in preds_cmp:
tmp = AcsAbacPermission.objects.filter(rule=it.rule)
for t in tmp:
if t not in p:
p.append(t)
'''
Here we should remove assertionData of profile on such a
source
'''
else:
messages.add_message(request, messages.ERROR,
_('Unable to find the object'))
@ -413,21 +469,23 @@ def del_any(request):
We check here that the user has the permission to delete the
permissions that should be removed if the object is removed
'''
if not check_permission(request.user, policy, it):
if not it in denied:
denied.append(it)
else:
delegateds = return_all_derived_delegated_permissions(it)
if len(delegateds) > 1:
logger.debug('del_permission: delegated permissions %s' \
% delegateds)
for i in delegateds:
if not i in collateral_damages:
collateral_damages.append(i)
if type_entity != 'source':
if not check_permission(request.user, policy, it):
if not it in denied:
denied.append(it)
else:
logger.debug('del_permission: no delegated permissions')
if not it in collateral_damages:
collateral_damages.append(it)
delegateds = return_all_derived_delegated_permissions(it)
if len(delegateds) > 1:
logger.debug('del_permission: delegated permissions \
%s' % delegateds)
for i in delegateds:
if not i in collateral_damages:
collateral_damages.append(i)
else:
logger.debug('del_permission: no delegated \
permissions')
if not it in collateral_damages:
collateral_damages.append(it)
if denied:
messages.add_message(request, messages.ERROR,
@ -456,6 +514,8 @@ def del_any(request):
'''Here we remove the permissions by hand'''
for it in collateral_damages:
try:
if type_entity == 'source':
remove_rule(it.rule)
it.delete()
messages.add_message(request, messages.INFO,
_('Object %s successfully deleted') %it)
@ -466,6 +526,25 @@ def del_any(request):
_('Unable to delete the the object: %s') %it)
try:
if type_entity == 'user':
for u in UserAlias.objects.filter(user=item):
logger.debug('del_any_final: User %s will be \
deleted - deletion of alias: %s' % (item, u))
u.delete()
elif type_entity == 'source':
try:
ns = Namespace.objects.get(name=item.name)
for u in UserAlias.objects.filter(namespace=ns):
logger.debug('del_any_final: Source %s will be \
deleted - deletion of alias: %s' % (item, u))
u.delete()
logger.debug('del_any_final: Source %s will be \
deleted - deletion of namespace: %s' % (item, ns))
ns.delete()
except Exception, err:
logger.error('del_any_final: \
Error removing a source due to: %s' % err)
pass
item.delete()
except:
logger.error('del_any_final: \

58
acs/forms.py
View File

@ -25,7 +25,7 @@ from registration.forms import RegistrationForm
from models import Action, Activity, AcsObject, Role, View
from abac.models import Source
from abac.models import Source, LdapSource
attrs_dict = {'class': 'required'}
@ -273,6 +273,62 @@ class AddSourceForm(forms.ModelForm):
return source
class AddLdapSourceForm(forms.ModelForm):
name = forms.RegexField(label=_("name"),
max_length=30, regex=r'^[\w.@+-]+$',
help_text = \
_("30 characters or fewer. Letters, digits and @/./+/-/_ only."),
error_messages = \
{'invalid': _("This value may contain only letters, \
numbers and @/./+/-/_ characters.")})
server = forms.RegexField(label=_("LDAP host"),
max_length=100, regex=r'^[\w.@+-]+$',
help_text = \
_("Provide a hostname or an IP address."),
error_messages = \
{'invalid': _("This value may contain only letters, \
numbers and @/./+/-/_ characters.")})
user = forms.RegexField(label=_("Username"), required=False,
max_length=100, regex=r'^[\w.@+-]+$',
help_text = \
_("Provide a user account if it is necessary to authenticate for binding."),
error_messages = \
{'invalid': _("This value may contain only letters, \
numbers and @/./+/-/_ characters.")})
password = forms.RegexField(label=_("Password"), required=False,
max_length=100, regex=r'^[\w.@+-]+$',
help_text = \
_("Provide a user account if it is necessary to authenticate for binding."),
error_messages = \
{'invalid': _("This value may contain only letters, \
numbers and @/./+/-/_ characters.")})
base = forms.RegexField(label=_("Base DN"),
max_length=100, regex=r'^[\w.@+-,=]+$',
help_text = \
_("Provide the base DN for searching, e.g. dc=organization,dc=org."),
error_messages = \
{'invalid': _("This value may contain only letters, \
numbers and @/./+/-/_ characters.")})
is_auth_backend = forms.BooleanField(required=False,
help_text = \
_("Check if this LDAP is also used as an authentication backend"))
class Meta:
model = LdapSource
fields = ("name", "server", "user", "password", "base", "is_auth_backend")
def save(self, commit=True):
source = super(AddLdapSourceForm, self).save(commit=False)
if commit:
source.save()
return source
class RoleChangeForm(forms.ModelForm):
def __init__(self, *args, **kwargs):

13
acs/main_views.py
View File

@ -53,6 +53,8 @@ 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
@ -159,10 +161,15 @@ def index(request):
list_power_services['Generic user management'] = \
{'add_user': "Add a user",
'list_users': "Modify or delete a user",
'add_abac_source': "Add a source of attributes"}
if policies:
'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 and \
'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 \

107
acs/signals.py
View File

@ -1,3 +1,108 @@
import logging
import ldap
from django.dispatch import Signal
attributes_call = Signal(providing_args = ["request","user"])
from models import UserAlias
from abac.models import LdapSource, AttributeNamespace
logger = logging.getLogger('acs')
attributes_call = Signal(providing_args = ["request", "user"])
def ldap_sources(request, user, **kwargs):
if not user:
logger.error('ldap_source: No user provided')
return {}
logger.debug('ldap_source: Searching attributes for user %s' % user)
sources = LdapSource.objects.all()
if not sources:
logger.debug('ldap_source: No LDAP source configured')
return {}
att_ns = AttributeNamespace.objects.get(friendly_name='LDAP')
attribute_namespace = att_ns.identifier
attributes = {}
for source in sources:
logger.debug('ldap_source: The LDAP source is known as %s' \
% source.name)
'''
Grab the DN
'''
identifier = None
if isinstance(user, UserAlias):
'''
If the request is on a user alias with no django user,
attributes can be retrieved if the alias is an ldap dn
'''
identifier = user.alias
else:
from abac.core import get_identifier_in_source
identifier = get_identifier_in_source(user, source)
if not identifier:
logger.error('ldap_source: No user identifier known into that \
source')
else:
logger.debug('ldap_source: the user is known as %s in source %s' \
% (identifier, source.name))
try:
l = ldap.open(source.server)
l.protocol_version = ldap.VERSION3
username = source.user
password = source.password
l.simple_bind(username, password)
except ldap.LDAPError, err:
logger.error('ldap_source: an error occured at binding due \
to %s' % err)
else:
base_dn = source.base
search_scope = ldap.SCOPE_SUBTREE
retrieve_attributes = None
dn = ldap.dn.explode_dn(identifier,
flags=ldap.DN_FORMAT_LDAPV3)
search_filter = dn[0]
logger.debug('ldap_source: rdn is %s' % search_filter)
data = []
try:
ldap_result_id = l.search(base_dn, search_scope,
search_filter, retrieve_attributes)
result_type, result_data = l.result(ldap_result_id, 0)
logger.debug('ldap_source: result %s %s' % (result_type,
result_data))
for d, dic in result_data:
logger.debug('ldap_source: found %s' % d)
if d == identifier:
logger.debug('ldap_source: Attributes are %s' \
% dic)
for key in dic.keys():
attr = {}
attr['name'] = key
attr['values'] = [\
a.decode('utf-8'). \
encode('ascii', 'ignore') \
for a in dic[key]]
attr['namespace'] = attribute_namespace
data.append(attr)
except ldap.LDAPError, err:
logger.error('ldap_source: an error occured at searching \
due to %s' % err)
else:
if not data:
logger.error('ldap_source: no attribute found')
else:
attributes[source.name] = data
logger.debug('ldap_source: the attributes returned are %s' % attributes)
return attributes
attributes_call.connect(ldap_sources)

34
acs/templates/list_aliases_and_add.html
View File

@ -18,6 +18,8 @@
</ul>
{% endif %}
{% if list_any or namespaces_not_in %}
<h3>{% trans "Manage aliases in policies" %}</h3>
{% if list_any %}
<ul>
{% for any, self_admin in list_any %}
@ -51,7 +53,7 @@
{% for ns in namespaces_not_in %}
<li>
<p><form method="post" action="/set_user_in_policy_from_home">
<p><label for="id_name">{% trans "Declare the user in the policy: " %} <strong>{{ ns }}</strong></label> {% trans "Alias" %}: <input id="id_alias" type="text" name="alias" maxlength="30" /> If not given, the username will be taken.</p>
<p><label for="id_name">{% trans "Declare the user in the policy: " %} <strong>{{ ns }}</strong></label> {% trans "Alias" %}: <input id="id_alias" type="text" name="alias" maxlength="30" value="{{ user.username }}"/> If not given, the username will be taken.</p>
<input type="hidden" name="user_id" value="{{ user.id }}"/>
<input type="hidden" name="namespace_id" value="{{ ns.id }}"/>
<input type="submit" name="{{ submit_name }}" value="{% trans "Add" %}"/>
@ -60,5 +62,35 @@
{% endfor %}
</ul>
{% endif %}
{% endif %}
{% if aliases_sources %}
<h3>{% trans "Manage aliases in sources" %}</h3>
<ul>
{% for ns, alias in aliases_sources %}
<li>
<p>{% trans "In " %}<strong>{{ ns.name }}</strong></p>
{% if alias %}
<form method="post" action="/del_any">
<input type="hidden" name="type_entity" value="{{ type_entity }}"/>
<input type="hidden" name="id" value="{{ alias.id }}"/>
{% if back_url %}<input type="hidden" name="back_url" value="{{ back_url }}"/>{% endif %}
<input type="submit" name="Delete" value="{% trans "Delete" %} {{ alias.alias }}"/>
</form>
{% else %}
<p>
<form method="post" action="/set_user_in_source">
<p><label for="id_name">{% trans "The user is known as (provide DN if LDAP):" %}</label><input id="id_alias" type="text" name="alias" maxlength="100" value="{{ user.username }}"/> If not given, the username will be taken.</p>
<input type="hidden" name="user_id" value="{{ user.id }}"/>
<input type="hidden" name="namespace_id" value="{{ ns.id }}"/>
<input type="submit" name="{{ submit_name }}" value="{% trans "Add" %}"/>
</form></p>
{% endif %}
</li>
{% endfor %}
</ul>
{% endif %}
<p><a href="{{ backlink }}">{% trans "Back" %}</a></p>
{% endblock %}

6
acs/templates/list_users_for_aliases.html
View File

@ -26,8 +26,10 @@
<p>{{ p }}</p>
<form method="post" action="">
<input id="id_policy" type="hidden" name="policy" value="{{ p.id }}"/>
<input type="submit" name="Synchronize" value="{% trans "Synchronize" %}"/>
<input type="submit" name="All_self_admin" value="{% trans "All users are self admininistrators" %}"/>
<p><label>{% trans "Declare all users in the policy with their username as alias:" %}<label></p>
<p><input type="submit" name="Synchronize" value="{% trans "Synchronize" %}"/></p>
<p><label>{% trans "Declare that users in the policy are self-administrators:" %}<label></p>
<p><input type="submit" name="All_self_admin" value="{% trans "All users are self admininistrators" %}"/></p>
</form>
</li>
{% endfor %}

5
acs/urls.py
View File

@ -71,6 +71,8 @@ urlpatterns = patterns('',
'acs.alias_mgmt_views.list_aliases_in_policy'),
url(r'^set_user_in_policy$',
'acs.alias_mgmt_views.set_user_in_policy'),
url(r'^set_user_in_source$',
'acs.alias_mgmt_views.set_user_in_source'),
url(r'^list_accesses$',
@ -124,6 +126,9 @@ urlpatterns = patterns('',
url(r'^list_abac_permissions$', 'acs.abac_views.list_abac_permissions'),
url(r'^del_abac_permission$', 'acs.abac_views.del_abac_permission'),
url(r'^add_abac_source$', 'acs.views.add_abac_source'),
url(r'^add_abac_ldap_source$', 'acs.views.add_abac_ldap_source'),
url(r'^list_abac_sources$', 'acs.views.list_abac_sources'),
url(r'^mod_source$', 'acs.views.mod_source'),
url(r'^add_admin_view$',
'acs.acs_administration_views.add_admin_view'),

112
acs/views.py
View File

@ -33,9 +33,7 @@ from django.conf import settings
from forms import AddRoleForm, AddObjectForm, AddViewForm, AddActionForm, \
AddActivityForm, RoleChangeForm, AcsObjectChangeForm, \
ViewChangeForm, ActionChangeForm, ActivityChangeForm, \
AddSourceForm
AddSourceForm, AddLdapSourceForm
from core import is_policy_action_creator, is_policy_object_creator, \
is_policy_user_administrator, is_in_policy, is_valid_regex, \
@ -55,7 +53,9 @@ from core import is_policy_action_creator, is_policy_object_creator, \
get_alias_in_policy
from models import UserAlias, Role, AcsObject, View, Action, Activity, \
AcsPermission
AcsPermission, Namespace
from abac.models import Source
from decorators import prevent_access_to_normal_users,\
check_policy_in_session, \
@ -106,6 +106,108 @@ def add_abac_source(request):
return return_add_any(request, form, title)
@csrf_exempt
@prevent_access_to_not_user_administrators
def add_abac_ldap_source(request):
title = _('Add a new ABAC LDAP source')
if request.method == 'POST':
if 'cancel' in request.POST:
messages.add_message(request, messages.INFO,
_('Operation canceled'))
return HttpResponseRedirect(root_url)
logger.debug('add_abac_ldap_source: request.POST %s ' % request.POST)
form = AddLdapSourceForm(request.POST)
logger.debug('add_abac_ldap_source: form %s ' % form)
if form.is_valid():
source = form.save()
Namespace.objects.get_or_create(name=source.name)
logger.debug('add_abac_ldap_source: Ldap source %s created' \
% source)
messages.add_message(request, messages.INFO,
_('The LDAP source %s has been correctly created.') % source)
return HttpResponseRedirect(root_url)
else:
form = AddLdapSourceForm()
return return_add_any(request, form, title)
@csrf_exempt
@prevent_access_to_not_user_administrators
def list_abac_sources(request):
title = _('Modify or delete an ABAC source')
type_entity = 'source'
objects = Source.objects.all()
return return_list_any(request, objects, title, type_entity,
backlink=root_url, back_url='list_abac_sources')
@csrf_exempt
@prevent_access_to_not_user_administrators
def mod_source(request):
form = None
source = None
if request.method == 'GET':
if 'source' in request.GET and request.GET['source']:
try:
source = Source.objects.get(id=request.GET['source'])
except:
messages.add_message(request, messages.ERROR,
_('Unknown source'))
return HttpResponseRedirect('/list_abac_sources')
else:
messages.add_message(request, messages.ERROR,
_('Unknown source'))
return HttpResponseRedirect('/list_abac_sources')
is_ldap = source.get_source_instance()
if is_ldap:
form = AddLdapSourceForm(instance=is_ldap)
else:
form = AddSourceForm(instance=source)
elif request.method == 'POST':
if 'cancel' in request.POST:
messages.add_message(request, messages.INFO,
_('Operation canceled'))
return HttpResponseRedirect('/list_abac_sources')
if 'id' in request.POST and request.POST['id']:
try:
source = Source.objects.get(id=request.POST['id'])
except:
messages.add_message(request, messages.ERROR,
_('Unknown source'))
return HttpResponseRedirect('/list_abac_sources')
else:
messages.add_message(request, messages.ERROR,
_('Unknown source'))
return HttpResponseRedirect('/list_abac_sources')
name = source.name
is_ldap = source.get_source_instance()
if is_ldap:
form = AddLdapSourceForm(request.POST, instance=is_ldap)
else:
form = AddSourceForm(request.POST, instance=source)
if form.is_valid():
s = form.save()
s.save()
messages.add_message(request, messages.INFO,
_('Source %s modified') % form.cleaned_data['name'])
else:
logger.error('mod_source: Error validating form %s' % form)
else:
messages.add_message(request, messages.ERROR,
_('Unknown HTTP method %s') %request.method)
return HttpResponseRedirect('/list_abac_sources')
title = _('Modify the source %s') % source
tpl_p = {'item': source, 'backlink': 'list_abac_sources'}
return return_mod_any(request, form, title, tpl_p=tpl_p,
template_name='mod_any_new.html')
@csrf_exempt
@prevent_access_to_not_user_administrators
def list_users(request):
@ -419,6 +521,8 @@ def list_any(request, type_entity):
return list_actions(request)
elif type_entity == 'activity':
return list_activities(request)
elif type_entity == 'source':
return list_abac_sources(request)
return HttpResponseRedirect(root_url)