Scripts for Linuxmuster import of custom E-Mail Adresses.

README.md

Scripts for Linuxmuster import of custom E-Mail Adresses.

CAUTION!

This may break your LDAP, use at your own risk and only if you know what you are doing!

Usage

• Put initializeUsersCreatedInLastImport.py, ldapHelper.py and lmnListImportFilterScript.py into /root/scripts on the school server.
• Link initializeUsersCreatedInLastImport.py to /etc/linuxmuster/sophomorix/hooks/sophomorix-add.d/50-set-firstpassword-and-email.py
• Go to the WebUI in Globale Einstellungen > Listenimport > Pfad zum Filterskript (at the very bottom)
• Enter /root/scripts/lmnListImportFilterScript.py into the students.csv field and save
• You can then use the Listenverwaltung > CSV > CSV Laden function to upload your custom CSV file:
• The CSV you upload has to look like this: class;lastname;firstname;birthdate;id;email
• All the emails should be in /etc/linuxmuster/sophomorix/default-school/students.extra.csv after the upload
• Once you import the users, the email of all new users will be written to sophomorixCustom1
• If you want to set the emails of existing users, you may use the importAllMailAddresses.py Script

importAllMailAddresses.py

#!/usr/bin/python3

import ldapHelper
import sys
import csv

if not ldapHelper.bindAsAdmin():
    sys.exit(1)

rc, users = ldapHelper.search("(sophomorixRole=student)")
#rc, users = ldapSearch("(&(!(sophomorixCustom1=*linuxmuster.lan))(sophomorixRole=student)(sophomorixAdminClass=mlm))")

EXTRA_USER_ATTRIBUTES_STORE="/etc/linuxmuster/sophomorix/default-school/students.extra.csv"

if not rc:
    print("Error")
    sys.exit(1)

def getExtraUserAttributes():
    """
    Get all extra user attributes from import
    """
    global EXTRA_USER_ATTRIBUTES_STORE

    with open(EXTRA_USER_ATTRIBUTES_STORE, encoding="utf-8") as extraAttributesFile:
        csvReader = csv.reader(extraAttributesFile, delimiter=';')

        extraUserAttributes = {}
        for line in csvReader:
            extraUserAttributes[line[0]] = {
                "email": line[1]
            }

        return True, extraUserAttributes

def setFirstEmailOfUser(user):
    username = user["username"]
    print(f"  * Importing email of {username}... ", end="")
    rc, ldapUser = ldapHelper.searchOne(f"sAMAccountName={username}")
    if not rc:
        print("-> Error reading user from LDAP!")
        return False

    if "email" in user:
        newMail = user["email"]
    else:
        newMail = f"{username}@linuxmuster.lan"

    rc = ldapHelper.setAttribute(ldapUser["distinguishedName"], "sophomorixCustom1", newMail)

    print("-> OK" if rc else "-> Error writing to LDAP")

rc, extraUserAttributes = getExtraUserAttributes()

for user in users:
    if user["sophomorixUnid"] != "---" and user["sophomorixUnid"] in extraUserAttributes:
        setFirstEmailOfUser({"username": user["sAMAccountName"], "email": extraUserAttributes[user["sophomorixUnid"]]["email"]})

initializeUsersCreatedInLastImport.py

#!/usr/bin/python3

import ldapHelper
import sys, csv

EXTRA_USER_ATTRIBUTES_STORE="/etc/linuxmuster/sophomorix/default-school/students.extra.csv"

print("Start")

# Functions

# Sophomorix logfile stuff

def getRecentlyAddedUsers():
    """
    Get all recently added users from sophomorix logfile
    """

    rc, extraUserAttributes = getExtraUserAttributes()
    if not rc:
        return False, None

    with open("/var/log/sophomorix/userlog/user-add.log", encoding="utf-8") as sophomorixAddLogfile:
        csvReader = csv.reader(sophomorixAddLogfile, delimiter=':')

        recentlyAddedUsers = []
        for line in csvReader:
            if line[0] != "ADD":
                continue

            user = {
                "importTimestamp": line[2],
                "school": line[6],
                "username": line[8],
                "lastName": line[10],
                "firstName": line[12],
                "class": line[14],
                "role": line[16],
                "id": line[18]
            }

            if user["id"] != "---" and user["id"] in extraUserAttributes:
                user = {**user, **extraUserAttributes[user["id"]]}

            recentlyAddedUsers.append(user)


        return True, recentlyAddedUsers

def getUsersAddedInLastImport():
    """
    Get all users that were added during the last import
    """
    rc, recentlyAddedUsers = getRecentlyAddedUsers()
    if not rc:
        return False, None

    usersAddedInLastImport = []
    timestampOfLastImport = -1

    for recentlyAddedUser in reversed(recentlyAddedUsers):
        if timestampOfLastImport == -1:
            timestampOfLastImport = recentlyAddedUser["importTimestamp"]

        if timestampOfLastImport != recentlyAddedUser["importTimestamp"]:
            break

        usersAddedInLastImport.append(recentlyAddedUser)

    return True, usersAddedInLastImport

def getExtraUserAttributes():
    """
    Get all extra user attributes from import
    """
    global EXTRA_USER_ATTRIBUTES_STORE

    with open(EXTRA_USER_ATTRIBUTES_STORE, encoding="utf-8") as extraAttributesFile:
        csvReader = csv.reader(extraAttributesFile, delimiter=';')

        extraUserAttributes = {}
        for line in csvReader:
            extraUserAttributes[line[0]] = {
                "email": line[1]
            }

        return True, extraUserAttributes

# Ldap user stuff

