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

Ticket #1879 - IPAdmin undefined anonymous parameter lists

Browse Source 
The IPAdmin class in ipaserver/ipaldap.py has methods with anonymous
undefined parameter lists.

For example:

    def getList(self,*args):

In Python syntax this means you can call getList with any positional
parameter list you want.

This is bad because:

1) It's not true, *args gets passed to an ldap function with a well
defined parameter list, so you really do have to call it with a
defined parameter list. *args will let you pass anything, but once it
gets passed to the ldap function it will blow up if the parameters do
not match (what parameters are those you're wondering? see item 2).

2) The programmer does not know what the valid parameters are unless
they are defined in the formal parameter list.

3) Without a formal parameter list automatic documentation generators
cannot produce API documentation (see item 2)

4) The Python interpreter cannot validate the parameters being passed
because there is no formal parameter list. Note, Python does not
validate the type of parameters, but it does validate the correct
number of postitional parameters are passed and only defined keyword
parameters are passed. Bypassing the language support facilities leads
to programming errors.

5) Without a formal parameter list program checkers such as pylint
cannot validate the program which leads to progamming errors.

6) Without a formal parameter list which includes default keyword
parameters it's not possible to use keyword arguments nor to know what
their default values are (see item 2). One is forced to pass a keyword
argument as a positional argument, plus you must then pass every
keyword argument between the end of the positional argument list and
keyword arg of interest even of the other keyword arguments are not of
interest. This also demands you know what the default value of the
intermediate keyword arguments are (see item 2) and hope they don't
change.

Also the *args anonymous tuple get passed into the error handling code
so it can report what the called values were. But because the tuple is
anonymous the error handler cannot not describe what it was passed. In
addition the error handling code makes assumptions about the possible
contents of the anonymous tuple based on current practice instead of
actual defined values. Things like "if the number of items in the
tuple is 2 or less then the first tuple item must be a dn
(Distinguished Name)" or "if the number of items in the tuple is
greater than 2 then the 3rd item must be an ldap search filter". These
are constructs which are not robust and will fail at some point in the
future.

This patch also fixes the use of IPAdmin.addEntry(). It was sometimes
being called with (dn, modlist), sometimes a Entry object, or
sometimes a Entity object. Now it's always called with either a Entry
or Entity object and IPAdmin.addEntry() validates the type of the
parameter passed.
This commit is contained in:
John Dennis
2011-09-26 10:39:15 -04:00
committed by Martin Kosek
parent 1b0b9645d1
commit e1c1fcf543
2 changed files with 45 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ipaserver/install/ldapupdate.py
View File

@@ -344,7 +344,7 @@ class LDAPUpdate:
root_logger.debug("Task id: %s", dn) root_logger.debug("Task id: %s", dn)
if self.live_run: if self.live_run:
self.conn.addEntry(e.dn, e.toTupleList()) self.conn.addEntry(e)
return dn return dn
@@ -644,7 +644,7 @@ class LDAPUpdate:
# addifexist may result in an entry with only a # addifexist may result in an entry with only a
# dn defined. In that case there is nothing to do. # dn defined. In that case there is nothing to do.
# It means the entry doesn't exist, so skip it. # It means the entry doesn't exist, so skip it.
self.conn.addEntry(entry.dn, entry.toTupleList()) self.conn.addEntry(entry)
self.modified = True self.modified = True
except Exception, e: except Exception, e:
root_logger.error("Add failure %s", e) root_logger.error("Add failure %s", e)

94
ipaserver/ipaldap.py
View File

