Source code for univention.s4connector.s4.user

#!/usr/bin/python3
#
# Univention S4 Connector
#  sid sync
#
# SPDX-FileCopyrightText: 2004-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only

import base64
from logging import getLogger

from univention.logging import Structured


log = Structured(getLogger("LDAP").getChild(__name__))


map_prefdev = [b'any', b'mhs', b'physical', b'telex', b'teletex', b'g3fax', b'g4fax', b'ia5', b'videotex', b'telephone']



[docs]
def prefdev_sync_s4_to_ucs(s4connector, key, s4_object):
    attr = 'preferredDeliveryMethod'
    s4_prefdev = s4_object['attributes'].get(attr)

    if s4_prefdev is None:
        return

    ucs_pref = ''
    for pref in s4_prefdev:
        try:
            ucs_pref += map_prefdev[int(pref)].decode('ASCII') + ' $ '
        except (IndexError, ValueError):
            log.warning("Ignoring invalid value %r of attribute preferredDeliveryMethod. Value must be a number between 0 and 10.", pref)
    ucs_pref = ucs_pref[:-3]
    return [ucs_pref.encode('ASCII')]




[docs]
def prefdev_sync_ucs_to_s4(s4connector, key, ucs_object):
    attr = 'preferredDeliveryMethod'

    ucs_prefdev = ucs_object['attributes'].get(attr)
    if ucs_prefdev is None:
        return

    s4_prefdev = []
    try:
        for pref in ucs_prefdev:
            int(pref.decode('ASCII'))
    except ValueError:
        # map values
        for pref in ucs_prefdev[0].replace(b' ', b'').split(b'$'):
            if pref in map_prefdev:
                s4_prefdev.append(str(map_prefdev.index(pref)).encode('ASCII'))
    else:
        # just return the already mapped value
        return ucs_prefdev
    return s4_prefdev




[docs]
def userCertificate_sync_s4_to_ucs(s4connector, key, s4_object):
    attr = 'userCertificate'
    s4_cert = s4_object['attributes'].get(attr)
    if s4_cert is None:
        return
    ucs_cert = base64.b64encode(s4_cert[0])
    return [ucs_cert]




[docs]
def userCertificate_sync_ucs_to_s4(s4connector, key, ucs_object):
    attr = 'userCertificate;binary'
    try:
        ucs_cert = ucs_object['attributes'][attr][0]
    except (KeyError, IndexError):
        return
    return [ucs_cert]




[docs]
def jpegPhoto_sync_s4_to_ucs(s4connector, key, s4_object):
    attr = 'jpegPhoto'

    s4_photo = s4_object['attributes'].get(attr)
    if s4_photo is None:
        return

    ucs_photo = []
    for photo in s4_photo:
        ucs_photo.append(base64.b64encode(photo))
    return ucs_photo




[docs]
def jpegPhoto_sync_ucs_to_s4(s4connector, key, ucs_object):
    attr = 'jpegPhoto'

    ucs_photo = ucs_object['attributes'].get(attr)
    if ucs_photo is None:
        return
    return ucs_photo[:]




[docs]
def secretary_sync_s4_to_ucs(s4connector, key, s4_object):
    attr = 'secretary'

    s4_secretary = s4_object['attributes'].get(attr)
    if s4_secretary is None:
        return
    else:
        ucs_secretary = []
        for s4_dn in s4_secretary:
            s4_dn = s4_dn.decode('UTF-8')
            dn_mapped = s4_dn
            if s4connector._get_dn_by_con(dn_mapped):
                dn_mapped = s4connector._get_dn_by_con(dn_mapped)
                dn_mapped = s4connector.dn_mapped_to_base(dn_mapped, s4connector.lo.base)
            if hasattr(s4connector.property['user'], 'position_mapping'):
                for mapping in s4connector.property['user'].position_mapping:
                    dn_mapped = s4connector._subtree_replace(dn_mapped, mapping[1], mapping[0])
                if dn_mapped == s4_dn and not (s4connector.lo.base.lower() == dn_mapped[-len(s4connector.lo.base):].lower() and len(s4connector.lo.base) > len(s4connector.lo_s4.base)):
                    dn_mapped = s4connector._subtree_replace(dn_mapped, s4connector.lo_s4.base, s4connector.lo.base)
            ucs_secretary.append(dn_mapped)
        # remove multiples
        ucs_secretary = list(dict.fromkeys(ucs_secretary))
        return ucs_secretary




[docs]
def secretary_sync_ucs_to_s4(s4connector, key, ucs_object):
    attr = 'secretary'

    ucs_secretary = ucs_object['attributes'][attr]
    if ucs_secretary is None:
        return
    else:
        s4_secretary = []
        for ucs_dn in ucs_secretary:
            ucs_dn = ucs_dn.decode('UTF-8')
            dn_mapped = ucs_dn
            if s4connector._get_dn_by_ucs(ucs_dn):
                dn_mapped = s4connector._get_dn_by_ucs(ucs_dn)
                dn_mapped = s4connector.dn_mapped_to_base(dn_mapped, s4connector.lo_s4.base)
            if hasattr(s4connector.property['user'], 'position_mapping'):
                for mapping in s4connector.property['user'].position_mapping:
                    dn_mapped = s4connector._subtree_replace(dn_mapped, mapping[0], mapping[1])
                if dn_mapped == ucs_dn and not (s4connector.lo_s4.base.lower() == dn_mapped[-len(s4connector.lo_s4.base):].lower() and len(s4connector.lo_s4.base) > len(s4connector.lo.base)):
                    dn_mapped = s4connector._subtree_replace(dn_mapped, s4connector.lo.base, s4connector.lo_s4.base)
            s4_secretary.append(dn_mapped.encode('UTF-8'))
        return list(set(s4_secretary))
Univention Corporate Server Python API 5.2 documentation

Source code for univention.s4connector.s4.user
