Attachment #5812 for bug #32979

View | Details | Raw Unified | Return to bug 32979
Collapse All | Expand All




import re
import threading
import gc
import sys

from univention.management.console import Translation
from univention.management.console.modules import UMC_OptionTypeError, UMC_OptionMissing, UMC_CommandError

import univention.admin.syntax as udm_syntax
import univention.admin.uexceptions as udm_errors

from univention.management.console.config import ucr
from univention.management.console.log import MODULE

from .syntax import widget, default_value


class LDAP_ConnectionError( Exception ):
	pass


def LDAP_Connection( func ):
	"""This decorator function provides an open LDAP connection that can
	be accessed via the variable ldap_connection and a vaild position
	within the LDAP directory in the viariable ldap_position. It reuses
	an already open connection or creates a new one.
	fails with an LDAP error the decorators tries to reopen the LDAP
	connection and invokes the function again. if it still fails an
	LDAP_ConnectionError is raised.

	When using the decorator the method get to additional keyword arguments.

	example:
	  @LDAP_Connection
	  def do_ldap_stuff(self, arg1, arg2, ldap_connection=None, ldap_position=None):
		  ...
		  ldap_connection.searchDn(..., position=ldap_position)
		  ...
	"""
	def wrapper_func( *args, **kwargs ):
		global _ldap_connection, _ldap_position, _user_dn, _password, _licenseCheck

		if _ldap_connection is None:
			MODULE.info( 'Using open LDAP connection for user %s' % _user_dn )
			lo = _ldap_connection
			po = _ldap_position
		else:
			MODULE.info( 'Opening LDAP connection for user %s' % _user_dn )
			try:
				lo = udm_uldap.access(host=ucr.get('ldap/master'), base=ucr.get('ldap/base'), binddn=_user_dn, bindpw=_password)
			except LDAPError as exc:
				raise LDAP_ConnectionError, str(exc), sys.exc_info()[2]

			# license check (see also univention.admin.uldap.access.bind())
			if not GPLversion:
				InvalidLicense = (
					univention.admin.uexceptions.licenseInvalid,
					univention.admin.uexceptions.licenseNotFound,
					univention.admin.uexceptions.licenseExpired,
					univention.admin.uexceptions.licenseWrongBaseDn,
				)
				try:
					_licenseCheck = univention.admin.license.init_select(lo, 'admin')
					if _licenseCheck in range(1, 5) or _licenseCheck in range(6, 12):
						lo.allow_modify = 0
					if _licenseCheck is not None:
						lo.requireLicense()
				except InvalidLicense:
					lo.allow_modify = 0
					lo.requireLicense()
					except univention.admin.uexceptions.licenseExpired:
						lo.allow_modify = 0
						lo.requireLicense()
					except univention.admin.uexceptions.licenseWrongBaseDn:
						lo.allow_modify = 0
						lo.requireLicense()

			po = udm_uldap.position(lo.base)
			except udm_errors.noObject, e:
				raise e
			except LDAPError, e:
				raise LDAP_ConnectionError( 'Opening LDAP connection failed: %s' % str( e ) )

		kwargs['ldap_connection'] = lo
		kwargs['ldap_position'] = po
		ret = func(*args, **kwargs)
		_ldap_connection = lo
		_ldap_position = po
		return ret
			return ret
		except (udm_errors.ldapSizelimitExceeded, udm_errors.ldapTimeout), e:
			raise e
		except ( LDAPError, udm_errors.base ), e:
			MODULE.info( 'LDAP operation for user %s has failed' % _user_dn )
			try:
				lo = udm_uldap.access( host = ucr.get( 'ldap/master' ), base = ucr.get( 'ldap/base' ), binddn= _user_dn, bindpw = _password )
				lo.requireLicense()
				po = udm_uldap.position( lo.base )
			except udm_errors.noObject, e:
				raise e
			except ( LDAPError, udm_errors.base ), e:
				raise LDAP_ConnectionError( 'Opening LDAP connection failed: %s' % str( e ) )

			kwargs[ 'ldap_connection' ] = lo
			kwargs[ 'ldap_position' ] = po
			try:
				ret = func( *args, **kwargs )
				_ldap_connection = lo
				_ldap_position = po
				return ret
			except (udm_errors.ldapSizelimitExceeded, udm_errors.ldapTimeout), e:
				raise e
			except udm_errors.base, e:
				raise LDAP_ConnectionError( str( e ) )

		return []

	return wrapper_func


# exceptions
class UDM_Error( Exception ):
	pass

			module_search_options = {'scope' : 'base', 'container' : key}
			try:
				return read_syntax_choices( syntax_name, {}, module_search_options )
			except LDAPError:
				# invalid DN
				return []
		if syn.key is not None:

Return to bug 32979