def check_ldap_info()

in Templates/kerberosSideCar/krb_side_car.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

def check_ldap_info(env_vars):
    """
    Sanity check of user info from LDAP
    :param env_vars: Environment variables
    :type env_vars: Dictionary
    :return: Prints warnings, no return value
    :rtype: nothing
    """

    ldap_check_status = False
    # Kerberos ticket must be available since the query is over Kerberos
    print("Start LDAP check", flush=True)

    servers = get_dc_server_names(env_vars[DIRECTORY_NAME])
    servers += [env_vars[DIRECTORY_NAME]]

    for server in servers:
        server = "ldap://" + server
        print("DNS server name = " + server, flush=True)
        try:
            conn = Connection(server, sasl_credentials=(
                ReverseDnsSetting.REQUIRE_RESOLVE_ALL_ADDRESSES,),
                              authentication=SASL,
                              sasl_mechanism=KERBEROS, auto_bind=True)
            ldap_filter = '(objectclass=group)'
            ldap_attrs = ["cn", "sn", "givenName"]
            distinguished_name_list = env_vars[DIRECTORY_NAME].split('.')
            distinguished_name = ""
            for name in distinguished_name_list:
                distinguished_name += "dc=" + name + ','
            distinguished_name = distinguished_name.strip(',')
            print("Getting LDAP info for DN = " + distinguished_name)
            conn.search(distinguished_name, ldap_filter, attributes=ldap_attrs)
            for entry in conn.entries:
                if re.findall("Administrator", str(entry)):
                    print("WARNING: User in administrator group: " + str(entry),
                          flush=True)
            print("LDAP check done", flush=True)
            ldap_check_status = True
            break
        except:
            print("LDAP check failed using DNS server = " + server, flush=True)
            continue
    if not ldap_check_status:
        print("Warning** LDAP check failed", flush=True)
        raise NameError("Warning**: LDAP check failed")
    else:
        print("LDAP check succeeded")
    return