#!/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)