LDAP Backup und Restore, GCStruct-Backup

backup.py

@@ -0,0 +1,29 @@
+import typer
+
+import config
+from misc import apache_ldap, file_backup
+
+app = typer.Typer()
+cfg = config.Config()
+
+
+@app.command()
+def gcstruct():
+    file_backup.gcstruct_backup(cfg.portal_dir)
+
+
+@app.command()
+def ldap():
+    cred = cfg.cognos11.credentials
+    apache_ldap.ldap_backup(cred.username, cred.password, cfg.cognos11.config_dir + "\\apacheds_backup.ldif")
+
+
+@app.command()
+def deployment():
+    file_backup.copy_deployment(cfg.cognos11.server_dir, cfg.portal_dir + "\\Content\\Deployment")
+
+
+if __name__ == "__main__":
+    # app()
+    # ldap_backup()
+    deployment()

gctools.py

@@ -1,14 +1,19 @@
 import typer
-import c11
+
+import backup
 import c7
+import c11
 import db
-import xls
+import misc2
 import status
+import xls
 
 app = typer.Typer()
+app.add_typer(backup.app, name="backup")
 app.add_typer(c11.app, name="c11")
 app.add_typer(c7.app, name="c7")
 app.add_typer(db.app, name="db")
+app.add_typer(misc2.app, name="misc")
 app.add_typer(xls.app, name="excel")
 app.add_typer(status.app, name="status")
 

misc/apache_ldap.py

@@ -1,42 +1,99 @@
-from ldap3 import ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, Connection, Server
-
-# from ldap3.core.exceptions import LDAPCursorError
-
-
-def connect_ldap3():
-    server = Server("localhost:10389", get_info=ALL, use_ssl=False)
-    # conn = Connection(server, 'uid=admin,ou=system', 'gc01gapsC$', auto_bind='TLS_AFTER_BIND')  # 'uid=admin,ou=system'
-    # status = conn.search('ou=cognos,dc=ibm,dc=com', '(objectclass=person)', 'SUBTREE')
-    # print(conn.entries)
-    conn = Connection(server, user="uid=Global1,ou=cognos,dc=ibm,dc=com", password="Cognos#11")
-    if conn.bind():
-        conn.search(
-            "ou=cognos,dc=ibm,dc=com",
-            "(objectclass=person)",
-            "SUBTREE",
-            attributes=[ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES],
-        )
-        format_string = "{:15} {:25} {:19} {:25} {}"
-        for e in conn.entries:
-            desc = ""
-            email = ""
-            if "description" in e:
-                desc = e.description
-            if "mail" in e:
-                email = e.mail
-            # print(e.entry_to_json())
-            print(format_string.format(str(e.uid), str(e.givenName), str(e.createTimestamp)[:19], str(email), desc))
-    else:
+import base64
+import hashlib
+import os
+
+import ldap3
+
+LDAP_SERVER = "localhost:10389"
+LDAP_BASE_DN = "ou=cognos,dc=ibm,dc=com"
+
+
+def ldap_connect(username: str, password: str) -> ldap3.Connection:
+    if not username.startswith("uid="):
+        username = f"uid={username},{LDAP_BASE_DN}"
+    server = ldap3.Server(LDAP_SERVER, get_info=ldap3.ALL, use_ssl=False)
+    conn = ldap3.Connection(server, user=username, password=password)
+    if not conn.bind():
         print(conn.result)
