IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Add custom mapping object for LDAP entry data.

Browse Source
This commit is contained in:
Jan Cholasta 2013-01-16 14:14:58 +01:00 committed by Martin Kosek
parent e2120c31c3
commit ef1cd6e247
10 changed files with 155 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ipalib/plugins/automember.py
View File

@ -305,7 +305,7 @@ class automember_add_condition(LDAPUpdate):
try:
(dn, old_entry) = ldap.get_entry(
dn, [attr], normalize=self.obj.normalize_dn)
for regex in old_entry:
for regex in old_entry.keys():
if not isinstance(entry_attrs[regex], (list, tuple)):
entry_attrs[regex] = [entry_attrs[regex]]
duplicate = set(old_entry[regex]) & set(entry_attrs[regex])
@ -319,7 +319,7 @@ class automember_add_condition(LDAPUpdate):
# Set failed and completed to they can be harvested in the execute super
setattr(context, 'failed', failed)
setattr(context, 'completed', completed)
setattr(context, 'entry_attrs', entry_attrs)
setattr(context, 'entry_attrs', dict(entry_attrs))
# Make sure to returned the failed results if there is nothing to remove
if completed == 0:
@ -409,7 +409,7 @@ class automember_remove_condition(LDAPUpdate):
# Set failed and completed to they can be harvested in the execute super
setattr(context, 'failed', failed)
setattr(context, 'completed', completed)
setattr(context, 'entry_attrs', entry_attrs)
setattr(context, 'entry_attrs', dict(entry_attrs))
# Make sure to returned the failed results if there is nothing to remove
if completed == 0:

26
ipalib/plugins/baseldap.py
View File