@@ -1,5 +1,6 @@
# Authors: Rich Megginson <richm@redhat.com> # Authors: Rich Megginson <richm@redhat.com>
# Rob Crittenden <rcritten@redhat.com # Rob Crittenden <rcritten@redhat.com>
# John Dennis <jdennis@redhat.com>
# #
# Copyright (C) 2007 Red Hat # Copyright (C) 2007 Red Hat
# see file 'COPYING' for use and warranty information # see file 'COPYING' for use and warranty information
@@ -35,6 +36,7 @@ from ldap.ldapobject import SimpleLDAPObject
from ipaserver import ipautil from ipaserver import ipautil
from ipalib import errors from ipalib import errors
from ipapython.ipautil import format_netloc from ipapython.ipautil import format_netloc
from ipapython.entity import Entity
# Global variable to define SASL auth # Global variable to define SASL auth
SASL_AUTH = ldap.sasl.sasl({},'GSSAPI') SASL_AUTH = ldap.sasl.sasl({},'GSSAPI')
@@ -263,7 +265,7 @@ class IPAdmin(SimpleLDAPObject):
self.dbdir = os.path.dirname(ent.getValue('nsslapd-directory')) self.dbdir = os.path.dirname(ent.getValue('nsslapd-directory'))
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
def __str__(self): def __str__(self):
return self.host + ":" + str(self.port) return self.host + ":" + str(self.port)
@@ -304,8 +306,8 @@ class IPAdmin(SimpleLDAPObject):
# re-raise the error so we can handle it # re-raise the error so we can handle it
raise e raise e
except ldap.NO_SUCH_OBJECT, e: except ldap.NO_SUCH_OBJECT, e:
args = kw.get('args', ["entry not found"]) arg_desc = kw.get('arg_desc', "entry not found")
raise errors.NotFound(reason=notfound(args)) raise errors.NotFound(reason=arg_desc)
except ldap.ALREADY_EXISTS, e: except ldap.ALREADY_EXISTS, e:
raise errors.DuplicateEntry() raise errors.DuplicateEntry()
except ldap.CONSTRAINT_VIOLATION, e: except ldap.CONSTRAINT_VIOLATION, e:
@@ -344,7 +346,7 @@ class IPAdmin(SimpleLDAPObject):
self.principal = principal self.principal = principal
self.proxydn = None self.proxydn = None
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
def do_simple_bind(self, binddn="cn=directory manager", bindpw=""): def do_simple_bind(self, binddn="cn=directory manager", bindpw=""):
self.binddn = binddn self.binddn = binddn
@@ -361,7 +363,7 @@ class IPAdmin(SimpleLDAPObject):
self.sasl_interactive_bind_s("", auth_tokens) self.sasl_interactive_bind_s("", auth_tokens)
self.__lateinit() self.__lateinit()
def getEntry(self,*args): def getEntry(self, base, scope, filterstr='(objectClass=*)', attrlist=None, attrsonly=0):
"""This wraps the search function. It is common to just get one entry""" """This wraps the search function. It is common to just get one entry"""
sctrl = self.__get_server_controls() sctrl = self.__get_server_controls()
@@ -370,21 +372,22 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
try: try:
res = self.search(*args) res = self.search(base, scope, filterstr, attrlist, attrsonly)
objtype, obj = self.result(res) objtype, obj = self.result(res)
except ldap.LDAPError, e: except ldap.LDAPError, e:
kw = {'args': args} arg_desc = 'base="%s", scope=%s, filterstr="%s"' % (base, scope, filterstr)
self.__handle_errors(e, **kw) self.__handle_errors(e, arg_desc=arg_desc)
if not obj: if not obj:
raise errors.NotFound(reason=notfound(args)) arg_desc = 'base="%s", scope=%s, filterstr="%s"' % (base, scope, filterstr)
raise errors.NotFound(reason=arg_desc)
elif isinstance(obj,Entry): elif isinstance(obj,Entry):
return obj return obj
else: # assume list/tuple else: # assume list/tuple
return obj[0] return obj[0]
def getList(self,*args): def getList(self, base, scope, filterstr='(objectClass=*)', attrlist=None, attrsonly=0):
"""This wraps the search function to find multiple entries.""" """This wraps the search function to find multiple entries."""
sctrl = self.__get_server_controls() sctrl = self.__get_server_controls()
@@ -392,14 +395,15 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
try: try:
res = self.search(*args) res = self.search(base, scope, filterstr, attrlist, attrsonly)
objtype, obj = self.result(res) objtype, obj = self.result(res)
except ldap.LDAPError, e: except ldap.LDAPError, e:
kw = {'args': args} arg_desc = 'base="%s", scope=%s, filterstr="%s"' % (base, scope, filterstr)
self.__handle_errors(e, **kw) self.__handle_errors(e, arg_desc=arg_desc)
if not obj: if not obj:
raise errors.NotFound(reason=notfound(args)) arg_desc = 'base="%s", scope=%s, filterstr="%s"' % (base, scope, filterstr)
raise errors.NotFound(reason=arg_desc)
entries = [] entries = []
for s in obj: for s in obj:
@@ -407,7 +411,8 @@ class IPAdmin(SimpleLDAPObject):
return entries return entries
def getListAsync(self,*args): def getListAsync(self, base, scope, filterstr='(objectClass=*)', attrlist=None, attrsonly=0,
serverctrls=None, clientctrls=None, timeout=-1, sizelimit=0):
"""This version performs an asynchronous search, to allow """This version performs an asynchronous search, to allow
results even if we hit a limit. results even if we hit a limit.
@@ -423,7 +428,8 @@ class IPAdmin(SimpleLDAPObject):
partial = 0 partial = 0
try: try:
msgid = self.search_ext(*args) msgid = self.search_ext(base, scope, filterstr, attrlist, attrsonly,
serverctrls, clientctrls, timeout, sizelimit)
objtype, result_list = self.result(msgid, 0) objtype, result_list = self.result(msgid, 0)
while result_list: while result_list:
for result in result_list: for result in result_list:
@@ -433,11 +439,13 @@ class IPAdmin(SimpleLDAPObject):
ldap.TIMELIMIT_EXCEEDED), e: ldap.TIMELIMIT_EXCEEDED), e:
partial = 1 partial = 1
except ldap.LDAPError, e: except ldap.LDAPError, e:
kw = {'args': args} arg_desc = 'base="%s", scope=%s, filterstr="%s", timeout=%s, sizelimit=%s' % \
self.__handle_errors(e, **kw) (base, scope, filterstr, timeout, sizelimit)
self.__handle_errors(e, arg_desc=arg_desc)
if not entries: if not entries:
raise errors.NotFound(reason=notfound(args)) arg_desc = 'base="%s", scope=%s, filterstr="%s"' % (base, scope, filterstr)
raise errors.NotFound(reason=arg_desc)
if partial == 1: if partial == 1:
counter = -1 counter = -1
@@ -446,19 +454,22 @@ class IPAdmin(SimpleLDAPObject):
return [counter] + entries return [counter] + entries
def addEntry(self,*args): def addEntry(self, entry):
"""This wraps the add function. It assumes that the entry is already """This wraps the add function. It assumes that the entry is already
populated with all of the desired objectclasses and attributes""" populated with all of the desired objectclasses and attributes"""
if not isinstance(entry, (Entry, Entity)):
raise TypeError('addEntry expected an Entry or Entity object, got %s instead' % entry.__class__)
sctrl = self.__get_server_controls() sctrl = self.__get_server_controls()
try: try:
if sctrl is not None: if sctrl is not None:
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.add_s(*args) self.add_s(entry.dn, entry.toTupleList())
except ldap.LDAPError, e: except ldap.LDAPError, e:
kw = {'args': args} arg_desc = 'entry=%s, modlist=%s' % (entry)
self.__handle_errors(e, **kw) self.__handle_errors(e, arg_desc=arg_desc)
return True return True
def updateRDN(self, dn, newrdn): def updateRDN(self, dn, newrdn):
@@ -475,7 +486,7 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.modrdn_s(dn, newrdn, delold=1) self.modrdn_s(dn, newrdn, delold=1)
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
return True return True
def updateEntry(self,dn,oldentry,newentry): def updateEntry(self,dn,oldentry,newentry):
@@ -494,7 +505,7 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.modify_s(dn, modlist) self.modify_s(dn, modlist)
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
return True return True
def generateModList(self, old_entry, new_entry): def generateModList(self, old_entry, new_entry):
@@ -575,10 +586,10 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.modify_s(dn, modlist) self.modify_s(dn, modlist)
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
return True return True
def deleteEntry(self,*args): def deleteEntry(self, dn):
"""This wraps the delete function. Use with caution.""" """This wraps the delete function. Use with caution."""
sctrl = self.__get_server_controls() sctrl = self.__get_server_controls()
@@ -586,10 +597,10 @@ class IPAdmin(SimpleLDAPObject):
try: try:
if sctrl is not None: if sctrl is not None:
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.delete_s(*args) self.delete_s(dn)
except ldap.LDAPError, e: except ldap.LDAPError, e:
kw = {'args': args} arg_desc = 'dn=%s' % (dn)
self.__handle_errors(e, **kw) self.__handle_errors(e, arg_desc=arg_desc)
return True return True
def modifyPassword(self,dn,oldpass,newpass): def modifyPassword(self,dn,oldpass,newpass):
@@ -607,7 +618,7 @@ class IPAdmin(SimpleLDAPObject):
self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl) self.set_option(ldap.OPT_SERVER_CONTROLS, sctrl)
self.passwd_s(dn, oldpass, newpass) self.passwd_s(dn, oldpass, newpass)
except ldap.LDAPError, e: except ldap.LDAPError, e:
self.__handle_errors(e, **{}) self.__handle_errors(e)
return True return True
def __wrapmethods(self): def __wrapmethods(self):
@@ -737,22 +748,3 @@ class IPAdmin(SimpleLDAPObject):
keys.sort(reverse=reverse) keys.sort(reverse=reverse)
return map(res.get, keys) return map(res.get, keys)
def notfound(args):
"""Return a string suitable for displaying as an error when a
search returns no results.
This just returns whatever is after the equals sign"""
if len(args) > 2:
searchfilter = args[2]
try:
# Python re doesn't do paren counting so the string could
# have a trailing paren "foo)"
target = re.match(r'\(.*=(.*)\)', searchfilter).group(1)
target = target.replace(")","")
except:
target = searchfilter
return "%s not found" % str(target)
else:
return args[0]