+        return None
+    return conn
+
+
+def ldap_backup(username: str, password: str, backup_file: str):
+    conn = ldap_connect(username, password)
+    if not conn:
+        return
+
+    conn.search(
+        LDAP_BASE_DN,
+        "(objectclass=person)",
+        "SUBTREE",
+        attributes=[ldap3.ALL_ATTRIBUTES, ldap3.ALL_OPERATIONAL_ATTRIBUTES],
+    )
+    with open(backup_file, "w", encoding="latin-1", newline="") as fwh:
+        fwh.write(conn.response_to_ldif())
+
+    format_string = "{:15} {:25} {:19} {:25} {}"
+    print(format_string.format("UID", "Name", "erstellt", "E-Mail", "Details"))
+    for e in conn.entries:
+        # print(e.entry_to_ldif())
+        # print(e.entry_to_json())
+        desc = ""
+        email = ""
+        if "description" in e:
+            desc = e.description
+        if "mail" in e:
+            email = e.mail
+        print(format_string.format(str(e.uid), str(e.givenName), str(e.createTimestamp)[:19], str(email), desc))
+    conn.unbind()
+
+
+def ldap_restore(username: str, password: str, backup_file: str):
+    conn = ldap_connect(username, password)
+    if not conn:
+        return
+
+    with open(backup_file, "r", encoding="latin-1") as ldif_file:
+        ldif_data = ldif_file.read()
+
+    ldif = ldap3.LDIF(ldif_data)
+    for entry in ldif.entries:
+        dn = entry["dn"]
+        object_class = entry["objectClass"]
+        attributes = entry["attributes"]
+
+        conn.search(dn, "(objectClass=*)", search_scope=ldap3.SUBTREE)
+        if conn.entries:
+            print(f"Eintrag {dn} existiert bereits.")
+        else:
+            conn.add(dn, object_class, attributes)
+
+    conn.unbind()
+
+
+def ldap_change_admin_password(old_password: str, new_password: str):
+    admin_user = "uid=admin,ou=system"
+    conn = ldap_connect(admin_user, old_password)
+    if not conn:
+        print("Admin-Passwort falsch!")
+        return False
+    if new_password == "":
+        return True
+    ssha_password = create_ssha_password(new_password)
+    conn.modify(admin_user, {"userPassword": [(ldap3.MODIFY_REPLACE, [ssha_password])]})
+    conn.unbind()
+    print("Passwort-Aenderung erfolgreich!")
+    return True
+
+
+def create_ssha_password(password: str) -> str:
+    salt = os.urandom(4)
+    sha = hashlib.sha1(password.encode("utf-8"))
+    sha.update(salt)
+    return "{SSHA}" + base64.b64encode(sha.digest() + salt).decode("utf-8")
 
 
 if __name__ == "__main__":
-    # connect_pyldap()
-    connect_ldap3()
-
-# from ldap3 import Server, Connection, AUTH_SIMPLE, STRATEGY_SYNC, ALL
-# s = Server(HOST, port=389, get_info=ALL)
-# c = Connection(s, authentication=AUTH_SIMPLE, user=user_dn, password=PASSWORD, check_names=True,
-# lazy=False, client_strategy=STRATEGY_SYNC, raise_exceptions=True)
-# c.open()
-# c.bind()
+    # ldap_backup()
+    ldap_change_admin_password("test12test", "gc01gapsC$")
+    # ldap_recovery("uid=admin,ou=system", "gc01gapsC$", "test.ldif")

misc/file_backup.py

