Disallow RDN change and single-value bypass using setattr/addattr.

When setting or adding an attribute wiht setatt/addattr check to
see if there is a Param for the attribute and enforce the multi-value.
If there is no Param check the LDAP schema for SINGLE-VALUE.

Catch RDN mods and try to return a more reasonable error message.

Ticket #230
Ticket #246
This commit is contained in:
Pavel Zuna 2010-10-14 13:05:43 -04:00 committed by Rob Crittenden
parent 267e803cdf
commit dff2ff8300
4 changed files with 79 additions and 18 deletions

View File

@ -1165,14 +1165,14 @@ class DatabaseError(ExecutionError):
For example:
>>> raise DatabaseError(desc="Can't contact LDAP server", info='')
>>> raise DatabaseError(desc="Can't contact LDAP server", info='Info goes here')
Traceback (most recent call last):
...
DatabaseError: Can't contact LDAP server:
DatabaseError: Can't contact LDAP server: Info goes here
"""
errno = 4203
format = _('%(desc)s:%(info)s')
format = _('%(desc)s: %(info)s')
class LimitsExceeded(ExecutionError):
@ -1205,6 +1205,37 @@ class ObjectclassViolation(ExecutionError):
errno = 4205
format = _('%(info)s')
class NotAllowedOnRDN(ExecutionError):
"""
**4206** Raised when an RDN value is modified.
For example:
>>> raise NotAllowedOnRDN()
Traceback (most recent call last):
...
NotAllowedOnRDN: modifying primary key is not allowed
"""
errno = 4206
format = _('modifying primary key is not allowed')
class OnlyOneValueAllowed(ExecutionError):
"""
**4207** Raised when trying to set more than one value to single-value attributes
For example:
>> raise OnlyOneValueAllowed(attr='ipasearchtimelimit')
Traceback (most recent call last):
...
OnlyOneValueAllowed: ipasearchtimelimit: Only one value allowed.
"""
errno = 4207
format = _('%(attr)s: Only one value allowed.')
class CertificateError(ExecutionError):
"""

View File

@ -504,7 +504,7 @@ class Command(HasParam):
a dictionary. The incoming attribute may be a string or
a list.
Any attribute found that is also a param is silently dropped.
Any attribute found that is also a param is validated.
append controls whether this returns a list of values or a single
value.

View File

@ -157,6 +157,14 @@ _attr_options = (
),
)
# addattr can cause parameters to have more than one value even if not defined
# as multivalue, make sure this isn't the case
def _check_single_value_attrs(params, entry_attrs):
for (a, v) in entry_attrs.iteritems():
if isinstance(v, (list, tuple)) and len(v) > 1:
if a in params and not params[a].multivalue:
raise errors.OnlyOneValueAllowed(attr=a)
class CallbackInterface(Method):
"""
@ -277,6 +285,8 @@ class LDAPCreate(CallbackInterface, crud.Create):
self, ldap, dn, entry_attrs, attrs_list, *keys, **options
)
_check_single_value_attrs(self.params, entry_attrs)
try:
ldap.add_entry(dn, entry_attrs, normalize=self.obj.normalize_dn)
except errors.ExecutionError, e:
@ -464,7 +474,7 @@ class LDAPUpdate(LDAPQuery, crud.Update):
except errors.ExecutionError, e:
try:
(dn, old_entry) = self._call_exc_callbacks(
keys, options, e, ldap.get_entry, dn, attrs_list,
keys, options, e, ldap.get_entry, dn, [],
normalize=self.obj.normalize_dn
)
except errors.NotFound:
@ -491,6 +501,8 @@ class LDAPUpdate(LDAPQuery, crud.Update):
self, ldap, dn, entry_attrs, attrs_list, *keys, **options
)
_check_single_value_attrs(self.params, entry_attrs)
try:
ldap.update_entry(dn, entry_attrs, normalize=self.obj.normalize_dn)
except errors.ExecutionError, e:

View File

@ -70,21 +70,21 @@ def _handle_errors(e, **kw):
try:
# re-raise the error so we can handle it
raise e
except _ldap.NO_SUCH_OBJECT, e:
except _ldap.NO_SUCH_OBJECT:
# args = kw.get('args', '')
# raise errors.NotFound(msg=notfound(args))
raise errors.NotFound(reason='no such entry')
except _ldap.ALREADY_EXISTS, e:
except _ldap.ALREADY_EXISTS:
raise errors.DuplicateEntry()
except _ldap.CONSTRAINT_VIOLATION, e:
except _ldap.CONSTRAINT_VIOLATION:
# This error gets thrown by the uniqueness plugin
if info == 'Another entry with the same attribute value already exists':
raise errors.DuplicateEntry()
else:
raise errors.DatabaseError(desc=desc, info=info)
except _ldap.INSUFFICIENT_ACCESS, e:
except _ldap.INSUFFICIENT_ACCESS:
raise errors.ACIError(info=info)
except _ldap.INVALID_CREDENTIALS, e:
except _ldap.INVALID_CREDENTIALS:
raise errors.ACIError(info="%s %s" % (info, desc))
except _ldap.NO_SUCH_ATTRIBUTE:
# this is raised when a 'delete' attribute isn't found.
@ -93,12 +93,14 @@ def _handle_errors(e, **kw):
raise errors.MidairCollision()
except _ldap.OBJECT_CLASS_VIOLATION:
raise errors.ObjectclassViolation(info=info)
except _ldap.ADMINLIMIT_EXCEEDED, e:
except _ldap.ADMINLIMIT_EXCEEDED:
raise errors.LimitsExceeded()
except _ldap.SIZELIMIT_EXCEEDED, e:
except _ldap.SIZELIMIT_EXCEEDED:
raise errors.LimitsExceeded()
except _ldap.TIMELIMIT_EXCEEDED, e:
except _ldap.TIMELIMIT_EXCEEDED:
raise errors.LimitsExceeded()
except _ldap.NOT_ALLOWED_ON_RDN:
raise errors.NotAllowedOnRDN(attr=info)
except _ldap.SUCCESS:
pass
except _ldap.LDAPError, e:
@ -255,6 +257,20 @@ class ldap2(CrudBackend, Encoder):
else:
return None
def get_single_value(self, attr):
"""
Check the schema to see if the attribute is single-valued.
If the attribute is in the schema then returns True/False
If there is a problem loading the schema or the attribute is
not in the schema return None
"""
if self.schema:
obj = self.schema.get_obj(_ldap.schema.AttributeType, attr)
return obj and obj.single_value
return None
@encode_args(2, 3, 'bind_dn', 'bind_pw')
def create_connection(self, ccache=None, bind_dn='', bind_pw='',
tls_cacertfile=None, tls_certfile=None, tls_keyfile=None,
@ -690,13 +706,15 @@ class ldap2(CrudBackend, Encoder):
adds = list(v.difference(old_v))
rems = list(old_v.difference(v))
is_single_value = self.get_single_value(k)
value_count = len(old_v) + len(adds) - len(rems)
if is_single_value and value_count > 1:
raise errors.OnlyOneValueAllowed(attr=k)
force_replace = False
if k in self._FORCE_REPLACE_ON_UPDATE_ATTRS:
if k in self._FORCE_REPLACE_ON_UPDATE_ATTRS or is_single_value:
force_replace = True
elif self.schema:
obj = self.schema.get_obj(_ldap.schema.AttributeType, k)
if obj and obj.single_value:
force_replace = True
elif len(adds) == 1 and len(rems) == 1:
force_replace = True