Produits Entr'ouvert » Authentic 2

import json

        # extra attributes retrieve by making other LDAP search using user object informations

        'extra_attributes': {},

        for extra_at in block.get('extra_attributes', {}):

            if 'loop_over_attribute' in block['extra_attributes'][extra_at]:

                attributes.add(block['extra_attributes'][extra_at]['loop_over_attribute'])

            at_mapping = block['extra_attributes'][extra_at].get('mapping', {})

            for key in at_mapping:

                if at_mapping[key] != 'dn':

                    attributes.add(at_mapping[key])

        # extra attributes

        attribute_map = cls.get_ldap_extra_attributes(block, conn, dn, attribute_map)

        return attribute_map

    @classmethod

    def get_ldap_extra_attributes(cls, block, conn, dn, attribute_map):

        '''Retrieve extra attributes from LDAP'''

        ldap_scopes = {

            'base': ldap.SCOPE_BASE,

            'one': ldap.SCOPE_ONELEVEL,

            'sub': ldap.SCOPE_SUBTREE,

        }

        log.debug(u'Attrs before extra attributes : %s' % attribute_map)

        for extra_attribute_name in block.get('extra_attributes', {}):

            extra_attribute_config = block['extra_attributes'][extra_attribute_name]

            extra_attribute_values = []

            if 'loop_over_attribute' in extra_attribute_config:

                extra_attribute_config['loop_over_attribute'] = extra_attribute_config['loop_over_attribute'].lower()

                if extra_attribute_config['loop_over_attribute'] not in attribute_map:

                    log.debug('loop_over_attribute %s not present (or empty) in user object attributes retreived. Pass.' % extra_attribute_config['loop_over_attribute'])

                    continue

                if 'filter' not in extra_attribute_config and 'basedn' not in extra_attribute_config:

                    log.warning('Extra attribute %s not correctly configured : you need to defined at least one of filter or basedn parameters' % extra_attribute_name)

                for item in attribute_map[extra_attribute_config['loop_over_attribute']]:

                    ldap_filter = force_text(extra_attribute_config.get('filter', 'objectClass=*')).format(item=item, **attribute_map)

                    ldap_basedn = force_text(extra_attribute_config.get('basedn', block.get('basedn'))).format(item=item, **attribute_map)

                    ldap_scope = ldap_scopes.get(extra_attribute_config.get('scope', 'sub'), ldap.SCOPE_SUBTREE)

                    ldap_attributes_mapping = extra_attribute_config.get('mapping', {})

                    ldap_attributes_names = filter(lambda a: a != 'dn', ldap_attributes_mapping.values())

                    try:

                        results = conn.search_s(ldap_basedn, ldap_scope, ldap_filter, list(ldap_attributes_names))

                    except ldap.LDAPError:

                        log.exception('unable to retrieve extra attribute %s for item %s' % (extra_attribute_name, item))

                        continue

                    item_value = {}

                    for obj in results:

                        log.debug(u'Object retrieved for extra attr %s with item %s : %s' % (extra_attribute_name, item, obj))

                        obj_attributes = cls.normalize_ldap_results(obj[1])

                        obj_attributes[dn] = obj[0]

                        log.debug(u'Object attributes normalized for extra attr %s with item %s : %s' % (extra_attribute_name, item, obj_attributes))

                        for key in ldap_attributes_mapping:

                            item_value[key] = obj_attributes.get(ldap_attributes_mapping[key].lower())

                            log.debug(u'Object attribute %s value retrieved for extra attr %s with item %s : %s' % (ldap_attributes_mapping[key], extra_attribute_name, item, item_value[key]))

                            if not item_value[key]:

                                del item_value[key]

                            elif len(item_value[key]) == 1:

                                item_value[key] = item_value[key][0]

                    extra_attribute_values.append(item_value)

            else:

                log.warning('loop_over_attribute not defined for extra attribute %s' % extra_attribute_name)

            extra_attribute_serialization = extra_attribute_config.get('serialization', None)

            if extra_attribute_serialization is None:

                attribute_map[extra_attribute_name] = extra_attribute_values

            elif extra_attribute_serialization == 'json':

                attribute_map[extra_attribute_name] = json.dumps(extra_attribute_values)

            else:

                log.warning('Invalid serialization type "%s" for extra attribute %s' % (extra_attribute_serialization, extra_attribute_name))

            names.update(map_text(block.get('extra_attributes', {})).keys())

import json

CARLICENSE = '123445ABC'

EO_O = "EO"

EO_STREET = "169 rue du Chateau"

EO_POSTALCODE = "75014"

EO_CITY = "PARIS"

EE_O = "EE"

EE_STREET = "44 rue de l'Ouest"

EE_POSTALCODE = "75014"

EE_CITY = "PARIS"

carLicense: {cl}

o: EO

o: EE

dn: o={eo_o},o=orga

objectClass: organization

o: {eo_o}

postalAddress: {eo_street}

postalCode: {eo_postalcode}

l: {eo_city}

dn: o={ee_o},o=orga

objectClass: organization

o: {ee_o}

postalAddress: {ee_street}

postalCode: {ee_postalcode}

l: {ee_city}

'''.format(dn=DN, uid=UID, password=PASS, cl=CARLICENSE,

           eo_o=EO_O, eo_street=EO_STREET, eo_postalcode=EO_POSTALCODE, eo_city=EO_CITY,

           ee_o=EE_O, ee_street=EE_STREET, ee_postalcode=EE_POSTALCODE, ee_city=EE_CITY

          )

    )

def test_get_mapped_attributes(slapd, settings, db, client):

    settings.LDAP_AUTH_SETTINGS = [{

        'url': [slapd.ldap_url],

        'basedn': 'o=ôrga',

        'use_tls': False,

        'groupstaff': ['cn=group1,o=orga'],

        'attributes': ['uid', 'carLicense'],

    }]

    user = authenticate(username='etienne.michu', password=UPASS)

    fetched_attrs = user.get_attributes(object(), {})

    assert UID in fetched_attrs.get('uid')

    assert CARLICENSE in fetched_attrs.get('carlicense')

def test_get_extra_mapped_attributes(slapd, settings, db, client):

    settings.LDAP_AUTH_SETTINGS = [{

        'url': [slapd.ldap_url],

        'basedn': 'o=ôrga',

        'use_tls': False,

        'groupstaff': ['cn=group1,o=orga'],

        'attributes': ['uid'],

        'extra_attributes': {

            'orga': {

                'loop_over_attribute': 'o',

                'filter': '(&(objectclass=organization)(o={item}))',

                'basedn': 'o=orga',

                'scope': 'sub',

                'mapping': {

                    'id': 'o',

                    'street': 'postalAddress',

                    'city': 'l',

                    'postal_code': 'postalCode',

                },

                'serialization': 'json'

            }

        },

    }]

    user = authenticate(username='etienne.michu', password=UPASS)

    fetched_attrs = user.get_attributes(object(), {})

    assert UID in fetched_attrs.get('uid')

    assert 'orga' in fetched_attrs

    orgas = json.loads(fetched_attrs.get('orga'))

    assert isinstance(orgas, list)

    assert len(orgas) == 2

    assert {'id': EO_O, 'street': EO_STREET, 'city': EO_CITY, 'postal_code': EO_POSTALCODE} in orgas

    assert {'id': EE_O, 'street': EE_STREET, 'city': EE_CITY, 'postal_code': EE_POSTALCODE} in orgas