@ -228,7 +228,7 @@ def entry_from_entry(entry, newentry):
del entry[e]
# Re-populate it with new wentry
for e in newentry:
for e in newentry.keys():
entry[e] = newentry[e]
def wait_for_value(ldap, dn, attr, value):
@ -868,7 +868,7 @@ last, after all sets and adds."""),
# Provide a nice error message when user tries to delete an
# attribute that does not exist on the entry (and user is not
# adding it)
names = set(n.lower() for n in old_entry)
names = set(n.lower() for n in old_entry.keys())
del_nonexisting = delattrs - (names | setattrs | addattrs)
if del_nonexisting:
raise errors.ValidationError(name=del_nonexisting.pop(),
@ -1070,8 +1070,8 @@ class LDAPCreate(BaseLDAPCommand, crud.Create):
self.obj.convert_attribute_members(entry_attrs, *keys, **options)
if self.obj.primary_key and keys[-1] is not None:
return dict(result=entry_attrs, value=keys[-1])
return dict(result=entry_attrs, value=u'')
return dict(result=dict(entry_attrs), value=keys[-1])
return dict(result=dict(entry_attrs), value=u'')
def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
assert isinstance(dn, DN)
@ -1195,8 +1195,8 @@ class LDAPRetrieve(LDAPQuery):
assert isinstance(dn, DN)
entry_attrs['dn'] = dn
if self.obj.primary_key and keys[-1] is not None:
return dict(result=entry_attrs, value=keys[-1])
return dict(result=entry_attrs, value=u'')
return dict(result=dict(entry_attrs), value=keys[-1])
return dict(result=dict(entry_attrs), value=u'')
def pre_callback(self, ldap, dn, attrs_list, *keys, **options):
assert isinstance(dn, DN)
@ -1324,8 +1324,8 @@ class LDAPUpdate(LDAPQuery, crud.Update):
self.obj.convert_attribute_members(entry_attrs, *keys, **options)
if self.obj.primary_key and keys[-1] is not None:
return dict(result=entry_attrs, value=keys[-1])
return dict(result=entry_attrs, value=u'')
return dict(result=dict(entry_attrs), value=keys[-1])
return dict(result=dict(entry_attrs), value=u'')
def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
assert isinstance(dn, DN)
@ -1552,7 +1552,7 @@ class LDAPAddMember(LDAPModMember):
return dict(
completed=completed,
failed=failed,
result=entry_attrs,
result=dict(entry_attrs),
)
def pre_callback(self, ldap, dn, found, not_found, *keys, **options):
@ -1651,7 +1651,7 @@ class LDAPRemoveMember(LDAPModMember):
return dict(
completed=completed,
failed=failed,
result=entry_attrs,
result=dict(entry_attrs),
)
def pre_callback(self, ldap, dn, found, not_found, *keys, **options):
@ -1861,7 +1861,7 @@ class LDAPSearch(BaseLDAPCommand, crud.Search):
for e in entries:
assert isinstance(e[0], DN)
e[1]['dn'] = e[0]
entries = [e for (dn, e) in entries]
entries = [dict(e) for (dn, e) in entries]
return dict(
result=entries,
@ -2000,7 +2000,7 @@ class LDAPAddReverseMember(LDAPModReverseMember):
return dict(
completed=completed,
failed=failed,
result=entry_attrs,
result=dict(entry_attrs),
)
def pre_callback(self, ldap, dn, *keys, **options):
@ -2100,7 +2100,7 @@ class LDAPRemoveReverseMember(LDAPModReverseMember):
return dict(
completed=completed,
failed=failed,
result=entry_attrs,
result=dict(entry_attrs),
)
def pre_callback(self, ldap, dn, *keys, **options):

14
ipalib/plugins/dns.py
View File

@ -2124,7 +2124,7 @@ class dnsrecord(LDAPObject):
assert isinstance(dn, DN)
ldap = self.api.Backend.ldap2
for rtype in entry_attrs:
for rtype in entry_attrs.keys():
rtype_cb = getattr(self, '_%s_pre_callback' % rtype, None)
if rtype_cb:
rtype_cb(ldap, dn, entry_attrs, *keys, **options)
@ -2267,9 +2267,9 @@ class dnsrecord(LDAPObject):
def check_record_type_collisions(self, old_entry, entry_attrs):
# Test that only allowed combination of record types was created
attrs = set(attr for attr in entry_attrs if attr in _record_attributes
attrs = set(attr for attr in entry_attrs.keys() if attr in _record_attributes
and entry_attrs[attr])
attrs.update(attr for attr in old_entry if attr not in entry_attrs)
attrs.update(attr for attr in old_entry.keys() if attr not in entry_attrs)
try:
attrs.remove('cnamerecord')
except KeyError:
@ -2404,7 +2404,7 @@ class dnsrecord_add(LDAPCreate):
self.obj.run_precallback_validators(dn, entry_attrs, *keys, **options)
# run precallback also for all new RR type attributes in entry_attrs
for attr in entry_attrs:
for attr in entry_attrs.keys():
try:
param = self.params[attr]
except KeyError:
@ -2437,7 +2437,7 @@ class dnsrecord_add(LDAPCreate):
except errors.NotFound:
pass
else:
for attr in entry_attrs:
for attr in entry_attrs.keys():
if attr not in _record_attributes:
continue
if entry_attrs[attr] is None:
@ -2568,7 +2568,7 @@ class dnsrecord_mod(LDAPUpdate):
normalize=self.obj.normalize_dn)
del_all = True
for attr in old_entry:
for attr in old_entry.keys():
if old_entry[attr]:
del_all = False
break
@ -2709,7 +2709,7 @@ class dnsrecord_del(LDAPUpdate):
del_all = False
if not self.obj.is_pkey_zone_record(*keys):
record_found = False
for attr in old_entry:
for attr in old_entry.keys():
if old_entry[attr]:
record_found = True
break

4
ipalib/plugins/krbtpolicy.py
View File

@ -177,7 +177,7 @@ class krbtpolicy_reset(LDAPQuery):
(dn, entry_attrs) = ldap.get_entry(dn, self.obj.default_attributes)
if keys[-1] is not None:
return dict(result=entry_attrs, value=keys[-1])
return dict(result=entry_attrs, value=u'')
return dict(result=dict(entry_attrs), value=keys[-1])
return dict(result=dict(entry_attrs), value=u'')
api.register(krbtpolicy_reset)

4
ipalib/plugins/sudorule.py
View File

@ -642,7 +642,7 @@ class sudorule_add_option(LDAPQuery):
dn, attrs_list, normalize=self.obj.normalize_dn
)
return dict(result=entry_attrs)
return dict(result=dict(entry_attrs))
def output_for_cli(self, textui, result, cn, **options):
textui.print_dashed(_('Added option "%(option)s" to Sudo Rule "%(rule)s"') % \
@ -697,7 +697,7 @@ class sudorule_remove_option(LDAPQuery):
dn, attrs_list, normalize=self.obj.normalize_dn
)
return dict(result=entry_attrs)
return dict(result=dict(entry_attrs))
def output_for_cli(self, textui, result, cn, **options):
textui.print_dashed(_('Removed option "%(option)s" from Sudo Rule "%(rule)s"') % \

2
ipalib/plugins/trust.py
View File

@ -311,7 +311,7 @@ sides.
base_dn = DN(api.env.container_trusts, api.env.basedn),
filter = trust_filter)
result['result'] = trusts[0][1]
result['result'] = dict(trusts[0][1])
result['result']['trusttype'] = [trust_type_string(result['result']['ipanttrusttype'][0])]
result['result']['trustdirection'] = [trust_direction_string(result['result']['ipanttrustdirection'][0])]
result['result']['truststatus'] = [trust_status_string(result['verified'])]

2
ipapython/entity.py
View File

@ -53,6 +53,8 @@ class Entity:
self.dn = DN(entrydata)
self.data = ipautil.CIDict()
elif isinstance(entrydata,dict):
if hasattr(entrydata, 'dn'):
entrydata['dn'] = entrydata.dn
self.dn = entrydata['dn']
del entrydata['dn']
self.data = ipautil.CIDict(entrydata)

9
ipaserver/ipaldap.py
View File

@ -40,7 +40,7 @@ from ipalib import errors
from ipapython.ipautil import format_netloc, wait_for_open_socket, wait_for_open_ports
from ipapython.dn import DN
from ipapython.entity import Entity
from ipaserver.plugins.ldap2 import IPASimpleLDAPObject
from ipaserver.plugins.ldap2 import IPASimpleLDAPObject, LDAPEntry
# Global variable to define SASL auth
SASL_AUTH = ldap.sasl.sasl({},'GSSAPI')
@ -54,7 +54,7 @@ class IPAEntryLDAPObject(IPASimpleLDAPObject):
objtype, data = IPASimpleLDAPObject.result(self, msgid, all, timeout)
# data is either a 2-tuple or a list of 2-tuples
if data:
if isinstance(data, tuple):
if isinstance(data, (LDAPEntry, tuple)):
return objtype, Entry(data)
elif isinstance(data, list):
return objtype, [Entry(x) for x in data]
@ -102,7 +102,10 @@ class Entry:
a search result entry or a reference or None.
If creating a new empty entry, data is the string DN."""
if entrydata:
if isinstance(entrydata,tuple):
if isinstance(entrydata,LDAPEntry):
self.dn = entrydata.dn
self.data = entrydata
elif isinstance(entrydata,tuple):
self.dn = entrydata[0]
self.data = ipautil.CIDict(entrydata[1])
elif isinstance(entrydata,DN):