@@ -0,0 +1,86 @@
+import hashlib
+import os
+import shutil
+from datetime import datetime
+from pathlib import Path
+
+
+def gcstruct_backup(base_dir: str) -> None:
+    print("Sicherung fuer GCStruct")
+    for f in Path(base_dir).rglob("Kontenrahmen.csv"):
+        if f.parent.name == "Kontenrahmen":
+            print("* " + str(f))
+            copy_file_with_timestamp(str(f), str(f.parent / "Backup"))
+
+
+def copy_file_with_timestamp(file_path: str, backup_folder: str) -> None:
+    """
+    Copies a file to a backup folder with a timestamp in the filename
+    only if the file has changed since last run.
+    """
+    if not Path(backup_folder).exists():
+        os.makedirs(backup_folder, exist_ok=True)
+
+    file_mod_timestamp = datetime.fromtimestamp(Path(file_path).stat().st_mtime).strftime("%Y%m%d_%H%M%S")
+    file_extension = Path(file_path).suffix
+    file_basename = Path(file_path).stem
+    backup_filename = f"{file_basename}__{file_mod_timestamp}{file_extension}"
+
+    if Path(backup_folder + "\\" + backup_filename).exists():
+        # backup folder is already up-to-date
+        return
+
+    latest_file = get_file_with_latest_timestamp(backup_folder, file_basename + "__")
+    if latest_file != "" and hash(file_path) == hash(latest_file):
+        # different timestamp, but same content. Keep old timestamp and backup file.
+        return
+
+    shutil.copy2(file_path, backup_folder + "\\" + backup_filename)
+    return
+
+
+def get_file_with_latest_timestamp(folder_path: str, filename_pattern: str) -> str:
+    """
+    Gets the latest timestamp from files in a folder with a matching filename.
+    """
+    files = [str(f) for f in Path(folder_path).iterdir() if f.name.startswith(filename_pattern)]
+    if len(files) == 0:
+        return ""
+    files.sort()
+    return files[-1]
+
+
+def hash(filename: str) -> str:
+    with open(filename, "rb") as frh:
+        return hashlib.sha256(frh.read()).hexdigest()
+
+
+def main() -> None:
+    gcstruct_backup("C:\\GlobalCube_Entwicklung")
+
+
+def copy_deployment(server_folder: str, backup_folder: str):
+    filename = "GC_BACKUP_CONTENT_STORE_DAILY.zip"
+    source_file = Path(server_folder + "\\deployment\\" + filename)
+    if not source_file.exists():
+        print(f"!! Datei {filename} existiert nicht !!")
+        return
+    file_ts = source_file.stat().st_mtime
+    print(filename + " - " + datetime.fromtimestamp(file_ts).strftime("%d.%m.%Y %H:%M:%S"))
+
+    if file_ts < datetime.now().timestamp() - 3 * 24 * 60 * 60:
+        print(f"!! Datei {filename} ist aelter als 3 Tage !!")
+        return
+
+    if not Path(backup_folder).exists():
+        os.makedirs(backup_folder, exist_ok=True)
+
+    dest_file = Path(f"{backup_folder}\\{filename}")
+    if dest_file.exists() and file_ts == dest_file.stat().st_mtime:
+        print(f"Datei {filename} ist bereits up-to-date.")
+        return
+    shutil.copy2(source_file, dest_file)
+
+
+if __name__ == "__main__":
+    main()

misc/headers.py

@@ -2,17 +2,16 @@ import os
 from pathlib import Path
 
 
-def main(base_dir):
-    path = Path(base_dir)
-    target = path.joinpath("Kopfzeilen")
-    os.makedirs(target, exist_ok=True)
+def create_headerfiles(base_dir: str):
+    """opens every *.csv in base_dir and saves first line to subfolder "Kopfzeilen" """
+    dest = Path(f"{base_dir}\\Kopfzeilen")
+    os.makedirs(dest, exist_ok=True)
 
-    for filename in Path(base_dir).glob("*.csv"):
-        with open(filename, "r", encoding="latin-1") as frh, open(
-            target.joinpath(filename.name), "w", encoding="latin-1"
-        ) as fwh:
+    for source_file in Path(base_dir).glob("*.csv"):
+        dest_file = dest / source_file.name
+        with source_file.open("r", encoding="latin-1") as frh, dest_file.open("w", encoding="latin-1") as fwh:
             fwh.write(frh.readline())
 
 
 if __name__ == "__main__":
-    main("C:/GlobalCube/System/OPTIMA/Export")
+    create_headerfiles("C:/GlobalCube/System/OPTIMA/Export")

misc2.py

@@ -0,0 +1,43 @@
+import typer
+
+import config
+from misc import apache_ldap
+from misc.headers import create_headerfiles
+
+app = typer.Typer()
+cfg = config.Config()
+
+
+@app.command()
+def headers():
+    create_headerfiles(cfg.system_dir + "\\Export")
+
+
+@app.command()
+def ldap_backup():
+    cred = cfg.cognos11.credentials
+    apache_ldap.ldap_backup(cred.username, cred.password, cfg.cognos11.config_dir + "\\apacheds_backup.ldif")
+
+
+@app.command()
+def ldap_restore(backup_file: str):
+    cred = cfg.cognos11.credentials
+    apache_ldap.restore_ldap(cred.username, cred.password, backup_file)
+
+
+@app.command()
+def ldap_password(old_password: str, new_password: str) -> bool:
+    return apache_ldap.ldap_change_admin_password(old_password, new_password)
+
+
+@app.command()
+def ldap_admin():
+    if ldap_password("gc01gapsC$", ""):
+        print("Passwort ist bereits aktuell.")
+    for p in ["Cognos#11Cognos#11", "mEn$q3b%P9p1j!A-ku", "Never4getpw#!!##"]:
+        if ldap_password(p, "gc01gapsC$"):
+            break
+
+
+if __name__ == "__main__":
+    headers()