Hi,
In relation to ticket 47757, I have started work on a deref control for Noriko.
The idea is to get it working in lib389, then get it upstreamed into pyldap.
At this point it's all done, except that the actual request control doesn't
appear to work. Could one of the lib389 / ldap python experts cast their eye
over this and let me know where I've gone wrong?
For one, I don't think that the DerferenceControl decodeControlValue function is
ever called, as I'm never seeing the encodedControlValue printed in my logs.
Second, the results I get from result3 are:
[('cn=testgroup,dc=example,dc=com', {'objectClass': ['top', 'extensibleobject'],
'uniqueMember': ['uid=test,dc=example,dc=com'], 'cn': ['testgroup']})]
[]
Which again, doesn't seem correct, as there should be a result from the control.
Additionally, any tips on how to make the code nicer would be appreciated.
I've attached the complete patch, with a unit test to trigger the search, but
the request control looks like:
"""
controlValue ::= SEQUENCE OF derefSpec DerefSpec
DerefSpec ::= SEQUENCE {
derefAttr attributeDescription, ; with DN syntax
attributes AttributeList }
AttributeList ::= SEQUENCE OF attr AttributeDescription
"""
class AttributeList(univ.SequenceOf):
componentType = AttributeDescription()
class DerefSpec(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('derefAttr', AttributeDescription()),
namedtype.NamedType('attributes', AttributeList()),
)
class ControlValue(univ.SequenceOf):
componentType = DerefSpec()
class DereferenceControl(LDAPControl):
"""
Dereference Control
"""
def __init__(self, criticality, deref):
LDAPControl.__init__(self, CONTROL_DEREF, criticality)
self.deref = deref
def encodeControlValue(self):
# How does -E ask for many values?
derefAttr, attributes = self.deref.split(':')
attributes = attributes.split(',')
al = AttributeList()
i = 0
while len(attributes) > 0:
al.setComponentByPosition(i, attributes.pop())
i += 1
ds = DerefSpec()
ds.setComponentByName('derefAttr', derefAttr)
ds.setComponentByName('attributes', al)
cv = ControlValue()
cv.setComponentByPosition(0, ds)
print(cv.prettyPrint())
return encoder.encode(cv)
def decodeControlValue(self,encodedControlValue):
print(encodedControlValue)
From eb3dede21afcfb930d9c45ee0d83d309683f02ce Mon Sep 17 00:00:00 2001
From: William Brown <will...@blackhats.net.au>
Date: Tue, 25 Aug 2015 15:38:11 +0930
Subject: [PATCH] Add dereference request control to lib389 for testing.
---
lib389/__init__.py | 99 +++++++++++++++++++++++++++++++++++++++++++++++
lib389/_constants.py | 6 +++
tests/dereference_test.py | 75 +++++++++++++++++++++++++++++++++++
3 files changed, 180 insertions(+)
create mode 100644 tests/dereference_test.py
diff --git a/lib389/__init__.py b/lib389/__init__.py
index 8dfe28f..1f3e6a4 100644
--- a/lib389/__init__.py
+++ b/lib389/__init__.py
@@ -73,6 +73,65 @@ MAJOR, MINOR, _, _, _ = sys.version_info
if MAJOR >= 3 or (MAJOR == 2 and MINOR >= 7):
from ldap.controls.simple import GetEffectiveRightsControl
+ from ldap.controls import LDAPControl
+
+ from pyasn1.type import namedtype,univ
+ from pyasn1.codec.ber import encoder,decoder
+ from pyasn1_modules.rfc2251 import AttributeDescription
+
+ # Could use AttributeDescriptionList
+
+ """
+ controlValue ::= SEQUENCE OF derefSpec DerefSpec
+
+ DerefSpec ::= SEQUENCE {
+ derefAttr attributeDescription, ; with DN syntax
+ attributes AttributeList }
+
+ AttributeList ::= SEQUENCE OF attr AttributeDescription
+ """
+ class AttributeList(univ.SequenceOf):
+ componentType = AttributeDescription()
+
+ class DerefSpec(univ.Sequence):
+ componentType = namedtype.NamedTypes(
+ namedtype.NamedType('derefAttr', AttributeDescription()),
+ namedtype.NamedType('attributes', AttributeList()),
+ )
+
+ class ControlValue(univ.SequenceOf):
+ componentType = DerefSpec()
+
+ class DereferenceControl(LDAPControl):
+ """
+ Dereference Control
+ """
+
+ def __init__(self, criticality, deref):
+ LDAPControl.__init__(self, CONTROL_DEREF, criticality)
+ self.deref = deref
+
+ def encodeControlValue(self):
+ # How does -E ask for many values?
+ derefAttr, attributes = self.deref.split(':')
+ attributes = attributes.split(',')
+ al = AttributeList()
+ i = 0
+ while len(attributes) > 0:
+ al.setComponentByPosition(i, attributes.pop())
+ i += 1
+ ds = DerefSpec()
+ ds.setComponentByName('derefAttr', derefAttr)
+ ds.setComponentByName('attributes', al)
+ cv = ControlValue()
+ cv.setComponentByPosition(0, ds)
+ print(cv.prettyPrint())
+ return encoder.encode(cv)
+
+ def decodeControlValue(self,encodedControlValue):
+ print('DEREF decode')
+ print(encodedControlValue)
+
RE_DBMONATTR = re.compile(r'^([a-zA-Z]+)-([1-9][0-9]*)$')
RE_DBMONATTRSUN = re.compile(r'^([a-zA-Z]+)-([a-zA-Z]+)$')
@@ -2401,6 +2460,46 @@ class DirSrv(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, [])
return ldap_result
+ # Is there a better name for this function?
+ def dereference(self, deref, base=DEFAULT_SUFFIX, scope=ldap.SCOPE_SUBTREE, *args, **kwargs):
+ """
+ Perform a search which dereferences values from attributes such as member
+ or unique member.
+ For arguments to this function, please see LDAPObject.search_s. For example:
+
+ @param deref - Dereference query
+ @param base - Base DN of the suffix to check
+ @param scope - search scope
+ @param args -
+ @param kwargs -
+ @return - ldap result
+
+ LDAPObject.search_s(base, scope[, filterstr='(objectClass=*)'[, attrlist=None[, attrsonly=0]]]) -> list|None
+
+ A deref query is of the format:
+
+ "<attribute to derference>:<deref attr1>,<deref attr2>..."
+
+ "uniqueMember:dn,objectClass"
+
+ This will return the dn's and objectClasses of the dereferenced members of the group.
+ """
+ if not (MAJOR >= 3 or (MAJOR == 2 and MINOR >= 7)):
+ raise Exception("UNSUPPORTED EXTENDED OPERATION ON THIS VERSION OF PYTHON")
+ ldap_result = None
+ # This may not be thread safe. Is there a better way to do this?
+ try:
+ drc = DereferenceControl(True, deref=deref.encode('UTF-8'))
+ sctrl = [drc]
+ self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
+
+ #ldap_result = self.search_s(base, scope, *args, **kwargs)
+ res = self.search(base, scope, *args, **kwargs)
+ resp_type, resp_data, resp_msgid, decoded_resp_ctrls = self.result3(res, resp_ctrl_classes={CONTROL_DEREF: DereferenceControl})
+ finally:
+ self.set_option(ldap.OPT_SERVER_CONTROLS, [])
+ return resp_data, decoded_resp_ctrls
+
def buildLDIF(self, num, ldif_file, suffix='dc=example,dc=com'):
"""Generate a simple ldif file using the dbgen.pl script, and set the
ownership and permissions to match the user that the server runs as.
diff --git a/lib389/_constants.py b/lib389/_constants.py
index efabfc6..69c305b 100644
--- a/lib389/_constants.py
+++ b/lib389/_constants.py
@@ -114,6 +114,12 @@ RDN_REPLICA = "cn=replica"
RETROCL_SUFFIX = "cn=changelog"
+##################################
+###
+### Request Control OIDS
+###
+##################################
+CONTROL_DEREF = '1.3.6.1.4.1.4203.666.5.16'
##################################
###
diff --git a/tests/dereference_test.py b/tests/dereference_test.py
new file mode 100644
index 0000000..4def10a
--- /dev/null
+++ b/tests/dereference_test.py
@@ -0,0 +1,75 @@
+'''
+Created on Aug 1, 2015
+
+@author: William Brown
+'''
+from lib389._constants import *
+from lib389 import DirSrv,Entry
+
+INSTANCE_PORT = 54321
+INSTANCE_SERVERID = 'dereferenceds'
+INSTANCE_PREFIX = '/srv'
+
+class Test_dereference():
+ def setUp(self):
+ instance = DirSrv(verbose=False)
+ instance.log.debug("Instance allocated")
+ args = {SER_HOST: LOCALHOST,
+ SER_PORT: INSTANCE_PORT,
+ SER_DEPLOYED_DIR: INSTANCE_PREFIX,
+ SER_SERVERID_PROP: INSTANCE_SERVERID
+ }
+ instance.allocate(args)
+ if instance.exists():
+ instance.delete()
+ instance.create()
+ instance.open()
+ instance.config.loglevel( (LOG_DEFAULT, LOG_PLUGIN, LOG_TRACE), service='error' )
+ # I believe deref is a default on
+ #instance.plugins.enable(name=PLUGIN_DEREF)
+ #instance.restart(300)
+ self.instance = instance
+
+ def tearDown(self):
+ #if self.instance.exists():
+ # self.instance.db2ldif(bename='userRoot', suffixes=[DEFAULT_SUFFIX], excludeSuffixes=[], encrypt=False, \
+ # repl_data=False, outputfile='%s/ldif/%s.ldif' % (self.instance.dbdir,INSTANCE_SERVERID ))
+ # self.instance.clearBackupFS()
+ # self.instance.backupFS()
+ # self.instance.delete()
+ pass
+
+ def add_user(self):
+ # Create a user entry
+ uentry = Entry('uid=test,%s' % DEFAULT_SUFFIX)
+ uentry.setValues('objectclass', 'top', 'extensibleobject')
+ uentry.setValues('uid', 'test')
+ self.instance.add_s(uentry)
+ #self.instance.log.debug("Created user entry as:" ,uentry.dn)
+
+ def add_group(self):
+ # Create a group for the user to have some rights to
+ gentry = Entry('cn=testgroup,%s' % DEFAULT_SUFFIX)
+ gentry.setValues('objectclass', 'top', 'extensibleobject')
+ gentry.setValues('cn', 'testgroup')
+ gentry.setValues('uniqueMember', 'uid=test,%s' % DEFAULT_SUFFIX)
+ self.instance.add_s(gentry)
+
+ def test_dereference(self):
+ # Run an effective rights search
+ result, control_response = self.instance.dereference('uniqueMember:dn,uid', filterstr='(cn=testgroup)')
+
+ #rights = result[0]
+ print(result)
+ print(control_response)
+ #assert rights.getValue('attributeLevelRights') == 'cn:rsc'
+ #assert rights.getValue('entryLevelRights') == 'v'
+
+if __name__ == "__main__":
+ test = Test_dereference()
+ test.setUp()
+ test.add_user()
+ test.add_group()
+ test.test_dereference()
+ test.tearDown()
+
--
2.4.3
--
389-devel mailing list
389-devel@lists.fedoraproject.org
https://admin.fedoraproject.org/mailman/listinfo/389-devel
