Python v3.5 script to compare UID in IPA and DSEE (LDAP)

• Intro
• Design
• Usage
• Script

Intro

There are two LDAP directories {(Free)IPA and DSEE} and the task is to compare UID. It also checks if user is with company, present in corporate Aria's LDAP. This script does that.

Design

This is python 3.5 script, it's to be run on FreeIPA client (see function admin_kinit), since first it gets list of all IPA users. Then it goes through that list and check is user exists in DSEE and Aria.
The script creates 4 output files in the same directory where script is run. The files are (with examples):

NOT_in_DSEE_<date_time> (users not in DSEE)

 

No DSEE account: admin
No DSEE account: amandabackup
No DSEE account: dhcp-service


ALL_<date_time> (complete output)

 

 ---- IPA-----: milan
 IPA UID: 1120149
DSEE UID: 1120149
Aria login: milang and email: milan.misic@comp.com
 ---- IPA-----: alisa
 IPA UID: 647915
DSEE UID: 647915
Aria login: alisa and email: alisa.yo@ocomp.com
 ---- IPA-----: tota
 IPA UID: 33588
DSEE UID: 33588
Aria login: tota and email: tota.mama@comp.com


NOT_in_ARIA_<date_time> (users not in Aria, not with Company)

 

No ARIA account: zarko
No ARIA account: zarkod


IPA-DSEE-diff_<date_time> (UID difference)

saska: IPA: 1636 DSEE: 648645
voja: IPA: 485526048 DSEE: 642980
luka: IPA: 485526105 DSEE: 713293

Usage

Run the script without any arguments.

Script

 

#!/bin/python3.5

import sys
import ldap
import subprocess
import logging, logging.handlers
from time import gmtime, strftime
import datetime

# --- Define log file
log_ipa_dsee_diff = ("IPA-DSEE-diff_" + datetime.datetime.now().strftime("%m-%d-%Y_%Hh%Mm%Ss"))
log_not_in_aria = ("NOT_in_ARIA_" + datetime.datetime.now().strftime("%m-%d-%Y_%Hh%Mm%Ss"))
log_not_in_dsee = ("NOT_in_DSEE_" + datetime.datetime.now().strftime("%m-%d-%Y_%Hh%Mm%Ss"))
log_all = ("ALL_" + datetime.datetime.now().strftime("%m-%d-%Y_%Hh%Mm%Ss"))
# create logger (interface that script uses for logging)
logger1 = logging.getLogger("diff")
logger1.setLevel(logging.INFO)
logger2 = logging.getLogger("notaria")
logger2.setLevel(logging.INFO)
logger3 = logging.getLogger("notdsee")
logger3.setLevel(logging.INFO)
logger4 = logging.getLogger("all")
logger4.setLevel(logging.INFO)
# create file handler (define log destination)
handler1 = logging.handlers.TimedRotatingFileHandler(log_ipa_dsee_diff, when='MIDNIGHT', backupCount=50, utc=False)
handler1.setLevel(logging.INFO)
handler2 = logging.handlers.TimedRotatingFileHandler(log_not_in_aria, when='MIDNIGHT', backupCount=50, utc=False)
handler2.setLevel(logging.INFO)
handler3 = logging.handlers.TimedRotatingFileHandler(log_not_in_dsee, when='MIDNIGHT', backupCount=50, utc=False)
handler3.setLevel(logging.INFO)
handler4 = logging.handlers.TimedRotatingFileHandler(log_all, when='MIDNIGHT', backupCount=50, utc=False)
handler4.setLevel(logging.INFO)
# create formatter (define layout of logs)
formatter = logging.Formatter('%(message)s')
handler1.setFormatter(formatter)
handler2.setFormatter(formatter)
handler3.setFormatter(formatter)
handler4.setFormatter(formatter)
# add handler to logger
logger1.addHandler(handler1)
logger2.addHandler(handler2)
logger3.addHandler(handler3)
logger4.addHandler(handler4)

def admin_kinit():
    """
    Obtain and cache Kerberos ticket-granting ticket for admin
    admin is IPA administrator account
    """
    try:
        print("Obtaining Kerberos ticket-granting ticket for admin (IPA Administrator)")
        subprocess.call(['sudo -u admin kinit -kt /homelocal/admin/.ipa/admin.kt admin'], shell=True)
    except subprocess.CalledProcessError as err:
        print("Admin account (IPA Administrator) cannot obtain Kerberos ticket-granting ticket" )
        sys.exit("subprocess.CalledProcessError: {0}".format(err))