97
ipaserver/plugins/ldap2.py
View File

@ -47,7 +47,6 @@ import ldap.filter as _ldap_filter
import ldap.sasl as _ldap_sasl
from ipapython.dn import DN, RDN
from ipapython.ipautil import CIDict
from collections import namedtuple
from ipalib.errors import NetworkError, DatabaseError
@ -80,7 +79,94 @@ _debug_log_ldap = False
# r = result[0]
# r[0] == r.dn
# r[1] == r.data
LDAPEntry = namedtuple('LDAPEntry', ['dn', 'data'])
class LDAPEntry(dict):
__slots__ = ('_dn',)
def __init__(self, _dn=None, _obj=None, **kwargs):
if isinstance(_dn, LDAPEntry):
assert _obj is None
_obj = _dn
self._dn = DN(_obj._dn) #pylint: disable=E1103
else:
assert isinstance(_dn, DN)
if _obj is None:
_obj = {}
self._dn = _dn
super(LDAPEntry, self).__init__(self._init_iter(_obj, **kwargs))
# properties for Entry and Entity compatibility
@property
def dn(self):
return self._dn
@dn.setter
def dn(self, value):
assert isinstance(value, DN)
self._dn = value
@property
def data(self):
return self
def _attr_name(self, name):
if not isinstance(name, basestring):
raise TypeError(
"attribute name must be unicode or str, got %s object %r" % (
name.__class__.__name__, name))
if isinstance(name, str):
name = name.decode('ascii')
return name.lower()
def _init_iter(self, _obj, **kwargs):
_obj = dict(_obj, **kwargs)
for (k, v) in _obj.iteritems():
yield (self._attr_name(k), v)
def __repr__(self):
dict_repr = super(LDAPEntry, self).__repr__()
return '%s(%s, %s)' % (type(self).__name__, repr(self._dn), dict_repr)
def copy(self):
return LDAPEntry(self)
def __setitem__(self, name, value):
super(LDAPEntry, self).__setitem__(self._attr_name(name), value)
def setdefault(self, name, default):
return super(LDAPEntry, self).setdefault(self._attr_name(name), default)
def update(self, _obj={}, **kwargs):
super(LDAPEntry, self).update(self._init_iter(_obj, **kwargs))
def __getitem__(self, name):
# for python-ldap tuple compatibility
if name == 0:
return self._dn
elif name == 1:
return self
return super(LDAPEntry, self).__getitem__(self._attr_name(name))
def get(self, name, default=None):
return super(LDAPEntry, self).get(self._attr_name(name), default)
def __delitem__(self, name):
super(LDAPEntry, self).__delitem__(self._attr_name(name))
def pop(self, name, *default):
return super(LDAPEntry, self).pop(self._attr_name(name), *default)
def __contains__(self, name):
return super(LDAPEntry, self).__contains__(self._attr_name(name))
def has_key(self, name):
return super(LDAPEntry, self).has_key(self._attr_name(name))
# for python-ldap tuple compatibility
def __iter__(self):
yield self._dn
yield self
# Group Member types
@ -459,14 +545,13 @@ class IPASimpleLDAPObject(object):
original_dn = dn_tuple[0]
original_attrs = dn_tuple[1]
ipa_dn = DN(original_dn)
ipa_attrs = dict()
ipa_entry = LDAPEntry(DN(original_dn))
for attr, original_values in original_attrs.items():
target_type = self._SYNTAX_MAPPING.get(self.get_syntax(attr), unicode_from_utf8)
ipa_attrs[attr.lower()] = self.convert_value_list(attr, target_type, original_values)
ipa_entry[attr] = self.convert_value_list(attr, target_type, original_values)
ipa_result.append(LDAPEntry(ipa_dn, ipa_attrs))
ipa_result.append(ipa_entry)
if _debug_log_ldap:
self.debug('ldap.result: %s', ipa_result)

29
tests/test_ipaserver/test_ldap.py
View File

@ -27,7 +27,7 @@
import nose
import os
from ipaserver.plugins.ldap2 import ldap2
from ipaserver.plugins.ldap2 import ldap2, LDAPEntry
from ipalib.plugins.service import service, service_show
from ipalib.plugins.host import host
import nss.nss as nss
@ -145,3 +145,30 @@ class test_ldap(object):
cert = cert[0]
serial = unicode(x509.get_serial_number(cert, x509.DER))
assert serial is not None
def test_entry(self):
"""
Test the LDAPEntry class
"""
cn1 = [u'test1']
cn2 = [u'test2']
dn1 = DN(('cn', cn1[0]))
dn2 = DN(('cn', cn2[0]))
e = LDAPEntry(dn1, cn=cn1)
assert e.dn is dn1
assert 'CN' in e
assert e['CN'] is cn1
assert e['CN'] is e[u'cn']
e.dn = dn2
assert e.dn is dn2
e['cn'] = cn2
assert 'CN' in e
assert e['CN'] is cn2
assert e['CN'] is e[u'cn']
del e['CN']
assert 'CN' not in e
assert u'cn' not in e