diff --git a/nettmanage.py b/nettmanage.py new file mode 100644 index 0000000..b0d11a7 --- /dev/null +++ b/nettmanage.py @@ -0,0 +1,1147 @@ +#!/usr/bin/env python3 +import json +import os +import re +import shutil +import subprocess +import sys +import tarfile +import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request +import zipfile +from datetime import datetime + +# ========================================== +# Standard-Konfiguration +# ========================================== +CONFIG = { + "netbox_path": "/opt/netbox", + "db_host": "localhost", + "db_name": "netbox", + "db_user": "netbox", + "db_pass": "", + "backup_dir": "/tmp", + # Leer lassen, damit upgrade.sh den Standard-Python-Interpreter verwendet. + # Beispiel für NetBox 4.5/4.6: /usr/bin/python3.12 + "python_path": "", +} + +NETBOX_SERVICES = ("netbox", "netbox-rq") + +# Diese Einträge werden beim Upgrade auf Wunsch ergänzt. Bereits vorhandene +# Einträge (auch mit Versions-Pin) werden nicht doppelt hinzugefügt. +DEFAULT_PLUGIN_REQUIREMENTS = ( + "netbox-topology-views", + "netbox-attachments", + "netbox-ping", + "netbox-reorder-rack", + "netbox-secrets", + "netbox-inventory", + "git+https://git.mrblake.cc/MrBlake/NetBox-VM-Import.git#egg=netbox_vmware_importer", +) +LDAP_REQUIREMENT = "django-auth-ldap" + + +# ========================================== +# Hilfsfunktionen für die Shell-GUI +# ========================================== +class Colors: + HEADER = "\033[95m" + BLUE = "\033[94m" + GREEN = "\033[92m" + WARNING = "\033[93m" + FAIL = "\033[91m" + ENDC = "\033[0m" + BOLD = "\033[1m" + + +def clear_screen(): + os.system("clear" if os.name == "posix" else "cls") + + +def print_header(title): + clear_screen() + print(Colors.HEADER + Colors.BOLD + "=" * 60 + Colors.ENDC) + print(Colors.HEADER + Colors.BOLD + f" {title}".ljust(59) + Colors.ENDC) + print(Colors.HEADER + Colors.BOLD + "=" * 60 + Colors.ENDC + "\n") + + +def log_info(msg): + print(Colors.BLUE + f"[*] {msg}" + Colors.ENDC) + + +def log_success(msg): + print(Colors.GREEN + Colors.BOLD + f"[+] {msg}" + Colors.ENDC) + + +def log_error(msg): + print(Colors.FAIL + Colors.BOLD + f"[-] {msg}" + Colors.ENDC) + + +def log_warn(msg): + print(Colors.WARNING + f"[!] {msg}" + Colors.ENDC) + + +def pause(): + input(Colors.BOLD + "\nDrücke [ENTER], um ins Hauptmenü zurückzukehren..." + Colors.ENDC) + + +def run_command(command, *, env=None, cwd=None, capture=False): + """Führt einen Befehl ohne Shell aus und liefert CompletedProcess zurück.""" + kwargs = { + "env": env, + "cwd": cwd, + "text": True, + "check": False, + } + if capture: + kwargs["stdout"] = subprocess.PIPE + kwargs["stderr"] = subprocess.PIPE + return subprocess.run(command, **kwargs) + + +def unique_path(path): + """Erzeugt einen noch nicht belegten Pfad durch Anhängen einer Nummer.""" + if not os.path.lexists(path): + return path + + counter = 1 + while True: + candidate = f"{path}_{counter}" + if not os.path.lexists(candidate): + return candidate + counter += 1 + + +def safe_extract_zip(zipf, destination): + """Verhindert Zip-Slip beim Entpacken von Backup-Dateien.""" + destination_real = os.path.realpath(destination) + for member in zipf.infolist(): + target = os.path.realpath(os.path.join(destination, member.filename)) + if os.path.commonpath((destination_real, target)) != destination_real: + raise ValueError(f"Unsicherer ZIP-Pfad erkannt: {member.filename}") + zipf.extractall(destination) + + +def safe_extract_tar(tarf, destination): + """Entpackt ein Release-Archiv ohne Pfad-Traversal oder Archiv-Symlinks.""" + destination_real = os.path.realpath(destination) + for member in tarf.getmembers(): + target = os.path.realpath(os.path.join(destination, member.name)) + if os.path.commonpath((destination_real, target)) != destination_real: + raise ValueError(f"Unsicherer TAR-Pfad erkannt: {member.name}") + if member.issym() or member.islnk(): + raise ValueError(f"Symlink/Hardlink im TAR-Archiv nicht erlaubt: {member.name}") + tarf.extractall(destination) + + +# ========================================== +# Kernfunktionen: Backup & Restore +# ========================================== +def execute_backup(): + print_header("BACKUP PROZESS") + + log_info("Konfiguriere den Backup-Umfang:") + do_db = input(" 1. Datenbank sichern? (J/n): ").strip().lower() != "n" + + exclude_changelog = False + exclude_users = False + + if do_db: + exclude_changelog = ( + input( + Colors.BLUE + + " -> Changelog (extras_objectchange) auslassen? Spart viel Platz. (j/N): " + + Colors.ENDC + ) + .strip() + .lower() + == "j" + ) + print( + Colors.WARNING + + " Hinweis: Das Auslassen von Benutzern kann beim Restore zu Foreign-Key-Fehlern führen, falls Objekte diesen Benutzern zugewiesen sind." + + Colors.ENDC + ) + exclude_users = ( + input( + Colors.BLUE + + " -> Benutzer (User, Gruppen, Tokens) auslassen? (j/N): " + + Colors.ENDC + ) + .strip() + .lower() + == "j" + ) + + do_media = input("\n 2. Media-Dateien (Bilder/Dokumente) sichern? (J/n): ").strip().lower() != "n" + + if not do_db and not do_media: + log_warn("Keine Komponenten zum Sichern ausgewählt. Abbruch.") + pause() + return + + media_path = os.path.join(CONFIG["netbox_path"], "netbox/media") + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + archive_name = f"netbox_backup_{stamp}" + zip_file_path = os.path.join(CONFIG["backup_dir"], f"{archive_name}.zip") + tmp_dir = os.path.join(CONFIG["backup_dir"], f"nb_tmp_{stamp}") + + os.makedirs(tmp_dir, exist_ok=True) + + try: + # 1. Datenbank-Dump + if do_db: + log_info("\nSchritt 1: Erstelle PostgreSQL-Datenbank-Dump...") + dump_file = os.path.join(tmp_dir, "netbox_database.dump") + + env = os.environ.copy() + if CONFIG["db_pass"]: + env["PGPASSWORD"] = CONFIG["db_pass"] + + cmd = [ + "pg_dump", + "-Fc", + "-h", + CONFIG["db_host"], + "-U", + CONFIG["db_user"], + "-d", + CONFIG["db_name"], + "-f", + dump_file, + ] + + # --exclude-table-data lässt die Tabellenstruktur intakt, + # exportiert aber keine Inhalte der gewählten Tabellen. + if exclude_changelog: + cmd.append("--exclude-table-data=extras_objectchange") + log_info(" -> Changelog-Daten werden übersprungen.") + if exclude_users: + cmd.extend( + [ + "--exclude-table-data=auth_user", + "--exclude-table-data=auth_group", + "--exclude-table-data=users_token", + ] + ) + log_info(" -> Benutzer-Daten werden übersprungen.") + + process = run_command(cmd, env=env, capture=True) + + if process.returncode != 0: + log_error(f"pg_dump fehlgeschlagen: {process.stderr}") + raise RuntimeError("Datenbank-Backup fehlgeschlagen.") + log_success("PostgreSQL-Dump erfolgreich erstellt.") + else: + log_info("\nSchritt 1: Datenbank-Backup übersprungen.") + + # 2. Media-Dateien + if do_media: + log_info("Schritt 2: Kopiere Media-Dateien (Bilder, Dokumente)...") + if os.path.exists(media_path): + dst_media = os.path.join(tmp_dir, "media") + shutil.copytree(media_path, dst_media, dirs_exist_ok=True) + log_success("Media-Ordner erfolgreich gesichert.") + else: + log_warn(f"Kein Media-Verzeichnis unter {media_path} gefunden.") + else: + log_info("Schritt 2: Media-Backup übersprungen.") + + # 3. Metadaten + log_info("Schritt 3: Erstelle Metadaten...") + meta = { + "version": "1.2", + "timestamp": stamp, + "source_db": CONFIG["db_name"], + "contains_db": do_db, + "contains_media": do_media, + "excluded_changelog": exclude_changelog, + "excluded_users": exclude_users, + } + with open(os.path.join(tmp_dir, "metadata.json"), "w", encoding="utf-8") as file_handle: + json.dump(meta, file_handle, indent=4) + + # 4. ZIP erstellen + log_info(f"Schritt 4: Komprimiere Daten nach {zip_file_path} ...") + with zipfile.ZipFile(zip_file_path, "w", zipfile.ZIP_DEFLATED) as zipf: + for root, _, files in os.walk(tmp_dir): + for filename in files: + full_path = os.path.join(root, filename) + rel_path = os.path.relpath(full_path, tmp_dir) + zipf.write(full_path, rel_path) + + print("\n" + "=" * 60) + log_success("BACKUP ERFOLGREICH ABGESCHLOSSEN!") + log_success(f"Datei liegt hier: {zip_file_path}") + print("=" * 60) + + except Exception as exc: + log_error(f"Kritischer Fehler: {exc}") + finally: + if os.path.exists(tmp_dir): + shutil.rmtree(tmp_dir) + pause() + + +def execute_restore(): + print_header("RESTORE PROZESS") + log_warn("ACHTUNG: Dieser Vorgang überschreibt bestehende Daten auf der Ziel-Instanz!") + confirm = input("Bist du sicher, dass du fortfahren möchtest? (j/N): ") + + if confirm.lower() != "j": + log_info("Abbruch durch Benutzer.") + pause() + return + + restore_file = input("\nPfad zur Backup-ZIP-Datei (z. B. /tmp/netbox_backup.zip): ").strip() + + if not restore_file or not os.path.exists(restore_file): + log_error(f"Die Datei '{restore_file}' existiert nicht!") + pause() + return + + media_path = os.path.join(CONFIG["netbox_path"], "netbox/media") + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + tmp_extract_dir = f"/tmp/nb_restore_{stamp}" + os.makedirs(tmp_extract_dir, exist_ok=True) + + try: + log_info(f"\nSchritt 1: Entpacke {restore_file} ...") + with zipfile.ZipFile(restore_file, "r") as zipf: + safe_extract_zip(zipf, tmp_extract_dir) + + meta_file = os.path.join(tmp_extract_dir, "metadata.json") + if not os.path.exists(meta_file): + raise RuntimeError("Ungültiges Backup: metadata.json fehlt.") + + with open(meta_file, "r", encoding="utf-8") as file_handle: + meta = json.load(file_handle) + + # 2. Datenbank-Restore + if meta.get("contains_db", True): + log_info("Schritt 2: Stelle PostgreSQL-Datenbank wieder her...") + dump_file = os.path.join(tmp_extract_dir, "netbox_database.dump") + if not os.path.exists(dump_file): + raise RuntimeError("Ungültiges Backup: netbox_database.dump fehlt.") + + env = os.environ.copy() + if CONFIG["db_pass"]: + env["PGPASSWORD"] = CONFIG["db_pass"] + + log_info("Lösche und erstelle Datenbank neu (Bereinigung)...") + drop_proc = run_command( + [ + "dropdb", + "-h", + CONFIG["db_host"], + "-U", + CONFIG["db_user"], + "--if-exists", + CONFIG["db_name"], + ], + env=env, + capture=True, + ) + if drop_proc.returncode != 0: + raise RuntimeError(f"dropdb fehlgeschlagen: {drop_proc.stderr.strip()}") + + create_proc = run_command( + [ + "createdb", + "-h", + CONFIG["db_host"], + "-U", + CONFIG["db_user"], + "-O", + CONFIG["db_user"], + CONFIG["db_name"], + ], + env=env, + capture=True, + ) + if create_proc.returncode != 0: + raise RuntimeError(f"createdb fehlgeschlagen: {create_proc.stderr.strip()}") + + log_info("Importiere Dump...") + proc_restore = run_command( + [ + "pg_restore", + "-h", + CONFIG["db_host"], + "-U", + CONFIG["db_user"], + "-d", + CONFIG["db_name"], + "--clean", + "--no-owner", + dump_file, + ], + env=env, + capture=True, + ) + + if proc_restore.returncode != 0: + log_warn( + f"pg_restore wurde mit Code {proc_restore.returncode} beendet:\n" + f"{proc_restore.stderr.strip()}" + ) + log_success("Datenbank-Wiederherstellung abgeschlossen.") + else: + log_info("Schritt 2: Übersprungen (Keine Datenbank im Backup enthalten).") + + # 3. Media-Restore + if meta.get("contains_media", True): + log_info("Schritt 3: Stelle Media-Dateien wieder her...") + src_media = os.path.join(tmp_extract_dir, "media") + if os.path.exists(src_media): + if os.path.exists(media_path): + shutil.rmtree(media_path) + shutil.copytree(src_media, media_path, dirs_exist_ok=True) + run_command(["chown", "-R", "netbox:netbox", media_path], capture=True) + log_success("Media-Dateien wiederhergestellt.") + else: + log_warn("Backup enthält laut Metadaten Media-Dateien, der Ordner fehlt jedoch.") + else: + log_info("Schritt 3: Übersprungen (Keine Media-Dateien im Backup enthalten).") + + # 4. NetBox upgrade.sh + log_info("Schritt 4: Führe NetBox upgrade.sh aus (Cache-Clear & Migrationen)...") + upgrade_script = os.path.join(CONFIG["netbox_path"], "upgrade.sh") + if os.path.exists(upgrade_script): + upgrade_env = os.environ.copy() + if CONFIG.get("python_path"): + upgrade_env["PYTHON"] = CONFIG["python_path"] + upgrade_proc = run_command( + [upgrade_script], + env=upgrade_env, + cwd=CONFIG["netbox_path"], + capture=False, + ) + if upgrade_proc.returncode != 0: + raise RuntimeError(f"upgrade.sh fehlgeschlagen (Code {upgrade_proc.returncode}).") + log_success("Upgrade-Skript erfolgreich durchgelaufen.") + else: + log_warn(f"Kein upgrade.sh unter {upgrade_script} gefunden.") + + # 5. Services-Restart + log_info("Schritt 5: Starte NetBox-Dienste neu...") + restart_services(strict=True) + + print("\n" + "=" * 60) + log_success("RESTORE ERFOLGREICH ABGESCHLOSSEN!") + print("=" * 60) + + except Exception as exc: + log_error(f"Fehler beim Restore: {exc}") + finally: + if os.path.exists(tmp_extract_dir): + shutil.rmtree(tmp_extract_dir) + pause() + + +# ========================================== +# Kernfunktionen: NetBox-Upgrade +# ========================================== +def normalize_version(raw_version): + version = raw_version.strip() + if version.lower().startswith("v"): + version = version[1:] + if not re.fullmatch(r"\d+\.\d+\.\d+", version): + raise ValueError("Bitte eine stabile Version im Format X.Y.Z eingeben, z. B. 4.6.5.") + return version + + +def version_tuple(version): + return tuple(int(part) for part in version.split(".")) + + +def detect_netbox_version(netbox_root): + """Versucht die installierte NetBox-Version ohne Django-Start zu erkennen.""" + basename = os.path.basename(os.path.realpath(netbox_root)) + match = re.search(r"netbox-v?(\d+\.\d+\.\d+)$", basename) + if match: + return match.group(1) + + git_dir = os.path.join(netbox_root, ".git") + if os.path.isdir(git_dir): + proc = run_command( + ["git", "-C", netbox_root, "describe", "--tags", "--exact-match"], + capture=True, + ) + if proc.returncode == 0: + value = proc.stdout.strip().lstrip("v") + if re.fullmatch(r"\d+\.\d+\.\d+", value): + return value + + python_bin = os.path.join(netbox_root, "venv", "bin", "python") + package_root = os.path.join(netbox_root, "netbox") + if os.path.isfile(python_bin) and os.path.isdir(package_root): + code = ( + "import sys; " + f"sys.path.insert(0, {package_root!r}); " + "import netbox; " + "print(getattr(netbox, '__version__', ''))" + ) + proc = run_command([python_bin, "-c", code], capture=True) + value = proc.stdout.strip().lstrip("v") if proc.returncode == 0 else "" + if re.fullmatch(r"\d+\.\d+\.\d+", value): + return value + + candidate_files = ( + os.path.join(netbox_root, "netbox", "netbox", "version.py"), + os.path.join(netbox_root, "netbox", "netbox", "__init__.py"), + ) + pattern = re.compile(r"(?:__version__|VERSION)\s*=\s*['\"]v?(\d+\.\d+\.\d+)['\"]") + for candidate in candidate_files: + if not os.path.isfile(candidate): + continue + try: + with open(candidate, "r", encoding="utf-8") as file_handle: + match = pattern.search(file_handle.read()) + if match: + return match.group(1) + except OSError: + pass + + return None + + +def download_release(version, destination): + url = f"https://github.com/netbox-community/netbox/archive/refs/tags/v{version}.tar.gz" + request = urllib.request.Request( + url, + headers={"User-Agent": "netbox-backup-restore-upgrade-tool/1.2"}, + ) + + try: + with urllib.request.urlopen(request, timeout=90) as response, open(destination, "wb") as output: + total = int(response.headers.get("Content-Length", "0") or 0) + downloaded = 0 + while True: + chunk = response.read(1024 * 1024) + if not chunk: + break + output.write(chunk) + downloaded += len(chunk) + if total: + percent = downloaded * 100 / total + print(f"\r Download: {percent:5.1f}%", end="", flush=True) + if total: + print() + except urllib.error.HTTPError as exc: + if exc.code == 404: + raise RuntimeError(f"NetBox-Release v{version} wurde auf GitHub nicht gefunden.") from exc + raise RuntimeError(f"HTTP-Fehler beim Download: {exc}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"Download fehlgeschlagen: {exc.reason}") from exc + + +def find_extracted_release(staging_dir, version): + expected = os.path.join(staging_dir, f"netbox-{version}") + if os.path.isdir(expected): + return expected + + directories = [ + os.path.join(staging_dir, name) + for name in os.listdir(staging_dir) + if os.path.isdir(os.path.join(staging_dir, name)) + ] + if len(directories) == 1: + return directories[0] + raise RuntimeError("Das entpackte NetBox-Verzeichnis konnte nicht eindeutig ermittelt werden.") + + +def copy_file_preserve(source, destination): + os.makedirs(os.path.dirname(destination), exist_ok=True) + proc = run_command(["cp", "-a", "--", source, destination], capture=True) + if proc.returncode != 0: + raise RuntimeError(f"Kopieren fehlgeschlagen ({source}): {proc.stderr.strip()}") + + +def copy_directory_contents_preserve(source, destination): + os.makedirs(destination, exist_ok=True) + proc = run_command(["cp", "-a", "--", os.path.join(source, "."), destination], capture=True) + if proc.returncode != 0: + raise RuntimeError(f"Kopieren fehlgeschlagen ({source}): {proc.stderr.strip()}") + + +def copy_upgrade_data(current_root, new_root): + copied = [] + + required_files = ( + ( + os.path.join(current_root, "netbox", "netbox", "configuration.py"), + os.path.join(new_root, "netbox", "netbox", "configuration.py"), + True, + ), + ( + os.path.join(current_root, "netbox", "netbox", "ldap_config.py"), + os.path.join(new_root, "netbox", "netbox", "ldap_config.py"), + False, + ), + ( + os.path.join(current_root, "local_requirements.txt"), + os.path.join(new_root, "local_requirements.txt"), + False, + ), + ( + os.path.join(current_root, "gunicorn.py"), + os.path.join(new_root, "gunicorn.py"), + False, + ), + ) + + for source, destination, mandatory in required_files: + if os.path.isfile(source): + copy_file_preserve(source, destination) + copied.append(os.path.relpath(source, current_root)) + elif mandatory: + raise RuntimeError(f"Pflichtdatei fehlt: {source}") + else: + log_info(f" -> Optional nicht vorhanden: {source}") + + directory_pairs = ( + ( + os.path.join(current_root, "netbox", "media"), + os.path.join(new_root, "netbox", "media"), + ), + ( + os.path.join(current_root, "netbox", "scripts"), + os.path.join(new_root, "netbox", "scripts"), + ), + ( + os.path.join(current_root, "netbox", "reports"), + os.path.join(new_root, "netbox", "reports"), + ), + ) + + for source, destination in directory_pairs: + if os.path.isdir(source): + copy_directory_contents_preserve(source, destination) + copied.append(os.path.relpath(source, current_root) + "/") + else: + log_info(f" -> Optional nicht vorhanden: {source}") + + return copied + + +def requirement_key(requirement): + value = requirement.strip() + if not value or value.startswith("#") or value.startswith("-"): + return None + + egg_match = re.search(r"#egg=([A-Za-z0-9_.-]+)", value, flags=re.IGNORECASE) + if egg_match: + return egg_match.group(1).lower().replace("_", "-") + + value = re.split(r"\s+#", value, maxsplit=1)[0].strip() + pep508_match = re.match(r"([A-Za-z0-9_.-]+)\s*@", value) + if pep508_match: + return pep508_match.group(1).lower().replace("_", "-") + + package_match = re.match(r"([A-Za-z0-9_.-]+)", value) + if package_match: + return package_match.group(1).lower().replace("_", "-") + return value.lower() + + +def merge_local_requirements(new_root, include_default_plugins, include_ldap): + requirements_file = os.path.join(new_root, "local_requirements.txt") + existing_lines = [] + + if os.path.exists(requirements_file): + with open(requirements_file, "r", encoding="utf-8") as file_handle: + existing_lines = file_handle.read().splitlines() + + existing_keys = { + key for key in (requirement_key(line) for line in existing_lines) if key is not None + } + + desired = [] + if include_default_plugins: + desired.extend(DEFAULT_PLUGIN_REQUIREMENTS) + if include_ldap: + desired.append(LDAP_REQUIREMENT) + + added = [] + for requirement in desired: + key = requirement_key(requirement) + if key in existing_keys: + continue + existing_lines.append(requirement) + existing_keys.add(key) + added.append(requirement) + + if desired or os.path.exists(requirements_file): + with open(requirements_file, "w", encoding="utf-8") as file_handle: + content = "\n".join(existing_lines).rstrip() + if content: + file_handle.write(content + "\n") + + return added + + +def has_csrf_setting(configuration_file): + try: + with open(configuration_file, "r", encoding="utf-8") as file_handle: + content = file_handle.read() + except OSError: + return False + return bool(re.search(r"^\s*CSRF_TRUSTED_ORIGINS\s*=", content, flags=re.MULTILINE)) + + +def parse_csrf_origins(raw_value): + origins = [] + for value in raw_value.split(","): + origin = value.strip().rstrip("/") + if not origin: + continue + parsed = urllib.parse.urlparse(origin) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise ValueError(f"Ungültiger CSRF-Origin: {origin}") + if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment: + raise ValueError(f"CSRF-Origin darf keinen Pfad/Query enthalten: {origin}") + if origin not in origins: + origins.append(origin) + return origins + + +def append_csrf_setting(configuration_file, origins): + with open(configuration_file, "a", encoding="utf-8") as file_handle: + file_handle.write("\n# Automatisch durch das NetBox-Upgrade-Tool gesetzt\n") + file_handle.write("CSRF_TRUSTED_ORIGINS = [\n") + for origin in origins: + file_handle.write(f" {origin!r},\n") + file_handle.write("]\n") + + +def stop_services(): + log_info("Stoppe NetBox-Dienste für die Umschaltung...") + for service in NETBOX_SERVICES: + proc = run_command(["systemctl", "stop", service], capture=True) + if proc.returncode != 0: + log_warn(f"{service} konnte nicht sauber gestoppt werden: {proc.stderr.strip()}") + + status = run_command(["systemctl", "is-active", service], capture=True) + if status.stdout.strip() == "active": + raise RuntimeError(f"Dienst {service} läuft trotz Stop-Befehl weiter.") + + +def restart_services(strict=False): + failures = [] + for service in NETBOX_SERVICES: + proc = run_command(["systemctl", "restart", service], capture=True) + if proc.returncode != 0: + failures.append(f"{service}: {proc.stderr.strip()}") + continue + + status = run_command(["systemctl", "is-active", service], capture=True) + if status.stdout.strip() != "active": + failures.append(f"{service}: Status ist '{status.stdout.strip() or 'unbekannt'}'") + else: + log_success(f"Dienst {service} ist aktiv.") + + if failures: + message = " | ".join(failures) + if strict: + raise RuntimeError(f"Dienst-Neustart fehlgeschlagen: {message}") + log_warn(f"Dienst-Neustart mit Problemen: {message}") + + +def switch_installation(netbox_path, extracted_root, version, stamp): + """Schaltet entweder einen Symlink oder ein echtes /opt/netbox-Verzeichnis um.""" + parent = os.path.dirname(netbox_path.rstrip("/")) or "/" + + if os.path.islink(netbox_path): + old_link_target = os.readlink(netbox_path) + version_path = os.path.join(parent, f"netbox-{version}") + if os.path.lexists(version_path): + raise RuntimeError( + f"Zielverzeichnis existiert bereits: {version_path}. " + "Bitte zuerst prüfen oder umbenennen." + ) + + os.rename(extracted_root, version_path) + temporary_link = unique_path(os.path.join(parent, f".netbox-link-{stamp}")) + os.symlink(version_path, temporary_link) + os.replace(temporary_link, netbox_path) + + return { + "mode": "symlink", + "netbox_path": netbox_path, + "old_link_target": old_link_target, + "new_path": version_path, + "old_path": os.path.realpath( + old_link_target + if os.path.isabs(old_link_target) + else os.path.join(parent, old_link_target) + ), + } + + old_path = unique_path(os.path.join(parent, f"netbox_old_{stamp}")) + os.rename(netbox_path, old_path) + try: + os.rename(extracted_root, netbox_path) + except Exception: + os.rename(old_path, netbox_path) + raise + + return { + "mode": "directory", + "netbox_path": netbox_path, + "old_path": old_path, + "new_path": netbox_path, + } + + +def rollback_filesystem_switch(switch_state, stamp): + """Nur vor Start von upgrade.sh sicher; danach könnten DB-Migrationen erfolgt sein.""" + if switch_state["mode"] == "symlink": + parent = os.path.dirname(switch_state["netbox_path"].rstrip("/")) or "/" + temporary_link = unique_path(os.path.join(parent, f".netbox-rollback-{stamp}")) + os.symlink(switch_state["old_link_target"], temporary_link) + os.replace(temporary_link, switch_state["netbox_path"]) + return + + failed_path = unique_path(f"{switch_state['netbox_path']}_failed_{stamp}") + os.rename(switch_state["netbox_path"], failed_path) + os.rename(switch_state["old_path"], switch_state["netbox_path"]) + switch_state["failed_path"] = failed_path + + +def execute_upgrade(): + print_header("NETBOX UPGRADE") + + if os.name != "posix" or os.geteuid() != 0: + log_error("Das Upgrade muss unter Linux als root bzw. mit sudo ausgeführt werden.") + pause() + return + + netbox_path = os.path.abspath(CONFIG["netbox_path"]) + if not os.path.lexists(netbox_path): + log_error(f"NetBox-Pfad existiert nicht: {netbox_path}") + pause() + return + + current_root = os.path.realpath(netbox_path) + current_config = os.path.join(current_root, "netbox", "netbox", "configuration.py") + if not os.path.isfile(current_config): + log_error(f"Keine gültige NetBox-Installation erkannt; configuration.py fehlt: {current_config}") + pause() + return + + current_version = detect_netbox_version(current_root) + if current_version: + log_info(f"Erkannte installierte Version: v{current_version}") + else: + log_warn("Die installierte NetBox-Version konnte nicht sicher erkannt werden.") + + try: + target_version = normalize_version(input("Zielversion (X.Y.Z, optional mit v): ")) + except ValueError as exc: + log_error(str(exc)) + pause() + return + + if current_version: + current_tuple = version_tuple(current_version) + target_tuple = version_tuple(target_version) + if target_tuple <= current_tuple: + log_error( + f"Die Zielversion v{target_version} ist nicht neuer als v{current_version}. " + "Downgrades werden absichtlich nicht automatisiert." + ) + pause() + return + if target_tuple[0] > current_tuple[0]: + log_warn( + "Major-Upgrade erkannt. NetBox erlaubt einen Major-Sprung nur von der " + "jeweils letzten Minor-Version des bisherigen Major-Releases." + ) + + print() + log_warn("Vor jedem Upgrade sind ein aktuelles Datenbank-/Media-Backup und die Release Notes Pflicht.") + log_info(f"Release Notes: https://github.com/netbox-community/netbox/releases/tag/v{target_version}") + if input("Backup erstellt und Release Notes geprüft? (j/N): ").strip().lower() != "j": + log_info("Upgrade abgebrochen. Bitte zuerst Backup und Release Notes prüfen.") + pause() + return + + include_plugins = ( + input("Genannte Standard-Plugins in local_requirements.txt ergänzen? (J/n): ") + .strip() + .lower() + != "n" + ) + + ldap_present = os.path.isfile( + os.path.join(current_root, "netbox", "netbox", "ldap_config.py") + ) + if ldap_present: + log_info("LDAP-Konfiguration erkannt; django-auth-ldap wird sichergestellt.") + + csrf_present = has_csrf_setting(current_config) + log_info( + "CSRF_TRUSTED_ORIGINS ist in configuration.py vorhanden." + if csrf_present + else "CSRF_TRUSTED_ORIGINS wurde nicht als direkte Zuweisung erkannt." + ) + csrf_raw = input( + "Neue CSRF_TRUSTED_ORIGINS kommagetrennt eingeben " + "(leer = vorhandene Konfiguration übernehmen): " + ).strip() + + try: + csrf_origins = parse_csrf_origins(csrf_raw) if csrf_raw else [] + except ValueError as exc: + log_error(str(exc)) + pause() + return + + if not csrf_present and not csrf_origins: + log_warn( + "Kein CSRF_TRUSTED_ORIGINS erkannt. Die Einstellung kann allerdings auch " + "dynamisch/importiert gesetzt sein." + ) + if input("Trotzdem fortfahren? (j/N): ").strip().lower() != "j": + log_info("Upgrade abgebrochen.") + pause() + return + + print() + log_warn( + "Plugin-Kompatibilität wird erst während upgrade.sh/Pip geprüft. " + "Nicht kompatible Plugins können das Upgrade stoppen." + ) + if input(f"Upgrade auf NetBox v{target_version} jetzt starten? (j/N): ").strip().lower() != "j": + log_info("Upgrade abgebrochen.") + pause() + return + + install_parent = os.path.dirname(netbox_path.rstrip("/")) or "/" + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + staging_dir = None + archive_path = None + switch_state = None + services_stopped = False + upgrade_started = False + + try: + if not shutil.which("cp"): + raise RuntimeError("Der Befehl 'cp' wurde nicht gefunden.") + if not shutil.which("systemctl"): + raise RuntimeError("Der Befehl 'systemctl' wurde nicht gefunden.") + + os.makedirs(CONFIG["backup_dir"], exist_ok=True) + fd, archive_path = tempfile.mkstemp( + prefix=f"netbox-v{target_version}-", + suffix=".tar.gz", + dir=CONFIG["backup_dir"], + ) + os.close(fd) + staging_dir = tempfile.mkdtemp(prefix=".netbox-upgrade-", dir=install_parent) + + log_info(f"Schritt 1: Lade NetBox v{target_version} herunter...") + download_release(target_version, archive_path) + log_success("Release-Archiv heruntergeladen.") + + log_info("Schritt 2: Entpacke und prüfe Release-Archiv...") + with tarfile.open(archive_path, "r:gz") as tarf: + safe_extract_tar(tarf, staging_dir) + extracted_root = find_extracted_release(staging_dir, target_version) + + required_upgrade_script = os.path.join(extracted_root, "upgrade.sh") + if not os.path.isfile(required_upgrade_script): + raise RuntimeError("Das Release enthält kein upgrade.sh und ist nicht plausibel.") + log_success("Release-Archiv erfolgreich geprüft.") + + log_info("Schritt 3: Übernehme Konfiguration, Media, Skripte und Reports...") + copied = copy_upgrade_data(current_root, extracted_root) + for item in copied: + log_info(f" -> übernommen: {item}") + + added_requirements = merge_local_requirements( + extracted_root, + include_default_plugins=include_plugins, + include_ldap=ldap_present, + ) + if added_requirements: + for requirement in added_requirements: + log_info(f" -> Requirement ergänzt: {requirement}") + else: + log_info(" -> Keine neuen Requirements erforderlich.") + + new_configuration = os.path.join( + extracted_root, "netbox", "netbox", "configuration.py" + ) + if csrf_origins: + append_csrf_setting(new_configuration, csrf_origins) + log_success("CSRF_TRUSTED_ORIGINS wurde in der neuen Konfiguration gesetzt.") + elif has_csrf_setting(new_configuration): + log_success("Vorhandenes CSRF_TRUSTED_ORIGINS wurde übernommen.") + else: + log_warn("CSRF_TRUSTED_ORIGINS bleibt unverändert bzw. dynamisch konfiguriert.") + + log_info("Schritt 4: Stoppe Dienste und schalte auf die neue Version um...") + stop_services() + services_stopped = True + switch_state = switch_installation( + netbox_path, + extracted_root, + target_version, + stamp, + ) + log_success("NetBox-Verzeichnis wurde umgeschaltet.") + + log_info("Schritt 5: Führe /opt/netbox/upgrade.sh aus...") + upgrade_script = os.path.join(netbox_path, "upgrade.sh") + upgrade_env = os.environ.copy() + if CONFIG.get("python_path"): + if not os.path.isfile(CONFIG["python_path"]): + raise RuntimeError( + f"Konfigurierter Python-Pfad existiert nicht: {CONFIG['python_path']}" + ) + upgrade_env["PYTHON"] = CONFIG["python_path"] + log_info(f"Verwende PYTHON={CONFIG['python_path']}") + + upgrade_started = True + upgrade_proc = run_command( + [upgrade_script], + env=upgrade_env, + cwd=netbox_path, + capture=False, + ) + if upgrade_proc.returncode != 0: + raise RuntimeError( + f"NetBox upgrade.sh wurde mit Code {upgrade_proc.returncode} beendet." + ) + log_success("upgrade.sh wurde erfolgreich abgeschlossen.") + + log_info("Schritt 6: Starte NetBox-Dienste neu und prüfe den Status...") + restart_services(strict=True) + services_stopped = False + + print("\n" + "=" * 60) + log_success(f"UPGRADE AUF NETBOX v{target_version} ERFOLGREICH!") + if switch_state["mode"] == "directory": + log_info(f"Alte Installation verbleibt unter: {switch_state['old_path']}") + else: + log_info(f"Vorheriges Release verbleibt unter: {switch_state['old_path']}") + log_info(f"Aktives Release: {switch_state['new_path']}") + print("=" * 60) + + except Exception as exc: + log_error(f"Upgrade fehlgeschlagen: {exc}") + + if switch_state and not upgrade_started: + log_warn("Fehler trat vor Start von upgrade.sh auf; Dateisystem-Umschaltung wird zurückgenommen.") + try: + rollback_filesystem_switch(switch_state, stamp) + log_success("Vorherige Installation wurde wieder aktiviert.") + if services_stopped: + restart_services(strict=False) + services_stopped = False + except Exception as rollback_exc: + log_error(f"Automatischer Dateisystem-Rollback fehlgeschlagen: {rollback_exc}") + elif switch_state and upgrade_started: + log_error( + "Kein automatischer Rollback nach Start von upgrade.sh: " + "Datenbank-Migrationen könnten bereits angewendet worden sein." + ) + log_info(f"Vorherige Installation liegt weiterhin unter: {switch_state['old_path']}") + log_info("Für einen vollständigen Rollback Datenbank/Media aus dem Backup wiederherstellen.") + elif services_stopped: + restart_services(strict=False) + services_stopped = False + + finally: + if archive_path and os.path.exists(archive_path): + os.remove(archive_path) + if staging_dir and os.path.exists(staging_dir): + shutil.rmtree(staging_dir, ignore_errors=True) + + pause() + + +# ========================================== +# Einstellungen +# ========================================== +def settings_menu(): + while True: + print_header("EINSTELLUNGEN") + keys = list(CONFIG.keys()) + for index, key in enumerate(keys, 1): + value = "*****" if key == "db_pass" and CONFIG[key] else CONFIG[key] + print( + f" {Colors.BOLD}{index}.{Colors.ENDC} " + f"{key.ljust(15)}: {Colors.BLUE}{value}{Colors.ENDC}" + ) + + print(f"\n {Colors.BOLD}S.{Colors.ENDC} Speichern & zurück zum Hauptmenü") + + choice = input("\nWelchen Wert möchtest du ändern? (Zahl/S): ").strip().lower() + + if choice == "s": + break + if choice.isdigit() and 1 <= int(choice) <= len(keys): + key_to_edit = keys[int(choice) - 1] + current_display = "*****" if key_to_edit == "db_pass" and CONFIG[key_to_edit] else CONFIG[key_to_edit] + new_value = input(f"Neuer Wert für {key_to_edit} (aktuell: {current_display}): ") + if new_value.strip() != "": + CONFIG[key_to_edit] = new_value.strip() + + +# ========================================== +# Hauptmenü +# ========================================== +def main_menu(): + while True: + print_header("NETBOX BACKUP, RESTORE & UPGRADE TOOL") + print(f" {Colors.BOLD}1.{Colors.ENDC} Backup erstellen") + print(f" {Colors.BOLD}2.{Colors.ENDC} Backup wiederherstellen") + print(f" {Colors.BOLD}3.{Colors.ENDC} NetBox upgraden") + print(f" {Colors.BOLD}4.{Colors.ENDC} Einstellungen prüfen / anpassen") + print(f" {Colors.BOLD}5.{Colors.ENDC} Beenden\n") + + choice = input(Colors.BOLD + "Wähle eine Aktion (1-5): " + Colors.ENDC).strip() + + if choice == "1": + execute_backup() + elif choice == "2": + execute_restore() + elif choice == "3": + execute_upgrade() + elif choice == "4": + settings_menu() + elif choice == "5": + clear_screen() + print("Beendet. Auf Wiedersehen!\n") + sys.exit(0) + else: + log_error("Ungültige Eingabe.") + time.sleep(1) + + +if __name__ == "__main__": + if os.geteuid() != 0: + print( + Colors.WARNING + + "Hinweis: Dieses Skript sollte idealerweise als 'root' oder mit 'sudo' ausgeführt werden," + + Colors.ENDC + ) + print( + Colors.WARNING + + "da systemctl, chown und der Zugriff auf /opt/netbox erhöhte Rechte erfordern.\n" + + Colors.ENDC + ) + time.sleep(2) + + try: + main_menu() + except KeyboardInterrupt: + print("\n\nAbbruch durch Benutzer. Ciao!") + sys.exit(0) \ No newline at end of file diff --git a/nettrans.py b/nettrans.py deleted file mode 100644 index 2896da1..0000000 --- a/nettrans.py +++ /dev/null @@ -1,312 +0,0 @@ -#!/usr/bin/env python3 -import os -import sys -import json -import zipfile -import subprocess -import shutil -import time -from datetime import datetime - -# ========================================== -# Standard Konfiguration -# ========================================== -CONFIG = { - 'netbox_path': '/opt/netbox', - 'db_host': 'localhost', - 'db_name': 'netbox', - 'db_user': 'netbox', - 'db_pass': '', - 'backup_dir': '/tmp' -} - -# ========================================== -# Hilfsfunktionen für die Shell-GUI -# ========================================== -class Colors: - HEADER = '\033[95m' - BLUE = '\033[94m' - GREEN = '\033[92m' - WARNING = '\033[93m' - FAIL = '\033[91m' - ENDC = '\033[0m' - BOLD = '\033[1m' - -def clear_screen(): - os.system('clear' if os.name == 'posix' else 'cls') - -def print_header(title): - clear_screen() - print(Colors.HEADER + Colors.BOLD + "=" * 60 + Colors.ENDC) - print(Colors.HEADER + Colors.BOLD + f" {title}".ljust(59) + Colors.ENDC) - print(Colors.HEADER + Colors.BOLD + "=" * 60 + Colors.ENDC + "\n") - -def log_info(msg): - print(Colors.BLUE + f"[*] {msg}" + Colors.ENDC) - -def log_success(msg): - print(Colors.GREEN + Colors.BOLD + f"[+] {msg}" + Colors.ENDC) - -def log_error(msg): - print(Colors.FAIL + Colors.BOLD + f"[-] {msg}" + Colors.ENDC) - -def log_warn(msg): - print(Colors.WARNING + f"[!] {msg}" + Colors.ENDC) - -def pause(): - input(Colors.BOLD + "\nDrücke [ENTER], um ins Hauptmenü zurückzukehren..." + Colors.ENDC) - -# ========================================== -# Kernfunktionen: Backup & Restore -# ========================================== -def execute_backup(): - print_header("BACKUP PROZESS") - - log_info("Konfiguriere den Backup-Umfang:") - do_db = input(" 1. Datenbank sichern? (J/n): ").strip().lower() != 'n' - - exclude_changelog = False - exclude_users = False - - if do_db: - exclude_changelog = input(Colors.BLUE + " -> Changelog (extras_objectchange) auslassen? Spart viel Platz. (j/N): " + Colors.ENDC).strip().lower() == 'j' - print(Colors.WARNING + " Hinweis: Das Auslassen von Benutzern kann beim Restore zu Foreign-Key Fehlern führen, falls Objekte diesen Benutzern zugewiesen sind." + Colors.ENDC) - exclude_users = input(Colors.BLUE + " -> Benutzer (User, Gruppen, Tokens) auslassen? (j/N): " + Colors.ENDC).strip().lower() == 'j' - - do_media = input("\n 2. Media-Dateien (Bilder/Dokumente) sichern? (J/n): ").strip().lower() != 'n' - - if not do_db and not do_media: - log_warn("Keine Komponenten zum Sichern ausgewählt. Abbruch.") - pause() - return - - media_path = os.path.join(CONFIG['netbox_path'], "netbox/media") - stamp = datetime.now().strftime("%Y%m%d_%H%M%S") - archive_name = f"netbox_backup_{stamp}" - zip_file_path = os.path.join(CONFIG['backup_dir'], f"{archive_name}.zip") - tmp_dir = os.path.join(CONFIG['backup_dir'], f"nb_tmp_{stamp}") - - os.makedirs(tmp_dir, exist_ok=True) - - try: - # 1. Datenbank-Dump - if do_db: - log_info("\nSchritt 1: Erstelle PostgreSQL Datenbank-Dump...") - dump_file = os.path.join(tmp_dir, "netbox_database.dump") - - env = os.environ.copy() - if CONFIG['db_pass']: env["PGPASSWORD"] = CONFIG['db_pass'] - - cmd = ["pg_dump", "-Fc", "-h", CONFIG['db_host'], "-U", CONFIG['db_user'], "-d", CONFIG['db_name'], "-f", dump_file] - - # Ausnahmen hinzufügen (--exclude-table-data lässt die Tabellenstruktur intakt, aber exportiert keine Inhalte) - if exclude_changelog: - cmd.append("--exclude-table-data=extras_objectchange") - log_info(" -> Changelog-Daten werden übersprungen.") - if exclude_users: - cmd.extend(["--exclude-table-data=auth_user", "--exclude-table-data=auth_group", "--exclude-table-data=users_token"]) - log_info(" -> Benutzer-Daten werden übersprungen.") - - process = subprocess.run(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - if process.returncode != 0: - log_error(f"pg_dump fehlgeschlagen: {process.stderr}") - raise Exception("Datenbank-Backup fehlgeschlagen.") - log_success("PostgreSQL Dump erfolgreich erstellt.") - else: - log_info("\nSchritt 1: Datenbank-Backup übersprungen.") - - # 2. Media-Dateien - if do_media: - log_info("Schritt 2: Kopiere Media-Dateien (Bilder, Dokumente)...") - if os.path.exists(media_path): - dst_media = os.path.join(tmp_dir, "media") - shutil.copytree(media_path, dst_media, dirs_exist_ok=True) - log_success("Media-Ordner erfolgreich gesichert.") - else: - log_warn(f"Kein Media-Verzeichnis unter {media_path} gefunden.") - else: - log_info("Schritt 2: Media-Backup übersprungen.") - - # 3. Metadaten - log_info("Schritt 3: Erstelle Metadaten...") - meta = { - "version": "1.1", - "timestamp": stamp, - "source_db": CONFIG['db_name'], - "contains_db": do_db, - "contains_media": do_media, - "excluded_changelog": exclude_changelog, - "excluded_users": exclude_users - } - with open(os.path.join(tmp_dir, "metadata.json"), "w") as f: - json.dump(meta, f, indent=4) - - # 4. ZIP erstellen - log_info(f"Schritt 4: Komprimiere Daten nach {zip_file_path} ...") - with zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf: - for root, _, files in os.walk(tmp_dir): - for file in files: - full_path = os.path.join(root, file) - rel_path = os.path.relpath(full_path, tmp_dir) - zipf.write(full_path, rel_path) - - print("\n" + "="*60) - log_success("BACKUP ERFOLGREICH ABGESCHLOSSEN!") - log_success(f"Datei liegt hier: {zip_file_path}") - print("="*60) - - except Exception as e: - log_error(f"Kritischer Fehler: {str(e)}") - finally: - if os.path.exists(tmp_dir): - shutil.rmtree(tmp_dir) - pause() - -def execute_restore(): - print_header("RESTORE PROZESS") - log_warn("ACHTUNG: Dieser Vorgang überschreibt bestehende Daten auf der Ziel-Instanz!") - confirm = input("Bist du sicher, dass du fortfahren möchtest? (j/N): ") - - if confirm.lower() != 'j': - log_info("Abbruch durch Benutzer.") - pause() - return - - restore_file = input(f"\nPfad zur Backup ZIP-Datei (z.B. /tmp/netbox_backup.zip): ").strip() - - if not restore_file or not os.path.exists(restore_file): - log_error(f"Die Datei '{restore_file}' existiert nicht!") - pause() - return - - media_path = os.path.join(CONFIG['netbox_path'], "netbox/media") - stamp = datetime.now().strftime("%Y%m%d_%H%M%S") - tmp_extract_dir = f"/tmp/nb_restore_{stamp}" - os.makedirs(tmp_extract_dir, exist_ok=True) - - try: - log_info(f"\nSchritt 1: Entpacke {restore_file} ...") - with zipfile.ZipFile(restore_file, 'r') as zipf: - zipf.extractall(tmp_extract_dir) - - meta_file = os.path.join(tmp_extract_dir, "metadata.json") - if not os.path.exists(meta_file): - raise Exception("Ungültiges Backup: metadata.json fehlt.") - - with open(meta_file, 'r') as f: - meta = json.load(f) - - # 2. Datenbank Restore - if meta.get("contains_db", True): - log_info("Schritt 2: Stelle PostgreSQL Datenbank wieder her...") - dump_file = os.path.join(tmp_extract_dir, "netbox_database.dump") - env = os.environ.copy() - if CONFIG['db_pass']: env["PGPASSWORD"] = CONFIG['db_pass'] - - log_info("Lösche und erstelle Datenbank neu (Bereinigung)...") - subprocess.run(["dropdb", "-h", CONFIG['db_host'], "-U", CONFIG['db_user'], "--if-exists", CONFIG['db_name']], env=env, stderr=subprocess.PIPE) - subprocess.run(["createdb", "-h", CONFIG['db_host'], "-U", CONFIG['db_user'], "-O", CONFIG['db_user'], CONFIG['db_name']], env=env, stderr=subprocess.PIPE) - - log_info("Importiere Dump...") - proc_restore = subprocess.run(["pg_restore", "-h", CONFIG['db_host'], "-U", CONFIG['db_user'], "-d", CONFIG['db_name'], "--clean", "--no-owner", dump_file], env=env, stderr=subprocess.PIPE, text=True) - - # pg_restore wirft oft Warnungen bei fehlenden Rechten, die aber meist unkritisch sind - if proc_restore.returncode not in [0, 1]: - log_warn(f"Import abgeschlossen mit Hinweisen (Code {proc_restore.returncode}). Dies kann passieren, wenn User exkludiert wurden.") - log_success("Datenbank erfolgreich wiederhergestellt.") - else: - log_info("Schritt 2: Übersprungen (Keine Datenbank im Backup enthalten).") - - # 3. Media Restore - if meta.get("contains_media", True): - log_info("Schritt 3: Stelle Media-Dateien wieder her...") - src_media = os.path.join(tmp_extract_dir, "media") - if os.path.exists(src_media): - if os.path.exists(media_path): shutil.rmtree(media_path) - shutil.copytree(src_media, media_path, dirs_exist_ok=True) - subprocess.run(["chown", "-R", "netbox:netbox", media_path], stderr=subprocess.PIPE) - log_success("Media-Dateien wiederhergestellt.") - else: - log_info("Schritt 3: Übersprungen (Keine Media-Dateien im Backup enthalten).") - - # 4. NetBox Upgrade - log_info("Schritt 4: Führe NetBox upgrade.sh aus (Cache-Clear & Migrations)...") - upgrade_script = os.path.join(CONFIG['netbox_path'], "upgrade.sh") - if os.path.exists(upgrade_script): - subprocess.run([upgrade_script], stdout=subprocess.PIPE, stderr=subprocess.PIPE) - log_success("Upgrade-Skript durchgelaufen.") - - # 5. Services Restart - log_info("Schritt 5: Starte NetBox Dienste neu...") - subprocess.run(["systemctl", "restart", "netbox", "netbox-rq"], stderr=subprocess.PIPE) - - print("\n" + "="*60) - log_success("RESTORE ERFOLGREICH ABGESCHLOSSEN!") - print("="*60) - - except Exception as e: - log_error(f"Fehler beim Restore: {str(e)}") - finally: - if os.path.exists(tmp_extract_dir): - shutil.rmtree(tmp_extract_dir) - pause() - -def settings_menu(): - while True: - print_header("EINSTELLUNGEN") - keys = list(CONFIG.keys()) - for i, key in enumerate(keys, 1): - val = "*****" if key == 'db_pass' and CONFIG[key] else CONFIG[key] - print(f" {Colors.BOLD}{i}.{Colors.ENDC} {key.ljust(15)}: {Colors.BLUE}{val}{Colors.ENDC}") - - print(f"\n {Colors.BOLD}S.{Colors.ENDC} Speichern & Zurück zum Hauptmenü") - - choice = input("\nWelchen Wert möchtest du ändern? (Zahl/S): ").strip().lower() - - if choice == 's': - break - elif choice.isdigit() and 1 <= int(choice) <= len(keys): - key_to_edit = keys[int(choice)-1] - new_val = input(f"Neuer Wert für {key_to_edit} (aktuell: {CONFIG[key_to_edit]}): ") - if new_val.strip() != "": - CONFIG[key_to_edit] = new_val.strip() - -# ========================================== -# Hauptmenü -# ========================================== -def main_menu(): - while True: - print_header("NETBOX BACKUP & RESTORE TOOL") - print(f" {Colors.BOLD}1.{Colors.ENDC} Backup erstellen") - print(f" {Colors.BOLD}2.{Colors.ENDC} Backup wiederherstellen") - print(f" {Colors.BOLD}3.{Colors.ENDC} Einstellungen prüfen / anpassen") - print(f" {Colors.BOLD}4.{Colors.ENDC} Beenden\n") - - choice = input(Colors.BOLD + "Wähle eine Aktion (1-4): " + Colors.ENDC).strip() - - if choice == '1': - execute_backup() - elif choice == '2': - execute_restore() - elif choice == '3': - settings_menu() - elif choice == '4': - clear_screen() - print("Beendet. Auf Wiedersehen!\n") - sys.exit(0) - else: - log_error("Ungültige Eingabe.") - time.sleep(1) - -if __name__ == "__main__": - if os.geteuid() != 0: - print(Colors.WARNING + "Hinweis: Dieses Skript sollte idealerweise als 'root' oder mit 'sudo' ausgeführt werden," + Colors.ENDC) - print(Colors.WARNING + "da systemctl, chown und der Zugriff auf /opt/netbox erhöhte Rechte erfordern.\n" + Colors.ENDC) - time.sleep(2) - - try: - main_menu() - except KeyboardInterrupt: - print("\n\nAbbruch durch Benutzer. Ciao!") - sys.exit(0) \ No newline at end of file