def setFirstPasswordOfUser(username):
    print("  * Processing first password ... ", end="")
    rc, ldapUser = ldapHelper.searchOne("sAMAccountName={}".format(user["username"]))
    if not rc:
        print("-> Error reading user from LDAP!")
        return False

    desiredFirstPassword = ldapUser["sophomorixBirthdate"] + "Muster!"

    if desiredFirstPassword == ldapUser["sophomorixFirstPassword"]:
        print("-> nothing to do")
        return True

    rc = ldapHelper.setUserPassword(ldapUser["distinguishedName"], desiredFirstPassword, isFirstPassword=True)

    print("-> OK" if rc else "-> Error writing to LDAP")
    return rc

def setFirstEmailOfUser(user):
    username = user["username"]
    print("  * Processing first email ... ", end="")
    rc, ldapUser = ldapHelper.searchOne(f"sAMAccountName={username}")
    if not rc:
        print("-> Error reading user from LDAP!")
        return False

    if "sophomorixCustom1" in ldapUser:
        print("-> nothing to do")
        return True

    if "email" in user:
        newMail = user["email"]
    else:
        print("-> No mail found in file!")
        newMail = f"{username}@linuxmuster.lan"
        return True

    rc = ldapHelper.setAttribute(ldapUser["distinguishedName"], "sophomorixCustom1", newMail)

    print("-> OK" if rc else "-> Error writing to LDAP")
    return rc

# end - Functions

if not ldapHelper.bindAsAdmin():
    sys.exit(1)

rc, usersAddedInLastImport = getUsersAddedInLastImport()
if not rc:
    sys.exit(1)

for user in usersAddedInLastImport:
    if user["class"] not in ["mlm", "extern"]:
        continue

    print("* Processing user {}".format(user["username"]))
    #setFirstPasswordOfUser(user["username"])
    setFirstEmailOfUser(user)

ldapHelper.py

#!/usr/bin/python3

import csv
import ldap
import sys
import base64

_ldapConnection = None

def bindAsAdmin():
    global _ldapConnection

    try:
        with open('/etc/linuxmuster/.secret/administrator') as f: ldapPassword = f.read()
    except:
        print("Error getting ldap password!")
        return False

    try:
        ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
        _ldapConnection = ldap.initialize("ldaps://localhost")
        _ldapConnection.protocol_version = 3
        _ldapConnection.set_option(ldap.OPT_REFERRALS, 0)
        _ldapConnection.simple_bind_s("administrator@linuxmuster.lan", ldapPassword)
        return True
    except:
        print("Error binding to LDAP!")
        return False

def search(filter):
    global _ldapConnection
    if _ldapConnection == None:
        print("Cannot talk to LDAP")
        return False, None

    try:
        rawResults = _ldapConnection.search_s(
                "DC=linuxmuster,DC=lan",
                ldap.SCOPE_SUBTREE,
                filter
                )
    except Exception as e:
        print("Error executing LDAP search!")
        print(e)
        return False, None

    try:
        processedResults =  []

        if len(rawResults) <= 0 or rawResults[0][0] == None:
            print("Search \"{}\" did not return any objects".format(filter))
            return False, None

        for dn, rawResult in rawResults:
            if not dn:
                continue

            processedResult = {}
            
            for attribute, rawValue in rawResult.items():
                try:
                    if len(rawValue) == 1:
                        processedResult[attribute] = str(rawValue[0].decode())
                    elif len(rawValue) > 0:
                        processedResult[attribute] = []
                        for rawItem in rawValue:
                            processedResult[attribute].append(str(rawItem.decode()))

                except UnicodeDecodeError:
                    continue

            processedResults.append(processedResult)
                
        return True, processedResults

    except Exception as e:
        print(e)
        return False, None

def searchOne(filter):
    ret, results = search(filter)
    if not ret:
        return False, None
    
    return True, results[0]

def setUserPassword(dn, newPassword, isFirstPassword=False):
    encodedPassword =  '"{0}"'.format(newPassword).encode('utf-16-le')
    ldif = [(ldap.MOD_REPLACE, 'unicodePwd', [encodedPassword])]

    if isFirstPassword:
        ldif.append((ldap.MOD_REPLACE, 'sophomorixFirstPassword', [newPassword.encode('utf-8')]))

    try:
        _ldapConnection.modify_s(dn, ldif)
        return True
    except Exception as e:
        print(e)
        return False

def setAttribute(dn, attribute, value):
    ldif = [(ldap.MOD_REPLACE, attribute, [value.encode('utf-8')])]
    try:
        _ldapConnection.modify_s(dn, ldif)
        return True
    except Exception as e:
        print(e)
        return False

lmnListImportFilterScript.py

#!/usr/bin/python3

import argparse, csv

parser = argparse.ArgumentParser(description='This is a filter script for use with linuxmuster.net\nSee: https://github.com/linuxmuster/sophomorix4/wiki/Export-and-Filters#filters')
parser.add_argument('input', help='Absolute file path of the input file')
parser.add_argument('output', help='Absolute file path of the output file')

args = parser.parse_args()

EXTRA_USER_ATTRIBUTES_STORE="/etc/linuxmuster/sophomorix/default-school/students.extra.csv"

with open(args.input) as inputFile, open(args.output, "w", newline='') as outputFile, open(EXTRA_USER_ATTRIBUTES_STORE, "w", newline="") as extraOutputFile:
    csvReader = csv.reader(inputFile, delimiter=';')
    next(csvReader, None) # skip header

    csvWriter = csv.writer(outputFile, delimiter=';')
    extraCsvWriter = csv.writer(extraOutputFile, delimiter=';')

    for line in csvReader:
        csvWriter.writerow([line[0], line[1], line[2], line[3], line[4]])
        extraCsvWriter.writerow([line[4], line[5]])