def admin_kerberos_ticket():
    """
    Does admin  have  valid kerberos ticket?
    """
    return True if subprocess.call(['sudo', '-u', 'admin', 'klist', '-s']) == 0 else False

def dsee_search(user):
    """
    Search DSEE with search filter as argument
    """
    DSEE_SRV="dsee-server.comp.com"
    base_DN = "l=amer,dc=comp,dc=com"
    search_Scope = ldap.SCOPE_SUBTREE # subtree search
    search_Filter = "uid=" + user  # this is login, not UID
    retrieve_Attributes = None
    global dsee_uid

    # create ldap object
    try:
        ldapobject = ldap.initialize('ldap://' + DSEE_SRV)
        ldapobject.protocol_version = ldap.VERSION3
    except ldap.LDAPError as err:
        sys.exit("ldap.LDAPError: {0}".format(err))

    try:
        l_search = ldapobject.search(base_DN, search_Scope, search_Filter, retrieve_Attributes)
        result_status, result_data = ldapobject.result(l_search, 0)
        if not result_data:
            print("No DSEE account: " + user)
            dsee_uid = None
            logger3.info("No DSEE account: " + user)
        else:
            dsee_uid = (result_data[0][1])["uidNumber"][0].decode()
            print("DSEE UID: " + (result_data[0][1])["uidNumber"][0].decode())
            logger4.info("DSEE UID: " + (result_data[0][1])["uidNumber"][0].decode())

    except ldap.LDAPError as err:
        sys.exit("ldap.LDAPError: {0}".format(err))


def aria_search(user):
    """
    Search Aria with search filter as argument
    """
    ARIA_SRV="aria-server.comp.com"
    base_DN = "dc=comp,dc=com"
    search_Scope = ldap.SCOPE_SUBTREE # subtree search
    search_Filter = "uid=" + user  # this is login, not UID
    retrieve_Attributes = None

    # create ldap object
    try:
        ldapobject = ldap.initialize('ldap://' + ARIA_SRV)
        ldapobject.protocol_version = ldap.VERSION3
    except ldap.LDAPError as err:
        sys.exit("ldap.LDAPError: {0}".format(err))

    try:
        l_search = ldapobject.search(base_DN, search_Scope, search_Filter, retrieve_Attributes)
        result_status, result_data = ldapobject.result(l_search, 0)
        if not result_data:
            print("A user is not with Company.")
            logger2.info("No ARIA account: " + user)
        else:
            print("Aria login: " + (result_data[0][1])["uid"][0].decode() + " and email: " + (result_data[0][1])["mail"][0].decode() )
            logger4.info("Aria login: " + (result_data[0][1])["uid"][0].decode() + " and email: " + (result_data[0][1])["mail"][0].decode() )
    except ldap.LDAPError as err:
        sys.exit("ldap.LDAPError: {0}".format(err))


def compare_uid():
    """
    List all IPA users
    """
    print("Getting IPA users")
    logger4.info("Getting IPA users")
    try:
        ipa_users = subprocess.check_output(['sudo -u admin ipa user-find | grep \"User login\" | awk \'{print $3}\' '], shell=True)
    except subprocess.SubprocessError as err:
        print("Can't get list of IPA users")
        logger4.info("Can't get list of IPA users")
        sys.exit("SubprocessError: {0}".format(err))

    print(type(ipa_users.decode().split("\n")))  # convert string into list
    print(ipa_users.decode().split("\n"))
    for ipa_login in ipa_users.decode().split("\n"):
        ipa_uid = subprocess.check_output(['sudo -u admin ipa user-show %s --raw | grep uidnumber | \
                                             awk \'{print $2}\'' % ipa_login], shell=True)
        try:
            print(" ---- IPA-----: " + ipa_login)
            logger4.info(" ---- IPA-----: " + ipa_login)
            print(" IPA UID: " + ipa_uid.decode().strip())
            logger4.info(" IPA UID: " + ipa_uid.decode().strip())
            ipa_uid = ipa_uid.decode().strip()
            dsee_search(ipa_login)
            aria_search(ipa_login)
            # compare IPA and DSEE uid
            if dsee_uid != None and dsee_uid != ipa_uid:
                logger1.info(ipa_login + ": IPA: " + ipa_uid + " DSEE: " + dsee_uid)
         except:
            print("problem ....")


if __name__ == '__main__':

    admin_kinit()

    print("Check if admin has kerberos ticket")
    if not admin_kerberos_ticket():
        sys.exit("Admin doesn't have a kerberos ticket!")
    print("OK, admin has kerberos ticket")

    compare_uid()

sys.exit(0)



Back to the main page